# 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）开始。