perf(intraday): 分时链路增量传输 + SSE 续画 + 视口感知

- minute-batch 支持 since 增量响应: 只回 >= since 的K (含动态最后一根),
  客户端按 (symbol, datetime) upsert 合并, 轮询流量从全天序列降到每轮一两根
- 自选分时 SSE 续画: enriched tick 前端续写分钟序列最后一根/滚动追加新K
  (纯视图叠加不写缓存, 服务端定版值始终权威); tick 新鲜时轮询降为 30s 兜底
- 视口感知: 从虚拟器直读渲染中 symbol 集合 (可见+overscan), 数据层只拉视口内;
  since 只按本轮请求集合取最旧尾部, 视口外落后不放大窗口; 集合变化 300ms
  防抖补拉; 剔除指数保持端点契约
- since 过滤专项测试: 含 since 当根 / 早于全部K / 无 since 三态

backend/app/api/kline.py

@@ -583,6 +583,15 @@ def get_minute_batch(request: Request, body: dict):
 
     symbols: list[str] = body.get("symbols", [])
     trade_date_str: str | None = body.get("date")
+    # 增量响应: since (ISO datetime) 之前的K不回传, 客户端本地缓存合并。
+    # since 应传客户端已持有的最后一根时间 — 形成中的动态K >= since, 每轮覆盖。
+    since_str = body.get("since")
+    since_dt: datetime | None = None
+    if since_str:
+        try:
+            since_dt = datetime.fromisoformat(str(since_str))
+        except ValueError:
+            since_dt = None
     # 自选分时本地优先标志: 全量分钟服务健康时, 股票缺口不再批量补拉
     # (本地分区由服务按间隔持续写入, 下一轮自然补全; 停牌/临停票补拉也是空, 无损)。
     # ETF 不在全量分钟 universe 内, 恒走补拉。服务不健康时回落现状补拉兜底。
@@ -757,8 +766,20 @@ def get_minute_batch(request: Request, body: dict):
         if sym not in result:
             result[sym] = live.to_dicts()
 
+    # since 增量过滤: 只回 >= since 的K (含动态最后一根), 无新增的 symbol 不回
+    if since_dt is not None:
+        result = {
+            sym: [r for r in rows if r["datetime"] >= since_dt]
+            for sym, rows in result.items()
+        }
+        result = {sym: rows for sym, rows in result.items() if rows}
+
     # full_minute_local: 本轮 prefer_local 生效 (本地分区由全量分钟服务供给, 股票未做补拉)
-    return {"data": result, "full_minute_local": full_minute_healthy}
+    return {
+        "data": result,
+        "full_minute_local": full_minute_healthy,
+        "incremental": since_dt is not None,
+    }
 
 
 @router.get("/minute-range")

backend/tests/test_minute_routing.py

@@ -413,6 +413,34 @@ def test_minute_batch_fresh_local_skips_pull(monkeypatch):
     assert len(result["data"]["600519.SH"]) == 240
 
 
+def test_minute_batch_since_returns_only_new_bars(monkeypatch):
+    """since 增量响应: 只回 >= since 的K (含 since 本身 — 形成中的动态K需覆盖),
+    无新增的 symbol 不出现在 data; incremental 标志为 True。"""
+    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)   # fresh: 不触发补拉, 直读本地后做 since 过滤
+    _, _, req = _endpoint_mocks(monkeypatch, local, sync_ret=pl.DataFrame())
+
+    result = kline_api.get_minute_batch(req, {
+        "symbols": ["600519.SH"], "date": "2026-01-15",
+        "since": "2026-01-15T15:00:00",
+    })
+    rows = result["data"]["600519.SH"]
+    assert [r["datetime"] for r in rows] == [datetime(2026, 1, 15, 15, 0)]   # 含 since 当根
+    assert result["incremental"] is True
+
+    # since 早于全部K → 全量返回; 无 since → incremental False
+    full = kline_api.get_minute_batch(req, {
+        "symbols": ["600519.SH"], "date": "2026-01-15", "since": "2026-01-15T09:00:00",
+    })
+    assert len(full["data"]["600519.SH"]) == 240
+    plain = kline_api.get_minute_batch(req, {"symbols": ["600519.SH"], "date": "2026-01-15"})
+    assert plain["incremental"] is False
+    assert len(plain["data"]["600519.SH"]) == 240
+
+
 # ---------- 测试 10: sync_minute_batch 自定义源成功时调 on_segment (Issue 1) ----------
 
 def test_sync_minute_batch_custom_calls_on_segment(monkeypatch):

frontend/src/lib/api.ts

@@ -2052,10 +2052,10 @@ export const api = {
       method: 'POST',
       body: JSON.stringify({ symbols, days }),
     }),
-  klineMinuteBatch: (symbols: string[], date?: string, preferLocal?: boolean) =>
-    request<{ data: Record<string, MinuteKlineRow[]>; full_minute_local?: boolean }>('/api/kline/minute-batch', {
+  klineMinuteBatch: (symbols: string[], date?: string, preferLocal?: boolean, since?: string) =>
+    request<{ data: Record<string, MinuteKlineRow[]>; full_minute_local?: boolean; incremental?: boolean }>('/api/kline/minute-batch', {
       method: 'POST',
-      body: JSON.stringify({ symbols, date, ...(preferLocal ? { prefer_local: true } : {}) }),
+      body: JSON.stringify({ symbols, date, ...(preferLocal ? { prefer_local: true } : {}), ...(since ? { since } : {}) }),
     }),
   instrumentSearch: (q: string, limit = 20, assetTypes?: string) =>
     request<{ results: { symbol: string; name: string; code: string; asset_type?: string }[] }>(

frontend/src/lib/minuteBatchIncremental.ts

@@ -0,0 +1,53 @@
+import type { QueryClient } from '@tanstack/react-query'
+import { api, type MinuteKlineRow } from '@/lib/api'
+
+// 分钟批量分时的增量轮询助手:
+// 后端 /api/kline/minute-batch 已支持 since 增量 (只回 >= since 的K, 含形成中
+// 动态最后一根)。这里在 queryFn 内读取 react-query 缓存里的上一轮全量序列,
+// 以"各 symbol 最后一根的最旧时间"为 since 请求增量, 并按 (symbol, datetime)
+// upsert 合并后返回 — 组件层拿到的仍是完整序列, 代码零改动。
+// 缓存不存在 (首次/换标的池/换日) 时不带 since, 全量拉取。
+
+type MinuteBatchData = Record<string, MinuteKlineRow[]>
+
+function lastBarTs(data: MinuteBatchData, symbols?: string[]): number | null {
+  // since 只按本轮请求的 symbol 取最旧最后一根: 视口感知下不可见 symbol 可能
+  // 落后很多分钟, 把它们计入会把整个批量窗口拉大重拉
+  const scope = symbols ? new Set(symbols) : null
+  let min: number | null = null
+  for (const [sym, rows] of Object.entries(data)) {
+    if (scope && !scope.has(sym)) continue
+    const last = rows[rows.length - 1]
+    if (!last) continue
+    const t = new Date(last.datetime).getTime()
+    if (Number.isFinite(t) && (min === null || t < min)) min = t
+  }
+  return min
+}
+
+function mergeInto(base: MinuteBatchData, patch: MinuteBatchData): MinuteBatchData {
+  const merged: MinuteBatchData = { ...base }
+  for (const [sym, rows] of Object.entries(patch)) {
+    const cache = merged[sym] ?? []
+    const byTs = new Map(cache.map(r => [r.datetime, r]))
+    for (const r of rows) byTs.set(r.datetime, r)   // 新值覆盖 (动态K定版)
+    merged[sym] = Array.from(byTs.values())
+      .sort((a, b) => (a.datetime < b.datetime ? -1 : a.datetime > b.datetime ? 1 : 0))
+  }
+  return merged
+}
+
+/** queryFn 用: 读缓存 → 增量请求 → 合并返回 { data: 完整序列 } (保持端点响应形状, 下游零改动) */
+export async function fetchMinuteBatchIncremental(
+  qc: QueryClient,
+  cacheKey: readonly unknown[],
+  symbols: string[],
+  preferLocal?: boolean,
+): Promise<{ data: MinuteBatchData }> {
+  const prev = qc.getQueryData<{ data: MinuteBatchData }>(cacheKey)?.data
+  const minTs = prev ? lastBarTs(prev, symbols) : null
+  const since = minTs !== null ? new Date(minTs).toISOString() : undefined
+  const resp = await api.klineMinuteBatch(symbols, undefined, preferLocal, since)
+  if (!since || !resp.incremental) return { data: resp.data ?? {} }
+  return { data: mergeInto(prev ?? {}, resp.data ?? {}) }
+}

frontend/src/pages/Screener.tsx

@@ -3,6 +3,7 @@ import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query'
 import { motion } from 'framer-motion'
 import { ScanSearch, Clock, TrendingUp, Star, Filter, Layers, Network, Sparkles, RefreshCw, Settings2, Store, RotateCcw, X } from 'lucide-react'
 import { api, genRuleId, type ScreenerStrategy, type ScreenerResult } from '@/lib/api'
+import { fetchMinuteBatchIncremental } from '@/lib/minuteBatchIncremental'
 import { DEFAULT_STRATEGY_NOTIFY_EVENTS } from '@/lib/strategyMonitorEvents'
 import { toast } from '@/components/Toast'
 import { useDataStatus, usePreferences, useCapabilities, useQuoteStatus } from '@/lib/useSharedQueries'
@@ -460,7 +461,8 @@ export function Screener() {
 
   const minuteBatch = useQuery({
     queryKey: QK.minuteBatch(intradaySymbolsKey),
-    queryFn: () => api.klineMinuteBatch(intradayRequestSymbols),
+    // 增量轮询: 读缓存以最后一根为 since 只拉新增, 本地合并为完整序列
+    queryFn: () => fetchMinuteBatchIncremental(qc, QK.minuteBatch(intradaySymbolsKey), intradayRequestSymbols),
     enabled: intradayVisible && intradayRequestSymbols.length > 0,
     staleTime: 10_000,
     placeholderData: previousData => previousData,

frontend/src/pages/Watchlist.tsx

@@ -5,6 +5,7 @@ import { useVirtualizer } from '@tanstack/react-virtual'
 import { motion, AnimatePresence } from 'framer-motion'
 import { Trash2, RefreshCw, Star, X, Search, LayoutGrid, List, Rows3, BarChart3, Settings2, Plus, Check, Filter, Eye, EyeOff, Minus, ChevronsUp, Clock, RotateCcw, ImagePlus, FolderOpen, FolderMinus, FolderPlus } from 'lucide-react'
 import { api, type KlineRow, type MinuteKlineRow, type WatchlistGroup, type WatchlistGroupColor } from '@/lib/api'
+import { fetchMinuteBatchIncremental } from '@/lib/minuteBatchIncremental'
 import { QK } from '@/lib/queryKeys'
 import { storage } from '@/lib/storage'
 import { fmtPrice, fmtPct, fmtBigNum, priceColorClass, formatExtNumber } from '@/lib/format'
@@ -878,16 +879,74 @@ export function Watchlist() {
   const { data: prefsData } = usePreferences()
   const intradayRefreshEnabled = prefsData?.minute_intraday_refresh ?? false
   const intradayRefreshInterval = prefsData?.minute_intraday_refresh_interval ?? 6
+  // 视口感知: 虚拟器渲染中的 symbol 集合 (可见 + overscan 缓冲), 由下方虚拟器
+  // 区域的 effect 写入; null = 未就绪/非虚拟化视图, 沿用全列表。首轮全量播种
+  // 缓存 (queryFn 在挂载时先于 ref 填充执行), 之后各轮只拉视口集合。
+  const minuteRequestSymbolsRef = useRef<string[] | null>(null)
   const minuteBatch = useQuery({
     queryKey: QK.minuteBatch(minuteSymbolsKey),
-    // prefer_local: 全量分钟服务健康时股票缺口不再批量补拉 (本地分区由服务持续写入),
-    // 大自选(数百上千只)不再持续打批量分钟接口
-    queryFn: () => api.klineMinuteBatch(minuteSymbols, undefined, true),
+    // 增量轮询 (prefer_local): 读缓存以最后一根为 since 只拉新增, 本地合并为
+    // 完整序列; 全量分钟健康时服务端零补拉, 不健康时也只增量。
+    // 请求集合 = 视口内 symbol (缓存按 symbol upsert, 视口外保留旧序列)
+    queryFn: () => {
+      const reqSymbols = minuteRequestSymbolsRef.current ?? minuteSymbols
+      return fetchMinuteBatchIncremental(qc, QK.minuteBatch(minuteSymbolsKey), reqSymbols, true)
+    },
     enabled: intradayVisible && minuteSymbols.length > 0 && !groupCardsOpen,
     staleTime: 10_000,
-    refetchInterval: (intradayRefreshEnabled && realtimeRunning) ? intradayRefreshInterval * 1000 : false,
+    // SSE tick 新鲜 (enriched 10s 内被行情推送刷新过) → 分时图已由下方续画本地
+    // 跳动, 轮询降为 30s 兜底校准; tick 断流时回到用户设定间隔
+    refetchInterval: () => {
+      if (!(intradayRefreshEnabled && realtimeRunning)) return false
+      const tickFresh = Date.now() - enriched.dataUpdatedAt < 10_000
+      return tickFresh ? 30_000 : intradayRefreshInterval * 1000
+    },
   })
-  const minuteData = intradayVisible ? (minuteBatch.data?.data ?? {}) : {}
+
+  // 分时图 SSE 续画: enriched 行情列每 tick 刷新 (SSE 触发), 前端本地续写分钟序列
+  // 的最后一根 / 分钟滚动追加新K — 轮询间隔内分时图也随实时价跳动, 零额外请求。
+  // 纯视图叠加不写回缓存: 轮询拉回的服务端定版K始终是权威值, 覆盖续画值。
+  const minuteData = useMemo(() => {
+    const base: Record<string, MinuteKlineRow[]> = intradayVisible ? (minuteBatch.data?.data ?? {}) : {}
+    const liveRows = enriched.data?.rows
+    if (!intradayVisible || !liveRows?.length) return base
+    const liveBySymbol = new Map<string, any>(liveRows.map((r: any) => [r.symbol, r]))
+    // 仅连续竞价时段续画 (9:31-11:30, 13:01-15:00); 收盘后 rt_price 即收盘价无需续写
+    const now = new Date()
+    const hh = now.getHours(), mm = now.getMinutes()
+    const inSession =
+      (hh === 9 && mm >= 31) || hh === 10 ||
+      (hh === 11 && mm <= 30) || (hh === 13 && mm >= 1) || hh === 14
+    if (!inSession) return base
+    // 与服务端同构的 naive 北京时间戳 (手工拼本地时间, 不能用 toISOString — 那是 UTC)
+    const pad = (n: number) => String(n).padStart(2, '0')
+    const barTs = `${now.getFullYear()}-${pad(now.getMonth() + 1)}-${pad(now.getDate())}T${pad(hh)}:${pad(mm)}:00`
+    const patched: Record<string, MinuteKlineRow[]> = {}
+    for (const sym of Object.keys(base)) {
+      const arr = base[sym]
+      if (!Array.isArray(arr) || arr.length === 0) { patched[sym] = arr; continue }
+      const live = liveBySymbol.get(sym)
+      const price = live?.rt_price ?? live?.close
+      if (typeof price !== 'number' || !Number.isFinite(price)) { patched[sym] = arr; continue }
+      const last = arr[arr.length - 1]
+      if (last.datetime === barTs) {
+        patched[sym] = [...arr.slice(0, -1), {
+          ...last,
+          close: price,
+          high: Math.max(last.high, price),
+          low: Math.min(last.low, price),
+        }]
+      } else if (barTs > last.datetime) {
+        patched[sym] = [...arr, {
+          datetime: barTs, open: price, high: price, low: price, close: price,
+          volume: 0, amount: 0,   // 量/额由下一轮轮询定版覆盖
+        }]
+      } else {
+        patched[sym] = arr   // 轮询数据已新于本地时钟 (钟差兜底), 不动
+      }
+    }
+    return patched
+  }, [intradayVisible, minuteBatch.data, enriched.data])
 
   const addMutation = useMutation({
     mutationFn: ({ symbol, groupId }: { symbol: string; groupId: string | null }) =>
@@ -1214,6 +1273,40 @@ export function Watchlist() {
     scrollMargin: cardScrollMargin,
   })
 
+  // 视口感知 (数据层): 从虚拟器派生"正在渲染的 symbol" (可见 + overscan 缓冲,
+  // 滚动前已就绪)。非虚拟化视图 (小列表/表格/分组) 为 null → 沿用全列表。
+  // 每次渲染直读 getVirtualItems (滚动不换 deps, 不能 useMemo), 副作用集中在 effect。
+  const visibleCardSymbols = (() => {
+    if (!virtualizeCards) return null
+    // 与 minuteSymbols 同口径: 剔除指数 (minute-batch 契约只收股票/ETF)
+    const scope = new Set(minuteSymbols as string[])
+    const items = cardRowVirtualizer.getVirtualItems()
+    const out: string[] = []
+    for (const item of items) {
+      for (let i = item.index * cardColumns; i < (item.index + 1) * cardColumns && i < sortedRows.length; i++) {
+        const s = (sortedRows[i] as any)?.symbol
+        if (typeof s === 'string' && scope.has(s)) out.push(s)
+      }
+    }
+    return out.length ? out : null
+  })()
+  const minuteVisibleKey = visibleCardSymbols?.join(',') ?? ''
+  const lastVisibleKeyRef = useRef<string | null>(null)
+  useEffect(() => {
+    minuteRequestSymbolsRef.current = visibleCardSymbols
+    if (!minuteVisibleKey) return
+    if (lastVisibleKeyRef.current === minuteVisibleKey) return
+    const prevKey = lastVisibleKeyRef.current
+    lastVisibleKeyRef.current = minuteVisibleKey
+    if (prevKey === null) return   // 首次: 挂载播种轮已按全列表发出, 不额外触发
+    // 视口集合变化 (滚动到新区段) → 防抖 300ms 补拉一次, 新滚入的 symbol 即时就绪
+    const t = setTimeout(() => {
+      qc.invalidateQueries({ queryKey: QK.minuteBatch(minuteSymbolsKey) })
+    }, 300)
+    return () => clearTimeout(t)
+    // eslint-disable-next-line react-hooks/exhaustive-deps
+  }, [minuteVisibleKey, minuteSymbolsKey])
+
   // 可见的 ext 列（卡片视图使用）
   const visibleExtCols = useMemo(
     () => visibleColumns.filter(c => c.source.type === 'ext'),