你是否想用 Python 验证自己的交易策略，却苦于没有合适的工具？BackTrader 作为一款功能强大的开源回测框架，能帮你解决这个问题。本文将带你从零开始，快速掌握 BackTrader 的核心用法。
一、开发环境构建
一个独立、干净的 Python 环境是开始一切的前提。我们推荐使用 MiniConda 来管理。
MiniConda下载：
wget -c https://repo.continuum.io/miniconda/Miniconda3-latest-Linux-x86_64.sh
赋予可执行权限：
chmod 755 Miniconda3-latest-Linux-x86_64.sh
MiniConda安装：
sh Miniconda3-latest-Linux-x86_64.sh
安装完成后，将 Conda 路径添加到环境变量中。
.bashrc文件导出Conda路径
export PATH=$PATH:/usr/local/anaconda/bin
为了提高包下载速度，建议添加国内镜像源。
增加镜像：
conda config --add channels bioconda
conda config --add channels conda-forge
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/free/
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main/
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/conda-forge/
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/bioconda/
查看已添加Channels
conda config --get channels
接下来，开始管理你的 Python 环境。
查看Conda环境
conda env list
创建Conda环境
conda create -n python3 python=3
切换至Conda环境
conda activate python3
退出Conda环境
conda deactivate
删除Conda环境
conda remove -n python3 --all
查看和管理已安装的包：
查看当前Conda环境已经安装的包
conda list
查看指定Conda环境的已经安装包
conda list -n python3
安装包到指定Conda环境
conda install -n python3 numpy
最后，安装我们所需的量化回测相关库。
安装BackTrader
pip install BackTrader plotting
安装tushare
pip install tushare
在当前Conda环境安装numpy
conda install numpy
二、BackTrader简介
1、BackTrader简介
BackTrader 是一款于 2015 年开源的 Python 量化回测框架，它不仅支持历史回测，也能接入实盘交易。其特点非常鲜明：
- 品种多：覆盖股票、期货、期权、外汇、数字货币等多种资产。
- 周期全：支持从 Tick 级、秒级到分钟、日、周、月、年等全周期数据。
- 速度快：利用 pandas 进行矢量运算，并支持多策略并行计算。
- 组件丰富：内置了 Ta-lib 技术指标库、PyFlio 分析模块、绘图模块以及参数自动优化等功能。
- 扩展灵活：可以轻松集成第三方库，如 pyfolio、empyrica、alphalens 等分析模块，甚至能结合 TensorFlow、PyTorch 等机器学习框架。
- 社区活跃：拥有完善的帮助文档和活跃的社区支持。
官网：https://www.BackTrader.com/
BackTrader安装：
pip install BackTrader [plotting]
pip index versions BackTrader # 版本查看
安装后，可以通过 __path__ 属性查看源码路径。
import BackTrader as bt
print(bt.__path__)
2、BackTrader模块
理解 BackTrader 的架构是熟练使用它的关键。其核心模块如下图所示：
- Cerebro：作为框架的“大脑”和调度核心，负责收集所有数据、运行策略、执行回测与交易、分析绩效和可视化。
- DataFeeds：数据馈送模块，负责将不同格式（如 CSV、Pandas DataFrame）的行情数据加载到回测框架中。
- Strategy：策略模块。在这里，你将根据行情数据和指标，制定并执行具体的买卖决策。
- Indicators：技术指标模块。内置了如 RSI、KDJ、MACD 等常见指标，并集成了 TA-Lib 库。
- Orders：订单管理模块。它将策略发出的信号转换为具体的订单指令，并跟踪订单状态。
- Broker：经纪人/交易所模块。模拟真实交易环境，管理账户资金，收取手续费、滑点等。其下包含：
Trade：记录每一笔成交的详细信息。Position：记录当前的持仓情况。
- Analyzers：分析器模块。用于评估策略绩效，计算如年化收益率、夏普比率、最大回撤等关键指标。
- Observers：观测器模块。实时观测并记录回测过程中的现金、权益、交易动作等数据。
- Sizers：仓位管理模块，用于控制每次交易的头寸大小。
- Writers：日志记录模块。
- 回测引擎：通过
Cerebro.run() 启动，模拟真实交易，执行策略逻辑。
- 可视化分析模块：通过
Cerebro.plot() 可将回测结果以图表形式直观展示，包括资金曲线、交易点位、损益情况等。
三、Data Feeds
1、行情数据
数据是回测的基础。我们可以使用 Tushare 这类库来获取历史行情数据。以下示例展示了如何获取一只股票的数据并保存为 CSV。
import tushare as ts
import pandas as pd
if __name__ == '__main__':
# 设置Token
ts.set_token('b31e0ac207a5a45e0f7503aff25bf6bd929b88fe1d017a034ee0d530')
# 初始化接口
ts_api = ts.pro_api()
data = ts_api.daily(ts_code='300750.SZ',
start_date='20180611', end_date='20210713')
data.to_csv("300750.csv", header=None, mode='a')
2、Data Feeds
BackTrader 的 Feeds 模块提供了灵活的数据加载功能。它要求数据至少包含 7 个标准字段：datetime (索引), open, high, low, close, volume, openinterest。
框架内置了多种 Feed 类来适配不同数据源：
GenericCSVData：加载通用 CSV 格式数据。YahooFinanceData：从雅虎财经下载数据。PandasData：从 Pandas DataFrame 加载数据，这是最常用、最方便的方式之一。IBData：从 Interactive Brokers API 获取实时数据。
由于 Pandas 是 Python 数据处理的利器，我们通常先将数据预处理成 DataFrame，再用 PandasData 加载。
stock_hfq_df = pd.read_csv("../data/000300.SH.csv",index_col='date',parse_dates=True)
start_date = datetime(2021, 9, 1) # 回测开始时间
end_date = datetime(2021, 9, 30) # 回测结束时间
data = bt.feeds.PandasData(dataname=stock_hfq_df, fromdate=start_date, todate=end_date) # 加载数据
四、BackTrader回测流程
BackTrader 的回测流程围绕 Cerebro 核心展开，清晰且模块化：
- 使用
DataFeeds 加载数据，并添加到 Cerebro。
- 将编写好的交易策略添加到
Cerebro。
- 按需添加分析器 (
Analyzers) 和观测器 (Observers)。
- 运行
Cerebro.run() 执行回测。
- 运行
Cerebro.plot() 对结果进行可视化。
一个完整的回测代码骨架如下：
import BackTrader as bt # 导入 BackTrader
import BackTrader.indicators as btind # 导入策略分析模块
import BackTrader.feeds as btfeeds # 导入数据模块
# 创建策略
class TestStrategy(bt.Strategy):
# 可选，设置回测的可变参数：如移动均线的周期
params = (
(...,...), # 最后一个“,”最好别删！
)
def log(self, txt, dt=None):
'''可选，构建策略打印日志的函数：可用于打印订单记录或交易记录等'''
dt = dt or self.datas[0].datetime.date(0)
print('%s, %s' % (dt.isoformat(), txt))
def __init__(self):
'''必选，初始化属性、计算指标等'''
pass
def notify_order(self, order):
'''可选，打印订单信息'''
pass
def notify_trade(self, trade):
'''可选，打印交易信息'''
pass
def next(self):
'''必选，编写交易策略逻辑'''
sma = btind.SimpleMovingAverage(...) # 计算均线
pass
# 实例化 Cerebro
Cerebro = bt.Cerebro()
# 通过 feeds 读取数据
data = btfeeds.BackTraderCSVData(...)
# 将数据传递给 “大脑”
Cerebro.adddata(data)
# 通过经纪商设置初始资金
Cerebro.broker.setcash(...)
# 设置单笔交易的数量
Cerebro.addsizer(...)
# 初始资金 100,000,000
Cerebro.broker.setcash(100000.0)
# 设置交易佣金，双边各0.0003
Cerebro.broker.setcommission(commission=0.0003)
# 设置滑点：双边各0.0001
Cerebro.broker.set_slippage_perc(perc=0.0001)
# 添加策略
Cerebro.addstrategy(TestStrategy)
# 添加策略分析指标
Cerebro.addanalyzer(...)
# 添加观测器
Cerebro.addobserver(...)
# 启动回测
Cerebro.run()
# 可视化回测结果
Cerebro.plot()
下面我们来分解这个流程中的关键步骤。
1、Cerebro实例
Cerebro = bt.Cerebro(**kwargs)
Cerebro = bt.Cerebro(runonce=True, preload=True)
参数 runonce 和 preload 决定了回测的模式和效率。runonce 模式速度更快，但必须启用 preload（预加载数据）。
2、添加数据
data = bt.BackTraderCSVData(dataname=**'path.days'**, timeframe=bt.TimeFrame.Days)
Cerebro.adddata(data)
注意：数据中的 datetime 字段必须确保是正确的时间格式。
3、重采样数据
data = bt.BackTraderCSVData(dataname=**'mypath.min'**, timeframe=bt.TimeFrame.Minutes)
Cerebro.resampledata(data, timeframe=bt.TimeFrame.Days)
重采样用于将小周期数据（如分钟线）合成大周期数据（如日线）。
4、数据回放
data = bt.BackTraderCSVData(dataname=**'mypath.min'**, timeframe=bt.TimeFrame.Minutes)
Cerebro.replaydata(data, timeframe=bt.TimeFrame.Days)
数据回放则是用小周期数据来“模拟”大周期的价格变动过程，常用于检验在更细粒度上的策略表现。
5、添加交易策略
策略类是 BackTrader 的灵魂。你需要继承 bt.Strategy 类并实现关键方法：
params：定义策略的可调参数，用于优化。__init__：初始化，声明和计算指标。next：核心，在每个 bar 上执行交易逻辑。notify_order / notify_trade：接收订单和交易的状态通知。prenext / nextstart / start / stop：生命周期钩子函数，用于特殊处理。
添加策略到大脑：
Cerebro.addstrategy(TestStrategy, myparam1=value1, myparam2=value2)
6、Indicators指标
BackTrader 内置了丰富的技术指标。例如，实现一个简单的均线交叉策略：
class TestStrategy(bt.Strategy):
# 自定义均线的实践间隔，默认是5天
params = (
('maperiod', 5),
)
def log(self, txt, dt=None):
''' Logging function for this strategy'''
dt = dt or self.datas[0].datetime.date(0)
print('%s, %s' % (dt.isoformat(), txt))
def __init__(self):
# Keep a reference to the "close" line in the data[0] dataseries
self.dataclose = self.datas[0].close
# To keep track of pending orders
self.order = None
# buy price
self.buyprice = None
# buy commission
self.buycomm = None
# 增加均线，简单移动平均线（SMA）又称“算术移动平均线”，是指对特定期间的收盘价进行简单平均化
self.sma = bt.indicators.SimpleMovingAverage(
self.datas[0], period=self.params.maperiod)
#订单状态改变回调方法 be notified through notify_order(order) of any status change in an order
def notify_order(self, order):
if order.status in [order.Submitted, order.Accepted]:
# Buy/Sell order submitted/accepted to/by broker - Nothing to do
return
# Check if an order has been completed
# Attention: broker could reject order if not enough cash
if order.status in [order.Completed]:
if order.isbuy():
self.log(
'BUY EXECUTED, Price: %.2f, Cost: %.2f, Comm %.2f' %
(order.executed.price,
order.executed.value,
order.executed.comm))
self.buyprice = order.executed.price
self.buycomm = order.executed.comm
elif order.issell():
self.log('SELL EXECUTED, Price: %.2f, Cost: %.2f, Comm %.2f' %
(order.executed.price,
order.executed.value,
order.executed.comm))
self.bar_executed = len(self)
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log('Order Canceled/Margin/Rejected')
# Write down: no pending order
self.order = None
#交易状态改变回调方法 be notified through notify_trade(trade) of any opening/updating/closing trade
def notify_trade(self, trade):
if not trade.isclosed:
return
# 每笔交易收益 毛利和净利
self.log('OPERATION PROFIT, GROSS %.2f, NET %.2f' %
(trade.pnl, trade.pnlcomm))
def next(self):
# Simply log the closing price of the series from the reference
self.log('Close, %.2f' % self.dataclose[0])
# Check if an order is pending ... if yes, we cannot send a 2nd one
if self.order:
return
# Check if we are in the market(当前账户持股情况，size，price等等)
if not self.position:
# Not yet ... we MIGHT BUY if ...
if self.dataclose[0] >= self.sma[0]:
#当收盘价，大于等于均线的价格
# BUY, BUY, BUY!!! (with all possible default parameters)
self.log('BUY CREATE, %.2f' % self.dataclose[0])
# Keep track of the created order to avoid a 2nd order
self.order = self.buy()
else:
# Already in the market ... we might sell
if self.dataclose[0] < self.sma[0]:
#当收盘价，小于均线价格
# SELL, SELL, SELL!!! (with all possible default parameters)
self.log('SELL CREATE, %.2f' % self.dataclose[0])
# Keep track of the created order to avoid a 2nd order
self.order = self.sell()
这个策略的逻辑很直观：当收盘价上穿均线时买入，下穿均线时卖出。
7、参数优化
想要寻找最优的均线周期？BackTrader 提供了便捷的参数优化接口。
Cerebro.optstrategy(MyStrategy, myparam1=range(10, 20))
使用 optstrategy 替代 addstrategy，框架会自动遍历参数范围进行多次回测。
8、交易管理
在回测前，需要配置好交易环境，这包括初始资金、佣金、滑点、仓位大小等。这些都是影响最终结果的关键因素。
# 设置投资金额1000000.0
Cerebro.broker.setcash(1000000.0)
# 每笔交易使用固定交易量
Cerebro.addsizer(bt.sizers.FixedSize, stake=100)
# 设置佣金为0.001,除以100去掉%号
Cerebro.broker.setcommission(commission=0.001)
9、观测器
观测器帮助我们在回测过程中实时观察关键信息。BackTrader 默认会添加三个标准观测器：
bt.observers.Broker：现金和资产总值。bt.observers.Trades：交易盈亏。bt.observers.BuySell：买卖点标记。
你也可以手动添加或禁用它们：
Cerebro = bt.Cerebro() #默认参数: stdstats=True
Cerebro.addobserver(bt.observers.Broker)
Cerebro.addobserver(bt.observers.Trades)
Cerebro.addobserver(bt.observers.BuySell)
10、评价指标
回测结束后，我们需要用客观的指标来评价策略的好坏。Analyzers 模块提供了全面的绩效分析工具。
# 添加分析评价指标
Cerebro.addanalyzer(ancls, *args, **kwargs)
tframes = dict(
days=bt.TimeFrame.Days,
weeks=bt.TimeFrame.Weeks,
months=bt.TimeFrame.Months,
years=bt.TimeFrame.Years)
Cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='SharpeRatio')
Cerebro.addanalyzer(bt.analyzers.DrawDown,_name = 'DrawDown')
Cerebro.addanalyzer(bt.analyzers.AnnualReturn,_name = 'AnnualReturn')
Cerebro.addanalyzer(bt.analyzers.SQN,_name = 'SQN')
Cerebro.addanalyzer(bt.analyzers.TradeAnalyzer,_name = 'TradeAnalyzer')
Cerebro.addanalyzer(bt.analyzers.PositionsValue,_name = 'PositionsValue')
Cerebro.addanalyzer(bt.analyzers.Returns,_name = 'Returns')
Cerebro.addanalyzer(bt.analyzers.LogReturnsRolling,timeframe=tframes['years'],_name = 'LogReturnsRolling')
Cerebro.addanalyzer(bt.analyzers.Transactions, _name='Transactions')
以下是一个结合了策略、数据加载、评价的完整示例：
# -*- coding: utf-8 -*-
from __future__ import (absolute_import, division, print_function,
unicode_literals)
import datetime # For datetime objects
import pandas as pd
import BackTrader as bt
import numpy as np
# SMA Stratey
class SMAStrategy(bt.Strategy):
params = (
('ssa_window', 15),
('maperiod', 15),
)
def log(self, txt, dt=None):
''' Logging function fot this strategy'''
dt = dt or self.datas[0].datetime.date(0)
print('%s, %s' % (dt.isoformat(), txt))
def __init__(self):
# Keep a reference to the "close" line in the data[0] dataseries
self.dataclose = self.datas[0].close
# To keep track of pending orders and buy price/commission
self.order = None
self.buyprice = None
self.buycomm = None
self.sma = bt.indicators.SimpleMovingAverage(
self.datas[0], period=self.params.maperiod)
def notify_order(self, order):
if order.status in [order.Submitted, order.Accepted]:
return
if order.status in [order.Completed]:
if order.isbuy():
self.log(
'BUY EXECUTED, Price: %.2f, Cost: %.2f, Comm %.2f' %
(order.executed.price,
order.executed.value,
order.executed.comm))
self.buyprice = order.executed.price
self.buycomm = order.executed.comm
else: # Sell
self.log('SELL EXECUTED, Price: %.2f, Cost: %.2f, Comm %.2f' %
(order.executed.price,
order.executed.value,
order.executed.comm))
self.bar_executed = len(self)
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log('Order Canceled/Margin/Rejected')
self.order = None
def notify_trade(self, trade):
if not trade.isclosed:
return
self.log('OPERATION PROFIT, GROSS %.2f, NET %.2f' %
(trade.pnl, trade.pnlcomm))
def next(self):
self.log('Close, %.2f' % self.dataclose[0])
if self.order:
return
if not self.position:
if self.dataclose[0] > self.sma[0]:
self.log('BUY CREATE, %.2f' % self.dataclose[0])
self.order = self.buy()
else:
if self.dataclose[0] < self.sma[0]:
self.log('SELL CREATE, %.2f' % self.dataclose[0])
self.order = self.sell()
if __name__ == '__main__':
Cerebro = bt.Cerebro()
Cerebro.addstrategy(SMAStrategy)
dataframe = pd.read_csv('dfqc.csv', index_col=0, parse_dates=True)
dataframe['openinterest'] = 0
data = bt.feeds.PandasData(dataname=dataframe,
fromdate = datetime.datetime(2015, 1, 1),
todate = datetime.datetime(2016, 12, 31)
)
Cerebro.adddata(data)
Cerebro.broker.setcash(100.0)
Cerebro.addsizer(bt.sizers.FixedSize, stake=10)
Cerebro.broker.setcommission(commission=0.0)
print('Starting Portfolio Value: %.2f' % Cerebro.broker.getvalue())
# 增加SharpeRatio，名称为SharpeRatio
Cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name = 'SharpeRatio')
# 增加DrawDown，名称为DW
Cerebro.addanalyzer(bt.analyzers.DrawDown, _name='DW')
results = Cerebro.run()
strat = results[0]
print('Final Portfolio Value: %.2f' % Cerebro.broker.getvalue())
# 查看SharpeRatio
print('SR:', strat.analyzers.SharpeRatio.get_analysis())
# 查看DW
print('DW:', strat.analyzers.DW.get_analysis())
Cerebro.plot()
11、策略回测
当一切配置就绪，一行代码即可启动回测。
result = Cerebro.run(**kwargs)
12、结果可视化
最后，让我们直观地看到策略的表现。plot() 函数会生成包含价格、指标、买卖点、资金曲线等信息的图表。
Cerebro.plot()
注意：如果在 Jupyter Notebook 中绘图报错 Javascript Error: IPython is not defined，需要在代码开头添加 %matplotlib inline，并修改绘图命令为 Cerebro.plot(iplot=False)。
结语
至此，你已经走过了 BackTrader 从环境搭建、数据加载、策略编写到回测分析的全过程。作为一个模块化设计优秀的框架，BackTrader 为你验证交易思想提供了强大的工具。当然，本文只是一个起点，深入掌握还需要你不断实践，探索其更高级的特性，如自定义分析器、复杂仓位管理、多资产组合回测等。
希望这篇指南能帮助你在量化交易的学习道路上迈出坚实的一步。如果你在实践过程中有任何心得或疑问，欢迎到 云栈社区 与更多开发者交流讨论。
发表于 2 小时前
|
查看: 3|
回复: 0
