"""全天候策略回测入口。 用法(Mac 默认 baostock;VPS Windows / miniQMT 已连用 miniqmt): # Mac 默认 baostock(跨平台,不依赖 miniQMT 客户端) python -m sanguo_portfolio.runner_backtest \\ --start 2020-01-01 --end 2024-12-31 --cash 1000000 # VPS miniQMT(实盘/精准 xtquant) python -m sanguo_portfolio.runner_backtest --provider miniqmt \\ --start 2020-01-01 --end 2024-12-31 --cash 1000000 JSON 输出(供 SSH 捕获,前端 MVP 用): python -m sanguo_portfolio.runner_backtest --json \\ --start 2024-01-01 --end 2024-02-29 --cash 1000000 Mac 跑 baostock 默认链路;miniQMT 链路仍保留(实盘 runner_live 用)。 """ from __future__ import annotations # ENV GUARD 必须早于任何 bullet_trade import # bullet_trade __init__ 加载时 _create_provider() 读 DEFAULT_DATA_PROVIDER 创建默认 provider: # miniqmt → import xtquant(周六休市 miniQMT 客户端不响应→卡死) # jqdata → import jqdatasdk(用户铁律不装→ModuleNotFoundError) # 方案: ENV 设 jqdata + 预插 mock jqdatasdk, 让 import 走 jqdata 分支拿 mock 不崩不卡; # 真实 provider 由 set_data_provider 运行时注入覆盖(local/baostock/miniqmt)。 import os import sys as _sys from unittest.mock import MagicMock as _MagicMock os.environ.setdefault("DEFAULT_DATA_PROVIDER", "jqdata") if "jqdatasdk" not in _sys.modules: _m = _MagicMock() # jqdata.py 用 @jq.utils.assert_auth 装饰器;MagicMock 的 assert_* 前缀被保护→AttributeError _m.utils.assert_auth = lambda func: func # passthrough 装饰器 _sys.modules["jqdatasdk"] = _m import argparse import json import logging from typing import Any, Dict logger = logging.getLogger(__name__) def parse_args() -> argparse.Namespace: p = argparse.ArgumentParser(description="sanguo_portfolio 组合回测") p.add_argument("--start", default="2020-01-01", help="回测开始日期 YYYY-MM-DD") p.add_argument("--end", default="2024-12-31", help="回测结束日期 YYYY-MM-DD") p.add_argument("--cash", type=float, default=1_000_000.0, help="初始资金(元)") p.add_argument("--benchmark", default="000300.XSHG", help="基准代码") p.add_argument("--max-pool", type=int, default=0, help="限制选股池前N只(0=不限,MVP验证用)") p.add_argument("--frequency", default="day", help="回测频率 day/minute") # A 股费用(对齐个股回测;BulletTrade 默认仅印花税千1+最低5元,这里显式可配) p.add_argument("--commission-rate", type=float, default=0.0003, help="佣金率双边(万3=0.0003)") p.add_argument("--stamp-duty-rate", type=float, default=0.001, help="印花税率卖出(千1=0.001)") p.add_argument("--min-commission", type=float, default=5.0, help="单笔最低佣金(元)") p.add_argument("--slippage", type=float, default=0.0, help="滑点比率(万10=0.001,0=不加)") p.add_argument( "--strategy", default="all_weather", choices=["all_weather", "momentum_timing", "value_selection", "small_cap"], help="策略: all_weather(全天候轮动) / momentum_timing(牛熊分界+取强舍弱+均线动量) / value_selection(价值精选6条月度调仓) / small_cap(小市值20只轮动,无对冲)", ) p.add_argument( "--provider", default="local", choices=["local", "baostock", "miniqmt", "unified"], help="数据 provider:local(parquet,旧) / baostock(Mac 跨平台) / miniqmt(VPS 实盘) / unified(方案A 权威层)", ) p.add_argument( "--provider-config", default="{}", help="provider 配置 JSON 字符串,如 '{\"data_dir\":\"D:/xtdata\"}'", ) p.add_argument( "--result-file", default="docs/portfolio_backtest_result.md", help="结果输出文件(.md)", ) p.add_argument( "--json", action="store_true", help="JSON 模式:print(json.dumps(result)) 到 stdout,供 SSH 捕获", ) return p.parse_args() def build_provider(provider_name: str, config_str: str) -> Any: """构造 provider 实例。 Args: provider_name: "baostock"(Mac 默认) 或 "miniqmt"(VPS 实盘) config_str: provider 配置 JSON 字符串 """ import json from .providers import ( BaostockProvider, LocalParquetProvider, LocalUnifiedProvider, SanguoMiniQmtProvider, ) cfg: Dict[str, Any] = {} if config_str and config_str != "{}": try: cfg = json.loads(config_str) except Exception as exc: logger.warning("provider-config 解析失败,用默认: %s", exc) cfg.setdefault("mode", "backtest") name = (provider_name or "baostock").lower() if name == "miniqmt": return SanguoMiniQmtProvider(cfg) if name == "baostock": return BaostockProvider(cfg) if name == "local": return LocalParquetProvider(cfg) if name == "unified": return LocalUnifiedProvider(cfg) raise ValueError(f"未知 provider: {name}(支持: local / baostock / miniqmt / unified)") def build_broker_facade(engine: Any) -> Any: """把 BacktestEngine 的聚宽风格 API 包成 BrokerFacade。""" from .strategies.all_weather import BrokerFacade # bullet_trade 的 BacktestEngine 把 set_benchmark/run_daily 等顶层函数注入到策略 # 模块 globals 里。这里把 engine 持有的对应函数转发给 BrokerFacade。 def _order_target_value(code: str, value: float): try: return engine.api.order_target_value(code, value) except Exception: try: return engine.order_target_value(code, value) except Exception as exc: logger.warning("order_target_value 失败 %s=%s: %s", code, value, exc) return None def _order_value(code: str, value: float): try: return engine.api.order_value(code, value) except Exception: try: return engine.order_value(code, value) except Exception as exc: logger.warning("order_value 失败 %s=%s: %s", code, value, exc) return None return BrokerFacade( order_target_value=_order_target_value, order_value=_order_value, ) def _build_strategy(args: argparse.Namespace, provider: Any) -> Any: """根据 --strategy 构造策略实例(分发)。""" name = args.strategy if name == "all_weather": from .strategies import AllWeatherConfig, AllWeatherStrategy return AllWeatherStrategy( provider=provider, config=AllWeatherConfig(max_pool=args.max_pool), ) if name == "momentum_timing": from .strategies import MomentumTimingConfig, MomentumTimingStrategy return MomentumTimingStrategy( provider=provider, config=MomentumTimingConfig(max_pool=args.max_pool), ) if name == "value_selection": from .strategies import ValueSelectionConfig, ValueSelectionStrategy return ValueSelectionStrategy( provider=provider, config=ValueSelectionConfig(max_pool=args.max_pool), ) if name == "small_cap": from .strategies import SmallCapConfig, SmallCapStrategy return SmallCapStrategy( provider=provider, config=SmallCapConfig(max_pool=args.max_pool), ) raise ValueError( f"未知 strategy: {name}(支持: all_weather / momentum_timing / value_selection / small_cap)" ) def _register_schedule(strategy: Any) -> None: """按策略类型注册 bullet_trade 顶层 run_daily/run_monthly 定时任务。""" try: from bullet_trade.core import run_daily, run_monthly # type: ignore except Exception as exc: logger.warning("注册定时任务失败(回测可能不触达): %s", exc) return try: from .strategies import ( AllWeatherStrategy, MomentumTimingStrategy, SmallCapStrategy, ValueSelectionStrategy, ) if isinstance(strategy, AllWeatherStrategy): run_daily(strategy.prepare_stock_list, "9:05") run_monthly(strategy.monthly_adjustment, 1, "9:30") run_daily(strategy.stop_loss, "14:00") return if isinstance(strategy, MomentumTimingStrategy): # 原策略 handle_data 单位时间触发 → 每日 9:30 run_daily(strategy.handle_data, "9:30") return if isinstance(strategy, ValueSelectionStrategy): # 原策略 run_monthly 第 5 个交易日(月度调仓) run_monthly(strategy.monthly_adjustment, 5, "9:30") return if isinstance(strategy, SmallCapStrategy): # 原策略 handle_data 单位时间触发 → 每日 9:30 # 5 日调仓周期由 handle_data 内部 day_count % tc == 0 控制(对齐 g.t % g.tc) run_daily(strategy.handle_data, "9:30") return except Exception as exc: logger.warning("注册定时任务失败(%s): %s", type(strategy).__name__, exc) return logger.warning("未知策略类型 %s,未注册定时任务", type(strategy).__name__) def run_backtest(args: argparse.Namespace) -> Dict[str, Any]: """跑回测,返回结果 dict。 BulletTrade 的 BacktestEngine 接受 strategy_file 或 initialize 等函数。 我们把策略类包成 initialize 函数:initialize 闭包挂 run_daily 等。 """ from bullet_trade import BacktestEngine # type: ignore from bullet_trade.data.api import set_data_provider # type: ignore provider = build_provider(args.provider, args.provider_config) set_data_provider(provider) # 占位策略:initialize 里把 self(strategy)挂到聚宽风格定时器 holder: Dict[str, Any] = {} def initialize(context): strategy = _build_strategy(args, provider) holder["strategy"] = strategy # 注册定时任务(按策略类型分发) _register_schedule(strategy) # 先注入 broker(含 set_option 委托) 再 initialize: initialize 里 set_option("use_real_price",True) # 才能真正设到 bullet_trade settings → fq_mode=pre 与 get_current_data 一致, 买入才成交 holder["broker"] = build_broker_facade_inner(strategy, context) strategy.broker = holder["broker"] # A 股费用 + 滑点(聚宽风格全局函数,对齐个股回测;BulletTrade 默认费用不全) from bullet_trade.core.api import set_order_cost, set_slippage # type: ignore from bullet_trade.core.settings import OrderCost, FixedSlippage # type: ignore set_order_cost( OrderCost( open_tax=0.0, close_tax=args.stamp_duty_rate, open_commission=args.commission_rate, close_commission=args.commission_rate, min_commission=args.min_commission, ), type="stock", ) if args.slippage: set_slippage(FixedSlippage(value=args.slippage)) strategy.initialize(context) def build_broker_facade_inner(strategy: Any, context: Any): from .strategies.all_weather import BrokerFacade # 在回测内,聚宽风格 order_target_value 来自 bullet_trade 顶层 from bullet_trade.core.api import ( # type: ignore order_target_value as bt_otv, order_value as bt_ov, ) from bullet_trade.core.settings import set_option as bt_set_option # type: ignore return BrokerFacade( order_target_value=lambda c, v: bt_otv(c, v), order_value=lambda c, v: bt_ov(c, v), # 注入 set_option 委托 bullet_trade settings: 让策略 set_option("use_real_price",True) # 真正生效 → engine fq_mode=pre 与 get_current_data(fq=pre) 一致, 避免保护价<当前价不成交 set_option=lambda k, v: bt_set_option(k, v), ) print(f"[runner] ENGINE_BUILD_PRE strategy={args.strategy}", flush=True) engine = BacktestEngine( initialize=initialize, start_date=args.start, end_date=args.end, frequency=args.frequency, initial_cash=args.cash, benchmark=args.benchmark, ) print("[runner] RUN_START", flush=True) result = engine.run() print("[runner] RUN_DONE type=%s" % type(result).__name__, flush=True) # 输出结果摘要到 markdown(JSON 模式时 result_file="" 跳过) if getattr(args, "result_file", ""): _write_result_md(result, args.result_file, args) return result def _write_result_md(result: Dict[str, Any], path: str, args: argparse.Namespace) -> None: """把回测关键指标写成 markdown(给 docs/portfolio_backtest_result.md)。""" try: summary = result.get("summary", {}) if isinstance(result, dict) else {} strategy_name = getattr(args, "strategy", "all_weather") title_map = { "all_weather": "全天候轮动", "momentum_timing": "牛熊分界+均线动量", "value_selection": "价值精选6条月度调仓", "small_cap": "小市值20只轮动(无对冲)", } title = title_map.get(strategy_name, strategy_name) lines = [ f"# sanguo_portfolio {title}回测结果", "", f"- 策略: {strategy_name}", f"- 区间: {args.start} ~ {args.end}", f"- 初始资金: {args.cash:,.0f}", f"- 基准: {args.benchmark}", "", "## 关键指标", "", "| 指标 | 值 |", "|---|---|", ] for k in ( "total_returns", "annual_returns", "benchmark_returns", "alpha", "beta", "sharpe", "sortino", "max_drawdown", "win_rate", "turnover", ): if k in summary: lines.append(f"| {k} | {summary[k]} |") content = "\n".join(lines) with open(path, "w") as f: f.write(content + "\n") logger.info("回测结果写入 %s", path) except Exception as exc: logger.warning("写结果文件失败: %s", exc) def run_backtest_json(params: Dict[str, Any]) -> Dict[str, Any]: """JSON 入口(供 SSH 触发,前端 MVP 用)。 Args: params: { pool: 标的池(暂未实际使用,占位), start_date, end_date: YYYY-MM-DD, initial_cash: 初始资金, } Returns: { "strategy": "all_weather" | "momentum_timing", "period": {"start": ..., "end": ..., "trading_days": N}, "stocks_selected": [{"code":..., "name":...}, ...], # 末日持仓 "trades": [{date, code, side, amount, price, ...}, ...], "equity_curve": [{"date":..., "equity":...}, ...], "metrics": {total_return, annual_return, max_drawdown, sharpe, ...}, } """ # 构造一个 Namespace 复用 run_backtest strategy_name = params.get("strategy", "all_weather") args = argparse.Namespace( start=params.get("start_date", "2024-01-01"), end=params.get("end_date", "2024-02-29"), cash=float(params.get("initial_cash", 1_000_000.0)), benchmark=params.get("benchmark", "000300.XSHG"), frequency="day", strategy=strategy_name, provider=params.get("provider", "local"), provider_config=params.get("provider_config", "{}"), result_file="", # JSON 模式不写 md max_pool=int(params.get("max_pool", 0)), commission_rate=float(params.get("commission_rate", 0.0003)), stamp_duty_rate=float(params.get("stamp_duty_rate", 0.001)), min_commission=float(params.get("min_commission", 5.0)), slippage=float(params.get("slippage", 0.0)), ) raw = run_backtest(args) summary = raw.get("summary", {}) if isinstance(raw, dict) else {} metrics = _extract_metrics(summary) # 净值曲线:daily_records 是 DataFrame,index=date,列含 total_value equity_curve = _extract_equity_curve(raw.get("daily_records")) # 选股(末日持仓):daily_positions 最后一日 stocks_selected = _extract_last_positions(raw.get("daily_positions")) # 成交明细 trades = _extract_trades(raw.get("trades")) meta = raw.get("meta", {}) if isinstance(raw, dict) else {} return { "strategy": strategy_name, "period": { "start": meta.get("start_date", args.start), "end": meta.get("end_date", args.end), "trading_days": len(equity_curve), }, "stocks_selected": stocks_selected, "trades": trades, "equity_curve": equity_curve, "metrics": metrics, "raw_summary": summary, } def _to_float(v: Any) -> float | None: """从 string/number 提取 float,失败返 None。bullet-trade summary 多为 '12.34%' 字符串。""" if v is None: return None if isinstance(v, (int, float)): return float(v) s = str(v).strip().replace("%", "").replace(",", "") try: return float(s) except (TypeError, ValueError): return None def _extract_metrics(summary: Dict[str, Any]) -> Dict[str, float | None]: """bullet-trade summary 用中文 key('策略收益'/'最大回撤'/...)。 百分比按字面数值(12.34% → 12.34),前端按需 /100 显示。 """ return { "total_return": _to_float(summary.get("策略收益")), "annual_return": _to_float(summary.get("策略年化收益")), "max_drawdown": _to_float(summary.get("最大回撤")), "sharpe": _to_float(summary.get("夏普比率")), "win_rate_daily": _to_float(summary.get("日胜率")), "win_rate_trade": _to_float(summary.get("交易胜率")), "trading_days": _to_float(summary.get("交易天数")), } def _extract_equity_curve(daily_records: Any) -> list[Dict[str, Any]]: """daily_records: DataFrame,index=date,列含 total_value。""" out: list[Dict[str, Any]] = [] if daily_records is None: return out try: import pandas as pd # type: ignore if isinstance(daily_records, pd.DataFrame): df = daily_records.reset_index() date_col = "date" if "date" in df.columns else df.columns[0] val_col = "total_value" if "total_value" in df.columns else None if val_col is None: return out for _, row in df.iterrows(): d = row[date_col] out.append({ "date": getattr(d, "strftime", lambda f: str(d))("%Y-%m-%d"), "equity": float(row[val_col]), }) except Exception as exc: logger.warning("解析 equity_curve 失败: %s", exc) return out def _extract_last_positions(daily_positions: Any) -> list[Dict[str, Any]]: """daily_positions: DataFrame,列含 date/code/amount/avg_cost/price/value。 取最后一日的非零持仓作为选股名单。""" out: list[Dict[str, Any]] = [] if daily_positions is None: return out try: import pandas as pd # type: ignore if isinstance(daily_positions, pd.DataFrame) and not daily_positions.empty: df = daily_positions if "date" in df.columns: last_date = df["date"].max() df = df[df["date"] == last_date] for _, row in df.iterrows(): amt = row.get("amount", 0) if amt is None or float(amt) <= 0: continue out.append({ "code": str(row.get("code", "")), "name": str(row.get("code", "")), # name 字段 bullet-trade 没存,前端展示 code "amount": float(amt), "avg_cost": float(row.get("avg_cost", 0) or 0), "price": float(row.get("price", 0) or 0), "value": float(row.get("value", 0) or 0), }) except Exception as exc: logger.warning("解析 last_positions 失败: %s", exc) return out def _extract_trades(trades: Any) -> list[Dict[str, Any]]: """trades: list[Trade],用 __dict__ 或属性兜底提取关键字段。""" out: list[Dict[str, Any]] = [] if not trades: return out keys = ("datetime", "date", "code", "side", "action", "amount", "filled_amount", "price", "filled_price", "commission", "status") for t in trades: item: Dict[str, Any] = {} for k in keys: v = None if hasattr(t, k): v = getattr(t, k) elif isinstance(t, dict): v = t.get(k) if v is None: continue # datetime 类转字符串 if hasattr(v, "strftime"): v = v.strftime("%Y-%m-%d %H:%M:%S") try: if isinstance(v, (int, float)): v = float(v) except Exception: pass item[k] = v if item: out.append(item) return out def main() -> None: logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") args = parse_args() if args.json: # JSON 模式:stderr 仍打日志,stdout 只输出 JSON(供 SSH 捕获) result = run_backtest_json({ "strategy": args.strategy, "start_date": args.start, "end_date": args.end, "initial_cash": args.cash, "benchmark": args.benchmark, "provider": args.provider, "provider_config": args.provider_config, "max_pool": args.max_pool, "commission_rate": args.commission_rate, "stamp_duty_rate": args.stamp_duty_rate, "min_commission": args.min_commission, "slippage": args.slippage, }) print(json.dumps(result, ensure_ascii=False, default=str)) else: run_backtest(args) if __name__ == "__main__": main()