feat(pipeline): P1 双机对拍器——t/ic 容差比对,分歧只开排查不开告警 [vps] [no-doc]

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-23 09:46:20 +08:00
parent 1538a23be2
commit fe4042572f
2 changed files with 128 additions and 0 deletions
@@ -0,0 +1,82 @@
+# sanguo_factor/review_compare.py
+"""双机对拍——流水线 spec §4.2 决议 H.
+
+关键指标一致→结果落档进告警流程;分歧超阈→不出降权告警、只开排查 issue
+(宁可误报排查,不可错发告警;分歧大概率=数据版本漂移或环境差异,本身就该被抓).
+对拍两指标: t(t_stat) 与 ic_mean,容差起步默认 tol_t=0.25 / tol_ic=0.02(首年校准).
+双方 t 均缺的因子跳过(数据面同缺口非环境分歧).
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+
+DEFAULT_TOL_T = 0.25
+DEFAULT_TOL_IC = 0.02
+
+
+def compare(a: dict, b: dict, tol_t: float = DEFAULT_TOL_T,
+            tol_ic: float = DEFAULT_TOL_IC) -> dict:
+    fa, fb = a.get("factors") or {}, b.get("factors") or {}
+    matches: list[str] = []
+    mismatches: list[dict] = []
+    missing: list[dict] = []
+    for name in sorted(set(fa) & set(fb)):
+        pa, pb = fa[name], fb[name]
+        ta, tb = pa.get("t"), pb.get("t")
+        if ta is None and tb is None:
+            matches.append(name)  # 双缺=同缺口,非分歧
+            continue
+        if ta is None or tb is None:
+            missing.append({"factor": name, "side": "a" if tb is None else "b",
+                            "t": ta if ta is not None else tb})
+            continue
+        ia, ib = pa.get("ic_mean"), pb.get("ic_mean")
+        delta_t = abs(ta - tb)
+        delta_ic = abs((ia or 0.0) - (ib or 0.0))
+        if delta_t > tol_t or delta_ic > tol_ic:
+            mismatches.append({"factor": name, "a_t": ta, "b_t": tb,
+                               "delta_t": delta_t, "a_ic": ia, "b_ic": ib})
+        else:
+            matches.append(name)
+    for name in sorted(set(fa) - set(fb)):
+        missing.append({"factor": name, "side": "a", "t": fa[name].get("t")})
+    for name in sorted(set(fb) - set(fa)):
+        missing.append({"factor": name, "side": "b", "t": fb[name].get("t")})
+    return {"consistent": not mismatches and not missing,
+            "matches": matches, "mismatches": mismatches, "missing": missing}
+
+
+def main(argv: list[str] | None = None) -> int:
+    ap = argparse.ArgumentParser(description="月度批评双机对拍")
+    ap.add_argument("json_a")
+    ap.add_argument("json_b")
+    ap.add_argument("--tol-t", type=float, default=DEFAULT_TOL_T)
+    ap.add_argument("--tol-ic", type=float, default=DEFAULT_TOL_IC)
+    args = ap.parse_args(argv)
+    try:
+        with open(args.json_a, encoding="utf-8") as f:
+            a = json.load(f)
+        with open(args.json_b, encoding="utf-8") as f:
+            b = json.load(f)
+    except (OSError, json.JSONDecodeError) as e:
+        print(f"[review_compare] 读件失败: {e}", file=sys.stderr)
+        return 1
+    r = compare(a, b, tol_t=args.tol_t, tol_ic=args.tol_ic)
+    print(f"[review_compare] 一致 {len(r['matches'])} · 分歧 {len(r['mismatches'])} · "
+          f"缺面 {len(r['missing'])}(容差 t±{args.tol_t} ic±{args.tol_ic})")
+    for m in r["mismatches"]:
+        print(f"  ⚠ 排查 {m['factor']}: a(t={m['a_t']:.3f},ic={m['a_ic']}) vs "
+              f"b(t={m['b_t']:.3f},ic={m['b_ic']}) Δt={m['delta_t']:.3f}")
+    for m in r["missing"]:
+        print(f"  ⚠ 缺面 {m['factor']}: 仅 {m['side']} 侧有(t={m['t']})")
+    if r["consistent"]:
+        print("[review_compare] 一致→结果落档进告警流程(决议 H)")
+        return 0
+    print("[review_compare] 分歧→不出降权告警,开排查 issue(决议 H)")
+    return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,46 @@
+# tests/factor/test_review_compare.py
+"""双机对拍 TDD——决议 H: 分歧超阈不开告警只开排查."""
+from sanguo_factor.review_compare import compare
+
+
+def _doc(factors, host="x"):
+    return {"host": host, "as_of": "2026-09-30", "factors": factors}
+
+
+def test_consistent_within_tolerance():
+    a = _doc({"p": {"t": 3.90, "ic_mean": 0.030, "count": 512}})
+    b = _doc({"p": {"t": 3.75, "ic_mean": 0.038, "count": 512}})
+    r = compare(a, b)   # |Δt|=0.15<=0.25, |Δic|=0.008<=0.02
+    assert r["consistent"] is True and not r["mismatches"] and not r["missing"]
+    assert r["matches"] == ["p"]
+
+
+def test_t_mismatch_over_tolerance():
+    a = _doc({"p": {"t": 3.90, "ic_mean": 0.03, "count": 512}})
+    b = _doc({"p": {"t": 3.30, "ic_mean": 0.03, "count": 512}})
+    r = compare(a, b)   # |Δt|=0.60>0.25
+    assert r["consistent"] is False
+    m = r["mismatches"][0]
+    assert m["factor"] == "p" and abs(m["delta_t"] - 0.60) < 1e-9
+
+
+def test_ic_mismatch_over_tolerance():
+    a = _doc({"p": {"t": 3.9, "ic_mean": 0.030, "count": 512}})
+    b = _doc({"p": {"t": 3.9, "ic_mean": 0.055, "count": 512}})
+    assert compare(a, b)["consistent"] is False
+
+
+def test_missing_factor_on_one_side():
+    a = _doc({"p": {"t": 3.9, "ic_mean": 0.03, "count": 512},
+              "q": {"t": 2.0, "ic_mean": 0.01, "count": 500}})
+    b = _doc({"p": {"t": 3.9, "ic_mean": 0.03, "count": 512}})
+    r = compare(a, b)
+    assert r["consistent"] is False
+    assert r["missing"] == [{"factor": "q", "side": "a", "t": 2.0}]
+
+
+def test_none_t_skipped_as_match():
+    # 双方 t 都缺(数据面同缺口)不算分歧——对拍只对「两边都算得出来」的比
+    a = _doc({"s": {"t": None, "ic_mean": None, "count": 0}})
+    b = _doc({"s": {"t": None, "ic_mean": None, "count": 0}})
+    assert compare(a, b)["consistent"] is True