From b2c5d8fd79a6c9c74162f339b5dcb8146e78d444 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Tue, 7 Jul 2026 11:57:16 +0800 Subject: [PATCH] =?UTF-8?q?feat(data):=20read=5Fparquet=5F15min=20+=20trad?= =?UTF-8?q?er=20data=5Fsource(qfq/raw=E5=8F=8C=E6=BA=90)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - datareader: +read_parquet_15min(sh/sz前缀+15min.parquet), get_database lazy(去tzlocal等依赖) - data_source: iter_bars cross-section yield(date,{symbol:Bar}), raw首版fallback qfq+warning(spec§17) - 本机 mock _read_fn 测调度逻辑, read_parquet_15min 容器冒烟 4 tests passed. --- sanguo_data/datareader.py | 32 +++++++++++++++++- sanguo_trader/data_source.py | 57 +++++++++++++++++++++++++++++++ tests/trader/test_data_source.py | 58 ++++++++++++++++++++++++++++++++ 3 files changed, 146 insertions(+), 1 deletion(-) create mode 100644 sanguo_trader/data_source.py create mode 100644 tests/trader/test_data_source.py diff --git a/sanguo_data/datareader.py b/sanguo_data/datareader.py index b7ab15b..ff5d6bf 100644 --- a/sanguo_data/datareader.py +++ b/sanguo_data/datareader.py @@ -12,7 +12,6 @@ import pandas as pd from datetime import datetime from vnpy.trader.object import BarData from vnpy.trader.constant import Exchange, Interval -from vnpy.trader.database import get_database from vnpy.trader.setting import SETTINGS def read_parquet_daily(symbol: str, start: str, end: str, cfg) -> list[BarData]: @@ -56,6 +55,7 @@ def guess_exchange(symbol: str) -> Exchange: def read_db_daily(symbol: str, start: str, end: str, cfg) -> list[BarData]: + from vnpy.trader.database import get_database # lazy：避免模块 import 依赖数据库驱动 # Configure vnpy database SETTINGS before calling get_database() SETTINGS["database.name"] = "sqlite" SETTINGS["database.database"] = cfg.data_paths["vnpy_db"] @@ -70,3 +70,33 @@ def read_db_daily(symbol: str, start: str, end: str, cfg) -> list[BarData]: start=start_dt, end=end_dt, ) + + +def read_parquet_15min(symbol: str, start: str, end: str, cfg) -> list[BarData]: + """读 15min parquet（NAS /volume1/stock/minute_kline/15min/sh{symbol}_15min.parquet）。""" + minute_dir = Path(cfg.data_paths["minute_15_dir"]) + start_dt = datetime.strptime(start, "%Y-%m-%d") + end_dt = datetime.strptime(end, "%Y-%m-%d") + prefix = "sh" if guess_exchange(symbol) == Exchange.SSE else "sz" + f = minute_dir / f"{prefix}{symbol}_15min.parquet" + if not f.exists(): + return [] + df = pd.read_parquet(f) + time_col = "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]) + if start_dt <= d <= end_dt: + bars.append(BarData( + symbol=symbol, + exchange=guess_exchange(symbol), + datetime=d.to_pydatetime(), + interval=Interval.MINUTE, + open_price=float(row["open"]), + high_price=float(row["high"]), + low_price=float(row["low"]), + close_price=float(row["close"]), + volume=float(row["volume"]), + gateway_name="DATA", + )) + return bars diff --git a/sanguo_trader/data_source.py b/sanguo_trader/data_source.py new file mode 100644 index 0000000..e9791af --- /dev/null +++ b/sanguo_trader/data_source.py @@ -0,0 +1,57 @@ +"""模拟盘行情统一接口（qfq/raw 双源，spec §3.3 / §5）。 + +_read_fn 内 lazy import datareader，避免模块级依赖 vnpy 链（tzlocal 等）， +本机无 vnpy 完整依赖时仍可 import + 单测（mock _read_fn）。 +""" +import logging +from collections.abc import Iterator + +logger = logging.getLogger(__name__) + + +def _read_fn(interval: str): + """按 interval 返回读取函数（lazy import）。""" + if interval == "d": + from sanguo_data.datareader import read_parquet_daily + return read_parquet_daily + if interval == "15m": + from sanguo_data.datareader import read_parquet_15min + return read_parquet_15min + raise ValueError(f"不支持的 interval: {interval}（首版仅 d / 15m）") + + +def iter_bars( + symbols: list[str], + start: str, + end: str, + interval: str, + adjust: str = "qfq", + cfg=None, +) -> Iterator[tuple]: + """按日期 cross-section yield (date, {symbol: BarData})。 + + raw 首版 fallback qfq + warning（NAS 暂无 raw parquet，spec §17 开放项）。 + """ + if adjust == "raw": + logger.warning( + "raw 模式首版 fallback qfq（NAS 暂无 raw parquet，spec §17 开放项）" + ) + read_fn = _read_fn(interval) + by_date: dict = {} + for sym in symbols: + for bar in read_fn(sym, start, end, cfg): + dt = bar.datetime + key = dt.date() if hasattr(dt, "date") else dt + by_date.setdefault(key, {})[sym] = bar + for date in sorted(by_date.keys()): + yield date, by_date[date] + + +def fetch_day(symbol: str, date: str, interval: str, + adjust: str = "qfq", cfg=None): + """实走模式拉当日 bar（C-S3 用）。""" + if adjust == "raw": + logger.warning("raw fallback qfq（spec §17）") + read_fn = _read_fn(interval) + bars = read_fn(symbol, date, date, cfg) + return bars[-1] if bars else None diff --git a/tests/trader/test_data_source.py b/tests/trader/test_data_source.py new file mode 100644 index 0000000..703fbf0 --- /dev/null +++ b/tests/trader/test_data_source.py @@ -0,0 +1,58 @@ +"""data_source 测试（mock _read_fn，不依赖 vnpy 链，spec §5/§3.3）。 + +read_parquet_15min 的真测试依赖 vnpy BarData + NAS parquet，本机无完整依赖， +在容器内冒烟（spec §17）；本文件只测 iter_bars 调度逻辑。 +""" +import logging +from datetime import datetime +from types import SimpleNamespace + +import pytest + +from sanguo_trader.data_source import iter_bars + + +def _mock_bar(sym: str, date_str: str, close: float): + return SimpleNamespace( + symbol=sym, + datetime=datetime.strptime(date_str, "%Y-%m-%d"), + close_price=close, + ) + + +def test_iter_bars_cross_section_multi_symbol(monkeypatch): + def mock_read(sym, start, end, cfg): + return [_mock_bar(sym, "2024-01-02", 10.5 if sym == "600000" else 15.5)] + monkeypatch.setattr("sanguo_trader.data_source._read_fn", lambda iv: mock_read) + sections = list(iter_bars(["600000", "000001"], "2024-01-01", "2024-01-31", "d")) + assert len(sections) == 1 + _date, d = sections[0] + assert "600000" in d and "000001" in d + assert d["600000"].close_price == 10.5 + + +def test_iter_bars_raw_fallback_warning(monkeypatch, caplog): + monkeypatch.setattr( + "sanguo_trader.data_source._read_fn", + lambda iv: lambda s, st, e, c: [], + ) + with caplog.at_level(logging.WARNING): + list(iter_bars(["600000"], "2024-01-01", "2024-01-31", "d", adjust="raw")) + assert "raw" in caplog.text + + +def test_unsupported_interval_rejected(): + with pytest.raises(ValueError): + # _read_fn 在生成器首次 next 时才执行，用 list 触发 + list(iter_bars(["600000"], "2024-01-01", "2024-01-31", "5m")) + + +def test_fetch_day_returns_last_bar(monkeypatch): + from sanguo_trader.data_source import fetch_day + monkeypatch.setattr( + "sanguo_trader.data_source._read_fn", + lambda iv: lambda s, st, e, c: [_mock_bar(s, "2024-01-02", 10.0), + _mock_bar(s, "2024-01-02", 10.5)], + ) + bar = fetch_day("600000", "2024-01-02", "d") + assert bar.close_price == 10.5 # 取最后一个