sanguo/sanguo_vnpy_v2
2
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ae3a76809158a40c76d6f2c3fdef9866bf0d9272
sanguo_vnpy_v2/docs/superpowers/plans/2026-07-05-phase2-factor-backtest.md
T

32 KiB
Raw Blame History

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 脚本
# 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: 测试占位
# 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 脚本
# 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: 写失败测试
# 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: 实现
# 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: 写测试
# 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")

# 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: 实现
# 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]

# 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）
# 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()

# 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: 实现
# 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)

# 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）
# 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 封装"
# 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

# 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 任务调度器"
# 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 路由）→ 通过
# 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

# 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

# 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}

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

Reference in New Issue View Git Blame Copy Permalink