fix(factor): 判定与档案五项收敛——_PROMOTION_BAR 单源化(第二处硬编码改引常量)+exam_gate 缺周期 fail-closed(单周期对比不再 SUGGEST_PROMOTE,INSUFFICIENT_PERIODS 明示)+报告文件名带时间戳(同日重跑不覆盖)+v2a 条目补 evidence 指针+等权交集语义 docstring 明示;gap台账坏行容错(T6-N1非dict行降级跳过);因子档 §7.7 台账 (审计 P3-15/16/17/18/19+T6-N1) [vps]

config/factor_registry.yaml

@@ -119,3 +119,4 @@ factors:
       effective_to: null
     promoted_at: '2026-09-11'
     promotion_t: 26.32
+    evidence_run: ev_20260911_021832_2f68   # promotion_t 锚批(NAS factor_eval.db label=composite-v2a-h2,1d t=26.3192)

docs/factor_research/factor-system-design.md

@@ -241,6 +241,8 @@
 
 首月观察点（10-04 首班起）：composite_quant12 应走 observations「新鲜且覆盖足」、v2a 走 verdicts（promotion_t=26.32 生效；衰减线=双条件 AND：t<13.16 且 t<2.0，地板主导误报极低）；exam_gate 生产首跑=NAS 真跑一次验收（搭首班窗顺车）。
 
+审计杂项收敛批（2026-10-05，P3-15/16/17/18/19+T6-N1；上段 defer 台账「pre_gate_table 硬编码 2.0」就此收编）：exam_gate 缺周期语义改 fail-closed——1/5/10d 任一周期任一侧缺 ICIR→INSUFFICIENT_PERIODS 不建议晋级（原「缺周期跳过不计」可退化单周期对比出 SUGGEST_PROMOTE，已废）；考场报告文件名带 HHMMSS（同日重跑不覆盖，eval run label 不变）；pre_gate REVIEW 阈值收编 _PROMOTION_BAR 常量（负 t 侧无镜像门边带=负向晋级不存在，维持现状）；v2a 条目补 `evidence_run: ev_20260911_021832_2f68`（=promotion_t 26.32 锚批，NAS factor_eval.db label=composite-v2a-h2，1d t=26.3192 实查核验；**惯例：promoted 条目 promotion_t 必须可溯源到 evidence_run**）；composite_library.build_composite_expression docstring 明示等权交集语义（任一源缺→null，分母恒 N 不重归一）；data_gap_check 台账坏行（合法 JSON 非 dict）降级跳过+stderr（T6-N1）。
+
 ## 8. 前端与任务态
 
 - 投研三页（IC 排行榜/因子详情/批量评估）：终端风 tokens（深空黑/荧光青），路由挂侧边栏「投研」组；mockup 三份在 `docs/factor_research/*mockup.html`。**先读 frontend/src/styles/tokens.css 再做前端设计**（第一版亮色被打回的教训）。

sanguo_factor/composite_library.py

@@ -89,7 +89,13 @@ def _embed_term(expression: str, direction: str) -> str:
 
 
 def build_composite_expression(terms: list[str]) -> str:
-    """等权平均拼接: (项1 + 项2 + ...) / N."""
+    """等权平均拼接: (项1 + 项2 + ...) / N.
+
+    交集语义(P3-19,勿误读): polars 算术 null 传染——任一源当日缺值→合成值
+    当日 null;分母恒 N(源数)不按当日可得源重归一。已知限制: 事件流源
+    (gdhs/研报/质押等)上市早期或数据起始前覆盖受限,合成因子覆盖=全源交集,
+    短于单源;看覆盖先看 N 里各源的 null 界,勿按「部分源缺也出值」理解。
+    """
     return "(" + " + ".join(terms) + f") / {len(terms)}"
 
 

sanguo_factor/data_gap_check.py

@@ -133,6 +133,10 @@ def load_ledger_fps(path: str) -> set[str]:
                     fp = json.loads(line).get("fingerprint")
                 except json.JSONDecodeError:
                     continue
+                except AttributeError:
+                    print(f"[data_gap_check] 台账坏行(合法 JSON 非 dict)跳过: "
+                          f"{line}", file=sys.stderr)      # T6-N1 降级留痕
+                    continue
                 if fp:
                     fps.add(fp)
     except OSError as e:

sanguo_factor/exam_gate.py

@@ -3,7 +3,8 @@
 
 候选与基准同一 run 同批快照(微扰混沌+跨批不可比教训,§9 工程坑#4);
 判定=1/5/10d 三周期 ICIR 全不输且至少一胜→SUGGEST_PROMOTE(仅建议,
-promotion 永不自动化=流水线宪法);报告落 report-dir,md 自包含.
+promotion 永不自动化=流水线宪法);三周期缺任一→INSUFFICIENT_PERIODS
+(fail-closed,单周期对比不再退化出晋级建议);报告落 report-dir,md 自包含.
 """
 from __future__ import annotations
 
@@ -22,12 +23,17 @@ DEFAULT_BASELINE = "composite_quant12_v2a"   # 生产冠军(在位基准)
 DEFAULT_START = "2021-07-01"                 # h2 考场惯例(L2 UI 预填同款)
 VERDICT_TEXT = {"SUGGEST_PROMOTE": "建议晋级(人裁,promotion 永不自动化)",
                 "NO_PROMOTE": "不晋级(考场未全胜)",
-                "DATA_INSUFFICIENT": "数据不足,人工研判"}
+                "DATA_INSUFFICIENT": "数据不足,人工研判",
+                "INSUFFICIENT_PERIODS": "周期覆盖不足,不建议晋级(fail-closed)"}
 
 
 def judge_exam(cand_row: dict, base_row: dict) -> dict:
-    """三周期 ICIR 对照:全不输+至少一胜→建议晋级;缺周期跳过不计."""
-    rows, wins, losses, compared = [], 0, 0, 0
+    """三周期 ICIR 对照:全不输+至少一胜→建议晋级.
+
+    P3-16 fail-closed: 1/5/10d 任一周期任一侧缺 ICIR→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 {}
@@ -36,6 +42,7 @@ def judge_exam(cand_row: dict, base_row: dict) -> dict:
                      "cand_t": cm.get("t_stat"), "base_t": bm.get("t_stat"),
                      "cand_n": cm.get("count"), "base_n": bm.get("count")})
         if cm.get("icir") is None or bm.get("icir") is None:
+            missing.append(h)
             continue
         compared += 1
         if cm["icir"] > bm["icir"]:
@@ -44,11 +51,13 @@ def judge_exam(cand_row: dict, base_row: dict) -> dict:
             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,
+    return {"verdict": verdict, "rows": rows, "missing_periods": missing,
             "compared": compared, "wins": wins, "losses": losses}
 
 
@@ -63,9 +72,12 @@ def render_report(candidate: str, baseline: str, start: str, end: str,
              f"{datetime.now().isoformat(timespec='seconds')}_", "",
              f"**判定: {VERDICT_TEXT[judged['verdict']]}**"
              f"(胜 {judged['wins']} / 负 {judged['losses']} / 计 "
-             f"{judged['compared']} 周期)", "",
-             "| 周期 | 候选 ICIR | 基准 ICIR | 候选 t | 基准 t | 候选 n | 基准 n |",
-             "|------|-----------|-----------|--------|--------|--------|--------|"]
+             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)} "
@@ -104,7 +116,8 @@ def main(argv: list[str] | None = None, *,
     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, f"{label}.md")
+    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))

sanguo_factor/monthly_review.py

@@ -23,7 +23,8 @@ from sanguo_factor.observe import observe_factor
 from sanguo_factor.version_registry import load_registry
 
 _EDGE_LO = 1.8        # 门边下沿:晋级 bar 2.0 的 10% 边带(起步默认,首年校准)
-_PROMOTION_BAR = 2.0  # pre_gate 同款晋级 t 门槛(单一事实来源引用,勿改一处)
+_PROMOTION_BAR = 2.0  # 晋级 t 门槛(单一事实来源,pre_gate_table 阈值引此常量)
+# 负 t 侧无镜像门边带:负向晋级在本体系不存在(审计 P3-15 注,维持现状)
 
 
 def pick_eval_run(db_path: str, as_of: str) -> dict | None:
@@ -131,7 +132,8 @@ def pre_gate_table(registry: dict, monthly_points: dict[str, list[dict]],
                    current_points: dict[str, dict]) -> dict[str, dict]:
     """J3 pre-gate:晋级候选(assessable/promoted/decaying)硬指标+与已晋级因子月度 t 相关.
 
-    共线阈 |r|>=0.95(起步默认,IC 0.961 教训);t<2.0 或共线→REVIEW(人审只看门边案例).
+    共线阈 |r|>=0.95(起步默认,IC 0.961 教训);t<_PROMOTION_BAR 或共线→REVIEW
+    (人审只看门边案例;阈值单一事实来源=模块级 _PROMOTION_BAR).
     """
     peer_names = [n for n, e in registry["factors"].items()
                   if e["status"] in ("promoted", "decaying")]
@@ -155,7 +157,7 @@ def pre_gate_table(registry: dict, monthly_points: dict[str, list[dict]],
         flag = "PASS"
         if cur.get("t") is None:
             flag = "NO_DATA"
-        elif cur.get("t") < 2.0:
+        elif cur.get("t") < _PROMOTION_BAR:
             flag = "REVIEW"
         elif max_r is not None and abs(max_r) >= 0.95:
             flag = "REVIEW"

tests/factor/test_data_gap_check.py

@@ -290,3 +290,14 @@ def test_append_ledger_failure_degrades_not_blocks(monkeypatch, tmp_path,
     assert dgc.maybe_open_gap_issue({"fa_x": ["s_new"]}, "m.yaml",
                                     ledger_path=str(bad_ledger)) == 77
     assert "台账写入失败" in capsys.readouterr().err
+
+
+def test_load_ledger_fps_tolerates_non_dict_json_line(tmp_path, capsys):
+    """T6-N1: 台账坏行=合法 JSON 非 dict(如 [1,2])→降级跳过该行+stderr,正常行仍生效."""
+    import sanguo_factor.data_gap_check as dgc
+    p = tmp_path / "ledger.jsonl"
+    p.write_text("[1, 2]\n"
+                 + _json.dumps({"fingerprint": "abcd1234", "issue": 7}) + "\n",
+                 encoding="utf-8")
+    assert dgc.load_ledger_fps(str(p)) == {"abcd1234"}
+    assert "非 dict" in capsys.readouterr().err

tests/factor/test_exam_gate.py

@@ -24,10 +24,8 @@ def test_judge_any_loss_no_promote():
     assert j["verdict"] == "NO_PROMOTE" and j["losses"] == 1
 
 
-def test_judge_missing_horizon_skipped():
-    j = judge_exam(_row({"1": 0.30, "5": 0.40}),                # 候选缺 10d
-                   _row({"1": 0.28, "5": 0.38, "10": 0.35}))
-    assert j["verdict"] == "SUGGEST_PROMOTE" and j["compared"] == 2
+# (P3-16 后缺周期不再跳过不计,原 test_judge_missing_horizon_skipped 的
+#  SUGGEST_PROMOTE 语义由 test_judge_missing_any_horizon_insufficient_periods 取代)
 
 
 def test_judge_no_overlap_data_insufficient():
@@ -71,6 +69,72 @@ def test_main_runs_candidate_and_baseline_same_run(tmp_path, capsys, monkeypatch
               run=fake_run, fetch_rows=fake_rows)
     assert rc == 0
     assert sorted(calls["names"]) == ["base_y", "cand_x"]     # 同批铁律
-    assert calls["label"] == "exam_cand_x_2026-10-03"
-    assert (tmp_path / "exam_cand_x_2026-10-03.md").exists()
+    assert calls["label"] == "exam_cand_x_2026-10-03"   # eval run label 不变(P3-17 只动报告件名)
+    assert len(list(tmp_path.glob("exam_cand_x_2026-10-03_??????.md"))) == 1
     assert "建议晋级" in capsys.readouterr().out
+
+
+# —— 审计 P3-16: 缺任一周期→INSUFFICIENT_PERIODS fail-closed(单周期对比不再晋级) ——
+def test_judge_missing_any_horizon_insufficient_periods():
+    j = judge_exam(_row({"1": 0.30, "5": 0.40}),                # 候选缺 10d
+                   _row({"1": 0.28, "5": 0.38, "10": 0.35}))
+    assert j["verdict"] == "INSUFFICIENT_PERIODS"
+    assert j["missing_periods"] == ["10"]
+    assert (j["wins"], j["compared"]) == (2, 2)   # 有对比分但判定 fail-closed,不作晋级依据
+
+
+def test_judge_missing_on_base_side_also_fails_closed():
+    j = judge_exam(_row({"1": 0.30, "5": 0.40, "10": 0.35}),
+                   _row({"1": 0.28, "5": 0.38}))                # 基准缺 10d
+    assert j["verdict"] == "INSUFFICIENT_PERIODS" and j["missing_periods"] == ["10"]
+
+
+def test_render_report_names_missing_periods():
+    j = judge_exam(_row({"1": 0.30, "5": 0.40}),
+                   _row({"1": 0.28, "5": 0.38, "10": 0.35}))
+    md = render_report("cand_x", "base_y", "2021-07-01", "2026-10-03", j)
+    assert "周期覆盖不足" in md and "缺周期" in md and "10d" in md   # 明示缺哪期
+
+
+# —— 审计 P3-17: 报告文件名带 HHMMSS,同日重跑不覆盖既有报告 ——
+class _FakeStamp:
+    def __init__(self, n):
+        self._n = n
+
+    def strftime(self, fmt):                       # exam_gate 只用 %H%M%S
+        return f"1200{self._n:02d}" if fmt == "%H%M%S" else "000000"
+
+    def isoformat(self, timespec=None):
+        return f"2026-10-05T12:00:0{self._n}"
+
+
+class _FakeDT:
+    seq = 0
+
+    @staticmethod
+    def now():
+        _FakeDT.seq += 1
+        return _FakeStamp(_FakeDT.seq)
+
+
+def test_main_rerun_same_day_keeps_both_reports(tmp_path, capsys, monkeypatch):
+    import sanguo_factor.exam_gate as eg
+    monkeypatch.setattr(eg, "datetime", _FakeDT)
+
+    def fake_run(names, start, end, eval_db, label, **kw):
+        return {"run_id": "r1", "factors_total": 2, "factors_done": 2}
+
+    def fake_rows(db, run_id):
+        return [_row({"1": 0.30, "5": 0.40, "10": 0.35}, factor="cand_x"),
+                _row({"1": 0.28, "5": 0.38, "10": 0.30}, factor="base_y")]
+
+    monkeypatch.setattr(eg, "get_factor", lambda n: {"expr": "x"})
+    argv = ["--candidate", "cand_x", "--baseline", "base_y",
+            "--end", "2026-10-03", "--eval-db", str(tmp_path / "e.db"),
+            "--report-dir", str(tmp_path)]
+    assert main(argv, run=fake_run, fetch_rows=fake_rows) == 0
+    assert main(argv, run=fake_run, fetch_rows=fake_rows) == 0
+    reports = sorted(tmp_path.glob("exam_cand_x_2026-10-03_??????.md"))
+    assert len(reports) == 2                        # 两跑两件,不覆盖
+    bodies = [p.read_text(encoding="utf-8") for p in reports]
+    assert len(set(bodies)) == 2                    # 生成时刻不同,内容确不同件

tests/factor/test_monthly_review.py

@@ -241,3 +241,15 @@ def test_edge_cases_render_markdown():
     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