perf(kline): 分钟批量补拉取到即落盘 + 尾部增量窗口

- 补拉结果经 _write_minute_partition upsert 落盘 (持仓库写锁, 与全量分钟
  服务/盘后同步同一纪律), 下一轮命中本地, 大自选从每轮全天重拉降为增量
- 完整性判定由 0.9 根数比例改为 期望-2 根: 比例阈值会让持久化数据在 90%
  处冻结尾巴
- 分类三态: fresh 直读本地零请求 / 尾部落后从最后一根+1min 增量拉 /
  中间有洞 (间距非 1min/午休 91min) 退回全天重拉回填, 防增量窗口漏数据
- 落盘失败降级为仅返回本轮数据 (持久化是优化而非正确性前提)

backend/app/api/kline.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 import logging
 import math
 from datetime import date, timedelta
+from pathlib import Path
 from functools import lru_cache
 from typing import Optional
 
@@ -599,11 +600,9 @@ def get_minute_batch(request: Request, body: dict):
     trade_date = date.fromisoformat(trade_date_str) if trade_date_str else cn_today()
 
     # 非交易日(周末/节假日)才回退到最近有数据的交易日; 否则盘中会显示昨天而非今天。
-    # 注意: 不能用 latest_minute_date_global() 判断盘中是否为交易日 —— 批量实时补拉
-    # 不落库 (见下方 sync_minute_batch 无 on_segment), 盘中它恒返回上次全量同步日,
-    # 用它做判据会导致 trade_date 永久回退到昨天, 再因 expected=240 判定昨日"完整"
-    # 而不再补拉今天, 形成永远显示昨日的死循环。
-    # 判据改为: 周末必回退; 工作日收盘后(>=15:30)仍无今日日K → 节假日, 回退。
+    # 判据: 周末必回退; 工作日收盘后(>=15:30)仍无今日日K → 节假日, 回退。
+    # (下方补拉已改为取到即落盘, 但只有真实交易时段才会写入当日分区,
+    #  节假日当日分区恒为空, 不影响该回退判据。)
     if not trade_date_str:
         today = cn_today()
         need_fallback = today.weekday() >= 5  # 周六/周日必非交易日
@@ -649,19 +648,36 @@ def get_minute_batch(request: Request, body: dict):
     else:
         expected = 240
 
-    # 按 symbol 分组, 判定哪些不完整需要补拉 (partition_by 一次切分, 同 daily-batch)
+    # 本地状态分类 (补拉已改为取到即落盘, 完整性判定随之收紧):
+    # - fresh:  根数 >= 期望-2 (时间边界容差), 直接用本地。原 0.9 比例阈值会让
+    #           持久化数据在 90% 处冻结尾巴, 必须按根数差判。
+    # - holes:  中间缺K (相邻间距非 1 分钟 / 非午休 91 分钟) → 全天重拉回填,
+    #           否则"最后一根+1min"的增量窗口永远不会回看中间的洞。
+    # - stale:  仅尾部落后 → 增量拉, 请求量从"每轮全天"降为"每轮一根"量级。
+    _LUNCH_GAP_MIN = 91  # 11:30 → 13:01
+
+    def _has_holes(sub: pl.DataFrame) -> bool:
+        gaps = sub["datetime"].diff().dt.total_minutes().drop_nulls()
+        return gaps.filter((gaps != 1) & (gaps != _LUNCH_GAP_MIN)).len() > 0
+
     result: dict[str, list[dict]] = {}
-    incomplete: list[str] = []
+    full_pull: list[str] = []          # 无数据或中间有洞 → 全天拉
+    stale_last: dict[str, datetime] = {}  # 尾部落后 → symbol → 最后一根时间
     local_parts: dict[str, pl.DataFrame] = {}
     if not df_local.is_empty():
         for part in df_local.partition_by("symbol", maintain_order=True):
             local_parts[part["symbol"][0]] = part.sort("datetime")
+    fresh_floor = max(0, expected - 2)
     for sym in symbols:
         sub = local_parts.get(sym, pl.DataFrame())
-        if expected > 0 and (sub.is_empty() or len(sub) < expected * 0.9):
-            incomplete.append(sym)
-        elif not sub.is_empty():
-            result[sym] = sub.to_dicts()
+        if expected == 0 or sub.height >= fresh_floor:
+            if not sub.is_empty():
+                result[sym] = sub.to_dicts()
+            continue
+        if sub.is_empty() or _has_holes(sub):
+            full_pull.append(sym)
+        else:
+            stale_last[sym] = sub["datetime"][-1]
 
     # prefer_local 生效判定: 仅当全量分钟服务健康 (freshness 契约, 见 minute_refresh.is_healthy)
     full_minute_healthy = False
@@ -669,59 +685,76 @@ def get_minute_batch(request: Request, body: dict):
         svc = getattr(request.app.state, "minute_refresh", None)
         full_minute_healthy = bool(svc is not None and svc.is_healthy())
     if full_minute_healthy:
-        # 股票 incomplete 不补拉, 本地有多少给多少 (服务下一轮写入补全);
-        # ETF 维持 incomplete 走下方补拉
-        for sym in incomplete:
-            sub = local_parts.get(sym)
-            if sym not in etf_set and sub is not None and not sub.is_empty():
-                result[sym] = sub.to_dicts()
-        incomplete = [s for s in incomplete if s in etf_set]
-
-    # Step 2: 缺失的 symbol 批量实时拉取 (不落库)
-    if incomplete:
-        start_time = datetime(trade_date.year, trade_date.month, trade_date.day, 9, 25, 0)
-        end_time = datetime(trade_date.year, trade_date.month, trade_date.day, 15, 5, 0)
-        lim = capset.limits(Cap.KLINE_MINUTE_BATCH)
-        # etf_set 已在上方获取, 直接复用 — 按 asset_type 拆分调用 sync_minute_batch
-        # (自定义源 / TickFlow 路由均依赖 asset_type 正确传递)
-        # 契约: 本端点只接受 stock/ETF (指数分钟K走 /api/index/minute 独立路径),
-        # 故两分支已覆盖全部 incomplete。若未来放开指数支持, 需额外加 index 分支
-        # 以避免被误路由为 stock。
-        stock_incomplete = [s for s in incomplete if s not in etf_set]
-        etf_incomplete = [s for s in incomplete if s in etf_set]
-        live_parts: list[pl.DataFrame] = []
-        if stock_incomplete:
-            df_s = kline_sync.sync_minute_batch(
-                stock_incomplete,
-                start_time=start_time,
-                end_time=end_time,
-                batch_size=lim.batch if lim else None,
-                rpm=lim.rpm if lim else None,
-                asset_type="stock",
-            )
-            if not df_s.is_empty():
-                live_parts.append(df_s)
-        if etf_incomplete:
-            df_e = kline_sync.sync_minute_batch(
-                etf_incomplete,
-                start_time=start_time,
-                end_time=end_time,
-                batch_size=lim.batch if lim else None,
-                rpm=lim.rpm if lim else None,
-                asset_type="etf",
-            )
-            if not df_e.is_empty():
-                live_parts.append(df_e)
-        if live_parts:
-            live_df = pl.concat(live_parts, how="diagonal_relaxed")
-            live_map: dict[str, pl.DataFrame] = {
-                part["symbol"][0]: part.sort("datetime")
-                for part in live_df.partition_by("symbol", maintain_order=True)
-            }
-            for sym in incomplete:
-                sub = live_map.get(sym)
+        # 股票缺口不补拉, 本地有多少给多少 (服务下一轮写入补全);
+        # ETF 不在 universe 内, 维持补拉
+        for sym in [*full_pull, *stale_last]:
+            if sym not in etf_set:
+                sub = local_parts.get(sym)
                 if sub is not None and not sub.is_empty():
                     result[sym] = sub.to_dicts()
+        full_pull = [s for s in full_pull if s in etf_set]
+        stale_last = {s: t for s, t in stale_last.items() if s in etf_set}
+
+    # Step 2: 补拉并落盘 (取到即写, upsert 语义; 下一轮命中本地, 请求量骤降)。
+    # 落盘失败只降级 (log 后继续返回本轮数据), 不影响响应 —— 持久化是优化而非正确性前提。
+    # 契约: 本端点只接受 stock/ETF (指数分钟K走 /api/index/minute 独立路径),
+    # 按 asset_type 拆分调用 (自定义源 / TickFlow 路由均依赖 asset_type 正确传递)。
+    day_start = datetime(trade_date.year, trade_date.month, trade_date.day, 9, 25, 0)
+    session_end = datetime(trade_date.year, trade_date.month, trade_date.day, 15, 5, 0)
+    lim = capset.limits(Cap.KLINE_MINUTE_BATCH)
+    minute_dirs = {
+        "stock": repo.store.data_dir / "kline_minute",
+        "etf": repo.store.data_dir / "kline_etf_minute",
+    }
+    live_map: dict[str, pl.DataFrame] = {}
+
+    def _pull(asset: str, sym_list: list[str], start: datetime) -> None:
+        if not sym_list:
+            return
+        df_live = kline_sync.sync_minute_batch(
+            sym_list,
+            start_time=start,
+            end_time=session_end,
+            batch_size=lim.batch if lim else None,
+            rpm=lim.rpm if lim else None,
+            asset_type=asset,
+        )
+        if df_live.is_empty():
+            return
+        try:
+            # 读-改-写必须持仓库写锁 (与全量分钟服务/盘后同步同一纪律, Windows 临时文件占用)。
+            # 仅在拿到真实目录时落盘: data_dir 异常 (非 Path) 时跳过, 只返回本轮数据。
+            minute_dir = minute_dirs[asset]
+            if isinstance(minute_dir, Path):
+                with repo._write_lock:
+                    kline_sync._write_minute_partition(df_live, minute_dir)
+        except Exception as e:  # noqa: BLE001
+            logger.warning("minute-batch 补拉落盘失败 (降级为仅返回): %s", e)
+        for part in df_live.partition_by("symbol", maintain_order=True):
+            live_map[part["symbol"][0]] = part.sort("datetime")
+
+    _pull("stock", [s for s in full_pull if s not in etf_set], day_start)
+    _pull("etf", [s for s in full_pull if s in etf_set], day_start)
+    if stale_last:
+        # 增量公共起点 = 最旧尾部 + 1min; 起点更早的重叠由落盘/合并去重吸收
+        inc_start = min(stale_last.values()) + timedelta(minutes=1)
+        if inc_start < session_end:
+            _pull("stock", [s for s in stale_last if s not in etf_set], inc_start)
+            _pull("etf", [s for s in stale_last if s in etf_set], inc_start)
+
+    # 合并: 有增量/回填的 symbol = 本地 + 拉取 upsert; 仅拉到的 (missing) 直接进结果
+    for sym, sub in local_parts.items():
+        live = live_map.get(sym)
+        if live is not None:
+            merged = (
+                pl.concat([sub, live])
+                .unique(subset=["symbol", "datetime"], keep="last")
+                .sort("datetime")
+            )
+            result[sym] = merged.to_dicts()
+    for sym, live in live_map.items():
+        if sym not in result:
+            result[sym] = live.to_dicts()
 
     # full_minute_local: 本轮 prefer_local 生效 (本地分区由全量分钟服务供给, 股票未做补拉)
     return {"data": result, "full_minute_local": full_minute_healthy}

backend/tests/test_minute_routing.py

@@ -12,6 +12,7 @@ mock 范式沿用 test_stocksdk_provider.py (monkeypatch 模块属性)。
 from __future__ import annotations
 
 from datetime import date, datetime, timedelta
+from pathlib import Path
 from threading import Lock
 from unittest.mock import MagicMock
 
@@ -316,6 +317,102 @@ def test_get_minute_batch_splits_stock_and_etf(monkeypatch):
     assert "510300.SH" in result["data"]
 
 
+# ---------- 测试 9b: 取到即落盘 + 增量拉取 (尾部落后 / 中间洞 / fresh) ----------
+
+def _bars(symbol: str, dts: list) -> pl.DataFrame:
+    """构造 canonical 8 列分钟K帧 (dts 为 datetime 列表)。"""
+    n = len(dts)
+    return pl.DataFrame({
+        "symbol": [symbol] * n,
+        "datetime": dts,
+        "open": [10.0] * n, "high": [10.1] * n, "low": [9.9] * n, "close": [10.0] * n,
+        "volume": [100.0] * n, "amount": [1000.0] * n,
+    })
+
+
+def _endpoint_mocks(monkeypatch, local_df: pl.DataFrame, sync_ret: pl.DataFrame | None):
+    """get_minute_batch 的最小 mock: repo/capset + sync/落盘 spy。返回 (捕获, 落盘, request)。"""
+    from app.api import kline as kline_api
+
+    captured: list[dict] = []
+
+    def fake_sync(symbols, *, start_time, end_time, batch_size, rpm, asset_type):
+        captured.append({"symbols": list(symbols), "start": start_time, "asset": asset_type})
+        return sync_ret if sync_ret is not None else pl.DataFrame()
+
+    monkeypatch.setattr(kline_api.kline_sync, "sync_minute_batch", fake_sync)
+    writes: list[pl.DataFrame] = []
+    monkeypatch.setattr(kline_api.kline_sync, "_write_minute_partition",
+                        lambda df, d: writes.append(df))
+
+    mock_repo = MagicMock()
+    mock_repo.get_etf_symbol_set.return_value = set()
+    mock_repo.get_minute_batch.return_value = local_df
+    mock_repo._write_lock = Lock()
+    # 真实 Path 才会触发落盘分支 (kline.py 的 isinstance 守卫)
+    mock_repo.store.data_dir = Path("data")
+
+    mock_capset = MagicMock()
+    mock_capset.has.return_value = True
+    mock_capset.limits.return_value = None
+
+    mock_request = MagicMock()
+    mock_request.app.state.repo = mock_repo
+    mock_request.app.state.capabilities = mock_capset
+    return captured, writes, mock_request
+
+
+def test_minute_batch_tail_stale_pulls_incremental_and_persists(monkeypatch):
+    """尾部落后 (本地连续但根数不足) → 从最后一根+1min 增量拉;
+    拉取结果落盘, 响应为本地+增量合并去重。"""
+    from app.api import kline as kline_api
+
+    local = _bars("600519.SH", [
+        datetime(2026, 1, 15, 9, 31), datetime(2026, 1, 15, 9, 32), datetime(2026, 1, 15, 9, 33),
+    ])
+    inc = _bars("600519.SH", [datetime(2026, 1, 15, 9, 34), datetime(2026, 1, 15, 9, 35)])
+    captured, writes, req = _endpoint_mocks(monkeypatch, local, sync_ret=inc)
+
+    result = kline_api.get_minute_batch(req, {"symbols": ["600519.SH"], "date": "2026-01-15"})
+
+    assert len(captured) == 1
+    assert captured[0]["start"] == datetime(2026, 1, 15, 9, 34)   # 最后一根 + 1min
+    assert captured[0]["asset"] == "stock"
+    assert writes and writes[0].height == 2                        # 取到即落盘
+    rows = result["data"]["600519.SH"]
+    assert len(rows) == 5                                          # 3 本地 + 2 增量
+    assert rows[-1]["datetime"] == datetime(2026, 1, 15, 9, 35)
+
+
+def test_minute_batch_middle_hole_falls_back_to_full_day(monkeypatch):
+    """中间缺K (间距 2min) → 增量窗口永远回看不到洞, 必须退回全天重拉。"""
+    from app.api import kline as kline_api
+
+    local = _bars("600519.SH", [datetime(2026, 1, 15, 9, 31), datetime(2026, 1, 15, 9, 33)])
+    captured, _, req = _endpoint_mocks(monkeypatch, local, sync_ret=pl.DataFrame())
+
+    kline_api.get_minute_batch(req, {"symbols": ["600519.SH"], "date": "2026-01-15"})
+
+    assert len(captured) == 1
+    assert captured[0]["start"] == datetime(2026, 1, 15, 9, 25)    # 全天窗口
+
+
+def test_minute_batch_fresh_local_skips_pull(monkeypatch):
+    """本地完整 (240 根, 含午休 91min 间距) → 不发任何拉取请求, 直读本地。"""
+    from app.api import kline as kline_api
+
+    dts = ([datetime(2026, 1, 15, 9, 31) + timedelta(minutes=i) for i in range(120)]
+           + [datetime(2026, 1, 15, 13, 1) + timedelta(minutes=i) for i in range(120)])
+    local = _bars("600519.SH", dts)
+    captured, writes, req = _endpoint_mocks(monkeypatch, local, sync_ret=pl.DataFrame())
+
+    result = kline_api.get_minute_batch(req, {"symbols": ["600519.SH"], "date": "2026-01-15"})
+
+    assert captured == []                                          # 零请求
+    assert writes == []
+    assert len(result["data"]["600519.SH"]) == 240
+
+
 # ---------- 测试 10: sync_minute_batch 自定义源成功时调 on_segment (Issue 1) ----------
 
 def test_sync_minute_batch_custom_calls_on_segment(monkeypatch):