Files

T

claude_dev 3369ea2433 docs(phase3a): Web API 完整化 design（异步 pool + WS 阶段 + JWT 单用户 + tears 完整化）

2026-07-06 17:22:45 +08:00

5.7 KiB

Raw Blame History

Phase 3a Web API 完整化 Design

对应 PRD：docs/superpowers/specs/2026-07-05-vnpy-quant-platform-prd.md 承接：Phase 2 因子/回测层（已 DONE）范围：回测异步 + WS 阶段进度 + JWT 单用户 + alpha tears 完整化 + optimize/factor 路由补完日期：2026-07-06 状态：Draft（待 review）

0. 范围决策（brainstorming 已确认）

需求（业务语言）	决策
回测改"后台跑"	异步：提交即返回 task_id，后台排队跑，跑完存库凭号查
回测跑时看什么	阶段进度（数据加载→算因子→回测→完成），不做百分比/实时日志
登录	单用户起步（一个账号密码）
因子分析报告	完整 tears（IC/IR + 分层收益图 + 打开即看的 html）

1. 架构

客户端 → FastAPI (JWT 校验) → Orchestrator → ProcessPoolExecutor → run_cta_backtest / run_factor_analysis
                                          ↘ 阶段回调 → WS 广播给订阅者
结果 → result_store → GET /task/{id}/result

核心：Phase 2 同步 _run_sync → 异步 submit_*（ProcessPoolExecutor + asyncio bridge）；任务状态/阶段变更经 WS 推订阅者；路由加 JWT 依赖。

2. 组件改动

sanguo_orchestrator（异步化）

pool.py：TaskPool 加 ProcessPoolExecutor(max_workers, mp_context=spawn) + 阶段追踪字段（stage: str）
runner.py：async submit_cta/submit_optimize/submit_factor → pool.submit + asyncio.wrap_future；任务包装器在关键节点回调 on_stage(task_id, stage) 推 WS

sanguo_api（新增 auth/ws + 路由补完）

auth.py（新）：JWT 单用户——create_token(username) / verify_token(token) 依赖；用户名/密码 hash/jwt_secret 配 config/backtest.yaml
ws.py（新）：WS 连接池（dict[task_id, set[WebSocket]]）+ broadcast(task_id, msg)；YAGNI 不做重连/心跳
routes.py：+ POST /auth/login、WS /ws/task/{id}；optimize/factor 路由真调用 submit_optimize/submit_factor；其他业务路由加 Depends(verify_token)

sanguo_factor（tears 完整化）

alpha_lab.py：补 compute_factors(symbols, factor_names, start, end, cfg) —— AlphaSession 加载 + add_feature + prepare_data + fetch_raw → 返回因子值 DataFrame
analyzer.py：tears pipeline 完整——compute_factors → get_clean_factor_and_forward_returns → create_full_tear_sheet → 输出 html 报告 + IC/IR 数值入 FactorReport

config/backtest.yaml

auth: {username, password_hash, jwt_secret, token_expire_minutes}
pool: {max_workers: 2}（NAS braswell 2 核）

3. 数据流

回测异步：

客户端 POST /backtest/cta (JWT) → orchestrator.submit_cta → pool.submit(run_cta_backtest)
  → task=pending → asyncio.wrap_future → task=running (WS 推)
  → run_cta_backtest 内阶段回调 (WS 推: 加载数据/回测中)
  → done/failed (WS 推) → 结果落 result_store
客户端 GET /task/{id}/result (JWT) → statistics + equity

因子分析：

POST /factor/analyze → submit_factor → pool 跑 run_factor_analysis
  → compute_factors(多 symbol) → alphalens tears → html 报告
  → 报告路径入 result_store → GET /task/{id}/result 返回 report_path

4. API（完整）

路由	鉴权	功能
`POST /api/v1/auth/login`	无	用户名密码 → `{token}`
`POST /api/v1/backtest/cta`	JWT	异步提交 CTA 回测 → `{task_id}`
`POST /api/v1/backtest/optimize`	JWT	异步提交参数优化 → `{task_id}`
`POST /api/v1/factor/analyze`	JWT	异步提交因子分析 → `{task_id}`
`GET /api/v1/task/{id}`	JWT	查任务状态 + 阶段
`GET /api/v1/task/{id}/result`	JWT	查结果（统计/报告路径）
`WS /ws/task/{id}`	JWT（query param）	推 `{status, stage}` 变更

5. 测试

类型	范围
异步 pool	mock `ProcessPoolExecutor`，验证 submit + future + 状态转换 + 阶段回调
JWT	签发/校验/401（TestClient + DependencyOverride）
WS	TestClient websocket 连接收消息
tears	mock alphalens（本地）+ 容器真实 tears 端到端
路由	optimize/factor 真调用 Orchestrator

6. 风险

风险	缓解
ProcessPoolExecutor + uvicorn fork 坑	`mp_context=spawn`（vnpy/alpha 已用 spawn，兼容）
WS 连接管理	简单 dict 连接池（YAGNI，不做重连/心跳）
JWT secret 泄露	config 配置（生产换环境变量）
tears 依赖容器（alphalens）	本地 mock 测试 + 容器端到端（同 Phase 2 策略）

7. 测试策略（同 Phase 2）

本地 Python 3.14 mock 测试（factor/api/orchestrator）+ 容器 Python 3.10 真实依赖测试（polars/alphalens/vnpy）+ 端到端冒烟。

8. 不做（YAGNI / 留后续 phase）

Vue 前端（Phase 3b）
多用户/角色权限（Phase 4+）
多因子组合回测（选股→加权→调仓，Phase 4+）
国金模拟（独立 spec，Phase 5+）
WS 重连/心跳/断线恢复（简单连接池够 Phase 3a 用）
回测百分比进度（vnpy 引擎不报告总进度，强行做易踩坑）

9. Phase 2 衔接

复用 result_store（存统计 + html 报告路径）
复用 cta_engine/cta_optimizer/registry/library
runner._run_sync → 改 async submit_*（保留 _run_sync 逻辑作为 pool 内执行的 worker 函数）
analyzer.run_factor_analysis 骨架 → 补 tears pipeline

变更记录

日期	变更	来源
2026-07-06	初稿，基于 brainstorming 4 轮业务对话确认	superpowers brainstorming

5.7 KiB Raw Blame History Unescape Escape