sanguo/sanguo_vnpy_v2
3
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(factor): IC 正交闸可观测性+判定收敛——report 增 skipped_missing_values/compared_pairs 计数+checked<len stderr 告警(系统性 eval 失败不再呈现绿灯)+有效 corr 日数下限 20(常数日滤除后少数日不触发置疑,insufficient_days 留痕)+origin 判定改条目级(分解因子不再每月重考,手工升版不退闸视野)+plan_batch docstring 勘正副作用 (审计 P2-9/P3-10/11/12) [vps] [no-doc]

Browse Source
This commit is contained in:
Claude Dev
2026-10-05 16:05:32 +08:00
parent b954b02e75
commit 02acb485c6
4 changed files with 224 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sanguo_factor/ic_gate.py
+36 -8
View File
@@ -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:
sanguo_factor/monthly_batch.py
+41 -5
View File
@@ -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']})"
tests/factor/test_ic_gate.py
+50
View File
@@ -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"]
tests/factor/test_monthly_batch.py
+97
View File
@@ -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<len(new_names)
触发 stderr 显式告警(告警属信号语义,rc 仍 0);report 计数如实."""
import json
import numpy as np
import pandas as pd
import yaml
import sanguo_factor # noqa: F401 内存库注册(供 fund 因子名解析)
import sanguo_factor.monthly_batch as mb
reg_path = tmp_path / "reg.yaml"
reg_path.write_text(yaml.safe_dump({"factors": {
"llm_x": {"name": "llm_x", "hypothesis": "hyp-1", "status": "incubating",
"versions": [{"v": 1, "commit": "decompose", "params": {
"expression": "cs_rank(close)", "source": "bars_daily",
"origin": "decomposer"}}]},
"llm_y": {"name": "llm_y", "hypothesis": "hyp-2", "status": "incubating",
"versions": [{"v": 1, "commit": "decompose", "params": {
"expression": "cs_rank(volume)", "source": "bars_daily",
"origin": "decomposer"}}]},
# 手写库真名同批在库(闸对照腿)
"fund_gross_margin": {
"name": "fund_gross_margin", "hypothesis": "hyp-0",
"status": "incubating",
"versions": [{"v": 1, "commit": "hand", "params": {}}]}}}),
encoding="utf-8")
def fake_run(factor_names, start, end, eval_db, label, **kw):
vdir = kw["factor_values_out"]
os.makedirs(vdir, exist_ok=True)
rng = np.random.default_rng(3)
w = pd.DataFrame(rng.normal(size=(25, 12)),
index=pd.date_range("2026-09-01", periods=25),
columns=[f"{i}.SSE" for i in range(12)])
# llm_y 缺件:eval 单因子失败不落盘的形态
for n in ("llm_x", "fund_gross_margin"):
w.to_parquet(os.path.join(vdir, f"{n}.parquet"))
return {"done": 2}
monkeypatch.setattr(mb, "run_batch_eval", fake_run)
rc = mb.main(["--as-of", "2026-10-31", "--registry", str(reg_path),
"--eval-db", str(tmp_path / "e.db"),
"--report-dir", str(tmp_path / "rep")])
assert rc == 0 # 告警不改退出码
err = capsys.readouterr().err
assert "IC 正交闸" in err and "缺件" in err
rep = json.loads((tmp_path / "rep" / "ic_gate_monthly_2026-10.json"
).read_text())
assert rep["checked"] == 1 and rep["skipped_missing_values"] == 1
assert rep["compared_pairs"] == 1
# —— P3-11: 新因子判定改条目级 origin(双向漂移收敛) ——
def test_decomposer_origin_stable_across_manual_bump():
"""手工升版(新 params 不带 origin)不再无声退出闸视野;同注册表状态下
连续两月判定一致(不再随最新版 params 翻转)."""
import sanguo_factor.monthly_batch as mb
reg = {"factors": {
"llm_x": {"name": "llm_x", "hypothesis": "h", "status": "incubating",
"versions": [{"v": 1, "commit": "decompose", "params": {
"expression": "cs_rank(close)", "source": "bars_daily",
"origin": "decomposer"}}]}}}
month1 = mb.decomposer_new_names(reg, ["llm_x"])
# 月 2: 手工升版——v2 params 不携带 origin(旧判定在此无声退出闸视野)
reg["factors"]["llm_x"]["versions"].append(
{"v": 2, "commit": "hand", "params": {"expression": "cs_rank(volume)",
"source": "bars_daily"}})
month2 = mb.decomposer_new_names(reg, ["llm_x"])
assert month1 == ["llm_x"]
assert month2 == ["llm_x"]
def test_decomposer_origin_entry_level_wins_hand_factor_excluded():
"""条目级 origin 存在则以条目级为准;手写因子(origin 全无)不入闸视野."""
import sanguo_factor.monthly_batch as mb
reg = {"factors": {
# 条目级 origin=decomposer,版本 params 不带(未来写入形态)
"llm_entry": {"name": "llm_entry", "hypothesis": "h",
"status": "incubating", "origin": "decomposer",
"versions": [{"v": 1, "commit": "hand", "params": {}}]},
# 条目级显式改判 manual:压过旧版本 params 的 decomposer 戳
"llm_override": {"name": "llm_override", "hypothesis": "h",
"status": "incubating", "origin": "manual",
"versions": [{"v": 1, "commit": "decompose",
"params": {"origin": "decomposer"}}]},
# 手写因子:origin 全无,不入闸视野
"fa_hand": {"name": "fa_hand", "hypothesis": "h",
"status": "incubating",
"versions": [{"v": 1, "commit": "hand", "params": {}}]}}}
picked = mb.decomposer_new_names(reg, ["llm_entry", "llm_override", "fa_hand"])
assert picked == ["llm_entry"]