diff --git a/backend/app/api/indices.py b/backend/app/api/indices.py
index e78c2a2..febd6bb 100644
--- a/backend/app/api/indices.py
+++ b/backend/app/api/indices.py
@@ -113,7 +113,7 @@ def get_index_minute(
     repo = request.app.state.repo
     info = _index_info(repo, symbol)
     day = trade_date or date.today()
-    df = kline_sync.fetch_minute_single(symbol, day)
+    df = kline_sync.fetch_minute_single(symbol, day, asset_type="index")
     return {
         "symbol": symbol,
         "name": info.get("name"),
diff --git a/backend/app/api/kline.py b/backend/app/api/kline.py
index b4cfe1a..2018d6f 100644
--- a/backend/app/api/kline.py
+++ b/backend/app/api/kline.py
@@ -540,14 +540,38 @@ def get_minute_batch(request: Request, body: dict):
         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)
-        live_df = kline_sync.sync_minute_batch(
-            incomplete,
-            start_time=start_time,
-            end_time=end_time,
-            batch_size=lim.batch if lim else None,
-            rpm=lim.rpm if lim else None,
-        )
-        if not live_df.is_empty():
+        # 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")
             for sym in incomplete:
                 sub = live_df.filter(pl.col("symbol") == sym).sort("datetime")
                 if not sub.is_empty():
@@ -594,7 +618,7 @@ def get_minute(
     if trade_date is None:
         # 本地无任何分钟K，尝试从 TickFlow 拉取当天
         trade_date = cn_today()
-        df = kline_sync.fetch_minute_single(symbol, trade_date)
+        df = kline_sync.fetch_minute_single(symbol, trade_date, asset_type=asset_type)
         price_limit = _get_price_limit_info(
             repo, symbol, trade_date, asset_type, stock_name,
         )
@@ -636,7 +660,7 @@ def get_minute(
         }
 
     # 本地不完整或无数据 → 从 TickFlow 实时拉取
-    live_df = kline_sync.fetch_minute_single(symbol, trade_date)
+    live_df = kline_sync.fetch_minute_single(symbol, trade_date, asset_type=asset_type)
     return {
         "symbol": symbol, "name": stock_name, "stock_info": stock_info,
         "date": str(trade_date), "rows": live_df.to_dicts(),
diff --git a/backend/app/data_providers/base.py b/backend/app/data_providers/base.py
index c17ca21..53b59f4 100644
--- a/backend/app/data_providers/base.py
+++ b/backend/app/data_providers/base.py
@@ -6,6 +6,7 @@ backtests stay data-source agnostic.
 """
 from __future__ import annotations
 
+from collections.abc import Callable
 from dataclasses import dataclass
 from datetime import datetime
 from typing import Literal, Protocol
@@ -57,8 +58,14 @@ class MarketDataProvider(Protocol):
         end_time: datetime | None,
         asset_type: AssetType,
         freq: str = "1m",
+        on_chunk_done: Callable[[int, int], None] | None = None,
     ) -> 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/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 53b258e..d726c5b 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,15 +111,31 @@ 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",
+        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/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 530e088..f7c9f6a 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,88 @@ 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,
+    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 逻辑干扰。
+
+    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()
+    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)
+
+    # 包装 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 call 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 +621,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 +639,20 @@ 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:
+        # 自定义源成功: 遵守与 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()
 
@@ -572,7 +660,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 +677,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 +689,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,11 +825,26 @@ 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:
-    """从 TickFlow 实时拉取单股单日分钟 K（不写入本地）。"""
+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)
     end_time = datetime(trade_date.year, trade_date.month, trade_date.day, 15, 5, 0)
+
+    # 自定义数据源分流: 与 sync_minute_batch 一致, 配了自定义分钟源时走 custom provider,
+    # 避免无 TickFlow Pro+ 权限的用户分时图首次打开(本地无数据)时补拉失败返回空。
+    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:
         raw = tf.klines.batch(
@@ -901,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):
@@ -964,6 +1071,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..7b6ea3e
--- /dev/null
+++ b/backend/tests/test_minute_routing.py
@@ -0,0 +1,641 @@
+"""自定义分钟数据源路由回归测试。
+
+对应设计文档 §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 自定义源成功 + 未传 on_segment → 原样返回 df (实时补拉契约)。
+
+    传了 on_segment 时走流式落盘分支 (见测试 10), 此处验证未传时的实时补拉契约。
+    """
+    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"]
+
+
+# ---------- 测试 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
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}
             </div>
           )}
-          {showSidebarQuotes && !isWatchlistMode && !isNoneTier && (
+          {showSidebarQuotes && !isWatchlistMode && (!isNoneTier || !!realtimeProviderName) && (
             <SidebarIndexQuotes rows={sidebarIndexQuotes?.rows} items={sidebarIndexes} />
           )}
         </div>
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 (
       <div className="flex flex-col items-center justify-center py-20 text-center">
         <div className="inline-flex items-center justify-center w-14 h-14 rounded-2xl