diff --git a/backend/app/api/kline.py b/backend/app/api/kline.py index eb4fb64..a54e3b4 100644 --- a/backend/app/api/kline.py +++ b/backend/app/api/kline.py @@ -459,51 +459,54 @@ def _attach_ext(resp: dict, repo, symbol: str, ext_columns: Optional[str]) -> di return resp -def _maybe_inject_live_candle(request: Request, symbol: str, rows: list[dict], asset_type: str = "stock") -> list[dict]: - """如果有当日实时 enriched 数据, 用实时数据生成今日蜡烛并追加/覆盖。 +def _latest_live_candle( + request: Request, + symbol: str, + asset_type: str = "stock", + *, + refresh_asset: bool = True, +) -> dict | None: + """从内存缓存读取单只标的的当日实时 enriched 行。""" - stock 走 QuoteService 的股票实时缓存; etf 走 ETF enriched 缓存 (开启实时 ETF - 拉取时为盘中数据, 否则为磁盘最新日, 由下方"非今日不注入"守卫自然跳过)。 - """ if asset_type == "stock": qs = getattr(request.app.state, "quote_service", None) if not qs: - return rows + return None df_today, enriched_date = qs.get_enriched_today() elif asset_type == "etf": - df_today, enriched_date = request.app.state.repo.get_enriched_latest_asset("etf") + df_today, enriched_date = request.app.state.repo.get_enriched_latest_asset( + "etf", refresh=refresh_asset, + ) else: - return rows + return None if df_today.is_empty(): - return rows + return None - # 非交易日（周末/假日）缓存的行情日期 != 今天，跳过注入避免产生重复蜡烛 + # 非交易日(周末/假日)缓存日期 != 今天, 跳过注入避免产生重复蜡烛 if not enriched_date or enriched_date != date.today(): - return rows + return None # 查找该 symbol 的实时 enriched 行 import polars as pl try: q = df_today.filter(pl.col("symbol") == symbol).to_dicts() if not q: - return rows + return None q = q[0] - except Exception: # noqa: BLE001 - return rows + except Exception: + return None close_price = q.get("close") if not close_price or close_price <= 0: - return rows + return None - today_str = str(enriched_date) - - # enriched 行已包含 OHLCV + 全套指标, 直接用它 - # 修复: API 在非交易时段可能返回 open/high/low=0, 用 close 填充避免异常蜡烛 + # 沿用完整日K接口原有的实时行投影, 避免增量接口形成第二套字段契约。 + # API 在非交易时段可能返回 open/high/low=0, 用 close 填充避免异常蜡烛。 raw_open = q.get("open") raw_high = q.get("high") raw_low = q.get("low") - live_row: dict = { - "date": today_str, + live_row = { + "date": str(enriched_date), "symbol": symbol, "open": raw_open if raw_open and raw_open > 0 else close_price, "high": raw_high if raw_high and raw_high > 0 else close_price, @@ -514,7 +517,6 @@ def _maybe_inject_live_candle(request: Request, symbol: str, rows: list[dict], a "change_pct": q.get("change_pct"), "is_live": True, } - # 补上 enriched 的技术指标字段 for key in ("ma5", "ma10", "ma20", "ma30", "ma60", "macd_dif", "macd_dea", "macd_hist", "kdj_k", "kdj_d", "kdj_j", @@ -523,11 +525,19 @@ def _maybe_inject_live_candle(request: Request, symbol: str, rows: list[dict], a "atr_14", "vol_ratio_5d"): if key in q and q[key] is not None: live_row[key] = q[key] + return live_row + + +def _maybe_inject_live_candle(request: Request, symbol: str, rows: list[dict], asset_type: str = "stock") -> list[dict]: + """如果有当日实时 enriched 数据, 用实时数据生成今日蜡烛并追加/覆盖。""" + live_row = _latest_live_candle(request, symbol, asset_type) + if live_row is None: + return rows # 如果已有今天的 enriched 行, 覆盖; 否则追加 found = False - for i, r in enumerate(rows): - if str(r.get("date")) == today_str: + for r in rows: + if str(r.get("date")) == live_row["date"]: r.update(live_row) found = True break @@ -538,6 +548,22 @@ def _maybe_inject_live_candle(request: Request, symbol: str, rows: list[dict], a return rows +@router.get("/daily/latest") +def get_daily_latest( + request: Request, + symbol: str = Query(..., description="标的代码,如 000001.SZ"), +): + """返回内存中的当日单行 K 线, 供详情页实时增量更新。""" + repo = request.app.state.repo + asset_type = repo.resolve_asset_type(symbol) + row = _latest_live_candle(request, symbol, asset_type, refresh_asset=False) + return { + "symbol": symbol, + "row": row, + "source": "live" if row is not None else "none", + } + + class DailyBatchRequest: """批量日K请求。""" symbols: list[str] diff --git a/backend/tests/test_kline_daily_latest.py b/backend/tests/test_kline_daily_latest.py new file mode 100644 index 0000000..1837dd0 --- /dev/null +++ b/backend/tests/test_kline_daily_latest.py @@ -0,0 +1,140 @@ +"""个股详情日 K 最新行接口测试。""" +from __future__ import annotations + +from datetime import date, timedelta + +import polars as pl +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from app.api.kline import router + + +class _FakeQuoteService: + def __init__(self, frame: pl.DataFrame, trade_date: date | None) -> None: + self._frame = frame + self._trade_date = trade_date + + def get_enriched_today(self): + return self._frame, self._trade_date + + +class _FakeRepo: + def __init__( + self, + asset_type: str = "stock", + latest_asset: tuple[pl.DataFrame, date | None] | None = None, + ) -> None: + self.asset_type = asset_type + self.latest_asset = latest_asset + self.latest_asset_calls = 0 + self.latest_asset_refresh: bool | None = None + + def resolve_asset_type(self, symbol: str) -> str: + return self.asset_type + + def get_enriched_latest_asset(self, asset_type: str, refresh: bool = True): + self.latest_asset_calls += 1 + self.latest_asset_refresh = refresh + if self.latest_asset is not None: + return self.latest_asset + raise AssertionError("stock latest row must use QuoteService's memory cache") + + +def _client( + frame: pl.DataFrame, + trade_date: date | None, + *, + asset_type: str = "stock", + latest_asset: tuple[pl.DataFrame, date | None] | None = None, +) -> tuple[TestClient, _FakeRepo]: + repo = _FakeRepo(asset_type, latest_asset) + app = FastAPI() + app.include_router(router) + app.state.repo = repo + app.state.quote_service = _FakeQuoteService(frame, trade_date) + return TestClient(app), repo + + +def _live_frame() -> pl.DataFrame: + return pl.DataFrame({ + "symbol": ["600000.SH"], + "date": [date.today()], + "open": [10.0], + "high": [10.8], + "low": [9.9], + "close": [10.6], + "volume": [123_456.0], + "amount": [1_234_560.0], + "ma5": [10.2], + "signal_limit_up": [False], + }) + + +def test_daily_latest_returns_only_current_memory_row() -> None: + client, repo = _client(_live_frame(), date.today()) + + response = client.get("/api/kline/daily/latest", params={"symbol": "600000.SH"}) + + assert response.status_code == 200 + body = response.json() + assert body["symbol"] == "600000.SH" + assert body["source"] == "live" + assert body["row"] == { + "symbol": "600000.SH", + "date": date.today().isoformat(), + "open": 10.0, + "high": 10.8, + "low": 9.9, + "close": 10.6, + "volume": 123_456.0, + "amount": 1_234_560.0, + "change_pct": None, + "ma5": 10.2, + "is_live": True, + } + assert repo.latest_asset_calls == 0 + + +def test_daily_latest_returns_none_for_stale_cache() -> None: + client, _ = _client(_live_frame(), date.today() - timedelta(days=1)) + + response = client.get("/api/kline/daily/latest", params={"symbol": "600000.SH"}) + + assert response.status_code == 200 + assert response.json() == { + "symbol": "600000.SH", + "row": None, + "source": "none", + } + + +def test_daily_latest_returns_none_when_symbol_is_missing() -> None: + client, _ = _client(_live_frame(), date.today()) + + response = client.get("/api/kline/daily/latest", params={"symbol": "600001.SH"}) + + assert response.status_code == 200 + assert response.json() == { + "symbol": "600001.SH", + "row": None, + "source": "none", + } + + +def test_daily_latest_uses_etf_enriched_cache() -> None: + etf = _live_frame().with_columns(pl.lit("510300.SH").alias("symbol")) + client, repo = _client( + pl.DataFrame(), + None, + asset_type="etf", + latest_asset=(etf, date.today()), + ) + + response = client.get("/api/kline/daily/latest", params={"symbol": "510300.SH"}) + + assert response.status_code == 200 + assert response.json()["row"]["close"] == 10.6 + assert response.json()["source"] == "live" + assert repo.latest_asset_calls == 1 + assert repo.latest_asset_refresh is False diff --git a/frontend/src/components/StockDailyKChart.tsx b/frontend/src/components/StockDailyKChart.tsx index 142a1a5..8e0950e 100644 --- a/frontend/src/components/StockDailyKChart.tsx +++ b/frontend/src/components/StockDailyKChart.tsx @@ -44,8 +44,6 @@ interface Props { onPriceDoubleClick?: (price: number, currentPrice: number) => void /** 扩展数据列参数（逗号分隔 config_id.field_name），透传给 klineDaily 接口 */ extColumns?: string - /** 日K自动刷新间隔(ms)。undefined = 不轮询(默认)。个股对话框实时刷新时传入, 盘中今日蜡烛随之更新 */ - refetchIntervalMs?: number } function isValidRow(r: any): boolean { @@ -121,7 +119,6 @@ export function StockDailyKChart({ onDateClick, onPriceDoubleClick, extColumns, - refetchIntervalMs, }: Props) { const [activeIndicators, setActiveIndicators] = useState(['vol']) const [showMarkers, setShowMarkers] = useState(true) @@ -131,7 +128,7 @@ export function StockDailyKChart({ const dateRange = externalDateRange ?? getDefaultRange() // 查询配置统一来自 klineDailyQueryOptions, 与 StockPanel 信息条/邻近预取共享同一 cache key (只发一次请求) - const kline = useQuery({ ...klineDailyQueryOptions(symbol, dateRange, extColumns), enabled: !!symbol, refetchInterval: refetchIntervalMs }) + const kline = useQuery({ ...klineDailyQueryOptions(symbol, dateRange, extColumns), enabled: !!symbol }) const rows = useMemo(() => toOHLC(kline.data?.rows ?? []), [kline.data?.rows]) const stockInfo = kline.data?.stock_info diff --git a/frontend/src/components/StockIntradayChart.tsx b/frontend/src/components/StockIntradayChart.tsx index 7791be8..ba6aed9 100644 --- a/frontend/src/components/StockIntradayChart.tsx +++ b/frontend/src/components/StockIntradayChart.tsx @@ -40,7 +40,7 @@ export function StockIntradayChart({ // 避免读到分钟增量落盘的上一轮本地分区; 历史日期后端自行忽略 live。 ...klineMinuteQueryOptions(symbol, date ?? undefined, refetchIntervalMs != null), enabled: !!symbol && !!date, - refetchInterval: refetchIntervalMs, + refetchInterval: query => query.state.data?.source === 'none' ? false : refetchIntervalMs, }) const fetchMinute = useMutation({ diff --git a/frontend/src/components/StockPanel.tsx b/frontend/src/components/StockPanel.tsx index 160b1e1..a2a0bb8 100644 --- a/frontend/src/components/StockPanel.tsx +++ b/frontend/src/components/StockPanel.tsx @@ -215,7 +215,6 @@ export function StockPanel({ onPriceDoubleClick={onPriceDoubleClick} visibleBars={showIntraday ? 40 : 60} extColumns={extColumns} - refetchIntervalMs={refetchIntervalMs} /> {showIntraday && selectedDate && !intradayDismissed && ( diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 593b9e9..e09af3d 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -234,6 +234,20 @@ export interface KlineRow { [key: string]: any } +export interface KlineDailyResponse { + symbol: string + name?: string + stock_info?: { name?: string; total_shares?: number; float_shares?: number; ext?: Record } + rows: KlineRow[] + source?: string +} + +export interface KlineDailyLatestResponse { + symbol: string + row: KlineRow | null + source: 'live' | 'none' +} + // ===== Watchlist ===== export interface WatchlistEntry { symbol: string @@ -2082,18 +2096,16 @@ export const api = { request('/api/capabilities/redetect', { method: 'POST' }), klineDaily: (symbol: string, days = 120, dateRange?: { start: string; end: string }, extColumns?: string) => - request<{ - symbol: string - name?: string - stock_info?: { name?: string; total_shares?: number; float_shares?: number; ext?: Record } - rows: KlineRow[] - source?: string - }>( + request( (dateRange ? `/api/kline/daily?symbol=${encodeURIComponent(symbol)}&start_date=${dateRange.start}&end_date=${dateRange.end}` : `/api/kline/daily?symbol=${encodeURIComponent(symbol)}&days=${days}`) + (extColumns ? `&ext_columns=${encodeURIComponent(extColumns)}` : ''), ), + klineDailyLatest: (symbol: string) => + request( + `/api/kline/daily/latest?symbol=${encodeURIComponent(symbol)}`, + ), klineDailyBatch: (symbols: string[], days = 12) => request<{ data: Record }>('/api/kline/daily-batch', { method: 'POST', diff --git a/frontend/src/lib/kline.ts b/frontend/src/lib/kline.ts index 37e2327..6bc7ac1 100644 --- a/frontend/src/lib/kline.ts +++ b/frontend/src/lib/kline.ts @@ -7,7 +7,7 @@ * placeholderData 内置"仅同 symbol 占位"守卫: 改日期范围/扩展字段时旧数据可暂显(不闪), * 切股时不透传上一只股票的数据(不误显示)。 */ -import { api } from '@/lib/api' +import { api, type KlineDailyLatestResponse, type KlineDailyResponse } from '@/lib/api' import { QK } from '@/lib/queryKeys' /** 分时 tab 多日分时默认周期 (StockPanel 预取与弹窗存储回退共用, 避免魔数两处漂移) */ @@ -29,6 +29,34 @@ export function klineDailyQueryOptions( } } +export function klineDailyLatestQueryOptions(symbol: string) { + return { + queryKey: QK.klineLatest(symbol), + queryFn: () => api.klineDailyLatest(symbol), + } +} + +export function mergeLatestKlineRow( + current: KlineDailyResponse | undefined, + latest: KlineDailyLatestResponse, +): KlineDailyResponse | undefined { + if (!current || !latest.row || current.symbol !== latest.symbol) return current + + const latestDate = String(latest.row.date).slice(0, 10) + const last = current.rows.at(-1) + if (!last) return { ...current, rows: [{ ...latest.row, date: latestDate }] } + + const lastDate = String(last.date).slice(0, 10) + if (latestDate < lastDate) return current + if (latestDate === lastDate) { + return { + ...current, + rows: [...current.rows.slice(0, -1), { ...last, ...latest.row, date: latestDate }], + } + } + return { ...current, rows: [...current.rows, { ...latest.row, date: latestDate }] } +} + /** * 单日分时查询配置 — 与 klineDailyQueryOptions 同风格的单源 options (date 为空 = 最新日内)。 * diff --git a/frontend/src/lib/queryKeys.ts b/frontend/src/lib/queryKeys.ts index 74e4bcc..f43fb42 100644 --- a/frontend/src/lib/queryKeys.ts +++ b/frontend/src/lib/queryKeys.ts @@ -79,6 +79,7 @@ export const QK = { // Kline kline: (symbol: string, start: string, end: string, extColumns?: string) => ['kline', symbol, start, end, extColumns ?? ''] as const, + klineLatest: (symbol: string) => ['kline-latest', symbol] as const, stockLevels: (symbol: string, days?: number) => ['stock-levels', symbol, days ?? 120] as const, klineMinute: (symbol: string, date: string) => ['kline-minute', symbol, date] as const, diff --git a/frontend/src/lib/useQuoteStream.ts b/frontend/src/lib/useQuoteStream.ts index 8eb5432..3840b01 100644 --- a/frontend/src/lib/useQuoteStream.ts +++ b/frontend/src/lib/useQuoteStream.ts @@ -1,11 +1,12 @@ import { useEffect, useRef, useCallback, useSyncExternalStore } from 'react' import { useQueryClient } from '@tanstack/react-query' import { SSE_INVALIDATE_PREFIXES, QK } from './queryKeys' +import { klineDailyLatestQueryOptions, mergeLatestKlineRow } from './kline' import { getQueryConfig } from './useQueryConfig' import { toast } from '@/components/Toast' import { pushAlertToasts } from '@/components/AlertToast' import { feedReviewEvent } from './reviewStore' -import type { StrategyAlertEvent } from './api' +import type { KlineDailyResponse, StrategyAlertEvent } from './api' // ===== 全局 SSE 连接状态 (模块级 store, 仿 AlertToast.tsx 模式) ===== // 实时行情 SSE 断开时 UI 无感知 → 会漏掉策略告警。这里暴露连接状态, @@ -47,8 +48,8 @@ export function useQuoteStreamStatus(): QuoteStreamStatus { } // ===== 焦点股票注册表 (个股对话框用) ===== -// 个股对话框打开时注册当前 symbol, SSE quotes_updated 推送时精准 invalidate -// 该 symbol 的日K查询 (['kline', symbol]), 让日K最后一根蜡烛随实时价变化。 +// 个股对话框打开时注册当前 symbol, SSE quotes_updated 推送时只取当日最新行， +// 再原位更新该 symbol 的日K查询缓存，避免重复下载整段历史。 // 不加进 SSE_INVALIDATE_PREFIXES 全局列表 —— 避免回测弹窗等也每秒重拉。 let _focusSymbol: string | null = null @@ -161,10 +162,25 @@ export function useQuoteStream( ), }) } - // 焦点股票日K精准刷新: 个股对话框打开时, 日K最后一根蜡烛随实时价变化。 - // 后端 _maybe_inject_live_candle 只读内存缓存, 不调 TickFlow, 秒级重拉零额外成本。 + // 焦点股票日K增量刷新: 只取内存中的当日行并合并缓存尾部。 if (_focusSymbol) { - qc.invalidateQueries({ queryKey: ['kline', _focusSymbol] }) + const symbol = _focusSymbol + void qc.fetchQuery({ ...klineDailyLatestQueryOptions(symbol), staleTime: 0 }) + .then((latest) => { + if (_focusSymbol !== symbol || !latest.row) return + const latestDate = String(latest.row.date).slice(0, 10) + const queries = qc.getQueryCache().findAll({ queryKey: ['kline', symbol] }) + for (const query of queries) { + const start = String(query.queryKey[2] ?? '') + const end = String(query.queryKey[3] ?? '') + if (latestDate < start || latestDate > end) continue + qc.setQueryData( + query.queryKey, + current => mergeLatestKlineRow(current, latest), + ) + } + }) + .catch(() => {}) } })