Merge pull request #266 from 0112020179/codex/fix-kline-transfer-minute-capability

fix(kline): 压缩详情响应并限制无权限分钟回退
2026-09-12 16:44:15 +08:00 · 2026-09-08 21:06:40 +08:00
parent de250f72bc a8bb7a25bf
commit cdd41a6d15
9 changed files with 380 additions and 58 deletions
@@ -70,9 +70,12 @@ def get_index_minute(
 ):
    """实时读取指数分钟 K。不写入股票分钟 parquet。"""
    repo = request.app.state.repo
+    capset = request.app.state.capabilities
    info = _index_info(repo, symbol)
    day = trade_date or date.today()
-    df = kline_sync.fetch_minute_single(symbol, day, asset_type="index")
+    df = kline_sync.fetch_minute_single(
+        symbol, day, asset_type="index", capset=capset,
+    )
    return {
        "symbol": symbol,
        "name": info.get("name"),
@@ -27,7 +27,7 @@ router = APIRouter(prefix="/api/kline", tags=["kline"])
 def _gzip_payload(request: Request, payload: dict, *, pref_key: str) -> dict | Response:
    """大 JSON 响应的传输压缩: 偏好开启 + 客户端接受 gzip + 响应超阈值才压。

-    分时/日K批量各自独立偏好键 (网络设置里大开关批量、子开关单独控制)。
+    分时/日K各自使用独立偏好键 (沿用已有 *_batch_compress 存储键保证兼容)。
    level 6 实测 13MB ≈ 290ms CPU 压掉 87%; level 9 要 2.5s 不可用。
    datetime → isoformat, 与 FastAPI jsonable_encoder 输出一致
    (前端 since 增量按字符串字典序比较, 格式必须与非压缩路径相同)。
@@ -393,7 +393,11 @@ def get_daily(
        except Exception as e:
            raise HTTPException(status_code=502, detail=f"TickFlow fetch failed: {e}") from e
        if raw.is_empty():
-            return {"symbol": symbol, "name": stock_name, "stock_info": stock_info, "rows": []}
+            return _gzip_payload(
+                request,
+                {"symbol": symbol, "name": stock_name, "stock_info": stock_info, "rows": []},
+                pref_key="daily_batch_compress",
+            )
        # 拉除权因子做前复权 (Starter+ 有权限), 否则空 df → compute_enriched 退回未复权
        factors = pl.DataFrame()
        capset = getattr(request.app.state, "capabilities", None)
@@ -408,7 +412,11 @@ def get_daily(
        # 即使 live 模式也尝试追加实时蜡烛
        rows = _maybe_inject_live_candle(request, symbol, rows, asset_type)
        resp = {"symbol": symbol, "name": stock_name, "stock_info": stock_info, "rows": rows, "source": "live"}
-        return _attach_ext(resp, repo, symbol, ext_columns)
+        return _gzip_payload(
+            request,
+            _attach_ext(resp, repo, symbol, ext_columns),
+            pref_key="daily_batch_compress",
+        )

    rows = df.to_dicts()

@@ -416,7 +424,11 @@ def get_daily(
    rows = _maybe_inject_live_candle(request, symbol, rows, asset_type)

    resp = {"symbol": symbol, "name": stock_name, "stock_info": stock_info, "rows": rows, "source": "enriched"}
-    return _attach_ext(resp, repo, symbol, ext_columns)
+    return _gzip_payload(
+        request,
+        _attach_ext(resp, repo, symbol, ext_columns),
+        pref_key="daily_batch_compress",
+    )


 def _attach_ext(resp: dict, repo, symbol: str, ext_columns: Optional[str]) -> dict:
@@ -879,13 +891,21 @@ def get_minute_range(

    # 指数分钟 K 不落本地仓库, 最新分时仍由 /api/index/minute 实时读取。
    if asset_type == "index":
-        return {**base_response, "sessions": [], "source": "none"}
+        return _gzip_payload(
+            request,
+            {**base_response, "sessions": [], "source": "none"},
+            pref_key="minute_batch_compress",
+        )

    end = cn_today()
    start = end - timedelta(days=days * 3 + 20)
    minute = repo.get_minute_range([symbol], start, end, asset_type=asset_type)
    if minute.is_empty() or "datetime" not in minute.columns:
-        return {**base_response, "sessions": [], "source": "none"}
+        return _gzip_payload(
+            request,
+            {**base_response, "sessions": [], "source": "none"},
+            pref_key="minute_batch_compress",
+        )

    minute = minute.with_columns(
        pl.col("datetime").dt.date().alias("_trade_date"),
@@ -914,11 +934,15 @@ def get_minute_range(
                "rows": rows,
            })

-    return {
-        **base_response,
-        "sessions": sessions,
-        "source": "local" if sessions else "none",
-    }
+    return _gzip_payload(
+        request,
+        {
+            **base_response,
+            "sessions": sessions,
+            "source": "local" if sessions else "none",
+        },
+        pref_key="minute_batch_compress",
+    )


@router.get("/minute")
@@ -931,12 +955,14 @@ def get_minute(
    """读取某只股票某天的分钟 K 线。

    - 本地有完整数据(240条) → 直接返回
-    - 本地无数据或不完整 → 从 TickFlow 实时拉取返回（不写入）
+    - 本地无数据或不完整 → 从有效分钟数据源实时拉取返回(不写入)
+    - 自定义源失败时, 仅具备 TickFlow 单股分钟能力才回退 TickFlow
    - live=true 且当日连续竞价时段 → 跳过本地优先直接实时拉取:
      盘中分钟增量落盘的本地分区按 ≥60s 轮次更新, 90% 完整度启发式会让
      详情分时图停在上一增量轮, 与行情列表的节奏脱节
    """
    repo = request.app.state.repo
+    capset = request.app.state.capabilities
    asset_type = repo.resolve_asset_type(symbol)
    stock_info = _get_stock_info(repo, symbol) if asset_type == "stock" else _get_asset_info(repo, symbol, asset_type)
    stock_name = stock_info.get("name")
@@ -961,22 +987,29 @@ def get_minute(
        else:
            trade_date = today
    if trade_date is None:
-        # 本地无任何分钟K，尝试从 TickFlow 拉取当天
+        # 本地无任何分钟K, 尝试从当前有效分钟源拉取当天
        trade_date = cn_today()
-        df = kline_sync.fetch_minute_single(symbol, trade_date, asset_type=asset_type)
+        df = kline_sync.fetch_minute_single(
+            symbol, trade_date, asset_type=asset_type, capset=capset,
+        )
        price_limit = _get_price_limit_info(
            repo, symbol, trade_date, asset_type, stock_name,
        )
        prev_close = _get_previous_closes(
            repo, symbol, [trade_date], asset_type,
        ).get(trade_date)
-        return {
-            "symbol": symbol, "name": stock_name, "stock_info": stock_info,
-            "date": str(trade_date), "rows": df.to_dicts(), "source": "live",
-            "asset_type": asset_type,
-            "price_limit": price_limit,
-            "prev_close": prev_close,
-        }
+        return _gzip_payload(
+            request,
+            {
+                "symbol": symbol, "name": stock_name, "stock_info": stock_info,
+                "date": str(trade_date), "rows": df.to_dicts(),
+                "source": "live" if not df.is_empty() else "none",
+                "asset_type": asset_type,
+                "price_limit": price_limit,
+                "prev_close": prev_close,
+            },
+            pref_key="minute_batch_compress",
+        )

    prev_close = _get_previous_closes(
        repo, symbol, [trade_date], asset_type,
@@ -988,14 +1021,20 @@ def get_minute(
    if live and trade_date == cn_today() and in_continuous_session():
        # 详情分时轮询: 当日盘中实时拉取最新一根K, 不落盘; 拉空(源侧延迟/
        # 时段边界)则落回下方本地优先路径。
-        live_df = kline_sync.fetch_minute_single(symbol, trade_date, asset_type=asset_type)
+        live_df = kline_sync.fetch_minute_single(
+            symbol, trade_date, asset_type=asset_type, capset=capset,
+        )
        if not live_df.is_empty():
-            return {
-                "symbol": symbol, "name": stock_name, "stock_info": stock_info,
-                "date": str(trade_date), "rows": live_df.to_dicts(),
-                "source": "live", "asset_type": asset_type,
-                "price_limit": price_limit, "prev_close": prev_close,
-            }
+            return _gzip_payload(
+                request,
+                {
+                    "symbol": symbol, "name": stock_name, "stock_info": stock_info,
+                    "date": str(trade_date), "rows": live_df.to_dicts(),
+                    "source": "live", "asset_type": asset_type,
+                    "price_limit": price_limit, "prev_close": prev_close,
+                },
+                pref_key="minute_batch_compress",
+            )

    df = repo.get_minute(symbol, trade_date, asset_type=asset_type)

@@ -1019,24 +1058,34 @@ def get_minute(
    is_complete = not df.is_empty() and len(df) >= expected * 0.9  # 允许 10% 容差

    if is_complete:
-        return {
+        return _gzip_payload(
+            request,
+            {
+                "symbol": symbol, "name": stock_name, "stock_info": stock_info,
+                "date": str(trade_date), "rows": df.to_dicts(), "source": "local",
+                "asset_type": asset_type,
+                "price_limit": price_limit,
+                "prev_close": prev_close,
+            },
+            pref_key="minute_batch_compress",
+        )
+
+    # 本地不完整或无数据 → 从当前有效分钟源实时拉取
+    live_df = kline_sync.fetch_minute_single(
+        symbol, trade_date, asset_type=asset_type, capset=capset,
+    )
+    return _gzip_payload(
+        request,
+        {
            "symbol": symbol, "name": stock_name, "stock_info": stock_info,
-            "date": str(trade_date), "rows": df.to_dicts(), "source": "local",
+            "date": str(trade_date), "rows": live_df.to_dicts(),
+            "source": "live" if not live_df.is_empty() else "none",
            "asset_type": asset_type,
            "price_limit": price_limit,
            "prev_close": prev_close,
-        }
-
-    # 本地不完整或无数据 → 从 TickFlow 实时拉取
-    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(),
-        "source": "live" if not live_df.is_empty() else "none",
-        "asset_type": asset_type,
-        "price_limit": price_limit,
-        "prev_close": prev_close,
-    }
+        },
+        pref_key="minute_batch_compress",
+    )


@router.post("/sync")
@@ -810,7 +810,7 @@ def update_data_source_job_timeouts(req: DataSourceJobTimeoutPrefs) -> dict:

@router.put("/preferences/minute-batch-compress")
 def update_minute_batch_compress(req: MinuteBatchCompressPrefs) -> dict:
-    """保存分时批量响应的 gzip 传输压缩开关。逐请求即时读取, 保存后立即生效。"""
+    """保存分时详情与批量响应的 gzip 传输压缩开关。逐请求即时读取, 保存后立即生效。"""
    from app.services import preferences
    preferences.save({"minute_batch_compress": req.minute_batch_compress})
    return {"minute_batch_compress": preferences.get_minute_batch_compress()}
@@ -818,7 +818,7 @@ def update_minute_batch_compress(req: MinuteBatchCompressPrefs) -> dict:

@router.put("/preferences/daily-batch-compress")
 def update_daily_batch_compress(req: DailyBatchCompressPrefs) -> dict:
-    """保存日K批量响应的 gzip 传输压缩开关 (与分时独立)。逐请求即时读取。"""
+    """保存日K详情与批量响应的 gzip 传输压缩开关 (与分时独立)。逐请求即时读取。"""
    from app.services import preferences
    preferences.save({"daily_batch_compress": req.daily_batch_compress})
    return {"daily_batch_compress": preferences.get_daily_batch_compress()}
@@ -761,10 +761,8 @@ def _try_custom_minute(
      (None, True)   → 未配自定义源 / 未配 minute dataset / 自定义源异常 → 走 TickFlow
      (df, False)    → 自定义源成功(含空 df) → 直接用, 不回退

-    降级策略 (C): 自定义源异常时无条件 fall through 到 TickFlow,
-    由 TickFlow 路径自身 try/except 兜底。Pro+ 用户 TickFlow 成功返回数据,
-    None 档用户 TickFlow 失败返回空。不显式判断 tier, 避免 #126 augmented
-    capability 逻辑干扰。
+    自定义源异常时返回 fallback=True。单股拉取调用方另行检查 TickFlow 原生
+    能力, 避免自定义源增广能力误放行无权限请求。

    resolver 异常边界由 _resolve_minute_provider 统一兜底; 业务调用
    (provider.get_minute) 仍在本函数 try 块内, 与 resolver 异常分离
@@ -1220,8 +1218,14 @@ def fetch_minute_single(
    symbol: str,
    trade_date: date,
    asset_type: AssetType = "stock",
+    *,
+    capset: CapabilitySet,
 ) -> pl.DataFrame:
-    """实时拉取单股单日分钟 K(不写入本地)。优先自定义分钟源, 回退 TickFlow。"""
+    """实时拉取单股单日分钟 K(不写入本地)。
+
+    优先使用当前自定义分钟源。仅当 TickFlow 原生单股分钟能力存在时才允许
+    回退 TickFlow; 自定义源增广只授予 batch 能力, 不会误放行该回退路径。
+    """
    from datetime import datetime
    # 北京时间窗口必须带时区: naive datetime 会被 .timestamp() 按服务器本地时区解释,
    # UTC 容器上窗口整体偏移 8 小时, 分时补拉必然为空。
@@ -1238,6 +1242,9 @@ def fetch_minute_single(
        # 见 sync_minute_batch 同分支注释: df 在此必非 None。
        return df if df is not None else pl.DataFrame()

+    if not capset.has(Cap.KLINE_MINUTE_BY_SYMBOL):
+        return pl.DataFrame()
+
    tf = get_client()
    try:
        raw = tf.klines.batch(
@@ -248,7 +248,7 @@ def get_data_source_long_job_timeout_s() -> int:


 def get_minute_batch_compress() -> bool:
-    """分时批量响应是否启用 gzip 传输压缩。默认开启 (公网部署传输是大头);
+    """分时详情与批量响应是否启用 gzip 传输压缩。默认开启 (公网部署传输是大头);
    本机/内网可关闭省服务端 CPU。每次请求即时读取, 开关保存后立即生效。
    """
    raw = load().get("minute_batch_compress", True)
@@ -256,7 +256,7 @@ def get_minute_batch_compress() -> bool:


 def get_daily_batch_compress() -> bool:
-    """日K批量响应是否启用 gzip 传输压缩 (与分时各自独立配置)。默认开启。"""
+    """日K详情与批量响应是否启用 gzip 传输压缩 (与分时各自独立配置)。默认开启。"""
    raw = load().get("daily_batch_compress", True)
    return bool(raw)

@@ -0,0 +1,235 @@
+"""个股详情 K 线传输压缩与分钟源能力门控回归测试。"""
+
+from __future__ import annotations
+
+from datetime import date, datetime, timedelta
+from unittest.mock import MagicMock
+
+import polars as pl
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from app.api import indices, kline
+from app.tickflow.capabilities import Cap, CapabilityLimits, CapabilitySet
+
+_SYMBOL = "600000.SH"
+_TRADE_DATE = date(2026, 1, 15)
+
+
+@pytest.fixture(autouse=True)
+def _isolated_settings_and_clock(monkeypatch):
+    monkeypatch.setattr("app.services.preferences.get_daily_batch_compress", lambda: True)
+    monkeypatch.setattr("app.services.preferences.get_minute_batch_compress", lambda: True)
+    monkeypatch.setattr(kline, "cn_today", lambda: _TRADE_DATE)
+    monkeypatch.setattr(kline, "cn_now", lambda: datetime(2026, 1, 15, 10, 30))
+    monkeypatch.setattr(kline, "in_continuous_session", lambda: True)
+
+
+def _minute_rows(count: int = 240) -> pl.DataFrame:
+    start = datetime(2026, 1, 15, 9, 30)
+    return pl.DataFrame(
+        {
+            "symbol": [_SYMBOL] * count,
+            "datetime": [start + timedelta(minutes=i) for i in range(count)],
+            "open": [10.0] * count,
+            "high": [10.1] * count,
+            "low": [9.9] * count,
+            "close": [10.05] * count,
+            "volume": [1_000.0] * count,
+            "amount": [10_050.0] * count,
+        }
+    )
+
+
+class _DetailRepo:
+    def __init__(self, minute: pl.DataFrame | None = None) -> None:
+        self.minute = minute if minute is not None else _minute_rows()
+
+    def resolve_asset_type(self, symbol: str) -> str:
+        return "stock"
+
+    def get_instruments(self) -> pl.DataFrame:
+        return pl.DataFrame(
+            {
+                "symbol": [_SYMBOL],
+                "name": ["浦发银行"],
+                "total_shares": [1.0],
+                "float_shares": [1.0],
+            }
+        )
+
+    def get_daily_asset(self, asset_type, symbol, start, end, columns=None):
+        frame = pl.DataFrame(
+            {
+                "symbol": [_SYMBOL] * 30,
+                "date": [date(2025, 12, 15) + timedelta(days=i) for i in range(30)],
+                "open": [10.0] * 30,
+                "high": [10.2] * 30,
+                "low": [9.8] * 30,
+                "close": [10.1] * 30,
+                "volume": [1_000.0] * 30,
+                "amount": [10_100.0] * 30,
+                "ma5": [10.0] * 30,
+                "ma10": [10.0] * 30,
+                "ma20": [10.0] * 30,
+            }
+        )
+        return frame.select(columns) if columns else frame
+
+    def get_minute(self, symbol, trade_date, asset_type="stock") -> pl.DataFrame:
+        return self.minute
+
+    def get_minute_range(self, symbols, start, end, asset_type="stock") -> pl.DataFrame:
+        return self.minute
+
+
+class _IndexRepo:
+    def get_index_instruments(self) -> pl.DataFrame:
+        return pl.DataFrame({"symbol": ["000001.SH"], "name": ["上证指数"]})
+
+
+def _client(repo, capset: CapabilitySet | None = None) -> TestClient:
+    app = FastAPI()
+    app.include_router(kline.router)
+    app.include_router(indices.router)
+    app.state.repo = repo
+    app.state.capabilities = capset or CapabilitySet()
+    app.state.quote_service = None
+    return TestClient(app)
+
+
+@pytest.mark.parametrize(
+    ("path", "daily_compress", "minute_compress"),
+    [
+        (f"/api/kline/daily?symbol={_SYMBOL}&days=120", True, False),
+        (f"/api/kline/minute?symbol={_SYMBOL}&date={_TRADE_DATE}", False, True),
+        (f"/api/kline/minute-range?symbol={_SYMBOL}&days=10", False, True),
+    ],
+)
+def test_detail_kline_responses_use_configured_gzip(
+    monkeypatch,
+    path,
+    daily_compress,
+    minute_compress,
+):
+    monkeypatch.setattr(
+        "app.services.preferences.get_daily_batch_compress",
+        lambda: daily_compress,
+    )
+    monkeypatch.setattr(
+        "app.services.preferences.get_minute_batch_compress",
+        lambda: minute_compress,
+    )
+
+    response = _client(_DetailRepo()).get(
+        path,
+        headers={"Accept-Encoding": "gzip"},
+    )
+
+    assert response.status_code == 200
+    assert response.headers["content-encoding"] == "gzip"
+    assert response.json()["symbol"] == _SYMBOL
+    plain = _client(_DetailRepo()).get(path, headers={"Accept-Encoding": "identity"})
+    assert "content-encoding" not in plain.headers
+    assert response.json() == plain.json()
+    assert int(response.headers["content-length"]) < len(plain.content)
+
+    monkeypatch.setattr("app.services.preferences.get_daily_batch_compress", lambda: False)
+    monkeypatch.setattr("app.services.preferences.get_minute_batch_compress", lambda: False)
+    disabled = _client(_DetailRepo()).get(path, headers={"Accept-Encoding": "gzip"})
+    assert "content-encoding" not in disabled.headers
+    assert disabled.json() == plain.json()
+
+
+@pytest.mark.parametrize("live", [False, True])
+def test_free_tier_skips_tickflow_minute_fallback(monkeypatch, live):
+    get_client = MagicMock(side_effect=AssertionError("must not call TickFlow"))
+    monkeypatch.setattr("app.services.preferences.get_minute_data_provider", lambda: "tickflow")
+    monkeypatch.setattr("app.services.kline_sync.get_client", get_client)
+
+    response = _client(_DetailRepo(pl.DataFrame())).get(
+        "/api/kline/minute",
+        params={"symbol": _SYMBOL, "date": str(_TRADE_DATE), "live": live},
+    )
+
+    assert response.status_code == 200
+    assert response.json()["source"] == "none"
+    get_client.assert_not_called()
+
+
+def test_custom_minute_source_remains_available_without_tickflow_capability(monkeypatch):
+    provider = MagicMock()
+    provider.get_minute.return_value = _minute_rows(1)
+    monkeypatch.setattr("app.services.preferences.get_minute_data_provider", lambda: "custom")
+    monkeypatch.setattr(
+        "app.data_providers.custom.provider_has_dataset", lambda name, dataset: True
+    )
+    monkeypatch.setattr("app.data_providers.custom.get_provider", lambda name: provider)
+    get_client = MagicMock(side_effect=AssertionError("must not call TickFlow"))
+    monkeypatch.setattr("app.services.kline_sync.get_client", get_client)
+
+    response = _client(_DetailRepo(pl.DataFrame())).get(
+        "/api/kline/minute",
+        params={"symbol": _SYMBOL, "date": str(_TRADE_DATE)},
+    )
+
+    assert response.status_code == 200
+    assert response.json()["source"] == "live"
+    provider.get_minute.assert_called_once()
+    get_client.assert_not_called()
+
+
+def test_failed_custom_source_does_not_fall_back_to_unsupported_tickflow(monkeypatch):
+    provider = MagicMock()
+    provider.get_minute.side_effect = RuntimeError("custom source unavailable")
+    monkeypatch.setattr("app.services.preferences.get_minute_data_provider", lambda: "custom")
+    monkeypatch.setattr(
+        "app.data_providers.custom.provider_has_dataset", lambda name, dataset: True
+    )
+    monkeypatch.setattr("app.data_providers.custom.get_provider", lambda name: provider)
+    get_client = MagicMock(side_effect=AssertionError("must not call TickFlow"))
+    monkeypatch.setattr("app.services.kline_sync.get_client", get_client)
+
+    capset = CapabilitySet({Cap.KLINE_MINUTE_BATCH: CapabilityLimits()})
+    response = _client(_DetailRepo(pl.DataFrame()), capset).get(
+        "/api/kline/minute",
+        params={"symbol": _SYMBOL, "date": str(_TRADE_DATE)},
+    )
+
+    assert response.status_code == 200
+    assert response.json()["source"] == "none"
+    get_client.assert_not_called()
+
+
+def test_pro_tier_keeps_tickflow_minute_fallback(monkeypatch):
+    capset = CapabilitySet({Cap.KLINE_MINUTE_BY_SYMBOL: CapabilityLimits()})
+    tickflow = MagicMock()
+    tickflow.klines.batch.side_effect = RuntimeError("upstream unavailable")
+    get_client = MagicMock(return_value=tickflow)
+    monkeypatch.setattr("app.services.preferences.get_minute_data_provider", lambda: "tickflow")
+    monkeypatch.setattr("app.services.kline_sync.get_client", get_client)
+
+    response = _client(_DetailRepo(pl.DataFrame()), capset).get(
+        "/api/kline/minute",
+        params={"symbol": _SYMBOL, "date": str(_TRADE_DATE)},
+    )
+
+    assert response.status_code == 200
+    assert response.json()["source"] == "none"
+    get_client.assert_called_once()
+
+
+def test_free_tier_skips_tickflow_index_minute_fallback(monkeypatch):
+    get_client = MagicMock(side_effect=AssertionError("must not call TickFlow"))
+    monkeypatch.setattr("app.services.preferences.get_minute_data_provider", lambda: "tickflow")
+    monkeypatch.setattr("app.services.kline_sync.get_client", get_client)
+
+    response = _client(_IndexRepo()).get(
+        "/api/index/minute",
+        params={"symbol": "000001.SH", "date": str(_TRADE_DATE)},
+    )
+
+    assert response.status_code == 200
+    assert response.json()["source"] == "none"
+    get_client.assert_not_called()
@@ -15,6 +15,7 @@ from fastapi.testclient import TestClient

 from app.api.kline import router
 from app.market_time import CN_TZ, in_continuous_session
+from app.tickflow.capabilities import Cap, CapabilityLimits, CapabilitySet

 # 2026-08-26 是周三; 10:00 处于上午连续竞价, expected(已交易分钟) = 30
 _NOW = datetime(2026, 8, 26, 10, 0, tzinfo=CN_TZ)
@@ -48,6 +49,9 @@ def _client() -> TestClient:
    app = FastAPI()
    app.include_router(router)
    app.state.repo = _FakeRepo()
+    app.state.capabilities = CapabilitySet({
+        Cap.KLINE_MINUTE_BY_SYMBOL: CapabilityLimits(),
+    })
    return TestClient(app)


@@ -62,7 +66,7 @@ def _patch_market(monkeypatch, *, in_session: bool) -> None:
 def _patch_live_fetch(monkeypatch) -> None:
    import app.api.kline as kline_api

-    def _fake_fetch(symbol, trade_date, asset_type="stock"):
+    def _fake_fetch(symbol, trade_date, asset_type="stock", *, capset):
        return pl.DataFrame({
            "datetime": [datetime(2026, 8, 26, 9, 59)],
            "close": [11.11],
@@ -9,6 +9,7 @@ from datetime import date, datetime

 from app.market_time import CN_TZ
 from app.services import kline_sync
+from app.tickflow.capabilities import Cap, CapabilityLimits, CapabilitySet


 def test_fetch_minute_single_window_is_beijing_wall_clock(monkeypatch):
@@ -30,7 +31,11 @@ def test_fetch_minute_single_window_is_beijing_wall_clock(monkeypatch):
    monkeypatch.setattr(kline_sync, "_try_custom_minute", _fake_try_custom_minute)
    monkeypatch.setattr(kline_sync, "get_client", lambda: _FakeClient())

-    kline_sync.fetch_minute_single("600000.SH", date(2026, 8, 14))
+    kline_sync.fetch_minute_single(
+        "600000.SH",
+        date(2026, 8, 14),
+        capset=CapabilitySet({Cap.KLINE_MINUTE_BY_SYMBOL: CapabilityLimits()}),
+    )

    start = datetime.fromtimestamp(captured["start_ms"] / 1000, tz=CN_TZ)
    end = datetime.fromtimestamp(captured["end_ms"] / 1000, tz=CN_TZ)
@@ -40,6 +40,18 @@ def _mock_minute_df(symbol: str = "600519.SH") -> pl.DataFrame:
    })


+def _empty_capset():
+    from app.tickflow.capabilities import CapabilitySet
+
+    return CapabilitySet()
+
+
+def _tickflow_minute_capset():
+    from app.tickflow.capabilities import Cap, CapabilityLimits, CapabilitySet
+
+    return CapabilitySet({Cap.KLINE_MINUTE_BY_SYMBOL: CapabilityLimits()})
+
+
 def _setup_custom_provider(monkeypatch, provider: object, has_dataset: bool = True) -> None:
    """统一 mock 自定义分钟源路由前置: preferences + provider_has_dataset + get_provider。

@@ -131,6 +143,7 @@ def test_custom_provider_exception_no_500(monkeypatch):
    # fetch_minute_single: 自定义源异常 → fall through → TickFlow 异常 → 返回空
    df_single = kline_sync.fetch_minute_single(
        "600519.SH", date(2026, 1, 15), asset_type="stock",
+        capset=_tickflow_minute_capset(),
    )
    assert isinstance(df_single, pl.DataFrame)
    assert df_single.is_empty()
@@ -173,9 +186,15 @@ def test_asset_type_threaded_to_provider(monkeypatch):
    _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")
+    kline_sync.fetch_minute_single(
+        "600519.SH", date(2026, 1, 15), asset_type="stock", capset=_empty_capset(),
+    )
+    kline_sync.fetch_minute_single(
+        "510300.SH", date(2026, 1, 15), asset_type="etf", capset=_empty_capset(),
+    )
+    kline_sync.fetch_minute_single(
+        "000001.SH", date(2026, 1, 15), asset_type="index", capset=_empty_capset(),
+    )

    # spy 被调 3 次, 每次收到对应 asset_type
    assert spy.call_count == 3
@@ -197,7 +216,7 @@ def test_custom_success_skips_tickflow(monkeypatch):
    monkeypatch.setattr(kline_sync, "get_client", get_client_spy)

    df = kline_sync.fetch_minute_single(
-        "600519.SH", date(2026, 1, 15), asset_type="stock",
+        "600519.SH", date(2026, 1, 15), asset_type="stock", capset=_empty_capset(),
    )

    # 返回的是 mock provider 的 df