324 lines
15 KiB
Python
324 lines
15 KiB
Python
# tests/factor/test_monthly_review.py
|
|
"""月度批评 TDD——spec §4.2 决议 H: 双机各跑,数据截止对齐自然月末."""
|
|
import json
|
|
import math
|
|
import os
|
|
|
|
import pytest
|
|
|
|
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,
|
|
render_markdown,
|
|
)
|
|
|
|
|
|
@pytest.fixture()
|
|
def eval_db(tmp_path):
|
|
p = str(tmp_path / "factor_eval.db")
|
|
eval_store.init_db(p)
|
|
# 8 月批(先建,墙钟/数据端双旧):用于验证 pick 选最新
|
|
# (顺序勘正 09-24:原 fixture 反序建批——sep 先建 aug 后建,created_at 秒级
|
|
# 存储下两批跨秒则 aug 墙钟更新被 pick 选中,同秒则靠 SQLite 平序侥幸,
|
|
# test_extract_points_shape 间歇红的根因;自然序+pick 同秒 end 破平=确定)
|
|
r0 = eval_store.create_run(p, label="aug", universe="zzq800", symbols_count=500,
|
|
factors_total=2, start="2025-08-01", end="2026-08-20",
|
|
params={})
|
|
eval_store.save_results(p, r0, [
|
|
{"factor": "pledge_net_chg", "category": "alt", "expression": "e1",
|
|
"metrics": {"1": {"count": 500, "t_stat": 1.4, "ic_mean": 0.030}}},
|
|
])
|
|
eval_store.finish_run(p, r0, "done", 1)
|
|
# 9 月批(后建=end/created_at 双新):end=09-18(未到月末也可跑,as_of 门控),两因子
|
|
r1 = eval_store.create_run(p, label="sep", universe="zzq800", symbols_count=512,
|
|
factors_total=2, start="2025-09-01", end="2026-09-18",
|
|
params={})
|
|
eval_store.save_results(p, r1, [
|
|
{"factor": "pledge_net_chg", "category": "alt", "expression": "e1",
|
|
"metrics": {"1": {"count": 512, "t_stat": 1.2, "ic_mean": 0.028}}},
|
|
{"factor": "fa_gross_margin", "category": "fa", "expression": "e2",
|
|
"metrics": {"1": {"count": 498, "t_stat": 0.9, "ic_mean": 0.011}}},
|
|
])
|
|
eval_store.finish_run(p, r1, "done", 2)
|
|
return p
|
|
|
|
|
|
@pytest.fixture()
|
|
def registry_yaml(tmp_path):
|
|
# pledge_net_chg: promoted(基准 4.01);s01_buzz: incubating(观察)
|
|
from sanguo_factor.version_registry import (add_version, load_registry,
|
|
save_registry, transition,
|
|
upsert_factor)
|
|
reg = {"factors": {}}
|
|
upsert_factor(reg, "pledge_net_chg", hypothesis="H-03")
|
|
transition(reg, "pledge_net_chg", "assessable")
|
|
transition(reg, "pledge_net_chg", "promoted", promoted_at="2026-09-22",
|
|
promotion_t=4.01)
|
|
add_version(reg, "pledge_net_chg", commit="574808c",
|
|
params={"window": 180}, effective_from="2026-09-22")
|
|
upsert_factor(reg, "s01_buzz", hypothesis="H-07")
|
|
p = str(tmp_path / "factor_registry.yaml")
|
|
save_registry(p, reg)
|
|
return p
|
|
|
|
|
|
def test_pick_eval_run_prefers_latest_before_asof(eval_db):
|
|
run = pick_eval_run(eval_db, as_of="2026-09-30")
|
|
assert run["label"] == "sep"
|
|
assert pick_eval_run(eval_db, as_of="2026-01-01") is None
|
|
|
|
|
|
def test_pick_eval_run_same_second_tiebreaks_by_end(eval_db):
|
|
"""created_at 秒级存储,同秒双批以 end 破平(数据端更新的批胜)——
|
|
不依赖 SQLite ORDER BY 平序(未定义行为,跨版本/平台可漂=flaky 源)."""
|
|
import sqlite3
|
|
conn = sqlite3.connect(eval_db)
|
|
conn.execute("UPDATE eval_runs SET created_at='2026-09-24T10:00:00'")
|
|
conn.commit()
|
|
conn.close()
|
|
run = pick_eval_run(eval_db, as_of="2026-09-30")
|
|
assert run["label"] == "sep"
|
|
|
|
|
|
def test_extract_points_shape(eval_db):
|
|
run = pick_eval_run(eval_db, as_of="2026-09-30")
|
|
pts = extract_points(eval_db, run)
|
|
assert pts["pledge_net_chg"] == {"t": 1.2, "ic_mean": 0.028, "count": 512}
|
|
assert pts["fa_gross_margin"]["t"] == 0.9
|
|
|
|
|
|
def test_build_report_verdict_and_collective(eval_db, registry_yaml):
|
|
from sanguo_factor.version_registry import load_registry
|
|
reg = load_registry(registry_yaml)
|
|
run = pick_eval_run(eval_db, "2026-09-30")
|
|
cur = extract_points(eval_db, run)
|
|
rep = build_report(reg, cur, history_points={}, as_of="2026-09-30", host="nas")
|
|
# pledge_net_chg: 当月 t=1.2 < 4.01*0.5 且 <2.0 → 第 1 个月 yellow
|
|
v = rep["verdicts"]["pledge_net_chg"]
|
|
assert v["state"] == "yellow" and v["months_consecutive"] == 1
|
|
# 集体水位: 只有一个 promoted,t=1.2 → median=1.2 不破 1.0
|
|
assert rep["collective"]["is_collective_decay"] is False
|
|
# 观察面: s01_buzz 无批记录 → red
|
|
assert rep["observations"]["s01_buzz"]["status"] == "red"
|
|
# 当月点已存入 monthly_points(下月续算原料;codex review 后带 ic_mean)
|
|
assert rep["monthly_points"]["pledge_net_chg"][-1] == {
|
|
"month": "2026-09", "t": 1.2, "ic_mean": 0.028}
|
|
|
|
|
|
def test_history_points_extend_streak(eval_db, registry_yaml, tmp_path):
|
|
from sanguo_factor.version_registry import load_registry
|
|
# 往月 JSON:7-8 月都已弱 → 本月再弱即第 3 个月=alert
|
|
hist = {"pledge_net_chg": [
|
|
{"month": "2026-07", "t": 1.1}, {"month": "2026-08", "t": 1.0}]}
|
|
out_dir = str(tmp_path / "reports")
|
|
os.makedirs(out_dir)
|
|
with open(os.path.join(out_dir, "vps_2026-08-31.json"), "w") as f:
|
|
json.dump({"host": "vps", "as_of": "2026-08-31", "monthly_points": hist}, f)
|
|
got = load_history_points(out_dir, exclude_file="nas_2026-09-30.json")
|
|
assert got["pledge_net_chg"] == hist["pledge_net_chg"]
|
|
reg = load_registry(registry_yaml)
|
|
run = pick_eval_run(eval_db, "2026-09-30")
|
|
rep = build_report(reg, extract_points(eval_db, run), history_points=got,
|
|
as_of="2026-09-30", host="nas")
|
|
assert rep["verdicts"]["pledge_net_chg"]["state"] == "alert"
|
|
|
|
|
|
def test_main_end_to_end_exit_codes(eval_db, registry_yaml, tmp_path):
|
|
out_dir = str(tmp_path / "out")
|
|
rc0 = main(["--as-of", "2026-09-30", "--registry", registry_yaml,
|
|
"--eval-db", eval_db, "--out-dir", out_dir, "--host", "nas"])
|
|
assert rc0 == 0 # 当月仅 yellow(第 1 个月),无 alert 无集体信号
|
|
jf = os.path.join(out_dir, "nas_2026-09-30.json")
|
|
mf = os.path.join(out_dir, "nas_2026-09-30.md")
|
|
assert os.path.exists(jf) and os.path.exists(mf)
|
|
with open(jf) as f:
|
|
rep = json.load(f)
|
|
assert rep["host"] == "nas" and rep["as_of"] == "2026-09-30"
|
|
|
|
|
|
def test_unregistered_factor_flagged(eval_db, registry_yaml):
|
|
"""J2:评估批有记录但注册表无条目→unregistered 警告,不进晋级候选."""
|
|
from sanguo_factor.version_registry import load_registry
|
|
reg = load_registry(registry_yaml)
|
|
run = pick_eval_run(eval_db, "2026-09-30")
|
|
cur = extract_points(eval_db, run)
|
|
cur["ghost_alpha"] = {"t": 9.9, "ic_mean": 0.05, "count": 500} # 批里的野因子
|
|
rep = build_report(reg, cur, history_points={}, as_of="2026-09-30", host="nas")
|
|
# fixture 的 sep 批含未注册的 fa_gross_margin+人工塞入的 ghost_alpha,都属无血统
|
|
assert rep["unregistered"] == ["fa_gross_margin", "ghost_alpha"]
|
|
assert "ghost_alpha" not in rep["verdicts"] # 不判衰减
|
|
assert "ghost_alpha" not in rep["pre_gate"] # 不进晋级候选
|
|
md = render_markdown(rep)
|
|
assert "无血统" in md and "ghost_alpha" in md
|
|
|
|
|
|
def test_pre_gate_collinear_review(eval_db, registry_yaml):
|
|
"""J3:与已晋级因子月度 t 序列完全同步→共线 REVIEW(|r|>=0.95 起步默认)."""
|
|
from sanguo_factor.version_registry import (load_registry, transition,
|
|
upsert_factor)
|
|
reg = load_registry(registry_yaml)
|
|
upsert_factor(reg, "fa_gross_margin", hypothesis="H-08") # fixture 无此因子,自建 assessable
|
|
transition(reg, "fa_gross_margin", "assessable")
|
|
months = [f"2026-0{m}" for m in range(1, 7)]
|
|
strong = [3.0 + 0.1 * i for i in range(6)] # pledge(已晋级)半年 t
|
|
copy_ = [2.5 + 0.1 * i for i in range(6)] # fa_gross_margin 完全同步
|
|
hist = {"pledge_net_chg": [{"month": m, "t": v} for m, v in zip(months, strong)],
|
|
"fa_gross_margin": [{"month": m, "t": v} for m, v in zip(months, copy_)]}
|
|
run = pick_eval_run(eval_db, "2026-09-30")
|
|
cur = extract_points(eval_db, run) # fa 当月 t=0.9(批里有记录)
|
|
rep = build_report(reg, cur, history_points=hist, as_of="2026-09-30", host="nas")
|
|
pg = rep["pre_gate"]["fa_gross_margin"] # assessable→进 pre-gate
|
|
assert pg["flag"] == "REVIEW"
|
|
assert pg["corr_with"] == "pledge_net_chg" and pg["max_corr"] >= 0.95
|
|
assert "Pre-gate" in render_markdown(rep)
|
|
|
|
|
|
def test_pre_gate_sample_short_is_not_correlation(eval_db, registry_yaml):
|
|
"""J3:月度点 n<6→相关不算(样本不足),不算误判共线."""
|
|
from sanguo_factor.version_registry import (load_registry, transition,
|
|
upsert_factor)
|
|
reg = load_registry(registry_yaml)
|
|
upsert_factor(reg, "fa_gross_margin", hypothesis="H-08")
|
|
transition(reg, "fa_gross_margin", "assessable")
|
|
hist = {"pledge_net_chg": [{"month": "2026-08", "t": 3.0},
|
|
{"month": "2026-09", "t": 3.1}],
|
|
"fa_gross_margin": [{"month": "2026-08", "t": 2.0},
|
|
{"month": "2026-09", "t": 2.1}]}
|
|
run = pick_eval_run(eval_db, "2026-09-30")
|
|
rep = build_report(reg, extract_points(eval_db, run), history_points=hist,
|
|
as_of="2026-09-30", host="nas")
|
|
pg = rep["pre_gate"]["fa_gross_margin"]
|
|
assert pg["max_corr"] is None and "样本不足" in pg["corr_note"]
|
|
|
|
|
|
# —— #70① v2a=promoted 走衰减判定路径;v1=assessable 走观察路径 ——
|
|
def test_real_config_composite_verdict_paths():
|
|
import sanguo_factor # noqa: F401
|
|
from sanguo_factor.monthly_review import build_report
|
|
from sanguo_factor.version_registry import load_registry
|
|
|
|
reg_path = os.path.join(
|
|
os.path.dirname(os.path.dirname(os.path.abspath(sanguo_factor.__file__))),
|
|
"config", "factor_registry.yaml")
|
|
reg = load_registry(reg_path)
|
|
pts = {"composite_quant12_v2a": {"t": 20.0, "ic_mean": 0.05, "count": 200},
|
|
"composite_quant12": {"t": 19.0, "ic_mean": 0.05, "count": 200}}
|
|
rep = build_report(reg, pts, {}, as_of="2026-10-31", host="t")
|
|
v = rep["verdicts"]["composite_quant12_v2a"]
|
|
assert v["promotion_t"] == 26.32 and v["state"] == "ok"
|
|
assert "composite_quant12" in rep["observations"] # assessable=观察
|
|
assert "composite_quant12_v2a" not in rep["observations"]
|
|
|
|
|
|
# —— #70② 门边案例:assessable 且 t∈[1.8,2.0) 才进清单 ——
|
|
def _mini_reg():
|
|
return {"factors": {
|
|
"cand_edge": {"name": "cand_edge", "hypothesis": "x", "status": "assessable"},
|
|
"cand_far": {"name": "cand_far", "hypothesis": "x", "status": "assessable"},
|
|
"cand_low": {"name": "cand_low", "hypothesis": "x", "status": "assessable"},
|
|
"prom_low": {"name": "prom_low", "hypothesis": "x", "status": "promoted",
|
|
"promotion_t": 4.0},
|
|
}}
|
|
|
|
|
|
def test_edge_cases_band_filter():
|
|
from sanguo_factor.monthly_review import build_report
|
|
pts = {"cand_edge": {"t": 1.9, "ic_mean": 0.04, "count": 100},
|
|
"cand_far": {"t": 2.5, "ic_mean": 0.05, "count": 100},
|
|
"cand_low": {"t": 1.2, "ic_mean": 0.02, "count": 100},
|
|
"prom_low": {"t": 1.9, "ic_mean": 0.04, "count": 100}}
|
|
rep = build_report(_mini_reg(), pts, {}, as_of="2026-10-31", host="t")
|
|
assert list(rep["edge_cases"]) == ["cand_edge"] # 只收带内 assessable
|
|
assert rep["edge_cases"]["cand_edge"]["t"] == 1.9
|
|
|
|
|
|
def test_edge_cases_render_markdown():
|
|
from sanguo_factor.monthly_review import build_report, render_markdown
|
|
pts = {"cand_edge": {"t": 1.9, "ic_mean": 0.04, "count": 100}}
|
|
rep = build_report(_mini_reg(), pts, {}, as_of="2026-10-31", host="t")
|
|
md = render_markdown(rep)
|
|
assert "门边 1" in md # 摘要行计数
|
|
assert "cand_edge" in md and "1.90" in md # 清单一行一只
|
|
|
|
|
|
# —— 审计 P3-15: pre_gate REVIEW 阈值须引 _PROMOTION_BAR 常量(单一事实来源) ——
|
|
def test_pre_gate_threshold_follows_promotion_bar_constant(monkeypatch):
|
|
"""REVIEW 的 t 阈值= _PROMOTION_BAR:改常量即改门(硬编码 2.0 时本测试红)."""
|
|
import sanguo_factor.monthly_review as mr
|
|
monkeypatch.setattr(mr, "_PROMOTION_BAR", 2.5)
|
|
reg = {"factors": {"fa_a": {"name": "fa_a", "hypothesis": "x",
|
|
"status": "assessable"}}}
|
|
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 不进逐年
|
|
|
|
|
|
# —— codex review 修批(2026-10-10):point 带 ic_mean/非法 month 校验/退化样本 ——
|
|
def test_factor_stats_degenerate_samples():
|
|
"""n=2 同值→方差 0→tAll 如实 None(不除零不造 0)."""
|
|
s = factor_stats(_ammo_pts(("2025-09", 0.05), ("2025-10", 0.05)))
|
|
assert s["icAll"] == 0.05 and s["tAll"] is None
|
|
assert s["positiveRatio"] == 1.0
|
|
|
|
|
|
def test_factor_stats_invalid_month_not_in_by_year():
|
|
"""非法 month(bad-2025)不切前 4 字符进 byYear;ic 值聚合不受月串影响."""
|
|
pts = [{"month": "bad-2025", "ic_mean": 0.1, "t": None},
|
|
{"month": "2025-09", "ic_mean": 0.04, "t": None}]
|
|
s = factor_stats(pts)
|
|
assert s["byYear"] == {"2025": 0.04} # bad-2025 整条剔出逐年
|
|
assert s["icAll"] == round((0.1 + 0.04) / 2, 6) # 值聚合照收
|
|
|
|
|
|
def test_build_report_current_point_carries_ic_mean():
|
|
"""当月 point 须带 ic_mean——否则 factor_stats 漏最新一期、首跑全空."""
|
|
reg = {"factors": {"fa_x": {"name": "fa_x", "hypothesis": "h",
|
|
"status": "assessable"}}}
|
|
rep = build_report(reg, {"fa_x": {"t": 2.0, "ic_mean": 0.05, "count": 100}},
|
|
history_points={}, as_of="2026-09-30", host="t")
|
|
assert rep["monthly_points"]["fa_x"][-1]["ic_mean"] == 0.05
|
|
s = rep["factor_stats"]["fa_x"]
|
|
assert s["icAll"] == 0.05 # 首跑单点即有弹药
|
|
assert s["positiveRatio"] == 1.0
|
|
assert s["tAll"] is None # n=1 无合成 t,如实
|