docs(phase2): 因子/回测层实现计划（13 task：3 spike + 10 实现，spike 先行）

2026-07-06 07:25:53 +08:00
parent 83ca15b57b
commit a480fd389f
1 changed files with 783 additions and 0 deletions
@@ -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）开始。