diff --git a/sanguo_api/routes.py b/sanguo_api/routes.py index e90b0d7..b508749 100644 --- a/sanguo_api/routes.py +++ b/sanguo_api/routes.py @@ -78,7 +78,8 @@ async def submit_cta(req: CtaBacktestRequest): cfg=None, benchmark=req.benchmark, capital=req.capital, - position_pct=req.position_pct + position_pct=req.position_pct, + interval=req.interval, ) return {"task_id": tid} diff --git a/sanguo_api/schemas.py b/sanguo_api/schemas.py index 54b6ea0..ad7941a 100644 --- a/sanguo_api/schemas.py +++ b/sanguo_api/schemas.py @@ -14,6 +14,8 @@ class CtaBacktestRequest(BaseModel): benchmark: str = "hs300" capital: float = 1_000_000 position_pct: float = 0.95 + # K 线周期："d"=日线(默认) / "5m" / "15m"。5m/15m 走 AShareBacktestingEngine 适配。 + interval: str = "d" # A 股费用参数（可选，前端先不暴露，给默认值） commission_rate: float = 0.00025 # 万 2.5 min_commission: float = 5.0 # 最低 5 元 diff --git a/sanguo_backtest/ashare_engine.py b/sanguo_backtest/ashare_engine.py index 96cd251..7a700d7 100644 --- a/sanguo_backtest/ashare_engine.py +++ b/sanguo_backtest/ashare_engine.py @@ -6,13 +6,15 @@ vnpy 源码零修改，全部覆写在本文件。 - AShareBacktestingEngine： - send_order 覆写 → 做空拦截（SHORT+OPEN 拒单）+ 定寸重算 volume - update_daily_close 覆写 → 工厂换 AShareDailyResult（父类在 :647 实例化 DailyResult） + - load_data 覆写 → 支持 5m/15m 周期（vnpy Interval enum 不认 '5m'/'15m'） """ from __future__ import annotations import logging +from datetime import datetime from vnpy_ctastrategy.backtesting import BacktestingEngine, DailyResult -from vnpy.trader.constant import Direction, Offset +from vnpy.trader.constant import Direction, Offset, Interval, Exchange logger = logging.getLogger(__name__) @@ -123,6 +125,11 @@ class AShareBacktestingEngine(BacktestingEngine): self.stamp_duty_rate: float = 0.0005 # 卖方 0.05% self.transfer_fee_rate: float = 0.00001 # 沪市 0.001% self.is_sse: bool = False + # 5m/15m 适配：vnpy Interval enum 不认 '5m'/'15m'，cta_engine 把 engine.interval + # 映射成 Interval.MINUTE 让父类 set_parameters 校验通过，真实 DB interval 字符串 + # 存此字段供 load_data 自定义路径使用。默认 "d" 走 vnpy 原生日线路径。 + self.raw_interval: str = "d" + self.sqlite_db_path: str | None = None # 由 cta_engine 注入（_dcfg.data_paths["vnpy_db"]） def send_order( self, @@ -174,3 +181,93 @@ class AShareBacktestingEngine(BacktestingEngine): transfer_fee_rate=self.transfer_fee_rate, is_sse=self.is_sse, ) + + def load_data(self) -> None: + """覆写父类：支持 5m/15m 周期（绕过 vnpy Interval enum 限制）。 + + vnpy Interval enum 只有 MINUTE('1m')/HOUR('1h')/DAILY('d') 等，不认 '5m'/'15m'。 + 父类 load_data 调 ``INTERVAL_DELTA_MAP[self.interval]`` 和 + ``load_bar_data(..., self.interval, ...)`` 都依赖 enum，传 '15m' 会 ValueError。 + + 适配思路：self.interval 已被父类 set_parameters 映射成 Interval.MINUTE（enum + 校验通过），真实 DB interval 存 self.raw_interval。当 raw_interval 是 5m/15m + 时，直接 sqlite 查 dbbardata 表，把每行转 BarData(interval=Interval.MINUTE)， + 绕过 vnpy database_manager 的 enum 限制；其余周期（d/1m 等）走 vnpy 原生路径。 + """ + if self.raw_interval not in ("5m", "15m"): + super().load_data() + return + self._load_intraday_data() + + def _load_intraday_data(self) -> None: + """5m/15m 直查 SQLite → BarData(MINUTE)，绕过 vnpy enum 限制。""" + from vnpy.trader.object import BarData + import sqlite3 + + self.output(f"开始加载 {self.raw_interval} 历史数据（ashare 适配）") + + if not self.end: + self.end = datetime.now() + if self.start >= self.end: + self.output("起始日期必须小于结束日期") + return + + db_path = self.sqlite_db_path + if not db_path: + try: + from vnpy.trader.setting import SETTINGS + db_path = SETTINGS.get("database.database") + except Exception: + db_path = None + if not db_path: + raise RuntimeError( + "5m/15m 回测需要 sqlite_db_path（或 SETTINGS['database.database']），" + "cta_engine 应在 load_data 前注入。" + ) + + symbol, exchange_str = self.vt_symbol.split(".") + # peewee DateTimeField 存的是 ISO 字符串；start/end 用 datetime 比较即可 + # （SQLite 会把参数转成可比较的字符串形式）。 + conn = sqlite3.connect(db_path) + try: + cur = conn.execute( + "SELECT datetime, volume, turnover, open_interest, " + "open_price, high_price, low_price, close_price " + "FROM dbbardata " + "WHERE symbol=? AND exchange=? AND interval=? " + "AND datetime>=? AND datetime<=? " + "ORDER BY datetime", + (symbol, exchange_str, self.raw_interval, self.start, self.end), + ) + rows = cur.fetchall() + finally: + conn.close() + + exchange = Exchange(exchange_str) + bars: list[BarData] = [] + for dt, vol, turnover, oi, o, h, l, c in rows: + if isinstance(dt, str): + try: + dt = datetime.fromisoformat(dt) + except ValueError: + continue + # vnpy_sqlite save 时 convert_tz 改成 UTC，回测时按本地时间跑即可 + # （日线回测也是直接读 DB datetime，行为一致）。 + bars.append(BarData( + symbol=symbol, + exchange=exchange, + datetime=dt, + interval=Interval.MINUTE, # 5m/15m 不在 enum，统一标 MINUTE + volume=float(vol or 0), + turnover=float(turnover or 0), + open_interest=float(oi or 0), + open_price=float(o or 0), + high_price=float(h or 0), + low_price=float(l or 0), + close_price=float(c or 0), + gateway_name="sqlite", + )) + + self.history_data.clear() + self.history_data.extend(bars) + self.output(f"历史数据加载完成，数据量：{len(self.history_data)}") diff --git a/sanguo_backtest/cta_engine.py b/sanguo_backtest/cta_engine.py index 03c841f..0f854b9 100644 --- a/sanguo_backtest/cta_engine.py +++ b/sanguo_backtest/cta_engine.py @@ -67,6 +67,27 @@ def guess_exchange(symbol: str) -> Exchange: return Exchange("SSE") +def _read_vt_setting_db() -> str | None: + """从 vnpy 原生 vt_setting.json 直接读 database.database 路径。 + + vnpy SETTINGS 字典会被 datareader.read_index_daily 等代码覆写为 yaml 的 + NAS Linux 路径（Windows VPS 上不存在但有 0 字节空文件，os.path.exists 误判）， + 导致连续多次回测时第二次起路径污染。vt_setting.json 是机器本地配置，权威。 + """ + try: + import json + from pathlib import Path + # vnpy 约定：~/.vntrader/vt_setting.json + home = Path.home() / ".vntrader" / "vt_setting.json" + if not home.exists(): + return None + with open(home, "r", encoding="utf-8") as f: + data = json.load(f) + return data.get("database.database") + except Exception: + return None + + def run_cta_backtest( strategy_class, symbol: str, @@ -83,6 +104,7 @@ def run_cta_backtest( min_commission: float = 5.0, stamp_duty_rate: float = 0.0005, transfer_fee_rate: float = 0.00001, + interval: str = "d", ) -> BacktestResult: """ Run CTA strategy backtest using AShareBacktestingEngine (vnpy 子类化). @@ -104,6 +126,9 @@ def run_cta_backtest( min_commission: 单笔最低佣金（默认 5 元） stamp_duty_rate: 印花税率卖方（默认 0.0005） transfer_fee_rate: 过户费率沪市（默认 0.00001） + interval: K 线周期，"d"=日线(默认) / "5m" / "15m"。 + 5m/15m 走 AShareBacktestingEngine 适配层（直查 dbbardata interval='5m'/'15m'， + 绕过 vnpy Interval enum 不认非标周期的限制）。 Returns: BacktestResult: Result object with backtest statistics and status @@ -127,10 +152,18 @@ def run_cta_backtest( # Create and configure A-share backtesting engine engine = AShareBacktestingEngine() + # 周期映射：vnpy Interval enum 只有 '1m'/'1h'/'d' 等，不认 '5m'/'15m'。 + # 5m/15m 时把 enum 传 MINUTE 让父类 set_parameters 校验通过，真实 DB interval + # 字符串存 engine.raw_interval 给 AShareBacktestingEngine.load_data 自定义路径用。 + if interval in ("5m", "15m"): + engine_interval = "1m" # Interval.MINUTE.value + else: + engine_interval = interval # "d" / "1m" / "1h" 等 vnpy 原生支持的值 + # Set parameters with A-share specific values engine.set_parameters( vt_symbol=vt_symbol, - interval="d", # Interval.DAILY.value — vnpy enum uses "d" not "1d" + interval=engine_interval, start=start_dt, end=end_dt, rate=commission_rate, # 佣金率（AShareDailyResult 用自身 commission_rate，此处仅保持一致） @@ -139,6 +172,9 @@ def run_cta_backtest( pricetick=0.01, # Minimum price tick (0.01 yuan for A-shares) capital=capital, # Starting capital ) + # raw_interval 在 5m/15m 时触发 ashare_engine 适配路径；其它周期与 engine.interval + # 一致，走 vnpy 原生 load_data。 + engine.raw_interval = interval # A 股适配参数（定寸 + 费用） engine.position_pct = position_pct @@ -155,13 +191,32 @@ def run_cta_backtest( # doesn't inherit main-process SETTINGS, so set before engine.load_data. # _dcfg is also reused by the metrics branch (benchmark data_paths) since the # cfg param can be None when called via the API. + # + # 路径解析优先级：vnpy 原生 vt_setting.json(机器相关，正确) > + # yaml data_paths.vnpy_db（模板可能是 NAS Linux 路径，Windows VPS 上不存在）。 + # 任一不存在时回退到另一个，避免 VPS 上 yaml 写 NAS 路径导致回测加载 0 数据。 + # + # 不能直接用 SETTINGS.get("database.database")：它会被 datareader.read_index_daily + # （基准加载）覆写为 yaml NAS 路径，污染后续调用。这里每次重新读 vt_setting.json + # 原始值（machine-truth），不依赖被污染的 SETTINGS 缓存。 _dcfg = None try: from vnpy.trader.setting import SETTINGS from sanguo_data.config import load_config, find_config_path _dcfg = load_config(find_config_path()) SETTINGS["database.name"] = "sqlite" - SETTINGS["database.database"] = _dcfg.data_paths["vnpy_db"] + _vt_db = _read_vt_setting_db() # 从 vt_setting.json 直接读（不依赖 SETTINGS dict） + _yaml_db = _dcfg.data_paths.get("vnpy_db") + if _vt_db and os.path.exists(_vt_db): + _resolved_db = _vt_db + elif _yaml_db and os.path.exists(_yaml_db): + _resolved_db = _yaml_db + else: + # 都不存在时保留 vnpy SETTINGS（让下游报出真实错误，而不是被 yaml 覆盖） + _resolved_db = _vt_db or _yaml_db + SETTINGS["database.database"] = _resolved_db + # 5m/15m 适配：ashare_engine._load_intraday_data 直查 SQLite 需要 DB 物理路径。 + engine.sqlite_db_path = _resolved_db except Exception as e: logging.warning("vnpy 数据库配置加载失败（回测可能无法加载历史数据）: %s", e) diff --git a/sanguo_orchestrator/runner.py b/sanguo_orchestrator/runner.py index 14b8dbb..8a74cc2 100644 --- a/sanguo_orchestrator/runner.py +++ b/sanguo_orchestrator/runner.py @@ -32,7 +32,8 @@ class Orchestrator: async def submit_cta(self, strategy_class, symbol: str, params: dict, start: str, end: str, cfg, benchmark: str = "hs300", - capital: float = 1_000_000, position_pct: float = 0.95) -> str: + capital: float = 1_000_000, position_pct: float = 0.95, + interval: str = "d") -> str: """Submit a CTA backtesting task asynchronously""" # Stable uuid up front → reused as the persisted DB task_id, so runner-id == # DB task_id (durable across restarts; previously used id(params) memory addr). @@ -48,6 +49,7 @@ class Orchestrator: benchmark=benchmark, capital=capital, position_pct=position_pct, + interval=interval, ) await self._notify_stage(task_id, "排队中") @@ -55,7 +57,7 @@ class Orchestrator: fut: Future = self.pool.submit_work( task_id, _cta_worker, spec["strategy_class"], spec["symbol"], spec["params"], spec["start"], spec["end"], spec["cfg"], spec["benchmark"], - self.db_path, task_id, spec["capital"], spec["position_pct"] + self.db_path, task_id, spec["capital"], spec["position_pct"], spec["interval"] ) task = self.pool.get_task(task_id) @@ -194,10 +196,10 @@ class Orchestrator: # Module-level worker functions (must be top-level for ProcessPoolExecutor pickle) -def _cta_worker(strategy_class, symbol: str, params: dict, start: str, end: str, cfg, benchmark: str, db_path: str, task_id: str, capital: float = 1_000_000, position_pct: float = 0.95) -> any: +def _cta_worker(strategy_class, symbol: str, params: dict, start: str, end: str, cfg, benchmark: str, db_path: str, task_id: str, capital: float = 1_000_000, position_pct: float = 0.95, interval: str = "d") -> any: """Worker for CTA backtest (lazy import, spawn-friendly)""" from sanguo_backtest.cta_engine import run_cta_backtest - return run_cta_backtest(strategy_class, symbol, params, start, end, cfg, db_path, benchmark=benchmark, task_id=task_id, capital=capital, position_pct=position_pct) + return run_cta_backtest(strategy_class, symbol, params, start, end, cfg, db_path, benchmark=benchmark, task_id=task_id, capital=capital, position_pct=position_pct, interval=interval) def _opt_worker(strategy_class, symbol: str, grid: dict, start: str, end: str, cfg, db_path: str, task_id: str) -> any: