Merge pull request #237 from 0112020179/codex/incremental-detail-kline

perf(kline): 个股详情仅增量更新最后一根日 K
2026-09-12 16:44:15 +08:00 · 2026-09-04 11:26:55 +08:00
parent 4ca655279a 8721bd2e83
commit 2140209506
9 changed files with 263 additions and 44 deletions
@@ -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]
@@ -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
@@ -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<string[]>(['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
@@ -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({
@@ -215,7 +215,6 @@ export function StockPanel({
          onPriceDoubleClick={onPriceDoubleClick}
          visibleBars={showIntraday ? 40 : 60}
          extColumns={extColumns}
-          refetchIntervalMs={refetchIntervalMs}
        />

        {showIntraday && selectedDate && !intradayDismissed && (
@@ -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<string, unknown> }
+  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<CapabilitiesResponse>('/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<string, unknown> }
-      rows: KlineRow[]
-      source?: string
-    }>(
+    request<KlineDailyResponse>(
      (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<KlineDailyLatestResponse>(
+      `/api/kline/daily/latest?symbol=${encodeURIComponent(symbol)}`,
+    ),
  klineDailyBatch: (symbols: string[], days = 12) =>
    request<{ data: Record<string, KlineRow[]> }>('/api/kline/daily-batch', {
      method: 'POST',
@@ -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 为空 = 最新日内)。
 *
@@ -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,
@@ -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<KlineDailyResponse>(
+                  query.queryKey,
+                  current => mergeLatestKlineRow(current, latest),
+                )
+              }
+            })
+            .catch(() => {})
        }
      })