diff --git a/sanguo_factor/review_compare.py b/sanguo_factor/review_compare.py new file mode 100644 index 00000000..c5262435 --- /dev/null +++ b/sanguo_factor/review_compare.py @@ -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()) diff --git a/tests/factor/test_review_compare.py b/tests/factor/test_review_compare.py new file mode 100644 index 00000000..7caaac5b --- /dev/null +++ b/tests/factor/test_review_compare.py @@ -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