fix(数据源): 支持自定义五挡盘口路由 (#479)

Co-authored-by: shy3130 <415333856@qq.com>

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

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

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

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"