From 4bbc7d07a3ed1a743c9d4a34b7e19c5360a609c0 Mon Sep 17 00:00:00 2001 From: kevin9327 <5299031+kevin9327@users.noreply.github.com> Date: Wed, 9 Sep 2026 07:25:35 +0900 Subject: [PATCH] =?UTF-8?q?fix(kline):=20=E5=AE=9E=E6=97=B6=E8=A1=8C?= =?UTF-8?q?=E6=83=85=E8=A6=86=E5=86=99=E5=BD=93=E6=97=A5=E6=97=A5K?= =?UTF-8?q?=E5=88=86=E5=8C=BA=E6=97=B6=E8=A1=A5=E5=86=99=20quote=5Fts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit sync_daily_by_quotes 手工拼 record 时漏了行情响应的 timestamp, 而 flush_live_daily 是整分区覆写, 会把 QuoteService 之前写入的 quote_ts 一并 抹掉。data_integrity 的判据全建立在 quote_ts 上, 分区因此从"盘中快照" 退化为"权威历史": 盘中触发同步后停机, 次日自检漏判, 停机时刻的 close/volume 永久留存并污染 lookback 指标。 --- backend/app/services/kline_sync.py | 4 +++ backend/tests/test_data_integrity.py | 37 ++++++++++++++++++++++++++++ 2 files changed, 41 insertions(+) diff --git a/backend/app/services/kline_sync.py b/backend/app/services/kline_sync.py index 06e3728..00867be 100644 --- a/backend/app/services/kline_sync.py +++ b/backend/app/services/kline_sync.py @@ -318,11 +318,15 @@ def sync_daily_by_quotes(repo: KlineRepository) -> int: "close": q.get("last_price"), "volume": q.get("volume"), "amount": q.get("amount"), + # 快照时刻标记: data_integrity 靠 quote_ts 区分盘中快照与盘后权威历史, + # 缺失会让盘中覆写的分区在停机后被当成完整历史, 永远不进修复。 + "quote_ts": q.get("timestamp"), }) df = pl.DataFrame(records) if df.is_empty(): return 0 + df = df.with_columns(pl.col("quote_ts").cast(pl.Int64, strict=False)) # 分区日期用北京交易日 (与 quote_service._build_daily 的 cn_today 一致), # 避免 UTC 服务器在盘中把日分区写成服务器本地日期。 diff --git a/backend/tests/test_data_integrity.py b/backend/tests/test_data_integrity.py index 30353d2..6f7a9e4 100644 --- a/backend/tests/test_data_integrity.py +++ b/backend/tests/test_data_integrity.py @@ -624,3 +624,40 @@ def test_pipeline_self_heals_snapshot_day(tmp_path, monkeypatch): ) assert enriched_left == [f"date={yesterday.isoformat()}", f"date={today.isoformat()}"] assert result["enriched_days"] > 0 + + +def test_quotes_flush_partition_keeps_quote_ts_for_integrity_scan(tmp_path, monkeypatch): + """实时行情覆写当日分区必须写 quote_ts, 否则停机后自检漏判盘中快照。 + + 盘中手动触发盘后管道时, "今天已有数据 → 实时行情覆写"分支会用 + tf.quotes.get_by_universes 整分区覆写。若覆写不带 quote_ts, 停机后次日 + 启动自检把这份半日数据当成 batch 权威历史, 停机时刻的 close/volume 永久 + 留存并污染 lookback 指标 —— 正是本模块要拦的场景。 + """ + from app.services import kline_sync + from app.tickflow import client as tf_client + from app.tickflow.repository import DataStore, KlineRepository + + snapshot_ms = _ts_ms(FRIDAY, time(11, 58)) + + class _FakeQuotes: + @staticmethod + def get_by_universes(universes): + return [{ + "symbol": "600001.SH", + "open": 10.0, "high": 10.2, "low": 9.8, "last_price": 10.1, + "volume": 1000.0, "amount": 10100.0, + "timestamp": snapshot_ms, + }] + + monkeypatch.setattr(tf_client, "get_client", lambda: SimpleNamespace(quotes=_FakeQuotes)) + monkeypatch.setattr(kline_sync, "cn_today", lambda: FRIDAY) + + repo = KlineRepository(DataStore(tmp_path)) + assert kline_sync.sync_daily_by_quotes(repo) == 1 + + issues = scan_recent_integrity(tmp_path, today=TODAY) + assert [(i.day, i.table, i.kind) for i in issues] == [(FRIDAY, "kline_daily", "snapshot")] + + part_dir = tmp_path / "kline_daily" / f"date={FRIDAY.isoformat()}" + assert _quote_ts_max_ms(part_dir) == snapshot_ms