From e2ffde99d57d58d99e754f8fc926e89b7085a039 Mon Sep 17 00:00:00 2001 From: intfoo Date: Wed, 15 Jul 2026 16:34:23 +0800 Subject: [PATCH 1/4] =?UTF-8?q?fix(minute,monitoring):=20=E8=87=AA?= =?UTF-8?q?=E5=AE=9A=E4=B9=89=E6=95=B0=E6=8D=AE=E6=BA=90=E7=94=A8=E6=88=B7?= =?UTF-8?q?=E5=8D=95=E8=82=A1=E8=A1=A5=E6=8B=89/=E7=9B=91=E6=8E=A7?= =?UTF-8?q?=E9=A1=B5=E6=89=93=E9=80=9A=20(#121)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 承接 #126: 能力探测已补 KLINE_MINUTE_BATCH, 但两处取数/UI 仍按 TickFlow 档位拦截, 配了自定义分钟/实时源的 None 档用户依旧被挡。 - kline_sync.fetch_minute_single: 与 sync_minute_batch 一致, 先查 preferences.get_minute_data_provider(), 非 tickflow 且自定义源有 minute dataset 时走 custom provider, 避免分时图首次打开(本地无数据) 补拉返回空。 - Monitoring.tsx: None 档但配了自定义实时源时, 后端 is_realtime_allowed(realtime_mode=full_market) 仍返回 True。 以 quoteStatus.realtime_allowed 作为最终判据, 不再用 isNoneTier 一刀切拦截实时监控页。 Co-authored-by: shy3130 --- backend/app/services/kline_sync.py | 13 ++++++++++++- frontend/src/pages/settings/Monitoring.tsx | 5 ++++- 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/backend/app/services/kline_sync.py b/backend/app/services/kline_sync.py index 530e088..e2ba0db 100644 --- a/backend/app/services/kline_sync.py +++ b/backend/app/services/kline_sync.py @@ -738,10 +738,21 @@ def fetch_intraday_monitor_batch( def fetch_minute_single(symbol: str, trade_date: date) -> pl.DataFrame: - """从 TickFlow 实时拉取单股单日分钟 K（不写入本地）。""" + """实时拉取单股单日分钟 K(不写入本地)。优先自定义分钟源, 回退 TickFlow。""" from datetime import datetime 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) + + # 自定义数据源分流: 与 sync_minute_batch 一致, 配了自定义分钟源时走 custom provider, + # 避免无 TickFlow Pro+ 权限的用户分时图首次打开(本地无数据)时补拉失败返回空。 + provider_name = preferences.get_minute_data_provider() + if provider_name != "tickflow": + from app.data_providers import custom as custom_sources + if custom_sources.provider_has_dataset(provider_name, "minute"): + provider = custom_sources.get_provider(provider_name) + return provider.get_minute([symbol], start_time=start_time, end_time=end_time) + # 未配置 minute dataset → 回退 TickFlow + tf = get_client() try: raw = tf.klines.batch( diff --git a/frontend/src/pages/settings/Monitoring.tsx b/frontend/src/pages/settings/Monitoring.tsx index 6fc59ba..9b085f6 100644 --- a/frontend/src/pages/settings/Monitoring.tsx +++ b/frontend/src/pages/settings/Monitoring.tsx @@ -49,6 +49,9 @@ export function SettingsMonitoringPanel({ highlight }: { highlight?: string } = const toggleQuote = useToggleRealtimeQuotes() const tier = tierRank(caps?.label ?? '') const isNoneTier = tier < 0 + // None 档但配了自定义实时源时, 后端 is_realtime_allowed 仍返回 True (realtime_mode=full_market) + // 此时不应拦截实时监控页 — 用 quoteStatus.realtime_allowed 作为最终判据 + const realtimeAllowed = quoteStatus?.realtime_allowed ?? !isNoneTier const isFreeTier = tier === 0 const realtimeEnabled = prefs?.realtime_quotes_enabled ?? false // 分时图实时刷新间隔 (秒), 与后端 [3,60] clamp 对齐; 默认 6 @@ -278,7 +281,7 @@ export function SettingsMonitoringPanel({ highlight }: { highlight?: string } = } }, [highlight]) - if (isNoneTier) { + if (isNoneTier && !realtimeAllowed) { return (
pl.DataFrame: - """Return normalized minute K rows. Implementations may return empty.""" + """Return normalized minute K rows. Implementations may return empty. + + on_chunk_done 契约: provider 实现内部以 2 参 (cur, total) 调用; + 3 参 seg_label 适配由 kline_sync._try_custom_minute 包装层负责, + 不应泄漏到 provider 契约层。 + """ def get_realtime( self, diff --git a/backend/app/data_providers/custom/provider.py b/backend/app/data_providers/custom/provider.py index 53b258e..1f80e91 100644 --- a/backend/app/data_providers/custom/provider.py +++ b/backend/app/data_providers/custom/provider.py @@ -3,6 +3,7 @@ from __future__ import annotations import logging import os +from collections.abc import Callable from datetime import datetime from pathlib import Path from typing import Any @@ -11,6 +12,7 @@ import httpx import polars as pl from app.config import settings +from app.data_providers.base import AssetType from app.data_providers.custom.config import CustomSourceConfig, DatasetConfig from app.data_providers.custom.mapper import apply_transforms, datetime_payload, extract_rows, map_rows from app.data_providers.normalizer import normalize_adj_factors, normalize_daily @@ -109,8 +111,9 @@ class GenericHTTPProvider: symbols: list[str], start_time: datetime | None, end_time: datetime | None, - asset_type: str = "stock", # noqa: ARG002 - on_chunk_done=None, + asset_type: AssetType = "stock", # noqa: ARG002 + freq: str = "1m", # noqa: ARG002 + on_chunk_done: Callable[[int, int], None] | None = None, ) -> pl.DataFrame: cfg = self._dataset("minute") frames: list[pl.DataFrame] = [] diff --git a/backend/app/data_providers/tickflow_provider.py b/backend/app/data_providers/tickflow_provider.py index a086949..c69bd73 100644 --- a/backend/app/data_providers/tickflow_provider.py +++ b/backend/app/data_providers/tickflow_provider.py @@ -2,6 +2,7 @@ from __future__ import annotations import logging +from collections.abc import Callable from datetime import datetime import polars as pl @@ -97,8 +98,9 @@ class TickFlowProvider: symbols: list[str], start_time: datetime | None, end_time: datetime | None, - asset_type: AssetType, # noqa: ARG002 + asset_type: AssetType = "stock", # noqa: ARG002 freq: str = "1m", # noqa: ARG002 + on_chunk_done: Callable[[int, int], None] | None = None, # noqa: ARG002 ) -> pl.DataFrame: # Existing minute sync remains in app.services.kline_sync for now. return pl.DataFrame() diff --git a/backend/app/plugins/stocksdk/provider.py b/backend/app/plugins/stocksdk/provider.py index 6922055..c3481a4 100644 --- a/backend/app/plugins/stocksdk/provider.py +++ b/backend/app/plugins/stocksdk/provider.py @@ -10,11 +10,13 @@ Original implementation by @forrany (PR #57), migrated to plugin architecture. from __future__ import annotations import logging +from collections.abc import Callable from dataclasses import dataclass, field from datetime import datetime import polars as pl +from app.data_providers.base import AssetType from app.data_providers.normalizer import normalize_adj_factors, normalize_daily from app.plugins.stocksdk import bridge from app.tickflow.rate_limits import chunked @@ -135,13 +137,13 @@ class StockSDKProvider: symbols: list[str], start_time: datetime | None, end_time: datetime | None, - asset_type: str = "stock", # noqa: ARG002 - on_chunk_done=None, - freq: str = "5m", + asset_type: AssetType = "stock", # noqa: ARG002 + freq: str = "1m", + on_chunk_done: Callable[[int, int], None] | None = None, ) -> pl.DataFrame: if not symbols: return pl.DataFrame() - period = "".join(ch for ch in str(freq) if ch.isdigit()) or "5" + period = "".join(ch for ch in str(freq) if ch.isdigit()) or "1" logger.info("stock-sdk minute 拉取开始(%d symbols, period=%s)", len(symbols), period) frames: list[pl.DataFrame] = [] chunks = chunked(symbols, _BATCH) diff --git a/backend/app/services/kline_sync.py b/backend/app/services/kline_sync.py index e2ba0db..2489718 100644 --- a/backend/app/services/kline_sync.py +++ b/backend/app/services/kline_sync.py @@ -9,10 +9,11 @@ from __future__ import annotations import logging from collections.abc import Callable -from datetime import datetime, timedelta +from datetime import date, datetime, timedelta import polars as pl +from app.data_providers.base import AssetType from app.indicators.pipeline import filter_halt_days from app.market_time import cn_now from app.services import preferences @@ -528,6 +529,56 @@ def _write_minute_partition(df: pl.DataFrame, minute_dir) -> int: return written +def _try_custom_minute( + symbols: list[str], + start_time: datetime | None, + end_time: datetime | None, + asset_type: AssetType, + freq: str = "1m", + on_chunk_done: Callable[[int, int, str], None] | None = None, +) -> tuple[pl.DataFrame | None, bool]: + """尝试从自定义分钟源拉取。返回 (df, should_fallback_to_tickflow)。 + + 返回契约: + (None, True) → 未配自定义源 / 未配 minute dataset / 自定义源异常 → 走 TickFlow + (df, False) → 自定义源成功(含空 df) → 直接用, 不回退 + + 降级策略 (C): 自定义源异常时无条件 fall through 到 TickFlow, + 由 TickFlow 路径自身 try/except 兜底。Pro+ 用户 TickFlow 成功返回数据, + None 档用户 TickFlow 失败返回空。不显式判断 tier, 避免 #126 augmented + capability 逻辑干扰。 + + on_chunk_done 适配: 上层回调是 3 参 (cur, total, seg_label), provider + 实现内部以 2 参 (cur, total) 调用。这里包装一层, provider 调 2 参时补 + 默认 seg_label="custom" 转发给上层, 保证进度展示不降级。 + """ + provider_name = preferences.get_minute_data_provider() + if provider_name == "tickflow": + return (None, True) + from app.data_providers import custom as custom_sources + if not custom_sources.provider_has_dataset(provider_name, "minute"): + return (None, True) + provider = custom_sources.get_provider(provider_name) + + # 包装 on_chunk_done: provider 调 2 参 → 补 seg_label="custom" → 转发上层 3 参 + wrapped_cb: Callable[[int, int], None] | None = None + if on_chunk_done is not None: + def _wrapped_cb(cur: int, total: int) -> None: + on_chunk_done(cur, total, "custom") + wrapped_cb = _wrapped_cb + + try: + df = provider.get_minute( + symbols, start_time=start_time, end_time=end_time, + asset_type=asset_type, freq=freq, on_chunk_done=wrapped_cb, + ) + return (df, False) + except Exception as e: # noqa: BLE001 + logger.warning("custom minute provider %s failed, falling back to TickFlow: %s", + provider_name, e) + return (None, True) + + def sync_minute_batch( symbols: list[str], start_time: datetime | None = None, @@ -538,6 +589,7 @@ def sync_minute_batch( on_chunk_done: Callable[[int, int, str], None] | None = None, segment_trading_days: int = 20, on_segment: Callable[[pl.DataFrame], None] | None = None, + asset_type: AssetType = "stock", ) -> pl.DataFrame: """批量拉取多股分钟 K。 @@ -555,16 +607,15 @@ def sync_minute_batch( 不进入全局 out → 内存峰值从「全量」降到「单段」。适用于 sync_and_persist_minute。 不传时 (如 get_minute_batch 的实时补拉) 保持原契约: 累积进 out 末尾一次性返回。 """ - # 自定义数据源分流: minute provider - provider_name = preferences.get_minute_data_provider() - if provider_name != "tickflow": - from app.data_providers import custom as custom_sources - if custom_sources.provider_has_dataset(provider_name, "minute"): - provider = custom_sources.get_provider(provider_name) - return provider.get_minute( - symbols, start_time=start_time, end_time=end_time, on_chunk_done=on_chunk_done, - ) - # 未配置 minute → 回退 TickFlow + df, fallback = _try_custom_minute( + symbols, start_time=start_time, end_time=end_time, + asset_type=asset_type, freq="1m", on_chunk_done=on_chunk_done, + ) + if not fallback: + # _try_custom_minute 成功时返回 (df, False), df 非 None; + # fallback=True 时才返回 (None, True)。故此处 df 必非 None。 + # (旧版 `df or pl.DataFrame()` 触发 polars DataFrame __bool__ TypeError, 已修。) + return df if df is not None else pl.DataFrame() tf = get_client() @@ -572,7 +623,7 @@ def sync_minute_batch( # 按 segment_trading_days 交易日分段 (交易日→自然日 ×7/5 换算, 含节假日余量)。 seg_calendar_days = max(1, int(segment_trading_days * 7 / 5)) SEG_CHUNK = timedelta(days=seg_calendar_days) - time_segments: list[tuple[datetime, datetime]] = [] + time_segments: list[tuple[datetime | None, datetime | None]] = [] if start_time and end_time: seg_start = start_time while seg_start < end_time: @@ -589,10 +640,10 @@ def sync_minute_batch( # 段内累积: 每段拉完即 flush, 避免全量攒内存 (OOM 根因) seg_out: list[pl.DataFrame] = [] - for seg_idx, (seg_start, seg_end) in enumerate(time_segments): + for seg_idx, (cur_start, cur_end) in enumerate(time_segments): # 当前的日期段描述 (供进度展示) - if seg_start and seg_end: - seg_label = f"{seg_start.strftime('%m-%d')}~{seg_end.strftime('%m-%d')}" + if cur_start and cur_end: + seg_label = f"{cur_start.strftime('%m-%d')}~{cur_end.strftime('%m-%d')}" else: seg_label = "最新" seg_total = len(time_segments) @@ -601,11 +652,11 @@ def sync_minute_batch( sleep_between_batches(step, rpm) step += 1 try: - if seg_start and seg_end: + if cur_start and cur_end: raw = tf.klines.batch( chunk, period="1m", - start_time=_datetime_to_ms(seg_start), - end_time=_datetime_to_ms(seg_end), + start_time=_datetime_to_ms(cur_start), + end_time=_datetime_to_ms(cur_end), count=10000, adjust="forward", as_dataframe=True, show_progress=False, @@ -737,7 +788,11 @@ def fetch_intraday_monitor_batch( return pl.concat(frames, how="diagonal_relaxed") if frames else pl.DataFrame() -def fetch_minute_single(symbol: str, trade_date: date) -> pl.DataFrame: +def fetch_minute_single( + symbol: str, + trade_date: date, + asset_type: AssetType = "stock", +) -> pl.DataFrame: """实时拉取单股单日分钟 K(不写入本地)。优先自定义分钟源, 回退 TickFlow。""" from datetime import datetime start_time = datetime(trade_date.year, trade_date.month, trade_date.day, 9, 25, 0) @@ -745,13 +800,13 @@ def fetch_minute_single(symbol: str, trade_date: date) -> pl.DataFrame: # 自定义数据源分流: 与 sync_minute_batch 一致, 配了自定义分钟源时走 custom provider, # 避免无 TickFlow Pro+ 权限的用户分时图首次打开(本地无数据)时补拉失败返回空。 - provider_name = preferences.get_minute_data_provider() - if provider_name != "tickflow": - from app.data_providers import custom as custom_sources - if custom_sources.provider_has_dataset(provider_name, "minute"): - provider = custom_sources.get_provider(provider_name) - return provider.get_minute([symbol], start_time=start_time, end_time=end_time) - # 未配置 minute dataset → 回退 TickFlow + df, fallback = _try_custom_minute( + [symbol], start_time=start_time, end_time=end_time, + asset_type=asset_type, freq="1m", + ) + if not fallback: + # 见 sync_minute_batch 同分支注释: df 在此必非 None。 + return df if df is not None else pl.DataFrame() tf = get_client() try: @@ -975,6 +1030,7 @@ def sync_and_persist_minute( on_chunk_done=on_chunk_done, segment_trading_days=segment_days, on_segment=_persist, + asset_type="stock", ) if written_box[0] == 0: diff --git a/backend/tests/test_minute_routing.py b/backend/tests/test_minute_routing.py new file mode 100644 index 0000000..f4e6c10 --- /dev/null +++ b/backend/tests/test_minute_routing.py @@ -0,0 +1,312 @@ +"""自定义分钟数据源路由回归测试。 + +对应设计文档 §4 测试矩阵 (docs/superpowers/specs/2026-07-18-minute-provider-unification-design.md)。 + +覆盖三个阻断问题: +1. stock-sdk 默认 freq 漂移 (5m → 1m) +2. 自定义源异常直接 500 (无 try/except) +3. 插件化路由重复 + asset_type 未透传 + +mock 范式沿用 test_stocksdk_provider.py (monkeypatch 模块属性)。 +""" +from __future__ import annotations + +from datetime import date, datetime +from unittest.mock import MagicMock + +import httpx +import polars as pl + +from app.plugins.stocksdk import provider as sp +from app.plugins.stocksdk.provider import StockSDKProvider +from app.services import kline_sync + + +# ---------- 辅助 ---------- + +def _mock_minute_df(symbol: str = "600519.SH") -> pl.DataFrame: + """构造非空分钟 K df, 用于 mock provider.get_minute 返回值。""" + return pl.DataFrame({ + "symbol": [symbol], + "datetime": [datetime(2026, 1, 15, 9, 35, 0)], + "open": [100.0], + "high": [101.0], + "low": [99.5], + "close": [100.5], + "volume": [1000.0], + "amount": [100500.0], + }) + + +def _setup_custom_provider(monkeypatch, provider: object, has_dataset: bool = True) -> None: + """统一 mock 自定义分钟源路由前置: preferences + provider_has_dataset + get_provider。 + + - preferences.get_minute_data_provider → "mock_src" + - custom.provider_has_dataset → has_dataset + - custom.get_provider → provider + """ + monkeypatch.setattr( + kline_sync.preferences, + "get_minute_data_provider", + lambda: "mock_src", + ) + monkeypatch.setattr( + "app.data_providers.custom.provider_has_dataset", + lambda name, ds: has_dataset, + ) + monkeypatch.setattr( + "app.data_providers.custom.get_provider", + lambda name: provider, + ) + + +# ---------- 测试 1: 自定义源成功返回 1 分钟 K ---------- + +def test_custom_minute_provider_returns_1m_k(monkeypatch): + """§4 测试 1: 自定义源成功返回 1m K, 且 provider 收到 freq="1m"。""" + spy = MagicMock(return_value=_mock_minute_df()) + mock_provider = MagicMock() + mock_provider.get_minute = spy + _setup_custom_provider(monkeypatch, mock_provider, has_dataset=True) + + df, fallback = kline_sync._try_custom_minute( + ["600519.SH"], + datetime(2026, 1, 15, 9, 25, 0), + datetime(2026, 1, 15, 15, 5, 0), + asset_type="stock", + ) + + assert fallback is False + assert df is not None + assert not df.is_empty() + # spy 收到 freq="1m" 和 asset_type="stock" + spy.assert_called_once() + _, kwargs = spy.call_args + assert kwargs.get("freq") == "1m" + assert kwargs.get("asset_type") == "stock" + + +# ---------- 测试 2: stock-sdk 收到 freq=1m → bridge job period="1" ---------- + +def test_stocksdk_get_minute_receives_freq_1m(monkeypatch): + """§4 测试 2: StockSDKProvider.get_minute(freq="1m") → bridge job period == "1"。 + + bridge.mjs opMinute 用 String(period), 1m → "1"。 + """ + captured: dict = {} + + def fake_run_job(job, timeout=None): + captured["job"] = job + # 返回空结果, 测试只验证 job.period + return {"ok": True, "op": job["op"], "rows": {}} + + monkeypatch.setattr(sp.bridge, "run_job", fake_run_job) + + StockSDKProvider().get_minute( + ["600519.SH"], None, None, freq="1m", + ) + + assert captured["job"]["op"] == "minute" + assert captured["job"]["period"] == "1" + + +# ---------- 测试 3: 自定义源异常 + TickFlow 也失败 → 返回空 (非 500) ---------- + +def test_custom_provider_exception_no_500(monkeypatch): + """§4 测试 3: 自定义源抛异常 + TickFlow 也失败, + fetch_minute_single / sync_minute_batch 返回空 df。 + """ + # 自定义源抛异常 + mock_provider = MagicMock() + mock_provider.get_minute.side_effect = httpx.TimeoutException("timeout") + _setup_custom_provider(monkeypatch, mock_provider, has_dataset=True) + + # mock get_client 返回 mock client, 其 klines.batch raise (TickFlow 也失败) + mock_tf = MagicMock() + mock_tf.klines.batch.side_effect = Exception("tickflow fail") + monkeypatch.setattr(kline_sync, "get_client", lambda: mock_tf) + + # fetch_minute_single: 自定义源异常 → fall through → TickFlow 异常 → 返回空 + df_single = kline_sync.fetch_minute_single( + "600519.SH", date(2026, 1, 15), asset_type="stock", + ) + assert isinstance(df_single, pl.DataFrame) + assert df_single.is_empty() + + # sync_minute_batch: 同一路径, 返回空 + df_batch = kline_sync.sync_minute_batch( + ["600519.SH"], + start_time=datetime(2026, 1, 15, 9, 25, 0), + end_time=datetime(2026, 1, 15, 15, 5, 0), + asset_type="stock", + ) + assert isinstance(df_batch, pl.DataFrame) + assert df_batch.is_empty() + + +# ---------- 测试 4: 未配 minute dataset → 回退 TickFlow ---------- + +def test_provider_without_minute_dataset_fallback(monkeypatch): + """§4 测试 4: provider_has_dataset 返回 False → (None, True) 回退 TickFlow。""" + mock_provider = MagicMock() + _setup_custom_provider(monkeypatch, mock_provider, has_dataset=False) + + df, fallback = kline_sync._try_custom_minute( + ["600519.SH"], None, None, asset_type="stock", + ) + + assert fallback is True + assert df is None + # provider.get_minute 不应被调用 (回退决策在前) + mock_provider.get_minute.assert_not_called() + + +# ---------- 测试 5: asset_type 透传到 provider ---------- + +def test_asset_type_threaded_to_provider(monkeypatch): + """§4 测试 5: stock/etf/index asset_type 透传到 provider.get_minute。""" + spy = MagicMock(return_value=_mock_minute_df()) + mock_provider = MagicMock() + mock_provider.get_minute = spy + _setup_custom_provider(monkeypatch, mock_provider, has_dataset=True) + + # 三次调用不同 asset_type + kline_sync.fetch_minute_single("600519.SH", date(2026, 1, 15), asset_type="stock") + kline_sync.fetch_minute_single("510300.SH", date(2026, 1, 15), asset_type="etf") + kline_sync.fetch_minute_single("000001.SH", date(2026, 1, 15), asset_type="index") + + # spy 被调 3 次, 每次收到对应 asset_type + assert spy.call_count == 3 + received_assets = [call.kwargs.get("asset_type") for call in spy.call_args_list] + assert received_assets == ["stock", "etf", "index"] + + +# ---------- 测试 6: 自定义源成功时不调 TickFlow ---------- + +def test_custom_success_skips_tickflow(monkeypatch): + """§4 测试 6: fetch_minute_single 自定义源成功 → 不调 get_client。""" + expected_df = _mock_minute_df() + mock_provider = MagicMock() + mock_provider.get_minute.return_value = expected_df + _setup_custom_provider(monkeypatch, mock_provider, has_dataset=True) + + # get_client 设为 spy, 若被调说明路由失败 + get_client_spy = MagicMock(name="get_client_spy") + monkeypatch.setattr(kline_sync, "get_client", get_client_spy) + + df = kline_sync.fetch_minute_single( + "600519.SH", date(2026, 1, 15), asset_type="stock", + ) + + # 返回的是 mock provider 的 df + assert df is expected_df + # TickFlow 路径未进入 + get_client_spy.assert_not_called() + + +# ---------- 测试 7: sync_minute_batch 自定义源成功直接返回 ---------- + +def test_sync_minute_batch_custom_success_returns_directly(monkeypatch): + """§4 测试 7: sync_minute_batch 自定义源成功 → 直接 return, 不走 segment 逻辑。""" + expected_df = _mock_minute_df() + mock_provider = MagicMock() + mock_provider.get_minute.return_value = expected_df + _setup_custom_provider(monkeypatch, mock_provider, has_dataset=True) + + get_client_spy = MagicMock(name="get_client_spy") + monkeypatch.setattr(kline_sync, "get_client", get_client_spy) + + df = kline_sync.sync_minute_batch( + ["600519.SH"], + start_time=datetime(2026, 1, 15, 9, 25, 0), + end_time=datetime(2026, 1, 15, 15, 5, 0), + asset_type="stock", + ) + + # 返回 mock provider 的 df, 不走 segment + assert df is expected_df + get_client_spy.assert_not_called() + + +# ---------- 测试 8: on_chunk_done 包装 (2参 → 3参补 seg_label='custom') ---------- + +def test_on_chunk_done_wrapped_to_3_args(monkeypatch): + """on_chunk_done 包装: provider 内部以 2 参 (cur, total) 调用 → + 上层 3 参 (cur, total, seg_label) spy 收到 seg_label='custom'。 + + 设计文档 §2: 保证自定义源路径进度展示不降级 (与 TickFlow 路径 3 参回调对齐)。 + """ + upper_cb = MagicMock(name="upper_3arg_cb") + + def provider_get_minute_side_effect(symbols, *, start_time, end_time, + asset_type, freq, on_chunk_done): + # 模拟 provider 实现内部以 2 参调用 on_chunk_done + # (如 GenericHTTPProvider/provider.py:127 / StockSDKProvider/provider.py:166) + if on_chunk_done is not None: + on_chunk_done(1, 3) + return _mock_minute_df() + + mock_provider = MagicMock() + mock_provider.get_minute.side_effect = provider_get_minute_side_effect + _setup_custom_provider(monkeypatch, mock_provider, has_dataset=True) + + df, fallback = kline_sync._try_custom_minute( + ["600519.SH"], + datetime(2026, 1, 15, 9, 25, 0), + datetime(2026, 1, 15, 15, 5, 0), + asset_type="stock", + on_chunk_done=upper_cb, + ) + + assert fallback is False + assert df is not None + # 上层 3 参 spy 被调用一次, 收到 (1, 3, "custom") + upper_cb.assert_called_once_with(1, 3, "custom") + + +# ---------- 测试 9: get_minute_batch 按 asset_type 拆分调用 sync_minute_batch ---------- + +def test_get_minute_batch_splits_stock_and_etf(monkeypatch): + """get_minute_batch 把 incomplete 拆成 stock/ETF 两组, 分别以 + asset_type='stock'/'etf' 调用 sync_minute_batch, 结果 concat 返回。 + + 覆盖 kline.py get_minute_batch 的双调用拼接逻辑 (本次提交改动量最大的部分)。 + 契约: 本端点只接受 stock/ETF (指数走 /api/index/minute), 故两分支覆盖全部 incomplete。 + """ + from app.api import kline as kline_api + + # mock sync_minute_batch: stock 返回 df_s, etf 返回 df_e (不同 symbol 便于 concat 后 filter 验证) + def fake_sync(symbols, *, start_time, end_time, batch_size, rpm, asset_type): + if asset_type == "stock": + return _mock_minute_df(symbol="600519.SH") + if asset_type == "etf": + return _mock_minute_df(symbol="510300.SH") + return pl.DataFrame() + sync_spy = MagicMock(side_effect=fake_sync) + monkeypatch.setattr(kline_api.kline_sync, "sync_minute_batch", sync_spy) + + # mock repo: ETF 集合含 510300.SH; 本地分钟K返回空 (强制走 incomplete 补拉) + mock_repo = MagicMock() + mock_repo.get_etf_symbol_set.return_value = {"510300.SH"} + mock_repo.get_minute_batch.return_value = pl.DataFrame() + + # mock capset: 有权限, limits 返回 None (lim.batch 访问被 `if lim else` 守护) + 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 + + body = {"symbols": ["600519.SH", "510300.SH"], "date": "2026-01-15"} + result = kline_api.get_minute_batch(mock_request, body) + + # sync_minute_batch 被调 2 次, asset_type 分别为 stock 和 etf + assert sync_spy.call_count == 2 + call_assets = sorted(call.kwargs.get("asset_type") for call in sync_spy.call_args_list) + assert call_assets == ["etf", "stock"] + + # 两个 symbol 都在结果里 (concat 后按 symbol filter 命中) + assert "600519.SH" in result["data"] + assert "510300.SH" in result["data"] From 91e16db547537357d4d7e1035bd912ec6a91fe12 Mon Sep 17 00:00:00 2001 From: intfoo Date: Sat, 18 Jul 2026 21:27:28 +0800 Subject: [PATCH 3/4] =?UTF-8?q?fix(layout):=20=E4=BF=AE=E5=A4=8D=20None=20?= =?UTF-8?q?=E6=A1=A3+=E8=87=AA=E5=AE=9A=E4=B9=89=E5=AE=9E=E6=97=B6?= =?UTF-8?q?=E6=BA=90=E7=94=A8=E6=88=B7=E4=BE=A7=E8=BE=B9=E6=A0=8F=E6=8C=87?= =?UTF-8?q?=E6=95=B0=E8=A1=8C=E6=83=85=E4=B8=8D=E6=98=BE=E7=A4=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 承接 #121 review: Layout.tsx:631 的 !isNoneTier 一刀切判定与同文件 546 行实时行情开关判据不一致, 导致 None 档+自定义实时源用户监控页 放行、实时开关可用, 但侧边栏指数行情卡片不渲染。后端 /api/intraday/indices 从缓存读取不调 TickFlow, 对该用户有数据, 纯前端拦截导致 UI 体验不一致。 将 !isNoneTier 改为 (!isNoneTier || !!realtimeProviderName), 与实时行情开关判据对齐。 --- frontend/src/components/Layout.tsx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/frontend/src/components/Layout.tsx b/frontend/src/components/Layout.tsx index bbda075..6614bba 100644 --- a/frontend/src/components/Layout.tsx +++ b/frontend/src/components/Layout.tsx @@ -628,7 +628,7 @@ export function Layout() { ) : null}
)} - {showSidebarQuotes && !isWatchlistMode && !isNoneTier && ( + {showSidebarQuotes && !isWatchlistMode && (!isNoneTier || !!realtimeProviderName) && ( )}
From d9a77a534acaae08920dadce069b6c9d1f3b81c9 Mon Sep 17 00:00:00 2001 From: intfoo Date: Sun, 19 Jul 2026 15:53:00 +0800 Subject: [PATCH 4/4] =?UTF-8?q?fix(minute):=20=E6=89=BF=E6=8E=A5=20#122=20?= =?UTF-8?q?review=20=E4=B8=89=E5=A4=84=E9=98=BB=E6=96=AD=E9=97=AE=E9=A2=98?= =?UTF-8?q?=E6=95=B4=E6=94=B9=20+=20resolver=20=E5=BC=82=E5=B8=B8=E8=BE=B9?= =?UTF-8?q?=E7=95=8C=E5=8A=A0=E5=9B=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/data_providers/custom/config.py | 4 + backend/app/data_providers/custom/provider.py | 21 +- backend/app/services/kline_sync.py | 69 +++- backend/tests/test_minute_routing.py | 331 +++++++++++++++++- 4 files changed, 407 insertions(+), 18 deletions(-) diff --git a/backend/app/data_providers/custom/config.py b/backend/app/data_providers/custom/config.py index 078d650..ae52033 100644 --- a/backend/app/data_providers/custom/config.py +++ b/backend/app/data_providers/custom/config.py @@ -33,6 +33,8 @@ class DatasetConfig: symbols_param: str = "symbols" start_param: str = "start_time" end_param: str = "end_time" + asset_type_param: str | None = None + freq_param: str | None = None @dataclass(frozen=True) @@ -72,6 +74,8 @@ def _dataset_from_dict(raw: dict[str, Any]) -> DatasetConfig: symbols_param=str(raw.get("symbols_param", "symbols") or "symbols"), start_param=str(raw.get("start_param", "start_time") or "start_time"), end_param=str(raw.get("end_param", "end_time") or "end_time"), + asset_type_param=str(raw.get("asset_type_param")) if raw.get("asset_type_param") else None, + freq_param=str(raw.get("freq_param")) if raw.get("freq_param") else None, ) diff --git a/backend/app/data_providers/custom/provider.py b/backend/app/data_providers/custom/provider.py index 1f80e91..d726c5b 100644 --- a/backend/app/data_providers/custom/provider.py +++ b/backend/app/data_providers/custom/provider.py @@ -111,16 +111,31 @@ class GenericHTTPProvider: symbols: list[str], start_time: datetime | None, end_time: datetime | None, - asset_type: AssetType = "stock", # noqa: ARG002 - freq: str = "1m", # noqa: ARG002 + asset_type: AssetType = "stock", + freq: str = "1m", on_chunk_done: Callable[[int, int], None] | None = None, ) -> pl.DataFrame: + """拉取分钟 K。 + + asset_type / freq 默认不传上游 (minute dataset URL 应返回 1m 数据)。 + 在 dataset 配置中设置 asset_type_param / freq_param 后, 这两个参数会以 + 配置的参数名注入请求 (GET → params, POST → body), 用于上游需区分 + stock/ETF/index 或固定频率的场景。 + """ cfg = self._dataset("minute") + override: dict[str, Any] = {} + if cfg.asset_type_param: + override[cfg.asset_type_param] = asset_type + if cfg.freq_param: + override[cfg.freq_param] = freq frames: list[pl.DataFrame] = [] chunks = chunked(symbols, cfg.batch) for i, chunk in enumerate(chunks): sleep_between_batches(i, cfg.rpm) - rows = self._request_rows(cfg, symbols=chunk, start_time=start_time, end_time=end_time) + rows = self._request_rows( + cfg, symbols=chunk, start_time=start_time, end_time=end_time, + override_params=override or None, override_body=override or None, + ) df = self._mapped_frame(cfg, rows) df = self._normalize_minute(df) if not df.is_empty(): diff --git a/backend/app/services/kline_sync.py b/backend/app/services/kline_sync.py index 2489718..f7c9f6a 100644 --- a/backend/app/services/kline_sync.py +++ b/backend/app/services/kline_sync.py @@ -529,6 +529,34 @@ def _write_minute_partition(df: pl.DataFrame, minute_dir) -> int: return written +def _resolve_minute_provider( + provider_name: str, +) -> tuple[object | None, bool, str | None]: + """统一解析 custom minute provider, 把所有 resolver 调用纳入同一异常边界。 + + 供 _try_custom_minute 和 sync_and_persist_minute 共用, 避免两处分别调 + provider_has_dataset / get_provider 时漏掉异常边界 (Issue 2 加固项)。 + + 返回 (provider, should_fallback_to_tickflow, error_msg): + - provider_name == "tickflow" 或未配 minute dataset → (None, True, None) 静默降级 + - resolver 异常 (registry 损坏 / 插件失效 / provider name 不存在) → (None, True, str(e)) + - 成功 → (provider, False, None) + + 上层依据 error_msg 决定是否 logger.warning (区分"未配"与"异常")。 + 注意: provider.get_minute() 仍由调用方在自身 try 块内调用 (业务异常, 非解析异常)。 + """ + if provider_name == "tickflow": + return (None, True, None) + from app.data_providers import custom as custom_sources + try: + if not custom_sources.provider_has_dataset(provider_name, "minute"): + return (None, True, None) + provider = custom_sources.get_provider(provider_name) + return (provider, False, None) + except Exception as e: # noqa: BLE001 + return (None, True, str(e)) + + def _try_custom_minute( symbols: list[str], start_time: datetime | None, @@ -548,17 +576,21 @@ def _try_custom_minute( None 档用户 TickFlow 失败返回空。不显式判断 tier, 避免 #126 augmented capability 逻辑干扰。 + resolver 异常边界由 _resolve_minute_provider 统一兜底; 业务调用 + (provider.get_minute) 仍在本函数 try 块内, 与 resolver 异常分离 + 便于日志区分 ("resolution failed" vs "call failed")。 + on_chunk_done 适配: 上层回调是 3 参 (cur, total, seg_label), provider 实现内部以 2 参 (cur, total) 调用。这里包装一层, provider 调 2 参时补 默认 seg_label="custom" 转发给上层, 保证进度展示不降级。 """ provider_name = preferences.get_minute_data_provider() - if provider_name == "tickflow": + provider, fallback, err = _resolve_minute_provider(provider_name) + if fallback: + if err is not None: + logger.warning("custom minute provider %s resolution failed, falling back to TickFlow: %s", + provider_name, err) return (None, True) - from app.data_providers import custom as custom_sources - if not custom_sources.provider_has_dataset(provider_name, "minute"): - return (None, True) - provider = custom_sources.get_provider(provider_name) # 包装 on_chunk_done: provider 调 2 参 → 补 seg_label="custom" → 转发上层 3 参 wrapped_cb: Callable[[int, int], None] | None = None @@ -574,7 +606,7 @@ def _try_custom_minute( ) return (df, False) except Exception as e: # noqa: BLE001 - logger.warning("custom minute provider %s failed, falling back to TickFlow: %s", + logger.warning("custom minute provider %s call failed, falling back to TickFlow: %s", provider_name, e) return (None, True) @@ -612,10 +644,15 @@ def sync_minute_batch( asset_type=asset_type, freq="1m", on_chunk_done=on_chunk_done, ) if not fallback: - # _try_custom_minute 成功时返回 (df, False), df 非 None; - # fallback=True 时才返回 (None, True)。故此处 df 必非 None。 - # (旧版 `df or pl.DataFrame()` 触发 polars DataFrame __bool__ TypeError, 已修。) - return df if df is not None else pl.DataFrame() + # 自定义源成功: 遵守与 TickFlow 路径一致的 on_segment 契约。 + # 传了 on_segment (如 sync_and_persist_minute 流式落盘) → 调 on_segment, 返回空 df; + # 未传 on_segment (如 fetch_minute_single 实时补拉) 或空 df → 原样返回 df。 + df = df if df is not None else pl.DataFrame() + if on_segment and not df.is_empty(): + # 空 df 不调 on_segment, 与 TickFlow 路径 `if seg_out:` (L684) 对称 + on_segment(df) + return pl.DataFrame() + return df tf = get_client() @@ -967,10 +1004,14 @@ def sync_and_persist_minute( on_chunk_done(current, total) 每个 chunk 完成后回调。 """ minute_provider = preferences.get_minute_data_provider() - minute_is_custom = False - if minute_provider != "tickflow": - from app.data_providers import custom as custom_sources - minute_is_custom = custom_sources.provider_has_dataset(minute_provider, "minute") + # resolver 调用统一走 _resolve_minute_provider, 与 _try_custom_minute 共用异常边界。 + # resolver 异常时视为非 custom (minute_is_custom=False), 走 capset 检查 → + # sync_minute_batch 内 _try_custom_minute 会再次 resolver 异常 → fallback TickFlow。 + _, fallback, resolve_err = _resolve_minute_provider(minute_provider) + minute_is_custom = not fallback + if resolve_err is not None: + logger.warning("custom minute provider %s resolution failed at sync_and_persist_minute, treating as non-custom: %s", + minute_provider, resolve_err) if not symbols: return 0 if not minute_is_custom and not capset.has(Cap.KLINE_MINUTE_BATCH): diff --git a/backend/tests/test_minute_routing.py b/backend/tests/test_minute_routing.py index f4e6c10..7b6ea3e 100644 --- a/backend/tests/test_minute_routing.py +++ b/backend/tests/test_minute_routing.py @@ -207,7 +207,10 @@ def test_custom_success_skips_tickflow(monkeypatch): # ---------- 测试 7: sync_minute_batch 自定义源成功直接返回 ---------- def test_sync_minute_batch_custom_success_returns_directly(monkeypatch): - """§4 测试 7: sync_minute_batch 自定义源成功 → 直接 return, 不走 segment 逻辑。""" + """§4 测试 7: sync_minute_batch 自定义源成功 + 未传 on_segment → 原样返回 df (实时补拉契约)。 + + 传了 on_segment 时走流式落盘分支 (见测试 10), 此处验证未传时的实时补拉契约。 + """ expected_df = _mock_minute_df() mock_provider = MagicMock() mock_provider.get_minute.return_value = expected_df @@ -310,3 +313,329 @@ def test_get_minute_batch_splits_stock_and_etf(monkeypatch): # 两个 symbol 都在结果里 (concat 后按 symbol filter 命中) assert "600519.SH" in result["data"] assert "510300.SH" in result["data"] + + +# ---------- 测试 10: sync_minute_batch 自定义源成功时调 on_segment (Issue 1) ---------- + +def test_sync_minute_batch_custom_calls_on_segment(monkeypatch): + """Issue 1: sync_minute_batch 自定义源成功 + 传了 on_segment → + 调 on_segment(df), 返回空 df (数据已落盘)。 + """ + expected_df = _mock_minute_df() + mock_provider = MagicMock() + mock_provider.get_minute.return_value = expected_df + _setup_custom_provider(monkeypatch, mock_provider, has_dataset=True) + + get_client_spy = MagicMock(name="get_client_spy") + monkeypatch.setattr(kline_sync, "get_client", get_client_spy) + + on_segment_spy = MagicMock(name="on_segment_spy") + df = kline_sync.sync_minute_batch( + ["600519.SH"], + start_time=datetime(2026, 1, 15, 9, 25, 0), + end_time=datetime(2026, 1, 15, 15, 5, 0), + on_segment=on_segment_spy, + asset_type="stock", + ) + + on_segment_spy.assert_called_once_with(expected_df) + assert isinstance(df, pl.DataFrame) + assert df.is_empty() + get_client_spy.assert_not_called() + + +# ---------- 测试 11: 自定义源返回空 df 时不调 on_segment (Issue 1 边界) ---------- + +def test_sync_minute_batch_custom_empty_df_skips_on_segment(monkeypatch): + """Issue 1 边界: 自定义源返回空 df → 不调 on_segment (与 TickFlow `if seg_out:` 对称)。 + """ + mock_provider = MagicMock() + mock_provider.get_minute.return_value = pl.DataFrame() + _setup_custom_provider(monkeypatch, mock_provider, has_dataset=True) + + on_segment_spy = MagicMock(name="on_segment_spy") + df = kline_sync.sync_minute_batch( + ["600519.SH"], + start_time=datetime(2026, 1, 15, 9, 25, 0), + end_time=datetime(2026, 1, 15, 15, 5, 0), + on_segment=on_segment_spy, + asset_type="stock", + ) + + on_segment_spy.assert_not_called() + assert isinstance(df, pl.DataFrame) + assert df.is_empty() + + +# ---------- 测试 12: sync_and_persist_minute + custom provider 端到端落盘 (Issue 1) ---------- + +def test_sync_and_persist_minute_custom_persists(monkeypatch, tmp_path): + """Issue 1 端到端: sync_and_persist_minute + 自定义源 → + _write_minute_partition 被调, written > 0。 + """ + expected_df = _mock_minute_df() + mock_provider = MagicMock() + mock_provider.get_minute.return_value = expected_df + _setup_custom_provider(monkeypatch, mock_provider, has_dataset=True) + + # mock sync_and_persist_minute 内部依赖 (通过 monkeypatch kline_sync 模块属性) + monkeypatch.setattr(kline_sync, "_cleanup_null_datetime_minute", lambda repo: None) + monkeypatch.setattr(kline_sync, "_migrate_symbol_to_date_partition", lambda repo: None) + monkeypatch.setattr(kline_sync, "_latest_minute_datetime", lambda repo: None) + monkeypatch.setattr(kline_sync, "resolve_limit", lambda *a, **kw: MagicMock(batch=100, rpm=30)) + monkeypatch.setattr(kline_sync.preferences, "get_minute_sync_segment_days", lambda: 20) + + # _write_minute_partition spy: 记录调用, 返回行数 + write_spy = MagicMock(return_value=expected_df.height) + monkeypatch.setattr(kline_sync, "_write_minute_partition", write_spy) + + # get_client spy: 自定义源成功时不应走 TickFlow + get_client_spy = MagicMock(name="get_client_spy") + monkeypatch.setattr(kline_sync, "get_client", get_client_spy) + + # mock repo + mock_repo = MagicMock() + mock_repo.store.data_dir = tmp_path + mock_repo.db.execute = MagicMock() + + # mock capset (minute_is_custom=True 绕过 has() 检查, resolve_limit 已 mock) + mock_capset = MagicMock() + + written = kline_sync.sync_and_persist_minute( + ["600519.SH"], mock_repo, mock_capset, + ) + + assert write_spy.called + assert written == expected_df.height + assert written > 0 + get_client_spy.assert_not_called() + + +# ---------- 测试 13: get_provider 异常时 fall through TickFlow (Issue 2) ---------- + +def test_get_provider_exception_falls_back_to_tickflow(monkeypatch): + """Issue 2: get_provider raise ValueError → + _try_custom_minute 返回 (None, True), 无异常穿透。 + """ + monkeypatch.setattr( + kline_sync.preferences, + "get_minute_data_provider", + lambda: "mock_src", + ) + monkeypatch.setattr( + "app.data_providers.custom.provider_has_dataset", + lambda name, ds: True, # provider 存在, 但 get_provider 会抛 + ) + + def _raising_get_provider(name): + raise ValueError("not found") + monkeypatch.setattr( + "app.data_providers.custom.get_provider", + _raising_get_provider, + ) + + df, fallback = kline_sync._try_custom_minute( + ["600519.SH"], None, None, asset_type="stock", + ) + + assert fallback is True + assert df is None + + +# ---------- 测试 14: provider_has_dataset 异常时 fall through (Issue 2) ---------- + +def test_provider_has_dataset_exception_falls_back(monkeypatch): + """Issue 2: provider_has_dataset raise → + _try_custom_minute 返回 (None, True), 无异常穿透。 + """ + monkeypatch.setattr( + kline_sync.preferences, + "get_minute_data_provider", + lambda: "mock_src", + ) + + def _raising_has_dataset(name, ds): + raise RuntimeError("registry corrupted") + monkeypatch.setattr( + "app.data_providers.custom.provider_has_dataset", + _raising_has_dataset, + ) + + df, fallback = kline_sync._try_custom_minute( + ["600519.SH"], None, None, asset_type="stock", + ) + + assert fallback is True + assert df is None + + +# ---------- 测试 15-17: GenericHTTPProvider opt-in 参数传递 (Issue 3) ---------- + +from app.data_providers.custom.config import CustomSourceConfig, DatasetConfig +from app.data_providers.custom.provider import GenericHTTPProvider + + +def _make_minute_config(**extra) -> CustomSourceConfig: + """构造带 minute dataset 的最小 CustomSourceConfig, extra 传给 DatasetConfig。""" + field_map = {f: f for f in ( + "symbol", "datetime", "open", "high", "low", "close", "volume", "amount" + )} + return CustomSourceConfig( + name="test_src", + display_name="Test Source", + datasets={"minute": DatasetConfig( + url="http://example.com/minute", field_map=field_map, **extra, + )}, + ) + + +def _capture_request_rows(provider): + """替换 _request_rows 为捕获 spy, 返回 captured dict。""" + captured: dict = {} + + def fake_request_rows(cfg, *, symbols=None, start_time=None, end_time=None, + override_params=None, override_body=None): + captured["override_params"] = override_params + captured["override_body"] = override_body + return [] # 空行 → 空 df + + provider._request_rows = fake_request_rows + return captured + + +def test_generic_http_get_minute_passes_asset_type_when_configured(): + """Issue 3: 配了 asset_type_param="asset" → override 含 {"asset": "etf"}。""" + config = _make_minute_config(asset_type_param="asset") + provider = GenericHTTPProvider(config) + captured = _capture_request_rows(provider) + + provider.get_minute(["600519.SH"], None, None, asset_type="etf", freq="1m") + + assert captured["override_params"] == {"asset": "etf"} + assert captured["override_body"] == {"asset": "etf"} + + +def test_generic_http_get_minute_passes_freq_when_configured(): + """Issue 3: 配了 freq_param="period" → override 含 {"period": "1m"}。""" + config = _make_minute_config(freq_param="period") + provider = GenericHTTPProvider(config) + captured = _capture_request_rows(provider) + + provider.get_minute(["600519.SH"], None, None, asset_type="stock", freq="1m") + + assert captured["override_params"] == {"period": "1m"} + assert captured["override_body"] == {"period": "1m"} + + +def test_generic_http_get_minute_omits_params_when_not_configured(): + """Issue 3 向后兼容: 未配 asset_type_param/freq_param → override 为 None, 不传上游。""" + config = _make_minute_config() # 无 asset_type_param / freq_param + provider = GenericHTTPProvider(config) + captured = _capture_request_rows(provider) + + provider.get_minute(["600519.SH"], None, None, asset_type="etf", freq="1m") + + # override 为 None (空 dict → `override or None`), 不传上游 + assert captured["override_params"] is None + assert captured["override_body"] is None + + +# ---------- 测试 18: sync_and_persist_minute resolver 异常时优雅返回 0 (观察项加固) ---------- + +def test_sync_and_persist_minute_resolver_exception_returns_zero(monkeypatch, tmp_path): + """观察项加固: sync_and_persist_minute 开头 _resolve_minute_provider 异常 → + 不向接口抛 500, 优雅降级 (minute_is_custom=False → 走 capset 检查 → 无权限 return 0)。 + """ + monkeypatch.setattr( + kline_sync.preferences, + "get_minute_data_provider", + lambda: "mock_src", + ) + # provider_has_dataset 抛异常 (模拟 registry 损坏) + def _raising_has_dataset(name, ds): + raise RuntimeError("registry corrupted") + monkeypatch.setattr( + "app.data_providers.custom.provider_has_dataset", + _raising_has_dataset, + ) + + # 无 KLINE_MINUTE_BATCH 权限 → resolver 异常视为非 custom → capset 检查失败 → return 0 + mock_capset = MagicMock() + mock_capset.has.return_value = False + + mock_repo = MagicMock() + mock_repo.store.data_dir = tmp_path + + # 不应抛异常, 优雅降级到 0 + written = kline_sync.sync_and_persist_minute( + ["600519.SH"], mock_repo, mock_capset, + ) + + assert written == 0 + + +# ---------- 测试 19: _resolve_minute_provider helper 单元测试 ---------- + +def test_resolve_minute_provider_tickflow_returns_silent_fallback(): + """观察项加固: provider_name == "tickflow" → (None, True, None) 静默降级, 无 err。""" + provider, fallback, err = kline_sync._resolve_minute_provider("tickflow") + assert provider is None + assert fallback is True + assert err is None + + +def test_resolve_minute_provider_no_dataset_returns_silent_fallback(monkeypatch): + """观察项加固: 配了 custom 但未配 minute dataset → (None, True, None) 静默降级。""" + monkeypatch.setattr( + "app.data_providers.custom.provider_has_dataset", + lambda name, ds: False, # 已注册但未配 minute + ) + provider, fallback, err = kline_sync._resolve_minute_provider("mock_src") + assert provider is None + assert fallback is True + assert err is None # 未配 ≠ 异常, 不应触发 warning + + +def test_resolve_minute_provider_has_dataset_exception_returns_err(monkeypatch): + """观察项加固: provider_has_dataset 抛异常 → (None, True, str(e)), 上层据此 warning。""" + def _raising(name, ds): + raise RuntimeError("registry corrupted") + monkeypatch.setattr("app.data_providers.custom.provider_has_dataset", _raising) + provider, fallback, err = kline_sync._resolve_minute_provider("mock_src") + assert provider is None + assert fallback is True + assert err is not None + assert "registry corrupted" in err + + +def test_resolve_minute_provider_get_provider_exception_returns_err(monkeypatch): + """观察项加固: provider_has_dataset 返回 True 但 get_provider 抛 → (None, True, str(e))。""" + monkeypatch.setattr( + "app.data_providers.custom.provider_has_dataset", + lambda name, ds: True, + ) + def _raising_get(name): + raise ValueError("not found") + monkeypatch.setattr("app.data_providers.custom.get_provider", _raising_get) + provider, fallback, err = kline_sync._resolve_minute_provider("mock_src") + assert provider is None + assert fallback is True + assert err is not None + assert "not found" in err + + +def test_resolve_minute_provider_success_returns_provider(monkeypatch): + """观察项加固: 正常路径 → (provider, False, None)。""" + mock_provider = object() # 任意 truthy 对象即可 + monkeypatch.setattr( + "app.data_providers.custom.provider_has_dataset", + lambda name, ds: True, + ) + monkeypatch.setattr( + "app.data_providers.custom.get_provider", + lambda name: mock_provider, + ) + provider, fallback, err = kline_sync._resolve_minute_provider("mock_src") + assert provider is mock_provider + assert fallback is False + assert err is None