fix(kline): 实时行情覆写当日日K分区时补写 quote_ts

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

@@ -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 服务器在盘中把日分区写成服务器本地日期。

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