137 lines
6.4 KiB
Python
137 lines
6.4 KiB
Python
# sanguo_factor/exam_gate.py
|
||
"""考场一键化(#70③,决议 K 配套): 拟合候选→同批考场→建议判定报告.
|
||
|
||
候选与基准同一 run 同批快照(微扰混沌+跨批不可比教训,§9 工程坑#4);
|
||
判定=1/5/10d 三周期 ICIR 全不输且至少一胜→SUGGEST_PROMOTE(仅建议,
|
||
promotion 永不自动化=流水线宪法);三周期缺任一→INSUFFICIENT_PERIODS
|
||
(fail-closed,单周期对比不再退化出晋级建议);报告落 report-dir,md 自包含.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import math
|
||
import os
|
||
import sys
|
||
from datetime import datetime
|
||
from typing import Any, Callable
|
||
|
||
from sanguo_factor.batch_eval import run_batch_eval
|
||
from sanguo_factor.eval_store import default_eval_db_path, get_rows
|
||
from sanguo_factor.registry import get_factor
|
||
|
||
HORIZONS = ("1", "5", "10")
|
||
DEFAULT_BASELINE = "composite_quant12_v2a" # 生产冠军(在位基准)
|
||
DEFAULT_START = "2021-07-01" # h2 考场惯例(L2 UI 预填同款)
|
||
VERDICT_TEXT = {"SUGGEST_PROMOTE": "建议晋级(人裁,promotion 永不自动化)",
|
||
"NO_PROMOTE": "不晋级(考场未全胜)",
|
||
"DATA_INSUFFICIENT": "数据不足,人工研判",
|
||
"INSUFFICIENT_PERIODS": "周期覆盖不足,不建议晋级(fail-closed)"}
|
||
|
||
|
||
def _missing_icir(v: Any) -> bool:
|
||
"""ICIR 缺失判定(#91⑤): None 或 NaN 皆缺——NaN 与任何值比较恒 False,
|
||
不判胜负却计入 compared,可搭 SUGGEST_PROMOTE 顺风车,故与 None 同判."""
|
||
return v is None or (isinstance(v, float) and math.isnan(v))
|
||
|
||
|
||
def judge_exam(cand_row: dict, base_row: dict) -> dict:
|
||
"""三周期 ICIR 对照:全不输+至少一胜→建议晋级.
|
||
|
||
P3-16 fail-closed: 1/5/10d 任一周期任一侧缺 ICIR（缺=None 或 NaN，#91⑤）→INSUFFICIENT_PERIODS
|
||
(缺周期不再跳过不计,单周期对比不退化出晋级建议);三周期全缺=DATA_INSUFFICIENT.
|
||
"""
|
||
rows, wins, losses, compared, missing = [], 0, 0, 0, []
|
||
for h in HORIZONS:
|
||
cm = (cand_row.get("metrics") or {}).get(h) or {}
|
||
bm = (base_row.get("metrics") or {}).get(h) or {}
|
||
rows.append({"horizon": h,
|
||
"cand_icir": cm.get("icir"), "base_icir": bm.get("icir"),
|
||
"cand_t": cm.get("t_stat"), "base_t": bm.get("t_stat"),
|
||
"cand_n": cm.get("count"), "base_n": bm.get("count")})
|
||
if _missing_icir(cm.get("icir")) or _missing_icir(bm.get("icir")):
|
||
missing.append(h)
|
||
continue
|
||
compared += 1
|
||
if cm["icir"] > bm["icir"]:
|
||
wins += 1
|
||
elif cm["icir"] < bm["icir"]:
|
||
losses += 1
|
||
if compared == 0:
|
||
verdict = "DATA_INSUFFICIENT"
|
||
elif missing:
|
||
verdict = "INSUFFICIENT_PERIODS"
|
||
elif losses == 0 and wins >= 1:
|
||
verdict = "SUGGEST_PROMOTE"
|
||
else:
|
||
verdict = "NO_PROMOTE"
|
||
return {"verdict": verdict, "rows": rows, "missing_periods": missing,
|
||
"compared": compared, "wins": wins, "losses": losses}
|
||
|
||
|
||
def _fmt(v: Any, nd: int = 4) -> str:
|
||
return f"{v:.{nd}f}" if isinstance(v, (int, float)) else "n/a"
|
||
|
||
|
||
def render_report(candidate: str, baseline: str, start: str, end: str,
|
||
judged: dict) -> str:
|
||
lines = [f"# 考场判定 · {candidate} vs {baseline}",
|
||
f"_窗 {start}~{end} · 同批同 run 快照 · 生成 "
|
||
f"{datetime.now().isoformat(timespec='seconds')}_", "",
|
||
f"**判定: {VERDICT_TEXT[judged['verdict']]}**"
|
||
f"(胜 {judged['wins']} / 负 {judged['losses']} / 计 "
|
||
f"{judged['compared']} 周期)", ""]
|
||
if judged.get("missing_periods"):
|
||
lines.append(f"**缺周期: {', '.join(f'{h}d' for h in judged['missing_periods'])}"
|
||
"——ICIR 缺失,不进晋级对比(P3-16 fail-closed)**")
|
||
lines += ["| 周期 | 候选 ICIR | 基准 ICIR | 候选 t | 基准 t | 候选 n | 基准 n |",
|
||
"|------|-----------|-----------|--------|--------|--------|--------|"]
|
||
for r in judged["rows"]:
|
||
lines.append(f"| {r['horizon']}d | {_fmt(r['cand_icir'])} "
|
||
f"| {_fmt(r['base_icir'])} | {_fmt(r['cand_t'], 2)} "
|
||
f"| {_fmt(r['base_t'], 2)} | {r['cand_n'] or 'n/a'} "
|
||
f"| {r['base_n'] or 'n/a'} |")
|
||
lines += ["", "晋级走 version_registry transition + 人裁;"
|
||
"本报告永不自动转移(promotion 永不自动化)。"]
|
||
return "\n".join(lines) + "\n"
|
||
|
||
|
||
def main(argv: list[str] | None = None, *,
|
||
run: Callable[..., dict] | None = None,
|
||
fetch_rows: Callable[[str, str], list[dict]] | None = None) -> int:
|
||
import sanguo_factor # noqa: F401 导入即注册(合成层全链)
|
||
ap = argparse.ArgumentParser(description="考场一键化(#70③)")
|
||
ap.add_argument("--candidate", required=True, help="已注册因子名(含 composite_*/滚动/ui 档)")
|
||
ap.add_argument("--baseline", default=DEFAULT_BASELINE)
|
||
ap.add_argument("--start", default=DEFAULT_START)
|
||
ap.add_argument("--end", default=datetime.now().strftime("%Y-%m-%d"))
|
||
ap.add_argument("--eval-db", default=default_eval_db_path())
|
||
ap.add_argument("--report-dir", default="reports/factor_exam")
|
||
args = ap.parse_args(argv)
|
||
|
||
for role, name in (("候选", args.candidate), ("基准", args.baseline)):
|
||
if get_factor(name) is None:
|
||
print(f"[exam_gate] {role} {name} 不可解析(须为已注册因子名)",
|
||
file=sys.stderr)
|
||
return 2
|
||
label = f"exam_{args.candidate}_{args.end}"
|
||
runner = run or run_batch_eval
|
||
getter = fetch_rows or get_rows
|
||
summary = runner([args.candidate, args.baseline], args.start, args.end,
|
||
args.eval_db, label)
|
||
rows = {r["factor"]: r for r in getter(args.eval_db, summary["run_id"])}
|
||
cand_row = rows.get(args.candidate) or {"factor": args.candidate, "metrics": {}}
|
||
base_row = rows.get(args.baseline) or {"factor": args.baseline, "metrics": {}}
|
||
judged = judge_exam(cand_row, base_row)
|
||
os.makedirs(args.report_dir, exist_ok=True)
|
||
out = os.path.join(args.report_dir, # P3-17: 文件名带 HHMMSS
|
||
f"{label}_{datetime.now().strftime('%H%M%S')}.md")
|
||
with open(out, "w", encoding="utf-8") as f:
|
||
f.write(render_report(args.candidate, args.baseline,
|
||
args.start, args.end, judged))
|
||
print(f"[exam_gate] 判定={VERDICT_TEXT[judged['verdict']]} 报告={out}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|