一键回测整个策略库

来源: Obsidian Vault

一键回测整个策略库：让 Python 帮你自动筛选交易策略

原创 数据科学实战

2026年6月8日 11:48

用 Python 揭秘均值回归策略：你的收益从何而来？

引言

测试一个交易策略很有用，但当你的研究真正变得「认真」起来时，往往意味着你手里已经有了一堆策略文件、多个优化器和不同的参数空间。如果还要一个文件一个文件地手动跑回测，这个过程会很快变得枯燥且低效。一个自然的想法是：能不能写一个脚本，把「一整个文件夹的策略」自动变成「一次完整的研究运行」？

本文就来拆解这样一个自动化回测脚本的实现思路。它能够自动下载行情数据、发现每个策略模块、运行各自的优化器、与买入持有（Buy and Hold）基准对比、保存单独结果、导出资金曲线图、记录失败项，并最终生成一张排名汇总表。对正在学习 Python 量化的同学来说，这是一个非常完整的工程实践范例。

一、整体目标：从「单个文件」到「研究流水线」

这个脚本要解决的核心痛点是：规模化。

它的工作流程可以概括为下面这几步：

1. 1. 选择资产和时间周期
2. 2. 下载 OHLCV 行情数据
3. 3. 清洗 K 线数据
4. 4. 创建买入持有基准
5. 5. 自动发现每一个策略模块
6. 6. 运行每个优化器
7. 7. 保存每个策略的优化结果
8. 8. 保存「策略 vs 基准」资金曲线图
9. 9. 提取最优参数与指标
10. 10. 记录失败项但不中断整个运行
11. 11. 按收益对完成的策略排序
12. 12. 导出汇总 CSV 并打印 Top 结果

简单来说，它把一个装满策略文件的文件夹，变成了一份带排名的研究报告。

二、干净的研究起点：导入与基础配置

脚本开头导入了自动化、绘图、数据处理和策略执行所需的工具。

from pathlib import Path   # 管理文件夹与输出路径 import argparse            # 处理命令行参数 import importlib           # 动态加载策略文件（关键） import re                  # 正则，用于清洗文件名 import shutil              # 文件夹操作（删除/重建） import matplotlib matplotlib.use("Agg")     # 使用非交互式后端，适合在服务器/批处理中保存图片 import matplotlib.pyplot as plt import pandas as pd        # 处理结果表格 import yfinance as yf      # 下载行情数据 # 让所有策略使用统一的初始资金 from vectorbt_strategies.strategy_opt_utils import DEFAULT_INIT_CASH

这里几个核心库的分工很清晰：

• • importlib：让脚本能够动态加载策略文件，这是「自动发现」的基础

• • Path：管理文件夹与输出路径

• • pandas：处理结果表格

• • yfinance：下载市场数据

• • matplotlib：绘制资金曲线图

补充说明：matplotlib.use("Agg") 必须放在 import matplotlib.pyplot之前调用，这是新手常踩的坑。Agg 是一个不依赖图形界面的后端，专门用来把图片直接保存成文件，非常适合批量运行的场景。

接着定义默认的资产和周期：

asset = "BTC-USD"            # 默认测试比特币 period = "1y"                # 默认回测一年 INIT_CASH = DEFAULT_INIT_CASH  # 统一的初始资金，保证横向可比

默认测试比特币一年的数据，这些值之后可以通过命令行修改，但默认值让脚本「开箱即跑」。

三、排除核心工具模块

策略文件夹里并不是每个 .py 都是「可交易策略」，有些是工具、注册表、调度器或包初始化文件。脚本用一个集合把它们排除掉：

CORE_MODULES = {    "__init__",    "strategy_opt_utils",    "strategy_registry",    "strategy_dispatcher", }

如果没有这层过滤，脚本可能会去「优化」一个根本不包含策略的辅助文件，导致报错。

四、几个实用的小工具函数

1. 生成安全的文件名

策略名里可能包含空格、符号等不适合做文件路径的字符，需要先「清洗」：

def safe_name(text: str) -> str:    # 把非字母数字、下划线、点、连字符的字符统一替换为下划线    return re.sub(r"[^A-Za-z0-9_.-]+", "_", text).strip("_")

这样可以让文件名保持可预测，避免保存 CSV 或图片时出现损坏的路径。

2. 展平 yFinance 的多级列

Yahoo Finance 有时会返回多级索引（MultiIndex）列，尤其在分组下载时。脚本会先把它展平：

def flatten_yfinance_columns(data: pd.DataFrame) -> pd.DataFrame:    if isinstance(data.columns, pd.MultiIndex):        # 如果第二级只有一个取值，去掉第二级        if len(data.columns.get_level_values(1).unique()) == 1:             data = data.droplevel(1, axis=1)        # 如果第一级只有一个取值，去掉第一级        elif len(data.columns.get_level_values(0).unique()) == 1:             data = data.droplevel(0, axis=1)    return data

脚本后续期望的是 Open、High、Low、Close、Volume 这样的简单列名，而不是嵌套结构。

五、下载并标准化 OHLCV 数据

数据下载函数负责拉取历史 K 线，并把结果标准化：

def download_ohlcv(symbol: str, period: str, interval: str) -> pd.DataFrame:     data = yf.download(symbol, period=period, interval=interval, auto_adjust=False)     data = flatten_yfinance_columns(data)    # 如果没有返回任何数据，立刻报错，而不是让空数据悄悄流入优化器    if data.empty:        raise ValueError(f"No data returned for {symbol}")     data = data.dropna(subset=["Close"]).copy()     close = data["Close"].astype(float)    # 若 Open/High/Low 缺失，则用 Close 相关值补齐     data["Open"] = data.get("Open", close.shift(1)).reindex(close.index)\         .fillna(close.shift(1)).fillna(close).astype(float)     data["High"] = data.get("High", close).reindex(close.index)\         .fillna(close).astype(float)     data["Low"] = data.get("Low", close).reindex(close.index)\         .fillna(close).astype(float)     data["Close"] = close    # 若 Volume 缺失，则填 0    if "Volume" in data.columns:         data["Volume"] = data["Volume"].reindex(close.index).fillna(0).astype(float)    else:         data["Volume"] = 0.0    return data["Open", "High", "Low", "Close", "Volume"](/wiki/"Open", "High", "Low", "Close", "Volume")

这个函数确保每个策略都拿到结构一致、完整的 DataFrame。缺失值能补则补，从而让脚本在面对不同资产和数据怪癖时更加稳健。

易错点提醒：「空数据直接报错」是一个很好的工程习惯。如果让空 DataFrame 一路流进优化器，往往会在很深的调用栈里报出莫名其妙的错误，排查起来非常痛苦。在源头就拦截，能省下大量调试时间。

六、构建买入持有基准

没有基准的策略结果几乎没有意义。脚本用同样的收盘价构建一条买入持有资金曲线：

def buy_and_hold_equity(data: pd.DataFrame) -> pd.Series:     close = data["Close"].astype(float).dropna()    if close.empty:        raise ValueError("Cannot build buy-and-hold benchmark from empty Close data")    # 用同样的初始资金，按收盘价比例增长    return (INIT_CASH * close / close.iloc[0]).rename("Buy and Hold")

它和每个策略使用相同的初始资金，这样就能直接对比「优化后的策略」与「单纯持有资产」的差别。

七、保存资金曲线图

每个完成的策略都会生成一张资金曲线图：

def save_equity_plot(result, output_dir: Path, symbol: str, period: str, benchmark_equity):     portfolio = result.best_portfolio       # 取出最优组合     strategy_equity = portfolio.value()     # 策略的资金曲线     fig, ax = plt.subplots(figsize=(12, 6))    # 绘制优化策略曲线     ax.plot(strategy_equity.index, strategy_equity.values,             label="Optimized Strategy", linewidth=2.5)    # 绘制买入持有基准（虚线）     ax.plot(benchmark_equity.index, benchmark_equity.values,             label="Buy and Hold", linestyle="--", linewidth=2.2)     ax.set_title(f"{symbol} {period} - {result.name}")     ax.set_ylabel("Portfolio Value")     ax.grid(True, alpha=0.3)     ax.legend()     fig.tight_layout()    # 用安全文件名保存图片     path = output_dir / f"{safe_name(result.name)}_equity_vs_buy_hold.png"     fig.savefig(path, dpi=160)     plt.close(fig)    return path

为什么图很重要？因为一张 CSV 也许会告诉你某个策略收益不错，但图能揭示难看的回撤、漫长的横盘期，或者明显跑输基准的事实。

八、自动发现策略模块（核心自动化）

这是整个脚本最关键的一步。它不再手动 import 每个策略文件，而是直接扫描 vectorbt_strategies 文件夹：

def discover_strategy_modules():     strategy_dir = Path(__file__).resolve().parent / "vectorbt_strategies"     modules = []    for path in sorted(strategy_dir.glob("*.py")):        # 跳过核心工具模块        if path.stem in CORE_MODULES:            continue        # 动态导入策略模块         modules.append(importlib.import_module(f"vectorbt_strategies.{path.stem}"))    return modules

这正是脚本可扩展的精髓所在：只要往文件夹里新增一个符合接口约定的策略文件，脚本就能自动发现它，让它成为整次回测的一部分，无需改动主程序。

九、命令行控制：同一引擎，不同数据

脚本支持命令行参数，让你在不同市场和时间周期上复用同一套逻辑：

parser = argparse.ArgumentParser(     description="Backtest all clean vectorbt strategy optimizers on one download" ) parser.add_argument("--symbol", default=f"{asset}") parser.add_argument("--period", default=f"{period}") parser.add_argument("--interval", default="1d") args = parser.parse_args()

于是你可以这样跑：

# 测试以太坊，2 年，日线 python backtest_all_strategies.py --symbol ETH-USD --period 2y --interval 1d # 测试苹果股票，5 年，日线 python backtest_all_strategies.py --symbol AAPL --period 5y --interval 1d

引擎不变，变的只是数据集。

十、创建干净的输出目录

每次运行都会创建一个全新的输出目录，避免新旧结果混在一起：

folder_name = f"{safe_name(args.symbol)}-{safe_name(args.period)}" output_dir = Path(folder_name) # 如果目录已存在，先删除再重建，保证干净 if output_dir.exists():     shutil.rmtree(output_dir) output_dir.mkdir(exist_ok=True) # 单独存放每个策略优化结果的子目录 per_strategy_dir = output_dir / "per_strategy_results" per_strategy_dir.mkdir(exist_ok=True)

十一、案例：完整跑通一个策略

下面把「下载数据 → 跑策略 → 存结果」这一条主线串起来看。

1. 下载数据并保存基准

print(f"Downloading {args.symbol} for {args.period} at {args.interval}...") data = download_ohlcv(args.symbol, args.period, args.interval) # 构建基准并计算基准收益率 benchmark_equity = buy_and_hold_equity(data) benchmark_return_pct = (benchmark_equity.iloc[-1] / benchmark_equity.iloc[0] - 1) * 100 # 保存原始数据和基准曲线，保证可复现 data.to_csv(output_dir / "data.csv") benchmark_equity.to_csv(output_dir / "buy_and_hold_equity.csv", header=True)

2. 遍历每个策略并运行优化器

strategy_modules = discover_strategy_modules() summary_rows = []   # 汇总排名数据 failures = []       # 失败记录 for module in strategy_modules:    # 优先取 STRATEGY_NAME，否则用模块名     strategy_name = getattr(module, "STRATEGY_NAME", module.__name__.split(".")[-1])    print(f"Backtesting {strategy_name}...")    try:        # 每个模块都需暴露 optimize_strategy 函数         result = module.optimize_strategy(data, optimize_for="total_return_pct")        # 给结果补充初始资金和基准收益         result.results["init_cash"] = INIT_CASH         result.results["benchmark_return_pct"] = benchmark_return_pct        # 保存该策略的优化结果表         result.results.to_csv(             per_strategy_dir / f"{safe_name(result.name)}_optimization_results.csv",             index=False         )        # 保存资金曲线图         plot_path = save_equity_plot(             result, output_dir, args.symbol, args.period, benchmark_equity         )        # 取出最优一行         best_row = result.results.iloc[0].to_dict()        # 追加一条汇总行         summary_rows.append({            "strategy": result.name,            "plot": str(plot_path),            "init_cash": INIT_CASH,             **{f"best_{key}": value for key, value in result.best_params.items()},            "total_return_pct": best_row.get("total_return_pct"),            "benchmark_return_pct": benchmark_return_pct,            "sharpe": best_row.get("sharpe"),            "rank_score": best_row.get("_rank_score"),            "max_drawdown_pct": best_row.get("max_drawdown_pct"),            "total_trades": best_row.get("total_trades"),            "end_value": best_row.get("end_value"),         })    except Exception as exc:        # 一个策略失败，不影响整体运行         failures.append({"strategy": strategy_name, "error": str(exc)})        print(f"FAILED {strategy_name}: {exc}")

这里有两个关键设计值得学习：

• • 统一接口约定：每个策略模块都暴露一个 optimize_strategy 函数，主程序只需按同一方式调用即可，这就是「面向接口编程」的思想。

• • 失败隔离：用 try / except 把单个策略的异常隔离开。对于大型策略库来说，一个坏掉的模块不应该拖垮整次研究运行，记录下来继续跑就好。

十二、生成最终排行榜

所有模块测试完后，脚本构建汇总 DataFrame 并按收益排序：

summary = pd.DataFrame(summary_rows) if not summary.empty:    # 按总收益降序排列，缺失值排最后     summary = summary.sort_values(        "total_return_pct", ascending=False, na_position="last"     ) # 保存汇总表和失败表 summary.to_csv(output_dir / "summary.csv", index=False) pd.DataFrame(     failures, columns=["strategy", "error"] ).to_csv(output_dir / "failures.csv", index=False)

运行结束后，输出文件夹里会包含：

• • data.csv（原始数据）

• • buy_and_hold_equity.csv（基准曲线）

• • summary.csv（排名汇总）

• • failures.csv（失败记录）

• • 每个策略的优化结果 CSV

• • 每个策略的「资金 vs 买入持有」对比图

这就是一份完整的多策略回测报告。

最后，脚本还会在终端直接打印 Top 结果，省去手动打开 CSV 的麻烦：

print(f"\nSaved backtest folder: {output_dir.resolve()}") print(f"Strategies completed: {len(summary_rows)}") print(f"Strategies failed: {len(failures)}") if not summary.empty:    print("\nTop strategies by total return:")    print(         summary[            "strategy", "total_return_pct", "benchmark_return_pct",            "sharpe", "max_drawdown_pct", "total_trades",         ](/wiki/            "strategy", "total_return_pct", "benchmark_return_pct",            "sharpe", "max_drawdown_pct", "total_trades",         ).head(10).to_string(index=False)     )

延伸理解：从一个示例仪表盘可以看到，比如在 META 一年的回测中，38 个策略里有 27 个为正收益，最佳策略的优化后总收益甚至能远超买入持有基准。但这恰恰提醒我们——这些都是「优化后」的样本内结果，存在过拟合风险。在实盘前，务必再做样本外测试（Out-of-Sample）和前向滚动验证（Walk-Forward），否则漂亮的回测曲线很可能只是「事后诸葛亮」。

总结

当策略研究超过一两个想法时，手动回测就会变得低效。要把研究做扎实，你需要：

• • 自动化：把重复劳动交给脚本

• • 一致的数据集：所有策略用同一份数据

• • 相同的初始资金：保证横向可比

• • 相同的基准：用买入持有作为统一参照

• • 统一的输出结构：方便归档和复盘

• • 批量对比的能力：不必逐个打开模块

本文拆解的这个脚本，正好把上述要素串成了一条完整的研究流水线。它足够简单、易于理解，又足够强大、可以扩展到整个策略库。

它的核心价值，就是把「一个装满策略文件的文件夹」变成「一份带排名的研究报告」。当然，并不是每个策略都能存活下来——而这正是回测的意义所在：它帮你更快地知道哪些策略不行，从而把宝贵的时间花在真正值得深入的候选者上。

对于正在学习 Python 量化的你来说，即使暂时用不上具体的交易逻辑，这套「动态发现模块 + 统一接口 + 失败隔离 + 结果归档」的工程模式，也值得借鉴到任何需要批量处理的项目中。