18 KiB
因子日度/月度双层分工 B 班实现计划
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: 落地 2026-10-11 拍板的双层分工补丁（spec 4b4fc16 设计已载）：①速览读最新（icStats/icFullT 数据源=根层∪daily/ 取 as_of 新者，判定层不动）②月度链①段复用同 as_of 日度批（检测存在→跳过重算）。
Architecture: B1 只改速览读取面——_monthly_reports() 加 include_daily 参数合并 daily/ 子目录件，factors()/factor_detail() 两个速览端点启用，判定层三端点（monthly-review/ic-trend/graveyard）调用形态不变仍只读根层。B2 在 monthly_batch.main() 月度模式下、run_batch_eval 之前检测 eval_db 已有「end=月末 且 label=daily_{月末} 且因子集一致且完成」的日度批→跳过 stage1 重算与 ic_gate，只留复用日志。
Tech Stack: FastAPI + pytest（tests/api），Python CLI + pytest（tests/factor），Vue3 + vitest（标签一处）。
Global Constraints
- 判定层读取路径一行不动：
monthly_review_view/factor_ic_trend（衰减判定/研判卡/墓园触发语义=月末快照，只读根层）。它们对_monthly_reports()的调用不得传新参。 - daily/ 子目录件与根层件同命名
{host}_{as_of}.json同 schema（run_factor_daily_wrapper.ps1 用同一 monthly_review.main 落盘）。 - 测试触及 factor registry 必须
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY", <tmp>)（既有 conftest 模式）。 - 前端验证权威=
cd frontend && npm run build（vue-tsc -b 含 spec 文件）；vitest 同跑。 - 本仓 commit 不加 Co-Authored-By；commit subject 行尾打标签（本班两 task 均为纯实现，设计已在
4b4fc16落档 → 标签=[vps] [no-doc]，两标签都在 subject 行尾）。 - B2 复用条件必须同时满足：
label == f"daily_{plan['end']}"、end == plan['end']、factors_total == len(plan['factor_names'])且factors_done == factors_total（完整批），并经eval_store.get_rows校验因子名集合与 plan 集合一致——因子集漂移（日度批后注册了新因子）时必须回退自算，不许部分复用。 - B2 跳过时 stdout 打一行
[monthly_batch] 复用日度批 run_id=... label=...（决议H:物理分开≠逻辑重叠,重算白烧）,跳过 stage1 重算与 ic_gate；exit code 0 不变。
Task 1: 速览读最新——icStats/icFullT 根层∪daily/ 取 as_of 新者
Files:
- Modify:
sanguo_api/routes_pipeline.py（_monthly_reports~:602、factors~:547、factor_detail~:722） - Modify:
frontend/src/views/pipeline/FactorDetail.vue:151 - Test:
tests/api/test_routes_pipeline_factors.py（追加用例）
Interfaces:
-
Consumes:
_monthly_dir()（envSANGUO_FACTOR_MONTHLY_DIR可覆写，测试用） -
Produces:
_monthly_reports(include_daily: bool = False)——判定层三处不传参=行为不变；速览两处传include_daily=True -
Step 1: 写失败测试（后端）
在 tests/api/test_routes_pipeline_factors.py 末尾追加（沿用 test_factor_detail_ic_stats 的既有模式——_seed_factor_card 建 fa_x、SANGUO_FACTOR_MONTHLY_DIR 指 tmp）：
def test_factor_detail_ic_stats_prefers_newer_daily(client, tmp_path, monkeypatch):
"""2026-10-11 双层分工补丁:icStats 读根层∪daily/ 取 as_of 新者——
根层 10-04 旧 schema 无 factor_stats,daily 件有真值→速览出真值."""
import json as _json
name, _ = _seed_factor_card(client, name="fa_x")
rep_dir = tmp_path / "reports" / "factor_monthly"
rep_dir.mkdir(parents=True)
(rep_dir / "vps_2026-10-04.json").write_text(
_json.dumps({"as_of": "2026-10-04", "monthly_points": {}}), "utf-8") # 旧 schema
daily_dir = rep_dir / "daily"
daily_dir.mkdir()
(daily_dir / "vps_2026-10-09.json").write_text(_json.dumps({
"as_of": "2026-10-09",
"factor_stats": {"fa_x": {"icAll": 0.03, "tAll": 2.5,
"positiveRatio": 0.8,
"byYear": {"2026": 0.03}}}}), "utf-8")
monkeypatch.setenv("SANGUO_FACTOR_MONTHLY_DIR", str(rep_dir))
got = client.get(f"/api/v1/pipeline/factors/{name}/detail").json()["factor"]["icStats"]
assert got and got["tAll"] == 2.5 # daily 件新者胜,不再被旧根层件挡住
def test_factor_detail_ic_stats_root_newer_wins(client, tmp_path, monkeypatch):
"""根层件更新时仍胜 daily(取 as_of 新者,不是 daily 优先)."""
import json as _json
name, _ = _seed_factor_card(client, name="fa_x")
rep_dir = tmp_path / "reports" / "factor_monthly"
rep_dir.mkdir(parents=True)
(rep_dir / "vps_2026-11-04.json").write_text(_json.dumps({
"as_of": "2026-11-04",
"factor_stats": {"fa_x": {"icAll": 0.05, "tAll": 3.3,
"positiveRatio": 0.9, "byYear": {}}}}), "utf-8")
daily_dir = rep_dir / "daily"
daily_dir.mkdir()
(daily_dir / "vps_2026-10-20.json").write_text(_json.dumps({
"as_of": "2026-10-20",
"factor_stats": {"fa_x": {"icAll": 0.01, "tAll": 1.1,
"positiveRatio": 0.5, "byYear": {}}}}), "utf-8")
monkeypatch.setenv("SANGUO_FACTOR_MONTHLY_DIR", str(rep_dir))
got = client.get(f"/api/v1/pipeline/factors/{name}/detail").json()["factor"]["icStats"]
assert got and got["tAll"] == 3.3
def test_judgement_layer_ignores_daily_dir(client, tmp_path, monkeypatch):
"""判定层不动:monthly-review 仍只读根层,daily 件不得泄漏进速览卡/对拍."""
import json as _json
rep_dir = tmp_path / "reports" / "factor_monthly"
rep_dir.mkdir(parents=True)
(rep_dir / "vps_2026-10-04.json").write_text(_json.dumps({
"as_of": "2026-10-04", "factors": {"fa_old": {"t": 1.0}},
"collective": {"median": 2.1, "floor": 2.0}}), "utf-8")
daily_dir = rep_dir / "daily"
daily_dir.mkdir()
(daily_dir / "vps_2026-10-09.json").write_text(_json.dumps({
"as_of": "2026-10-09", "factors": {"fa_new": {"t": 9.9}},
"collective": {"median": 9.9, "floor": 2.0}}), "utf-8")
monkeypatch.setenv("SANGUO_FACTOR_MONTHLY_DIR", str(rep_dir))
got = client.get("/api/v1/pipeline/monthly-review").json()
assert got["runDate"] == "2026-10-04" # 仍是根层最新,不被 daily 顶掉
assert got["factorsEvaluated"] == 1 # daily 的 fa_new 不算进来
- Step 2: 跑测试确认失败
Run: venv310/bin/python -m pytest tests/api/test_routes_pipeline_factors.py -q -k "daily or root_newer or ignores_daily"
Expected: 前两例 FAIL（icStats 为 None/被旧根层挡）、第三例 FAIL（runDate 变成 2026-10-09——若 include_daily 默认误开）或 PASS（默认关时——那是防回归锚，保留）。
- Step 3: 实现
_monthly_reports(include_daily)+ 两速览端点接线
routes_pipeline.py _monthly_reports 改为：
def _monthly_reports(include_daily: bool = False) -> list[tuple[str, str, dict]]:
"""monthly dir 全部报告 [(host, as_of, doc)];按 as_of 降序,坏文件跳过.
文件名={host}_{as_of}.json(monthly_review.main 落盘原文)。
include_daily=True 时并入 daily/ 子目录件(2026-10-11 双层分工补丁:
速览读最新——根层∪daily 取 as_of 新者);判定层(月末快照语义)不传参,
仍只读根层,一行不动。
"""
d = _monthly_dir()
if not os.path.isdir(d):
return []
out: list[tuple[str, str, dict]] = []
def _scan(folder: str) -> None:
for fn in os.listdir(folder):
if not fn.endswith(".json"):
continue
host, _, as_of = fn[:-5].rpartition("_")
if not host or len(as_of) != 10 or as_of[4] != "-":
continue
try:
with open(os.path.join(folder, fn), encoding="utf-8") as f:
out.append((host, as_of, json.load(f)))
except (OSError, json.JSONDecodeError):
continue
_scan(d)
if include_daily:
daily = os.path.join(d, "daily")
if os.path.isdir(daily):
_scan(daily)
out.sort(key=lambda t: t[1], reverse=True)
return out
factors() 内 reports = _monthly_reports() → reports = _monthly_reports(include_daily=True)（连同上方注释补一句「∪daily/ 取新(2026-10-11 补丁)」）；factor_detail() 内同样改传 include_daily=True。monthly_review_view/factor_ic_trend 两处调用不改。
- Step 4: 跑后端测试全绿
Run: venv310/bin/python -m pytest tests/api/test_routes_pipeline_factors.py tests/api/test_routes_pipeline_monthly.py -q
Expected: 全 PASS（含既有 test_factor_detail_ic_stats/test_factors_list_ic_full_t_from_stats——根层单件形态行为不变）。
- Step 5: 前端标签改「末次评估批」
frontend/src/views/pipeline/FactorDetail.vue:151：<span class="k">末次月度批</span> → <span class="k">末次评估批</span>。检查 frontend/src/views/pipeline/ 下有无其他「末次月度批」字样（grep -rn 末次月度批 frontend/src/ 应零残留）与 FactorDetail.spec.ts 是否断言该文案（有则同步改「末次评估批」）。
- Step 6: 前端验证（权威=build）
Run: cd frontend && npm run build && npx vitest run src/views/pipeline/FactorDetail.spec.ts 2>/dev/null || npx vitest run
Expected: build 绿 + vitest 绿。
- Step 7: Commit
git add sanguo_api/routes_pipeline.py frontend/src/views/pipeline/FactorDetail.vue tests/api/test_routes_pipeline_factors.py frontend/src/views/pipeline/FactorDetail.spec.ts 2>/dev/null || git add -A
git commit -m "feat(api): 弹药速览读最新——icStats/icFullT数据源改根层∪daily/子目录取as_of新者(根层10-04旧schema无factor_stats不再挡住daily真值),判定层三端点仍只读根层一行不动;前端末次月度批标签改末次评估批(2026-10-11双层分工补丁B1,设计见4b4fc16) [vps] [no-doc]"
（commit 前 git log --format=%s -1 自查标签在 subject 行尾。）
Task 2: 月度链①段复用同 as_of 日度批——检测存在跳过重算
Files:
- Modify:
sanguo_factor/monthly_batch.py（main 内run_batch_eval调用段 ~:207-231） - Test:
tests/factor/test_monthly_batch.py（追加用例）
Interfaces:
-
Consumes:
eval_store.list_runs(db)（dict 含 label/end/factors_total/factors_done/run_id）、eval_store.get_rows(db, run_id)（dict 含 factor）、plandict（factor_names/start/end/label） -
Produces: 模块级函数
_reusable_daily_run(db: str, plan: dict) -> dict | None -
Step 1: 写失败测试
在 tests/factor/test_monthly_batch.py 末尾追加（沿用既有 registry/tmp_path fixture 与 monkeypatch.setattr(monthly_batch, "run_batch_eval", ...) 模式；用真实 sqlite eval_db 造日度批）：
def test_monthly_reuses_daily_batch_same_end(monkeypatch, registry, tmp_path, capsys):
"""2026-10-11 双层分工补丁:月度模式检测同 end 日度批已存在→跳过重算
(物理分开≠逻辑重叠,40min 重算白烧);ic_gate 一并跳过(日度链已跑)."""
from sanguo_factor import eval_store
db = str(tmp_path / "eval.db")
names = sorted(f["name"] for f in registry["factors"].values())
rid = eval_store.create_run(db, "daily_2026-09-30", "csi300", 500,
len(names), "2025-10-01", "2026-09-30", {})
eval_store.save_results(db, rid, [{"factor": n, "category": "c",
"expression": "e", "ic_mean": 0.01,
"ic_ir": 0.1} for n in names])
eval_store.finish_run(db, rid, "ok", len(names))
called = []
monkeypatch.setattr(monthly_batch, "run_batch_eval",
lambda *a, **k: called.append(a) or {})
rc = monthly_batch.main(["--as-of", "2026-10-04", "--registry",
os.environ["SANGUO_FACTOR_REGISTRY"],
"--eval-db", db, "--report-dir",
str(tmp_path / "rep")])
assert rc == 0 and not called # 未重算
assert "复用日度批" in capsys.readouterr().out
def test_monthly_reruns_when_daily_factor_set_drifts(monkeypatch, registry, tmp_path, capsys):
"""因子集漂移(日度批后注册新因子)→不许部分复用,回退自算."""
from sanguo_factor import eval_store
db = str(tmp_path / "eval.db")
names = sorted(f["name"] for f in registry["factors"].values())
partial = names[:-1] # 少一个=覆盖不全
rid = eval_store.create_run(db, "daily_2026-09-30", "csi300", 500,
len(partial), "2025-10-01", "2026-09-30", {})
eval_store.save_results(db, rid, [{"factor": n, "category": "c",
"expression": "e", "ic_mean": 0.01,
"ic_ir": 0.1} for n in partial])
eval_store.finish_run(db, rid, "ok", len(partial))
called = []
monkeypatch.setattr(monthly_batch, "run_batch_eval",
lambda *a, **k: called.append(a) or {"status": "ok"})
rc = monthly_batch.main(["--as-of", "2026-10-04", "--registry",
os.environ["SANGUO_FACTOR_REGISTRY"],
"--eval-db", db, "--report-dir",
str(tmp_path / "rep")])
assert rc == 0 and called # 回退自算
def test_daily_mode_never_reuses(monkeypatch, registry, tmp_path):
"""日度模式自身不进复用分支(同 label 重跑=幂等覆盖,不查旧批)."""
from sanguo_factor import eval_store
db = str(tmp_path / "eval.db")
names = sorted(f["name"] for f in registry["factors"].values())
rid = eval_store.create_run(db, "daily_2026-09-30", "csi300", 500,
len(names), "2025-10-01", "2026-09-30", {})
eval_store.save_results(db, rid, [{"factor": n, "category": "c",
"expression": "e", "ic_mean": 0.01,
"ic_ir": 0.1} for n in names])
eval_store.finish_run(db, rid, "ok", len(names))
called = []
monkeypatch.setattr(monthly_batch, "run_batch_eval",
lambda *a, **k: called.append(a) or {})
monthly_batch.main(["--as-of", "2026-09-30", "--daily", "--registry",
os.environ["SANGUO_FACTOR_REGISTRY"],
"--eval-db", db, "--report-dir",
str(tmp_path / "rep")])
assert called
注意：create_run/save_results/finish_run 的参数名以 sanguo_factor/eval_store.py 实签名为准（行内值不对就照实际改）；registry fixture 的因子若含 retired/graveyard 全被剔则 plan 因子集=被剔后的名单——测试断言前先看 plan_batch 实际产出的 names（必要时打 print 调整）。若既有测试里 main 调用还须其他参数（vnpy-db/fund-data-dir 缺省路径），照 test_main_daily_flag_reaches_plan 抄全参数形态。
- Step 2: 跑测试确认失败
Run: venv310/bin/python -m pytest tests/factor/test_monthly_batch.py -q -k "reuses or drifts or never_reuses"
Expected: test_monthly_reuses_daily_batch_same_end FAIL（当前必重算）、test_daily_mode_never_reuses PASS（现状即如此，防回归锚）。
- Step 3: 实现
_reusable_daily_run+ main 接线
monthly_batch.py 模块级加：
def _reusable_daily_run(db: str, plan: dict[str, Any]) -> dict | None:
"""月末批可复用的同 end 日度批(2026-10-11 双层分工补丁,决议H:
物理分开≠逻辑重叠——次月1日夜班已产出 end=月末 的日度批时,月度链
stage1 不重算 40min,直接复用)。全条件满足才复用:label/end 钉住同
as_of、因子集一致(漂移=回退自算,不许部分复用)、完整批。
"""
from sanguo_factor import eval_store
want = f"daily_{plan['end']}"
plan_names = set(plan["factor_names"])
for r in eval_store.list_runs(db):
if (r.get("label") == want and r.get("end") == plan["end"]
and r.get("factors_total") == len(plan_names)
and r.get("factors_done") == len(plan_names)):
got = {row["factor"] for row in eval_store.get_rows(db, r["run_id"])}
if got == plan_names:
return r
return None
main() 内 stage1 段改造——print(f"[monthly_batch] {len(plan['factor_names'])} 因子 ...") 之后、values_dir = ... 之前插：
if args.daily:
summary = run_batch_eval(...) # 原调用原样保留
elif (reuse := _reusable_daily_run(db, plan)) is not None:
print(f"[monthly_batch] 复用日度批 run_id={reuse['run_id']} "
f"label={reuse['label']}(决议H:物理分开≠逻辑重叠,重算白烧),"
f"跳过 stage1 重算与 ic_gate")
# stage2 monthly_review 经 pick_eval_run 自然选中该批(end<=as_of 最新)
return 0
else:
summary = run_batch_eval(...) # 原调用原样
（实现时把原 run_batch_eval 调用抽成局部变量/保持单处调用点+前置分支均可，以最小 diff 为准；跳过路径不写 values/ic_gate 产物——日度链跑时已出过同 as_of 的。）return 0 前确认函数尾部没有其他必须执行的收尾（对比原 main 的 return 路径）。
- Step 4: 跑 factor 域测试全绿
Run: venv310/bin/python -m pytest tests/factor/test_monthly_batch.py tests/factor/test_monthly_review.py -q
Expected: 全 PASS（含既有 test_main_invokes_run_batch_eval_with_plan——空 eval_db 时回退自算路径不变）。
- Step 5: Commit
git add sanguo_factor/monthly_batch.py tests/factor/test_monthly_batch.py
git commit -m "feat(factor): 月度链①段复用同as_of日度批——月末批stage1检测eval_db已有end=月末且label=daily_的完整批(因子集一致才复用,漂移回退自算)则跳过40min重算与ic_gate(2026-10-11双层分工补丁B2,设计见4b4fc16;11-01月度班首受益) [vps] [no-doc]"
收尾（controller 做）
venv310/bin/python -m pytest tests/api tests/factor -q全量回归cd frontend && npm run build终验- push → CI（nas-verify）绿后报用户：弹药速览有真值可去 VPS 验证（部署等用户口令或搭下班车）