From b78ba9940d9627ce5713dc02ab0c1df973513b72 Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Mon, 31 Aug 2026 20:53:23 +0800 Subject: [PATCH] =?UTF-8?q?perf(kline):=20=E5=88=86=E9=92=9F=E6=89=B9?= =?UTF-8?q?=E9=87=8F=E8=A1=A5=E6=8B=89=E5=8F=96=E5=88=B0=E5=8D=B3=E8=90=BD?= =?UTF-8?q?=E7=9B=98=20+=20=E5=B0=BE=E9=83=A8=E5=A2=9E=E9=87=8F=E7=AA=97?= =?UTF-8?q?=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 补拉结果经 _write_minute_partition upsert 落盘 (持仓库写锁, 与全量分钟 服务/盘后同步同一纪律), 下一轮命中本地, 大自选从每轮全天重拉降为增量 - 完整性判定由 0.9 根数比例改为 期望-2 根: 比例阈值会让持久化数据在 90% 处冻结尾巴 - 分类三态: fresh 直读本地零请求 / 尾部落后从最后一根+1min 增量拉 / 中间有洞 (间距非 1min/午休 91min) 退回全天重拉回填, 防增量窗口漏数据 - 落盘失败降级为仅返回本轮数据 (持久化是优化而非正确性前提) --- backend/app/api/kline.py | 157 ++++++++++++++++----------- backend/tests/test_minute_routing.py | 97 +++++++++++++++++ 2 files changed, 192 insertions(+), 62 deletions(-) diff --git a/backend/app/api/kline.py b/backend/app/api/kline.py index 7ecbb0f..84bea75 100644 --- a/backend/app/api/kline.py +++ b/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} diff --git a/backend/tests/test_minute_routing.py b/backend/tests/test_minute_routing.py index b2483fe..4c38ee5 100644 --- a/backend/tests/test_minute_routing.py +++ b/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):