ETH 15分钟K线量化回测：Backtrader实现100次滚动策略验证

这次我们来看一个很具体、也很实用的量化小项目：在以太坊（ETH）15分钟图表上做短线策略回测，并且连续记录100次回测结果。

为什么要单独盯15分钟级别？因为短线策略最怕两件事：一是周期太长导致信号太少，二是周期太短导致手续费和滑点吃掉利润。15分钟K线在两者之间相对平衡，既能产生足够的交易样本，又比1分钟级别稳定，是很多个人量化玩家做加密币短线验证的常用周期。

这个回测项目最值得关注的几个点：基于Backtrader框架做事件驱动回测、ETH 15分钟K线数据输入、策略采用“双均线趋势 + RSI过滤器 + 固定止损”的组合逻辑、通过脚本循环执行100次回测并自动汇总指标。整个项目跑完，你能得到的不只是一条净值曲线，而是一份包含胜率、盈亏比、最大回撤、连续亏损次数等维度的回测统计表。

这篇文章不会去论证“短线策略能不能稳定赚钱”，那种结论在没有实盘验证之前都不成立。本文要做的是把整套可复用的回测流程拆开：数据怎么取、环境怎么搭、策略怎么写、100次回测怎么批量执行、结果怎么汇总，以及最容易踩的坑在哪里。

如果你已经在写量化策略，但每次验证新想法都要手动开图表、记参数、反复点回测按钮，那这篇文章可以直接收藏。读完你可以照着一套脚本把历史回测流水线化，以后每个新策略都能在半小时内跑完一轮批量验证。

注意一句话：本文内容仅作技术研究和教学用途，不构成任何投资建议。加密货币市场波动极大，任何策略在实盘前都需要经过充分验证并遵守所在地区法律法规。

1. 核心能力速览

从能力表可以看出，这个项目的硬件门槛很低。回测是纯计算任务，没有模型推理，不依赖 GPU，也没有显存压力。100次回测的耗时主要取决于数据量、策略复杂度和是否启用 matplotlib 绘图。通常会比你跑一次深度学习推理快得多。

2. 适用场景与使用边界

2.1 适合谁用

这个项目适合三类人。

第一类是量化交易初学者。Backtrader 的学习曲线在量化框架里属于中等偏友好，15分钟K线数据量又不算大，很适合用来理解“事件驱动回测”的基本流程。

第二类是已经有策略想法、但不想每次手动点回测按钮的人。把“取数 -> 跑策略 -> 记录指标 -> 画图”写成脚本后，后续验证新参数只需要改一行配置。

第三类是做策略复盘的人。ETH 15分钟级别每天会产生96根K线，一个月大约2880根，足够做日线级别趋势跟踪的初步测试。如果你关心某个时间段内策略表现是否稳定，100次滚动回测比单次回测更有参考价值。

2.2 能解决什么问题

• 把历史数据上的策略信号可视化，确认交易指令是否按预期触发。
• 通过多次回测统计胜率和盈亏比，减少“单一区间回测表现好”造成的误判。
• 量化手续费和滑点对短线策略的影响。
• 快速比较不同参数组合下的回撤和收益率。

2.3 不适合什么场景

15分钟K线回测不适合以下几类情况：

• 高频交易级别的验证。秒级或毫秒级撮合模型需要逐笔成交数据，15分钟K线粒度太粗。
• 实盘交易信号推送。回测框架本身不负责实盘下单，接实盘需要另行开发交易执行模块。
• 没有任何历史数据的币种。新上线的小币种K线样本不够，回测结果没有统计意义。
• 忽略流动性和深度的情况。回测假设按收盘价成交，实际短线挂单会被盘口深度影响。

2.4 合规与风险边界

回测结果代表的是历史数据下的策略表现，不代表未来收益。加密货币交易在不同地区有不同监管要求，请确认你的研究和交易行为符合所在地区法律法规。不要使用回测结果向他人承诺收益，也不要基于未经实盘验证的策略投入大额资金。

如果后续把策略连接实盘交易API，还需要特别注意API密钥保管、下单权限限制和风控机制。建议先用模拟盘验证一段时间，再考虑小资金实盘。

3. 环境准备与前置条件

3.1 系统与Python版本

建议在 Windows 10/11、Ubuntu 20.04+ 或 macOS 上运行。Python 版本建议 3.9 到 3.11，Backtrader 对 Python 3.12 的兼容性需要实测，不推荐在新环境中直接冒险。

检查Python版本：

python --version

如果还没有 Python，建议安装 Anaconda 或 Miniconda，方便创建独立环境：

conda create -n eth_bt python=3.10 -y conda activate eth_bt

3.2 安装依赖库

需要安装的核心库有四个：backtrader、pandas、numpy、matplotlib。

pip install backtrader pandas numpy matplotlib

如果下载速度不理想，可以临时使用国内镜像源：

pip install backtrader pandas numpy matplotlib -i https://pypi.tuna.tsinghua.edu.cn/simple

安装完成后验证：

python -c "import backtrader; import pandas; print('deps ok')"

能正常输出 d eps ok 就说明环境就绪。

3.3 数据准备检查清单

• 是否存在至少连续3个月的ETH 15分钟K线数据。
• CSV列名是否包含 open_time、open、high、low、close、volume。
• open_time 是否为时间戳或可解析的日期时间字符串。
• 数据中是否存在跳空、重复时间戳、NaN值。
• 手续费和滑点参数是否按实际交易环境预留。

3.4 磁盘与内存

ETH 15分钟K线数据量不大。以一年数据为例，约 365 * 96 = 35040 行，CSV文件通常只有几MB。100次回测产生的中间数据可能会多一些，但整体磁盘占用可以控制在几百MB以内。内存方面，普通8GB内存的机器跑这个项目没有压力。

4. 数据准备与回测框架搭建

4.1 数据来源与格式

ETH 15分钟K线数据可以来自交易所公开历史数据接口、第三方数据服务或本地CSV文件。以常见交易所公开K线接口返回的结构为例，字段通常包含开盘时间、开盘价、最高价、最低价、收盘价、成交量等。

一个统一的数据清洗流程如下：

import pandas as pd def load_eth_15m_data(file_path): """ 加载ETH 15分钟K线CSV数据，统一列名和类型 实际文件需要按你的数据源调整列名 """ df = pd.read_csv(file_path) # 统一列名 df.rename( columns={ "timestamp": "open_time", "open": "open", "high": "high", "low": "low", "close": "close", "volume": "volume", }, inplace=True, ) # open_time 转成 datetime 并设为索引 df["open_time"] = pd.to_datetime(df["open_time"]) df.set_index("open_time", inplace=True) # 只保留必要列 df = df[["open", "high", "low", "close", "volume"]].astype(float) # 去重、排序、去空值 df = df[~df.index.duplicated(keep="first")].sort_index() df.dropna(subset=["open", "high", "low", "close", "volume"], inplace=True) return df

如果你还没有本地CSV，也可以通过接口拉取。下面是一个通用的接口拉取模板，实际接口地址、参数名和限频要求需要按你选用的数据源调整：

import requests import pandas as pd def fetch_eth_15m_from_api(symbol="ETHUSDT", limit=3000): """ 从公开K线接口拉取ETH 15分钟数据。 这里使用的是通用示例，实际地址请替换为你的数据源地址。 """ url = "https://your-data-source.example.com/api/v3/klines" params = { "symbol": symbol, "interval": "15m", "limit": limit, } resp = requests.get(url, params=params, timeout=30) resp.raise_for_status() data = resp.json() df = pd.DataFrame(data, columns=[ "open_time", "open", "high", "low", "close", "volume", "close_time", "quote_volume", "count", "taker_buy_volume", "taker_buy_quote_volume", "ignore", ]) df["open_time"] = pd.to_datetime(df["open_time"], unit="ms") df.set_index("open_time", inplace=True) df = df[["open", "high", "low", "close", "volume"]].astype(float) return df

这里要特别提醒：不同数据源的返回字段顺序可能不同，解析前先打印前几行确认字段结构，不要盲目套用列名。

4.2 Backtrader数据馈送

把清洗后的 pandas DataFrame 传入 Backtrader，需要注意列名映射。Backtrader 默认认识 open、high、low、close、volume 这些列名，但时间索引要通过timeframe和compression参数声明。

import backtrader as bt def create_bt_data(df): data = bt.feeds.PandasData( dataname=df, datetime=None, # 使用 DataFrame 索引 open="open", high="high", low="low", close="close", volume="volume", openinterest=None, timeframe=bt.TimeFrame.Minutes, compression=15, ) return data

这里设置timeframe=bt.TimeFrame.Minutes, compression=15后，Backtrader 能够按照15分钟周期正确驱动策略事件。如果没有正确设置，策略的 next() 调用频率会按每天而不是每根K线触发，导致信号严重失真。

5. 短线策略逻辑设计

这个项目里的策略不需要太复杂。回测的核心目的不是构建一个逻辑复杂的黑箱模型，而是建立一套可重复的验证流程。策略越简单，后期分析结果时越容易定位问题。

采用一个经典的组合逻辑：

• 快线均线：5周期SMA。
• 慢线均线：20周期SMA。
• 买入条件：快线上穿慢线，并且 RSI(14) 低于 50。
• 卖出条件：快线下穿慢线，或者 RSI(14) 高于 75。
• 止损规则：入场后价格向不利方向波动超过 2%，强制平仓。
• 仓位规则：每次固定投入总资金的 20%，不做加仓。

选择这个逻辑的原因是它足够简单，每个信号都有明确含义，适合作为第一批回测测试用例。如果你之后想测试其他策略，只需要替换 Backtrader 的 Strategy 类。

6. 回测代码实现

6.1 策略类实现

import backtrader as bt class EmaRsiShortLineStrategy(bt.Strategy): """ ETH 15分钟短线策略： - 快线下穿慢线且 RSI 超买时卖出 - 入场后固定比例止损 """ params = ( ("fast_period", 5), ("slow_period", 20), ("rsi_period", 14), ("rsi_buy_threshold", 50), ("rsi_sell_threshold", 75), ("stop_loss_pct", 0.02), ("risk_per_trade", 0.2), ) def __init__(self): self.fast_ma = bt.indicators.SMA(self.data.close, period=self.p.fast_period) self.slow_ma = bt.indicators.SMA(self.data.close, period=self.p.slow_period) self.rsi = bt.indicators.RSI(self.data.close, period=self.p.rsi_period) self.cross_up = bt.indicators.CrossOver(self.fast_ma, self.slow_ma) self.order = None def next(self): if self.order: return if not self.position: # 入场：金叉 + RSI低于阈值 if self.cross_up > 0 and self.rsi[0] < self.p.rsi_buy_threshold: size = self.broker.getvalue() * self.p.risk_per_trade / self.data.close[0] self.order = self.buy(size=size) else: # 出场：死叉 或 RSI超买 或 触发止损 if self.cross_up < 0 or self.rsi[0] > self.p.rsi_sell_threshold: self.order = self.close() else: current_price = self.data.close[0] entry_price = self.position.price drawdown_pct = (current_price - entry_price) / entry_price if drawdown_pct <= -self.p.stop_loss_pct: self.order = self.close() def notify_order(self, order): if order.status in [order.Completed, order.Canceled, order.Margin]: self.order = None

这段代码里值得注意的地方：notify_order必须把self.order重置为 None，否则策略会一直认为上一笔订单还在执行中，后续K线无法开仓。

止损比例 2% 是短线策略里比较常见的参数，但这不是推荐值，具体数值必须根据ETH的历史波动率做调整。15分钟级别的价格波动如果经常超过2%，这个止损很容易被噪音扫掉。

6.2 单次回测封装

把单次回测逻辑封装成函数，方便后续在循环中重复调用：

def run_backtest(df, strategy_params, cash=10000.0, commission=0.001): """ 执行一次Backtrader回测 df: ETH 15分钟K线 DataFrame strategy_params: 策略参数 dict cash: 初始资金 commission: 手续费比例 返回指标 dict """ cerebro = bt.Cerebro() cerebro.addstrategy(EmaRsiShortLineStrategy, **strategy_params) data = create_bt_data(df) cerebro.adddata(data) cerebro.broker.setcash(cash) cerebro.broker.setcommission(commission=commission) # 分析器：收益率、回撤、交易记录 cerebro.addanalyzer(bt.analyzers.Returns, _name="returns") cerebro.addanalyzer(bt.analyzers.DrawDown, _name="drawdown") cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name="trades") results = cerebro.run() strat = results[0] # 提取指标 ret = strat.analyzers.returns.get_analysis() dd = strat.analyzers.drawdown.get_analysis() trade_analysis = strat.analyzers.trades.get_analysis() total_trades = trade_analysis.total.closed if trade_analysis.total else 0 won = trade_analysis.won.total if trade_analysis.won else 0 lost = trade_analysis.lost.total if trade_analysis.lost else 0 win_rate = won / total_trades if total_trades > 0 else 0.0 return { "final_value": cerebro.broker.getvalue(), "total_return": ret["rtot"], "max_drawdown": dd["max"]["drawdown"], "total_trades": total_trades, "win_rate": win_rate, "won": won, "lost": lost, }

这里手续费按千分之一设置，这是很多现货交易场景常见费率，但具体以你的实际交易渠道为准。如果不设置佣金，回测结果会偏乐观，尤其是15分钟级别这种频繁交易的策略，手续费对最终收益影响很大。

7. 100次回测的执行与结果统计

7.1 为什么做100次

单次回测只能说明策略在某一段特定行情下的表现。ETH 不同时间段的市场状态差异很大，有趋势行情、震荡行情、急涨急跌行情。如果策略只在某一小段上涨行情中表现好，其他区间表现很差，单次回测很容易掩盖这个问题。

做100次回测的目的，是拿到一个指标分布。重点看三件事：

• 胜率和盈亏比的中位数是否稳定。
• 最大回撤的分布是否在所有区间内都可控。
• 是否存在某些极端行情区间导致策略完全失效。

7.2 滚动窗口方案

推荐使用滚动窗口方式：每次回测取连续3个月数据，每次向后滚动1周，这样100次回测覆盖约100周的数据范围。

import pandas as pd import numpy as np def generate_rolling_windows(df, window_days=90, step_days=7): windows = [] start = df.index[0] end = df.index[-1] current = start while current + pd.Timedelta(days=window_days) <= end: win_start = current win_end = current + pd.Timedelta(days=window_days) windows.append((win_start, win_end)) current += pd.Timedelta(days=step_days) return windows

7.3 批量执行100次

def batch_backtest(df, window_days=90, step_days=7, max_rounds=100): windows = generate_rolling_windows(df, window_days, step_days) records = [] for i, (start, end) in enumerate(windows[:max_rounds], start=1): sub_df = df.loc[start:end] if len(sub_df) < 500: continue params = { "fast_period": 5, "slow_period": 20, "rsi_period": 14, "rsi_buy_threshold": 50, "rsi_sell_threshold": 75, "stop_loss_pct": 0.02, "risk_per_trade": 0.2, } try: metrics = run_backtest(sub_df, params) metrics["round"] = i metrics["start"] = start.strftime("%Y-%m-%d") metrics["end"] = end.strftime("%Y-%m-%d") records.append(metrics) print(f"第 {i:3d} 次回测完成，交易 {metrics['total_trades']} 笔，胜率 {metrics['win_rate']:.2%}") except Exception as e: print(f"第 {i:3d} 次回测异常：{e}") result_df = pd.DataFrame(records) result_df.to_csv("eth_15m_backtest_100_records.csv", index=False, encoding="utf-8-sig") return result_df

执行：

df = load_eth_15m_data("eth_15m.csv") result_df = batch_backtest(df, window_days=90, step_days=7, max_rounds=100)

跑完后，eth_15m_backtest_100_records.csv就是100次回测的原始记录。每一行对应一次回测的关键指标。

7.4 结果统计与可视化

拿到100条记录后，做二次统计分析。重点看统计量：均值、中位数、标准差、25%分位数、75%分位数。

import matplotlib.pyplot as plt # 打印核心统计量 summary_cols = ["total_return", "max_drawdown", "total_trades", "win_rate"] summary = result_df[summary_cols].describe() print(summary) # 胜率分布直方图 plt.figure(figsize=(10, 5)) plt.hist(result_df["win_rate"], bins=20, alpha=0.7, edgecolor="black") plt.xlabel("Win Rate") plt.ylabel("Count") plt.title("ETH 15m Strategy Win Rate Distribution (100 Runs)") plt.grid(True, alpha=0.3) plt.tight_layout() plt.savefig("win_rate_distribution.png", dpi=150)

再画一个看更多维度的箱线图：

import matplotlib.pyplot as plt fig, axes = plt.subplots(2, 2, figsize=(12, 8)) axes[0, 0].boxplot(result_df["total_return"]) axes[0, 0].set_title("Total Return") axes[0, 1].boxplot(result_df["max_drawdown"]) axes[0, 1].set_title("Max Drawdown") axes[1, 0].boxplot(result_df["total_trades"]) axes[1, 0].set_title("Total Trades") axes[1, 1].boxplot(result_df["win_rate"]) axes[1, 1].set_title("Win Rate") plt.tight_layout() plt.savefig("backtest_100_boxplot.png", dpi=150)

统计结果出来后，按以下顺序读数据：

1. 先看胜率中位数，如果100次回测的胜率中位数在50%以下，策略本身可能不具备优势。
2. 再看盈亏比和总收益率，胜率低不代表亏钱，盈亏比高时低胜率也能盈利。
3. 然后看最大回撤的75%分位数，如果75%分位数的回撤已经超出心理承受范围，就要考虑降低仓位或加过滤条件。
4. 最后看交易次数，如果某次回测只有不到20笔交易，样本量不足，那次指标没有参考意义。

7.5 从100次到结论

100次回测的另一个价值是可以做参数敏感性分析。比如你可以把 fast_period 从 5 改为 10，slow_period 从 20 改为 40，重新跑一遍100次，对比两组指标分布。

最终形成一份完整的回测记录表：

round,start,end,total_return,max_drawdown,total_trades,win_rate,final_value 1,2024-01-01,2024-04-01,-0.035,0.081,42,0.381,9650 2,2024-01-08,2024-04-08,0.012,0.065,39,0.436,10120 ...

上表只是字段演示，实际数值以你本机回测输出为准。不要因为某一次回测结果好就急着实盘，也不要因为某一次结果差就立刻否定策略。

8. 资源占用与性能观察

8.1 计算与内存

ETH 15分钟K线回测的负载很低。Cerebro 在每次回测中只加载一个数据源，策略也只做均线和RSI计算，没有复杂矩阵运算。100次回测中超过90%的时间可能都花在了数据切片、CSV写入和matplotlib绘图上。

如果发现批量回测跑得很慢，优先检查两点：一是plt.savefig是否在每轮回测内部调用，二是是否在循环内重复加载全量CSV数据。建议把绘图放到100次全部完成后统一执行。

8.2 减少不必要开销的写法

• 在循环外读取全量数据，循环内只做df.loc[start:end]切片。
• 关闭 Cerebro 的stdstats绘图，用cerebro.run(exactbars=True)减少内存占用。
• 在批量执行时不启动 matplotlib 交互模式，只在最后保存图片。

cerebro = bt.Cerebro(exactbars=True)

加了exactbars=True后，Backtrader 不会在内存中保留完整的价格线用于绘图，内存占用会显著下降，代价是回测结束后无法直接从 Cerebro 内部绘图，需要自己保存指标。

8.3 日志输出节奏

100次循环不要每次 print 完整指标，只输出每次回测的交易数和胜率即可。如果某次回测异常，再打印异常信息。全量详细日志建议写入文件，方便事后排查。

9. 常见问题与排查方法

下表整理了最容易遇到的10个问题，按照出现频率从高到低排列：

9.1 关于手续费和滑点的一点点细节

Backtrader 的setcommission可以设置百分比手续费，但滑动点差通常需要自定义 CommissionInfo 类：

import backtrader as bt class CustomCommission(bt.CommInfoBase): params = ( ("commission", 0.001), ("slip_perc", 0.0005), ) def _getcommission(self, size, price, pseudoexec): commission = abs(size) * price * self.p.commission slippage = abs(size) * price * self.p.slip_perc return commission + slippage cerebro.broker.addcommissioninfo(CustomCommission())

这个示例演示了如何把滑点拆出来计算。实盘前务必根据你的实际交易渠道和流动性重新设置参数。

10. 最佳实践与使用建议

10.1 先小参数验证

第一次跑通流程，先用最短的时间窗口和最小参数集跑通代码。比如先用30天数据、10次回测验证流程，确认CSV能够正确输出后再扩展到100次。这样能避免在批量模式下花大量时间调试基础问题。

10.2 保留一套最小可运行配置

把数据清洗、策略、回测三个模块拆成独立函数，然后保留一份run_demo.py，里面写死一组默认参数。以后新想法只需要复制一份策略类，不用动其他模块。

项目目录建议这样组织：

eth_15m_backtest/ ├── data/ │ └── eth_15m.csv ├── strategies/ │ └── ema_rsi_strategy.py ├── backtest/ │ └── engine.py ├── results/ │ ├── eth_15m_backtest_100_records.csv │ ├── win_rate_distribution.png │ └── backtest_100_boxplot.png ├── run_demo.py └── requirements.txt

10.3 批量任务加日志与重试

100次回测中如果某次因为数据窗口不足而失败，不要让整个脚本崩溃。建议在循环里用 try-except 捕获异常，跳过异常窗口并记录原因。跑完后单独检查失败批次是否影响了统计分布。

10.4 接口服务与访问限制

如果你把回测封装成 Web 服务，比如通过 Flask 或 FastAPI 提供“指定日期区间回测”的接口，需要限制访问范围。回测接口不需要对外开放公网，建议只绑定127.0.0.1，或者在服务前面加一层鉴权，否则容易被滥用。

10.5 合规与授权

如果数据来源是某个数据服务商，请确认你的使用方式符合其服务条款，尤其不要批量下载后二次分发。后续如果把策略接人实盘交易，务必备份好API密钥，限制IP白名单，并设置下单金额上限。

10.6 从回测到实盘的检查清单

• 手续费、滑点是否已计入回测。
• 是否在多个不同行情区间下测试过策略。
• 100次回测的胜率和最大回撤分布是否在可接受范围。
• 是否做过滚动窗口的前视偏差检查，即策略只用当时已发生的数据。
• 是否在模拟盘中至少运行一段时间验证执行链路。

11. 小结

ETH 15分钟图表上的短线策略回测，核心价值不在于某一个策略有多神，而在于把回测这件事变成一条可重复、可批量、可统计的流水线。用Backtrader做事件驱动回测，用滚动窗口做100次批量验证，最后用表格和图表给出指标分布——这套流程可以直接复用到其他币种、其他周期、其他策略。

第一次跑通后，最先应该验证的是策略能否在数据窗口内产生足够的交易信号。如果100次回测中有不少窗口交易次数少于20笔，说明策略在实际行情中触发频率偏低，需要考虑缩短均线周期或放宽入场条件。最容易踩的坑是手续费和滑点设置不真实，导致回测收益虚高。

后续可以继续扩展的方向：

• 把单策略回测扩展到参数网格搜索，自动寻找参数最优区间。
• 把ETH单一品种扩展到ETH、BTC、SOL等多个交易对，Backtrader支持通过adddata组合多数据源，实现多股或多元资产组合回测，这也是它的一大优势。多元资产回测能看到策略在不同币种上的表现差异，避免单一币种拟合。
• 引入更多过滤器，比如基于ATR的动态止损、基于成交量的突破过滤。
• 把回测结果接入报告生成系统，每天定时跑一次数据更新并生成最新回测图表。

建议收藏备用。等你有新策略想法的时候，直接把这套流程搬出来跑一遍，节省下来的时间足够多看几十根K线了。