diff --git a/sanguo_factor/ic_gate.py b/sanguo_factor/ic_gate.py index a6daf348..51e262bc 100644 --- a/sanguo_factor/ic_gate.py +++ b/sanguo_factor/ic_gate.py @@ -16,6 +16,7 @@ import pandas as pd IC_THRESHOLD = 0.99 _MIN_DAYS = 20 # 共同日下限:少于此无统计意义 _MIN_SYMS = 10 # 单日截面标的下限 +_MIN_CORR_DAYS = 20 # 有效 corr 日下限(P3-10):常数截面滤除后少数日均值不置疑 def _load_wide(path: str) -> pd.DataFrame | None: @@ -25,27 +26,39 @@ def _load_wide(path: str) -> pd.DataFrame | None: return df if not df.empty else None -def _cross_sectional_ic(a: pd.DataFrame, b: pd.DataFrame) -> float | None: - """逐交易日截面 Pearson→均值;无共同日/截面太小→None.""" +def _cross_sectional_ic(a: pd.DataFrame, b: pd.DataFrame) -> tuple[float | None, int]: + """逐交易日截面 Pearson→均值;返回 (均值 IC, 有效 corr 日数). + + 无共同日/截面太小→(None, 0);常数截面日 corr=NaN 滤除后计入有效日数 + (P3-10:有效日数供报告层判 insufficient_days). + """ days = a.index.intersection(b.index) if len(days) < _MIN_DAYS: - return None + return None, 0 syms = a.columns.intersection(b.columns) if len(syms) < _MIN_SYMS: - return None + return None, 0 a2, b2 = a.loc[days, syms], b.loc[days, syms] corrs = [] for day in days: c = a2.loc[day].corr(b2.loc[day]) if c == c: # NaN 过滤(常数截面) corrs.append(c) - return sum(corrs) / len(corrs) if corrs else None + return (sum(corrs) / len(corrs) if corrs else None), len(corrs) def orthogonality_report(values_dir: str, new_names: list[str], library_names: list[str], *, threshold: float = IC_THRESHOLD) -> dict: - """逐新因子→对在库 max |IC|;json 形状 {threshold, checked, flagged}.""" + """逐新因子→对在库 max |IC|;json 形状 {threshold, checked, flagged, + skipped_missing_values, compared_pairs, insufficient_days}. + + P2-9: checked < len(new_names)(缺件跳过)与 compared_pairs(实际比对 + 对数)如实计数——「查过且干净」与「根本没查成」可区分,系统性 eval + 失败不再呈现绿灯(接线处 monthly_batch 据此 stderr 告警). + P3-10: 全部比对有效 corr 日数 <_MIN_CORR_DAYS 的新因子不进置疑清单, + 入 insufficient_days 留痕. + """ cache: dict[str, pd.DataFrame | None] = {} def wide(name: str) -> pd.DataFrame | None: @@ -55,22 +68,37 @@ def orthogonality_report(values_dir: str, new_names: list[str], return cache[name] flagged, checked = [], 0 + skipped = compared = 0 + insufficient: list[str] = [] for new in new_names: nf = wide(new) if nf is None: + skipped += 1 continue checked += 1 best_name, best_ic = "", None + judged = False # 至少一次比对达有效日数下限 for lib in library_names: if lib == new: continue - ic = _cross_sectional_ic(nf, wide(lib)) if wide(lib) is not None else None + lf = wide(lib) + if lf is None: + continue + compared += 1 + ic, n_valid = _cross_sectional_ic(nf, lf) + if n_valid < _MIN_CORR_DAYS: + continue + judged = True if ic is not None and (best_ic is None or abs(ic) > abs(best_ic)): best_name, best_ic = lib, ic if best_ic is not None and abs(best_ic) >= threshold: flagged.append({"factor": new, "vs": best_name, "ic": round(best_ic, 4)}) - return {"threshold": threshold, "checked": checked, "flagged": flagged} + elif not judged: + insufficient.append(new) + return {"threshold": threshold, "checked": checked, "flagged": flagged, + "skipped_missing_values": skipped, "compared_pairs": compared, + "insufficient_days": insufficient} def write_report(out_path: str, report: dict) -> None: diff --git a/sanguo_factor/monthly_batch.py b/sanguo_factor/monthly_batch.py index da42d7f4..d294eeb6 100644 --- a/sanguo_factor/monthly_batch.py +++ b/sanguo_factor/monthly_batch.py @@ -77,7 +77,12 @@ def resolve_with_yaml(registry: dict): def plan_batch(registry: dict, as_of: str, window_months: int = 12, resolver=None) -> dict[str, Any]: - """纯函数:注册表→批参数(main 只做 IO 编排,可单测). + """注册表→批参数(main 只做 IO 编排,可单测). + + 「纯函数」契约勘正(P3-12): resolver=get_factor(默认)时纯读; + resolver=resolve_with_yaml 时带进程内注册副作用——首次解析 miss 名 + 会动态注册进全局因子库(register_factor),对进程内其他消费者可见; + 跨调用幂等(已注册直接命中),但测试须留意跨用例泄漏. 血统名漂移防线(09-27 审计 F-3): 库中不可解析名剔除进 skipped_unregistered(不炸批,main 侧显式告警)——yaml 与因子库 @@ -94,6 +99,31 @@ def plan_batch(registry: dict, as_of: str, window_months: int = 12, "end": end, "label": f"monthly_{as_of[:7]}"} +def decomposer_new_names(registry: dict, factor_names: list[str]) -> list[str]: + """IC 闸「新因子」判定(审计 P3-11):条目级 origin 优先,旧数据回落 + version params 全版本扫描(非仅最新). + + 双向漂移根因=旧判定读「最新版 params.origin」:分解因子月月在闸视野 + 永不收敛、手工升版(新 params 不带 origin)即无声退出闸视野。改以稳定 + 身份判定:条目级 origin 存在则以条目级为准(人工可显式改判退出); + 否则回落扫全部版本 params(decompose 的 origin 戳落在含 v1 的各版本 + 上),手工升版不再影响判定——同注册表状态跨月判定一致. + """ + picked = [] + for name in factor_names: + entry = registry["factors"].get(name) or {} + origin = entry.get("origin") + if origin is not None: + if origin == "decomposer": + picked.append(name) + continue + versions = entry.get("versions") or [] + if any((v.get("params") or {}).get("origin") == "decomposer" + for v in versions): + picked.append(name) + return picked + + def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser(description="先批后评·第一段:注册表全量月度重评批") ap.add_argument("--as-of", required=True, help="数据截止日 YYYY-MM-DD(归一自然月末)") @@ -149,15 +179,21 @@ def main(argv: list[str] | None = None) -> int: fund_data_dir=args.fund_data_dir) print(f"[monthly_batch] 完成: {summary}") # IC 正交闸(D4): 新因子=origin decomposer;报告件不自动转移 - new_names = [n for n, e in registry["factors"].items() - if ((e.get("versions") or [{}])[-1].get("params") or {} - ).get("origin") == "decomposer" - and n in plan["factor_names"]] + new_names = decomposer_new_names(registry, plan["factor_names"]) if new_names: from sanguo_factor.ic_gate import orthogonality_report, write_report rep = orthogonality_report(values_dir, new_names, plan["factor_names"]) write_report(os.path.join(args.report_dir, f"ic_gate_{plan['label']}.json"), rep) + # P2-9: 「查过且干净」与「根本没查成」可区分——缺件/零比对时 + # 显式告警(系统性 eval 失败不再呈现绿灯);告警属信号语义, + # 不 exit 非零 + if rep["checked"] < len(new_names) or rep["compared_pairs"] == 0: + print(f"[monthly_batch] ⚠️ IC 正交闸未完整执行: 新因子 " + f"{len(new_names)} 实查 {rep['checked']}(缺件跳过 " + f"{rep['skipped_missing_values']}),比对对数 " + f"{rep['compared_pairs']}——疑系统性 eval 失败,勿当绿灯", + file=sys.stderr) if rep["flagged"]: print(f"[monthly_batch] ⚠️ IC 正交闸: " + ", ".join(f"{f['factor']}(vs {f['vs']} ic={f['ic']})" diff --git a/tests/factor/test_ic_gate.py b/tests/factor/test_ic_gate.py index 3a98ce6c..1a534cdf 100644 --- a/tests/factor/test_ic_gate.py +++ b/tests/factor/test_ic_gate.py @@ -51,3 +51,53 @@ def test_missing_parquet_skipped_and_write_report(tmp_path): out = tmp_path / "sub" / "ic.json" write_report(str(out), rep) assert json.loads(out.read_text())["threshold"] == 0.99 + + +# —— P2-9: 「查过且干净」与「根本没查成」报告可区分 —— + +def test_missing_new_factor_counted_in_report(tmp_path): + """新因子 parquet 缺件(eval 单因子失败不落盘形态)→计数如实: + skipped_missing_values+1, checked 不含它,比对对数只计实查.""" + rng = np.random.default_rng(5) + dup = _wide(rng) + dup.to_parquet(tmp_path / "llm_new.parquet") + dup.to_parquet(tmp_path / "f_lib.parquet") + rep = orthogonality_report(str(tmp_path), ["llm_new", "llm_lost"], + ["llm_new", "f_lib"]) + assert rep["checked"] == 1 + assert rep["skipped_missing_values"] == 1 + assert rep["compared_pairs"] == 1 + assert len(rep["flagged"]) == 1 + + +def test_compared_pairs_zero_when_all_values_missing(tmp_path): + rep = orthogonality_report(str(tmp_path), ["llm_new"], ["f_lib"]) + assert rep["checked"] == 0 + assert rep["skipped_missing_values"] == 1 + assert rep["compared_pairs"] == 0 + assert rep["flagged"] == [] + + +# —— P3-10: 有效 corr 日数下限(常数截面日滤除后少数日不置疑) —— + +def test_valid_corr_days_floor_19_rejected_20_flagged(tmp_path): + # 20 有效日(21 共同日滤 1 常数截面日)→照常置疑 + rng = np.random.default_rng(9) + a20 = _wide(rng, n_days=21) + dup20 = a20 + rng.normal(scale=1e-6, size=a20.shape) + a20.to_parquet(tmp_path / "llm_a.parquet") + dup20.to_parquet(tmp_path / "f_lib20.parquet") + rep = orthogonality_report(str(tmp_path), ["llm_a"], ["f_lib20"]) + assert rep["flagged"] and rep["flagged"][0]["factor"] == "llm_a" + assert rep["insufficient_days"] == [] + + # 19 有效日(20 共同日滤 1 常数截面日)→不置疑,insufficient_days 留痕 + a19 = _wide(rng, n_days=20) + dup19 = a19 + rng.normal(scale=1e-6, size=a19.shape) + a19.iloc[0] = 1.0 # 常数截面日(corr=NaN 被滤) + dup19.iloc[0] = 1.0 + a19.to_parquet(tmp_path / "llm_b.parquet") + dup19.to_parquet(tmp_path / "f_lib19.parquet") + rep2 = orthogonality_report(str(tmp_path), ["llm_b"], ["f_lib19"]) + assert rep2["flagged"] == [] + assert rep2["insufficient_days"] == ["llm_b"] diff --git a/tests/factor/test_monthly_batch.py b/tests/factor/test_monthly_batch.py index f5611707..e13ae823 100644 --- a/tests/factor/test_monthly_batch.py +++ b/tests/factor/test_monthly_batch.py @@ -330,3 +330,100 @@ def test_main_passes_fund_data_dir(tmp_path, monkeypatch): "--report-dir", str(tmp_path)]) assert rc == 0 assert seen.get("fund_data_dir") == "/tmp/static_root" + + +# —— P2-9: IC 闸系统性失败不再呈现绿灯(缺件计数+stderr 告警) —— + +def test_main_warns_when_new_factor_values_missing(tmp_path, capsys, monkeypatch): + """新因子 parquet 缺件(eval 单因子失败形态)→checked