mirror of
https://ghfast.top/https://github.com/aeroxw/tick-stock-panel.git
synced 2026-09-12 15:34:16 +08:00
fix(pipeline): 停机缺口检测与自动修复, 防止盘中快照数据永久损坏
场景: 用户盘中停机后次日启动并开实时行情, 实时 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。
This commit is contained in:
shy3130
@@ -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)
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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")
|
||||
@@ -8,7 +8,9 @@
|
||||
这样修正功能与盘后管道永远保持一致,不会出现遗漏。
|
||||
|
||||
落盘是 merge-upsert (按 symbol+date 去重 keep="last"), 新数据天然覆盖旧值,
|
||||
不会产生重复,也不需要先删分区。
|
||||
不会产生重复。日K/指数/ETF 族无需先删分区; 股票 enriched 例外 — 增量重算
|
||||
只算 enriched 里不存在的日期, "分区已存在但内容是盘中快照"的场景由管道内
|
||||
的 prune_enriched_partitions 先删坏分区再重算 (run_now 内处理, 此处无需关心)。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user