feat(data): 15min双源集成(data_source路由 + baostock格式适配)

- _resolve_dir_key: 15min支持raw/qfq双源(移除raw15min抛错)
- _check_adjust_cfg: 按interval查dir(minute_15_raw_dir/qfq_dir)
- read_parquet_15min: datetime列优先(baostock时分,旧date兼容)
- config: 加 minute_15_qfq_dir/minute_15_raw_dir
- test: 更新raw15min路由断言(6 passed)

实测容器: 600000 15min qfq 336bars close6.1977 / raw 336bars close6.6200,
datetime时分正确(09:45:00). 15min双源分期项落地.

config/data_platform.yaml

@@ -4,6 +4,8 @@ data_paths:
   raw_dir: /volume1/stock/A股数据/日线数据/raw
   qfq_dir: /volume1/stock/A股数据/日线数据/qfq
   minute_15_dir: /volume1/stock/minute_kline/15min
+  minute_15_qfq_dir: /volume1/stock/minute_kline/15min_qfq
+  minute_15_raw_dir: /volume1/stock/minute_kline/15min_raw
   vnpy_db: /volume1/stock/sanguo_vnpy/data/quant_trading.db
   stock_list: /volume1/stock/A股数据/stock_info/stock_basic_info_raw_20260326_113530.csv
 

sanguo_data/datareader.py

@@ -83,7 +83,9 @@ def read_parquet_15min(symbol: str, start: str, end: str, cfg, dir_key: str = "m
     if not f.exists():
         return []
     df = pd.read_parquet(f)
-    time_col = "date" if "date" in df.columns else df.columns[0]
+    # baostock 15min 有完整时分 datetime 列；旧数据用 date 列
+    time_col = ("datetime" if "datetime" in df.columns
+                else ("date" if "date" in df.columns else df.columns[0]))
     bars: list[BarData] = []
     for _, row in df.iterrows():
         d = pd.to_datetime(row[time_col])

sanguo_trader/data_source.py

@@ -26,29 +26,32 @@ def _read_fn(interval: str):
 
 
 def _resolve_dir_key(adjust: str, interval: str) -> str:
-    """adjust → cfg.data_paths 的目录 key。
-
-    raw 仅日线（raw 15min 待分期项）；qfq/默认按 interval 选。
-    """
-    if adjust == "raw":
-        if interval != "d":
-            raise ValueError(
-                f"raw 模式暂仅支持日线（interval='d'），got '{interval}'"
-                "（raw 15min 待分期项）"
-            )
-        return "raw_dir"
-    if adjust == "qfq":
-        return "qfq_dir"   # 干净 qfq（daily_dir 是 mixed，仅 backtest 兼容）
-    return "daily_dir" if interval == "d" else "minute_15_dir"
+    """adjust+interval → cfg.data_paths 目录 key。日线/15min 均支持 raw/qfq 双源。"""
+    if interval == "d":
+        if adjust == "raw":
+            return "raw_dir"
+        if adjust == "qfq":
+            return "qfq_dir"   # 干净 qfq（daily_dir 是 mixed，仅 backtest 兼容）
+        return "daily_dir"
+    if interval in ("15m", "15min"):
+        if adjust == "raw":
+            return "minute_15_raw_dir"
+        if adjust == "qfq":
+            return "minute_15_qfq_dir"
+        return "minute_15_dir"
+    raise ValueError(f"不支持的 interval: {interval}（仅 d / 15m）")
 
 
-def _check_adjust_cfg(adjust: str, cfg) -> None:
+def _check_adjust_cfg(adjust: str, cfg, interval: str = "d") -> None:
     """raw/qfq 需对应 dir 配置，缺失明确报错（不静默 fallback，避免混源）。"""
-    need = {"raw": "raw_dir", "qfq": "qfq_dir"}.get(adjust)
+    need = {
+        ("raw", "d"): "raw_dir", ("qfq", "d"): "qfq_dir",
+        ("raw", "15m"): "minute_15_raw_dir", ("qfq", "15m"): "minute_15_qfq_dir",
+    }.get((adjust, interval))
     if need and cfg and need not in getattr(cfg, "data_paths", {}):
         raise ValueError(
             f"{adjust} 模式需 cfg.data_paths['{need}']（未配置；"
-            f"先用 raw_redownload.py --adjust {adjust} 生成 parquet）"
+            f"先用对应下载脚本生成 parquet）"
         )
 
 
@@ -62,7 +65,7 @@ def iter_bars(
 ) -> Iterator[tuple]:
     """按日期 cross-section yield (date, {symbol: BarData})。"""
     dir_key = _resolve_dir_key(adjust, interval)
-    _check_adjust_cfg(adjust, cfg)
+    _check_adjust_cfg(adjust, cfg, interval)
     read_fn = _read_fn(interval)
     by_date: dict = {}
     for sym in symbols:
@@ -78,7 +81,7 @@ def fetch_day(symbol: str, date: str, interval: str,
               adjust: str = "qfq", cfg=None):
     """实走模式拉当日 bar（C-S3 用）。"""
     dir_key = _resolve_dir_key(adjust, interval)
-    _check_adjust_cfg(adjust, cfg)
+    _check_adjust_cfg(adjust, cfg, interval)
     read_fn = _read_fn(interval)
     bars = read_fn(symbol, date, date, cfg, dir_key)
     return bars[-1] if bars else None

tests/trader/test_data_source.py

@@ -65,15 +65,20 @@ def test_iter_bars_raw_missing_dir_raises(monkeypatch):
                        adjust="raw", cfg=_cfg({})))
 
 
-def test_iter_bars_raw_15min_unsupported(monkeypatch):
-    """raw 仅日线；15min+raw 报错（raw 15min 待分期项）。"""
+def test_iter_bars_raw_15min_routes_dedicated_dir(monkeypatch):
+    """raw 15min 路由 minute_15_raw_dir（baostock 双源，分期项已落地）。"""
+    seen = []
     monkeypatch.setattr(
         "sanguo_trader.data_source._read_fn",
-        lambda iv: lambda s, st, e, c, dir_key: [],
+        lambda iv: lambda s, st, e, c, dir_key: seen.append(dir_key) or [],
     )
-    with pytest.raises(ValueError, match="日线"):
+    list(iter_bars(["600000"], "2024-01-01", "2024-01-31", "15m",
+                   adjust="raw", cfg=_cfg({"minute_15_raw_dir": "/x"})))
+    assert seen == ["minute_15_raw_dir"]
+    # 缺 minute_15_raw_dir 配置 → 明确报错（不 fallback，防混源）
+    with pytest.raises(ValueError, match="minute_15_raw_dir"):
         list(iter_bars(["600000"], "2024-01-01", "2024-01-31", "15m",
-                       adjust="raw", cfg=_cfg({"raw_dir": "/x"})))
+                       adjust="raw", cfg=_cfg({})))
 
 
 def test_unsupported_interval_rejected():