From a480fd389f051717f635a4d1cae51ada52a0cda9 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Mon, 6 Jul 2026 07:25:53 +0800 Subject: [PATCH] =?UTF-8?q?docs(phase2):=20=E5=9B=A0=E5=AD=90/=E5=9B=9E?= =?UTF-8?q?=E6=B5=8B=E5=B1=82=E5=AE=9E=E7=8E=B0=E8=AE=A1=E5=88=92=EF=BC=88?= =?UTF-8?q?13=20task=EF=BC=9A3=20spike=20+=2010=20=E5=AE=9E=E7=8E=B0?= =?UTF-8?q?=EF=BC=8Cspike=20=E5=85=88=E8=A1=8C=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../2026-07-05-phase2-factor-backtest.md | 783 ++++++++++++++++++ 1 file changed, 783 insertions(+) create mode 100644 docs/superpowers/plans/2026-07-05-phase2-factor-backtest.md diff --git a/docs/superpowers/plans/2026-07-05-phase2-factor-backtest.md b/docs/superpowers/plans/2026-07-05-phase2-factor-backtest.md new file mode 100644 index 0000000..b124256 --- /dev/null +++ b/docs/superpowers/plans/2026-07-05-phase2-factor-backtest.md @@ -0,0 +1,783 @@ +# Phase 2 因子/回测层 Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 构建因子层（vnpy.alpha 封装）+ CTA 回测层 + 任务编排 + 轻量 FastAPI，形成因子研究与策略回测的完整闭环。 + +**Architecture:** 分层解耦（sanguo_factor / sanguo_backtest / sanguo_orchestrator / sanguo_api），vnpy 引擎零改造，适配层只包装。因子层封装 vnpy.alpha 的 AlphaLab/AlphaDataset（polars + 表达式因子 + 内置 alphalens）；CTA 回测层包装 BacktestingEngine + run_optimization；编排层 multiprocessing pool；API 层 5 路由 FastAPI。 + +**Tech Stack:** vnpy 4.4.0（源码 vnpy_v4.4.0/，sys.path.insert）、vnpy.alpha（polars + alphalens 内置）、vnpy_ctabacktester（run_optimization）、FastAPI、pytest、Python 3.14。 + +## Global Constraints + +- **vnpy 零改造**：vnpy 源码在 `vnpy_v4.4.0/`，通过 `sys.path.insert(0, _VNPY_SRC)` 引用（参照 sanguo_data/datareader.py:6-9） +- **Python 3.14**（Phase 1 已验证 + vnpy 4.4.0 不拉 Qt） +- **vnpy.alpha 真实 API**（源码探测）： + - `AlphaLab`（lab.py）：polars，数据存 `lab_path/daily/{vt_symbol}.parquet` + - `AlphaDataset`（dataset/template.py）：表达式因子，`add_feature(name, expression)`、`feature_expressions: dict[str, str | pl.Expr]` + - 因子范式：`cs_rank(feature: DataProxy) -> DataProxy`（横截面）、ts_function、ta_function + - 内置 alphalens：`from alphalens.tears import create_full_tear_sheet`（template.py:14） +- **数据转换必需**：Phase 1（SQLite DbBarData + parquet 年分区）→ AlphaLab（polars parquet by vt_symbol），转换在 `sanguo_factor/data_adapter.py` +- **run_optimization**：在 `vnpy_ctabacktester` 包（非 vnpy 核心），需 `pip install`，S2 spike 验证 +- **Phase 1 接口复用**：`read_db_daily(symbol, start, end, cfg) → list[BarData]` +- **因子层范围**：只到 Alphalens 分层分析（IC/IR/分层收益），不含组合回测 +- **测试覆盖率 ≥ 80%**，TDD，每 task commit +- **不引 Qlib、不用 Celery** + +--- + +## File Structure + +``` +sanguo_factor/{__init__,data_adapter,alpha_lab,registry,library,analyzer}.py +sanguo_backtest/{__init__,result_store,cta_engine,cta_optimizer}.py +sanguo_orchestrator/{__init__,task,pool,runner}.py +sanguo_api/{__init__,schemas,app,routes}.py +tests/{factor,backtest,orchestrator,api,spikes}/ +config/backtest.yaml +scripts/smoke_phase2.py +``` + +--- + +## Task 1: S3 Spike — peewee 版本冲突 + +**Files:** `tests/spikes/test_s3_peewee.py`, `scripts/spike_s3_peewee.py` +**验证**：alphalens（vnpy.alpha 内置 import）与 vnpy_sqlite（peewee 4.1.1）能否共存。 + +- [ ] **Step 1: 写 spike 脚本** + +```python +# scripts/spike_s3_peewee.py +"""S3: 验证 alphalens 与 vnpy_sqlite(peewee) 能否共存.""" +import sys, os +_VNPY_SRC = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "vnpy_v4.4.0")) +sys.path.insert(0, _VNPY_SRC) + +def main(): + results = {} + try: + from alphalens.utils import get_clean_factor_and_forward_returns + from alphalens.tears import create_full_tear_sheet + results["alphalens_import"] = "OK" + except Exception as e: + results["alphalens_import"] = f"FAIL: {type(e).__name__}: {e}" + try: + import peewee + results["peewee_version"] = peewee.__version__ + from vnpy_sqlite.sqlite_database import SqliteDatabase + results["vnpy_sqlite_import"] = "OK" + except Exception as e: + results["vnpy_sqlite_import"] = f"FAIL: {type(e).__name__}: {e}" + try: + from vnpy.alpha.dataset import AlphaDataset + results["alpha_dataset_import"] = "OK" + except Exception as e: + results["alpha_dataset_import"] = f"FAIL: {type(e).__name__}: {e}" + print("=== S3 Spike Result ===") + for k, v in results.items(): print(f" {k}: {v}") + ok = all(v == "OK" or v.startswith("4") for v in results.values()) + print(f"\nS3 VERDICT: {'PASS' if ok else 'FAIL/CONFLICT'}") + +if __name__ == "__main__": + main() +``` + +- [ ] **Step 2: 运行** — `python scripts/spike_s3_peewee.py` +- [ ] **Step 3: 测试占位** + +```python +# tests/spikes/test_s3_peewee.py +import pytest +def test_s3_verdict_recorded(): + S3_VERDICT = "PENDING_RUN_SPIKE" # 改 PASS/FAIL 后提交 + if S3_VERDICT == "PENDING_RUN_SPIKE": + pytest.skip("先运行 scripts/spike_s3_peewee.py，再填结论") + assert S3_VERDICT in ("PASS", "FAIL") +``` + +- [ ] **Step 4: 运行 spike + 记录结论到 progress.md** +- [ ] **Step 5: Commit** — `git add tests/spikes/test_s3_peewee.py scripts/spike_s3_peewee.py && git commit -m "spike(s3): peewee 版本冲突验证"` + +--- + +## Task 2: S1 Spike — vnpy.alpha A 股支撑度 + +**Files:** `tests/spikes/test_s1_vnpy_alpha_ashare.py`, `scripts/spike_s1_alpha_ashare.py` +**验证**：AlphaLab/AlphaDataset 能否处理 A 股（Phase 1 真实数据 → AlphaLab 格式 → 表达式因子）。 + +- [ ] **Step 1: 写 spike 脚本** + +```python +# scripts/spike_s1_alpha_ashare.py +"""S1: 验证 vnpy.alpha 能否处理 A 股数据。""" +import sys, os +_VNPY_SRC = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "vnpy_v4.4.0")) +sys.path.insert(0, _VNPY_SRC) +sys.path.insert(0, "/app") + +def main(): + results = {} + try: + from vnpy.alpha.lab import AlphaLab + AlphaLab("/tmp/alpha_lab_spike") + results["alphalab_init"] = "OK" + except Exception as e: + results["alphalab_init"] = f"FAIL: {e}"; print(results); return + try: + from sanguo_data.datareader import read_db_daily + from sanguo_data.config import load_config + cfg = load_config("/app/config/data_platform.yaml") + bars = read_db_daily("600000", "2024-01-01", "2024-06-30", cfg) + from vnpy.alpha.lab import AlphaLab + lab = AlphaLab("/tmp/alpha_lab_spike") + lab.save_bar_data(bars) + results["data_convert"] = f"OK ({len(bars)} bars)" + except Exception as e: + import traceback; traceback.print_exc() + results["data_convert"] = f"FAIL: {e}"; print(results); return + try: + from vnpy.alpha.dataset import AlphaDataset, Segment + results["alpha_dataset"] = "OK (需据源码细化 add_feature)" + except Exception as e: + results["alpha_dataset"] = f"FAIL: {e}" + print("=== S1 Spike Result ===") + for k, v in results.items(): print(f" {k}: {v}") + ok = all("OK" in v for v in results.values()) + print(f"\nS1 VERDICT: {'PASS' if ok else 'PARTIAL/FAIL'}") + +if __name__ == "__main__": + main() +``` + +- [ ] **Step 2: 容器运行** — `docker run --rm -v /volume1/stock:/app/data sanguo_vnpy_v2:with-sqlite python /app/scripts/spike_s1_alpha_ashare.py` +- [ ] **Step 3: 测试占位**（同 S3 模式，findings dict 记录 A 股特性支撑度） +- [ ] **Step 4: 运行 + 记录**（PARTIAL/FAIL → fallback 自建因子层） +- [ ] **Step 5: Commit** — `git commit -m "spike(s1): vnpy.alpha A 股支撑度验证"` + +--- + +## Task 3: S2 Spike — run_optimization 多进程 + +**Files:** `tests/spikes/test_s2_run_optimization.py`, `scripts/spike_s2_optimization.py` +**验证**：vnpy_ctabacktester run_optimization 容器内 multiprocessing 行为 + max_workers。 + +- [ ] **Step 1: 装 vnpy_ctabacktester + 写 spike**（探测 run_optimization 签名 + 小规模跑） +- [ ] **Step 2: 容器 `pip install vnpy_ctabacktester` + 运行**（NAS 弱 CPU 慢 → docker commit `:with-backtester`） +- [ ] **Step 3: 测试占位** +- [ ] **Step 4: 记录 max_workers 推荐值** +- [ ] **Step 5: Commit** — `git commit -m "spike(s2): run_optimization 多进程容器验证"` + +> spike 脚本骨架参照 S3（import + inspect.signature + 小规模调用），执行 subagent 据 run_optimization 实际签名填充。 + +--- + +## Task 4: sanguo_factor/data_adapter.py + +**Files:** `sanguo_factor/__init__.py`, `sanguo_factor/data_adapter.py`, `tests/factor/conftest.py`, `tests/factor/test_data_adapter.py` +**Interfaces:** Consumes `read_db_daily`; Produces `convert_bars_to_alpha_df(bars) → pl.DataFrame`、`save_alpha_lab_data(bars, lab_path)` + +- [ ] **Step 1: 写失败测试** + +```python +# tests/factor/test_data_adapter.py +import sys, os +_VNPY_SRC = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "vnpy_v4.4.0")) +sys.path.insert(0, _VNPY_SRC) +import polars as pl +from datetime import datetime +from vnpy.trader.object import BarData +from vnpy.trader.constant import Exchange, Interval + +def _make_bar(symbol, dt, close): + return BarData(symbol=symbol, exchange=Exchange.SSE, datetime=dt, + interval=Interval.DAILY, open_price=close, high_price=close, + low_price=close, close_price=close, volume=1000, gateway_name="TEST") + +def test_convert_bars_to_alpha_df_columns(): + bars = [_make_bar("600000", datetime(2024,1,i), 10.0+i) for i in range(1, 6)] + from sanguo_factor.data_adapter import convert_bars_to_alpha_df + df = convert_bars_to_alpha_df(bars) + assert isinstance(df, pl.DataFrame) + for col in ["vt_symbol","datetime","open_price","high_price","low_price", + "close_price","volume","turnover","open_interest"]: + assert col in df.columns + assert df.height == 5 + +def test_convert_empty_bars(): + from sanguo_factor.data_adapter import convert_bars_to_alpha_df + assert convert_bars_to_alpha_df([]).height == 0 +``` + +- [ ] **Step 2: 验证失败** +- [ ] **Step 3: 实现** + +```python +# sanguo_factor/data_adapter.py +"""Phase 1 BarData → vnpy.alpha AlphaLab polars 格式转换。""" +import sys, os +_VNPY_SRC = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "vnpy_v4.4.0")) +if _VNPY_SRC not in sys.path: + sys.path.insert(0, _VNPY_SRC) +import polars as pl +from pathlib import Path +from vnpy.trader.object import BarData + + +def convert_bars_to_alpha_df(bars: list[BarData]) -> pl.DataFrame: + if not bars: + return pl.DataFrame(schema={ + "vt_symbol": pl.Utf8, "datetime": pl.Datetime, + "open_price": pl.Float64, "high_price": pl.Float64, "low_price": pl.Float64, + "close_price": pl.Float64, "volume": pl.Float64, + "turnover": pl.Float64, "open_interest": pl.Float64, + }) + return pl.DataFrame({ + "vt_symbol": [b.vt_symbol for b in bars], + "datetime": [b.datetime for b in bars], + "open_price": [b.open_price for b in bars], + "high_price": [b.high_price for b in bars], + "low_price": [b.low_price for b in bars], + "close_price": [b.close_price for b in bars], + "volume": [b.volume for b in bars], + "turnover": [b.turnover for b in bars], + "open_interest": [b.open_interest for b in bars], + }) + + +def save_alpha_lab_data(bars: list[BarData], lab_path: str) -> Path: + from vnpy.alpha.lab import AlphaLab + lab = AlphaLab(lab_path) + lab.save_bar_data(bars) + return lab.daily_path +``` + +- [ ] **Step 4: 验证通过** — 2 passed +- [ ] **Step 5: Commit** — `git commit -m "feat(factor): data_adapter BarData→AlphaLab polars 转换"` + +--- + +## Task 5: sanguo_factor/registry.py + library.py + +**Files:** `sanguo_factor/registry.py`, `sanguo_factor/library.py`, `tests/factor/test_registry.py`, `tests/factor/test_library.py` +**Interfaces:** `register_factor(name, expression, category)`、`get_factor(name)`、`list_factors(category=None)` + +- [ ] **Step 1: 写测试** + +```python +# tests/factor/test_registry.py +def test_register_and_get_factor(): + from sanguo_factor.registry import register_factor, get_factor, _REGISTRY + _REGISTRY.clear() + register_factor("ma5", "pl.col('close_price').rolling_mean(5)", category="trend") + f = get_factor("ma5") + assert f["name"] == "ma5" and f["category"] == "trend" + +def test_list_factors_by_category(): + from sanguo_factor.registry import register_factor, list_factors, _REGISTRY + _REGISTRY.clear() + register_factor("ma5", "e1", category="trend") + register_factor("rsi", "e2", category="momentum") + assert len(list_factors(category="trend")) == 1 + assert len(list_factors()) == 2 + +def test_register_duplicate_raises(): + import pytest + from sanguo_factor.registry import register_factor, _REGISTRY + _REGISTRY.clear() + register_factor("ma5", "e", category="trend") + with pytest.raises(ValueError): + register_factor("ma5", "other", category="trend") +``` + +```python +# tests/factor/test_library.py +def test_library_registers_on_import(): + from sanguo_factor.registry import list_factors + import sanguo_factor.library + assert len(list_factors(category="builtin")) > 0 +``` + +- [ ] **Step 2: 验证失败** +- [ ] **Step 3: 实现** + +```python +# sanguo_factor/registry.py +"""因子表达式注册（包装 vnpy.alpha AlphaDataset.add_feature 范式）。""" +from typing import Any +_REGISTRY: dict[str, dict[str, Any]] = {} + +def register_factor(name: str, expression: str, category: str = "custom") -> None: + if name in _REGISTRY: + raise ValueError(f"因子已存在: {name}") + _REGISTRY[name] = {"name": name, "expression": expression, "category": category} + +def get_factor(name: str) -> dict[str, Any] | None: + return _REGISTRY.get(name) + +def list_factors(category: str | None = None) -> list[dict[str, Any]]: + if category is None: + return list(_REGISTRY.values()) + return [f for f in _REGISTRY.values() if f["category"] == category] +``` + +```python +# sanguo_factor/library.py +"""内置因子表达式库（Alpha158/101 子集，polars 表达式）。""" +from .registry import register_factor + +BUILTIN_FACTORS = [ + {"name": "ma5", "expression": "pl.col('close_price').rolling_mean(5)", "category": "builtin"}, + {"name": "ma10", "expression": "pl.col('close_price').rolling_mean(10)", "category": "builtin"}, + {"name": "ma20", "expression": "pl.col('close_price').rolling_mean(20)", "category": "builtin"}, + {"name": "vol_ma5", "expression": "pl.col('volume').rolling_mean(5)", "category": "builtin"}, + {"name": "return_1d", "expression": "pl.col('close_price').pct_change(1)", "category": "builtin"}, + {"name": "return_5d", "expression": "pl.col('close_price').pct_change(5)", "category": "builtin"}, + {"name": "high_low_5", "expression": "(pl.col('high_price').rolling_max(5) - pl.col('low_price').rolling_min(5)) / pl.col('close_price')", "category": "builtin"}, +] + +def _register_all(): + from .registry import _REGISTRY + for f in BUILTIN_FACTORS: + if f["name"] not in _REGISTRY: + register_factor(f["name"], f["expression"], f["category"]) + +_register_all() +``` + +- [ ] **Step 4: 验证通过** +- [ ] **Step 5: Commit** — `git commit -m "feat(factor): 因子表达式注册 + Alpha158 子集内置库"` + +--- + +## Task 6: sanguo_factor/alpha_lab.py + analyzer.py + +**Files:** `sanguo_factor/alpha_lab.py`, `sanguo_factor/analyzer.py`, `tests/factor/test_alpha_lab.py`, `tests/factor/test_analyzer.py` +**Interfaces:** `AlphaLabSession`、`run_factor_analysis(symbols, factor_names, start, end, cfg, output_dir) → FactorReport` +**依赖 S1 spike**：若 vnpy.alpha 不支持 A 股 → fallback 自建。 + +- [ ] **Step 1: 写测试（mock AlphaLab + alphalens）** + +```python +# tests/factor/test_alpha_lab.py +from unittest.mock import Mock, patch +def test_alpha_lab_session_init(tmp_path): + from sanguo_factor.alpha_lab import AlphaLabSession + with patch("sanguo_factor.alpha_lab.AlphaLab") as MockLab: + s = AlphaLabSession(lab_path=str(tmp_path)) + assert s.lab_path == str(tmp_path) + MockLab.assert_called_once() + +def test_load_symbols_calls_read_db_daily(tmp_path): + from sanguo_factor.alpha_lab import AlphaLabSession + with patch("sanguo_factor.alpha_lab.AlphaLab"), \ + patch("sanguo_factor.alpha_lab.read_db_daily") as mock_read: + mock_read.return_value = [] + s = AlphaLabSession(lab_path=str(tmp_path)) + s.load_symbols(["600000"], "2024-01-01", "2024-06-30", cfg=Mock()) + mock_read.assert_called() +``` + +```python +# tests/factor/test_analyzer.py +from unittest.mock import Mock, patch, MagicMock +def test_run_factor_analysis_returns_report(tmp_path): + from sanguo_factor.analyzer import run_factor_analysis + with patch("sanguo_factor.analyzer.create_full_tear_sheet"), \ + patch("sanguo_factor.analyzer.get_clean_factor_and_forward_returns") as mc: + mc.return_value = MagicMock() + r = run_factor_analysis(symbols=["600000"], factor_names=["ma5"], + start="2024-01-01", end="2024-06-30", cfg=Mock(), output_dir=str(tmp_path)) + assert r.factor_names == ["ma5"] +``` + +- [ ] **Step 2: 验证失败** +- [ ] **Step 3: 实现** + +```python +# sanguo_factor/alpha_lab.py +import sys, os +_VNPY_SRC = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "vnpy_v4.4.0")) +if _VNPY_SRC not in sys.path: + sys.path.insert(0, _VNPY_SRC) +from vnpy.alpha.lab import AlphaLab +from sanguo_data.datareader import read_db_daily +from .data_adapter import save_alpha_lab_data + + +class AlphaLabSession: + def __init__(self, lab_path: str): + self.lab_path = lab_path + self.lab = AlphaLab(lab_path) + + def load_symbols(self, symbols: list[str], start: str, end: str, cfg) -> None: + for symbol in symbols: + bars = read_db_daily(symbol, start, end, cfg) + if bars: + save_alpha_lab_data(bars, self.lab_path) +``` + +```python +# sanguo_factor/analyzer.py +import sys, os +_VNPY_SRC = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "vnpy_v4.4.0")) +if _VNPY_SRC not in sys.path: + sys.path.insert(0, _VNPY_SRC) +from dataclasses import dataclass, field +from alphalens.utils import get_clean_factor_and_forward_returns +from alphalens.tears import create_full_tear_sheet +from .alpha_lab import AlphaLabSession +from .registry import get_factor + + +@dataclass +class FactorReport: + factor_names: list[str] + output_dir: str + ic_summary: dict = field(default_factory=dict) + + +def run_factor_analysis(symbols, factor_names, start, end, cfg, output_dir) -> FactorReport: + session = AlphaLabSession(lab_path=output_dir) + session.load_symbols(symbols, start, end, cfg) + # 执行时据 AlphaDataset.add_feature（template.py:60）+ S1 spike 细化： + # for name in factor_names: dataset.add_feature(name, get_factor(name)["expression"]) + # → get_clean_factor_and_forward_returns → create_full_tear_sheet + return FactorReport(factor_names=factor_names, output_dir=output_dir) +``` + +- [ ] **Step 4: 验证通过** +- [ ] **Step 5: Commit** — `git commit -m "feat(factor): AlphaLab 封装 + Alphalens 分析器"` + +--- + +## Task 7: sanguo_backtest/result_store.py + +**Files:** `sanguo_backtest/__init__.py`, `sanguo_backtest/result_store.py`, `tests/backtest/conftest.py`, `tests/backtest/test_result_store.py` +**Interfaces:** `BacktestResult` dataclass、`save_result(result, db_path, file_dir) → int`、`load_result(id, db_path)`、`list_results(type_filter, db_path)` + +- [ ] **Step 1: 写测试**（save/load、list filter、failed error_msg 三例，完整代码见 design §2.2） +- [ ] **Step 2: 验证失败** +- [ ] **Step 3: 实现**（SQLite `backtest_stats` 表 + parquet 净值/成交，schema 含 `id, task_id, type, status, strategy, symbol, params(json), start, end, statistics(json), equity_path, trades_path, error_msg, created_at`） + +```python +# sanguo_backtest/result_store.py +import sqlite3, json +from dataclasses import dataclass +from pathlib import Path +import pandas as pd + + +@dataclass +class BacktestResult: + task_id: str; type: str; status: str; strategy: str; symbol: str + params: dict; start: str; end: str; statistics: dict + equity_curve: "pd.DataFrame | None" = None + trades: "pd.DataFrame | None" = None + error_msg: "str | None" = None + + +_SCHEMA = """CREATE TABLE IF NOT EXISTS backtest_stats ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id TEXT, type TEXT, status TEXT, strategy TEXT, symbol TEXT, + params TEXT, start TEXT, end TEXT, statistics TEXT, + equity_path TEXT, trades_path TEXT, error_msg TEXT, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP);""" + + +def _connect(db_path): + conn = sqlite3.connect(db_path); conn.executescript(_SCHEMA); return conn + + +def save_result(result, db_path, file_dir=None): + conn = _connect(db_path) + try: + equity_path = trades_path = None + if file_dir: + fdir = Path(file_dir); fdir.mkdir(parents=True, exist_ok=True) + if result.equity_curve is not None and not result.equity_curve.empty: + equity_path = str(fdir / f"{result.task_id}_equity.parquet") + result.equity_curve.to_parquet(equity_path) + if result.trades is not None and not result.trades.empty: + trades_path = str(fdir / f"{result.task_id}_trades.parquet") + result.trades.to_parquet(trades_path) + cur = conn.execute( + "INSERT INTO backtest_stats (task_id,type,status,strategy,symbol,params,start,end,statistics,equity_path,trades_path,error_msg) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", + (result.task_id, result.type, result.status, result.strategy, result.symbol, + json.dumps(result.params), result.start, result.end, + json.dumps(result.statistics), equity_path, trades_path, result.error_msg)) + conn.commit(); return cur.lastrowid + finally: + conn.close() + + +def load_result(rid, db_path): + conn = _connect(db_path) + try: + row = conn.execute("SELECT * FROM backtest_stats WHERE id=?", (rid,)).fetchone() + if not row: raise KeyError(f"result {rid} not found") + cols = [d[0] for d in conn.execute("SELECT * FROM backtest_stats LIMIT 0").description] + d = dict(zip(cols, row)) + equity = pd.read_parquet(d["equity_path"]) if d.get("equity_path") else None + trades = pd.read_parquet(d["trades_path"]) if d.get("trades_path") else None + return BacktestResult( + task_id=d["task_id"], type=d["type"], status=d["status"], + strategy=d["strategy"], symbol=d["symbol"], + params=json.loads(d["params"]), start=d["start"], end=d["end"], + statistics=json.loads(d["statistics"]) if d["statistics"] else {}, + equity_curve=equity, trades=trades, error_msg=d.get("error_msg")) + finally: + conn.close() + + +def list_results(type_filter=None, db_path=""): + conn = _connect(db_path) + try: + q, args = "SELECT id FROM backtest_stats", () + if type_filter: q, args = q + " WHERE type=?", (type_filter,) + return [load_result(r[0], db_path) for r in conn.execute(q, args).fetchall()] + finally: + conn.close() +``` + +- [ ] **Step 4: 验证通过** — 3 passed +- [ ] **Step 5: Commit** — `git commit -m "feat(backtest): result_store SQLite+parquet 结果存储"` + +--- + +## Task 8: sanguo_backtest/cta_engine.py + +**Files:** `sanguo_backtest/cta_engine.py`, `tests/backtest/test_cta_engine.py` +**Interfaces:** `run_cta_backtest(strategy_class, symbol, params, start, end, cfg, db_path) → BacktestResult` + +- [ ] **Step 1: 写测试**（mock BacktestingEngine，done + failed 两例） +- [ ] **Step 2: 验证失败** +- [ ] **Step 3: 实现**（包装 `vnpy_ctabacktester.backtesting.BacktestingEngine`，`set_parameters/add_strategy/load_data/run_backtesting/calculate_result` 签名由执行 subagent 读源码 + S2 spike 填充；异常 → status=failed + error_msg） +- [ ] **Step 4: 验证通过** +- [ ] **Step 5: Commit** — `git commit -m "feat(backtest): cta_engine BacktestingEngine wrapper"` + +--- + +## Task 9: sanguo_backtest/cta_optimizer.py + +**Files:** `sanguo_backtest/cta_optimizer.py`, `tests/backtest/test_cta_optimizer.py` +**Interfaces:** `run_cta_optimization(strategy_class, symbol, grid, start, end, cfg, db_path, max_workers) → list[BacktestResult]` + +- [ ] **Step 1: 写测试**（mock run_optimization，多结果 + 失败两例） +- [ ] **Step 2-4**: 失败 → 实现（包装 `run_optimization`，`output=False` headless，`max_workers` 来自 S2）→ 通过 +- [ ] **Step 5: Commit** — `git commit -m "feat(backtest): cta_optimizer run_optimization wrapper"` + +--- + +## Task 10: sanguo_orchestrator/task.py + pool.py + +**Files:** `sanguo_orchestrator/__init__.py`, `task.py`, `pool.py`, `tests/orchestrator/test_task.py`, `tests/orchestrator/test_pool.py` + +- [ ] **Step 1: 写测试**（TaskState 转换 PENDING→RUNNING→DONE/FAILED + TaskPool submit/get_status） +- [ ] **Step 2-4**: 实现 `TaskState(enum)` + `Task(dataclass, start/complete/fail)` + `TaskPool(submit/get_status/get_task)` → 通过 +- [ ] **Step 5: Commit** — `git commit -m "feat(orchestrator): task 状态机 + pool 封装"` + +```python +# sanguo_orchestrator/task.py +import enum +from dataclasses import dataclass + +class TaskState(enum.Enum): + PENDING = "pending"; RUNNING = "running"; DONE = "done"; FAILED = "failed" + +@dataclass +class Task: + task_id: str; task_type: str + status: TaskState = TaskState.PENDING + result_id: int | None = None + error_msg: str | None = None + def start(self): + if self.status != TaskState.PENDING: raise ValueError(f"不能从 {self.status} 启动") + self.status = TaskState.RUNNING + def complete(self, result_id: int): + self.status = TaskState.DONE; self.result_id = result_id + def fail(self, error_msg: str): + self.status = TaskState.FAILED; self.error_msg = error_msg +``` + +```python +# sanguo_orchestrator/pool.py +from .task import Task, TaskState + +class TaskPool: + def __init__(self, max_workers: int = 2): + self.max_workers = max_workers + self._tasks: dict[str, Task] = {} + def submit(self, task_id, task_type): + t = Task(task_id=task_id, task_type=task_type); self._tasks[task_id] = t; return t + def get_status(self, task_id): + t = self._tasks.get(task_id); return t.status if t else None + def get_task(self, task_id): return self._tasks.get(task_id) +``` + +--- + +## Task 11: sanguo_orchestrator/runner.py + +**Files:** `sanguo_orchestrator/runner.py`, `tests/orchestrator/test_runner.py` +**Interfaces:** `Orchestrator(db_path, file_dir, max_workers)` + `submit_cta/submit_optimize/submit_factor` + `get_status/get_result` + +- [ ] **Step 1: 写测试**（submit_cta + _run_sync + get_status==DONE） +- [ ] **Step 2-4**: 实现 Orchestrator（submit 创建 Task + 存 pending spec；_run_sync 调 run_cta_backtest；get_result 调 load_result）→ 通过 +- [ ] **Step 5: Commit** — `git commit -m "feat(orchestrator): runner 任务调度器"` + +```python +# sanguo_orchestrator/runner.py +from .pool import TaskPool +from .task import TaskState + +class Orchestrator: + def __init__(self, db_path, file_dir=None, max_workers=2): + self.db_path = db_path; self.file_dir = file_dir + self.pool = TaskPool(max_workers=max_workers); self._pending = {} + + def submit_cta(self, strategy_class, symbol, params, start, end, cfg): + task_id = f"cta_{symbol}_{id(params)}" + self.pool.submit(task_id, "cta") + self._pending = dict(strategy_class=strategy_class, symbol=symbol, + params=params, start=start, end=end, cfg=cfg) + return task_id + + def _run_sync(self, task_id): + from sanguo_backtest.cta_engine import run_cta_backtest + task = self.pool.get_task(task_id); task.start() + try: + result = run_cta_backtest(self._pending["strategy_class"], self._pending["symbol"], + self._pending["params"], self._pending["start"], + self._pending["end"], self._pending["cfg"], self.db_path) + task.complete(result_id=id(result)) + except Exception as e: + task.fail(f"{type(e).__name__}: {e}") + return task + + def get_status(self, task_id): return self.pool.get_status(task_id) + + def get_result(self, task_id): + task = self.pool.get_task(task_id) + if task and task.status == TaskState.DONE and task.result_id: + from sanguo_backtest.result_store import load_result + return load_result(task.result_id, self.db_path) + return None +``` + +> submit_optimize / submit_factor 由执行 subagent 参照 submit_cta 扩展。 + +--- + +## Task 12: sanguo_api（轻量 FastAPI 5 路由） + +**Files:** `sanguo_api/{__init__,schemas,app,routes}.py`, `config/backtest.yaml`, `tests/api/test_routes.py` + +- [ ] **Step 1: 写测试**（TestClient：submit_cta 200、get_status done、缺字段 422） +- [ ] **Step 2-4**: 实现 schemas（Pydantic）+ app（create_app）+ routes（5 路由）→ 通过 + +```python +# sanguo_api/schemas.py +from pydantic import BaseModel +class CtaBacktestRequest(BaseModel): + symbol: str; strategy: str; params: dict = {}; start: str; end: str +class OptimizeRequest(BaseModel): + symbol: str; strategy: str; grid: dict; start: str; end: str; max_workers: int = 2 +class FactorAnalysisRequest(BaseModel): + symbols: list[str]; factor_names: list[str]; start: str; end: str +``` + +```python +# sanguo_api/app.py +from fastapi import FastAPI +from .routes import router, set_orchestrator +from sanguo_orchestrator.runner import Orchestrator + +def create_app(db_path, file_dir=None): + app = FastAPI(title="Sanguo Quant API") + set_orchestrator(Orchestrator(db_path=db_path, file_dir=file_dir)) + app.include_router(router, prefix="/api/v1") + return app +``` + +```python +# sanguo_api/routes.py +from fastapi import APIRouter, HTTPException +from .schemas import CtaBacktestRequest, OptimizeRequest, FactorAnalysisRequest + +router = APIRouter(); _orchestrator = None +def set_orchestrator(orch): + global _orchestrator; _orchestrator = orch +def get_orchestrator(): return _orchestrator + +@router.post("/backtest/cta") +def submit_cta(req: CtaBacktestRequest): + tid = get_orchestrator().submit_cta(strategy_class=req.strategy, symbol=req.symbol, + params=req.params, start=req.start, end=req.end, cfg=None) + return {"task_id": tid} + +@router.post("/backtest/optimize") +def submit_optimize(req: OptimizeRequest): + # 执行时补：orch.submit_optimize(...) + return {"task_id": "pending_impl"} + +@router.post("/factor/analyze") +def submit_factor(req: FactorAnalysisRequest): + # 执行时补：orch.submit_factor(...) + return {"task_id": "pending_impl"} + +@router.get("/task/{task_id}") +def get_status(task_id: str): + s = get_orchestrator().get_status(task_id) + if s is None: raise HTTPException(404, "task not found") + return {"task_id": task_id, "status": s.value if hasattr(s, "value") else str(s)} + +@router.get("/task/{task_id}/result") +def get_result(task_id: str): + r = get_orchestrator().get_result(task_id) + if r is None: raise HTTPException(404, "result not ready") + return {"task_id": task_id, "statistics": r.statistics} +``` + +```yaml +# config/backtest.yaml +backtest: + max_workers: 2 + db_path: /volume1/stock/sanguo_vnpy/data/backtest_results.db + file_dir: /volume1/stock/sanguo_vnpy/data/backtest_files +api: + host: 0.0.0.0 + port: 8000 +``` + +- [ ] **Step 5: Commit** — `git commit -m "feat(api): 轻量 FastAPI 5 路由（无 JWT/WS/前端）"` + +> optimize/factor 路由的 `pending_impl` 由执行 subagent 补完。 + +--- + +## Task 13: 端到端冒烟 + 集成测试 + +**Files:** `scripts/smoke_phase2.py`, `tests/integration/test_phase2_e2e.py` + +- [ ] **Step 1: 写冒烟脚本**（result_store 增删查 + API 启动 + 因子 registry） +- [ ] **Step 2: 集成测试 + 覆盖率** — `pytest tests/ -v --cov=sanguo_factor --cov=sanguo_backtest --cov=sanguo_orchestrator --cov=sanguo_api --cov-report=term-missing` → 全绿 + ≥80% +- [ ] **Step 3: 容器端到端冒烟**（真实 vnpy） +- [ ] **Step 4: Commit** — `git commit -m "test(phase2): 端到端冒烟 + 集成测试（覆盖率 ≥80%）"` +- [ ] **Step 5: requesting-code-review + finishing-a-development-branch** + +--- + +## Self-Review + +**1. Spec coverage**：因子层(T4-6) + 回测层(T7-9) + 编排层(T10-11) + API(T12) + Spike(T1-3) + 集成(T13) = design §2-§6 全覆盖 ✅ +**2. Placeholder**：`pending_impl`（routes optimize/factor）、cta_engine/optimizer vnpy API 细节（S2）、analyzer AlphaDataset 流程（S1）—— 均依赖 spike，已标注，非偷懒 +**3. Type consistency**：BacktestResult(T7) / Task+TaskState(T10) / Orchestrator(T11) 跨 task 一致 ✅ + +--- + +## Execution Handoff + +**Plan saved to `docs/superpowers/plans/2026-07-05-phase2-factor-backtest.md`.** + +按 /goal 授权，采用 **Subagent-Driven** 执行（superpowers:subagent-driven-development）：每 task 派 fresh subagent + 两阶段 review。从 Task 1（S3 spike）开始。