diff --git a/scripts/data_platform/static_vintage_check.py b/scripts/data_platform/static_vintage_check.py index 6f13668..5b971e4 100644 --- a/scripts/data_platform/static_vintage_check.py +++ b/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: diff --git a/tests/data_platform/test_static_vintage_check.py b/tests/data_platform/test_static_vintage_check.py index 2428341..e56f475 100644 --- a/tests/data_platform/test_static_vintage_check.py +++ b/tests/data_platform/test_static_vintage_check.py @@ -2,8 +2,10 @@ """静态表 vintage 一致性自检测试 (2026-09-01 洗库撕裂事故防线④)。 核心契约: 跨表最大报告期不一致 → 告警 (退出码 3) + vintage_status.json -落盘 (策略侧盘前预判数据源); 一致 → OK (退出码 0); 估值断更 → 告警。 +落盘 (策略侧盘前预判数据源); 一致 → OK (退出码 0); 估值断更 → 告警; +涨停池三件套缺日 (09-02 入列): 可回补洞 → 告警, 永久缺 → 只入 json。 """ +import datetime import json import os import time @@ -27,6 +29,34 @@ def _mk_table(root, name, stocks_periods): return d +def _mk_ok_statements(root): + for t in ("balance", "income", "cashflow"): + _mk_table(root, t, {"a": ["2026-06-30"]}) + + +def _recent_weekdays(n, drop_dates=()): + """最近 n 个「昨天及以前」的工作日 (不写死日期), drop_dates 命中则跳过。""" + out = [] + cur = datetime.date.today() - datetime.timedelta(days=1) + while len(out) < n: + if cur.weekday() < 5 and cur not in drop_dates: + out.append(cur) + cur -= datetime.timedelta(days=1) + return out + + +def _weekdays_since(days_back): + """今天-days_back → 昨天 区间内全部工作日 (含边界对齐)。""" + out = [] + cur = datetime.date.today() - datetime.timedelta(days=days_back) + end = datetime.date.today() - datetime.timedelta(days=1) + while cur <= end: + if cur.weekday() < 5: + out.append(cur) + cur += datetime.timedelta(days=1) + return out + + # ======================== 单表扫描 ======================== class TestScanStatementTable: @@ -101,6 +131,58 @@ class TestCheckAll: assert any("估值表断更" in a for a in alerts) +class TestPanelGapCheck: + """涨停池三件套缺日检查 (30 交易日回补窗, 滑出=永久缺不可逆)。""" + + def _mk_dated_files(self, root, panel_type, dates): + d = root / panel_type + d.mkdir(parents=True, exist_ok=True) + for dt in dates: + (d / f"{dt.strftime('%Y%m%d')}_{panel_type}.parquet").write_bytes(b"x" * 10) + + def test_hole_within_window_alerts(self, tmp_path): + """近窗工作日缺 1 天 → 告警点名缺日 (schtask 漏跑)。""" + _mk_ok_statements(tmp_path) + hole = _recent_weekdays(5)[-1] # 5 个工作日前的那天挖掉 + dates = _recent_weekdays(30, drop_dates=(hole,)) + self._mk_dated_files(tmp_path, "zt_pool", dates) + _, alerts = svc.check_all(tmp_path) + assert any("zt_pool" in a and "缺日" in a + and hole.isoformat() in a for a in alerts) + + def test_all_present_no_alert(self, tmp_path): + """连续工作日全有 → 无缺日告警 (节假日空文件语义=文件存在)。""" + _mk_ok_statements(tmp_path) + self._mk_dated_files(tmp_path, "zt_pool", _recent_weekdays(30)) + _, alerts = svc.check_all(tmp_path) + assert not any("缺日" in a for a in alerts) + + def test_permanent_hole_no_alert_but_in_json(self, tmp_path): + """滑出回补窗的老洞 → 不刷告警 (避免永久 exit 3), 但入 json。 + + 构造: 首文件=120 天前的工作日, 近 45 天工作日全有, 中段全缺 → + 中段洞均 >42 天 = 永久, 回补窗内无洞。 + """ + _mk_ok_statements(tmp_path) + old = datetime.date.today() - datetime.timedelta(days=120) + while old.weekday() >= 5: + old -= datetime.timedelta(days=1) # 对齐到工作日才是「应有文件」日 + dates = [old] + _weekdays_since(45) + self._mk_dated_files(tmp_path, "zt_pool_dtgc", dates) + status, alerts = svc.check_all(tmp_path) + assert not any("zt_pool_dtgc" in a and "缺日" in a for a in alerts) + assert old.isoformat() not in status["panel"]["zt_pool_dtgc"]["holes_permanent"] + assert status["panel"]["zt_pool_dtgc"]["holes_permanent"] # 中段洞在列 + assert status["panel"]["zt_pool_dtgc"]["holes_backfillable"] == [] + + def test_missing_dir_no_crash_no_alert(self, tmp_path): + """目录不存在 (从未入列的环境) → 不崩不告警, json 记 n_files=0。""" + _mk_ok_statements(tmp_path) + status, alerts = svc.check_all(tmp_path) + assert not any("缺日" in a for a in alerts) + assert status["panel"]["zt_pool"]["n_files"] == 0 + + # ======================== main / JSON 契约 ======================== class TestMain: