feat(data): vintage自检扩涨停池缺日防线——策略session共识「断档=不可逆损失须尽快」;per-date家族语义=每工作日(含节假日,空数据也写空文件)应有1个parquet,文件缺=当晚schtask漏跑=洞;洞≤42日历日(≈30交易日回补窗)→告警点名缺日+手动--start/--end回补指引,滑出窗=永久缺只入json不刷告警(防永久exit3疲劳);检查窗=首文件日→昨天(当日19:30才落盘不算洞);panel块进vintage_status.json契约(additive零破坏);+4测试(近窗洞告警/连续无洞/永久洞静默入json/目录缺不崩)13绿 [vps]

scripts/data_platform/static_vintage_check.py

@@ -29,6 +29,11 @@ DEFAULT_ROOT = r"C:\sanguo_vnpy_v2\data\static"
 STATEMENT_TABLES = ("balance", "income", "cashflow")  # 有 REPORT_DATE 列
 VALUATION_STALE_HOURS = 48.0  # 估值表(每日19:00 ak-eod)最新mtime超此数=断更
 EMPTY_PARQUET_MIN_BYTES = 1024  # 与 akshare_static_download 同口径
+# 涨停池三件套 (2026-09-02 入列 ak-events): per-date 家族每个工作日(含节假日,
+# 空数据也写空文件)应有 1 个 parquet, 文件缺 = 当晚 schtask 漏跑 = 洞。
+# 端点只支持最近 ~30 交易日 → 洞滑出窗口 = 永久缺 (不可逆), 09-02 与策略侧共识防线。
+PANEL_TYPES = ("zt_pool", "zt_pool_zbgc", "zt_pool_dtgc")
+PANEL_BACKFILL_CALENDAR_DAYS = 42  # ≈30 交易日回补窗的日历日近似
 
 
 def _norm_period(v) -> str:
@@ -100,6 +105,42 @@ def valuation_freshness(root: Path) -> dict:
     return info
 
 
+def panel_gap_check(root: Path) -> dict:
+    """涨停池三件套缺日检查: 枚举首文件日 → 昨天的工作日, 文件缺 = 洞。
+
+    可回补洞 (≤PANEL_BACKFILL_CALENDAR_DAYS 天) → check_all 出告警; 更老的洞
+    判永久缺, 只入 json 不刷告警 (避免永久 exit 3 告警疲劳)。当日 19:30 才
+    落盘, 检查窗不含今天。
+    """
+    out = {}
+    today = datetime.date.today()
+    for t in PANEL_TYPES:
+        dates = set()
+        d = root / t
+        if d.is_dir():
+            for m in d.glob(f"*_{t}.parquet"):
+                try:
+                    dates.add(
+                        datetime.datetime.strptime(m.name[:8], "%Y%m%d").date())
+                except ValueError:
+                    continue
+        entry = {"n_files": len(dates), "first_date": None,
+                 "holes_backfillable": [], "holes_permanent": []}
+        if dates:
+            first = min(dates)
+            entry["first_date"] = first.isoformat()
+            cur = first
+            while cur < today:
+                if cur.weekday() < 5 and cur not in dates:
+                    key = ("holes_backfillable"
+                           if (today - cur).days <= PANEL_BACKFILL_CALENDAR_DAYS
+                           else "holes_permanent")
+                    entry[key].append(cur.isoformat())
+                cur += datetime.timedelta(days=1)
+        out[t] = entry
+    return out
+
+
 def check_all(root: Path) -> Tuple[dict, list]:
     """盘点三报表 + 估值新鲜度。返 (status, alerts); alerts 非空 → 退出码 3。"""
     status = {
@@ -139,6 +180,14 @@ def check_all(root: Path) -> Tuple[dict, list]:
         alerts.append(
             "估值表断更: 最新 mtime %s 距今 %.1fh (>%.0fh, ak-eod 未跑成?)"
             % (v["newest_mtime"], v["age_hours"], VALUATION_STALE_HOURS))
+    # 涨停池三件套缺日 (09-02 入列): 可回补洞告警, 永久缺只入 json
+    status["panel"] = panel_gap_check(root)
+    for t, e in status["panel"].items():
+        if e["holes_backfillable"]:
+            alerts.append(
+                "[%s] 缺日(回补窗内): %s (当晚 schtask 漏跑; 30 交易日内手动 "
+                "--start 缺日 --end 缺日 可补, 滑出窗=永久缺)" % (
+                    t, ",".join(e["holes_backfillable"])))
     return status, alerts
 
 
@@ -177,6 +226,10 @@ def main() -> int:
     v = status["valuation"]
     print("[vintage] valuation   newest=%s age=%sh" % (
         v.get("newest_mtime"), v.get("age_hours")))
+    for t, e in status["panel"].items():
+        print("[vintage] %-13s files=%d 洞(可回补)=%d 洞(永久)=%d first=%s" % (
+            t, e["n_files"], len(e["holes_backfillable"]),
+            len(e["holes_permanent"]), e["first_date"]))
     print("[vintage] json → %s" % args.json_out)
     if alerts:
         for a in alerts: