feat(factor): 月度链①段复用同as_of日度批——月末批stage1检测eval_db已有end=月末且label=daily_的完整批(因子集一致才复用,漂移回退自算)则跳过40min重算与ic_gate(2026-10-11双层分工补丁B2,设计见4b4fc16;11-01月度班首受益) [vps] [no-doc]

2026-10-11 09:52:52 +08:00
parent d374e03bb2
commit 9389d75d15
2 changed files with 107 additions and 0 deletions
@@ -13,6 +13,7 @@ import argparse
 import calendar
 import json
 import os
+import sqlite3
 import sys
 from typing import Any

@@ -132,6 +133,29 @@ def decomposer_new_names(registry: dict, factor_names: list[str]) -> list[str]:
    return picked


+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"])
+    try:
+        runs = eval_store.list_runs(db)
+    except sqlite3.OperationalError:   # 空/无 schema 库=无可复用,回退自算
+        return None
+    for r in runs:
+        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
+
+
 def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description="先批后评·第一段:注册表全量月度重评批")
    ap.add_argument("--as-of", required=True, help="数据截止日 YYYY-MM-DD(归一自然月末)")
@@ -205,6 +229,18 @@ def main(argv: list[str] | None = None) -> int:
                print(f"[monthly_batch] 已开数据缺口 issue #{num}")
    print(f"[monthly_batch] {len(plan['factor_names'])} 因子 "
          f"{plan['start']}~{plan['end']} → {db} (label={plan['label']})")
+    # 2026-10-11 双层分工补丁 B2(决议H): 月度链①段复用同 as_of 日度批——
+    # 日度模式自身不查(同 label 重跑=幂等覆盖);月度模式检测到 end=月末、
+    # label=daily_月末、因子集一致的完整批→跳过 stage1 重算与 ic_gate
+    # (不写 values/ic_gate 产物,日度链跑时已出过同 as_of 的;stage2
+    # monthly_review 经 pick_eval_run 自然选中该批)。日度批缺席/漂移→
+    # 落到下方原重算路径,行为不变。
+    reuse = None if args.daily else _reusable_daily_run(db, plan)
+    if reuse is not None:
+        print(f"[monthly_batch] 复用日度批 run_id={reuse['run_id']} "
+              f"label={reuse['label']}(决议H:物理分开≠逻辑重叠,重算白烧),"
+              f"跳过 stage1 重算与 ic_gate")
+        return 0
    values_dir = os.path.join(args.report_dir, "values", plan["label"])
    summary = run_batch_eval(plan["factor_names"], plan["start"], plan["end"],
                             db, plan["label"], factor_values_out=values_dir,
@@ -499,6 +499,77 @@ def test_decomposer_origin_stable_across_manual_bump():
    assert month2 == ["llm_x"]


+# —— 2026-10-11 双层分工补丁 B2(决议H): 月度链①段复用同 as_of 日度批 ——
+
+def _seed_daily_run(db, plan, names):
+    """eval_db 造一个 end=plan月末 的日度批(复用检测面对的既存事实).
+
+    create_run 前须 init_db(空文件无 schema);save_results 行须带 metrics 键
+    (eval_store 实签名,metrics JSON 落 eval_results.metrics_json)."""
+    from sanguo_factor import eval_store
+    eval_store.init_db(db)
+    rid = eval_store.create_run(db, f"daily_{plan['end']}", "csi300", 500,
+                                len(names), plan["start"], plan["end"], {})
+    eval_store.save_results(db, rid, [
+        {"factor": n, "category": "c", "expression": "e",
+         "metrics": {"1": {"ic_mean": 0.01}}} for n in names])
+    eval_store.finish_run(db, rid, "ok", len(names))
+
+
+def test_monthly_reuses_daily_batch_same_end(monkeypatch, registry, tmp_path, capsys):
+    """2026-10-11 双层分工补丁:月度模式检测同 end 日度批已存在→跳过重算
+    (物理分开≠逻辑重叠,40min 重算白烧);ic_gate 一并跳过(日度链已跑).
+
+    as_of=2026-09-30=wrapper 语义(次月首个周日跑,as_of 归一到上月月末)——
+    次月 1 日夜班日度批 end 恰=上月月末,月度链检测到即复用."""
+    from sanguo_factor import eval_store
+    db = str(tmp_path / "eval.db")
+    plan = plan_batch(registry, "2026-09-30")   # 实际可跑名单(剔 retired/graveyard)
+    _seed_daily_run(db, plan, plan["factor_names"])
+    called = []
+    monkeypatch.setattr(monthly_batch, "run_batch_eval",
+                        lambda *a, **k: called.append(a) or {})
+    p = str(tmp_path / "reg.yaml")
+    save_registry(p, registry)
+    rc = main(["--as-of", "2026-09-30", "--registry", p,
+               "--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):
+    """因子集漂移(日度批后注册新因子)→不许部分复用,回退自算."""
+    upsert_factor(registry, "fund_gross_margin", hypothesis="H-5")   # 库中真名,可解析
+    db = str(tmp_path / "eval.db")
+    plan = plan_batch(registry, "2026-09-30")
+    partial = plan["factor_names"][:-1]        # 少一个=覆盖不全
+    assert partial and set(partial) != set(plan["factor_names"])
+    _seed_daily_run(db, plan, partial)
+    called = []
+    monkeypatch.setattr(monthly_batch, "run_batch_eval",
+                        lambda *a, **k: called.append(a) or {})
+    p = str(tmp_path / "reg.yaml")
+    save_registry(p, registry)
+    rc = main(["--as-of", "2026-09-30", "--registry", p,
+               "--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 重跑=幂等覆盖,不查旧批)."""
+    db = str(tmp_path / "eval.db")
+    plan = plan_batch(registry, "2026-09-30")
+    _seed_daily_run(db, plan, plan["factor_names"])
+    called = []
+    monkeypatch.setattr(monthly_batch, "run_batch_eval",
+                        lambda *a, **k: called.append(a) or {})
+    p = str(tmp_path / "reg.yaml")
+    save_registry(p, registry)
+    main(["--as-of", "2026-09-30", "--daily", "--registry", p,
+          "--eval-db", db, "--report-dir", str(tmp_path / "rep")])
+    assert called
+
+
 def test_decomposer_origin_entry_level_wins_hand_factor_excluded():
    """条目级 origin 存在则以条目级为准;手写因子(origin 全无)不入闸视野."""
    import sanguo_factor.monthly_batch as mb