feat(factor): 月度批评报告判定弹药三件——factor_stats 聚合(全期合成t/正IC占比/逐年IC)+md 渲染 [nas]
This commit is contained in:
@@ -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']},"
|
||||
|
||||
@@ -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 不进逐年
|
||||
|
||||
Reference in New Issue
Block a user