From 8d8b66c7f887a0bb3c01ee4c9f72433db178579d Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Wed, 9 Sep 2026 12:13:04 +0800 Subject: [PATCH] =?UTF-8?q?fix(kline):=20=E5=88=86=E9=92=9F=E6=89=B9?= =?UTF-8?q?=E9=87=8F=E8=A1=A5=E5=85=A8=E8=AF=86=E5=88=AB=E5=89=8D=E9=83=A8?= =?UTF-8?q?=E7=A9=BA=E6=B4=9E,=20=E9=87=8D=E5=90=AF=E8=B7=A8=E5=BC=80?= =?UTF-8?q?=E7=9B=98=E5=90=8E=E5=88=86=E6=97=B6=E5=8F=AF=E8=87=AA=E6=84=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 盘中重启/停机跨开盘后, 本地当日分钟K只剩重启时刻起的连续尾巴。 缺口判定只检查相邻K间隔, 连续尾巴被判"仅尾部落后"走增量 — 增量 锚定本地最新时间, 永远不会回看缺的开头, 上午的洞永久残留, 自选 分时只显示几根K。 - _has_holes 增加前部洞判定: 当日首根 > 开盘+6min 容差即视为洞, 归入全天重拉 (取到即落盘, 一次自愈)。仅根数不足的分支才判定, 稳态零额外请求; 停牌晚开票命中亦幂等无害 - 全量分钟服务健康时不再压制空洞类补拉 (服务增量补不了洞), 纯尾部落后的压制维持不变 实测: 服务器重启窗口后策略页(本地为空)走全天拉取显示完整、 自选页(有连续尾巴)被增量锁死只显示 11 根, 修复后统一自愈。 --- backend/app/api/kline.py | 20 +++-- backend/tests/test_minute_routing.py | 118 +++++++++++++++++++++++++++ 2 files changed, 132 insertions(+), 6 deletions(-) diff --git a/backend/app/api/kline.py b/backend/app/api/kline.py index ac934e7..442d5b2 100644 --- a/backend/app/api/kline.py +++ b/backend/app/api/kline.py @@ -741,12 +741,19 @@ def get_minute_batch(request: Request, body: dict): # 本地状态分类 (补拉已改为取到即落盘, 完整性判定随之收紧): # - fresh: 根数 >= 期望-2 (时间边界容差), 直接用本地。原 0.9 比例阈值会让 # 持久化数据在 90% 处冻结尾巴, 必须按根数差判。 - # - holes: 中间缺K (相邻间距非 1 分钟 / 非午休 91 分钟) → 全天重拉回填, - # 否则"最后一根+1min"的增量窗口永远不会回看中间的洞。 + # - holes: 缺K → 全天重拉回填, 否则"最后一根+1min"的增量窗口永远不会 + # 回看洞。含两种: 中间的洞 (相邻间距非 1 分钟 / 非午休 91 分钟) + # 与前部的洞 (首根显著晚于开盘 — 盘中重启/停机跨开盘的残留, + # 连续的尾部K会被增量锚定锁死, 同样必须全天重拉)。 # - stale: 仅尾部落后 → 增量拉, 请求量从"每轮全天"降为"每轮一根"量级。 _LUNCH_GAP_MIN = 91 # 11:30 → 13:01 + # 前部洞基准: 开盘后 6 分钟 (容许无集合竞价K的数据源)。晚开/停牌复牌的票 + # 也会命中 → 全天拉幂等, 至多多一次批量请求, 与中间洞同一代价模型。 + day_open_floor = datetime(trade_date.year, trade_date.month, trade_date.day, 9, 36, 0) def _has_holes(sub: pl.DataFrame) -> bool: + if not sub.is_empty() and sub["datetime"][0] > day_open_floor: + return True gaps = sub["datetime"].diff().dt.total_minutes().drop_nulls() return gaps.filter((gaps != 1) & (gaps != _LUNCH_GAP_MIN)).len() > 0 @@ -775,14 +782,15 @@ 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: - # 股票缺口不补拉, 本地有多少给多少 (服务下一轮写入补全); - # ETF 不在 universe 内, 维持补拉 - for sym in [*full_pull, *stale_last]: + # 纯尾部落后 (stale_last): 服务的增量轮下一轮就会补上, 股票不补拉省请求。 + # 空洞 (full_pull: 空分区 / 中间洞 / 前部洞): 服务增量锚定本地最新时间, + # 永远不会回看洞 → 不压制, 由端点全天拉取并落盘修复。 + # ETF 不在服务 universe 内, 两类均维持补拉。 + for sym in 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 语义; 下一轮命中本地, 请求量骤降)。 diff --git a/backend/tests/test_minute_routing.py b/backend/tests/test_minute_routing.py index e848e21..e4f88a2 100644 --- a/backend/tests/test_minute_routing.py +++ b/backend/tests/test_minute_routing.py @@ -1001,6 +1001,124 @@ def test_get_minute_batch_no_flag_unaffected_even_if_healthy(monkeypatch): assert result["full_minute_local"] is False +# ---------- 测试: 前部洞 (盘中重启/停机跨开盘的残留) ---------- + + +def _mock_tail_rows(symbol: str, n: int, first: datetime) -> pl.DataFrame: + """n 根从 first 开始的连续分钟K — 模拟重启后实时写入的尾部序列。""" + return pl.DataFrame({ + "symbol": [symbol] * n, + "datetime": [first + 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 test_get_minute_batch_leading_hole_triggers_full_day_refetch(monkeypatch): + """前部洞: 首根显著晚于开盘的连续尾部K → 全天重拉, 而非"最后一根+1min"增量。 + + 场景: 盘中重启/停机跨开盘后, 本地只剩 11:20 起的连续尾巴 (11 根)。 + 旧逻辑判"仅尾部落后"走增量, 上午的洞永远不会被回看; 新逻辑判洞 → 全天拉。 + """ + from app.api import kline as kline_api + + sync_spy = MagicMock(return_value=_mock_minute_rows("600519.SH", 121)) + 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_tail_rows( + "600519.SH", 11, datetime(2026, 1, 15, 11, 20) + ) + + 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) + + # 全天拉: start_time = 当日开盘窗口 (09:25), 不是"最后一根 + 1min" (11:31) + assert sync_spy.call_count == 1 + assert sync_spy.call_args.kwargs.get("start_time") == datetime(2026, 1, 15, 9, 25) + # 合并结果包含上午: 首根回到开盘附近, 根数覆盖全天 + rows = result["data"]["600519.SH"] + assert rows[0]["datetime"] == datetime(2026, 1, 15, 9, 31) + assert len(rows) >= 121 + + +def test_get_minute_batch_healthy_does_not_suppress_leading_hole_refetch(monkeypatch): + """服务健康 + prefer_local: 前部洞的股票仍全天补拉 (服务增量锚定本地最新, + 补不了洞); 纯尾部落后的股票维持不补拉 (服务下一轮会补尾巴)。""" + from app.api import kline as kline_api + + sync_spy = MagicMock(return_value=_mock_minute_rows("600519.SH", 121)) + monkeypatch.setattr(kline_api.kline_sync, "sync_minute_batch", sync_spy) + + hole_local = _mock_tail_rows("600519.SH", 11, datetime(2026, 1, 15, 11, 20)) + stale_local = _mock_minute_rows("000001.SZ", 100) # 09:31 开头的连续序列, 仅根数不足 + + mock_repo = MagicMock() + mock_repo.get_etf_symbol_set.return_value = set() + mock_repo.get_minute_batch.return_value = pl.concat([hole_local, stale_local]) + + 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", "000001.SZ"], "date": "2026-01-15", "prefer_local": True} + result = kline_api.get_minute_batch(mock_request, body) + + # 洞票被全天补拉; 尾部票未被拉 (只调了一次, 只为 600519) + assert sync_spy.call_count == 1 + assert sync_spy.call_args.args[0] == ["600519.SH"] + assert sync_spy.call_args.kwargs.get("start_time") == datetime(2026, 1, 15, 9, 25) + # 尾部票返回本地 100 根 (健康压制原样生效) + assert len(result["data"]["000001.SZ"]) == 100 + # 洞票拿到全天 + assert result["data"]["600519.SH"][0]["datetime"] == datetime(2026, 1, 15, 9, 31) + assert result["full_minute_local"] is True + + +def test_get_minute_batch_normal_open_not_treated_as_leading_hole(monkeypatch): + """无集合竞价K的源首根 09:31/09:35 → 不算前部洞, 维持增量语义 (不全天重拉)。""" + 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() + # 首根 09:31, 仅 5 根 (历史日 expected=240, 根数不足但非洞) + mock_repo.get_minute_batch.return_value = _mock_minute_rows("600519.SH", 5) + + 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} + kline_api.get_minute_batch(mock_request, body) + + # 增量拉: start_time = 最后一根本身 (09:35), 不是 09:25 全天窗口 + assert sync_spy.call_count == 1 + assert sync_spy.call_args.kwargs.get("start_time") == datetime(2026, 1, 15, 9, 35) + + def test_minute_refresh_is_healthy_requires_recent_round(monkeypatch): """is_healthy 三条件: 偏好开 + 线程活 + 最近一轮距现在 ≤ max(2×间隔, 30s)。""" import time as time_mod