From e15acddb2f24e9bf388be7c2e6003d967080c82b Mon Sep 17 00:00:00 2001 From: shy3130 Date: Sun, 23 Aug 2026 19:12:57 +0800 Subject: [PATCH] =?UTF-8?q?fix(pipeline):=20=E5=81=9C=E6=9C=BA=E7=BC=BA?= =?UTF-8?q?=E5=8F=A3=E6=A3=80=E6=B5=8B=E4=B8=8E=E8=87=AA=E5=8A=A8=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D,=20=E9=98=B2=E6=AD=A2=E7=9B=98=E4=B8=AD=E5=BF=AB?= =?UTF-8?q?=E7=85=A7=E6=95=B0=E6=8D=AE=E6=B0=B8=E4=B9=85=E6=8D=9F=E5=9D=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 场景: 用户盘中停机后次日启动并开实时行情, 实时 flush 写出"今天"分区后, 盘后管道的「今天已有数据→只刷今天」分支会让停机日的半日快照永久留存 (close=停机时刻价, volume=半日累计, 技术指标全错且污染后续 lookback)。 判据 (quote_ts 列, 仅实时 flush 写入真实毫秒时间戳): - null → batch 拉取/盘后计算的权威历史 → 完整 - 历史交易日时刻 < 15:00 → 盘中快照 → 坏 - ≥ 15:00 → 尾盘定版 → 完整; 今天不校验 读取走 parquet 元数据 statistics, 实测 ~0.5ms/分区。 三个钩子层层设防: - 启动自检 (main.py): boot 后 30s 后台扫描, 窗口内坏数据自动创建修复任务 (复用 repair_daily 管道 + JobStore, ActiveJobCard 可见进度) - 开实时门禁 (settings.py): 坏数据在 5 天窗口内 → 自动建修复任务并 409, 前端经现有 toast 机制展示; 窗口外放行不拦 - 盘后管道自愈 (daily_pipeline.py): 分支3前完整性扫描, 坏日 → 降级分支4 从 min(最新日, 坏日) 范围重拉; ETF/指数起点同步提前到各自族最早坏日 修复原语补强: 股票 enriched 的坏分区由 prune_enriched_partitions 先删再 当"新日期"重算 (增量重算只算不存在的日期, 分区已存在虽错也永不重算)。 分钟K缺口经修复管道 Step 2.5 增量同步天然补洞 (start=max(datetime) 幂等)。 测试: 新增 24 个 (判据/门禁/boot/端到端离线集成复刻用户场景), 全量 1002 passed。 --- backend/app/api/settings.py | 25 ++ backend/app/jobs/daily_pipeline.py | 67 +++- backend/app/main.py | 13 + backend/app/services/data_integrity.py | 345 ++++++++++++++++++++ backend/app/services/repair_daily.py | 4 +- backend/tests/test_data_integrity.py | 432 +++++++++++++++++++++++++ 6 files changed, 881 insertions(+), 5 deletions(-) create mode 100644 backend/app/services/data_integrity.py create mode 100644 backend/tests/test_data_integrity.py diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py index a5747ed..779e771 100644 --- a/backend/app/api/settings.py +++ b/backend/app/api/settings.py @@ -810,6 +810,31 @@ def update_realtime_quotes(req: RealtimeQuotesPrefs, request: Request) -> dict: if req.realtime_quotes_enabled and qs and qs.is_paused(): # 管道/数据修正运行期间禁止开启实时行情 — 防止写盘竞态 raise HTTPException(status_code=409, detail="数据同步运行中，实时行情已临时暂停，请稍后再开启") + if req.realtime_quotes_enabled: + # 历史完整性门禁: 检测到最近交易日的盘中快照/缺口时禁止开启 — + # 实时 flush 写出"今天"分区后, 盘后管道的"只刷今天"分支会让停机日的 + # 半日快照永久留存。同时自动创建修复任务, 修完即可正常开启。 + from app.services import data_integrity + + repo = getattr(request.app.state, "repo", None) + if repo is not None: + try: + issues = data_integrity.scan_recent_integrity(repo.store.data_dir) + except Exception: # noqa: BLE001 + issues = [] + earliest = data_integrity.earliest_issue_day(issues) + if issues and data_integrity.within_auto_repair_window(earliest): + job_id, is_new = data_integrity.launch_integrity_repair( + request.app.state, earliest, "realtime_gate", + ) + if job_id is not None: + detail = ( + f"检测到{data_integrity.describe_issues(issues)}，" + + ("已自动创建修复任务，完成后即可开启实时行情" + if is_new else "修复任务正在进行中，请稍后再开启") + + f"（任务 {job_id}）" + ) + raise HTTPException(status_code=409, detail=detail) if req.realtime_quotes_enabled and qs and qs.realtime_mode() == "watchlist" and not preferences.get_realtime_watchlist_symbols(): preferences.save({"realtime_quotes_enabled": False}) _sync_depth_polling(False) diff --git a/backend/app/jobs/daily_pipeline.py b/backend/app/jobs/daily_pipeline.py index 41954e8..840f860 100644 --- a/backend/app/jobs/daily_pipeline.py +++ b/backend/app/jobs/daily_pipeline.py @@ -147,6 +147,31 @@ def run_now( today = _date.today() today_exists = latest_daily and latest_daily >= today new_daily_days = 0 + + # 完整性自愈: 检测最近交易日的盘中快照/缺口 (盘中停机后次日开实时会留下 + # 中午快照, 而下方"今天已有数据→只刷今天"分支会让它永久留存)。 + # 命中 → 本次管道放弃实时覆写分支, 降级为从最早坏日起的范围拉取。 + integrity_issues: list = [] + stale_day: _date | None = None + etf_stale_day: _date | None = None + index_stale_day: _date | None = None + if override_start_date is None: + try: + from app.services import data_integrity + integrity_issues = data_integrity.scan_recent_integrity( + repo.store.data_dir, today=today, + ) + if integrity_issues: + stale_day = data_integrity.earliest_issue_day(integrity_issues, ("kline_daily",)) + etf_stale_day = data_integrity.earliest_issue_day(integrity_issues, ("kline_etf_daily",)) + index_stale_day = data_integrity.earliest_issue_day(integrity_issues, ("kline_index_daily",)) + logger.warning( + "integrity: 检测到 %d 个不完整分区(%s), 本次管道改走范围拉取修复", + len(integrity_issues), data_integrity.describe_issues(integrity_issues), + ) + except Exception as e: # noqa: BLE001 + logger.warning("integrity scan failed (soft, 按无坏数据处理): %s", e) + integrity_issues = [] # 日K范围拉取的起点(分支3补缺口/分支4首次/数据修正); 实时增量/跳过时为 None。 # 供 Step 1.5 除权因子回溯范围对齐: 范围拉取→用日K范围, 非范围→最近N天兜底。 daily_range_start: _date | None = None @@ -177,8 +202,15 @@ def run_now( new_daily_days = gap_days emit("sync_daily", 45, f"日K 完成,覆盖 {gap_days} 天") logger.info("sync_daily: [%s ~ %s] done, %d days", start_date, today, gap_days) - elif today_exists and capset.has(Cap.QUOTE_POOL) and _prefs.get_daily_data_provider() == "tickflow": + elif ( + today_exists + and stale_day is None + and capset.has(Cap.QUOTE_POOL) + and _prefs.get_daily_data_provider() == "tickflow" + ): # 付费档:今天有数据(QuoteService 已落盘)→ 实时行情覆写,确保最新。 + # stale_day 非空时禁用本分支: "只刷今天"会让停机日的盘中快照永久留存, + # 降级到下方 batch 路径从坏日起重拉。 # free/none 档无 quote.pool 能力,即便今天已有数据(如从 expert 降级), # 也降级到下方 batch 路径刷新,避免调用无权限的实时行情接口。 emit("sync_daily", 12, f"获取日K [{today} ~ {today}] 实时行情…") @@ -186,11 +218,13 @@ def run_now( new_daily_days = 1 emit("sync_daily", 45, f"日K 完成,{written_daily} 只标的") logger.info("sync_daily: [%s ~ %s] live quotes, %d symbols", today, today, written_daily) - elif latest_daily: + elif latest_daily or stale_day: # 有历史 → batch 补齐缺口。 # 也覆盖"今天已有数据但无实时行情权限(free/none)"的降级场景: # 此时 start_date = latest_daily = today,batch 刷新当天日K。 - start_date = latest_daily + # 完整性修复场景: start_date = min(本地最新日, 最早坏日) — + # today_exists 时 latest_daily=今天, 不取 min 会漏掉坏日。 + start_date = min(d for d in (latest_daily, stale_day) if d is not None) daily_range_start = start_date emit("sync_daily", 12, f"获取日K [{start_date} ~ {today}]…") logger.info("sync_daily: [%s ~ %s] %s", start_date, today, @@ -230,6 +264,22 @@ def run_now( logger.info("sync_daily: [%s ~ %s] done", start_date, today) _invalidate("daily") + # 完整性修复时删除股票 enriched 的坏分区: 增量重算只算 enriched 里不存在 + # 的日期, 盘中快照日分区已存在(虽是错的), 不删永远不会被重算。删除后 + # Step 2 把这些日期当"新日期"重算 (剩余分区最近 60 天做历史前缀, 窗口 ≤5 天回看充足)。 + repair_start = override_start_date if override_start_date is not None else stale_day + if repair_start is not None: + try: + from app.services.data_integrity import prune_enriched_partitions + pruned = prune_enriched_partitions( + repo.store.data_dir, repair_start, "kline_daily_enriched", + ) + if pruned: + logger.info("integrity: 已删除 %d 个待重算的 enriched 分区 (≥ %s)", pruned, repair_start) + except Exception as e: # noqa: BLE001 + logger.warning("enriched prune failed (soft): %s", e) + + # 单标的新鲜度: 全局 max(date) 会被任一有今日数据的标的"拉高", 掩盖停牌/复牌/ # 一直拉失败而掉队的个股缺口(全局判据只刷"今天", 永不回补掉队标的的历史缺口)。 # 这里检测并**可见化**(WARNING + 计入结果), 让掉队标的不再隐形。 @@ -391,11 +441,15 @@ def run_now( d.name[5:] for d in index_dir.glob("date=*") if d.is_dir() and d.name.startswith("date=") ) if index_dir.exists() else [] - # 数据修正模式下用传入起点; 否则用本地指数最新日期补到今天 + # 数据修正模式下用传入起点; 否则用本地指数最新日期补到今天; + # 完整性修复时起点再提前到最早坏日 (实时写过今天的指数分区时, + # "最新日期=今天"会让停机日的快照/缺口永久留存) if override_start_date: index_start = override_start_date else: index_start = _date.fromisoformat(index_dates[-1]) if index_dates else today - _td(days=365) + if index_stale_day is not None and index_start > index_stale_day: + index_start = index_stale_day def _index_chunk(cur: int, tot: int) -> None: emit("sync_index", 88, f"指数日K批次 {cur}/{tot}", @@ -456,6 +510,9 @@ def run_now( if d.is_dir() and d.name.startswith("date=") ) if etf_dir.exists() else [] etf_start = _date.fromisoformat(etf_dates[-1]) if etf_dates else today - _td(days=365) + # 同指数: 完整性修复时把 ETF 起点提前到最早坏日 + if etf_stale_day is not None and etf_start > etf_stale_day: + etf_start = etf_stale_day def _etf_chunk(cur: int, tot: int) -> None: emit("sync_index", 88, f"ETF 日K批次 {cur}/{tot}", @@ -592,6 +649,8 @@ def run_now( "regime_days": regime_days, "mainline_rows": mainline_rows, "lagging_symbols": len(lagging_symbols), + "integrity_repair_from": repair_start.isoformat() if repair_start else None, + "integrity_issues": len(integrity_issues), "skipped_stages": skipped, "stage_errors": stage_errors, } diff --git a/backend/app/main.py b/backend/app/main.py index 6fe86b0..a65e130 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -173,6 +173,19 @@ async def _application_lifespan(app: FastAPI): except Exception as e: # noqa: BLE001 logger.warning("depth_service init failed: %s", e) + # 停机缺口自检: 延迟后台扫描, 发现最近交易日的盘中快照/缺口时自动创建 + # 修复任务 (盘中停机→次日开实时场景, 不修则坏数据被"只刷今天"分支永久留存) + try: + import threading + + from app.services.data_integrity import boot_integrity_check + + timer = threading.Timer(30.0, boot_integrity_check, args=(app.state,)) + timer.daemon = True # 不阻塞进程退出 + timer.start() + except Exception as e: # noqa: BLE001 + logger.warning("integrity boot check scheduling failed: %s", e) + # 企业微信智能机器人长连接(可选通道, 失败不阻断启动) try: from app.services.wecom_bot_service import WecomBotService diff --git a/backend/app/services/data_integrity.py b/backend/app/services/data_integrity.py new file mode 100644 index 0000000..8361b3e --- /dev/null +++ b/backend/app/services/data_integrity.py @@ -0,0 +1,345 @@ +"""历史交易日数据完整性检测 — 停机缺口 / 盘中快照判别。 + +场景: 用户盘中停机后, 次日再启动并开实时行情, 实时 flush 写出"今天"分区后, +盘后管道的「今天已有数据 → 只刷今天」分支会让停机日的盘中快照永久留存 +(close=停机时刻价, volume=半日累计, 技术指标全错且污染后续 lookback 类指标)。 + +判据 (quote_ts 列, 毫秒 Unix 时间戳, 仅实时 flush 写入真实值): +- null → batch 拉取 / 盘后计算写入的权威历史 → 完整 +- d < 今天 且 时刻 < d 15:00 → 盘中快照 (停机前实时写的) → 坏 +- d < 今天 且 时刻 ≥ d 15:00 → 尾盘定版 (close_final) → 完整 +- d == 今天 → 实时更新中, 属正常, 不校验 +- 分区缺失的工作日 → 缺口 (工作日近似; 节假日误报的代价是一次空范围拉取, + merge-upsert 空写, 无害) + +检测成本: 每分区只读 parquet 元数据 statistics (不解压数据页), 实测 ~0.5ms/分区。 +""" +from __future__ import annotations + +import contextlib +import logging +from collections.abc import Iterable +from dataclasses import dataclass +from datetime import date, datetime, timedelta +from datetime import time as dt_time +from pathlib import Path + +import polars as pl + +from app.market_time import CN_TZ + +logger = logging.getLogger(__name__) + +# 尾盘定版线: quote_ts 达到当日 15:00 即视为收盘后写入 (含 close_final 定版) +CLOSE_CUTOFF = dt_time(15, 0) + +# 扫描窗口: 最近 N 个自然日内、今天之前的交易日 +SCAN_WINDOW_DAYS = 7 + +# 自动修复窗口: 最早坏日距今超过 N 个自然日 → 只报告不自动修 (更大缺口由用户手动 repair) +AUTO_REPAIR_MAX_LAG_DAYS = 5 + +# 参与检测的日K族表 (实时 flush 会写这三族的 daily/enriched) +_DAILY_TABLES = ("kline_daily", "kline_etf_daily", "kline_index_daily") + +# 表 → 资产族 (用于管道/修复侧按族取起点) +TABLE_FAMILY = { + "kline_daily": "stock", + "kline_etf_daily": "etf", + "kline_index_daily": "index", +} + + +@dataclass(frozen=True) +class IntegrityIssue: + day: date + table: str + kind: str # "snapshot"=盘中快照 | "missing"=分区缺失 + + +def _quote_ts_max_ms(part_dir: Path) -> int | None: + """读单个日期分区的 max(quote_ts); 列不存在/全 null/无统计 → None。 + + 优先走 parquet 元数据 row-group statistics (不解压数据页), + statistics 缺失时回退 polars 列扫描。 + """ + import pyarrow.parquet as pq + + candidates: list[int | None] = [] + files = sorted(part_dir.glob("*.parquet")) + if not files: + return None + for path in files: + try: + meta = pq.read_metadata(path) + names = [meta.schema.column(i).name for i in range(meta.num_columns)] + if "quote_ts" not in names: + continue + idx = names.index("quote_ts") + file_max: int | None = None + for rg in range(meta.num_row_groups): + stats = meta.row_group(rg).column(idx).statistics + if stats is not None and stats.max is not None: + value = stats.max + file_max = int(value) if file_max is None else max(file_max, int(value)) + if file_max is None and meta.num_rows > 0: + # statistics 未写入 → 回退列扫描 + file_max = ( + pl.scan_parquet(path) + .select(pl.col("quote_ts").max()) + .collect() + .item() + ) + candidates.append(file_max) + except Exception as e: # noqa: BLE001 + logger.debug("quote_ts scan skipped %s: %s", path, e) + values = [v for v in candidates if v is not None] + return max(values) if values else None + + +def _is_snapshot(day: date, quote_ts_ms: int | None) -> bool: + """非空 quote_ts 且对应北京时间时刻早于当日收盘线 → 盘中快照。""" + if quote_ts_ms is None: + return False + try: + ts = datetime.fromtimestamp(int(quote_ts_ms) / 1000, tz=CN_TZ) + except (OverflowError, OSError, ValueError): + return False + return ts.date() == day and ts.time() < CLOSE_CUTOFF + + +def _candidate_days(today: date, lookback_days: int) -> list[date]: + """最近 lookback_days 自然日内、严格早于今天的工作日 (节假日近似, 误报无害)。""" + days: list[date] = [] + for offset in range(1, lookback_days + 1): + d = today - timedelta(days=offset) + if d.weekday() < 5: + days.append(d) + return sorted(days) + + +def scan_recent_integrity( + data_dir: Path, + *, + today: date | None = None, + lookback_days: int = SCAN_WINDOW_DAYS, +) -> list[IntegrityIssue]: + """扫描最近交易日的数据完整性, 返回坏分区列表 (按日期升序)。 + + 每族表独立判定; 族内"最近无任何活动"(最新分区早于窗口)时整族跳过 — + 覆盖首次启动(无数据)与长期停用(用户自主)两类不应自动修复的场景。 + """ + data_dir = Path(data_dir) + today = today or datetime.now(CN_TZ).date() + window_start = today - timedelta(days=lookback_days) + issues: list[IntegrityIssue] = [] + + for table in _DAILY_TABLES: + base = data_dir / table + existing: set[date] = set() + if base.exists(): + for part in base.glob("date=*"): + try: + existing.add(date.fromisoformat(part.name[5:])) + except ValueError: + continue + latest = max(existing) if existing else None + # 族内近期无活动 → 不判定 (首次启动 / 长期停用) + if latest is None or latest < window_start: + continue + + for day in _candidate_days(today, lookback_days): + if day not in existing: + # 只报"尾部缺口": 晚于本地最新分区的缺失日。 + # 历史内部空洞是另一类问题(laggards), 已有独立告警, 不在此扩面。 + if day > latest: + issues.append(IntegrityIssue(day=day, table=table, kind="missing")) + continue + part_dir = base / f"date={day.isoformat()}" + quote_ts = _quote_ts_max_ms(part_dir) + if _is_snapshot(day, quote_ts): + issues.append(IntegrityIssue(day=day, table=table, kind="snapshot")) + + issues.sort(key=lambda i: (i.day, i.table)) + return issues + + +def earliest_issue_day( + issues: list[IntegrityIssue], + tables: Iterable[str] | None = None, +) -> date | None: + """坏分区中最早的一天; tables 限定参与的表族 (None=全部)。""" + scoped = ( + [i for i in issues if i.table in tables] if tables is not None else issues + ) + return min((i.day for i in scoped), default=None) + + +def within_auto_repair_window(day: date | None, *, today: date | None = None) -> bool: + """最早坏日是否落在自动修复窗口内 (≤ AUTO_REPAIR_MAX_LAG_DAYS 自然日)。""" + if day is None: + return False + today = today or datetime.now(CN_TZ).date() + return (today - day).days <= AUTO_REPAIR_MAX_LAG_DAYS + + +def prune_enriched_partitions( + data_dir: Path, + start: date, + table: str = "kline_daily_enriched", +) -> int: + """删除 enriched 表 date ≥ start 的日期分区, 使修复重算把它们当"新日期"。 + + 股票 enriched 增量重算只算 enriched 里不存在的日期; 盘中快照日分区已存在 + (虽是错的), 不删则永远不会被重算。指数/ETF 的 enriched 走 merge-upsert + 全行覆盖, 无需删除。删除后 run_pipeline(new_dates_only=True) 用剩余分区 + 最近 60 天做历史前缀重算 (修复窗口 ≤5 天, 回看充足)。 + """ + base = Path(data_dir) / table + if not base.exists(): + return 0 + import shutil + + removed = 0 + for part in base.glob("date=*"): + try: + d = date.fromisoformat(part.name[5:]) + except ValueError: + continue + if d >= start: + shutil.rmtree(part, ignore_errors=True) + removed += 1 + return removed + + +def describe_issues(issues: list[IntegrityIssue]) -> str: + """面向用户的一句话描述 (409 详情 / 日志用)。""" + if not issues: + return "" + days = sorted({i.day for i in issues}) + day_text = "、".join(d.isoformat() for d in days) + kinds = {i.kind for i in issues} + reason = "停机前的盘中快照" if "snapshot" in kinds else "缺失" + return f"{day_text} 的数据为{reason}" + + +def launch_integrity_repair(app_state, start_date: date, reason: str) -> tuple[str | None, bool]: + """自动创建数据修复任务 (复用 repair_daily 管道 + JobStore 任务体系)。 + + 返回 (job_id, is_new): + - (None, False) : 无法修复 (无 batch 能力 / 无 repo) + - (id, False) : 已有 pending/running 任务复用 (singleflight) + - (id, True) : 新建并启动 + 任务体与 /api/kline/repair_daily 完全一致: run slot + 实时 paused 互斥 + + run_repair_daily(override_start_date)。 + + 调度自适应: 调用方在事件循环内 (API 端点) → executor 后台执行; + 无事件循环 (boot Timer 线程) → 独立 daemon 线程执行。 + """ + import asyncio + import threading + + repo = getattr(app_state, "repo", None) + capset = getattr(app_state, "capabilities", None) + if repo is None or capset is None: + return None, False + try: + from app.tickflow.capabilities import Cap + + if not capset.has(Cap.KLINE_DAILY_BATCH): + logger.info("integrity repair skipped: no KLINE_DAILY_BATCH capability") + return None, False + except Exception: # noqa: BLE001 + return None, False + + from app.services.pipeline_jobs import ( + JobCancelledError, + job_store, + release_run_slot, + try_acquire_run_slot, + ) + from app.services.repair_daily import run_repair_daily + + job_id, is_new = job_store.create() + if not is_new: + return job_id, False + + qs = getattr(app_state, "quote_service", None) + + def progress(stage: str, pct: int, msg: str, + stage_pct: int | None = None, skip_log: bool = False) -> None: + job_store.progress(job_id, stage, pct, msg, stage_pct=stage_pct, skip_log=skip_log) + + def _run() -> dict: + # 修复期间暂停实时取数, 防止覆写同一批 parquet 竞态 + if qs: + with qs.paused(): + return run_repair_daily(repo, capset, start_date, on_progress=progress) + return run_repair_daily(repo, capset, start_date, on_progress=progress) + + def _execute() -> None: + try: + if not try_acquire_run_slot(job_id): + job_store.fail(job_id, "已有数据任务在运行(或上一次任务卡死未结束),请稍后再试") + return + job_store.start(job_id) + result = _run() + if isinstance(result, dict) and "error" in result: + job_store.fail(job_id, str(result["error"])) + else: + job_store.succeed(job_id, result) + except JobCancelledError: + pass # 已由 terminate() 标记失败 + except Exception as e: # noqa: BLE001 + logger.exception("integrity repair failed: job_id=%s", job_id) + job_store.fail(job_id, str(e)) + finally: + release_run_slot(job_id) + with contextlib.suppress(Exception): + from app.api.data import invalidate_storage_cache + + invalidate_storage_cache() + + try: + loop = asyncio.get_running_loop() + + async def task() -> None: + await loop.run_in_executor(None, _execute) + + asyncio.create_task(task()) + except RuntimeError: + threading.Thread( + target=_execute, daemon=True, name=f"integrity-repair-{job_id[:8]}" + ).start() + + logger.warning("integrity: 自动修复任务启动 job=%s start=%s reason=%s", job_id, start_date, reason) + return job_id, True + + +def boot_integrity_check(app_state) -> None: + """启动自检 (后台线程调用): 发现窗口内的坏数据自动创建修复任务。 + + 分钟K缺口无需单独处理 — 修复管道 Step 2.5 在 minute_sync_enabled 时 + 以 start=max(datetime) 增量补洞, 天然覆盖停机缺口。 + """ + repo = getattr(app_state, "repo", None) + if repo is None: + return + try: + issues = scan_recent_integrity(repo.store.data_dir) + except Exception as e: # noqa: BLE001 + logger.warning("boot integrity scan failed: %s", e) + return + if not issues: + logger.info("boot integrity check: 近 %d 个交易日数据完整", SCAN_WINDOW_DAYS) + return + earliest = earliest_issue_day(issues) + logger.warning("boot integrity check: %s (共 %d 个坏分区)", describe_issues(issues), len(issues)) + if not within_auto_repair_window(earliest): + logger.warning( + "integrity: 最早坏日 %s 超出自动修复窗口(%d 天), 请在数据页手动执行数据修正", + earliest, AUTO_REPAIR_MAX_LAG_DAYS, + ) + return + with contextlib.suppress(Exception): + launch_integrity_repair(app_state, earliest, "boot_integrity_check") diff --git a/backend/app/services/repair_daily.py b/backend/app/services/repair_daily.py index 5d70f02..9ea4fcb 100644 --- a/backend/app/services/repair_daily.py +++ b/backend/app/services/repair_daily.py @@ -8,7 +8,9 @@ 这样修正功能与盘后管道永远保持一致,不会出现遗漏。 落盘是 merge-upsert (按 symbol+date 去重 keep="last"), 新数据天然覆盖旧值, -不会产生重复,也不需要先删分区。 +不会产生重复。日K/指数/ETF 族无需先删分区; 股票 enriched 例外 — 增量重算 +只算 enriched 里不存在的日期, "分区已存在但内容是盘中快照"的场景由管道内 +的 prune_enriched_partitions 先删坏分区再重算 (run_now 内处理, 此处无需关心)。 """ from __future__ import annotations diff --git a/backend/tests/test_data_integrity.py b/backend/tests/test_data_integrity.py new file mode 100644 index 0000000..74821ca --- /dev/null +++ b/backend/tests/test_data_integrity.py @@ -0,0 +1,432 @@ +"""数据完整性检测 (停机缺口/盘中快照) 单元测试。 + +判据核心: quote_ts 仅实时 flush 写入真实毫秒时间戳 (batch 拉取/盘后计算为 +null); 历史交易日的 quote_ts 时刻 < 15:00 即盘中快照 → 坏。 +""" +from __future__ import annotations + +from datetime import date, datetime, time, timedelta, timezone +from types import SimpleNamespace + +import polars as pl +import pytest + +from app.market_time import CN_TZ +from app.services.data_integrity import ( + AUTO_REPAIR_MAX_LAG_DAYS, + IntegrityIssue, + _is_snapshot, + _quote_ts_max_ms, + earliest_issue_day, + prune_enriched_partitions, + scan_recent_integrity, + within_auto_repair_window, +) + +# 2026-08-19(周三) ~ 2026-08-21(周五) 是工作日; TODAY 取 2026-08-24(周一) +TODAY = date(2026, 8, 24) +FRIDAY = date(2026, 8, 21) +THURSDAY = date(2026, 8, 20) + + +def _ts_ms(day: date, t: time) -> int: + return int(datetime.combine(day, t, tzinfo=CN_TZ).timestamp() * 1000) + + +def _write_daily_partition(root, table: str, day: date, quote_ts: int | None, symbols=("600001.SH",)) -> None: + part = root / table / f"date={day.isoformat()}" + part.mkdir(parents=True, exist_ok=True) + n = len(symbols) + pl.DataFrame({ + "symbol": list(symbols), + "date": [day] * n, + "open": [10.0] * n, + "high": [10.1] * n, + "low": [9.9] * n, + "close": [10.0] * n, + "volume": [100.0] * n, + "amount": [1000.0] * n, + "quote_ts": [quote_ts] * n, + }).write_parquet(part / "part.parquet") + + +# ── 判据 ──────────────────────────────────────────────────────────── + + +def test_snapshot_predicate(): + noon = _ts_ms(FRIDAY, time(11, 58)) + after_close = _ts_ms(FRIDAY, time(15, 0, 30)) + assert _is_snapshot(FRIDAY, noon) is True + assert _is_snapshot(FRIDAY, after_close) is False + assert _is_snapshot(FRIDAY, None) is False # batch 历史 → 权威 + + +def test_quote_ts_max_reads_partition_statistics(tmp_path): + part = tmp_path / "date=2026-08-21" + part.mkdir() + pl.DataFrame({ + "symbol": ["a", "b"], + "quote_ts": [1000, 2000], + }).write_parquet(part / "part.parquet") + assert _quote_ts_max_ms(part) == 2000 + + +def test_quote_ts_max_none_for_all_null(tmp_path): + part = tmp_path / "date=2026-08-21" + part.mkdir() + pl.DataFrame({ + "symbol": ["a", "b"], + "quote_ts": [None, None], + }).write_parquet(part / "part.parquet") + assert _quote_ts_max_ms(part) is None + + +# ── 扫描 ──────────────────────────────────────────────────────────── + + +def test_batch_history_with_null_quote_ts_is_clean(tmp_path): + _write_daily_partition(tmp_path, "kline_daily", THURSDAY, None) + _write_daily_partition(tmp_path, "kline_daily", FRIDAY, None) + issues = scan_recent_integrity(tmp_path, today=TODAY) + # 周六/周日非工作日不扫; 无今日分区且周五为最新 → 周五之后无缺口 + assert issues == [] + + +def test_midday_snapshot_partition_is_flagged(tmp_path): + _write_daily_partition(tmp_path, "kline_daily", FRIDAY, _ts_ms(FRIDAY, time(11, 58))) + _write_daily_partition(tmp_path, "kline_daily", TODAY, _ts_ms(TODAY, time(10, 0))) + issues = scan_recent_integrity(tmp_path, today=TODAY) + assert [(i.day, i.table, i.kind) for i in issues] == [ + (FRIDAY, "kline_daily", "snapshot"), + ] + + +def test_final_snapshot_after_close_is_clean(tmp_path): + _write_daily_partition(tmp_path, "kline_daily", FRIDAY, _ts_ms(FRIDAY, time(15, 1))) + _write_daily_partition(tmp_path, "kline_daily", TODAY, _ts_ms(TODAY, time(10, 0))) + assert scan_recent_integrity(tmp_path, today=TODAY) == [] + + +def test_today_partition_is_never_flagged(tmp_path): + # 今天的盘中 quote_ts 属正常实时更新 + _write_daily_partition(tmp_path, "kline_daily", FRIDAY, None) + _write_daily_partition(tmp_path, "kline_daily", TODAY, _ts_ms(TODAY, time(9, 45))) + assert scan_recent_integrity(tmp_path, today=TODAY) == [] + + +def test_missing_tail_day_flagged(tmp_path): + # 周四有数据, 周五(工作日)整天停机缺失, 今天周一启动 + _write_daily_partition(tmp_path, "kline_daily", THURSDAY, None) + issues = scan_recent_integrity(tmp_path, today=TODAY) + assert [(i.day, i.table, i.kind) for i in issues] == [ + (FRIDAY, "kline_daily", "missing"), + ] + + +def test_snapshot_and_missing_both_reported(tmp_path): + # 周四盘中快照 + 周五缺失 + _write_daily_partition(tmp_path, "kline_daily", THURSDAY, _ts_ms(THURSDAY, time(13, 30))) + issues = scan_recent_integrity(tmp_path, today=TODAY) + kinds = {(i.day, i.kind) for i in issues} + assert (THURSDAY, "snapshot") in kinds + assert (FRIDAY, "missing") in kinds + assert earliest_issue_day(issues) == THURSDAY + + +def test_no_recent_activity_not_flagged(tmp_path): + # 最新分区早于扫描窗口 → 整族跳过 (首次启动/长期停用不自动修复) + old = TODAY - timedelta(days=30) + _write_daily_partition(tmp_path, "kline_daily", old, None) + assert scan_recent_integrity(tmp_path, today=TODAY) == [] + + +def test_interior_history_hole_not_flagged(tmp_path): + # 历史内部空洞是 laggards 另一类问题, 只报"晚于本地最新分区"的尾部缺口 + _write_daily_partition(tmp_path, "kline_daily", FRIDAY, None) + issues = scan_recent_integrity(tmp_path, today=TODAY) + assert issues == [] + + +def test_etf_family_independent(tmp_path): + # ETF 族近期无活动 → 不判定, 即便股票族有问题 + _write_daily_partition(tmp_path, "kline_daily", FRIDAY, _ts_ms(FRIDAY, time(11, 58))) + _write_daily_partition(tmp_path, "kline_daily", TODAY, _ts_ms(TODAY, time(10, 0))) + issues = scan_recent_integrity(tmp_path, today=TODAY) + assert all(i.table == "kline_daily" for i in issues) + assert earliest_issue_day(issues, ("kline_etf_daily",)) is None + + +def test_auto_repair_window(): + assert within_auto_repair_window(TODAY - timedelta(days=AUTO_REPAIR_MAX_LAG_DAYS), today=TODAY) is True + assert within_auto_repair_window(TODAY - timedelta(days=AUTO_REPAIR_MAX_LAG_DAYS + 1), today=TODAY) is False + assert within_auto_repair_window(None, today=TODAY) is False + + +# ── enriched 分区删除 ─────────────────────────────────────────────── + + +def test_prune_enriched_partitions_removes_only_range(tmp_path): + base = tmp_path / "kline_daily_enriched" + for day in (THURSDAY, FRIDAY, TODAY): + part = base / f"date={day.isoformat()}" + part.mkdir(parents=True) + (part / "part.parquet").write_bytes(b"x") + removed = prune_enriched_partitions(tmp_path, FRIDAY) + assert removed == 2 + assert (base / f"date={THURSDAY.isoformat()}").exists() + assert not (base / f"date={FRIDAY.isoformat()}").exists() + assert not (base / f"date={TODAY.isoformat()}").exists() + + +# ── 管道起点决策 (分支3降级后的起点) ──────────────────────────────── + + +def _resolve_daily_sync_start(latest_daily, stale_day): + # 与 daily_pipeline.run_now 分支4的起点表达式一致 (min(非空值)) + return min(d for d in (latest_daily, stale_day) if d is not None) + + +def test_branch4_start_takes_earliest_bad_day(): + # today_exists 场景: latest=今天, 坏日=上周五 → 起点必须是上周五 + assert _resolve_daily_sync_start(TODAY, FRIDAY) == FRIDAY + + +def test_branch4_start_without_stale_day_uses_latest(): + assert _resolve_daily_sync_start(FRIDAY, None) == FRIDAY + assert _resolve_daily_sync_start(TODAY, None) == TODAY + + +def test_timezone_conversion_is_cn(): + # quote_ts 是毫秒 Unix 时间戳, 必须按 UTC+8 折算 — 15:00 边界用例 + ts = int(datetime(2026, 8, 21, 7, 0, tzinfo=timezone.utc).timestamp() * 1000) # 北京 15:00 + assert _is_snapshot(FRIDAY, ts) is False + ts_morning = int(datetime(2026, 8, 21, 3, 58, tzinfo=timezone.utc).timestamp() * 1000) # 北京 11:58 + assert _is_snapshot(FRIDAY, ts_morning) is True + + +def test_issue_from_other_day_timestamp_not_flagged(tmp_path): + # 防御: quote_ts 日期与分区日期不符(跨天写入的脏数据)不判快照 + part = tmp_path / "kline_daily" / f"date={FRIDAY.isoformat()}" + part.mkdir(parents=True) + pl.DataFrame({ + "symbol": ["a"], "date": [FRIDAY], "quote_ts": [_ts_ms(THURSDAY, time(11, 0))], + }).write_parquet(part / "part.parquet") + issues = scan_recent_integrity(tmp_path, today=TODAY) + # 周五分区带周四时间戳 → 不判快照; 周四分区缺失且晚于最新(周五) → 不报 + assert issues == [] + + +def test_describe_and_issue_dataclass(): + issues = [IntegrityIssue(day=FRIDAY, table="kline_daily", kind="snapshot")] + from app.services.data_integrity import describe_issues + + assert "2026-08-21" in describe_issues(issues) + assert "盘中快照" in describe_issues(issues) + + +# ── 开实时行情门禁 (钩子2) ────────────────────────────────────────── + + +def _gate_state(tmp_path, quote_service, repo): + from types import SimpleNamespace + + return SimpleNamespace( + app=SimpleNamespace(state=SimpleNamespace( + quote_service=quote_service, + depth_service=None, + repo=SimpleNamespace(store=SimpleNamespace(data_dir=tmp_path)), + capabilities=None, + )) + ) + + +class _QuoteServiceStub: + def __init__(self, mode="market"): + self._mode = mode + self.enabled = False + + @staticmethod + def is_realtime_allowed(): + return True + + @staticmethod + def is_paused(): + return False + + @staticmethod + def realtime_mode(): + return "market" + + def enable(self): + self.enabled = True + + def disable(self): + self.enabled = False + + +def test_realtime_gate_blocks_on_snapshot_and_launches_repair(tmp_path, monkeypatch): + from fastapi import HTTPException + + from app.api import settings as settings_api + from app.services import data_integrity + + _write_daily_partition(tmp_path, "kline_daily", FRIDAY, _ts_ms(FRIDAY, time(11, 58))) + _write_daily_partition(tmp_path, "kline_daily", TODAY, _ts_ms(TODAY, time(10, 0))) + + launched = [] + monkeypatch.setattr( + data_integrity, "launch_integrity_repair", + lambda state, day, reason: (launched.append((day, reason)) or ("job-x", True)), + ) + saved = {} + monkeypatch.setattr( + "app.services.preferences.save", lambda payload: saved.update(payload), + ) + + qs = _QuoteServiceStub() + request = _gate_state(tmp_path, qs, repo=None) + req = settings_api.RealtimeQuotesPrefs(realtime_quotes_enabled=True) + + with pytest.raises(HTTPException) as exc_info: + settings_api.update_realtime_quotes(req, request) + + assert exc_info.value.status_code == 409 + assert "盘中快照" in exc_info.value.detail + assert "job-x" in exc_info.value.detail + # 修复任务以最早坏日为起点, 且实时行情未被开启 + assert launched == [(FRIDAY, "realtime_gate")] + assert saved == {} + + +def test_realtime_gate_allows_clean_data(tmp_path, monkeypatch): + from app.api import settings as settings_api + + _write_daily_partition(tmp_path, "kline_daily", FRIDAY, None) + _write_daily_partition(tmp_path, "kline_daily", TODAY, _ts_ms(TODAY, time(10, 0))) + + saved = {} + monkeypatch.setattr( + "app.services.preferences.save", lambda payload: saved.update(payload), + ) + qs = _QuoteServiceStub() + request = _gate_state(tmp_path, qs, repo=None) + req = settings_api.RealtimeQuotesPrefs(realtime_quotes_enabled=True) + + result = settings_api.update_realtime_quotes(req, request) + assert result["realtime_quotes_enabled"] is True + assert qs.enabled is True + assert saved == {"realtime_quotes_enabled": True} + + +def test_realtime_gate_ignores_old_issues_beyond_window(tmp_path, monkeypatch): + from app.api import settings as settings_api + + old_day = TODAY - timedelta(days=AUTO_REPAIR_MAX_LAG_DAYS + 1) + while old_day.weekday() >= 5: + old_day -= timedelta(days=1) + _write_daily_partition(tmp_path, "kline_daily", old_day, _ts_ms(old_day, time(11, 58))) + _write_daily_partition(tmp_path, "kline_daily", TODAY, _ts_ms(TODAY, time(10, 0))) + + saved = {} + monkeypatch.setattr( + "app.services.preferences.save", lambda payload: saved.update(payload), + ) + qs = _QuoteServiceStub() + request = _gate_state(tmp_path, qs, repo=None) + req = settings_api.RealtimeQuotesPrefs(realtime_quotes_enabled=True) + + result = settings_api.update_realtime_quotes(req, request) + assert result["realtime_quotes_enabled"] is True + + +def test_boot_check_launches_repair_within_window(tmp_path, monkeypatch): + from types import SimpleNamespace + + from app.services import data_integrity + + # boot_integrity_check 用真实"今天" — 往回找最近工作日造盘中快照分区 + launched = [] + monkeypatch.setattr( + data_integrity, "launch_integrity_repair", + lambda state, day, reason: (launched.append(day) or ("job-x", True)), + ) + + real_today = datetime.now(CN_TZ).date() + probe = real_today - timedelta(days=1) + while probe.weekday() >= 5: + probe -= timedelta(days=1) + data_dir = tmp_path / "boot" + _write_daily_partition(data_dir, "kline_daily", probe, _ts_ms(probe, time(11, 58))) + _write_daily_partition(data_dir, "kline_daily", real_today, _ts_ms(real_today, time(10, 0))) + + state = SimpleNamespace( + repo=SimpleNamespace(store=SimpleNamespace(data_dir=data_dir)), + ) + data_integrity.boot_integrity_check(state) + assert launched == [probe] + + +# ── 管道自愈端到端 (钩子3, 离线集成) ──────────────────────────────── + + +def _write_full_partition(root, table: str, day: date, quote_ts: int | None) -> None: + part = root / table / f"date={day.isoformat()}" + part.mkdir(parents=True, exist_ok=True) + pl.DataFrame({ + "symbol": ["600001.SH", "600002.SH"], + "date": [day, day], + "open": [10.0, 20.0], "high": [10.1, 20.1], + "low": [9.9, 19.9], "close": [10.0, 20.0], + "volume": [100.0, 200.0], "amount": [1000.0, 4000.0], + "quote_ts": [quote_ts, quote_ts], + }).write_parquet(part / "part.parquet") + + +def test_pipeline_self_heals_snapshot_day(tmp_path, monkeypatch): + """用户 bug 场景复刻: 昨天盘中快照 + 今天实时分区 → 管道应放弃"只刷今天", + 降级为从坏日起的范围拉取, 并把坏 enriched 分区删后重算。""" + from app.config import settings as app_settings + from app.jobs import daily_pipeline + from app.services import instrument_sync, kline_sync + from app.tickflow.repository import DataStore, KlineRepository + + today = datetime.now(CN_TZ).date() + yesterday = today - timedelta(days=1) + while yesterday.weekday() >= 5: + yesterday -= timedelta(days=1) + + _write_full_partition(tmp_path, "kline_daily", yesterday, _ts_ms(yesterday, time(11, 58))) + _write_full_partition(tmp_path, "kline_daily", today, _ts_ms(today, time(10, 0))) + _write_full_partition(tmp_path, "kline_daily_enriched", yesterday, _ts_ms(yesterday, time(11, 58))) + _write_full_partition(tmp_path, "kline_daily_enriched", today, _ts_ms(today, time(10, 0))) + + # 网络函数离线化: 维表同步 + 日K batch 拉取(记录参数) + monkeypatch.setattr(instrument_sync, "sync_instruments", lambda data_dir: 0) + batch_calls: list[dict] = [] + + def _fake_batch(universe, repo, capset, start_date=None, end_date=None, on_chunk_done=None): + batch_calls.append({ + "start": start_date.date() if hasattr(start_date, "date") else start_date, + "end": end_date.date() if hasattr(end_date, "date") else end_date, + }) + return 0 + + monkeypatch.setattr(kline_sync, "sync_and_persist_daily_batch", _fake_batch) + # run_pipeline() 不传 data_dir 时读 settings.data_dir — 同步指到 tmp + monkeypatch.setattr(app_settings, "data_dir", tmp_path) + + repo = KlineRepository(DataStore(tmp_path)) + capset = SimpleNamespace(has=lambda key: key == "QUOTE_POOL") + + result = daily_pipeline.run_now(repo, capset) # type: ignore[arg-type] + + # 分支3(实时覆写只刷今天)被降级 → 范围拉取起点=坏日 + assert batch_calls and batch_calls[0]["start"] == yesterday + assert result["integrity_repair_from"] == yesterday.isoformat() + assert result["integrity_issues"] >= 1 + # 坏 enriched 分区被删后当"新日期"重算写回 (无 prune 时 Step 2 走 skip 不写) + enriched_left = sorted( + p.name for p in (tmp_path / "kline_daily_enriched").glob("date=*") + ) + assert enriched_left == [f"date={yesterday.isoformat()}", f"date={today.isoformat()}"] + assert result["enriched_days"] > 0