diff --git a/backend/app/api/kline.py b/backend/app/api/kline.py index c2e7b2e..9a5564e 100644 --- a/backend/app/api/kline.py +++ b/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") diff --git a/backend/tests/test_minute_routing.py b/backend/tests/test_minute_routing.py index 2d732b8..e5d635d 100644 --- a/backend/tests/test_minute_routing.py +++ b/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): diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 1d6194b..03dd8b8 100644 --- a/frontend/src/lib/api.ts +++ b/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; full_minute_local?: boolean }>('/api/kline/minute-batch', { + klineMinuteBatch: (symbols: string[], date?: string, preferLocal?: boolean, since?: string) => + request<{ data: Record; 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 }[] }>( diff --git a/frontend/src/lib/minuteBatchIncremental.ts b/frontend/src/lib/minuteBatchIncremental.ts new file mode 100644 index 0000000..f7330d4 --- /dev/null +++ b/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 + +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 ?? {}) } +} diff --git a/frontend/src/pages/Screener.tsx b/frontend/src/pages/Screener.tsx index 5f4d388..e2b05e5 100644 --- a/frontend/src/pages/Screener.tsx +++ b/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, diff --git a/frontend/src/pages/Watchlist.tsx b/frontend/src/pages/Watchlist.tsx index 57d5092..92ab058 100644 --- a/frontend/src/pages/Watchlist.tsx +++ b/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(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 = intradayVisible ? (minuteBatch.data?.data ?? {}) : {} + const liveRows = enriched.data?.rows + if (!intradayVisible || !liveRows?.length) return base + const liveBySymbol = new Map(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 = {} + 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(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'),