From 5a4702edcb33e6073cc0d6b98fa428056ba3b989 Mon Sep 17 00:00:00 2001 From: shy3130 Date: Fri, 30 Oct 2026 22:29:34 +0800 Subject: [PATCH] =?UTF-8?q?fix(tdxquant):=20=E4=BF=AE=E5=A4=8D=E5=88=86?= =?UTF-8?q?=E9=92=9F=E5=90=8C=E6=AD=A5=E6=97=B6=E5=8C=BA=E4=B8=8E=E5=A4=B1?= =?UTF-8?q?=E8=B4=A5=E7=8A=B6=E6=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 在 SDK 清理响应前保留并脱敏服务端错误，明确的鉴权拒绝立即停止请求并触发熔断。历史分钟同步显式传递数据源失败，避免任务显示成功但落盘为零行。 将带时区的请求窗口归一为北京墙钟，补充鉴权、任务失败、UTC/北京时间窗口及 Parquet 落盘回归测试，并更新插件契约文档。 Signed-off-by: shy3130 --- backend/app/data_providers/base.py | 4 + backend/app/main.py | 6 ++ backend/app/plugins/tdxquant/client.py | 43 ++++++++- backend/app/plugins/tdxquant/probe.py | 20 ++++- backend/app/plugins/tdxquant/provider.py | 39 ++++++--- backend/app/services/kline_sync.py | 4 +- .../test_minute_custom_streaming_persist.py | 77 ++++++++++++++++ backend/tests/test_tdxquant_auth_breaker.py | 11 ++- backend/tests/test_tdxquant_probe.py | 20 +++++ backend/tests/test_tdxquant_provider.py | 87 ++++++++++++++++++- docs/plugin-development.md | 4 +- 11 files changed, 293 insertions(+), 22 deletions(-) diff --git a/backend/app/data_providers/base.py b/backend/app/data_providers/base.py index 5c1bfafd..15abeee7 100644 --- a/backend/app/data_providers/base.py +++ b/backend/app/data_providers/base.py @@ -16,6 +16,10 @@ import polars as pl AssetType = Literal["stock", "index", "etf"] +class ProviderRequestError(RuntimeError): + """数据源明确请求失败; 调用方须标记失败, 不得转换成成功的空数据。""" + + @dataclass(frozen=True) class ProviderCapabilities: instruments: bool = False diff --git a/backend/app/main.py b/backend/app/main.py index 61db91c5..fe98b024 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -54,6 +54,7 @@ from app.api import settings as settings_api from app.api import upgrade_guard as upgrade_guard_api from app.api.routes import router as core_router from app.config import settings +from app.data_providers.base import ProviderRequestError from app.enriched_generation import EnrichedGenerationUnavailableError from app.extensions.loader import ( configure_backend_extensions, @@ -732,6 +733,11 @@ async def capability_denied_handler(request: Request, exc: CapabilityDenied) -> async def daily_integrity_error_handler(request: Request, exc: DailyIntegrityError) -> JSONResponse: return JSONResponse(status_code=409, content={"detail": str(exc), "code": "daily_integrity"}) + +@app.exception_handler(ProviderRequestError) +async def provider_request_error_handler(request: Request, exc: ProviderRequestError) -> JSONResponse: + return JSONResponse(status_code=502, content={"detail": str(exc), "code": "provider_request"}) + # 生产期静态文件(前端 dist) def _register_static_routes(target: FastAPI, static_dir: Path) -> None: """挂载前端构建产物: /assets 走静态服务, 根级文件直出, 其余回退 SPA。 diff --git a/backend/app/plugins/tdxquant/client.py b/backend/app/plugins/tdxquant/client.py index 87268314..383aef21 100644 --- a/backend/app/plugins/tdxquant/client.py +++ b/backend/app/plugins/tdxquant/client.py @@ -9,14 +9,18 @@ DLL 与运行时 ini 随 wheel 分发。两个实测约束决定本模块的形 from __future__ import annotations import hashlib +import json import logging import os +import re import shutil import sys import time from datetime import date, datetime from pathlib import Path +from app.data_providers.base import ProviderRequestError + logger = logging.getLogger(__name__) # 全 A 股票池: 五个市场名实测(2026-10)合计约 5579 只; ETF/指数不在枚举内 @@ -38,10 +42,41 @@ LIB_DLL_NAME = _DLL_BY_PLATFORM.get(sys.platform, "libTdxAiData.so") LIB_ENV = "TDX_AI_DATA_LIB" # tdxaidata 读取: 指向 DLL 文件路径 -class TdxQuantError(RuntimeError): +class TdxQuantError(ProviderRequestError): """tdxaidata 调用失败(网络/参数/协议)。""" +class TdxQuantAuthError(TdxQuantError): + """服务端明确拒绝凭据或额度; 重试同一请求不能恢复。""" + + +def _raise_response_error(raw, api_key: str = "") -> None: + """SDK._clean 会打印并抹掉 ErrorId, 必须在它之前保留失败语义。""" + value = raw + if isinstance(value, (str, bytes)): + try: + value = json.loads(value) + except (ValueError, UnicodeDecodeError): + return + if not isinstance(value, dict): + return + error_id = value.get("ErrorId") + if (error_id is None or str(error_id) == "0") and "error" not in value: + return + message = str(value.get("Error") or value.get("error") or "未知错误") + if api_key: + message = message.replace(api_key, "[redacted]") + message = re.sub(r"TDX-[\w-]+", "[redacted]", message, flags=re.IGNORECASE)[:200] + if re.search( + r"tokenkey|unauthori[sz]ed|invalid.*(?:key|token)|(?:key|token).*(?:error|expired)|权限|积分|额度|认证", + message, re.IGNORECASE, + ): + raise TdxQuantAuthError( + "TdxQuant 凭据、接口权限或额度不可用，请核对所配置 Key 与通达信后台状态。服务端: " + message + ) + raise TdxQuantError(f"TdxQuant 服务端请求失败: {message}") + + def package_lib_dir() -> Path: """tdxaidata 包自带的 lib 目录(平台 DLL + 运行时 ini)。""" import tdxaidata @@ -156,12 +191,16 @@ class TdxQuantClient: last: Exception = TdxQuantError(f"{method} 未知失败") for attempt in range(_RETRIES): try: - out = self._tqs._clean(getattr(self._native_cls.get_instance(), method)(*args)) + raw = getattr(self._native_cls.get_instance(), method)(*args) + _raise_response_error(raw, getattr(self, "api_key", "")) + out = self._tqs._clean(raw) if out is not None: return out last = TdxQuantError(f"{method} 返回空数据(第 {attempt + 1} 次)") if attempt + 1 >= _NONE_RETRIES: break + except TdxQuantAuthError: + raise except Exception as exc: # noqa: BLE001 — 原生 ctypes 层异常类型不可枚举 last = TdxQuantError(f"{method}: {exc}") time.sleep(_RETRY_SLEEP_S) diff --git a/backend/app/plugins/tdxquant/probe.py b/backend/app/plugins/tdxquant/probe.py index a612536d..d1494223 100644 --- a/backend/app/plugins/tdxquant/probe.py +++ b/backend/app/plugins/tdxquant/probe.py @@ -10,7 +10,8 @@ TDX_AI_DATA_LIB 指向), 结论以 JSON 回传。 父进程指定的进度文件, 父进程据此定位死点并附退出码, 不再返回"无输出"; - TDX 会话按 IP 复用: 同 IP 有活跃好 Key 会话时(常驻后端必然如此), 错误 Key 的日K也能返回 ⇒ "无效 Key 拒绝"不可靠。本探测只验证 - 格式 + 连通性, Key 真伪由首次实拉(test_dataset / 日常轮询)暴露。 + 格式 + 连通性; 服务端明确拒绝凭据/额度时拒绝保存。 + 通过连接探测仍不代表所有数据接口的权限均可用。 """ from __future__ import annotations @@ -76,9 +77,17 @@ def run_worker() -> int: reachable = False for _ in range(3): try: - if native.get_market_snapshot_batch("600519.SH", "Now"): + raw = native.get_market_snapshot_batch("600519.SH", "Now") + tc._raise_response_error(raw, key) + from tdxaidata import tqs + + if tqs._clean(raw): reachable = True break + except tc.TdxQuantAuthError as exc: + stage("done", ok=False) + print(json.dumps({"ok": False, "reason": str(exc)})) + return 0 except Exception: # noqa: BLE001 pass time.sleep(1.2) @@ -87,7 +96,7 @@ def run_worker() -> int: print(json.dumps({"ok": False, "reason": "行情服务器连接失败(网络不可达)"})) return 0 - # 日K仅作辅助信号(会话按 IP 复用时错误 Key 也能通过, 不能作拒绝依据) + # 日K成功仅作辅助信号(会话按 IP 复用时错误 Key 也能通过), 明确拒绝则保留原因。 stage("auth") aux = "" try: @@ -96,8 +105,13 @@ def run_worker() -> int: ) from tdxaidata import tqs + tc._raise_response_error(raw, key) v = tqs._clean(raw) aux = "his_ok" if isinstance(v, dict) and v.get("Date") else "his_empty" + except tc.TdxQuantAuthError as exc: + stage("done", ok=False) + print(json.dumps({"ok": False, "reason": str(exc)})) + return 0 except Exception: # noqa: BLE001 aux = "his_exc" diff --git a/backend/app/plugins/tdxquant/provider.py b/backend/app/plugins/tdxquant/provider.py index 5f853397..0b1dbcfe 100644 --- a/backend/app/plugins/tdxquant/provider.py +++ b/backend/app/plugins/tdxquant/provider.py @@ -40,7 +40,12 @@ from zoneinfo import ZoneInfo import polars as pl -from app.plugins.tdxquant.client import SNAPSHOT_FIELDS, TdxQuantClient, TdxQuantError +from app.plugins.tdxquant.client import ( + SNAPSHOT_FIELDS, + TdxQuantAuthError, + TdxQuantClient, + TdxQuantError, +) logger = logging.getLogger(__name__) @@ -108,8 +113,8 @@ def probe_api_key(api_key: str) -> tuple[bool, str]: """候选 Key 探测: 格式校验 + 子进程连通性实探。 TDX 会话按 IP 复用(同 IP 活跃好 Key 会话时错误 Key 也能拉到日K, 实测), - "无效 Key 拒绝"不可靠 — 格式明显不对直接拒绝, 连通性通过即保存, - Key 真伪由首次实拉(test_dataset / 日常轮询)暴露, 卡片可一键试拉复核。 + "无效 Key 拒绝"不可靠 — 格式明显不对或服务端明确拒绝时不保存; + 通过连通性不代表所有接口权限均可用, 卡片可一键试拉复核。 """ key = (api_key or "").strip() if not key.startswith("TDX-") or len(key) < 12: @@ -398,6 +403,7 @@ class TdxQuantProvider: self.config = _TdxQuantConfig() self._client: TdxQuantClient | None = None self._auth_failed_at: float | None = None + self._auth_failure_reason = "" def _auth_cooldown_active(self) -> bool: if self._auth_failed_at is None: @@ -409,8 +415,9 @@ class TdxQuantProvider: def _trip_auth_breaker(self, streak: int, first_error: str) -> None: self._auth_failed_at = time.monotonic() + self._auth_failure_reason = first_error logger.error( - "TdxQuant 连续 %d 只标的返回空数据, 疑似凭据失效(Key 无效/过期/额度耗尽), " + "TdxQuant 连续 %d 只标的分钟请求失败, 疑似凭据、权限或额度异常, " "已熔断 %d 分钟: 分钟增量/修复轮暂停拉取。请到 设置→数据源 重新保存 " "TdxQuant Key (保存即自动恢复)。首个错误: %s", streak, int(self._AUTH_COOLDOWN_S // 60), first_error or "未知", @@ -692,15 +699,16 @@ class TdxQuantProvider: on_chunk_done(0, 0) return client = self._get_client() - end_dt = end_time or datetime.now() + # 同步窗口来自 cn_now/单股 API, 可能带时区; 原生接口和返回列均用北京墙钟。 + if start_time is not None and start_time.tzinfo is not None: + start_time = start_time.astimezone(_BEIJING).replace(tzinfo=None) + end_dt = end_time or datetime.now(_BEIJING) + if end_dt.tzinfo is not None: + end_dt = end_dt.astimezone(_BEIJING).replace(tzinfo=None) count = _minute_count_for_window(start_time, end_dt) for i, symbol in enumerate(symbols): - try: - # 分钟周期必须传完整时刻: end 截到 00:00:00 会把当日分钟全部排除在窗口外 - raw = client.his_dats(symbol, end_dt, count, period="1m") - except TdxQuantError as e: - logger.warning("TdxQuant 分钟K %s 拉取失败: %s", symbol, e) - raw = {} + # 请求失败应中止历史同步并标记失败, 不得作为无数据继续遍历全市场。 + raw = client.his_dats(symbol, end_dt, count, period="1m") rows = _minute_rows(symbol, raw, start_time, end_dt) if not rows and raw.get("Date") and not _minute_rows(symbol, raw, None, None): _warn_minute_unparsed(symbol, raw) # 去掉窗口过滤仍 0 行才是解析失败 @@ -737,6 +745,9 @@ class TdxQuantProvider: for symbol in symbols: try: raw = client.his_dats(symbol, now, fetch, period="1m") + except TdxQuantAuthError as e: + self._trip_auth_breaker(1, str(e)) + break except TdxQuantError as e: failures += 1 first_error = first_error or str(e) @@ -792,6 +803,9 @@ class TdxQuantProvider: for j, symbol in enumerate(targets): try: raw = client.his_dats(symbol, now, count, period="1m") + except TdxQuantAuthError as e: + self._trip_auth_breaker(1, str(e)) + break except TdxQuantError as e: failures += 1 first_error = first_error or str(e) @@ -894,6 +908,9 @@ class TdxQuantProvider: else self.get_minute(syms, end - timedelta(days=3), end)) except TdxQuantError as e: return {"provider": self.name, "dataset": dataset, "rows": 0, "error": str(e)} + if dataset == "full_minute" and self._auth_cooldown_active(): + return {"provider": self.name, "dataset": dataset, "rows": 0, + "error": self._auth_failure_reason or "TdxQuant 凭据或额度不可用，请核对后重新保存 Key"} head = df.head(5).to_dicts() for row in head: for k, v in list(row.items()): diff --git a/backend/app/services/kline_sync.py b/backend/app/services/kline_sync.py index a2a92550..aa2553fb 100644 --- a/backend/app/services/kline_sync.py +++ b/backend/app/services/kline_sync.py @@ -20,7 +20,7 @@ from typing import Any import polars as pl -from app.data_providers.base import AssetType +from app.data_providers.base import AssetType, ProviderRequestError from app.indicators.pipeline import filter_halt_days from app.market_time import CN_TZ, cn_now, cn_today from app.services import minute_adjust, preferences @@ -1112,6 +1112,8 @@ def _try_custom_minute( symbols, start_time=start_time, end_time=end_time, asset_type=asset_type, freq=freq, on_chunk_done=wrapped_cb, ) + except ProviderRequestError: + raise except Exception as e: # noqa: BLE001 logger.warning( "custom minute provider %s call failed; not falling back to TickFlow: %s", diff --git a/backend/tests/test_minute_custom_streaming_persist.py b/backend/tests/test_minute_custom_streaming_persist.py index 8fab272e..a30c8437 100644 --- a/backend/tests/test_minute_custom_streaming_persist.py +++ b/backend/tests/test_minute_custom_streaming_persist.py @@ -10,6 +10,7 @@ from __future__ import annotations from datetime import datetime import polars as pl +import pytest from app.services import kline_sync @@ -130,3 +131,79 @@ def test_no_segment_path_returns_df_directly(monkeypatch): ) assert len(calls) == 1 and calls[0] == ["600519.SH", "000001.SZ"] assert out.height == 2 + + +@pytest.mark.parametrize("streaming", [False, True]) +def test_explicit_provider_failure_is_not_reported_as_successful_empty_sync(monkeypatch, streaming): + from types import SimpleNamespace + + from app.data_providers.base import ProviderRequestError + + calls = [] + def fail(*args, **kwargs): + calls.append(args[0]) + raise ProviderRequestError("数据源凭据或额度不可用") + + monkeypatch.setattr(kline_sync, "_selected_minute_provider_name", lambda: "example") + monkeypatch.setattr(kline_sync, "_resolve_minute_provider", lambda name: (SimpleNamespace(get_minute=fail), False, None)) + monkeypatch.setattr(kline_sync, "get_client", lambda: pytest.fail("不得跨源请求 TickFlow")) + persisted = [] + with pytest.raises(ProviderRequestError, match="凭据或额度"): + kline_sync.sync_minute_batch(["600519.SH", "000001.SZ"], on_segment=persisted.append if streaming else None) + assert len(calls) == 1 + assert persisted == [] + + +@pytest.mark.parametrize("days", [20, 365]) +def test_background_minute_sync_marks_provider_denial_as_failed(monkeypatch, tmp_path, days): + import asyncio + from types import SimpleNamespace + + from app.api import kline as kline_api + from app.data_providers.base import ProviderRequestError + from app.services import pipeline_jobs + + store = pipeline_jobs.JobStore(store_dir=tmp_path / "jobs") + monkeypatch.setattr(pipeline_jobs, "job_store", store) + monkeypatch.setattr(pipeline_jobs, "try_acquire_run_slot", lambda job_id: True) + monkeypatch.setattr(pipeline_jobs, "release_run_slot", lambda job_id: None) + monkeypatch.setattr(pipeline_jobs, "run_with_capacity", lambda job_id, fn: fn()) + monkeypatch.setattr(kline_api, "_minute_allowed", lambda capset: True) + monkeypatch.setattr("app.tickflow.pools.get_pool", lambda name: ["600519.SH"]) + monkeypatch.setattr("app.api.data.invalidate_storage_cache", lambda: None) + def denied(*args, **kwargs): + raise ProviderRequestError("TdxQuant 凭据、接口权限或额度不可用") + monkeypatch.setattr(kline_sync, "sync_and_persist_minute", denied) + repo = SimpleNamespace(store=SimpleNamespace(data_dir=tmp_path), get_index_symbol_set=lambda: set()) + async def run(): + tasks = [] + create_task = asyncio.create_task + def capture(coro): + task = create_task(coro) + tasks.append(task) + return task + monkeypatch.setattr(asyncio, "create_task", capture) + result = await kline_api.trigger_minute_sync(repo, object(), override_days=days) + await tasks[0] + return store.get(result["job_id"]) + result = asyncio.run(run()) + assert result["status"] == "failed" + assert "凭据、接口权限或额度" in result["error"] + assert not list(tmp_path.rglob("*.parquet")) + + +def test_provider_request_error_has_readable_http_response(): + from fastapi import FastAPI + from fastapi.testclient import TestClient + + from app.data_providers.base import ProviderRequestError + from app.main import provider_request_error_handler + + app = FastAPI() + app.add_exception_handler(ProviderRequestError, provider_request_error_handler) + @app.get("/minute") + def minute(): + raise ProviderRequestError("数据源凭据或额度不可用") + response = TestClient(app).get("/minute") + assert response.status_code == 502 + assert response.json() == {"detail": "数据源凭据或额度不可用", "code": "provider_request"} diff --git a/backend/tests/test_tdxquant_auth_breaker.py b/backend/tests/test_tdxquant_auth_breaker.py index 5edcea11..99c6a37f 100644 --- a/backend/tests/test_tdxquant_auth_breaker.py +++ b/backend/tests/test_tdxquant_auth_breaker.py @@ -10,7 +10,7 @@ from __future__ import annotations import pytest from app.plugins.tdxquant import provider as tdx_provider -from app.plugins.tdxquant.client import TdxQuantError +from app.plugins.tdxquant.client import TdxQuantAuthError, TdxQuantError class _FakeClient: @@ -103,3 +103,12 @@ def test_success_resets_blank_streak(): p = _provider_with(fake) p.get_intraday_latest(symbols=[f"S{i}" for i in range(19)]) assert p._auth_failed_at is None # 中间成功一次即清零, 不达熔断阈值 + + +@pytest.mark.parametrize("method", ["get_intraday_latest", "get_intraday_batch"]) +def test_explicit_auth_denial_stops_after_first_symbol(method): + fake = _FakeClient([TdxQuantAuthError("凭据或额度不可用")] * 50) + provider = _provider_with(fake) + assert getattr(provider, method)([f"S{i}" for i in range(50)]).is_empty() + assert fake.calls == 1 + assert provider._auth_cooldown_active() diff --git a/backend/tests/test_tdxquant_probe.py b/backend/tests/test_tdxquant_probe.py index a666b151..bfa2c2d5 100644 --- a/backend/tests/test_tdxquant_probe.py +++ b/backend/tests/test_tdxquant_probe.py @@ -129,6 +129,26 @@ def test_run_worker_exception_reports_stage(monkeypatch, tmp_path, capsys): assert progress["stage"] == "exception" +def test_run_worker_rejects_explicit_auth_denial_after_reachable_snapshot(monkeypatch, tmp_path, capsys): + _patch_worker_env(monkeypatch, tmp_path) + import tdxaidata.tdxaidata as tdxmod + from tdxaidata import tqs + + class Native: + def get_market_snapshot_batch(self, *args): + return {"price": 1} + + def get_his_dats(self, *args): + return json.dumps({"ErrorId": -1, "Error": "TokenKey Insufficient Or Key Error"}) + + monkeypatch.setattr(tdxmod._NativeTdxAiData, "get_instance", classmethod(lambda cls: Native())) + monkeypatch.setattr(tqs, "_clean", lambda raw: raw if isinstance(raw, dict) else None) + assert probe.run_worker() == 0 + payload = json.loads(capsys.readouterr().out.strip().splitlines()[-1]) + assert payload["ok"] is False + assert "凭据" in payload["reason"] or "额度" in payload["reason"] + + def test_desktop_main_dispatches_probe_worker(monkeypatch): monkeypatch.setenv("TDX_PROBE_WORKER", "1") monkeypatch.setattr(desktop, "_guard_streams", lambda: None) diff --git a/backend/tests/test_tdxquant_provider.py b/backend/tests/test_tdxquant_provider.py index d86437f5..b8605a3b 100644 --- a/backend/tests/test_tdxquant_provider.py +++ b/backend/tests/test_tdxquant_provider.py @@ -184,6 +184,87 @@ def test_raw_gives_up_on_persistent_none(monkeypatch): c.his_dats("600519.SH", date(2026, 1, 10), 5) +@pytest.mark.parametrize("encoded", [False, True]) +def test_explicit_auth_error_is_preserved_without_retry_or_key_leak(monkeypatch, encoded): + import json + + secret = "TDX-private-test-key" + response = {"ErrorId": -1, "Error": f"TokenKey Insufficient Or Key Error {secret}"} + sleeps = [] + c = _client_with_raw(monkeypatch, [json.dumps(response) if encoded else response], sleeps) + c.api_key = secret + clean = c._tqs._clean + c._tqs._clean = lambda raw: None if raw else clean(raw) # SDK 会把错误抹成 None。 + with pytest.raises(tc.TdxQuantAuthError, match=r"凭据|额度") as caught: + c.his_dats("600519.SH", date(2026, 1, 10), 1, period="1m") + assert secret not in str(caught.value) + assert sleeps == [] + + +def test_server_error_retains_reason_and_can_retry(monkeypatch): + sleeps = [] + c = _client_with_raw(monkeypatch, [{"error": "server busy", "error_code": 503}, {"Date": ["20260101"]}], sleeps) + assert c.his_dats("600519.SH", date(2026, 1, 10), 1)["Date"] == ["20260101"] + assert sleeps + + +def test_historical_minute_failure_stops_remaining_symbols(): + fake = _FakeClient() + fake.his_default = tc.TdxQuantError("网络连接失败") + with pytest.raises(tc.TdxQuantError, match="连接失败"): + _provider(fake).get_minute(["600519.SH", "000001.SZ"], None, datetime(2026, 10, 29, 15)) + assert len(fake.his_calls) == 1 + + +@pytest.mark.parametrize("dataset", ["minute", "full_minute"]) +def test_minute_dataset_probe_reports_explicit_auth_failure(dataset): + fake = _FakeClient() + fake.his_default = tc.TdxQuantAuthError("凭据或额度不可用") + result = _provider(fake).test_dataset(dataset, ["600519.SH"]) + assert result["rows"] == 0 + assert "凭据或额度" in result["error"] + assert len(fake.his_calls) == 1 + + +@pytest.mark.parametrize("zone", [ZoneInfo("Asia/Shanghai"), ZoneInfo("UTC")]) +def test_minute_aware_request_window_matches_beijing_wallclock(zone): + fake = _FakeClient() + fake.his = {"600519.SH": _minute_raw([ + ("20261009", 93100, 10.0, 10.1, 9.9, 10.0, 1000.0, 100.0), + ])} + start = datetime(2026, 10, 9, 9, 30, tzinfo=_BEIJING).astimezone(zone) + end = datetime(2026, 10, 9, 9, 32, tzinfo=_BEIJING).astimezone(zone) + frame = _provider(fake).get_minute(["600519.SH"], start, end) + assert frame.height == 1 + assert frame["datetime"][0] == datetime(2026, 10, 9, 9, 31) + assert fake.his_calls[0]["end"] == datetime(2026, 10, 9, 9, 32) + + +def test_tdx_minute_window_flows_through_service_to_parquet(monkeypatch, tmp_path): + import polars as pl + + from app.services import kline_sync + + fake = _FakeClient() + fake.his = {"600519.SH": {"Value": _minute_raw([ + ("20261009", 93100, 10.0, 10.1, 9.9, 10.0, 1000.0, 100.0), + ])}} + provider = _provider(fake) + monkeypatch.setattr(kline_sync, "_selected_minute_provider_name", lambda: "tdxquant") + monkeypatch.setattr(kline_sync, "_resolve_minute_provider", lambda name: (provider, False, None)) + monkeypatch.setattr(kline_sync, "get_client", lambda: pytest.fail("不得跨源请求")) + kline_sync.sync_minute_batch( + ["600519.SH"], datetime(2026, 10, 9, 9, 30, tzinfo=_BEIJING), + datetime(2026, 10, 9, 9, 32, tzinfo=_BEIJING), + on_segment=lambda frame: kline_sync._write_minute_partition(frame, tmp_path), + ) + stored = pl.read_parquet(tmp_path / "date=2026-10-09/part.parquet") + assert stored.height == 1 + assert stored["datetime"][0] == datetime(2026, 10, 9, 9, 31) + assert stored["volume"][0] == 10.0 + assert stored["amount"][0] == 1000000.0 + + def test_snapshot_batch_retries_transient_failure(monkeypatch): sleeps: list[float] = [] c = _client_with_raw(monkeypatch, [], sleeps) @@ -556,11 +637,11 @@ def test_minute_freq_and_asset_guards(): assert fake.his_calls == [] # 未发起取数 -def test_minute_transport_error_soft(): +def test_minute_transport_error_is_not_successful_empty_result(): fake = _FakeClient() fake.his = {"600519.SH": tc.TdxQuantError("boom")} - df = _provider(fake).get_minute(["600519.SH"], None, datetime.now()) - assert df.is_empty() # 软失败不阻断 + with pytest.raises(tc.TdxQuantError, match="boom"): + _provider(fake).get_minute(["600519.SH"], None, datetime.now()) # ---------------------------------------------------------------- minute: #509 返回形态容错 diff --git a/docs/plugin-development.md b/docs/plugin-development.md index 22089cf3..3371642a 100644 --- a/docs/plugin-development.md +++ b/docs/plugin-development.md @@ -326,7 +326,9 @@ uv run --extra dev python -m ruff check app/plugins// tests/test_