diff --git a/backend/app/plugins/tdxquant/client.py b/backend/app/plugins/tdxquant/client.py index 383aef21..798480de 100644 --- a/backend/app/plugins/tdxquant/client.py +++ b/backend/app/plugins/tdxquant/client.py @@ -11,6 +11,7 @@ from __future__ import annotations import hashlib import json import logging +import math import os import re import shutil @@ -282,29 +283,41 @@ class TdxQuantClient: """单标的除权事件, 每行 {Date 'YYYYMMDD', Type, Bonus, ShareBonus, AllotPrice, Allotment} — 金额/比例均为**每 10 股**口径(Bonus=10派X元)。 """ - start = f"{start_d:%Y-%m-%d}" if start_d else "" - end = f"{end_d:%Y-%m-%d}" if end_d else "" - try: - out = self._tqs.get_divid_factors(code, start, end) - except Exception as exc: # SDK 内部异常类型不可枚举 - raise TdxQuantError(f"get_divid_factors({code}): {exc}") from exc - if isinstance(out, list): - return out - # 环境装有 pandas 时 SDK 返回 DataFrame, 归一为 list[dict]。 - # 注意 Date 会成为 DataFrame 索引(同 get_market_data 的日期索引惯例), - # reset_index 后若列名漂移, 把首列恢复为 'Date'。 - to_dict = getattr(out, "to_dict", None) - reset_index = getattr(out, "reset_index", None) - if callable(to_dict) and callable(reset_index): + start = f"{start_d:%Y-%m-%d} 00:00:00" if start_d else "" + end = f"{end_d:%Y-%m-%d} 00:00:00" if end_d else "" + # SDK get_divid_factors 会把 _clean 拒绝的服务端错误和正常无事件 + # 都变成空 DataFrame。与日K共用原生重试/错误检查，避免丢失原因； + # 使用同一 get_cw_data 端点及 SDK 1.2.2 的四列顺序，单位不变。 + out = self._raw("get_cw_data", str(code).strip(), start, end) + if not isinstance(out, dict): + raise TdxQuantError(f"get_cw_data({code}) 返回异常结构: {type(out)!r}") + if not out: + return [] + dates, values, types = out.get("Date"), out.get("Value"), out.get("Type", []) + if (not all(isinstance(v, list) for v in (dates, values, types)) + or len(dates) != len(values)): + raise TdxQuantError(f"get_cw_data({code}) Date/Value/Type 列结构异常") + records = [] + fields = ("Bonus", "AllotPrice", "ShareBonus", "Allotment") + for i, item in enumerate(values): + date_key = "".join(c for c in str(dates[i]).strip() if c.isdigit())[:8] try: - records = reset_index().to_dict("records") - if isinstance(records, list) and records: - if "Date" not in records[0]: - first_key = next(iter(records[0])) - if first_key not in ("Type", "Bonus", "AllotPrice", "ShareBonus", "Allotment"): - for row in records: - row["Date"] = row.pop(first_key) - return records - except Exception: # noqa: BLE001 — 转换失败按结构异常处理 - pass - raise TdxQuantError(f"get_divid_factors({code}) 返回异常结构: {type(out)!r}") + event_date = datetime.strptime(date_key, "%Y%m%d").date() if len(date_key) == 8 else None + except ValueError: + event_date = None + if event_date is None or (start_d and event_date < start_d) or (end_d and event_date > end_d): + continue # SDK 同样过滤无效日期及服务端未裁剪的窗口外事件。 + if item is None: + item = [] + if not isinstance(item, list): + raise TdxQuantError(f"get_cw_data({code}) Value[{i}] 结构异常") + padded = item + [0] * max(0, 4 - len(item)) + try: + amounts = [float(v or 0) for v in padded[:4]] + except (TypeError, ValueError, OverflowError) as exc: + raise TdxQuantError(f"get_cw_data({code}) Value[{i}] 含无效数值") from exc + if not all(math.isfinite(v) for v in amounts): + raise TdxQuantError(f"get_cw_data({code}) Value[{i}] 含非有限数值") + records.append({"Date": date_key, "Type": types[i] if i < len(types) else "", + **dict(zip(fields, amounts, strict=True))}) + return records diff --git a/backend/app/plugins/tdxquant/provider.py b/backend/app/plugins/tdxquant/provider.py index 0b1dbcfe..583eb8fe 100644 --- a/backend/app/plugins/tdxquant/provider.py +++ b/backend/app/plugins/tdxquant/provider.py @@ -586,7 +586,9 @@ class TdxQuantProvider: events = client.divid_factors(symbol, start_d, end_d + timedelta(days=_ADJ_FUTURE_DAYS)) except TdxQuantError as e: logger.warning("TdxQuant 除权事件 %s 拉取失败: %s", symbol, e) - events = [] + # 失败不能当作「该股无除权」；由任务层标记失败，整轮不写入 + # 不完整因子，也避免权限/额度异常后继续请求其余数千只。 + raise events = [ev for ev in events if _compact_date(ev.get("Date")) is not None] if events: events.sort(key=lambda ev: str(ev.get("Date"))) @@ -610,7 +612,7 @@ class TdxQuantProvider: raw = client.his_dats(symbol, end_d, _bars_for_window((end_d - start_d).days)) except TdxQuantError as e: logger.warning("TdxQuant 除权配价 %s 日K失败: %s", symbol, e) - return {} + raise out: dict[date, float] = {} dates = raw.get("Date") or [] closes = raw.get("Close") or [] diff --git a/backend/tests/test_tdxquant_adj_response.py b/backend/tests/test_tdxquant_adj_response.py new file mode 100644 index 00000000..577dd235 --- /dev/null +++ b/backend/tests/test_tdxquant_adj_response.py @@ -0,0 +1,175 @@ +"""Replay native responses through the installed SDK without loading its DLL. + +The SDK turns both a valid empty result and a server error into empty pandas +frames. Exercise the actual SDK boundary, then provider math and persistence. +""" +from __future__ import annotations + +import json +from datetime import date, datetime +from types import SimpleNamespace + +import polars as pl +import pytest + +from app.plugins.tdxquant import client as tc +from app.plugins.tdxquant import provider as tp +from app.services import kline_sync + +sdk = pytest.importorskip("tdxaidata.tdxaidata") + +EVENT = {"ErrorId": 0, "Date": ["20240619"], "Type": ["1"], + "Value": [["2", "5", "1", "2"]]} +AUTH_ERROR = {"ErrorId": -1, "Error": "TokenKey Insufficient Or Key Error TDX-test-secret"} + + +@pytest.fixture +def make_client(monkeypatch): + monkeypatch.setattr(tc.time, "sleep", lambda _: None) + monkeypatch.setattr(tp, "_HIST_INTERVAL_S", 0) + + def make(responses, *, close_response=None): + responses = list(responses) + calls = [] + + def events(*args): + calls.append(args) + value = responses.pop(0) if len(responses) > 1 else responses[0] + return json.dumps(value) + + closes = close_response if close_response is not None else { + "ErrorId": 0, "Date": ["20240618", "20240619"], "Close": ["10", "8.35"], + } + native = SimpleNamespace(get_cw_data=events, get_his_dats=lambda *a: json.dumps(closes)) + + class ReplaySDK(sdk.tqs): + @classmethod + def _tdx(cls): + return native + + client = object.__new__(tc.TdxQuantClient) + client.api_key = "TDX-test-secret" + client._tqs = ReplaySDK + client._native_cls = SimpleNamespace(get_instance=lambda: native) + return client, calls + + return make + + +def test_event_values_and_dates_match_sdk_dataframe(make_client): + client, calls = make_client([EVENT]) + expected = client._tqs.get_divid_factors("600000.SH", "2024-01-01", "2024-12-31") + assert not expected.empty # The normal nonempty DataFrame was already supported. + actual = client.divid_factors("600000.SH", date(2024, 1, 1), date(2024, 12, 31)) + row = expected.reset_index().to_dict("records")[0] + row["Date"] = tp._compact_date(row["Date"]) + actual[0]["Date"] = tp._compact_date(actual[0]["Date"]) + assert actual == [row] + assert calls[-1] == ("600000.SH", "2024-01-01 00:00:00", "2024-12-31 00:00:00") + + +@pytest.mark.parametrize("payload", [{"ErrorId": 0}, {"ErrorId": 0, "Date": [], "Type": [], "Value": []}]) +def test_valid_empty_sdk_frame_means_no_events(make_client, payload): + client, _ = make_client([payload]) + assert client._tqs.get_divid_factors("000002.SZ").empty + assert client.divid_factors("000002.SZ") == [] + + +def test_auth_failure_is_preserved_without_retry_or_secret_output(make_client, capsys): + client, calls = make_client([AUTH_ERROR]) + with pytest.raises(tc.TdxQuantAuthError, match=r"凭据|额度") as caught: + client.divid_factors("000002.SZ") + assert len(calls) == 1 + assert "TDX-test-secret" not in str(caught.value) + assert "TDX-test-secret" not in capsys.readouterr().out + + +def test_transient_server_error_retries_and_recovers(make_client): + client, calls = make_client([{"ErrorId": 500, "Error": "server busy"}, EVENT]) + assert len(client.divid_factors("600000.SH")) == 1 + assert len(calls) == 2 + + +@pytest.mark.parametrize("payload", [None, [], {"unexpected": 1}, + {"Date": ["20240619"], "Type": ["1"], "Value": []}, + {"Date": ["20240619"], "Type": ["1"], "Value": [["invalid", 0, 0, 0]]}, + {"Date": ["20240619"], "Type": ["1"], "Value": [["NaN", 0, 0, 0]]}, + {"Date": ["20240619"], "Type": ["1"], "Value": [["Infinity", 0, 0, 0]]}, +]) +def test_malformed_response_remains_a_failure(make_client, payload): + client, _ = make_client([payload]) + with pytest.raises(tc.TdxQuantError): + client.divid_factors("600000.SH") + + +def test_date_window_is_inclusive_and_empty_window_is_valid(make_client): + client, _ = make_client([EVENT]) + assert len(client.divid_factors("600000.SH", date(2024, 6, 19), date(2024, 6, 19))) == 1 + assert client.divid_factors("600000.SH", date(2024, 6, 20)) == [] + + +def test_normalization_matches_sdk_without_pandas(make_client, monkeypatch): + monkeypatch.setattr(sdk, "pd", None) + client, _ = make_client([EVENT]) + assert client.divid_factors("600000.SH") == client._tqs.get_divid_factors("600000.SH") + + +def route_to(client, monkeypatch): + from app.data_providers import custom + + provider = tp.TdxQuantProvider() + provider._client = client + monkeypatch.setattr(kline_sync.preferences, "get_adj_factor_provider", lambda: "tdxquant") + monkeypatch.setattr(custom, "provider_has_dataset", lambda name, dataset: True) + monkeypatch.setattr(custom, "get_provider", lambda name: provider) + monkeypatch.setattr(kline_sync, "get_client", lambda: pytest.fail("不得跨源请求 TickFlow")) + return provider + + +def test_events_flow_to_parquet_with_existing_per_share_formula(make_client, monkeypatch, tmp_path): + client, _ = make_client([EVENT]) + route_to(client, monkeypatch) + repo = SimpleNamespace(store=SimpleNamespace(data_dir=tmp_path)) + added, affected = kline_sync.sync_adj_factor( + ["600000.SH"], repo, SimpleNamespace(has=lambda _: False), + start_time=datetime(2024, 1, 1), end_time=datetime(2024, 12, 31), + ) + assert added == 1 and affected == ["600000.SH"] + row = pl.read_parquet(tmp_path / "adj_factor/all.parquet").row(0, named=True) + assert row["trade_date"] == date(2024, 6, 19) + assert row["dividend"] == pytest.approx(0.2) + assert row["bonus"] == pytest.approx(0.1) + assert row["allot"] == pytest.approx(0.2) + assert row["allot_price"] == 5.0 + assert row["prev_close"] == 10.0 + assert row["ex_factor"] == pytest.approx(10.0 / 8.31) + + +def test_partial_failure_keeps_existing_factors_and_stops_remaining_symbols(make_client, monkeypatch, tmp_path): + client, calls = make_client([EVENT, AUTH_ERROR]) + route_to(client, monkeypatch) + path = tmp_path / "adj_factor/all.parquet" + path.parent.mkdir() + pl.DataFrame({"symbol": ["600000.SH"], "trade_date": [date(2023, 1, 1)], "ex_factor": [1.02]}).write_parquet(path) + before = path.read_bytes() + repo = SimpleNamespace(store=SimpleNamespace(data_dir=tmp_path)) + with pytest.raises(tc.TdxQuantAuthError): + kline_sync.sync_adj_factor(["600000.SH", "000002.SZ", "600519.SH"], repo, None, + end_time=datetime(2024, 12, 31)) + assert len(calls) == 2 + assert path.read_bytes() == before + + +def test_adj_dataset_probe_exposes_server_failure(make_client, monkeypatch): + client, calls = make_client([AUTH_ERROR]) + provider = route_to(client, monkeypatch) + result = provider.test_dataset("adj_factor", ["000002.SZ"]) + assert result["rows"] == 0 and "凭据" in result["error"] + assert len(calls) == 1 + + +def test_price_lookup_failure_is_not_reported_as_successful_zero_rows(make_client, monkeypatch): + client, _ = make_client([EVENT], close_response=AUTH_ERROR) + provider = route_to(client, monkeypatch) + with pytest.raises(tc.TdxQuantAuthError): + provider.get_adj_factors(["600000.SH"], None, datetime(2024, 12, 31)) diff --git a/docs/plugin-development.md b/docs/plugin-development.md index eb565044..df7d76c6 100644 --- a/docs/plugin-development.md +++ b/docs/plugin-development.md @@ -334,6 +334,8 @@ uv run --extra dev python -m ruff check app/plugins// tests/test_