feat(portfolio): 组合回测异步化(接入 orchestrator task 框架)

routes_portfolio 同步 subprocess.run(timeout=600) >10min 回测被 504 杀。改异步复用现有 sanguo_orchestrator:submit_portfolio 仿 submit_factor, 子进程执行搬至 portfolio_worker.py(三机自适应保留,timeout 600→3600), 复用 _wait_future/_on_done 持久化。前端 onSubmit 改 POST 拿 task_id → 轮询 /task/{id} → done 取 /portfolio/task/{id}/result。 - runner.py: + submit_portfolio + _portfolio_worker(lazy import) - portfolio_worker.py(新): 三机自适应 argv + 子进程 + stdout JSON → BacktestResult - routes_portfolio.py: POST async 返 task_id + GET /portfolio/task/{id}/result - 前端: postPortfolioBacktest 返 task_id + 轮询 status + 取 result(去 600s timeout) 不动策略逻辑/runner_backtest 核心/现有 task 接口。
2026-08-01 13:52:46 +08:00
parent 2ca2facce3
commit 4986e5c3c4
5 changed files with 329 additions and 120 deletions
@@ -58,11 +58,33 @@ export interface PortfolioBacktestResult {

 export async function postPortfolioBacktest(
  req: PortfolioBacktestReq,
-): Promise<PortfolioBacktestResult> {
-  const { data } = await apiClient.post<PortfolioBacktestResult>(
+): Promise<string> {
+  const { data } = await apiClient.post<{ task_id: string }>(
    '/portfolio/backtest',
    req,
-    { timeout: 600000 },
+  )
+  return data.task_id
+}
+
+export interface PortfolioTaskStatus {
+  task_id: string
+  status: string
+  stage: string
+  error_msg: string | null
+}
+
+export async function getPortfolioTaskStatus(
+  taskId: string,
+): Promise<PortfolioTaskStatus> {
+  const { data } = await apiClient.get<PortfolioTaskStatus>(`/task/${taskId}`)
+  return data
+}
+
+export async function getPortfolioResult(
+  taskId: string,
+): Promise<PortfolioBacktestResult> {
+  const { data } = await apiClient.get<PortfolioBacktestResult>(
+    `/portfolio/task/${taskId}/result`,
  )
  return data
 }
@@ -1,11 +1,13 @@
 <script setup lang="ts">
-import { ref, reactive, onMounted, watch } from 'vue'
+import { ref, reactive, onMounted, onUnmounted, watch } from 'vue'
 import { ElMessage } from 'element-plus'
 import type { EChartsCoreOption } from 'echarts'
 import { useChart } from '@/composables/useChart'
 import { darkTitle, darkTooltip, darkGrid, darkAxis } from '@/utils/echartsDark'
 import {
  postPortfolioBacktest,
+  getPortfolioTaskStatus,
+  getPortfolioResult,
  type PortfolioBacktestResult,
  type EquityPoint,
  type StockPicked,
@@ -21,6 +23,8 @@ const trades = ref<PortfolioTrade[]>([])
 const metrics = ref<PortfolioMetrics | null>(null)
 const period = ref<{ start: string; end: string; trading_days: number } | null>(null)
 const errorMsg = ref('')
+const stageHint = ref('')
+let pollTimer: ReturnType<typeof setInterval> | null = null

 const form = reactive({
  pool: 'hs300_subset',
@@ -65,6 +69,9 @@ function renderEquity(): void {
 onMounted(() => {
  // 等 dom 挂载后渲染(若有初始数据)
 })
+onUnmounted(() => {
+  if (pollTimer) clearInterval(pollTimer)
+})
 watch(equityCurve, renderEquity, { deep: true, flush: 'post' })

 function fmtPct(v: number | null | undefined): string {
@@ -79,19 +86,40 @@ function fmtNum(v: number | null | undefined, digits = 2): string {
 async function onSubmit(): Promise<void> {
  submitting.value = true
  errorMsg.value = ''
+  stageHint.value = ''
  result.value = null
  equityCurve.value = []
  stocks.value = []
  trades.value = []
  metrics.value = null
  try {
-    const r = await postPortfolioBacktest({
+    const tid = await postPortfolioBacktest({
      pool: form.pool,
      start_date: form.start,
      end_date: form.end,
      initial_cash: form.cash,
      benchmark: form.benchmark,
    })
+    // Poll task status until done/failed, then fetch result
+    const r = await new Promise<PortfolioBacktestResult>((resolve, reject) => {
+      pollTimer = setInterval(async () => {
+        try {
+          const s = await getPortfolioTaskStatus(tid)
+          if (s.stage) stageHint.value = s.stage
+          if (s.status === 'done') {
+            if (pollTimer) { clearInterval(pollTimer); pollTimer = null }
+            const res = await getPortfolioResult(tid)
+            resolve(res)
+          } else if (s.status === 'failed') {
+            if (pollTimer) { clearInterval(pollTimer); pollTimer = null }
+            reject(new Error(s.error_msg || '回测失败'))
+          }
+          // pending/running: keep polling
+        } catch (pollErr) {
+          // transient network error on poll — keep polling
+        }
+      }, 2500)
+    })
    result.value = r
    equityCurve.value = r.equity_curve || []
    stocks.value = r.stocks_selected || []
@@ -173,10 +201,10 @@ async function onSubmit(): Promise<void> {
          :loading="submitting"
          @click="onSubmit"
        >
-          {{ submitting ? '回测中(VPS 执行,请等待...)' : '开始回测' }}
+          {{ submitting ? `回测中${stageHint ? '(' + stageHint + ')' : '...'}` : '开始回测' }}
        </el-button>
        <span v-if="submitting" class="muted form-hint">
-          最长 600s,期间请勿关闭页面
+          异步执行,请勿关闭页面
        </span>
      </div>
    </el-card>
@@ -1,33 +1,25 @@
-"""组合策略 API 路由(MVP)。
+"""组合策略 API 路由(异步化 P0)。

-POST /portfolio/backtest: SSH 触发 VPS 跑 BulletTrade + all_weather,
-捕获 stdout JSON 返回前端。同步模式(回测耗时,前端 loading,timeout 600s)。
+POST /portfolio/backtest: 异步提交组合回测任务,返回 task_id 供前端轮询。
+GET  /portfolio/task/{task_id}/result: 取组合回测结果。

-风格参考 routes_paper.py / routes_live.py。
+回测子进程执行(三机自适应 NAS/VPS/Mac-SSH)搬至 portfolio_worker.py,
+超时从 600s 提升至 3600s。复用 sanguo_orchestrator task 框架。
 """
 from __future__ import annotations

-import json
 import logging
-import os
-import subprocess
-import sys
-from typing import Any, Optional
+from typing import Any

 from fastapi import APIRouter, Depends, Header, HTTPException
 from pydantic import BaseModel, Field

 from .auth import verify_token as verify_token_impl
+from .routes import get_orchestrator

 logger = logging.getLogger(__name__)
 router = APIRouter()

-# VPS SSH 连接配置(~/.ssh/config 已配 49.232.102.198 免密 key)
-_VPS_HOST = "49.232.102.198"
-_VPS_WORKDIR = r"C:\\sanguo_vnpy_v2"
-_VPS_PYTHON = "python"
-_VPS_TIMEOUT = 600  # 回测耗时,给 10 分钟
-

 async def verify_token(authorization: str | None = Header(None)):
    if authorization is None or not authorization.startswith("Bearer "):
@@ -44,106 +36,51 @@ class PortfolioBacktestRequest(BaseModel):


@router.post("/portfolio/backtest", dependencies=[Depends(verify_token)])
-def run_portfolio_backtest(req: PortfolioBacktestRequest):
-    """SSH 触发 VPS 跑 BulletTrade + all_weather 回测,同步返回 JSON 结果。
+async def run_portfolio_backtest(req: PortfolioBacktestRequest):
+    """异步提交组合回测任务,返回 task_id。前端轮询 GET /task/{id} 再取结果。"""
+    tid = await get_orchestrator().submit_portfolio(
+        start=req.start_date,
+        end=req.end_date,
+        cash=req.initial_cash,
+        benchmark=req.benchmark,
+        max_pool=30,
+        provider_config=None,
+    )
+    return {"task_id": tid}

-    Windows SSH 坑:
-    - GBK 编码:python -X utf8 避免中文 print 编码崩
-    - 引号嵌套:用 list 形式 argv,避免 shell 引号 escape 噩梦
-    - 没有 tail/head/grep:用 python 后处理(本函数在 Mac 端直接解析 stdout)
-    - 中文路径:VPS_WORKDIR / userdata_mini 走 env(DEFAULT_DATA_PROVIDER=miniqmt)
-    """
-    # 本地模式(VPS 后端直接跑 runner,避免 SSH 自连外网 IP 绕路);
-    # SSH 模式(Mac 后端 → VPS)。env SANGUO_PORTFOLIO_LOCAL=1 切本地。
-    # 自动检测三机自适应:
-    # - VPS(Windows): _VPS_WORKDIR(C:\sanguo_vnpy_v2) 存在 → 本地直接跑 runner(默认 provider,行为不变)
-    # - NAS(Linux 容器): /app 存在(docker-compose 代码挂载目录) → 本地跑 runner --provider unified 读 NAS 权威数据层
-    # - Mac(dev): 都不满足 → SSH 到 VPS(原行为)
-    if os.path.isdir(_VPS_WORKDIR) or os.path.isdir("/app"):
-        local_cwd = _VPS_WORKDIR if os.path.isdir(_VPS_WORKDIR) else "/app"
-        argv = [
-            sys.executable, "-X", "utf8", "-m", "sanguo_portfolio.runner_backtest",
-            "--json", "--start", req.start_date, "--end", req.end_date,
-            "--cash", str(req.initial_cash), "--benchmark", req.benchmark, "--max-pool", "30",
-        ]
-        # NAS 容器分支: unified provider 读 NAS 权威数据层(dbbardata + parquet);
-        # VPS 本地分支保持原状(默认 provider,cwd=_VPS_WORKDIR,行为零变化)
-        if local_cwd == "/app":
-            nas_provider_config = json.dumps({
-                "db_path": "/volume1/stock/sanguo_vnpy_v2/data_backup/quant_trading.db",
-                "data_dir": "/volume1/stock/sanguo_vnpy_v2/data",
-            })
-            argv += ["--provider", "unified", "--provider-config", nas_provider_config]
-        logger.info("[portfolio] 本地跑 runner: %s", " ".join(argv[3:]))
-        try:
-            proc = subprocess.run(
-                argv, cwd=local_cwd, capture_output=True, text=True,
-                timeout=_VPS_TIMEOUT, check=False,
-            )
-        except subprocess.TimeoutExpired:
-            raise HTTPException(504, f"回测超时(>{_VPS_TIMEOUT}s)")
-        except Exception as exc:
-            logger.exception("[portfolio] 本地 runner 调用失败")
-            raise HTTPException(500, f"本地 runner 调用失败: {exc}")
-    else:
-        # Mac 端 SSH 到 VPS(远端命令手动拼,不用 shlex.quote:产 POSIX 单引号 cmd 不认)
-        remote_cmd = (
-            f"cd {_VPS_WORKDIR} && "
-            f"{_VPS_PYTHON} -X utf8 -m sanguo_portfolio.runner_backtest --json "
-            f"--start {req.start_date} --end {req.end_date} "
-            f"--cash {req.initial_cash} --benchmark {req.benchmark} "
-            f"--max-pool 30"
-        )
-        ssh_argv = [
-            "ssh", "-o", "ConnectTimeout=15", "-o", "StrictHostKeyChecking=no",
-            _VPS_HOST, remote_cmd,
-        ]
-        logger.info("[portfolio] SSH 触发: %s", remote_cmd)
-        try:
-            proc = subprocess.run(
-                ssh_argv, capture_output=True, text=True,
-                timeout=_VPS_TIMEOUT, check=False,
-            )
-        except subprocess.TimeoutExpired:
-            raise HTTPException(504, f"VPS 回测超时(>{_VPS_TIMEOUT}s)")
-        except FileNotFoundError:
-            raise HTTPException(500, "本机未找到 ssh 命令")
-        except Exception as exc:
-            logger.exception("[portfolio] SSH 调用失败")
-            raise HTTPException(500, f"SSH 调用失败: {exc}")

-    if proc.returncode != 0:
-        stderr_tail = (proc.stderr or "")[-2000:]
-        logger.error("[portfolio] VPS 回测失败 rc=%s stderr=%s", proc.returncode, stderr_tail)
-        raise HTTPException(
-            500,
-            f"VPS 回测失败(rc={proc.returncode}): {stderr_tail}",
-        )
+@router.get("/portfolio/task/{task_id}/result", dependencies=[Depends(verify_token)])
+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")

-    # 从 stdout 提取最后一行 JSON(runner --json 只 print 一行)
-    stdout = proc.stdout or ""
-    import json as _json
-    result: Optional[dict[str, Any]] = None
-    parse_err: Optional[str] = None
-    for line in reversed(stdout.strip().splitlines()):
-        line = line.strip()
-        if not line.startswith("{"):
-            continue
-        try:
-            result = _json.loads(line)
-            break
-        except _json.JSONDecodeError as exc:
-            parse_err = str(exc)
-            continue
+    stats: dict = r.statistics or {}
+    metrics = stats.get("metrics", {})
+    stocks_selected = stats.get("stocks_selected", [])
+    period = stats.get("period", {})

-    if result is None:
-        logger.error(
-            "[portfolio] stdout 无 JSON 行。parse_err=%s stdout_tail=%s",
-            parse_err, stdout[-2000:],
-        )
-        raise HTTPException(
-            500,
-            f"VPS stdout 解析失败: {parse_err or 'no JSON line'}; stdout_tail={stdout[-500:]!r}",
-        )
+    return {
+        "task_id": task_id,
+        "strategy": r.strategy,
+        "period": period,
+        "metrics": metrics,
+        "equity_curve": _df_to_records(r.equity_curve),
+        "trades": _df_to_records(r.trades),
+        "stocks_selected": stocks_selected,
+    }

-    return result
+
+def _df_to_records(df: Any) -> list[dict]:
+    """DataFrame/list → list[dict] (empty-safe)."""
+    if df is None:
+        return []
+    if hasattr(df, "empty") and df.empty:
+        return []
+    if hasattr(df, "to_dict"):
+        return df.to_dict(orient="records")
+    if isinstance(df, list):
+        return df
+    return []
@@ -0,0 +1,181 @@
+"""Portfolio backtest worker (stateless, spawn-friendly).
+
+Runs runner_backtest as a subprocess, parses stdout JSON, wraps into
+BacktestResult and persists it. Three-machine adaptive (NAS/VPS/Mac-SSH),
+logic moved here from routes_portfolio.py.
+
+Called via runner._portfolio_worker → run_portfolio_task(spec).
+"""
+from __future__ import annotations
+
+import json
+import logging
+import os
+import subprocess
+import sys
+from typing import Any, Optional
+
+logger = logging.getLogger(__name__)
+
+# Three-machine adaptive config (mirrors routes_portfolio.py original logic)
+_VPS_HOST = "49.232.102.198"
+_VPS_WORKDIR = r"C:\\sanguo_vnpy_v2"
+_VPS_PYTHON = "python"
+_TIMEOUT = 3600  # 1 hour hard cap (was 600s sync in routes_portfolio)
+
+
+def run_portfolio_task(spec: dict) -> Any:
+    """Run portfolio backtest subprocess → parse stdout JSON → BacktestResult.
+
+    spec keys:
+        task_id:          orchestrator task id (used as BacktestResult.task_id)
+        start, end:       backtest period (YYYY-MM-DD)
+        cash:             initial capital
+        benchmark:        benchmark code
+        max_pool:         max stocks in pool
+        provider_config:  optional dict {db_path, data_dir} (None for VPS/Mac)
+        db_path:          orchestrator DB path for result persistence
+        file_dir:         directory for equity/trades JSON files
+
+    Returns:
+        BacktestResult (persisted, with .id set so _on_done can complete).
+    Raises:
+        RuntimeError on subprocess failure or stdout parse failure.
+    """
+    task_id = spec["task_id"]
+    start = spec["start"]
+    end = spec["end"]
+    cash = spec["cash"]
+    benchmark = spec["benchmark"]
+    max_pool = spec["max_pool"]
+    provider_config = spec.get("provider_config")
+    db_path = spec.get("db_path", "")
+    file_dir = spec.get("file_dir")
+
+    argv, cwd = _build_argv(start, end, cash, benchmark, max_pool, provider_config)
+    logger.info("[portfolio_worker] task=%s running: %s", task_id, " ".join(argv[3:]))
+
+    proc = subprocess.run(
+        argv, cwd=cwd, capture_output=True, text=True,
+        timeout=_TIMEOUT, check=False,
+    )
+
+    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)
+        raise RuntimeError(
+            f"runner_backtest failed (rc={proc.returncode}): {stderr_tail}"
+        )
+
+    data = _parse_stdout_json(proc.stdout or "")
+
+    # Lazy imports (spawn-friendly: heavy deps loaded only in worker process)
+    import pandas as pd
+    from sanguo_backtest.result_store import BacktestResult, save_result
+
+    equity_list = data.get("equity_curve") or []
+    trades_list = data.get("trades") or []
+    period = data.get("period", {}) or {}
+
+    br = BacktestResult(
+        task_id=task_id,
+        type="portfolio",
+        status="done",
+        strategy=data.get("strategy", ""),
+        symbol=benchmark,
+        params={"benchmark": benchmark, "max_pool": max_pool},
+        start=period.get("start", ""),
+        end=period.get("end", ""),
+        statistics={
+            "metrics": data.get("metrics", {}),
+            "raw_summary": data.get("raw_summary", {}),
+            "stocks_selected": data.get("stocks_selected", []),
+            "period": period,
+        },
+        equity_curve=pd.DataFrame(equity_list) if equity_list else None,
+        trades=pd.DataFrame(trades_list) if trades_list else None,
+    )
+
+    # Persist so get_result can reload via load_result(result.id).
+    # save_result sets br.id, which _on_done picks up as task.result_id.
+    if db_path:
+        save_result(br, db_path=db_path, file_dir=file_dir)
+
+    return br
+
+
+def _build_argv(
+    start: str, end: str, cash: float, benchmark: str,
+    max_pool: int, provider_config: Optional[dict],
+) -> tuple[list[str], Optional[str]]:
+    """Three-machine adaptive argv construction.
+
+    Returns (argv, cwd):
+        NAS (Linux container, /app exists): local runner + unified provider
+        VPS (Windows, C:\\sanguo_vnpy_v2 exists): local runner, default provider
+        Mac (neither): SSH to VPS
+    """
+    if os.path.isdir(_VPS_WORKDIR) or os.path.isdir("/app"):
+        cwd = _VPS_WORKDIR if os.path.isdir(_VPS_WORKDIR) else "/app"
+        argv = [
+            sys.executable, "-X", "utf8", "-m", "sanguo_portfolio.runner_backtest",
+            "--json", "--start", start, "--end", end,
+            "--cash", str(cash), "--benchmark", benchmark,
+            "--max-pool", str(max_pool),
+        ]
+        # NAS container: unified provider reads NAS authoritative data layer
+        # (dbbardata + parquet). VPS keeps default provider (cwd=_VPS_WORKDIR).
+        if cwd == "/app":
+            nas_provider_config = json.dumps({
+                "db_path": "/volume1/stock/sanguo_vnpy_v2/data_backup/quant_trading.db",
+                "data_dir": "/volume1/stock/sanguo_vnpy_v2/data",
+            })
+            argv += ["--provider", "unified", "--provider-config", nas_provider_config]
+        elif provider_config:
+            argv += ["--provider", "unified",
+                     "--provider-config", json.dumps(provider_config)]
+        return argv, cwd
+
+    # Mac: SSH to VPS (remote cmd hand-built, no shlex.quote: POSIX single
+    # quotes not understood by Windows cmd)
+    remote_cmd = (
+        f"cd {_VPS_WORKDIR} && "
+        f"{_VPS_PYTHON} -X utf8 -m sanguo_portfolio.runner_backtest --json "
+        f"--start {start} --end {end} "
+        f"--cash {cash} --benchmark {benchmark} "
+        f"--max-pool {max_pool}"
+    )
+    ssh_argv = [
+        "ssh", "-o", "ConnectTimeout=15", "-o", "StrictHostKeyChecking=no",
+        _VPS_HOST, remote_cmd,
+    ]
+    return ssh_argv, None
+
+
+def _parse_stdout_json(stdout: str) -> dict:
+    """Extract the last JSON line (starting with '{') from subprocess stdout.
+
+    runner_backtest --json prints exactly one JSON line as the last output.
+    Reversed scan handles trailing log noise.
+    """
+    parse_err: Optional[str] = None
+    for line in reversed(stdout.strip().splitlines()):
+        line = line.strip()
+        if not line.startswith("{"):
+            continue
+        try:
+            return json.loads(line)
+        except json.JSONDecodeError as exc:
+            if parse_err is None:
+                parse_err = str(exc)
+            continue
+
+    logger.error(
+        "[portfolio_worker] stdout has no JSON line. parse_err=%s stdout_tail=%s",
+        parse_err, stdout[-2000:],
+    )
+    raise RuntimeError(
+        f"stdout parse failed: {parse_err or 'no JSON line'}; "
+        f"stdout_tail={stdout[-500:]!r}"
+    )
@@ -3,6 +3,7 @@ Orchestrator for task coordination and execution
 Manages backtesting tasks with lazy imports
 """
 import asyncio
+import os
 import uuid
 from concurrent.futures import Future
 from .pool import TaskPool
@@ -120,6 +121,40 @@ class Orchestrator:
        asyncio.ensure_future(self._wait_future(task_id, fut))
        return task_id

+    async def submit_portfolio(self, start: str, end: str, cash: float,
+                               benchmark: str, max_pool: int = 30,
+                               provider_config=None) -> str:
+        """Submit a portfolio backtest task asynchronously.
+
+        Runs runner_backtest as a subprocess (3600s hard cap) inside the
+        process pool, reusing the same _wait_future/_on_done bridge as
+        CTA/optimize/factor.
+        """
+        task_id = f"portfolio_{uuid.uuid4().hex[:8]}"
+        self.pool.submit(task_id, "portfolio")
+        file_dir = os.path.dirname(os.path.abspath(self.db_path)) if self.db_path else None
+        self._pending[task_id] = dict(
+            task_id=task_id,
+            start=start,
+            end=end,
+            cash=cash,
+            benchmark=benchmark,
+            max_pool=max_pool,
+            provider_config=provider_config,
+            db_path=self.db_path,
+            file_dir=file_dir,
+        )
+        await self._notify_stage(task_id, "排队中")
+
+        spec = self._pending[task_id]
+        fut: Future = self.pool.submit_work(task_id, _portfolio_worker, spec)
+
+        task = self.pool.get_task(task_id)
+        task.start()
+        await self._notify_stage(task_id, "回测中")
+        asyncio.ensure_future(self._wait_future(task_id, fut))
+        return task_id
+
    async def _wait_future(self, task_id: str, fut: Future) -> None:
        """Wait for Future to complete and handle result/exception

@@ -212,3 +247,9 @@ def _factor_worker(symbols: list, factor_names: list, start: str, end: str, cfg,
    """Worker for factor analysis (lazy import, spawn-friendly)"""
    from sanguo_factor.analyzer import run_factor_analysis
    return run_factor_analysis(symbols, factor_names, start, end, cfg, output_dir)
+
+
+def _portfolio_worker(spec: dict) -> any:
+    """Worker for portfolio backtest (lazy import, spawn-friendly)"""
+    from sanguo_orchestrator.portfolio_worker import run_portfolio_task
+    return run_portfolio_task(spec)