fix(data): vintage自检补覆盖率撕裂告警——VPS真数实测暴露v1盲区:cashflow洗库中段8%vs两表100%但max_period同为Q2'26,仅看max报了OK漏掉撕裂;补=三报表coverage最大-最小>50pp告警(三报表同批披露稳态应同步,中段/某表大面积失败即触发);+1真形态测试 [vps]

scripts/data_platform/static_vintage_check.py

@@ -5,7 +5,7 @@
 背景: sanguo-ak-quarter 财报季全量洗库逐股重写 balance→income→cashflow,
 单表 ~8h、全程 20h+: 洗库中段两表 vintage 撕裂 (09-01 09:32 balance 已带
 中报而 income 停在 Q1'26 → value ROE 配对归零清仓)。本脚本盘前盘点各报表
-最新报告期分布, 跨表不一致/估值断更大声告警, 并落 vintage_status.json 供
+最新报告期分布, 跨表不一致/覆盖率撕裂/估值断更大声告警, 并落 vintage_status.json 供
 策略侧盘前预判 (契约: tables.<type> = {max_period, coverage, n_at_max,
 n_valid, n_empty, n_unreadable, newest_mtime}, checked_at 为扫描时刻)。
 
@@ -124,6 +124,15 @@ def check_all(root: Path) -> Tuple[dict, list]:
         alerts.append(
             "报表 vintage 撕裂: %s (跨表配对将缺期, 疑似洗库进行中/中段, "
             "策略盘前勿按新报告期决策)" % detail)
+    # 覆盖率撕裂: 三报表同批披露, 稳态 coverage 应同步; 最大-最小 >50pp
+    # = 洗库中段 / 某表大面积抓取失败 (09-01 晚实测形态: cashflow 中段 8%
+    # vs balance/income 100%, 但 max_period 同为 Q2'26, 仅看 max 会漏报)
+    covs = {t: status["tables"][t]["coverage"] for t in STATEMENT_TABLES}
+    if min(covs.values()) > 0 and max(covs.values()) - min(covs.values()) > 0.5:
+        detail = ", ".join(f"{t}={c:.0%}" for t, c in sorted(covs.items()))
+        alerts.append(
+            "表间覆盖率撕裂: %s (三报表同批披露应同步; 疑似洗库进行中"
+            "/某表大面积抓取失败)" % detail)
     status["valuation"] = valuation_freshness(root)
     v = status["valuation"]
     if v.get("age_hours") is not None and v["age_hours"] > VALUATION_STALE_HOURS:

tests/data_platform/test_static_vintage_check.py

@@ -76,6 +76,19 @@ class TestCheckAll:
         _, alerts = svc.check_all(tmp_path)
         assert alerts == []
 
+    def test_coverage_tear_alert(self, tmp_path):
+        """max_period 相同但覆盖率撕裂 (洗库中段形态): 某表大面积滞后 → 告警。
+
+        09-01 晚真数实测暴露的 v1 盲区: cashflow 洗到 8% vs 两表 100%,
+        max 同为 Q2'26, 仅看 max 会漏报。
+        """
+        _mk_table(tmp_path, "balance", {"a": ["2026-06-30"], "b": ["2026-06-30"]})
+        _mk_table(tmp_path, "income", {"a": ["2026-06-30"], "b": ["2026-06-30"]})
+        _mk_table(tmp_path, "cashflow", {
+            "a": ["2026-06-30"], "b": ["2026-03-31"], "c": ["2026-03-31"]})
+        _, alerts = svc.check_all(tmp_path)
+        assert any("覆盖率撕裂" in a and "cashflow=33%" in a for a in alerts)
+
     def test_valuation_stale_alert(self, tmp_path):
         """估值表 (每日 ak-eod) 超 48h 未刷新 → 断更告警。"""
         _mk_table(tmp_path, "balance", {"a": ["2026-06-30"]})