From a62013cb9cb6df851b07fde618fb86766a5958a9 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Mon, 24 Aug 2026 23:52:40 +0800 Subject: [PATCH] =?UTF-8?q?docs(factor):=20=E7=AC=AC=E4=B8=80=E6=89=B9?= =?UTF-8?q?=E5=AE=9E=E6=96=BD=E8=AE=A1=E5=88=92=E2=80=94=E2=80=9413?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1TDD=E6=8B=86=E8=A7=A3(=E6=8C=82=E8=BD=BD/?= =?UTF-8?q?=E8=82=A1=E7=A5=A8=E6=B1=A0/=E6=8C=87=E6=A0=87/=E5=AD=98?= =?UTF-8?q?=E5=82=A8/=E5=BC=95=E6=93=8E/CLI/API/=E5=89=8D=E7=AB=AF?= =?UTF-8?q?=E4=B8=89=E9=A1=B5/NAS=E9=AA=8C=E8=AF=81),=E5=90=ABST=E8=BF=87?= =?UTF-8?q?=E6=BB=A4=E5=81=8F=E5=B7=AE=E5=A3=B0=E6=98=8E=E4=B8=8Espawn?= =?UTF-8?q?=E6=B1=A0=E8=A7=84=E9=81=BF=E8=AE=BE=E8=AE=A1=20[nas]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../2026-08-24-factor-batch1-alpha101-158.md | 2215 +++++++++++++++++ 1 file changed, 2215 insertions(+) create mode 100644 docs/superpowers/plans/2026-08-24-factor-batch1-alpha101-158.md diff --git a/docs/superpowers/plans/2026-08-24-factor-batch1-alpha101-158.md b/docs/superpowers/plans/2026-08-24-factor-batch1-alpha101-158.md new file mode 100644 index 0000000..97c2b29 --- /dev/null +++ b/docs/superpowers/plans/2026-08-24-factor-batch1-alpha101-158.md @@ -0,0 +1,2215 @@ +# 因子研究第一批（Alpha101+Alpha158 挂载与全A重验）实施计划 + +> **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 自带的 Alpha101(100)+Alpha158(158) 因子表达式挂进 sanguo_factor 注册表，建成批量评估管线（全A 8.5 年 → IC 排行榜落盘）+ 前端「因子研究工作台」三页面，部署 NAS 验证。 + +**Architecture:** 评估引擎**不走** AlphaDataset 的 spawn 进程池（其 prepare_data 会对每个表达式 pickle 整个 9M 行 DataFrame），改为直接调用 `vnpy.alpha.dataset.utility.calculate_by_expression` 在进程内逐因子计算（内存天然有界，无 spawn 路径问题）。股票池由 dbbardata 直读（sqlite3 分块查询 → polars），vwap 在加载层派生（turnover/volume）。指标全部 numpy/pandas 向量化（RankIC=秩相关逐行向量化）。结果落 `factor_eval.db`（eval_runs + eval_results 两表，metrics 按周期存 JSON）。现有单因子分析链路（AlphaLabSession/alphalens tears）**零改动**。 + +**Tech Stack:** Python 3.10 (venv310) / polars 1.42 / pandas / sqlite3 / FastAPI / Vue3 + TS + Element Plus + echarts + +**Spec:** `docs/superpowers/specs/2026-08-24-factor-batch1-alpha101-158.md` +**前端视觉唯一真相源:** `docs/factor_research/factor-workbench-mockup.html`（用户已确认）+ `frontend/src/styles/tokens.css` + +## Global Constraints + +- 测试一律用 `venv310/bin/python -m pytest ...`（系统 python 的 pandas 3.0.1 假红） +- commit message 末尾必打标签：后端/数据链路改动 `[vps]`（VPS 运行时也触发），纯前端/CI/docker `[nas]` +- 不改：`sanguo_factor/analyzer.py`、`alpha_lab.py` 现有行为、`routes.py` 既有 /factor/list /factor/analyze 端点、vnpy_v4.4.0/ 任何文件 +- 新 API 路由放新文件 `sanguo_api/routes_factor.py`（仿 routes_paper.py 模式），在 `app.py` 注册 +- 前端页面样式 1:1 对齐 mockup：深空黑 `--bg`、荧光青 `--brand`、琥珀 `--amber`、红涨绿跌 `--up/--down`、近直角、mono 数字；读 `frontend/src/styles/tokens.css` 取变量名，不写死色值 +- 评估口径（spec 定稿）：主板+创业板（代码前缀 60/00/30，剔科创68/北交/ETF）、2018-01-01~2026-06-30、RankIC 前瞻 1/5/10 日、退市股保留、不做行业中性化 +- **⚠️ 对 spec 的一处有意识偏差**：不实现 ST 过滤。本地无 point-in-time 名称史，用「当前名称」回溯过滤会引入前视偏差，比不过滤更糟。代码前缀过滤 + 120 交易日预热已控住主要污染。在最终报告向用户说明 +- dbbardata 列名：`symbol, exchange, datetime, interval, volume, turnover, open_interest, open_price, high_price, low_price, close_price`，datetime 为 ISO 字符串，日线 interval='d'（既有直查范式见 `sanguo_backtest/ashare_engine.py:235`） +- NAS 容器名 `sanguo_vnpy_v2`；docker 全路径 `/var/packages/Docker/target/usr/bin/docker` +- YAGNI（spec 明确不做）：行业中性化、因子合成、Alpha360、GTJA191、聚宽财务因子、相关性聚类去重代码化（跑完数据后再分析） + +## File Structure + +``` +sanguo_factor/ + data_adapter.py [改] convert_bars_to_alpha_df 补 vwap 派生列 + alpha_datasets.py [新] 挂载 vnpy Alpha101/158 表达式进注册表 + universe.py [新] dbbardata 分块直读 → AlphaLab 格式 polars df（含 vwap/bar_idx） + metrics.py [新] 向量化 RankIC/ICIR/t/胜率/多空年化/换手/十分组 + eval_store.py [新] factor_eval.db 读写（eval_runs/eval_results） + batch_eval.py [新] 批量评估引擎（进程内逐因子算 → 指标 → 落盘） +sanguo_api/ + schemas.py [改] +FactorBatchEvalRequest + routes_factor.py [新] GET runs/leaderboard/detail + POST submit + app.py [改] 注册 routes_factor + set_eval_db_path 接线 +sanguo_orchestrator/ + runner.py [改] +submit_batch_eval +_batch_eval_worker + _persist_factor_eval +scripts/factor_research/ + run_eval.py [新] CLI（冒烟/全量，NAS docker exec 用） +frontend/src/ + api/factor.ts [改] +评估相关类型与请求 + router/index.ts [改] +3 路由 + views/Layout.vue [改] 投研组 +2 菜单项 +面包屑 + views/factor/Leaderboard.vue [新] IC 排行榜 + views/factor/LeaderboardDetail.vue [新] 因子详情 + views/factor/BatchEval.vue [新] 批量评估 +tests/factor/ + test_alpha_datasets.py [新] + test_vwap_adapter.py [新] + test_universe.py [新] + test_metrics.py [新] + test_eval_store.py [新] + test_batch_eval.py [新] +tests/api/ + test_factor_eval_routes.py [新] +.gitea/workflows/ci-cd.yml [改] pytest 门禁加 tests/factor +``` + +--- + +### Task 1: vwap 派生列（data_adapter） + +**Files:** +- Modify: `sanguo_factor/data_adapter.py` +- Test: `tests/factor/test_vwap_adapter.py` + +**Interfaces:** +- Produces: `convert_bars_to_alpha_df(bars) -> pl.DataFrame` 输出多一列 `vwap: Float64`（volume>0 时 = turnover/volume，否则 null）。下游 Task 2/6 依赖「df 有 vwap 列即可被表达式引用」（`calculate_by_expression` 按 df.columns 自动注入变量名）。 + +- [ ] **Step 1: 写失败测试** + +```python +# tests/factor/test_vwap_adapter.py +"""vwap 派生列:turnover/volume,volume=0 → null.""" +import sys, os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "vnpy_v4.4.0"))) + +from datetime import datetime +from vnpy.trader.object import BarData, Interval +from sanguo_factor.data_adapter import convert_bars_to_alpha_df + + +def _bar(symbol, dt, close=10.0, volume=100.0, turnover=1000.0): + return BarData( + symbol=symbol, exchange="SSE", datetime=dt, interval=Interval.DAILY, + open_price=9.9, high_price=10.1, low_price=9.8, close_price=close, + volume=volume, turnover=turnover, open_interest=0, gateway_name="T", + ) + + +def test_vwap_derived(): + df = convert_bars_to_alpha_df([_bar("600000", datetime(2024, 1, 5))]) + assert df["vwap"][0] == 10.0 # 1000 / 100 + + +def test_vwap_null_when_volume_zero(): + df = convert_bars_to_alpha_df([_bar("600000", datetime(2024, 1, 5), volume=0.0, turnover=0.0)]) + assert df["vwap"][0] is None + + +def test_empty_schema_has_vwap(): + df = convert_bars_to_alpha_df([]) + assert "vwap" in df.columns + assert df.height == 0 +``` + +- [ ] **Step 2: 跑测试确认失败** + +Run: `venv310/bin/python -m pytest tests/factor/test_vwap_adapter.py -v` +Expected: FAIL（`vwap` 不在 columns） + +- [ ] **Step 3: 最小实现** + +`convert_bars_to_alpha_df` 的空 schema dict 加 `"vwap": pl.Float64`；构造 dict 加： + +```python +"vwap": [(b.turnover / b.volume) if (b.turnover and b.volume) else None for b in bars], +``` + +docstring 补一行：`vwap: turnover/volume 派生(volume=0 置 null),Alpha101 表达式依赖`。 + +- [ ] **Step 4: 跑测试确认通过 + 回归** + +Run: `venv310/bin/python -m pytest tests/factor -v` +Expected: 全 PASS（含既有 5 个测试文件） + +- [ ] **Step 5: Commit** + +```bash +git add sanguo_factor/data_adapter.py tests/factor/test_vwap_adapter.py +git commit -m "feat(factor): data_adapter 补 vwap 派生列(turnover/volume,零量置null)——Alpha101 表达式 59 处引用 vwap,dbbardata 无此列 [vps]" +``` + +--- + +### Task 2: 挂载 Alpha101/Alpha158 表达式（alpha_datasets） + +**Files:** +- Create: `sanguo_factor/alpha_datasets.py` +- Test: `tests/factor/test_alpha_datasets.py` + +**Interfaces:** +- Consumes: `vnpy.alpha.dataset.datasets.alpha_101.Alpha101` / `alpha_158.Alpha158`（构造只注册表达式不计算，空 df 即可实例化） +- Produces: + - `mount_alpha101() -> int`、`mount_alpha158() -> int`、`mount_all() -> dict[str, int]`（幂等，已存在同名跳过） + - 注册后 `list_factors("alpha101")` 100 项、`list_factors("alpha158")` 158 项，factor 名保持 vnpy 原名（`alpha1..alpha100`、`kmid/klen/roc_5/ma_20/...`） + - 挂载入口必须在 `sanguo_factor/__init__.py` 导入即生效（`from . import alpha_datasets` 链上自动注册）——下游 `/factor/list`、批量评估、CLI 全靠它 + +- [ ] **Step 1: 写失败测试** + +```python +# tests/factor/test_alpha_datasets.py +"""Alpha101/158 全量挂载:数量/幂等/类别/表达式可用性.""" +import sys, os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "vnpy_v4.4.0"))) + +from sanguo_factor.alpha_datasets import mount_all +from sanguo_factor.registry import list_factors, get_factor, _REGISTRY + + +def test_mount_all_counts(): + counts = mount_all() + assert counts == {"alpha101": 100, "alpha158": 158} + + +def test_categories_and_expression(): + mount_all() + a101 = list_factors("alpha101") + assert {f["name"] for f in a101} >= {"alpha1", "alpha2", "alpha100"} + assert get_factor("alpha5")["expression"].count("vwap") >= 1 # alpha5 显式用 vwap + a158 = list_factors("alpha158") + names = {f["name"] for f in a158} + assert {"kmid", "klen", "roc_5", "ma_20", "std_20", "wvma_20", "vwap_0"} <= names + + +def test_mount_idempotent(): + mount_all() + n_before = len(_REGISTRY) + mount_all() + assert len(_REGISTRY) == n_before + + +def test_import_side_effect_registers(): + # sanguo_factor 包导入即挂载(下游 /factor/list 依赖) + import sanguo_factor + assert len(list_factors("alpha101")) == 100 +``` + +- [ ] **Step 2: 跑测试确认失败** + +Run: `venv310/bin/python -m pytest tests/factor/test_alpha_datasets.py -v` +Expected: FAIL（ModuleNotFoundError: sanguo_factor.alpha_datasets） + +- [ ] **Step 3: 实现** + +```python +# sanguo_factor/alpha_datasets.py +"""挂载 vnpy 自带 Alpha101/Alpha158 表达式因子到注册表. + +vnpy 的 AlphaDataset 子类在 __init__ 里只做 add_feature 字符串注册(不触发计算), +用空 schema df 实例化即可零成本提取全部表达式。 +""" +import sys +import 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 .registry import register_factor, list_factors, _REGISTRY + +_ALPHA_SCHEMA = { + "vt_symbol": __import__("polars").Utf8, + "datetime": __import__("polars").Datetime, + "open": __import__("polars").Float64, + "high": __import__("polars").Float64, + "low": __import__("polars").Float64, + "close": __import__("polars").Float64, + "volume": __import__("polars").Float64, + "turnover": __import__("polars").Float64, + "open_interest": __import__("polars").Float64, + "vwap": __import__("polars").Float64, +} + +_DUMMY_PERIOD = ("2018-01-01", "2018-01-02") + + +def _extract_expressions(dataset_cls) -> dict[str, str]: + """实例化数据集类,提取其注册的全部 {name: expression}.""" + import polars as pl + df = pl.DataFrame(schema=_ALPHA_SCHEMA) + ds = dataset_cls(df, _DUMMY_PERIOD, _DUMMY_PERIOD, _DUMMY_PERIOD) + return {name: str(expr) for name, expr in ds.feature_expressions.items()} + + +def _mount(dataset_cls, category: str) -> int: + """把一个数据集类的表达式挂进注册表(已存在同名跳过,幂等).""" + mounted = 0 + for name, expression in _extract_expressions(dataset_cls).items(): + if name in _REGISTRY: + continue + register_factor(name, expression, category) + mounted += 1 + return mounted + + +def mount_alpha101() -> int: + """挂载 WorldQuant Alpha101(100 个),category=alpha101.""" + from vnpy.alpha.dataset.datasets.alpha_101 import Alpha101 + return _mount(Alpha101, "alpha101") + + +def mount_alpha158() -> int: + """挂载 Qlib Alpha158(158 个),category=alpha158.""" + from vnpy.alpha.dataset.datasets.alpha_158 import Alpha158 + return _mount(Alpha158, "alpha158") + + +def mount_all() -> dict[str, int]: + """挂载全部内置数据集,返回各类新挂数量.""" + return {"alpha101": mount_alpha101(), "alpha158": mount_alpha158()} + + +# 模块导入即挂载(与 library.py 内置因子同一模式,下游 import sanguo_factor 即全量可见) +mount_all() +``` + +> 注：`__import__("polars")` 的写法太丑——实现时直接顶部 `import polars as pl`，schema 用 `pl.Utf8` 等（polars 是硬依赖，无需 lazy）。上面为计划示意，**实现以顶部 import polars 为准**。 + +`sanguo_factor/__init__.py` 末尾追加一行： + +```python +from . import alpha_datasets # noqa: F401 挂载 Alpha101/158(导入即注册) +``` + +（先读该文件现有内容，追加不重排。） + +- [ ] **Step 4: 跑测试确认通过** + +Run: `venv310/bin/python -m pytest tests/factor/test_alpha_datasets.py tests/factor -v` +Expected: 全 PASS + +- [ ] **Step 5: Commit** + +```bash +git add sanguo_factor/alpha_datasets.py sanguo_factor/__init__.py tests/factor/test_alpha_datasets.py +git commit -m "feat(factor): 挂载 vnpy Alpha101(100)+Alpha158(158) 全量表达式进注册表——空df实例化零成本提取,导入即注册,幂等 [vps]" +``` + +--- + +### Task 3: 评估股票池加载（universe） + +**Files:** +- Create: `sanguo_factor/universe.py` +- Test: `tests/factor/test_universe.py` + +**Interfaces:** +- Consumes: dbbardata 表结构（见 Global Constraints） +- Produces: + - `STOCK_PREFIXES = ("60", "00", "30")`、`WARMUP_BARS = 120` + - `load_universe_bars(vnpy_db: str, start: str, end: str, symbols: list[str] | None = None, limit: int | None = None) -> pl.DataFrame` + - start/end 为评估窗（YYYY-MM-DD）；内部自动加 lookback（start-300 天）与 forward 缓冲（end+45 天，供 10 日前瞻收益） + - 返回列：`vt_symbol(Utf8), datetime(Datetime, 已截断到日), open, high, low, close, volume, turnover, open_interest, vwap, bar_idx(Int64)` + - `vwap`：`volume>0 → turnover/volume 否则 null` + - `bar_idx`：按 symbol 分组、datetime 升序的 0 起行号（预热/次新过滤用） + - `limit`：随机抽样 N 只（`random.Random(42)` 固定种子，冒烟可复现）；`symbols` 显式给定时只查这些 + - `evaluation_filter(df: pl.DataFrame, start: str, end: str) -> pl.DataFrame`：`datetime ∈ [start, end] 且 bar_idx >= WARMUP_BARS`（IC 评估行集） + +- [ ] **Step 1: 写失败测试** + +```python +# tests/factor/test_universe.py +"""股票池加载:前缀过滤/时间缓冲/vwap/bar_idx/显式symbols/limit抽样.""" +import sqlite3 +import sys, os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "vnpy_v4.4.0"))) + +import pytest +from sanguo_factor.universe import load_universe_bars, evaluation_filter, WARMUP_BARS + +_DDL = """ +CREATE TABLE dbbardata( + symbol TEXT, exchange TEXT, datetime TEXT, interval TEXT, + volume REAL, turnover REAL, open_interest REAL, + open_price REAL, high_price REAL, low_price REAL, close_price REAL) +""" + + +def _mk_db(tmp_path, rows): + db = str(tmp_path / "qt.db") + conn = sqlite3.connect(db) + conn.execute(_DDL) + conn.executemany("INSERT INTO dbbardata VALUES(?,?,?,?,?,?,?,?,?,?,?)", rows) + conn.commit(); conn.close() + return db + + +def _row(sym, ex, day, close, volume=100.0, turnover=None): + return (sym, ex, f"{day} 00:00:00", "d", volume, + turnover if turnover is not None else close * volume, 0, + close, close, close, close) + + +@pytest.fixture() +def db(tmp_path): + rows = [] + # 600000:60根(2017-10~2017-12) + 评估窗内 3 根 + for i in range(60): + rows.append(_row("600000", "SSE", f"2017-10-{(i % 28) + 1:02d}", 10.0 + i * 0.01)) + for d in ("2018-01-02", "2018-01-03", "2018-01-04"): + rows.append(_row("600000", "SSE", d, 11.0)) + # 000001:只有 2 根(次新,bar_idx pl.DataFrame: + """读评估窗(含前后缓冲)全A日线 → AlphaLab 格式(含 vwap/bar_idx).""" + lookback_start = ( + datetime.strptime(start, "%Y-%m-%d") - timedelta(days=_LOOKBACK_DAYS) + ).strftime("%Y-%m-%d") + forward_end = ( + datetime.strptime(end, "%Y-%m-%d") + timedelta(days=_FORWARD_DAYS) + ).strftime("%Y-%m-%d") + + if symbols is None: + conn = sqlite3.connect(vnpy_db, timeout=30) + try: + likes = " OR ".join(f"symbol LIKE '{p}%'" for p in STOCK_PREFIXES) + cur = conn.execute( + f"SELECT DISTINCT symbol FROM dbbardata " + f"WHERE interval='d' AND datetime>=? AND datetime<=? AND ({likes})", + (lookback_start, forward_end), + ) + symbols = [r[0] for r in cur.fetchall()] + finally: + conn.close() + + if not symbols: + return _empty_alpha_df() + if limit is not None and limit < len(symbols): + symbols = sorted(random.Random(42).sample(symbols, limit)) + + chunks: list[pl.DataFrame] = [] + for i in range(0, len(symbols), _CHUNK): + part = symbols[i : i + _CHUNK] + ph = ",".join("?" * len(part)) + conn = sqlite3.connect(vnpy_db, timeout=30) + conn.execute("PRAGMA busy_timeout=30000") + try: + cur = conn.execute( + f"SELECT {_COLS} FROM dbbardata " + f"WHERE interval='d' AND datetime>=? AND datetime<=? AND symbol IN ({ph})", + (lookback_start, forward_end, *part), + ) + rows = cur.fetchall() + finally: + conn.close() + if rows: + chunks.append(pl.DataFrame( + rows, + schema={"symbol": pl.Utf8, "exchange": pl.Utf8, "dt": pl.Utf8, + "volume": pl.Float64, "turnover": pl.Float64, "open_interest": pl.Float64, + "open": pl.Float64, "high": pl.Float64, "low": pl.Float64, "close": pl.Float64}, + orient="row", + )) + if not chunks: + return _empty_alpha_df() + + df = pl.concat(chunks) + df = ( + df.with_columns( + pl.col("dt").str.slice(0, 10).str.to_datetime("%Y-%m-%d").alias("datetime"), + (pl.col("symbol") + "." + pl.col("exchange")).alias("vt_symbol"), + pl.when(pl.col("volume") > 0) + .then(pl.col("turnover") / pl.col("volume")) + .otherwise(None) + .alias("vwap"), + ) + .sort(["vt_symbol", "datetime"]) + .with_columns(pl.int_range(pl.len()).over("vt_symbol").alias("bar_idx")) + .select(["vt_symbol", "datetime", "open", "high", "low", "close", + "volume", "turnover", "open_interest", "vwap", "bar_idx"]) + ) + return df + + +def evaluation_filter(df: pl.DataFrame, start: str, end: str) -> pl.DataFrame: + """IC 评估行集:窗口内 + 预热期已过.""" + return df.filter( + (pl.col("datetime") >= pl.lit(start).str.to_date()) + & (pl.col("datetime") <= pl.lit(end).str.to_date()) + & (pl.col("bar_idx") >= WARMUP_BARS) + ) + + +def _empty_alpha_df() -> pl.DataFrame: + return pl.DataFrame(schema={ + "vt_symbol": pl.Utf8, "datetime": pl.Datetime, + "open": pl.Float64, "high": pl.Float64, "low": pl.Float64, "close": pl.Float64, + "volume": pl.Float64, "turnover": pl.Float64, "open_interest": pl.Float64, + "vwap": pl.Float64, "bar_idx": pl.Int64, + }) +``` + +> 实现注意：`pl.lit(start).str.to_date()` 与 `datetime` 列比较——若类型不匹配报错，改用 `datetime.strptime(start, "%Y-%m-%d")` python 对象做 `pl.lit(...)`。以测试通过为准，别死磕一种写法。 + +- [ ] **Step 4: 跑测试确认通过** + +Run: `venv310/bin/python -m pytest tests/factor/test_universe.py -v` +Expected: 6 个全 PASS + +- [ ] **Step 5: Commit** + +```bash +git add sanguo_factor/universe.py tests/factor/test_universe.py +git commit -m "feat(factor): 评估股票池加载器——dbbardata分块直读polars,前缀60/00/30,300天lookback+45天forward缓冲,vwap派生,bar_idx预热 [vps]" +``` + +--- + +### Task 4: 向量化评估指标（metrics） + +**Files:** +- Create: `sanguo_factor/metrics.py` +- Test: `tests/factor/test_metrics.py` + +**Interfaces:** +- Produces（全部操作 pandas DataFrame，index=日期 DatetimeIndex，columns=vt_symbol，值=因子/收益；NaN=缺失自动剔除）: + - `TRADING_DAYS_PER_YEAR = 244` + - `rank_corr_rows(A: pd.DataFrame, B: pd.DataFrame) -> pd.Series`：逐日 Spearman 秩相关（列自动取交集） + - `factor_turnover(F: pd.DataFrame) -> float`：`1 - 相邻两日因子秩相关` 均值 + - `long_short_annual_return(F, R) -> float | None`：top10%−bottom10% 等权日收益均值复利年化 + - `decile_annual_returns(F, R) -> list[float | None]`：十分组等权年化（D1 最低→D10 最高） + - `monthly_ic(ic: pd.Series) -> list[dict]`：`[{"month": "2018-01", "ic": 0.05}, ...]` + - `classify(icir, t_stat) -> str`：`"effective" | "watch" | "eliminated"`（|ICIR|≥0.3 且 |t|≥2 → effective；否则 |t|≥1.5 → watch） + - `summarize_factor(F, R1, R5, R10) -> dict`：`{"1": {...}, "5": {...}, "10": {...}, "turnover": float}`，每周期含 `count, ic_mean, ic_std, icir, t_stat, win_rate, ls_annual, monthly_ic, deciles, conclusion`；单因子异常时返回 `{"error": str}` 由调用方包装 + +- [ ] **Step 1: 写失败测试** + +```python +# tests/factor/test_metrics.py +"""向量化指标:已知输入的精确断言.""" +import sys, os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) + +import numpy as np +import pandas as pd +import pytest + +from sanguo_factor.metrics import ( + rank_corr_rows, factor_turnover, long_short_annual_return, + decile_annual_returns, monthly_ic, classify, summarize_factor, + TRADING_DAYS_PER_YEAR, +) + + +def _mat(values, cols=("A", "B", "C")): + idx = pd.to_datetime(["2024-01-01", "2024-01-02", "2024-01-03"]) + return pd.DataFrame(values, index=idx, columns=list(cols), dtype=float) + + +def test_rank_corr_perfect_monotonic(): + F = _mat([[1, 2, 3], [3, 2, 1], [1, 3, 2]]) + R = _mat([[10, 20, 30], [30, 20, 10], [10, 30, 20]]) + ic = rank_corr_rows(F, R) + assert (ic == 1.0).all() + + +def test_rank_corr_inverse(): + F = _mat([[1, 2, 3], [1, 2, 3]]) + R = _mat([[3, 2, 1], [30, 20, 10]]) + ic = rank_corr_rows(F, R) + assert (ic == -1.0).all() + + +def test_rank_corr_nan_propagates_row(): + F = _mat([[1, 2, 3], [np.nan, 2, 3]]) + R = _mat([[1, 2, 3], [1, 2, 3]]) + ic = rank_corr_rows(F, R) + assert not np.isnan(ic.iloc[0]) + assert np.isnan(ic.iloc[1]) # 2个有效值秩恒定 → 无方差 → nan + + +def test_turnover_zero_for_static(): + F = _mat([[1, 2, 3]] * 4) + assert factor_turnover(F) == pytest.approx(0.0) + + +def test_turnover_full_for_shuffled(): + F = _mat([[1, 2, 3], [3, 2, 1]]) # 完全逆序 → 秩相关-1 → 换手=2 + assert factor_turnover(F) == pytest.approx(2.0) + + +def test_long_short_direction(): + F = _mat([[1, 2, 3, 4, 5], [5, 4, 3, 2, 1]], cols=tuple("ABCDE")) + R = _mat([[0.01, 0.02, 0.03, 0.04, 0.05], + [0.01, 0.02, 0.03, 0.04, 0.05]], cols=tuple("ABCDE")) + # 每日 top10%(1只)=E bottom10%=A → 日均 0.04 + expected = (1 + 0.04) ** TRADING_DAYS_PER_YEAR - 1 + assert long_short_annual_return(F, R) == pytest.approx(expected) + + +def test_decile_monotonic(): + F = _mat([list(range(1, 11))] * 3, cols=tuple("ABCDEFGHIJ")) + R = _mat([[c / 100 for c in range(1, 11)]] * 3, cols=tuple("ABCDEFGHIJ")) + dec = decile_annual_returns(F, R) + assert len(dec) == 10 + assert all(d is not None for d in dec) + assert dec == sorted(dec) # D1最低收益 → D10最高收益 单调 + + +def test_monthly_ic_shape(): + idx = pd.to_datetime(["2024-01-05", "2024-01-10", "2024-02-01"]) + out = monthly_ic(pd.Series([0.1, 0.2, -0.1], index=idx)) + assert out[0] == {"month": "2024-01", "ic": pytest.approx(0.15)} + assert out[1] == {"month": "2024-02", "ic": pytest.approx(-0.1)} + + +def test_classify_rules(): + assert classify(0.5, 5.0) == "effective" + assert classify(-0.4, -3.0) == "effective" # 负 IC 强因子同样有效(反向) + assert classify(0.1, 1.8) == "watch" + assert classify(0.0, 0.5) == "eliminated" + + +def test_summarize_factor_structure(): + rng = np.random.default_rng(7) + idx = pd.bdate_range("2024-01-01", periods=60) + base = np.tile(np.arange(10.0, 90.0, 1.0), (60, 1)) + F = pd.DataFrame(base + rng.normal(0, 0.5, base.shape), index=idx) + R1 = pd.DataFrame(-0.001 * base + rng.normal(0, 0.001, base.shape), index=idx, columns=F.columns) + R5, R10 = R1 * 5, R1 * 10 + out = summarize_factor(F, R1, R5, R10) + assert set(out) == {"1", "5", "10", "turnover"} + p1 = out["1"] + for key in ("count", "ic_mean", "ic_std", "icir", "t_stat", "win_rate", + "ls_annual", "monthly_ic", "deciles", "conclusion"): + assert key in p1 + assert p1["count"] == 60 + assert p1["ic_mean"] < 0 # 构造为负相关 +``` + +- [ ] **Step 2: 跑测试确认失败** + +Run: `venv310/bin/python -m pytest tests/factor/test_metrics.py -v` +Expected: FAIL（ModuleNotFoundError） + +- [ ] **Step 3: 实现** + +```python +# sanguo_factor/metrics.py +"""批量评估指标:全向量化(RankIC=秩相关逐行),不依赖 alphalens. + +所有矩阵约定:pandas DataFrame,index=DatetimeIndex(日),columns=vt_symbol, +值=因子值或前瞻收益,NaN=缺失自动从当日截面剔除。 +""" +import warnings + +import numpy as np +import pandas as pd + +TRADING_DAYS_PER_YEAR = 244 + + +def _row_pearson(a: np.ndarray, b: np.ndarray, index: pd.Index) -> pd.Series: + """逐行 Pearson(输入已是秩),全 NaN/无方差行 → NaN.""" + mask = ~(np.isnan(a) | np.isnan(b)) + n = mask.sum(axis=1) + a0 = np.where(mask, a, np.nan) + b0 = np.where(mask, b, np.nan) + with warnings.catch_warnings(): + warnings.simplefilter("ignore", RuntimeWarning) + am = np.nanmean(a0, axis=1, keepdims=True) + bm = np.nanmean(b0, axis=1, keepdims=True) + ad = np.where(mask, a0 - am, 0.0) + bd = np.where(mask, b0 - bm, 0.0) + denom = np.sqrt((ad ** 2).sum(axis=1) * (bd ** 2).sum(axis=1)) + with np.errstate(invalid="ignore", divide="ignore"): + ic = np.where(denom > 0, (ad * bd).sum(axis=1) / denom, np.nan) + ic = np.where(n >= 3, ic, np.nan) # <3 只无意义 + return pd.Series(ic, index=index) + + +def rank_corr_rows(A: pd.DataFrame, B: pd.DataFrame) -> pd.Series: + """逐日 Spearman:先各自按行取秩再逐行 Pearson.""" + cols = A.columns.intersection(B.columns) + idx = A.index.intersection(B.index) + a = A.loc[idx, cols].rank(axis=1).to_numpy(dtype=float) + b = B.loc[idx, cols].rank(axis=1).to_numpy(dtype=float) + return _row_pearson(a, b, idx) + + +def factor_turnover(F: pd.DataFrame) -> float: + """换手率 = 1 - 相邻两日因子秩相关均值.""" + if len(F) < 2: + return 0.0 + corr = rank_corr_rows(F.iloc[1:], F.iloc[:-1]).replace([np.inf, -np.inf], np.nan).dropna() + return float(1.0 - corr.mean()) if len(corr) else 0.0 + + +def _quantile_mask(F: pd.DataFrame, lo_frac: float, hi_frac: float) -> pd.DataFrame: + """按行把因子值分位选mask(基于升序秩/当日有效数).""" + ranks = F.rank(axis=1, ascending=False) # 1=最大 + n = ranks.notna().sum(axis=1) + k = np.maximum((n * 0.1).round().astype(int), 1) + if lo_frac == 0.0: + return ranks.le(k, axis=0) & ranks.notna() + return ranks.ge(n - k + 1, axis=0) & ranks.notna() + + +def long_short_annual_return(F: pd.DataFrame, R: pd.DataFrame) -> float | None: + """多空年化:top10% - bottom10% 等权前瞻日收益均值,复利年化.""" + cols = F.columns.intersection(R.columns) + idx = F.index.intersection(R.index) + f, r = F.loc[idx, cols], R.loc[idx, cols] + top = _quantile_mask(f, 0.0, 0.1) + bot = _quantile_mask(f, 0.9, 1.0) + daily = (r.where(top).mean(axis=1) - r.where(bot).mean(axis=1)).dropna() + if daily.empty: + return None + return float((1.0 + daily.mean()) ** TRADING_DAYS_PER_YEAR - 1.0) + + +def decile_annual_returns(F: pd.DataFrame, R: pd.DataFrame) -> list[float | None]: + """十分组(D1因子最低→D10最高)等权年化收益.""" + cols = F.columns.intersection(R.columns) + idx = F.index.intersection(R.index) + f, r = F.loc[idx, cols], R.loc[idx, cols] + pct = f.rank(axis=1, ascending=True).div(f.notna().sum(axis=1), axis=0) + out: list[float | None] = [] + for d in range(10): + sel = (pct > d / 10) & (pct <= (d + 1) / 10) + daily = r.where(sel).mean(axis=1).dropna() + out.append( + float((1.0 + daily.mean()) ** TRADING_DAYS_PER_YEAR - 1.0) + if len(daily) else None + ) + return out + + +def monthly_ic(ic: pd.Series) -> list[dict]: + """IC 按月聚合(sparkline/详情图数据).""" + s = ic.dropna() + if s.empty: + return [] + g = s.groupby(s.index.to_period("M").to_timestamp()).mean() + return [{"month": t.strftime("%Y-%m"), "ic": round(float(v), 6)} for t, v in g.items()] + + +def classify(icir: float | None, t_stat: float | None) -> str: + """结论信号灯:|ICIR|>=0.3 且 |t|>=2 有效;否则 |t|>=1.5 观察;其余淘汰.""" + if icir is None or t_stat is None: + return "eliminated" + if abs(icir) >= 0.3 and abs(t_stat) >= 2.0: + return "effective" + if abs(t_stat) >= 1.5: + return "watch" + return "eliminated" + + +def _period_stats(ic: pd.Series) -> dict: + s = ic.dropna() + n = len(s) + if n < 2: + return {"count": int(n), "ic_mean": None, "ic_std": None, + "icir": None, "t_stat": None, "win_rate": None, + "monthly_ic": monthly_ic(ic)} + m = float(s.mean()) + sd = float(s.std()) + icir = m / sd if sd > 0 else None + t_stat = m / (sd / n ** 0.5) if sd > 0 else None + return { + "count": int(n), + "ic_mean": m, + "ic_std": sd, + "icir": icir, + "t_stat": t_stat, + "win_rate": float((s > 0).mean()), + "monthly_ic": monthly_ic(ic), + } + + +def summarize_factor(F: pd.DataFrame, R1: pd.DataFrame, R5: pd.DataFrame, R10: pd.DataFrame) -> dict: + """单因子全指标:三周期 IC 族 + 换手 + 多空 + 十分组 + 结论.""" + out: dict = {"turnover": factor_turnover(F)} + for p, R in (("1", R1), ("5", R5), ("10", R10)): + ic = rank_corr_rows(F, R) + stats = _period_stats(ic) + stats["ls_annual"] = long_short_annual_return(F, R) + stats["deciles"] = decile_annual_returns(F, R) + stats["conclusion"] = classify(stats["icir"], stats["t_stat"]) + out[p] = stats + return out +``` + +- [ ] **Step 4: 跑测试确认通过** + +Run: `venv310/bin/python -m pytest tests/factor/test_metrics.py -v` +Expected: 全 PASS（`test_rank_corr_nan_propagates_row` 那条：两列秩无方差 → denom=0 → NaN，与实现一致） + +- [ ] **Step 5: Commit** + +```bash +git add sanguo_factor/metrics.py tests/factor/test_metrics.py +git commit -m "feat(factor): 向量化评估指标——RankIC逐行秩相关/ICIR/t/胜率/多空年化/换手/十分组/月度IC/结论分级 [vps]" +``` + +--- + +### Task 5: 评估结果存储（eval_store） + +**Files:** +- Create: `sanguo_factor/eval_store.py` +- Test: `tests/factor/test_eval_store.py` + +**Interfaces:** +- Produces: + - `init_db(path)` + - `create_run(path, label, universe, symbols_count, factors_total, start, end, params: dict) -> str`（run_id 形如 `ev_20260824_1530_a1b2`） + - `save_results(path, run_id, rows: list[dict])`（rows 元素 = `{factor, category, expression, metrics: dict}`，UPSERT） + - `finish_run(path, run_id, status, factors_done)` + - `list_runs(path) -> list[dict]`（created_at 倒序） + - `get_rows(path, run_id, category=None, search=None) -> list[dict]`（metrics 已反序列化） + - `get_detail(path, run_id, factor) -> dict | None` + - `default_eval_db_path() -> str`：orchestrator backtest_results.db 同目录下 `factor_eval.db`；环境变量 `SANGUO_FACTOR_EVAL_DB` 可覆盖（NAS/VPS/本地三环境同一套代码） + +- [ ] **Step 1: 写失败测试** + +```python +# tests/factor/test_eval_store.py +"""eval_store:建库/建run/UPSERT落盘/查询过滤/finish.""" +import sys, os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) + +from sanguo_factor import eval_store + + +def _rows(): + return [ + {"factor": "alpha2", "category": "alpha101", "expression": "(-1)*ts_corr(...)", + "metrics": {"1": {"ic_mean": 0.05}, "turnover": 0.3}}, + {"factor": "kmid", "category": "alpha158", "expression": "(close-open)/open", + "metrics": {"1": {"ic_mean": -0.02}, "turnover": 0.6}}, + ] + + +def test_roundtrip(tmp_path): + db = str(tmp_path / "factor_eval.db") + eval_store.init_db(db) + run_id = eval_store.create_run(db, label="冒烟", universe="custom", symbols_count=50, + factors_total=2, start="2024-01-01", end="2024-12-31", + params={"periods": [1, 5, 10]}) + eval_store.save_results(db, run_id, _rows()) + eval_store.finish_run(db, run_id, "done", factors_done=2) + + runs = eval_store.list_runs(db) + assert len(runs) == 1 and runs[0]["run_id"] == run_id and runs[0]["status"] == "done" + + rows = eval_store.get_rows(db, run_id) + assert len(rows) == 2 + assert rows[0]["metrics"]["1"]["ic_mean"] == 0.05 + + assert eval_store.get_detail(db, run_id, "kmid")["category"] == "alpha158" + assert eval_store.get_detail(db, run_id, "nope") is None + + +def test_filter_by_category_and_search(tmp_path): + db = str(tmp_path / "factor_eval.db") + eval_store.init_db(db) + run_id = eval_store.create_run(db, label="x", universe="all_a", symbols_count=1, + factors_total=2, start="", end="", params={}) + eval_store.save_results(db, run_id, _rows()) + assert len(eval_store.get_rows(db, run_id, category="alpha101")) == 1 + assert len(eval_store.get_rows(db, run_id, search="corr")) == 1 # 搜表达式 + assert len(eval_store.get_rows(db, run_id, search="kmid")) == 1 # 搜因子名 + + +def test_upsert_same_run(tmp_path): + db = str(tmp_path / "factor_eval.db") + eval_store.init_db(db) + run_id = eval_store.create_run(db, label="x", universe="all_a", symbols_count=1, + factors_total=1, start="", end="", params={}) + eval_store.save_results(db, run_id, [_rows()[0]]) + eval_store.save_results(db, run_id, [{**_rows()[0], "metrics": {"1": {"ic_mean": 0.09}}}] * 1) + rows = eval_store.get_rows(db, run_id) + assert len(rows) == 1 and rows[0]["metrics"]["1"]["ic_mean"] == 0.09 +``` + +- [ ] **Step 2: 跑测试确认失败** + +Run: `venv310/bin/python -m pytest tests/factor/test_eval_store.py -v` +Expected: FAIL（ModuleNotFoundError） + +- [ ] **Step 3: 实现** + +```python +# sanguo_factor/eval_store.py +"""批量评估结果落盘 factor_eval.db(eval_runs + eval_results). + +metrics 按周期 {"1": {...}, "5": {...}, "10": {...}, "turnover": float} 存 JSON, +schema 平坦、加周期零迁移。 +""" +import json +import os +import sqlite3 +import uuid +from datetime import datetime + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS eval_runs( + run_id TEXT PRIMARY KEY, + label TEXT NOT NULL, + universe TEXT NOT NULL, + symbols_count INTEGER NOT NULL, + factors_total INTEGER NOT NULL, + start TEXT NOT NULL, + end TEXT NOT NULL, + status TEXT NOT NULL, + created_at TEXT NOT NULL, + finished_at TEXT, + factors_done INTEGER NOT NULL DEFAULT 0, + params_json TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS eval_results( + run_id TEXT NOT NULL, + factor TEXT NOT NULL, + category TEXT NOT NULL, + expression TEXT NOT NULL, + metrics_json TEXT NOT NULL, + PRIMARY KEY(run_id, factor) +); +""" + + +def default_eval_db_path() -> str: + """SANGUO_FACTOR_EVAL_DB 覆盖;默认与 backtest_results.db 同目录.""" + env = os.environ.get("SANGUO_FACTOR_EVAL_DB") + if env: + return env + from sanguo_data.config import load_config, find_config_path + cfg = load_config(find_config_path()) + vnpy_db = cfg.data_paths.get("vnpy_db", "") + return os.path.join(os.path.dirname(os.path.abspath(vnpy_db)), "factor_eval.db") + + +def _conn(path: str) -> sqlite3.Connection: + conn = sqlite3.connect(path, timeout=30) + conn.execute("PRAGMA busy_timeout=30000") + return conn + + +def init_db(path: str) -> None: + os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True) + conn = _conn(path) + try: + conn.executescript(_SCHEMA) + conn.commit() + finally: + conn.close() + + +def create_run(path, label, universe, symbols_count, factors_total, start, end, params) -> str: + run_id = f"ev_{datetime.now():%Y%m%d_%H%M%S}_{uuid.uuid4().hex[:4]}" + conn = _conn(path) + try: + conn.execute( + "INSERT INTO eval_runs VALUES(?,?,?,?,?,?,?,?,?,?,0,?)", + (run_id, label, universe, symbols_count, factors_total, start, end, + "running", datetime.now().isoformat(timespec='seconds'), None, + json.dumps(params, ensure_ascii=False)), + ) + conn.commit() + finally: + conn.close() + return run_id + + +def save_results(path, run_id, rows: list[dict]) -> None: + conn = _conn(path) + try: + conn.executemany( + "INSERT OR REPLACE INTO eval_results VALUES(?,?,?,?,?)", + [(run_id, r["factor"], r["category"], r["expression"], + json.dumps(r["metrics"], ensure_ascii=False)) for r in rows], + ) + cur = conn.execute( + "SELECT COUNT(*) FROM eval_results WHERE run_id=?", (run_id,)) + done = cur.fetchone()[0] + conn.execute("UPDATE eval_runs SET factors_done=? WHERE run_id=?", (done, run_id)) + conn.commit() + finally: + conn.close() + + +def finish_run(path, run_id, status, factors_done) -> None: + conn = _conn(path) + try: + conn.execute( + "UPDATE eval_runs SET status=?, finished_at=?, factors_done=? WHERE run_id=?", + (status, datetime.now().isoformat(timespec='seconds'), factors_done, run_id), + ) + conn.commit() + finally: + conn.close() + + +def list_runs(path) -> list[dict]: + conn = _conn(path) + conn.row_factory = sqlite3.Row + try: + rows = conn.execute( + "SELECT * FROM eval_runs ORDER BY created_at DESC").fetchall() + finally: + conn.close() + return [dict(r) for r in rows] + + +def get_rows(path, run_id, category=None, search=None) -> list[dict]: + q = "SELECT * FROM eval_results WHERE run_id=?" + args: list = [run_id] + if category: + q += " AND category=?" + args.append(category) + if search: + q += " AND (factor LIKE ? OR expression LIKE ?)" + args.extend([f"%{search}%", f"%{search}%"]) + conn = _conn(path) + conn.row_factory = sqlite3.Row + try: + rows = conn.execute(q, args).fetchall() + finally: + conn.close() + return [{**dict(r), "metrics": json.loads(r["metrics_json"])} for r in rows] + + +def get_detail(path, run_id, factor) -> dict | None: + conn = _conn(path) + conn.row_factory = sqlite3.Row + try: + r = conn.execute( + "SELECT * FROM eval_results WHERE run_id=? AND factor=?", + (run_id, factor)).fetchone() + finally: + conn.close() + if r is None: + return None + return {**dict(r), "metrics": json.loads(r["metrics_json"])} +``` + +- [ ] **Step 4: 跑测试确认通过** + +Run: `venv310/bin/python -m pytest tests/factor/test_eval_store.py -v` +Expected: 全 PASS + +- [ ] **Step 5: Commit** + +```bash +git add sanguo_factor/eval_store.py tests/factor/test_eval_store.py +git commit -m "feat(factor): 评估结果存储 factor_eval.db——runs/results两表,metrics按周期JSON,SANGUO_FACTOR_EVAL_DB可覆盖 [vps]" +``` + +--- + +### Task 6: 批量评估引擎（batch_eval） + +**Files:** +- Create: `sanguo_factor/batch_eval.py` +- Test: `tests/factor/test_batch_eval.py` + +**Interfaces:** +- Consumes: Task 2 `mount_all()`/`get_factor`、Task 3 `load_universe_bars/evaluation_filter`、Task 4 `summarize_factor`、Task 5 store API、vnpy `calculate_by_expression` +- Produces: + - `run_batch_eval(factor_names: list[str], start: str, end: str, eval_db: str, label: str, universe: str = "all_a", symbols: list[str] | None = None, limit: int | None = None, cfg=None, progress_cb=None) -> dict` + - 返回 `{"run_id", "factors_total", "factors_done", "errors", "elapsed_sec"}` + - `progress_cb(done: int, total: int, current: str)` 每因子一次（CLI 打印用；API 侧暂不接，YAGNI） + - 单因子计算异常**不中断**：该因子落一行 `metrics={"error": "..."}` 继续 + +- [ ] **Step 1: 写失败测试** + +```python +# tests/factor/test_batch_eval.py +"""批量引擎端到端:合成小库 → 2因子计算 → 指标落盘可查.""" +import sqlite3 +import sys, os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "vnpy_v4.4.0"))) + +import pytest + +from sanguo_factor import alpha_datasets # 挂载 +from sanguo_factor.batch_eval import run_batch_eval +from sanguo_factor import eval_store + +_DDL = """ +CREATE TABLE dbbardata( + symbol TEXT, exchange TEXT, datetime TEXT, interval TEXT, + volume REAL, turnover REAL, open_interest REAL, + open_price REAL, high_price REAL, low_price REAL, close_price REAL) +""" + + +@pytest.fixture(scope="module") +def db(tmp_path_factory): + """3 只 × ~400 交易日(2017-06~2018-12)合成库:一只单调涨/一只震荡/一只反着走.""" + p = tmp_path_factory.mktemp("b") + db = str(p / "qt.db") + conn = sqlite3.connect(db) + conn.execute(_DDL) + import pandas as pd + days = pd.bdate_range("2017-06-01", "2018-12-28") + for i, day in enumerate(days): + d = day.strftime("%Y-%m-%d") + rows = [ + ("600000", "SSE", f"{d} 00:00:00", "d", 100.0, 1_000_000.0, 0, + 10 + i * 0.01, 10 + i * 0.01, 10 + i * 0.01, 10 + i * 0.01), # 单调涨 + ("000001", "SZSE", f"{d} 00:00:00", "d", 100.0, 500_000.0, 0, + 5, 5, 5, 5), # 平盘(截面另一端) + ("300001", "SZSE", f"{d} 00:00:00", "d", 200.0, 900_000.0, 0, + 20 - i * 0.005, 20 - i * 0.005, 20 - i * 0.005, 20 - i * 0.005), # 缓跌 + ] + conn.executemany("INSERT INTO dbbardata VALUES(?,?,?,?,?,?,?,?,?,?,?)", rows) + conn.commit(); conn.close() + return db + + +def test_run_batch_eval_end_to_end(db, tmp_path, monkeypatch): + eval_db = str(tmp_path / "factor_eval.db") + out = run_batch_eval( + factor_names=["ma_20", "roc_5"], start="2018-01-01", end="2018-06-30", + eval_db=eval_db, label="t", symbols=None, cfg=None, vnpy_db_override=db, + ) + assert out["factors_done"] == 2 + assert out["errors"] == [] + rows = eval_store.get_rows(eval_db, out["run_id"]) + assert {r["factor"] for r in rows} == {"ma_20", "roc_5"} + # ma_20 = ts_mean(close,20)/close:涨股该值持续低(均价低于现价) 跌股高 → 与次日收益负相关为主 + m = eval_store.get_detail(eval_db, out["run_id"], "ma_20")["metrics"] + assert "1" in m and "5" in m and "10" in m and "turnover" in m + assert isinstance(m["1"]["ic_mean"], float) + + +def test_bad_factor_recorded_not_fatal(db, tmp_path): + eval_db = str(tmp_path / "factor_eval.db") + out = run_batch_eval( + factor_names=["不存在的因子", "kmid"], start="2018-01-01", end="2018-03-31", + eval_db=eval_db, label="t2", cfg=None, vnpy_db_override=db, + ) + assert out["factors_done"] == 2 + assert len(out["errors"]) == 1 + detail = eval_store.get_detail(eval_db, out["run_id"], "不存在的因子") + assert "error" in detail["metrics"] +``` + +> 注：测试用的 `vnpy_db_override` 参数是**测试注入数据库路径**的后门（生产 cfg 走 `cfg.data_paths["vnpy_db"]`）；实现签名里带上它，默认 None。 + +- [ ] **Step 2: 跑测试确认失败** + +Run: `venv310/bin/python -m pytest tests/factor/test_batch_eval.py -v` +Expected: FAIL（ModuleNotFoundError） + +- [ ] **Step 3: 实现** + +```python +# sanguo_factor/batch_eval.py +"""批量评估引擎:全A bars → 进程内逐因子 calculate_by_expression → 指标 → 落盘. + +不走 AlphaDataset.prepare_data(其 spawn 池对每个表达式 pickle 整个 DataFrame, +9M 行 × 258 因子的传输开销不可接受);calculate_by_expression 纯进程内 polars, +内存随单因子天然有界。现有单因子分析链路(analyzer/alphalens tears)零改动。 +""" +import sys +import os +import time + +_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 pandas as pd +import polars as pl + +from .universe import load_universe_bars, evaluation_filter +from .registry import get_factor +from . import eval_store +from .metrics import summarize_factor + + +def _forward_return_matrices(close_wide: pd.DataFrame, periods=(1, 5, 10)) -> dict[int, pd.DataFrame]: + out = {} + for p in periods: + out[p] = close_wide.shift(-p) / close_wide - 1.0 + return out + + +def run_batch_eval( + factor_names: list[str], + start: str, + end: str, + eval_db: str, + label: str, + universe: str = "all_a", + symbols: list[str] | None = None, + limit: int | None = None, + cfg=None, + progress_cb=None, + vnpy_db_override: str | None = None, +) -> dict: + """跑一轮批量评估,结果增量写入 eval_db,返回摘要.""" + from vnpy.alpha.dataset.utility import calculate_by_expression + + if cfg is None: + from sanguo_data.config import load_config, find_config_path + cfg = load_config(find_config_path()) + vnpy_db = vnpy_db_override or cfg.data_paths["vnpy_db"] + + t0 = time.time() + bars = load_universe_bars(vnpy_db, start, end, symbols=symbols, limit=limit) + if bars.height == 0: + raise ValueError(f"股票池为空: vnpy_db={vnpy_db} window={start}~{end}") + + alpha_df = bars.select(["vt_symbol", "datetime", "open", "high", "low", "close", + "volume", "turnover", "open_interest", "vwap"]) + eval_rows = evaluation_filter(bars, start, end).select(["vt_symbol", "datetime"]) + + close_wide = ( + bars.select(["datetime", "vt_symbol", "close"]) + .pivot(index="datetime", on="vt_symbol", values="close") + .sort("datetime").to_pandas().set_index("datetime") + ) + close_wide.index = pd.to_datetime(close_wide.index) + rets = _forward_return_matrices(close_wide) + + universe_label = universe if symbols is None else "custom" + eval_store.init_db(eval_db) + run_id = eval_store.create_run( + eval_db, label=label, universe=universe_label, + symbols_count=bars["vt_symbol"].n_unique(), factors_total=len(factor_names), + start=start, end=end, params={"limit": limit, "symbols": symbols[:20] if symbols else None}, + ) + + errors: list[str] = [] + done = 0 + buffer: list[dict] = [] + for i, name in enumerate(factor_names): + row = _eval_one(name, alpha_df, eval_rows, rets) + if "error" in row["metrics"]: + errors.append(name) + buffer.append(row) + done += 1 + if len(buffer) >= 20 or done == len(factor_names): + eval_store.save_results(eval_db, run_id, buffer) + buffer = [] + if progress_cb: + progress_cb(done, len(factor_names), name) + + eval_store.finish_run(eval_db, run_id, "done", factors_done=done) + return { + "run_id": run_id, + "factors_total": len(factor_names), + "factors_done": done, + "errors": errors, + "elapsed_sec": round(time.time() - t0, 1), + "symbols_count": int(bars["vt_symbol"].n_unique()), + } + + +def _eval_one(name: str, alpha_df: pl.DataFrame, eval_rows: pl.DataFrame, + rets: dict[int, pd.DataFrame]) -> dict: + factor = get_factor(name) + if factor is None: + return {"factor": name, "category": "unknown", "expression": "", + "metrics": {"error": f"因子未注册: {name}"}} + try: + res = calculate_by_expression(alpha_df, factor["expression"]) + f_long = res.join(eval_rows, on=["vt_symbol", "datetime"], how="inner") + F = ( + f_long.pivot(index="datetime", on="vt_symbol", values="data") + .sort("datetime").to_pandas().set_index("datetime") + ) + F.index = pd.to_datetime(F.index) + metrics = summarize_factor(F, rets[1], rets[5], rets[10]) + return {"factor": name, "category": factor["category"], + "expression": factor["expression"], "metrics": metrics} + except Exception as e: # 单因子失败不拖垮整批 + return {"factor": name, "category": factor["category"], + "expression": factor["expression"], + "metrics": {"error": f"{type(e).__name__}: {e}"}} +``` + +> 实现注意：`res` 是 `calculate_by_expression` 返回的 `datetime/vt_symbol/data` 三列 df；join `eval_rows`（已含窗口+预热过滤）即完成行集裁剪。若 polars pivot 参数名对不上（`on=`/`columns=`），以 `venv310/bin/python -c "import polars as pl; help(pl.DataFrame.pivot)"` 查 1.42 签名为准。 + +- [ ] **Step 4: 跑测试确认通过** + +Run: `venv310/bin/python -m pytest tests/factor/test_batch_eval.py -v` +Expected: 2 个全 PASS（首跑会触发 vnpy alpha import，几秒） + +- [ ] **Step 5: Commit** + +```bash +git add sanguo_factor/batch_eval.py tests/factor/test_batch_eval.py +git commit -m "feat(factor): 批量评估引擎——进程内逐因子calculate_by_expression(弃spawn池整df pickle),单因子失败不中断,结果增量落盘 [vps]" +``` + +--- + +### Task 7: CLI 脚本（scripts/factor_research/run_eval.py） + +**Files:** +- Create: `scripts/factor_research/run_eval.py`（目录新建，无 `__init__.py`，脚本以 `python scripts/factor_research/run_eval.py` 运行，顶部 sys.path 补 repo root——参考 scripts/diag_factor.py 的既有做法，先读它） + +**Interfaces:** +- Produces: CLI，参数 `--start --end --categories alpha101 alpha158 --factors a,b --symbols 600000,000001 --limit 50 --label --db --write-every --list-categories --list-factors [category]` +- 默认：`--start 2018-01-01 --end 2026-06-30 --categories alpha101 alpha158 --label batch1` +- 结束打印 Top 15（按 1D |icir| 降序）小表格 + run_id + +- [ ] **Step 1: 实现（CLI 薄壳，核心逻辑全在 Task 6，无需单测——用 --help + --list-factors 手验）** + +```python +#!/usr/bin/env python +"""批量因子评估 CLI(冒烟/全量;NAS docker exec / 本地直跑). + +用法示例: + 冒烟(50只×2024×10因子): + venv310/bin/python scripts/factor_research/run_eval.py \ + --start 2024-01-01 --end 2024-12-31 --limit 50 --label smoke-2024 \ + --factors alpha2,alpha6,alpha12,alpha18,kmid,klen,roc_5,ma_20,std_20,wvma_20 + 全量(Alpha101+158 × 全A × 8.5年): + venv310/bin/python scripts/factor_research/run_eval.py --label batch1-full +""" +import argparse +import os +import sys + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "vnpy_v4.4.0"))) + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--start", default="2018-01-01") + ap.add_argument("--end", default="2026-06-30") + ap.add_argument("--categories", nargs="*", default=["alpha101", "alpha158"]) + ap.add_argument("--factors", default="", help="逗号分隔,优先于 --categories") + ap.add_argument("--symbols", default="", help="逗号分隔裸代码;空=全A") + ap.add_argument("--limit", type=int, default=None, help="随机抽样 N 只(种子42)") + ap.add_argument("--label", default="batch1") + ap.add_argument("--db", default=None, help="factor_eval.db 路径;默认 default_eval_db_path()") + ap.add_argument("--list-factors", default="", metavar="CATEGORY", help="列出类目因子后退出") + args = ap.parse_args() + + from sanguo_factor.alpha_datasets import mount_all + mount_all() + from sanguo_factor.registry import list_factors + from sanguo_factor.batch_eval import run_batch_eval + from sanguo_factor.eval_store import default_eval_db_path, get_rows + + if args.list_factors: + for f in list_factors(args.list_factors): + print(f"{f['name']:16s} [{f['category']}] {f['expression'][:80]}") + return 0 + + if args.factors: + factor_names = [s.strip() for s in args.factors.split(",") if s.strip()] + else: + factor_names = [f["name"] for c in args.categories for f in list_factors(c)] + if not factor_names: + print("未找到因子(检查 --categories/--factors)", file=sys.stderr) + return 1 + + db_path = args.db or default_eval_db_path() + symbols = [s.strip() for s in args.symbols.split(",") if s.strip()] or None + print(f"[eval] {len(factor_names)} 因子 × {args.start}~{args.end} → {db_path}") + + def _cb(done: int, total: int, current: str) -> None: + print(f"[eval] {done}/{total} {current}", flush=True) + + out = run_batch_eval(factor_names, args.start, args.end, db_path, label=args.label, + symbols=symbols, limit=args.limit, cfg=None, progress_cb=_cb) + + print(f"[eval] run_id={out['run_id']} done={out['factors_done']} " + f"errors={len(out['errors'])} elapsed={out['elapsed_sec']}s symbols={out['symbols_count']}") + if out["errors"]: + print(f"[eval] 失败因子: {', '.join(out['errors'][:20])}") + + rows = get_rows(db_path, out["run_id"]) + scored = [] + for r in rows: + p1 = (r.get("metrics") or {}).get("1") or {} + if p1.get("icir") is not None: + scored.append((abs(p1["icir"]), r["factor"], p1)) + scored.sort(reverse=True) + print(f"\n{'因子':<14s} {'IC':>8s} {'ICIR':>8s} {'t':>8s} {'胜率':>7s} 结论") + for _, name, p1 in scored[:15]: + print(f"{name:<14s} {p1['ic_mean']:>8.4f} {p1['icir']:>8.3f} " + f"{p1['t_stat']:>8.2f} {p1['win_rate']:>7.1%} {p1['conclusion']}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) +``` + +- [ ] **Step 2: 手验** + +Run: `venv310/bin/python scripts/factor_research/run_eval.py --list-factors alpha101 | head -5 && venv310/bin/python scripts/factor_research/run_eval.py --help` +Expected: 列出 alpha1..alpha5；--help 正常 + +- [ ] **Step 3: Commit** + +```bash +git add scripts/factor_research/run_eval.py +git commit -m "feat(factor): 批量评估CLI——冒烟/全量/列因子,进度打印+Top15汇总表(NAS docker exec 冒烟入口) [vps]" +``` + +--- + +### Task 8: API 端点 + orchestrator 接线 + +**Files:** +- Modify: `sanguo_api/schemas.py`（追加 FactorBatchEvalRequest） +- Create: `sanguo_api/routes_factor.py` +- Modify: `sanguo_api/app.py`（注册路由 + set_eval_db_path 接线） +- Modify: `sanguo_orchestrator/runner.py`（追加 submit_batch_eval/_batch_eval_worker/_persist_factor_eval） +- Test: `tests/api/test_factor_eval_routes.py` + +**Interfaces:** +- Consumes: Task 5 `eval_store`、Task 6 `run_batch_eval`、Task 2 挂载 +- Produces: + - `GET /factor/eval/runs` → `{"runs": [...]}` + - `GET /factor/eval/leaderboard?run_id=&period=1|5|10&category=&search=&sort=abs_icir|ic_mean|t_stat|ls_annual&order=desc` → `{"tiles": {"factors_total","effective","watch","top_ls_annual","top_ls_factor"}, "rows": [...]}`（row = 扁平化所选周期：`factor,category,expression,ic_mean,icir,t_stat,win_rate,ls_annual,turnover,conclusion,monthly_ic`；run_id 缺省 = 最新 run） + - `GET /factor/eval/detail?run_id=&factor=` → `{"factor","category","expression","metrics"}`（全周期） + - `POST /factor/eval/submit` body `{"categories":["alpha101"],"factors":[],"symbols":[],"start":"2018-01-01","end":"2026-06-30","label":"batch1"}` → `{"task_id": "factor_eval_xxxx"}` + - Orchestrator: `async submit_batch_eval(factor_names, start, end, symbols, label) -> task_id`；完成时 `_persist_factor_eval` 落 backtest_results.db（type="factor_eval", statistics={"run_id", "factors_done", "errors"}）使任务中心可见 + - `set_eval_db_path(path)` 由 app.py 在启动时调（路径 = dirname(orchestrator db_path)/factor_eval.db） + +- [ ] **Step 1: 写失败测试** + +```python +# tests/api/test_factor_eval_routes.py +"""/factor/eval/* 端点:leaderboard 扁平化/tiles/排序/详情/提交校验.""" +import sys, os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from sanguo_factor import eval_store +from sanguo_api import routes_factor + + +@pytest.fixture() +def client(tmp_path, monkeypatch): + db = str(tmp_path / "factor_eval.db") + eval_store.init_db(db) + routes_factor.set_eval_db_path(db) + app = FastAPI() + app.include_router(routes_factor.router) + return TestClient(app) + + +def _seed(db): + run_id = eval_store.create_run(db, label="冒烟", universe="custom", symbols_count=50, + factors_total=3, start="2024-01-01", end="2024-12-31", params={}) + eval_store.save_results(db, run_id, [ + {"factor": "alpha2", "category": "alpha101", "expression": "e2", + "metrics": {"turnover": 0.2, "1": {"ic_mean": 0.05, "icir": 0.4, "t_stat": 3.0, + "win_rate": 0.6, "ls_annual": 0.12, "conclusion": "effective", + "monthly_ic": [{"month": "2024-01", "ic": 0.05}]}, + "5": {}, "10": {}}}, + {"factor": "kmid", "category": "alpha158", "expression": "ek", + "metrics": {"turnover": 0.5, "1": {"ic_mean": -0.01, "icir": -0.05, "t_stat": -0.4, + "win_rate": 0.45, "ls_annual": -0.02, "conclusion": "eliminated", + "monthly_ic": []}, "5": {}, "10": {}}}, + {"factor": "alpha6", "category": "alpha101", "expression": "e6", + "metrics": {"turnover": 0.3, "1": {"ic_mean": 0.03, "icir": 0.2, "t_stat": 1.7, + "win_rate": 0.55, "ls_annual": 0.06, "conclusion": "watch", + "monthly_ic": []}, "5": {}, "10": {}}}, + ]) + eval_store.finish_run(db, run_id, "done", 3) + return run_id + + +def test_leaderboard_default_latest_run(client): + _seed(routes_factor._eval_db_path) + r = client.get("/factor/eval/leaderboard") + assert r.status_code == 200 + body = r.json() + assert body["tiles"]["factors_total"] == 3 + assert body["tiles"]["effective"] == 1 + assert body["tiles"]["watch"] == 1 + assert body["tiles"]["top_ls_factor"] == "alpha2" + # 默认 abs_icir desc: alpha2(0.4) > alpha6(0.2) > kmid(0.05) + assert [x["factor"] for x in body["rows"]] == ["alpha2", "alpha6", "kmid"] + row = body["rows"][0] + for key in ("factor", "category", "expression", "ic_mean", "icir", "t_stat", + "win_rate", "ls_annual", "turnover", "conclusion", "monthly_ic"): + assert key in row + + +def test_leaderboard_filter_and_period(client): + _seed(routes_factor._eval_db_path) + r = client.get("/factor/eval/leaderboard", params={"category": "alpha158"}) + assert [x["factor"] for x in r.json()["rows"]] == ["kmid"] + r = client.get("/factor/eval/leaderboard", params={"search": "alpha6"}) + assert [x["factor"] for x in r.json()["rows"]] == ["alpha6"] + + +def test_detail(client): + _seed(routes_factor._eval_db_path) + r = client.get("/factor/eval/detail", params={"factor": "kmid"}) + assert r.status_code == 200 + assert r.json()["category"] == "alpha158" + assert "1" in r.json()["metrics"] + assert client.get("/factor/eval/detail", params={"factor": "nope"}).status_code == 404 + + +def test_runs_list(client): + _seed(routes_factor._eval_db_path) + r = client.get("/factor/eval/runs") + assert len(r.json()["runs"]) == 1 + + +def test_submit_validation_error(client): + # categories 与 factors 双空 → 422/400 + r = client.post("/factor/eval/submit", + json={"start": "2018-01-01", "end": "2026-06-30"}) + assert r.status_code in (400, 422) +``` + +- [ ] **Step 2: 跑测试确认失败** + +Run: `venv310/bin/python -m pytest tests/api/test_factor_eval_routes.py -v` +Expected: FAIL（no module routes_factor） + +- [ ] **Step 3: 实现** + +`sanguo_api/schemas.py` 末尾追加： + +```python +class FactorBatchEvalRequest(BaseModel): + """Factor batch evaluation request""" + categories: list[str] = [] + factors: list[str] = [] + symbols: list[str] = [] + start: str + end: str + label: str = "" +``` + +`sanguo_api/routes_factor.py`： + +```python +"""因子批量评估 API:runs/leaderboard/detail/submit.""" +from fastapi import APIRouter, HTTPException, Depends, Query + +from .schemas import FactorBatchEvalRequest +from .auth import verify_token as verify_token_impl + +router = APIRouter(dependencies=[Depends(verify_token_impl)]) + +_eval_db_path: str | None = None + +_SORT_KEYS = { + "abs_icir": lambda p: abs(p.get("icir") or 0.0), + "ic_mean": lambda p: p.get("ic_mean") if p.get("ic_mean") is not None else float("-inf"), + "t_stat": lambda p: abs(p.get("t_stat") or 0.0), + "ls_annual": lambda p: p.get("ls_annual") if p.get("ls_annual") is not None else float("-inf"), +} + + +def set_eval_db_path(path: str) -> None: + global _eval_db_path + _eval_db_path = path + + +def _db() -> str: + from sanguo_factor.eval_store import default_eval_db_path + return _eval_db_path or default_eval_db_path() + + +def _latest_run_id(db: str) -> str | None: + from sanguo_factor.eval_store import list_runs + runs = list_runs(db) + return runs[0]["run_id"] if runs else None + + +@router.get("/factor/eval/runs") +def eval_runs(): + from sanguo_factor.eval_store import list_runs + return {"runs": list_runs(_db())} + + +@router.get("/factor/eval/leaderboard") +def eval_leaderboard( + run_id: str | None = None, + period: str = Query("1", pattern="^(1|5|10)$"), + category: str | None = None, + search: str | None = None, + sort: str = Query("abs_icir", pattern="^(abs_icir|ic_mean|t_stat|ls_annual)$"), + order: str = Query("desc", pattern="^(asc|desc)$"), +): + from sanguo_factor.eval_store import get_rows + db = _db() + rid = run_id or _latest_run_id(db) + if not rid: + return {"tiles": {"factors_total": 0, "effective": 0, "watch": 0, + "top_ls_annual": None, "top_ls_factor": None}, "rows": []} + rows = get_rows(db, rid, category=category, search=search) + + flat: list[dict] = [] + for r in rows: + m = r.get("metrics") or {} + p = m.get(period) or {} + if "error" in m and not p: + flat.append({"factor": r["factor"], "category": r["category"], + "expression": r["expression"], "error": m["error"], + "ic_mean": None, "icir": None, "t_stat": None, "win_rate": None, + "ls_annual": None, "turnover": m.get("turnover"), + "conclusion": "eliminated", "monthly_ic": []}) + continue + flat.append({ + "factor": r["factor"], "category": r["category"], "expression": r["expression"], + "ic_mean": p.get("ic_mean"), "icir": p.get("icir"), "t_stat": p.get("t_stat"), + "win_rate": p.get("win_rate"), "ls_annual": p.get("ls_annual"), + "turnover": m.get("turnover"), "conclusion": p.get("conclusion", "eliminated"), + "monthly_ic": p.get("monthly_ic", []), + }) + + keyfn = _SORT_KEYS[sort] + flat.sort(key=lambda x: (keyfn(x) is not None, keyfn(x)), reverse=(order == "desc")) + for i, row in enumerate(flat, 1): + row["rank"] = i + + scored = [x for x in flat if x.get("ic_mean") is not None] + top_ls = max(scored, key=lambda x: (x.get("ls_annual") or float("-inf")), default=None) + tiles = { + "factors_total": len(flat), + "effective": sum(1 for x in flat if x["conclusion"] == "effective"), + "watch": sum(1 for x in flat if x["conclusion"] == "watch"), + "top_ls_annual": top_ls["ls_annual"] if top_ls else None, + "top_ls_factor": top_ls["factor"] if top_ls else None, + } + return {"tiles": tiles, "rows": flat} + + +@router.get("/factor/eval/detail") +def eval_detail(run_id: str | None = None, factor: str = ""): + from sanguo_factor.eval_store import get_detail + db = _db() + rid = run_id or _latest_run_id(db) + d = get_detail(db, rid, factor) if rid else None + if d is None: + raise HTTPException(status_code=404, detail="factor not found in run") + return d + + +@router.post("/factor/eval/submit") +async def eval_submit(req: FactorBatchEvalRequest): + if not req.categories and not req.factors: + raise HTTPException(status_code=400, detail="categories 与 factors 至少给一个") + from .routes import get_orchestrator + orch = get_orchestrator() + if orch is None: + raise HTTPException(status_code=503, detail="orchestrator 未就绪") + task_id = await orch.submit_batch_eval( + factor_names=req.factors, categories=req.categories, + start=req.start, end=req.end, symbols=req.symbols or None, label=req.label, + ) + return {"task_id": task_id} +``` + +`sanguo_orchestrator/runner.py`（追加；`_batch_eval_worker` 必须**模块级**——spawn 进程池按引用 pickle）： + +```python +def _batch_eval_worker(factor_names: list, categories: list, start: str, end: str, + symbols, label: str, eval_db: str) -> dict: + """进程池 worker:批量评估(spawn 环境,自行 load config/挂载因子).""" + from sanguo_factor.alpha_datasets import mount_all + mount_all() + if not factor_names: + from sanguo_factor.registry import list_factors + factor_names = [f["name"] for c in categories for f in list_factors(c)] + from sanguo_factor.batch_eval import run_batch_eval + return run_batch_eval(factor_names, start, end, eval_db, label=label, symbols=symbols) +``` + +`Orchestrator` 类内追加（仿 submit_factor）： + +```python + async def submit_batch_eval(self, factor_names: list, categories: list, + start: str, end: str, symbols, label: str) -> str: + """Submit a factor batch evaluation task asynchronously""" + from .eval_db_path import resolve_eval_db # 或直接内联(见下) + task_id = f"factor_eval_{uuid.uuid4().hex[:8]}" + self._record_submit(task_id) + self.pool.submit(task_id, "factor_eval") + eval_db = resolve_eval_db(self.db_path) + fut: Future = self.pool.submit_work( + task_id, _batch_eval_worker, factor_names, categories, + start, end, symbols, label or "batch", eval_db, + ) + task = self.pool.get_task(task_id) + task.start() + await self._notify_stage(task_id, "批量评估中") + asyncio.ensure_future(self._wait_future(task_id, fut)) + return task_id +``` + +`resolve_eval_db` 别单独建文件——放 runner.py 模块级函数： + +```python +def resolve_eval_db(backtest_db_path: str) -> str: + """factor_eval.db 与 backtest_results.db 同目录(SANGUO_FACTOR_EVAL_DB 可覆盖).""" + import os + env = os.environ.get("SANGUO_FACTOR_EVAL_DB") + if env: + return env + return os.path.join(os.path.dirname(os.path.abspath(backtest_db_path or ".")), "factor_eval.db") +``` + +完成回写（在 `_wait_future` 完成分支里既有 `if getattr(result, "ic_summary", ...)` 之后并列加）： + +```python + # 批量评估结果摘要落任务中心(完整数据在 factor_eval.db) + if isinstance(result, dict) and "run_id" in result: + self._persist_factor_eval(task_id, result) +``` + +```python + def _persist_factor_eval(self, task_id: str, summary: dict) -> None: + """批量评估摘要落 backtest_results.db(type=factor_eval)进任务列表.""" + from sanguo_backtest.result_store import save_result, BacktestResult + try: + save_result(BacktestResult( + task_id=task_id, type="factor_eval", status="done", + strategy=summary.get("label", "batch"), + symbol=f"{summary.get('symbols_count', 0)}只", + params={}, + start="", end="", + statistics={"run_id": summary.get("run_id"), + "factors_done": summary.get("factors_done"), + "errors": summary.get("errors", [])[:20]}, + equity_curve=None, trades=None, + ), db_path=self.db_path) + except Exception as e: + import logging + logging.getLogger(__name__).warning("persist factor_eval %s failed: %s", task_id, e) +``` + +> ⚠️ 先读 `_wait_future` 实际代码再插入——上面片段是意图，插入点以「worker 返回 dict 且带 run_id」为判据，别破坏既有 FactorReport/BacktestResult 分支。同样先读 `BacktestResult` dataclass 字段（sanguo_backtest/result_store.py）确认字段名一致。 + +`sanguo_api/app.py`：import 区加 `from .routes_factor import router as factor_eval_router, set_eval_db_path as set_factor_eval_db_path`；`include_router` 区加一行；创建 app/orchestrator 初始化处（找到 orchestrator db_path 设定的地方）加： + +```python +from sanguo_orchestrator.runner import resolve_eval_db +set_factor_eval_db_path(resolve_eval_db(orchestrator_db_path)) +``` + +> 先读 app.py 全文找现有 wiring 位置（routes_paper 的 set_db_path 怎么接就怎么接）。 + +- [ ] **Step 4: 跑测试确认通过 + 回归** + +Run: `venv310/bin/python -m pytest tests/api/test_factor_eval_routes.py tests/api/test_factor_routes.py tests/factor -v` +Expected: 全 PASS（既有 /factor/list、/factor/analyze 测试不回归） + +- [ ] **Step 5: Commit** + +```bash +git add sanguo_api/schemas.py sanguo_api/routes_factor.py sanguo_api/app.py sanguo_orchestrator/runner.py tests/api/test_factor_eval_routes.py +git commit -m "feat(api): 因子批量评估四端点(runs/leaderboard/detail/submit)+orchestrator任务接线——leaderboard服务端扁平化排序,worker落factor_eval.db+任务中心摘要 [vps]" +``` + +--- + +### Task 9: 前端接线（api/router/menu） + +**Files:** +- Modify: `frontend/src/api/factor.ts` +- Modify: `frontend/src/router/index.ts:25-27` 区域 +- Modify: `frontend/src/views/Layout.vue:31-33`（面包屑）与 `:69`（导航数组）与 `:171-174`（菜单） + +**Interfaces:** +- Produces（供 Task 10/11/12 消费的 TS API）: + +```typescript +// —— 追加到 frontend/src/api/factor.ts —— + +export interface EvalRun { + run_id: string + label: string + universe: string + symbols_count: number + factors_total: number + factors_done: number + start: string + end: string + status: string + created_at: string +} + +export interface MonthlyIcPoint { month: string; ic: number } + +export interface LeaderboardRow { + rank: number + factor: string + category: string + expression: string + ic_mean: number | null + icir: number | null + t_stat: number | null + win_rate: number | null + ls_annual: number | null + turnover: number | null + conclusion: string + monthly_ic: MonthlyIcPoint[] + error?: string +} + +export interface LeaderboardTiles { + factors_total: number + effective: number + watch: number + top_ls_annual: number | null + top_ls_factor: string | null +} + +export interface EvalDetail { + factor: string + category: string + expression: string + metrics: Record +} + +export interface BatchEvalSubmit { + categories: string[] + factors: string[] + symbols: string[] + start: string + end: string + label: string +} + +export async function getEvalRuns(): Promise { + const { data } = await apiClient.get<{ runs: EvalRun[] }>('/factor/eval/runs') + return data.runs +} + +export async function getLeaderboard(params: { + run_id?: string + period?: string + category?: string + search?: string + sort?: string + order?: string +}): Promise<{ tiles: LeaderboardTiles; rows: LeaderboardRow[] }> { + const { data } = await apiClient.get('/factor/eval/leaderboard', { params }) + return data +} + +export async function getEvalDetail(factor: string, runId?: string): Promise { + const { data } = await apiClient.get('/factor/eval/detail', { + params: { factor, ...(runId ? { run_id: runId } : {}) }, + }) + return data +} + +export async function submitBatchEval(req: BatchEvalSubmit): Promise { + const { data } = await apiClient.post<{ task_id: string }>('/factor/eval/submit', req) + return data.task_id +} +``` + +`router/index.ts` 在 `factor/new` 之前插三行： + +```typescript + { path: 'factor/leaderboard', name: 'fc-leaderboard', component: () => import('@/views/factor/Leaderboard.vue') }, + { path: 'factor/leaderboard/:factor', name: 'fc-leaderboard-detail', component: () => import('@/views/factor/LeaderboardDetail.vue') }, + { path: 'factor/batch', name: 'fc-batch', component: () => import('@/views/factor/BatchEval.vue') }, +``` + +`Layout.vue`： +- 面包屑数组（31-33 行区域）加： + +```typescript + { match: /^\/factor\/leaderboard\/.+$/, group: '投研', title: '因子详情' }, + { match: /^\/factor\/leaderboard$/, group: '投研', title: 'IC 排行榜' }, + { match: /^\/factor\/batch$/, group: '投研', title: '批量评估' }, +``` + +- 侧边导航数组（69 行区域）加： + +```typescript + { label: 'IC 排行榜', group: '投研', path: '/factor/leaderboard' }, + { label: '批量评估', group: '投研', path: '/factor/batch' }, +``` + +- 菜单（171-174）改为： + +```html + + + IC 排行榜 + 批量评估 + 因子分析 + +``` + +> 注意：先 Read Layout.vue 全文确认数组结构/键名与上面一致再改（摘要行号可能漂移）。此 task 页面文件尚不存在，router 懒加载不会构建失败，但**本 task 不 build**，build 统一放 Task 12。 + +- [ ] **Step 1: 改四个文件**（如上） +- [ ] **Step 2: Commit** + +```bash +git add frontend/src/api/factor.ts frontend/src/router/index.ts frontend/src/views/Layout.vue +git commit -m "feat(web): 因子工作台接线——eval API 客户端+3路由+投研组菜单/面包屑 [nas]" +``` + +--- + +### Task 10: IC 排行榜页（Leaderboard.vue） + +**Files:** +- Create: `frontend/src/views/factor/Leaderboard.vue` + +**Interfaces:** +- Consumes: Task 9 的 `getEvalRuns/getLeaderboard`、路由 `fc-leaderboard-detail`（行点击 `router.push(`/factor/leaderboard/${factor}?run=${runId}`)`） +- 视觉：**逐区块对照 mockup**（`docs/factor_research/factor-workbench-mockup.html` 视图①「IC 排行榜」）复刻——4 统计瓦片/筛选栏（run 下拉+类别 chips+搜索框+1D5D10D 切换）/主表格（IC 均值列内嵌横向色条 div 宽度=|ic_mean|×500%、红正绿负按 `--up/--down`）/IC 趋势 sparkline（inline SVG polyline，取 monthly_ic 后 24 点，正段青负段按 up/down）/信号灯（●`--lamp-ok`=effective ●`--lamp-warn`=watch ●`--lamp-idle`=eliminated） + +完整组件骨架（样式类名从 mockup 对应区块取值，token 全走 CSS 变量）： + +```vue + +``` + +template 结构（视觉细节照 mockup）：页面标题区 → 4 瓦片（`tiles.factors_total/effective/watch/top_ls_annual`+`top_ls_factor`）→ 筛选栏（el-select run / chips / el-input search（回车触发 load）/ 1D·5D·10D segmented）→ 表格（列：排名/因子名（副行表达式截断 48 字符）/类别 chip/IC 均值+色条/ICIR/t 值/胜率/多空年化（红正绿跌色）/换手/IC 趋势 svg/信号灯+结论字）→ 行 click 跳详情；`loading` 时 el-loading / 空态「暂无评估结果，去批量评估发起」链接 `/factor/batch`。 + +- [ ] **Step 1: 写组件（对照 mockup 逐区块复刻）** +- [ ] **Step 2: 手验**：`cd frontend && npm run dev` 登录后访问 `/factor/leaderboard`（空态可见、筛选交互正常、无 console 报错） +- [ ] **Step 3: Commit** + +```bash +git add frontend/src/views/factor/Leaderboard.vue +git commit -m "feat(web): IC排行榜页——瓦片+筛选chips+全指标表格+IC趋势sparkline+信号灯,终端风1:1 mockup [nas]" +``` + +--- + +### Task 11: 因子详情页（LeaderboardDetail.vue） + +**Files:** +- Create: `frontend/src/views/factor/LeaderboardDetail.vue` + +**Interfaces:** +- Consumes: `getEvalDetail`、route param `factor`、query `run`；周期切换本地态（默认 1） +- Produces: 图表用 echarts（平台既有模式：`import * as echarts from 'echarts'` 或 composables/useChart.ts + `@/utils/echartsDark` 的 darkTitle/darkTooltip/darkGrid/darkAxis，参考 `frontend/src/components/charts/EquityChart.vue` 的用法） + +页面结构（对照 mockup 视图②）： +1. 返回链接 ← IC 排行榜 + 因子名标题 + 类别 chip + 信号灯 +2. 表达式卡片：`--panel` 底、青色 mono、可展开全文 +3. 6 格指标条（当前周期）：IC 均值 / ICIR / t 值 / 胜率 / 多空年化 / 换手率 + 参评天数（`metrics[period].count`） +4. 月度 IC 柱状 + 累计 IC 折线（双 y 或同图双系列 echarts：柱=月度 IC 正红负绿、线=累计 IC 青 `#00e5ff`） +5. 十分组年化收益柱状（D1–D10，色阶从 `--down` 渐到 `--up`） +6. tears 入口：跳既有 `/factor/new`（提示「 Tears 报告请从因子分析页发起」按钮）——v1 不做深链 +7. 周期切换（1D/5D/10D）重渲染 3/4/5 区块 + +```vue + +``` + +- [ ] **Step 1: 写组件** +- [ ] **Step 2: 手验**：dev server 从排行榜点行进入，三区块渲染、周期切换重绘、无 console 报错 +- [ ] **Step 3: Commit** + +```bash +git add frontend/src/views/factor/LeaderboardDetail.vue +git commit -m "feat(web): 因子详情页——表达式卡+6格指标+月度/累计IC双系列图+十分组柱状,echarts终端暗色 [nas]" +``` + +--- + +### Task 12: 批量评估页 + 前端构建门禁（BatchEval.vue + build + CI 扩） + +**Files:** +- Create: `frontend/src/views/factor/BatchEval.vue` +- Modify: `.gitea/workflows/ci-cd.yml:57`（pytest 命令加 `tests/factor`） + +**Interfaces:** +- Consumes: `submitBatchEval`、`getFactors`（类目清单）、既有任务中心路由（提交后跳 `/backtest/history`） +- 页面结构（对照 mockup 视图③）：01 股票池（3 卡：全A `''` /自定义 symbols textarea——中证800 预设 v1 置灰「未就绪」）→ 02 因子集（按 category 分组勾选卡：alpha101/alpha158/builtin，显示每类因子数，全选/清空）→ 03 时间窗（el-date-picker 起止，默认 2018-01-01 / 2026-06-30；提示文案「全量约 4400 只 × 8.5 年，预计 1–3 小时，完成后结果进 IC 排行榜」）→ 提交按钮（校验：因子集非空）→ 成功 toast + 跳任务中心 +- CI：`pytest tests/data_platform tests/portfolio tests/factor -q`（tests/factor 全是纯逻辑小测，进 CI 门禁防回归） + +- [ ] **Step 1: 写组件（如上结构，el-card 分步布局 + 终端风覆盖样式照平台既有表单页如 backtest/New.vue 的覆盖写法）** +- [ ] **Step 2: 完整构建验证（必做，别只 dev）** + +Run: `cd frontend && npm run build` +Expected: vue-tsc + vite build 零 error（warning 可容忍）；失败先修再提交 + +- [ ] **Step 3: 后端全量回归 + CI 文件改** + +Run: `venv310/bin/python -m pytest tests/factor tests/api -q` +Expected: 全 PASS + +- [ ] **Step 4: Commit** + +```bash +git add frontend/src/views/factor/BatchEval.vue .gitea/workflows/ci-cd.yml +git commit -m "feat(web): 批量评估页——股票池/因子集/时间窗三步提交进任务中心;CI门禁加tests/factor [nas]" +``` + +--- + +### Task 13: 部署 NAS + 冒烟 + 全量 + 验证 + +**Files:** 无新文件（操作任务） + +- [ ] **Step 1: push 触发 CI** + +```bash +git push origin master +``` + +盯 `http://192.168.2.154:3000/sanguo/sanguo_vnpy_v2/actions`：test（pytest 含新 tests/factor）→ nas-deploy → nas-verify 全绿。红了就地修（Mac runner 睡眠会导致不跑——确认 caffeinate 在跑）。 + +- [ ] **Step 2: NAS 容器内冒烟（50 只 × 2024 × 10 因子）** + +```bash +ssh sanguo-nas '/var/packages/Docker/target/usr/bin/docker exec sanguo_vnpy_v2 \ + python scripts/factor_research/run_eval.py \ + --start 2024-01-01 --end 2024-12-31 --limit 50 --label smoke-2024 \ + --factors alpha2,alpha6,alpha12,alpha18,kmid,klen,roc_5,ma_20,std_20,wvma_20 \ + --db /volume1/stock/sanguo_vnpy_v2/data/factor_eval.db' +``` + +Expected: `[eval] run_id=ev_... done=10 errors=0`，Top15 表格打印，|IC| 普遍 < 0.1。记下 elapsed 秒数（全量外推依据）。 + +> 冒烟 db 路径显式给 `/volume1/stock/sanguo_vnpy_v2/data/factor_eval.db`（容器挂载卷内、API 容器同进程可见的持久位置）。若容器内该目录不存在，先 `docker exec ... mkdir -p`。**default_eval_db_path() 在容器里指向哪要先验证**：`docker exec sanguo_vnpy_v2 python -c "from sanguo_factor.eval_store import default_eval_db_path; print(default_eval_db_path())"`——如果和上面不一致，统一以 `SANGUO_FACTOR_EVAL_DB=/volume1/stock/sanguo_vnpy_v2/data/factor_eval.db` 环境变量注入容器（改 NAS 容器 env 或 app.py 启动处，优先不动 app.py，用 NAS 侧 env）。 + +- [ ] **Step 3: API 验证** + +```bash +TOKEN=$(curl -s -X POST http://192.168.2.154:8000/api/v1/auth/login -H 'Content-Type: application/json' \ + -d '{"username":"admin","password":"<从 .claude/gitea-config.json 或询问用户>"}' | python3 -c 'import json,sys;print(json.load(sys.stdin)["token"])') +curl -s "http://192.168.2.154:8000/api/v1/factor/eval/runs" -H "Authorization: Bearer $TOKEN" +curl -s "http://192.168.2.154:8000/api/v1/factor/eval/leaderboard" -H "Authorization: Bearer $TOKEN" | python3 -m json.tool | head -40 +``` + +Expected: runs 含 smoke-2024（status=done）；leaderboard tiles.factors_total=10、rows 10 行、无 error 字段。 + +- [ ] **Step 4: 前端页面验证（chrome-devtools MCP）** + +打开 `http://192.168.2.154:8000/`（或前端实际端口，问 runbook/nas-verify 用的地址）→ 登录 → 投研→IC 排行榜：冒烟 run 可选、瓦片数字、表格 10 行、信号灯正常 → 点行进详情：三图表渲染 → 批量评估页：表单结构完整。截图留档。 + +- [ ] **Step 5: 全量评估（走 API 任务中心，非 docker exec——长任务铁律）** + +```bash +curl -s -X POST http://192.168.2.154:8000/api/v1/factor/eval/submit \ + -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' \ + -d '{"categories":["alpha101","alpha158"],"factors":[],"symbols":[], + "start":"2018-01-01","end":"2026-06-30","label":"batch1-full"}' +``` + +Expected: `{"task_id":"factor_eval_..."}`；任务中心出现任务、stage=批量评估中。预计 1–3 小时（以冒烟实测外推），期间 Monitor 轮询： + +```bash +# 每 10 分钟查一次进度(eval_runs.factors_done 递增 = 活着) +ssh sanguo-nas '/var/packages/Docker/target/usr/bin/docker exec sanguo_vnpy_v2 python -c " +import sqlite3; c=sqlite3.connect(\"/volume1/stock/sanguo_vnpy_v2/data/factor_eval.db\"); +print(c.execute(\"SELECT run_id,factors_done,factors_total,status FROM eval_runs ORDER BY created_at DESC LIMIT 1\").fetchone())"' +``` + +- [ ] **Step 6: 全量结果验收** + +完成后 leaderboard：tiles.factors_total=258、effective/watch 有分布（全 0 effective = 数据或口径有问题要排查）；抽 3 个知名因子（alpha101 的 alpha6/alpha12、alpha158 的 ma_20）人工核对 IC 符号量级是否合理（量价反转类因子 A 股 RankIC 1D 通常 |0.03|~|0.1|）。 + +- [ ] **Step 7: 收尾汇报** + +- `git log --oneline` 列出全部 commit（确认每条有标签） +- 向用户汇报：冒烟/全量耗时、Top 因子表、有效/观察/淘汰分布、ST 过滤偏差说明、待推 VPS 清单（按 CLAUDE.md 铁律，**等用户原话「推vps」**） +- 更新 memory `factor-session-mainline.md` + +--- + +## Self-Review 结论 + +- **Spec 覆盖**：挂载(T2)/vwap(T1)/批量脚本(T6,7)/冒烟先行(T13S2)/缓存扩容→**不需要**（批量路径绕开 AlphaLabSession 的 50 只上限，spec 第 5 条「股票池缓存扩容」由新路径天然满足，零改动）/新 API(T8)/前端三页(T9-12)/验收标准 5 条对应 T13 S2-S6 +- **偏差（已声明）**：ST 过滤不做（前视偏差理由）；「SQLite 或 CSV+markdown」取 SQLite 单方案 +- **类型一致性**：metrics/period key "1"/"5"/"10" 全链一致；run_id 贯穿 store/API/前端 query +- **排序稳定性**：leaderboard 默认 abs_icir desc，rank 服务端生成