From 9588a77a013af272f54d34fadea6fff40fb7e985 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Sun, 2 Aug 2026 13:42:06 +0800 Subject: [PATCH] =?UTF-8?q?feat(portfolio):=20=E5=9B=9E=E6=B5=8B=E5=8F=AF?= =?UTF-8?q?=E8=A7=82=E6=B5=8B=E6=80=A7=20L1+L2+L3(=E6=B2=BB=E6=8E=92?= =?UTF-8?q?=E6=9F=A5=E9=BB=91=E7=9B=92)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit L1 worker logging: portfolio_worker 模块级 basicConfig(spawn子进程logger进docker logs) + subprocess stderr=None 继承(runner_backtest日志实时进docker logs,原capture_output吞了) + 失败处理适配(stderr未捕获,改提示查docker logs) L2 task状态区分: 加 GET /portfolio/task/{id}/status(返state+error) + get_result 区分 failed(422+error)/running·pending(404 not ready)/done(原全404分不清) L3 _wait_future日志: runner.py except 加 logger.error(原只task.fail静默,治非timeout异常排查) 前端无需改(轮询通用/task/{id}已处理failed) --- sanguo_api/routes_portfolio.py | 42 +++++++++++++++++++++++-- sanguo_orchestrator/portfolio_worker.py | 21 +++++++++---- sanguo_orchestrator/runner.py | 4 +++ 3 files changed, 58 insertions(+), 9 deletions(-) diff --git a/sanguo_api/routes_portfolio.py b/sanguo_api/routes_portfolio.py index 3ad4c85..0d23fc5 100644 --- a/sanguo_api/routes_portfolio.py +++ b/sanguo_api/routes_portfolio.py @@ -59,9 +59,28 @@ async def run_portfolio_backtest(req: PortfolioBacktestRequest): def get_portfolio_result(task_id: str): """取组合回测结果。从 BacktestResult.statistics 取 metrics/stocks_selected/period, 从 equity_curve/trades(DataFrame)转 records。""" - r = get_orchestrator().get_result(task_id) - if r is None: - raise HTTPException(status_code=404, detail="result not ready") + # 区分 failed(422+error)/running·pending(404 not ready)/done(result)/not found(404)。 + # 原实现对跑中/失败都 404,排查困难(分不清慢/死/失败)。 + orch = get_orchestrator() + state = orch.get_status(task_id) + if state is None: + r = orch.get_result(task_id) # 不在内存池(历史 task 持久化 DB)→直接查 DB + if r is None: + raise HTTPException(status_code=404, detail="task not found") + elif state.value == "failed": + task = orch.pool.get_task(task_id) + raise HTTPException( + status_code=422, + detail=f"task failed: {task.error_msg if task else 'unknown'}", + ) + elif state.value != "done": + raise HTTPException( + status_code=404, detail=f"result not ready (state={state.value})" + ) + else: + r = orch.get_result(task_id) + if r is None: + raise HTTPException(status_code=404, detail="done but result missing") stats: dict = r.statistics or {} metrics = stats.get("metrics", {}) @@ -79,6 +98,23 @@ def get_portfolio_result(task_id: str): } +@router.get("/portfolio/task/{task_id}/status", dependencies=[Depends(verify_token)]) +def get_portfolio_status(task_id: str): + """组合回测任务状态(区分 pending/running/done/failed + error 详情)。 + 通用 GET /task/{id} 也返 status,本接口聚焦 portfolio 命名空间 + error 字段。""" + orch = get_orchestrator() + state = orch.get_status(task_id) + if state is None: + raise HTTPException(status_code=404, detail="task not found") + task = orch.pool.get_task(task_id) + return { + "task_id": task_id, + "state": state.value if hasattr(state, "value") else str(state), + "stage": orch.pool.get_stage(task_id) or "", + "error": task.error_msg if (task and isinstance(task.error_msg, str)) else None, + } + + def _df_to_records(df: Any) -> list[dict]: """DataFrame/list → list[dict] (empty-safe).""" if df is None: diff --git a/sanguo_orchestrator/portfolio_worker.py b/sanguo_orchestrator/portfolio_worker.py index 0427136..d8b9d26 100644 --- a/sanguo_orchestrator/portfolio_worker.py +++ b/sanguo_orchestrator/portfolio_worker.py @@ -17,6 +17,13 @@ from typing import Any, Optional logger = logging.getLogger(__name__) +# ProcessPool spawn 子进程不继承主进程 logging 配置 → 回测日志黑盒。 +# 模块级 basicConfig 让本 worker 子进程 logger(INFO) 输出到 stderr → docker logs。 +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s: %(message)s", +) + # Three-machine adaptive config (mirrors routes_portfolio.py original logic) _VPS_HOST = "49.232.102.198" _VPS_WORKDIR = r"C:\\sanguo_vnpy_v2" @@ -57,9 +64,11 @@ def run_portfolio_task(spec: dict) -> Any: logger.info("[portfolio_worker] task=%s running: %s", task_id, " ".join(argv[3:])) try: + # stderr=None(继承): runner_backtest 日志实时进 docker logs(不用等结束); + # stdout=PIPE 捕获(取最后 JSON 行)。原 capture_output=True 吞了 stderr。 proc = subprocess.run( - argv, cwd=cwd, capture_output=True, text=True, - timeout=_TIMEOUT, check=False, + argv, cwd=cwd, stdout=subprocess.PIPE, stderr=None, + text=True, timeout=_TIMEOUT, check=False, ) except subprocess.TimeoutExpired: # 全市场全周期慢→3h仍超→显式log(否则_wait_future静默task.fail难排查) @@ -73,11 +82,11 @@ def run_portfolio_task(spec: dict) -> Any: ) if proc.returncode != 0: - stderr_tail = (proc.stderr or "")[-2000:] - logger.error("[portfolio_worker] task=%s rc=%s stderr=%s", - task_id, proc.returncode, stderr_tail) + # stderr 未捕获(继承→docker logs 实时),此处只记 rc;回测日志查 docker logs [runner] 行 + logger.error("[portfolio_worker] task=%s rc=%s (runner_backtest 日志已实时输出 docker logs)", + task_id, proc.returncode) raise RuntimeError( - f"runner_backtest failed (rc={proc.returncode}): {stderr_tail}" + f"runner_backtest failed (rc={proc.returncode}); 详情查 docker logs [runner] 日志" ) data = _parse_stdout_json(proc.stdout or "") diff --git a/sanguo_orchestrator/runner.py b/sanguo_orchestrator/runner.py index 62d0d85..149f4ba 100644 --- a/sanguo_orchestrator/runner.py +++ b/sanguo_orchestrator/runner.py @@ -166,6 +166,10 @@ class Orchestrator: result = await asyncio.wrap_future(fut) await self._on_done(task_id, result) except Exception as e: + import logging + logging.getLogger(__name__).error( + "task %s failed in pool: %s: %s", task_id, type(e).__name__, e + ) task = self.pool.get_task(task_id) if task: task.fail(f"{type(e).__name__}: {e}")