From 67ee83fba9710e14de66f8e8e901fbd766b86774 Mon Sep 17 00:00:00 2001 From: Break Zhang Date: Thu, 15 Oct 2026 15:45:39 +0800 Subject: [PATCH] =?UTF-8?q?fix(=E6=95=B0=E6=8D=AE=E6=BA=90):=20=E6=94=AF?= =?UTF-8?q?=E6=8C=81=E8=87=AA=E5=AE=9A=E4=B9=89=E4=BA=94=E6=8C=A1=E7=9B=98?= =?UTF-8?q?=E5=8F=A3=E8=B7=AF=E7=94=B1=20(#479)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: shy3130 <415333856@qq.com> --- backend/app/data_providers/custom/config.py | 4 +- backend/app/data_providers/custom/loader.py | 2 +- backend/app/data_providers/custom/provider.py | 49 +++++++++++++++++- backend/tests/test_custom_depth_provider.py | 51 +++++++++++++++++++ 4 files changed, 102 insertions(+), 4 deletions(-) diff --git a/backend/app/data_providers/custom/config.py b/backend/app/data_providers/custom/config.py index e265b25b..75c5dba2 100644 --- a/backend/app/data_providers/custom/config.py +++ b/backend/app/data_providers/custom/config.py @@ -7,7 +7,7 @@ from typing import Any, Literal import yaml -DatasetName = Literal["daily", "adj_factor", "realtime", "minute", "full_minute", "financial"] +DatasetName = Literal["daily", "adj_factor", "realtime", "minute", "depth5", "full_minute", "financial"] DEFAULT_TIMEOUT = 30.0 MAX_TIMEOUT = 300.0 @@ -106,7 +106,7 @@ def config_from_dict(raw: dict[str, Any], path: Path | None = None) -> CustomSou datasets = { name: _dataset_from_dict(cfg) for name, cfg in (raw.get("datasets") or {}).items() - if name in {"daily", "adj_factor", "realtime", "minute", "full_minute", "financial"} and isinstance(cfg, dict) + if name in {"daily", "adj_factor", "realtime", "minute", "depth5", "full_minute", "financial"} and isinstance(cfg, dict) } default_name = path.stem if path else "preview" name = str(raw.get("name", default_name) or default_name).lower() diff --git a/backend/app/data_providers/custom/loader.py b/backend/app/data_providers/custom/loader.py index ae183dde..42412b7b 100644 --- a/backend/app/data_providers/custom/loader.py +++ b/backend/app/data_providers/custom/loader.py @@ -424,7 +424,7 @@ def _sanitize_for_yaml(config: dict) -> dict: datasets_out: dict = {} for ds_name, ds_cfg in (config.get("datasets") or {}).items(): - if ds_name not in {"daily", "adj_factor", "realtime", "minute", "full_minute", "financial"}: + if ds_name not in {"daily", "adj_factor", "realtime", "minute", "depth5", "full_minute", "financial"}: continue if not isinstance(ds_cfg, dict): continue diff --git a/backend/app/data_providers/custom/provider.py b/backend/app/data_providers/custom/provider.py index a1d63466..47ed8955 100644 --- a/backend/app/data_providers/custom/provider.py +++ b/backend/app/data_providers/custom/provider.py @@ -34,6 +34,7 @@ _REQUIRED = { "minute": {"symbol", "datetime", "open", "high", "low", "close", "volume"}, # full_minute (全量分钟) 与 minute 同形: 当日窗口批量拉取, 字段映射一致 "full_minute": {"symbol", "datetime", "open", "high", "low", "close", "volume"}, + "depth5": {"symbol", "asks", "bids"}, # minute/full_minute 的 amount(成交额)按契约可空 (docs/plugin-development.md: # 无法提供可靠成交额时应返回 null 不得伪造, 均价线降级显示 —), 不列为必填 — # 否则缺成交额的源整个被 validate 拒绝, 保存后源从列表消失 @@ -121,7 +122,7 @@ class GenericHTTPProvider: errors.append(f"{dataset}: pct_unit 仅用于 realtime 数据集") elif cfg.pct_unit not in ("percent", "decimal"): errors.append(f"{dataset}: pct_unit 必须是 percent 或 decimal") - if dataset != "realtime": + if dataset not in {"realtime", "depth5"}: request_params = [cfg.symbols_param, cfg.start_param, cfg.end_param] if dataset in {"minute", "full_minute"}: request_params.extend( @@ -251,6 +252,36 @@ class GenericHTTPProvider: return [] return df.to_dicts() + def get_depth_batch(self, symbols: list[str]) -> dict[str, dict]: + """Return normalized five-level order books keyed by symbol.""" + cfg = self._dataset("depth5") + result: dict[str, dict] = {} + chunks = chunked(symbols, cfg.batch) + for i, chunk in enumerate(chunks): + sleep_between_batches(i, cfg.rpm) + try: + rows = self._request_rows_retry(cfg, chunk) + except Exception as e: # noqa: BLE001 + logger.warning( + "custom depth5: batch %d/%d failed (%d symbols), skipped: %s", + i + 1, len(chunks), len(chunk), e, + ) + continue + df = self._mapped_frame(cfg, rows) + for row in df.to_dicts(): + symbol = row.get("symbol") + if not symbol: + continue + asks = _normalize_depth_levels(row.get("asks")) + bids = _normalize_depth_levels(row.get("bids")) + result[str(symbol)] = { + "asks": asks, + "bids": bids, + "ask_volumes": [level[1] for level in asks], + "bid_volumes": [level[1] for level in bids], + } + return result + def get_minute( self, symbols: list[str], @@ -403,6 +434,8 @@ class GenericHTTPProvider: start_time = end_time - timedelta(days=7) if dataset == "realtime": rows = self._request_rows(cfg) + elif dataset == "depth5": + rows = self._request_rows(cfg, symbols=test_symbols) elif dataset in {"minute", "full_minute"}: override: dict[str, Any] = {} if cfg.asset_type_param: @@ -503,6 +536,20 @@ class GenericHTTPProvider: return {}, {} +def _normalize_depth_levels(value: Any) -> list[list[float]]: + if not isinstance(value, (list, tuple)): + return [] + levels: list[list[float]] = [] + for item in value: + if not isinstance(item, (list, tuple)) or len(item) < 2: + continue + try: + levels.append([float(item[0]), float(item[1])]) + except (TypeError, ValueError): + continue + return levels + + def _token_from_env(name: str | None) -> str | None: if not name: return None diff --git a/backend/tests/test_custom_depth_provider.py b/backend/tests/test_custom_depth_provider.py index 11372550..2620a7a9 100644 --- a/backend/tests/test_custom_depth_provider.py +++ b/backend/tests/test_custom_depth_provider.py @@ -132,3 +132,54 @@ def test_invalid_custom_depth_contract_fails_closed(monkeypatch): assert _service()._call_depth_batch(["A"]) == {} get_provider.assert_not_called() + + +def test_custom_depth_provider_maps_asks_and_bids(monkeypatch): + from app.data_providers.custom.config import config_from_dict + from app.data_providers.custom.provider import GenericHTTPProvider + + provider = GenericHTTPProvider(config_from_dict({ + "name": "eltdx", + "datasets": { + "depth5": { + "url": "http://example.test/depth5", + "response_path": "data", + "field_map": { + "symbol": "symbol", "asks": "asks", "bids": "bids", + }, + }, + }, + })) + monkeypatch.setattr(provider, "_request_rows", lambda cfg, **kwargs: [{ + "symbol": "000001.SZ", + "asks": [[11.33, 1256]], + "bids": [[11.32, 857]], + }]) + + assert provider.get_depth_batch(["000001.SZ"]) == { + "000001.SZ": { + "ask_volumes": [1256], + "bid_volumes": [857], + "asks": [[11.33, 1256]], + "bids": [[11.32, 857]], + }, + } + + +def test_depth5_config_is_accepted_and_does_not_require_dates(monkeypatch): + from app.data_providers.custom.config import config_from_dict + from app.data_providers.custom.provider import GenericHTTPProvider + + provider = GenericHTTPProvider(config_from_dict({ + "name": "eltdx", + "datasets": { + "depth5": { + "url": "http://example.test/depth5", + "field_map": {"symbol": "symbol", "asks": "asks", "bids": "bids"}, + }, + }, + })) + + assert provider.validate() == [] + monkeypatch.setattr(provider, "_request_rows", lambda cfg, **kwargs: []) + assert provider.test_dataset("depth5")["dataset"] == "depth5"