From 8dc3f494b7610b377855f27e16427c166e73351f Mon Sep 17 00:00:00 2001 From: claude_dev Date: Tue, 1 Sep 2026 19:03:53 +0800 Subject: [PATCH] =?UTF-8?q?feat(data):=20=E9=9D=99=E6=80=81=E8=A1=A8?= =?UTF-8?q?=E8=B0=83=E5=BA=A6=E6=8C=AA=E5=91=A8=E6=9C=AB+vintage=E7=9B=98?= =?UTF-8?q?=E5=89=8D=E8=87=AA=E6=A3=80=E2=80=94=E2=80=94=E2=91=A2sanguo-ak?= =?UTF-8?q?-quarter=E6=9C=88=E5=86=851=E5=8F=B702:00=E6=94=B9=E8=B4=A2?= =?UTF-8?q?=E6=8A=A5=E6=9C=88=E9=A6=96=E4=B8=AA=E5=91=A8=E5=85=AD02:00(/d?= =?UTF-8?q?=20SAT,=E5=91=A8=E5=85=AD02:00=E2=86=92=E5=91=A8=E4=B8=8009:30?= =?UTF-8?q?=E6=9C=8955h=E8=B7=91=E9=81=93;=E6=97=A7=E6=8E=92=E7=A8=8B7.5h?= =?UTF-8?q?=E5=A4=9C=E7=AA=97=E8=BF=9E=E5=8D=95=E8=A1=A88h=E9=83=BD?= =?UTF-8?q?=E8=A3=85=E4=B8=8D=E4=B8=8B,09-01=E6=B3=A8=E5=86=8C=E5=90=8E?= =?UTF-8?q?=E9=A6=96=E7=88=8616h+=E7=A9=BF=E6=95=B4=E4=B8=AA=E4=BA=A4?= =?UTF-8?q?=E6=98=93=E6=97=A5,09:32=E5=86=B3=E7=AD=96=E7=AA=97=E6=92=9Ebal?= =?UTF-8?q?ance(94%Q2'26)/income(0%)vintage=E6=92=95=E8=A3=82=E8=87=B4valu?= =?UTF-8?q?e=20ROE=E5=BD=92=E9=9B=B6=E6=B8=85=E4=BB=93;=E6=9C=88=E5=86=851?= =?UTF-8?q?-4=E5=8F=B7=E8=AF=BB=E6=97=A7=E4=BD=86=E4=B8=80=E8=87=B4,?= =?UTF-8?q?=E7=AD=96=E7=95=A5=E9=85=8D=E5=AF=B9=E4=BF=AE=E5=A4=8D=E4=BE=A7?= =?UTF-8?q?=E5=85=BC=E5=AE=B9)=E2=91=A3=E6=96=B0=E5=A2=9Estatic=5Fvintage?= =?UTF-8?q?=5Fcheck.py=E6=AF=8F=E6=97=A509:00=E8=87=AA=E6=A3=80:=E4=B8=89?= =?UTF-8?q?=E6=8A=A5=E8=A1=A8=E6=9C=80=E5=A4=A7=E6=8A=A5=E5=91=8A=E6=9C=9F?= =?UTF-8?q?+=E8=A6=86=E7=9B=96=E7=8E=87=E7=9B=98=E7=82=B9(REPORT=5FDATE?= =?UTF-8?q?=E5=88=97=E8=A3=81=E5=89=AA=E6=89=AB=E6=8F=8F),=E8=B7=A8?= =?UTF-8?q?=E8=A1=A8=E6=92=95=E8=A3=82/=E4=BC=B0=E5=80=BC=E6=96=AD?= =?UTF-8?q?=E6=9B=B4(>48h)=E5=91=8A=E8=AD=A6exit=203,=E5=8E=9F=E5=AD=90?= =?UTF-8?q?=E8=90=BDvintage=5Fstatus.json=E4=BE=9B=E7=AD=96=E7=95=A5?= =?UTF-8?q?=E7=9B=98=E5=89=8D=E9=A2=84=E5=88=A4(=E7=AD=96=E7=95=A5session?= =?UTF-8?q?=E5=8D=8F=E4=BD=9C=E5=A5=91=E7=BA=A6);register=E5=8A=A0sanguo-a?= =?UTF-8?q?k-vintage+=E6=96=B0wrapper;+8=E6=B5=8B=E8=AF=95=20[vps]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/data_platform/ak_quarter_wrapper.ps1 | 6 +- scripts/data_platform/ak_vintage_wrapper.ps1 | 13 ++ .../register_akshare_schtasks.ps1 | 11 +- scripts/data_platform/static_vintage_check.py | 181 ++++++++++++++++++ .../test_static_vintage_check.py | 121 ++++++++++++ 5 files changed, 328 insertions(+), 4 deletions(-) create mode 100644 scripts/data_platform/ak_vintage_wrapper.ps1 create mode 100644 scripts/data_platform/static_vintage_check.py create mode 100644 tests/data_platform/test_static_vintage_check.py diff --git a/scripts/data_platform/ak_quarter_wrapper.ps1 b/scripts/data_platform/ak_quarter_wrapper.ps1 index f780490..da63bf8 100644 --- a/scripts/data_platform/ak_quarter_wrapper.ps1 +++ b/scripts/data_platform/ak_quarter_wrapper.ps1 @@ -1,5 +1,7 @@ -# ak_quarter_wrapper.ps1 — sanguo-ak-quarter schtask wrapper (财报季 APR/MAY/SEP/NOV 02:00 akshare 三表+预告/快报 --force) -# balance+income+cashflow+forecast+express; per-stock 全量慢(5500×5), 财报季月夜间跑 +# ak_quarter_wrapper.ps1 — sanguo-ak-quarter schtask wrapper (财报季月首个周六 02:00 akshare 三表+预告/快报 --force) +# balance+income+cashflow+forecast+express; per-stock 全量(5500×5表)单趟 20h+, +# 只能周末跑 (2026-09-01 事故: 月内1号02:00 起 7.5h 夜窗装不下, 洗库穿交易日, +# 09:32 决策窗撞 balance/income vintage 撕裂; 首个周六起跑周一 09:30 前有 55h 跑道) $env:http_proxy = '' $env:https_proxy = '' $env:all_proxy = '' diff --git a/scripts/data_platform/ak_vintage_wrapper.ps1 b/scripts/data_platform/ak_vintage_wrapper.ps1 new file mode 100644 index 0000000..f8a2cb7 --- /dev/null +++ b/scripts/data_platform/ak_vintage_wrapper.ps1 @@ -0,0 +1,13 @@ +# ak_vintage_wrapper.ps1 — sanguo-ak-vintage schtask wrapper (每日 09:00 静态表 vintage 一致性自检) +# 三报表(balance/income/cashflow)最大报告期不一致(洗库中段撕裂)或估值表断更 → ALERT + 退出码 3; +# 落 data/static/vintage_status.json 供策略盘前预判 (2026-09-01 洗库撕裂事故防线④) +$env:http_proxy = '' +$env:https_proxy = '' +$env:all_proxy = '' +Set-Location C:\sanguo_vnpy_v2 +$ts = Get-Date -Format 'yyyyMMdd_HHmmss' +$logDir = 'C:\sanguo_vnpy_v2\data\migration_logs' +if (-not (Test-Path $logDir)) { New-Item -ItemType Directory -Path $logDir -Force | Out-Null } +$log = Join-Path $logDir "ak_vintage_$ts.txt" +C:\Python310\python.exe -X utf8 C:\sanguo_vnpy_v2\scripts\data_platform\static_vintage_check.py *>> $log +exit $LASTEXITCODE diff --git a/scripts/data_platform/register_akshare_schtasks.ps1 b/scripts/data_platform/register_akshare_schtasks.ps1 index 1cfd7de..13e7549 100644 --- a/scripts/data_platform/register_akshare_schtasks.ps1 +++ b/scripts/data_platform/register_akshare_schtasks.ps1 @@ -9,10 +9,17 @@ schtasks --% /create /tn sanguo-ak-eod /tr "powershell -ExecutionPolicy Bypass -File C:\sanguo_vnpy_v2\scripts\data_platform\ak_eod_wrapper.ps1" /sc daily /st 19:00 /ru SYSTEM /rl HIGHEST /f schtasks --% /create /tn sanguo-ak-events /tr "powershell -ExecutionPolicy Bypass -File C:\sanguo_vnpy_v2\scripts\data_platform\ak_events_wrapper.ps1" /sc daily /st 19:30 /ru SYSTEM /rl HIGHEST /f schtasks --% /create /tn sanguo-ak-stock /tr "powershell -ExecutionPolicy Bypass -File C:\sanguo_vnpy_v2\scripts\data_platform\ak_stock_wrapper.ps1" /sc weekly /d SAT /st 03:00 /ru SYSTEM /rl HIGHEST /f -schtasks --% /create /tn sanguo-ak-quarter /tr "powershell -ExecutionPolicy Bypass -File C:\sanguo_vnpy_v2\scripts\data_platform\ak_quarter_wrapper.ps1" /sc monthly /m APR,MAY,SEP,NOV /st 02:00 /ru SYSTEM /rl HIGHEST /f +# ak-quarter: 财报月首个周六 02:00 (2026-09-01 事故修正: 旧 /d 缺省=每月1号02:00, +# 夜间窗口 7.5h 连一张表(~8h)都装不下, 09-01 首爆 balance 02:00-10:01 / income +# 10:01-18:17 穿整个交易日, 09:32 决策窗撞两表 vintage 撕裂 → value 清仓。 +# 首个周六 02:00 → 周一 09:30 有 55h 跑道; 月内 1-4 号读旧但一致, 策略配对兼容) +schtasks --% /create /tn sanguo-ak-quarter /tr "powershell -ExecutionPolicy Bypass -File C:\sanguo_vnpy_v2\scripts\data_platform\ak_quarter_wrapper.ps1" /sc monthly /m APR,MAY,SEP,NOV /d SAT /st 02:00 /ru SYSTEM /rl HIGHEST /f +# ak-vintage: 每日 09:00 静态表 vintage 一致性自检 (防线④: 撕裂/断更告警退出码 3, +# 落 data/static/vintage_status.json 供策略盘前预判) +schtasks --% /create /tn sanguo-ak-vintage /tr "powershell -ExecutionPolicy Bypass -File C:\sanguo_vnpy_v2\scripts\data_platform\ak_vintage_wrapper.ps1" /sc daily /st 09:00 /ru SYSTEM /rl HIGHEST /f schtasks --% /create /tn sanguo-index /tr "powershell -ExecutionPolicy Bypass -File C:\sanguo_vnpy_v2\scripts\data_platform\index_monthly_wrapper.ps1" /sc monthly /d 16 /st 19:50 /ru SYSTEM /rl HIGHEST /f Write-Output "=== VERIFY Arguments 路径(应全路径, 非 \xxx.ps1)===" -foreach ($t in @('sanguo-ak-eod','sanguo-ak-events','sanguo-ak-stock','sanguo-ak-quarter','sanguo-index')) { +foreach ($t in @('sanguo-ak-eod','sanguo-ak-events','sanguo-ak-stock','sanguo-ak-quarter','sanguo-ak-vintage','sanguo-index')) { $xml = schtasks /query /tn $t /xml 2>&1 | Out-String if ($xml -match 'File\s+([^<"]+)') { Write-Output (" {0,-20} File {1}" -f $t, $matches[1]) diff --git a/scripts/data_platform/static_vintage_check.py b/scripts/data_platform/static_vintage_check.py new file mode 100644 index 0000000..2322bc3 --- /dev/null +++ b/scripts/data_platform/static_vintage_check.py @@ -0,0 +1,181 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""静态表 vintage 一致性自检 (2026-09-01 洗库撕裂事故防线④)。 + +背景: sanguo-ak-quarter 财报季全量洗库逐股重写 balance→income→cashflow, +单表 ~8h、全程 20h+: 洗库中段两表 vintage 撕裂 (09-01 09:32 balance 已带 +中报而 income 停在 Q1'26 → value ROE 配对归零清仓)。本脚本盘前盘点各报表 +最新报告期分布, 跨表不一致/估值断更大声告警, 并落 vintage_status.json 供 +策略侧盘前预判 (契约: tables. = {max_period, coverage, n_at_max, +n_valid, n_empty, n_unreadable, newest_mtime}, checked_at 为扫描时刻)。 + +只读, 不依赖 akshare。用法 (VPS 09:00 schtask sanguo-ak-vintage): + python static_vintage_check.py # 默认扫 C:\\sanguo_vnpy_v2\\data\\static + python static_vintage_check.py --root --json-out + +退出码: 0=一致; 3=有告警 (schtask Last Result 可见) +""" +import argparse +import datetime +import json +import os +import sys +from pathlib import Path +from typing import Tuple + +import pandas as pd + +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 同口径 + + +def _norm_period(v) -> str: + """REPORT_DATE 值 → 'YYYY-MM-DD' (兼容 str / datetime 两种 dtype)。""" + return str(v)[:10] + + +def scan_statement_table(table_dir: Path) -> dict: + """扫一张报表: 每股最大 REPORT_DATE → 表级分布。 + + 只读 REPORT_DATE 一列 (parquet 列裁剪, 5550 股 ~1min 级)。 + 空文件 (<1KB)/0 行/读失败分别计数, 不中断扫描。 + """ + stats = { + "n_total": 0, "n_empty": 0, "n_unreadable": 0, + "n_valid": 0, "n_at_max": 0, "max_period": None, + "newest_mtime": None, "coverage": 0.0, + } + if not table_dir.is_dir(): + stats["error"] = "目录不存在" + return stats + newest = 0.0 + per_max = [] + for f in sorted(table_dir.glob("*.parquet")): + stats["n_total"] += 1 + try: + st = f.stat() + newest = max(newest, st.st_mtime) + if st.st_size < EMPTY_PARQUET_MIN_BYTES: + stats["n_empty"] += 1 + continue + df = pd.read_parquet(f, columns=["REPORT_DATE"]) + if df.empty: + stats["n_empty"] += 1 + continue + per_max.append(_norm_period(df["REPORT_DATE"].max())) + stats["n_valid"] += 1 + except Exception: + stats["n_unreadable"] += 1 + if per_max: + mx = max(per_max) + stats["max_period"] = mx + stats["n_at_max"] = sum(1 for v in per_max if v == mx) + stats["coverage"] = round(stats["n_at_max"] / stats["n_valid"], 4) + if newest: + stats["newest_mtime"] = datetime.datetime.fromtimestamp( + newest).strftime("%Y-%m-%d %H:%M:%S") + return stats + + +def valuation_freshness(root: Path) -> dict: + """估值表 (每日 ak-eod 19:00) 新鲜度: 最新 parquet mtime 距今几小时。""" + info = {"newest_mtime": None, "age_hours": None} + d = root / "valuation" + if not d.is_dir(): + info["error"] = "目录不存在" + return info + newest = 0.0 + for f in d.glob("*.parquet"): + try: + newest = max(newest, f.stat().st_mtime) + except OSError: + continue + if newest: + ts = datetime.datetime.fromtimestamp(newest) + info["newest_mtime"] = ts.strftime("%Y-%m-%d %H:%M:%S") + info["age_hours"] = round( + (datetime.datetime.now() - ts).total_seconds() / 3600, 1) + return info + + +def check_all(root: Path) -> Tuple[dict, list]: + """盘点三报表 + 估值新鲜度。返 (status, alerts); alerts 非空 → 退出码 3。""" + status = { + "checked_at": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + "root": str(root), + "tables": {}, + "valuation": {}, + } + alerts = [] + periods = {} + for t in STATEMENT_TABLES: + s = scan_statement_table(root / t) + status["tables"][t] = s + if s["max_period"]: + periods[t] = s["max_period"] + else: + alerts.append( + "[%s] 无有效数据 (n_valid=0): %s" % ( + t, s.get("error", "全空/不可读"))) + if len(periods) == len(STATEMENT_TABLES) and len(set(periods.values())) > 1: + detail = ", ".join(f"{t}={p}" for t, p in sorted(periods.items())) + alerts.append( + "报表 vintage 撕裂: %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: + alerts.append( + "估值表断更: 最新 mtime %s 距今 %.1fh (>%.0fh, ak-eod 未跑成?)" + % (v["newest_mtime"], v["age_hours"], VALUATION_STALE_HOURS)) + return status, alerts + + +def write_json_atomic(path: Path, payload: dict) -> None: + """原子写 JSON (tmp → os.replace, 洗库同款语义)。""" + tmp = path.with_suffix(path.suffix + ".tmp") + tmp.parent.mkdir(parents=True, exist_ok=True) + tmp.write_text( + json.dumps(payload, ensure_ascii=False, indent=1), encoding="utf-8") + os.replace(tmp, path) + + +def main() -> int: + ap = argparse.ArgumentParser( + description="静态表 vintage 一致性自检 (三报表+估值)") + ap.add_argument( + "--root", default=os.environ.get("STATIC_VINTAGE_ROOT", DEFAULT_ROOT), + help="static 根目录 (默认 %s)" % DEFAULT_ROOT) + ap.add_argument( + "--json-out", + default=os.environ.get( + "STATIC_VINTAGE_JSON", + os.path.join(DEFAULT_ROOT, "vintage_status.json")), + help="vintage_status.json 输出路径") + args = ap.parse_args() + + status, alerts = check_all(Path(args.root)) + write_json_atomic(Path(args.json_out), status) + + for t, s in status["tables"].items(): + print("[vintage] %-11s max=%s coverage=%.1f%% (%d/%d) empty=%d " + "unreadable=%d newest=%s" % ( + t, s["max_period"], 100 * s["coverage"], + s["n_at_max"], s["n_valid"], s["n_empty"], + s["n_unreadable"], s["newest_mtime"])) + v = status["valuation"] + print("[vintage] valuation newest=%s age=%sh" % ( + v.get("newest_mtime"), v.get("age_hours"))) + print("[vintage] json → %s" % args.json_out) + if alerts: + for a in alerts: + print("[vintage][ALERT] %s" % a) + return 3 + print("[vintage] OK: 三报表 vintage 一致") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/data_platform/test_static_vintage_check.py b/tests/data_platform/test_static_vintage_check.py new file mode 100644 index 0000000..b49e3ca --- /dev/null +++ b/tests/data_platform/test_static_vintage_check.py @@ -0,0 +1,121 @@ +# -*- coding: utf-8 -*- +"""静态表 vintage 一致性自检测试 (2026-09-01 洗库撕裂事故防线④)。 + +核心契约: 跨表最大报告期不一致 → 告警 (退出码 3) + vintage_status.json +落盘 (策略侧盘前预判数据源); 一致 → OK (退出码 0); 估值断更 → 告警。 +""" +import json +import os +import time + +import pandas as pd + +import static_vintage_check as svc # noqa: E402 (conftest 已加 sys.path) + + +def _mk_pq(path, periods): + df = pd.DataFrame({"REPORT_DATE": list(periods), "v": range(len(periods))}) + df.to_parquet(path, index=False) + + +def _mk_table(root, name, stocks_periods): + """stocks_periods: {filename_stem: [报告期,...]}""" + d = root / name + d.mkdir(parents=True, exist_ok=True) + for stem, periods in stocks_periods.items(): + _mk_pq(d / f"{stem}.parquet", periods) + return d + + +# ======================== 单表扫描 ======================== + +class TestScanStatementTable: + + def test_basic_distribution(self, tmp_path): + d = _mk_table(tmp_path, "balance", { + "a": ["2025-12-31", "2026-06-30"], + "b": ["2026-06-30"], + "c": ["2026-03-31"], # 滞后股 (未披露中报) + }) + (d / "empty.parquet").write_bytes(b"x" * 10) # <1KB 空文件 + s = svc.scan_statement_table(d) + assert s["max_period"] == "2026-06-30" + assert s["n_valid"] == 3 and s["n_at_max"] == 2 + assert s["n_empty"] == 1 and s["n_unreadable"] == 0 + assert s["coverage"] == round(2 / 3, 4) # coverage 保留 4 位小数 + + def test_datetime_dtype_period_normalized(self, tmp_path): + """真实数据 REPORT_DATE 形如 '2026-06-30 00:00:00' → 归一 YYYY-MM-DD。""" + _mk_table(tmp_path, "income", { + "a": ["2026-06-30 00:00:00"], + }) + s = svc.scan_statement_table(tmp_path / "income") + assert s["max_period"] == "2026-06-30" + + def test_missing_dir(self, tmp_path): + s = svc.scan_statement_table(tmp_path / "nope") + assert "error" in s and s["max_period"] is None + + +# ======================== 跨表一致性 / 告警 ======================== + +class TestCheckAll: + + def test_vintage_tear_alert(self, tmp_path): + """09-01 事故签名: balance 已到 Q2'26, income 停在 Q1'26 → 撕裂告警。""" + _mk_table(tmp_path, "balance", {"a": ["2026-06-30"]}) + _mk_table(tmp_path, "income", {"a": ["2026-03-31"]}) + _mk_table(tmp_path, "cashflow", {"a": ["2026-06-30"]}) + _, alerts = svc.check_all(tmp_path) + assert any("撕裂" in a and "income=2026-03-31" in a for a in alerts) + + def test_consistent_ok(self, tmp_path): + _mk_table(tmp_path, "balance", {"a": ["2026-06-30"]}) + _mk_table(tmp_path, "income", {"a": ["2026-06-30"]}) + _mk_table(tmp_path, "cashflow", {"a": ["2026-06-30"]}) + _, alerts = svc.check_all(tmp_path) + assert alerts == [] + + def test_valuation_stale_alert(self, tmp_path): + """估值表 (每日 ak-eod) 超 48h 未刷新 → 断更告警。""" + _mk_table(tmp_path, "balance", {"a": ["2026-06-30"]}) + _mk_table(tmp_path, "income", {"a": ["2026-06-30"]}) + _mk_table(tmp_path, "cashflow", {"a": ["2026-06-30"]}) + d = _mk_table(tmp_path, "valuation", {"v": ["2026-08-31"]}) + old = time.time() - 100 * 3600 + os.utime(d / "v.parquet", (old, old)) + _, alerts = svc.check_all(tmp_path) + assert any("估值表断更" in a for a in alerts) + + +# ======================== main / JSON 契约 ======================== + +class TestMain: + + def test_ok_exit0_and_json_written(self, tmp_path, monkeypatch, capsys): + _mk_table(tmp_path, "balance", {"a": ["2026-06-30"]}) + _mk_table(tmp_path, "income", {"a": ["2026-06-30"]}) + _mk_table(tmp_path, "cashflow", {"a": ["2026-06-30"]}) + jp = tmp_path / "vintage_status.json" + monkeypatch.setattr( + "sys.argv", ["x", "--root", str(tmp_path), "--json-out", str(jp)]) + assert svc.main() == 0 + payload = json.loads(jp.read_text(encoding="utf-8")) + assert set(payload["tables"]) == {"balance", "income", "cashflow"} + assert payload["tables"]["balance"]["max_period"] == "2026-06-30" + assert "OK" in capsys.readouterr().out + # 幂等重跑 (原子覆盖) + monkeypatch.setattr( + "sys.argv", ["x", "--root", str(tmp_path), "--json-out", str(jp)]) + assert svc.main() == 0 + + def test_tear_exit3_with_alert_line(self, tmp_path, monkeypatch, capsys): + _mk_table(tmp_path, "balance", {"a": ["2026-06-30"]}) + _mk_table(tmp_path, "income", {"a": ["2026-03-31"]}) + _mk_table(tmp_path, "cashflow", {"a": ["2026-03-31"]}) + jp = tmp_path / "v.json" + monkeypatch.setattr( + "sys.argv", ["x", "--root", str(tmp_path), "--json-out", str(jp)]) + assert svc.main() == 3 + assert "[vintage][ALERT]" in capsys.readouterr().out + assert jp.exists() # 告警态也落 json (策略侧能看到撕裂详情)