From a8bb7a25bf10c260e4d45eae34d971e53802a86b Mon Sep 17 00:00:00 2001 From: 0112020179 Date: Tue, 8 Sep 2026 18:09:59 +0800 Subject: [PATCH] =?UTF-8?q?fix(kline):=20=E5=8E=8B=E7=BC=A9=E8=AF=A6?= =?UTF-8?q?=E6=83=85=E5=93=8D=E5=BA=94=E5=B9=B6=E9=99=90=E5=88=B6=E6=97=A0?= =?UTF-8?q?=E6=9D=83=E9=99=90=E5=88=86=E9=92=9F=E5=9B=9E=E9=80=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/indices.py | 5 +- backend/app/api/kline.py | 133 +++++++---- backend/app/api/settings.py | 4 +- backend/app/services/kline_sync.py | 17 +- backend/app/services/preferences.py | 4 +- backend/tests/test_kline_detail_transport.py | 235 +++++++++++++++++++ backend/tests/test_kline_minute_live.py | 6 +- backend/tests/test_kline_sync_timezone.py | 7 +- backend/tests/test_minute_routing.py | 27 ++- 9 files changed, 380 insertions(+), 58 deletions(-) create mode 100644 backend/tests/test_kline_detail_transport.py diff --git a/backend/app/api/indices.py b/backend/app/api/indices.py index 6d76b8a..5042bd7 100644 --- a/backend/app/api/indices.py +++ b/backend/app/api/indices.py @@ -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"), diff --git a/backend/app/api/kline.py b/backend/app/api/kline.py index a54e3b4..5fb6356 100644 --- a/backend/app/api/kline.py +++ b/backend/app/api/kline.py @@ -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") diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py index bb3af65..305b04d 100644 --- a/backend/app/api/settings.py +++ b/backend/app/api/settings.py @@ -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()} diff --git a/backend/app/services/kline_sync.py b/backend/app/services/kline_sync.py index acee83a..06e3728 100644 --- a/backend/app/services/kline_sync.py +++ b/backend/app/services/kline_sync.py @@ -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( diff --git a/backend/app/services/preferences.py b/backend/app/services/preferences.py index 99b59ae..bc9b9a2 100644 --- a/backend/app/services/preferences.py +++ b/backend/app/services/preferences.py @@ -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) diff --git a/backend/tests/test_kline_detail_transport.py b/backend/tests/test_kline_detail_transport.py new file mode 100644 index 0000000..ae39e84 --- /dev/null +++ b/backend/tests/test_kline_detail_transport.py @@ -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() diff --git a/backend/tests/test_kline_minute_live.py b/backend/tests/test_kline_minute_live.py index 7764024..a30b0a1 100644 --- a/backend/tests/test_kline_minute_live.py +++ b/backend/tests/test_kline_minute_live.py @@ -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], diff --git a/backend/tests/test_kline_sync_timezone.py b/backend/tests/test_kline_sync_timezone.py index 44d3c78..a25f78d 100644 --- a/backend/tests/test_kline_sync_timezone.py +++ b/backend/tests/test_kline_sync_timezone.py @@ -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) diff --git a/backend/tests/test_minute_routing.py b/backend/tests/test_minute_routing.py index 1145a4f..e848e21 100644 --- a/backend/tests/test_minute_routing.py +++ b/backend/tests/test_minute_routing.py @@ -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