From df5cae143e7bb94b5a09a72ccdab393a898434e7 Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Mon, 31 Aug 2026 17:10:02 +0800 Subject: [PATCH] =?UTF-8?q?feat(minute):=20=E8=87=AA=E9=80=89=E5=88=86?= =?UTF-8?q?=E6=97=B6=E6=8E=A5=E5=85=A5=E5=85=A8=E9=87=8F=E5=88=86=E9=92=9F?= =?UTF-8?q?=E6=9C=AC=E5=9C=B0=E4=BE=9B=E7=BB=99=20(prefer=5Flocal=20+=20fr?= =?UTF-8?q?eshness=20=E5=88=A4=E5=AE=9A)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - MinuteRefreshService.is_healthy(): 偏好开 + 线程活 + 最近一轮距今 ≤ max(2×间隔, 30s)。last_round_at 只在取数+落盘成功后更新, 连续失败 自动超时判不健康, 冷启动修复轮期间天然不健康 - /api/kline/minute-batch 加 prefer_local 标志: 服务健康时股票 incomplete 不再批量补拉, 本地有多少给多少 (服务下一轮落盘补全; 停牌票补拉也是空, 无损); ETF 不在全量分钟 universe 内维持补拉; 不健康回落现状兜底。 响应加 full_minute_local 标记 - 大自选 (数百上千只) 不再每 6s 轮询持续打批量分钟接口 - 4 个新用例: 健康跳过股票补拉/ETF 仍补, 不健康回落, 无标志不受影响, is_healthy 三条件边界 --- backend/app/api/kline.py | 21 +++- backend/app/services/minute_refresh.py | 20 ++++ backend/tests/test_minute_routing.py | 145 ++++++++++++++++++++++++- 3 files changed, 184 insertions(+), 2 deletions(-) diff --git a/backend/app/api/kline.py b/backend/app/api/kline.py index b3a61e6..7ecbb0f 100644 --- a/backend/app/api/kline.py +++ b/backend/app/api/kline.py @@ -582,6 +582,10 @@ def get_minute_batch(request: Request, body: dict): symbols: list[str] = body.get("symbols", []) trade_date_str: str | None = body.get("date") + # 自选分时本地优先标志: 全量分钟服务健康时, 股票缺口不再批量补拉 + # (本地分区由服务按间隔持续写入, 下一轮自然补全; 停牌/临停票补拉也是空, 无损)。 + # ETF 不在全量分钟 universe 内, 恒走补拉。服务不健康时回落现状补拉兜底。 + prefer_local = bool(body.get("prefer_local", False)) if not symbols: return {"data": {}} @@ -659,6 +663,20 @@ def get_minute_batch(request: Request, body: dict): elif not sub.is_empty(): result[sym] = sub.to_dicts() + # prefer_local 生效判定: 仅当全量分钟服务健康 (freshness 契约, 见 minute_refresh.is_healthy) + full_minute_healthy = False + if prefer_local: + 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) @@ -705,7 +723,8 @@ def get_minute_batch(request: Request, body: dict): if sub is not None and not sub.is_empty(): result[sym] = sub.to_dicts() - return {"data": result} + # full_minute_local: 本轮 prefer_local 生效 (本地分区由全量分钟服务供给, 股票未做补拉) + return {"data": result, "full_minute_local": full_minute_healthy} @router.get("/minute-range") diff --git a/backend/app/services/minute_refresh.py b/backend/app/services/minute_refresh.py index a26fe3e..cc809d8 100644 --- a/backend/app/services/minute_refresh.py +++ b/backend/app/services/minute_refresh.py @@ -295,6 +295,26 @@ class MinuteRefreshService: # 状态 # ------------------------------------------------------------------ + def is_healthy(self) -> bool: + """读侧 freshness 判定: 本地 kline_minute 分区是否正被服务持续写入。 + + 条件: 偏好开启 + 轮询线程存活 + 最近一轮距现在 ≤ max(2×间隔, 30s)。 + 健康时消费方 (自选分时等) 可信任本地分区、跳过批量补拉; + 连续失败轮不更新 last_round_at → 自动超时判不健康, 无需单独盯 last_error。 + """ + try: + if not preferences.get_minute_refresh_enabled(): + return False + except Exception: # noqa: BLE001 — 偏好文件异常按不健康处理 + return False + if self._thread is None or not self._thread.is_alive(): + return False + last = self._state.last_round_at + if last is None: + return False + interval = preferences.get_minute_refresh_interval() + return (time.time() - float(last)) <= max(2.0 * interval, 30.0) + def status(self) -> dict[str, Any]: import contextlib diff --git a/backend/tests/test_minute_routing.py b/backend/tests/test_minute_routing.py index 91516b3..b2483fe 100644 --- a/backend/tests/test_minute_routing.py +++ b/backend/tests/test_minute_routing.py @@ -11,7 +11,7 @@ mock 范式沿用 test_stocksdk_provider.py (monkeypatch 模块属性)。 """ from __future__ import annotations -from datetime import date, datetime +from datetime import date, datetime, timedelta from threading import Lock from unittest.mock import MagicMock @@ -735,3 +735,146 @@ def test_sync_minute_single_rejects_index_symbol(): asyncio.run(kline_api.sync_minute_single(mock_request, {"symbol": "000001.SH"})) assert exc_info.value.status_code == 400 assert "指数" in str(exc_info.value.detail) + + +# ---------- 测试: prefer_local (自选分时 × 全量分钟健康) ---------- + + +def _mock_minute_rows(symbol: str, n: int) -> pl.DataFrame: + """n 根分钟K (同日递增分钟), 用于构造'部分完整'的本地分区数据。""" + return pl.DataFrame({ + "symbol": [symbol] * n, + "datetime": [datetime(2026, 1, 15, 9, 31, 0) + timedelta(minutes=i) for i in range(n)], + "open": [100.0] * n, "high": [101.0] * n, "low": [99.5] * n, "close": [100.5] * n, + "volume": [1000.0] * n, "amount": [100500.0] * n, + }) + + +def _healthy_svc(monkeypatch, healthy: bool): + """mock app.state.minute_refresh (is_healthy 可控)。""" + svc = MagicMock() + svc.is_healthy.return_value = healthy + return svc + + +def test_get_minute_batch_prefer_local_healthy_skips_stock_refetch(monkeypatch): + """全量分钟服务健康 + prefer_local: 股票 incomplete 不再批量补拉, + 本地现有数据(哪怕 <90%)直接返回; ETF 不在服务 universe, 维持补拉。""" + from app.api import kline as kline_api + + sync_spy = MagicMock(return_value=_mock_minute_df(symbol="510300.SH")) + monkeypatch.setattr(kline_api.kline_sync, "sync_minute_batch", sync_spy) + + mock_repo = MagicMock() + mock_repo.get_etf_symbol_set.return_value = {"510300.SH"} + # 股票本地 100 根 (expected=240 → <90% 判 incomplete), ETF 本地空 + mock_repo.get_minute_batch.side_effect = ( + lambda syms, d, asset_type="stock": + _mock_minute_rows("600519.SH", 100) if asset_type == "stock" else pl.DataFrame() + ) + + 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 + mock_request.app.state.minute_refresh = _healthy_svc(monkeypatch, True) + + body = {"symbols": ["600519.SH", "510300.SH"], "date": "2026-01-15", "prefer_local": True} + result = kline_api.get_minute_batch(mock_request, body) + + # 股票未被补拉: sync_minute_batch 只为 ETF 调了一次 + assert sync_spy.call_count == 1 + assert sync_spy.call_args.kwargs.get("asset_type") == "etf" + assert sync_spy.call_args.args[0] == ["510300.SH"] + # 股票返回的是本地 100 根 (部分数据, 不因 incomplete 而缺失) + assert len(result["data"]["600519.SH"]) == 100 + assert result["full_minute_local"] is True + + +def test_get_minute_batch_prefer_local_unhealthy_falls_back(monkeypatch): + """服务不健康 (挂了/停了) + prefer_local: 回落现状补拉兜底, 行为与不带标志一致。""" + from app.api import kline as kline_api + + sync_spy = MagicMock(return_value=_mock_minute_df()) + monkeypatch.setattr(kline_api.kline_sync, "sync_minute_batch", sync_spy) + + mock_repo = MagicMock() + mock_repo.get_etf_symbol_set.return_value = set() + mock_repo.get_minute_batch.return_value = _mock_minute_rows("600519.SH", 100) + + 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 + mock_request.app.state.minute_refresh = _healthy_svc(monkeypatch, False) + + body = {"symbols": ["600519.SH"], "date": "2026-01-15", "prefer_local": True} + result = kline_api.get_minute_batch(mock_request, body) + + assert sync_spy.call_count == 1 # 股票照常补拉 + assert sync_spy.call_args.kwargs.get("asset_type") == "stock" + assert result["full_minute_local"] is False + + +def test_get_minute_batch_no_flag_unaffected_even_if_healthy(monkeypatch): + """不带 prefer_local (策略页等) 即使服务健康也维持现状补拉 — 本轮只辐射自选场景。""" + from app.api import kline as kline_api + + sync_spy = MagicMock(return_value=_mock_minute_df()) + monkeypatch.setattr(kline_api.kline_sync, "sync_minute_batch", sync_spy) + + mock_repo = MagicMock() + mock_repo.get_etf_symbol_set.return_value = set() + mock_repo.get_minute_batch.return_value = _mock_minute_rows("600519.SH", 100) + + 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 + mock_request.app.state.minute_refresh = _healthy_svc(monkeypatch, True) + + body = {"symbols": ["600519.SH"], "date": "2026-01-15"} + result = kline_api.get_minute_batch(mock_request, body) + + assert sync_spy.call_count == 1 + assert result["full_minute_local"] is False + + +def test_minute_refresh_is_healthy_requires_recent_round(monkeypatch): + """is_healthy 三条件: 偏好开 + 线程活 + 最近一轮距现在 ≤ max(2×间隔, 30s)。""" + import time as time_mod + from app.services import minute_refresh as mr + + svc = mr.MinuteRefreshService(MagicMock()) # is_healthy 不触达 repo + + monkeypatch.setattr(mr.preferences, "get_minute_refresh_enabled", lambda: True) + monkeypatch.setattr(mr.preferences, "get_minute_refresh_interval", lambda: 6) + # 线程未启动 → False + assert svc.is_healthy() is False + + svc._thread = MagicMock() + svc._thread.is_alive.return_value = True + # 无轮次记录 → False + assert svc.is_healthy() is False + + # 最近一轮在 10s 前 (≤ max(12, 30)) → True + svc._state.last_round_at = time_mod.time() - 10 + assert svc.is_healthy() is True + + # 最近一轮在 120s 前 (> 30) → False (连续失败不更新 last_round_at, 自动超时) + svc._state.last_round_at = time_mod.time() - 120 + assert svc.is_healthy() is False + + # 偏好关闭 → False + monkeypatch.setattr(mr.preferences, "get_minute_refresh_enabled", lambda: False) + svc._state.last_round_at = time_mod.time() - 10 + assert svc.is_healthy() is False