diff --git a/scripts/data_platform/static_vintage_check.py b/scripts/data_platform/static_vintage_check.py index 2322bc3..6f13668 100644 --- a/scripts/data_platform/static_vintage_check.py +++ b/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. = {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: diff --git a/tests/data_platform/test_static_vintage_check.py b/tests/data_platform/test_static_vintage_check.py index b49e3ca..2428341 100644 --- a/tests/data_platform/test_static_vintage_check.py +++ b/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"]})