说明:文中代码为通用工程骨架,品种代码与参数均为演示用示例值,不构成任何交易建议;涉及的收益/盈亏字段仅指回测程序的输出口径,非真实业绩。
我最开始只在单个品种上做回测,得出的结论看着很理想,换到其他标的上立刻失效。如果逐个手动运行回测脚本,不仅耗时久,还容易出现漏跑、参数不一致、结果文件散乱在各个文件夹里的问题,事后根本没法横向对比不同标的的表现。
借助 WorkBuddy 本地运行脚本、读取本地文件、批量生成执行任务的能力,我搭建了一整条自动化流水线:一次性遍历全部标的,逐个执行回测,独立保存每一份结果,最后自动合并所有数据并生成离线看板。整套流程全程不用人工介入,保证每一轮回测的执行口径完全统一。
回测脚本最容易出错的地方就是硬编码标的信息。我把品种合约映射、交易基础参数全部抽离为独立配置模块,主回测脚本只做导入调用,不再写死任何标的信息。
新建 instrument_config.py 配置模块,代码片段如下:
# 品种-主力合约映射表(示例,仅用于演示)
CONTRACT_MAP = {
"AU": "AU2612",
"SC": "SC2612",
"RB": "RB2612"
}
# 每品种基础交易参数
TRADE_PARAM = {
"AU": {"min_tick": 0.02, "default_lots": 1},
"SC": {"min_tick": 0.1, "default_lots": 1},
"RB": {"min_tick": 1, "default_lots": 2}
}
# 批量回测标的清单
TARGET_LIST = list(CONTRACT_MAP.keys())后续只要修改这份配置文件,就能增减标的、调整基础参数,不需要改动回测核心代码。
我直接让 WorkBuddy 生成主调度脚本,核心逻辑是循环遍历标的清单,单个标的出错不会中断整批任务,同时每一个标的单独输出一份 CSV 结果文件,避免内存溢出导致全部数据丢失。
调度脚本骨架代码:
import os
import traceback
import pandas as pd
from instrument_config import TARGET_LIST, CONTRACT_MAP
# 新建结果存放目录
OUTPUT_DIR = "batch_result"
os.makedirs(OUTPUT_DIR, exist_ok=True)
error_list = []
def run_single_backtest(instrument):
"""单个标的回测入口,输出独立 CSV 文件"""
# 调用回测核心函数(业务逻辑独立封装)
result_df = backtest_core(instrument, CONTRACT_MAP[instrument])
# 单标的结果单独落盘
save_path = os.path.join(OUTPUT_DIR, f"{instrument}_result.csv")
result_df.to_csv(save_path, encoding="utf-8-sig", index=False)
return result_df
if __name__ == "__main__":
for code in TARGET_LIST:
try:
run_single_backtest(code)
print(f"{code} 回测执行完成")
except Exception:
# 捕获异常,记录错误信息,继续执行下一个标的
error_list.append({"instrument": code,
"error": traceback.format_exc()})
# 保存错误清单
pd.DataFrame(error_list).to_csv("run_error_log.csv",
encoding="utf-8-sig", index=False)把这段代码丢给 WorkBuddy,它可以自动补全函数封装、日志打印、文件夹自动创建等细节。
所有单标的 CSV 文件生成完毕后,需要把全部文件合并成一张总统计表,再生成纯前端 HTML 看板,不需要部署后端,双击就能打开查看,支持本地表格筛选与排序。
数据合并 + 生成看板示例代码:
import glob
from pathlib import Path
from datetime import datetime
import pandas as pd
# 读取目录下所有结果 CSV
csv_files = glob.glob("batch_result/*.csv")
df_list = []
for file in csv_files:
df = pd.read_csv(file, encoding="utf-8-sig")
df_list.append(df)
total_df = pd.concat(df_list, ignore_index=True)
# 增加运行时间戳与数据区间,方便追溯
total_df["run_time"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
total_df["data_period"] = "2025-01-01 ~ 2026-10-01"
# 保存总汇总表
total_df.to_csv("total_summary.csv", encoding="utf-8-sig", index=False)
# 生成纯前端离线 HTML,不引入外部 CDN
html_content = f"""
<html>
<head>
<meta charset="utf-8">
<title>批量回测汇总看板</title>
</head>
<body>
{total_df.to_html(index=False)}
</body>
</html>
"""
Path("backtest_dashboard.html").write_text(html_content, encoding="utf-8")生成后的 HTML 文件完全离线可用,内网无网络环境也能正常打开查看表格。
为了验证额外筛选条件对整体表现的影响,我搭建了两组对照实验:第一组执行基础版本回测,第二组开启方向过滤条件,两组使用完全一致的标的清单、数据区间与基础配置。
两次运行分别生成两套独立的 CSV 结果文件,再把两份总表并排读取,做字段差值对比。这套对照实验的核心价值,是把筛选条件带来的变化量化出来,避免主观判断优化效果。全程只控制唯一变量,其余所有配置保持完全一致。
在反复调试整条流水线的过程中,我踩过不少隐蔽的工程问题,整理成表格方便自查:
问题描述 | 产生原因 | 解决办法 |
|---|---|---|
回测结果和实盘表现不一致 | 回测脚本与实盘脚本使用两份独立配置文件,参数口径不一致 | 回测、实盘共用同一份 |
批量运行程序崩溃,所有结果全部丢失 | 所有标的结果全部存入内存 DataFrame,没有分步落地文件 | 每个标的跑完立刻单独保存 CSV,只把文件名存入列表,不把全量数据留在内存 |
夜盘跨自然日的交易记录归属错误 | 直接用本地日期 | 严格使用交易所交易日历做时间归属,不依赖系统本地日期 |
生成的 HTML 看板在无网络电脑打开空白 | 网页代码引入了外部 CDN 在线资源 | 全部使用纯原生前端代码,不加载任何外部在线 JS、CSS 资源 |
CSV 文件用 Excel 打开中文乱码 | 文件默认使用 utf-8 编码,Windows 系统解析异常 | 所有文件落盘统一指定编码为 |
多轮回测结果互相混淆,分不清数据来源 | 统计表没有记录运行时间与数据区间 | 每一轮汇总表都追加运行时间戳与数据起止日期 |
这套自动化流水线最大的优势并不是单纯提升运行速度,而是建立起一套可复现、可追溯、口径统一的量化实验流程。每一次批量回测,标的清单、基础参数、数据区间完全固定,结果文件自动留存,横向对比多个标的的表现时,不会因为人工操作差异带来数据偏差。
只要把任务交给 WorkBuddy,从批量调度、结果落地到自动生成看板,全程自动化执行,我只需要最后打开汇总表格做数据分析。
文中用到的技能包与示例已开源在 Gitee:https://gitee.com/yu-tongming/quant-trading-skills ,含单技能版与五角色协作版,MIT 授权,可直接装进支持 Skill 的 AI 助手里。如果你只想拿现成的回测汇总与报告工具,也可以看量化策略工具箱。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。