feat(portfolio): 回测可观测性 L1+L2+L3(治排查黑盒)

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

@@ -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)
+    # 区分 failed(422+error)/running·pending(404 not ready)/done(result)/not found(404)。
-    if r is None:
+    # 原实现对跑中/失败都 404,排查困难(分不清慢/死/失败)。
-        raise HTTPException(status_code=404, detail="result not ready")
+    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:

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,
+            argv, cwd=cwd, stdout=subprocess.PIPE, stderr=None,
-            timeout=_TIMEOUT, check=False,
+            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:]
+        # stderr 未捕获(继承→docker logs 实时),此处只记 rc;回测日志查 docker logs [runner] 行
-        logger.error("[portfolio_worker] task=%s rc=%s stderr=%s",
+        logger.error("[portfolio_worker] task=%s rc=%s (runner_backtest 日志已实时输出 docker logs)",
-                     task_id, proc.returncode, stderr_tail)
+                     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 "")

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}")