sanguo_vnpy_v2/sanguo_portfolio/runner_backtest.py

"""全天候策略回测入口。

用法(VPS Windows / miniQMT 已连):
    set DEFAULT_DATA_PROVIDER=miniqmt
    python -m sanguo_portfolio.runner_backtest \\
        --start 2020-01-01 --end 2024-12-31 --cash 1000000

Mac 没装 xtquant,这里仅作为入口脚本(测试用 mock,实际跑 rsync 到 VPS)。
"""
from __future__ import annotations

# ENV GUARD 必须早于任何 bullet_trade import
import os
os.environ.setdefault("DEFAULT_DATA_PROVIDER", "miniqmt")

import argparse
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("--frequency", default="day", help="回测频率 day/minute")
    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)",
    )
    return p.parse_args()


def build_provider(config_str: str) -> Any:
    """构造 SanguoMiniQmtProvider。"""
    import json
    from .providers import 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")
    return SanguoMiniQmtProvider(cfg)


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 run_backtest(args: argparse.Namespace) -> Dict[str, Any]:
    """跑回测,返回结果 dict。

    BulletTrade 的 BacktestEngine 接受 strategy_file 或 initialize 等函数。
    我们把 AllWeatherStrategy 包成 initialize 函数:initialize 闭包挂 run_daily 等。
    """
    from bullet_trade.core import BacktestEngine  # type: ignore
    from bullet_trade.data.api import set_data_provider  # type: ignore

    from .strategies import AllWeatherStrategy

    provider = build_provider(args.provider_config)
    set_data_provider(provider)

    # 占位策略:initialize 里把 self(strategy)挂到聚宽风格定时器
    holder: Dict[str, Any] = {}

    def initialize(context):
        strategy = AllWeatherStrategy(provider=provider)
        holder["strategy"] = strategy

        # bullet-trade 的 run_daily/run_monthly 接受全局函数;把 method 暴露为模块级
        # 这里偷个懒:用 functools.partial 注册到 engine 的 scheduler
        import functools

        # bullet-trade 顶层 run_daily 等可调用,context._scheduler 暴露
        try:
            from bullet_trade.core import run_daily, run_monthly  # type: ignore
            run_daily(strategy.prepare_stock_list, "9:05")
            run_monthly(strategy.monthly_adjustment, 1, "9:30")
            run_daily(strategy.stop_loss, "14:00")
        except Exception as exc:
            logger.warning("注册定时任务失败(回测可能不触达): %s", exc)

        strategy.initialize(context)
        holder["broker"] = build_broker_facade_inner(strategy, context)
        strategy.broker = holder["broker"]

    def build_broker_facade_inner(strategy: AllWeatherStrategy, 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,
        )
        return BrokerFacade(
            order_target_value=lambda c, v: bt_otv(c, v),
            order_value=lambda c, v: bt_ov(c, v),
        )

    engine = BacktestEngine(
        initialize=initialize,
        start_date=args.start,
        end_date=args.end,
        frequency=args.frequency,
        initial_cash=args.cash,
        benchmark=args.benchmark,
    )
    result = engine.run()

    # 输出结果摘要到 markdown
    _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 {}
        lines = [
            "# sanguo_portfolio 全天候回测结果",
            "",
            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 main() -> None:
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
    args = parse_args()
    run_backtest(args)


if __name__ == "__main__":
    main()