From 86bfbcf283b958caea056985b5ebfdec75278fe7 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Sat, 10 Oct 2026 10:18:21 +0800 Subject: [PATCH] =?UTF-8?q?feat(factor):=20=E6=9C=88=E5=BA=A6=E6=89=B9?= =?UTF-8?q?=E8=AF=84=E6=8A=A5=E5=91=8A=E5=88=A4=E5=AE=9A=E5=BC=B9=E8=8D=AF?= =?UTF-8?q?=E4=B8=89=E4=BB=B6=E2=80=94=E2=80=94factor=5Fstats=20=E8=81=9A?= =?UTF-8?q?=E5=90=88(=E5=85=A8=E6=9C=9F=E5=90=88=E6=88=90t/=E6=AD=A3IC?= =?UTF-8?q?=E5=8D=A0=E6=AF=94/=E9=80=90=E5=B9=B4IC)+md=20=E6=B8=B2?= =?UTF-8?q?=E6=9F=93=20[nas]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- sanguo_factor/monthly_review.py | 46 +++++++++++++++++++++++++++++ tests/factor/test_monthly_review.py | 37 +++++++++++++++++++++++ 2 files changed, 83 insertions(+) diff --git a/sanguo_factor/monthly_review.py b/sanguo_factor/monthly_review.py index 75b990b9..1e72ec31 100644 --- a/sanguo_factor/monthly_review.py +++ b/sanguo_factor/monthly_review.py @@ -68,6 +68,34 @@ def load_history_points(out_dir: str, exclude_file: str) -> dict[str, list[dict] return {f: [pts[m] for m in sorted(pts)] for f, pts in merged.items()} +def factor_stats(points: list[dict]) -> dict: + """判定弹药聚合(2026-10-10 spec §4.2):全期合成 t/正 IC 占比/逐年. + + 口径:各月点 ic_mean(该批 12M 窗日均 IC 均值)的等权均值≈全史近似; + 合成 t=mean/(std(ddof=1)/√n),n<2 或 std=0 时如实 None(小样本判读弱 + 的诚实注记在 spec,判定动作不因此自动化). + """ + ics = [p["ic_mean"] for p in points if p.get("ic_mean") is not None] + if not ics: + return {"icAll": None, "tAll": None, "positiveRatio": None, "byYear": {}} + n = len(ics) + mean_ic = sum(ics) / n + t_all = None + if n >= 2: + var = sum((v - mean_ic) ** 2 for v in ics) / (n - 1) + if var > 0: + t_all = round(mean_ic / (var ** 0.5 / n ** 0.5), 4) + by_year: dict[str, list[float]] = {} + for p in points: + m, v = str(p.get("month") or ""), p.get("ic_mean") + if m and v is not None: + by_year.setdefault(m[:4], []).append(v) + return {"icAll": round(mean_ic, 6), "tAll": t_all, + "positiveRatio": round(sum(1 for v in ics if v > 0) / n, 4), + "byYear": {y: round(sum(v) / len(v), 6) + for y, v in sorted(by_year.items())}} + + def build_report(registry: dict, current_points: dict[str, dict], history_points: dict[str, list[dict]], as_of: str, host: str) -> dict: @@ -112,6 +140,8 @@ def build_report(registry: dict, current_points: dict[str, dict], "observations": observations, "collective": collective_level(promoted_ts), "monthly_points": monthly_points, + "factor_stats": {name: factor_stats(pts) + for name, pts in monthly_points.items()}, "unregistered": unregistered, "pre_gate": pre_gate_table(registry, monthly_points, current_points), "edge_cases": edge_cases, @@ -201,6 +231,22 @@ def render_markdown(report: dict) -> str: for k, v in report["observations"].items(): rs = ";".join(v["reasons"]) if v["reasons"] else "新鲜且覆盖足" lines.append(f"- {v['status']} {k}: {rs}") + stats = report.get("factor_stats") or {} + if stats: + lines.append("") + lines.append("## 判定弹药（全期口径）") + lines.append("") + lines.append("| 因子 | 全期IC | 合成t | 正IC占比 | 逐年IC |") + lines.append("|------|--------|-------|----------|--------|") + for name in sorted(stats): + s = stats[name] + years = ", ".join(f"{y}:{v:.4f}" for y, v in s["byYear"].items()) or "—" + + def _f(v, nd=4): + return "—" if v is None else f"{v:.{nd}f}" + + lines.append(f"| {name} | {_f(s['icAll'], 6)} | {_f(s['tAll'])} " + f"| {_f(s['positiveRatio'])} | {years} |") c = report["collective"] lines.append(f"\n## 集体水位(决议 C:跌破=研判卡数据,权重冻结不动)\n" f"- 已晋级 n={c['n']},t 中位数={c['median']},地板={c['floor']}," diff --git a/tests/factor/test_monthly_review.py b/tests/factor/test_monthly_review.py index 5787ef2a..040731e3 100644 --- a/tests/factor/test_monthly_review.py +++ b/tests/factor/test_monthly_review.py @@ -1,6 +1,7 @@ # tests/factor/test_monthly_review.py """月度批评 TDD——spec §4.2 决议 H: 双机各跑,数据截止对齐自然月末.""" import json +import math import os import pytest @@ -9,6 +10,7 @@ from sanguo_factor import eval_store from sanguo_factor.monthly_review import ( build_report, extract_points, + factor_stats, load_history_points, main, pick_eval_run, @@ -253,3 +255,38 @@ def test_pre_gate_threshold_follows_promotion_bar_constant(monkeypatch): pg = mr.pre_gate_table(reg, {"fa_a": [{"month": "2026-09", "t": 2.2}]}, {"fa_a": {"t": 2.2, "ic_mean": 0.05, "count": 200}}) assert pg["fa_a"]["flag"] == "REVIEW" # 2.2 < 2.5(常量)→REVIEW;硬编码 2.0 会 PASS + + +# —— 判定弹药三件(2026-10-10 spec §4.2): 全期合成t/正IC占比/逐年IC —— +def _ammo_pts(*pairs): + return [{"month": m, "ic_mean": v, "t": None, "count": 1} for m, v in pairs] + + +def test_factor_stats_basic(): + s = factor_stats(_ammo_pts(("2025-09", 0.04), ("2025-10", 0.06), ("2026-09", 0.02))) + assert s["icAll"] == round((0.04 + 0.06 + 0.02) / 3, 6) + assert s["positiveRatio"] == 1.0 + assert s["byYear"] == {"2025": round((0.04 + 0.06) / 2, 6), "2026": 0.02} + mean = 0.04 + var = ((0.0) + (0.02 ** 2) + ((-0.02) ** 2)) / 2 # ddof=1(离差平方) + t_all = mean / (math.sqrt(var) / math.sqrt(3)) + assert s["tAll"] == round(t_all, 4) + + +def test_factor_stats_negative_and_short(): + s = factor_stats(_ammo_pts(("2025-09", 0.05), ("2025-10", -0.01))) + assert s["positiveRatio"] == 0.5 + assert s["tAll"] is not None + single = factor_stats(_ammo_pts(("2025-09", 0.05))) + assert single["tAll"] is None # n<2 无合成 t + assert single["positiveRatio"] == 1.0 + + +def test_factor_stats_empty_and_dirty(): + assert factor_stats([]) == {"icAll": None, "tAll": None, + "positiveRatio": None, "byYear": {}} + dirty = [{"month": "2025-09", "ic_mean": None, "t": None}, + {"month": "", "ic_mean": 0.1, "t": None}] + s = factor_stats(dirty) + assert s["icAll"] == 0.1 + assert s["byYear"] == {} # 空 month 不进逐年