fix(tdxquant): 修复分钟同步时区与失败状态

在 SDK 清理响应前保留并脱敏服务端错误，明确的鉴权拒绝立即停止请求并触发熔断。历史分钟同步显式传递数据源失败，避免任务显示成功但落盘为零行。

将带时区的请求窗口归一为北京墙钟，补充鉴权、任务失败、UTC/北京时间窗口及 Parquet 落盘回归测试，并更新插件契约文档。

Signed-off-by: shy3130 <shy3130@qq.com>

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

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。

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)

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"
 

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()):

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",

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"}

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()

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)

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 返回形态容错

docs/plugin-development.md

@@ -326,7 +326,9 @@ uv run --extra dev python -m ruff check app/plugins/<your_plugin>/ tests/test_<y
   - 提供 `realtime`(五市场全 A 股票池枚举 + 600 只/批快照, 批间失败隔离)、`depth5`(快照自带买卖五档数组, 手口径透传)、`daily`(count 模式单标的拉取, 原始价, volume 股→手, amount 万元→元, `iter_daily` 逐标的流式)、`adj_factor`(事件接口每10股口径 ÷10 转每股 + 本地K线一次配价, 交易所公式推导单事件比值, 涨跌停自检)、`minute`(1m count 模式单标的, 240 根/日, 深度约 40 交易日, datetime=Date+Time 拼北京墙钟 naive, 与日K同口径 股→手/万元→元)、`full_minute`(修复轮 `get_intraday_batch` 当日窗口 + 稳态增量 `get_intraday_latest` 每股最新 N 根; 无批量端点, 全市场一轮 10 分钟量级); 指数快照直接支持(上证指数走 TDX 999999.SH 别名, `get_realtime_indices` 映射); 快照无名称列由维表关联; 未声明 financial → 回退 TickFlow; 清单声明 `billing: points` + `key_manage_url`(商城积分明细入口)
   - `client.py` — 原生层封装(lib 私有物化 + token 注入双 ini + `TDX_AI_DATA_LIB` 指向、空闲首连 "Invalid server response" 重试包装、股票池 12h 缓存)
   - `provider.py` — Provider 实现(快照昨收=LastClose/高低=Max/Min 字段映射、涨跌幅 (Now-LastClose)/LastClose 小数制、RefreshTime 北京墙钟→epoch ms、软失败)
-  - `probe.py` — 候选 Key 探测的独立子进程(原生 DLL 每进程单 Key): 分阶段进度留痕定位原生层死点; 因 TDX 会话按 IP 复用(错误 Key 在活跃会话下也能拉数), 探测语义为格式+连通性而非 Key 真伪, 真伪由首次实拉暴露
+  - `probe.py` — 候选 Key 探测的独立子进程(原生 DLL 每进程单 Key): 分阶段进度留痕定位原生层死点; 服务端明确拒绝凭据或额度时不保存。因 TDX 会话按 IP 复用(错误 Key 在活跃会话下也能拉数), 通过探测仅代表格式与连通性检查通过, 不保证所有数据接口均有权限。
+
+TdxQuant 在 SDK 清理响应前检查服务端错误, 保留错误原因并脱敏; 明确的凭据/额度拒绝不重试, 实时分钟轮立即熔断。历史分钟请求失败会通过 `app.data_providers.base.ProviderRequestError` 传递到任务层并标记失败, 不转换成成功的 0 行或跨源重试; 其他插件也可用该异常显式声明请求失败。带时区的分钟请求窗口在插件边界归一为北京墙钟, 再与原生分钟列比较。正常停牌、休市或窗口内无数据的空结果保留原语义。
 
 ## 路由机制(无需关心, 仅参考)