Backtrader（开源量化回测框架）

Backtrader 是一个强大的开源Python量化回测框架，支持多数据源、多策略、多周期回测与实盘交易。纯Python实现，无外部依赖，架构清晰且易于扩展。

文档：https://www.backtrader.com/docu/

安装

bash
pip install backtrader

如需绘图

pip install backtrader[plotting]

或者

pip install matplotlib

核心概念

Backtrader 使用面向对象的事件驱动架构：

• - Cerebro：策略引擎，负责协调数据、策略和经纪商
• Strategy：策略类，编写交易逻辑的地方
• Data Feed：数据源，支持CSV、Pandas和在线数据
• Broker：经纪商模拟，管理资金和订单
• Indicator：技术指标，内置100+常用指标
• Analyzer：分析器，计算策略绩效指标
• Observer：观察器，记录策略运行时状态

最简示例

python
import backtrader as bt

class MyStrategy(bt.Strategy):
简单均线策略
params = ((period, 20),) # 策略参数：均线周期

def init(self):
# 初始化指标（在init中定义，自动计算）
self.sma = bt.indicators.SimpleMovingAverage(self.data.close, period=self.params.period)

def next(self):
# 每根K线触发一次，在此编写交易逻辑
if self.data.close[0] > self.sma[0]:
if not self.position: # 无持仓则买入
self.buy()
elif self.data.close[0] < self.sma[0]:
if self.position: # 有持仓则卖出
self.sell()

创建引擎

cerebro = bt.Cerebro() cerebro.addstrategy(MyStrategy)

加载数据（Yahoo CSV格式）

data = bt.feeds.YahooFinanceCSVData(dataname=stock_data.csv) cerebro.adddata(data)

设置初始资金

cerebro.broker.setcash(100000.0)

设置手续费

cerebro.broker.setcommission(commission=0.001)

运行回测

print(f初始资金: {cerebro.broker.getvalue():.2f}) cerebro.run() print(f最终资金: {cerebro.broker.getvalue():.2f})

绘制结果

cerebro.plot()

---

数据源

从Pandas DataFrame加载

python
import backtrader as bt
import pandas as pd

从CSV读取数据

df = pd.readcsv(stockdata.csv, parsedates=[date], indexcol=date)

DataFrame必须包含列: open, high, low, close, volume（小写列名）

data = bt.feeds.PandasData(dataname=df)
cerebro.adddata(data)

从CSV文件加载

python

通用CSV格式

data = bt.feeds.GenericCSVData(
dataname=stock_data.csv,
dtformat=%Y-%m-%d, # 日期格式
datetime=0, # 日期列索引
open=1, # 开盘价列索引
high=2, # 最高价列索引
low=3, # 最低价列索引
close=4, # 收盘价列索引
volume=5, # 成交量列索引
openinterest=-1 # 持仓量列索引（-1表示无此列）
)
cerebro.adddata(data)

多股票 / 多周期

python

加载多只股票数据

data1 = bt.feeds.PandasData(dataname=df1, name=stock1)
data2 = bt.feeds.PandasData(dataname=df2, name=stock2)
cerebro.adddata(data1)
cerebro.adddata(data2)

在策略中访问多只股票

class MultiStockStrategy(bt.Strategy): def init(self): # self.datas[0]是第一只股票，self.datas[1]是第二只 self.sma1 = bt.indicators.SMA(self.datas[0].close, period=20) self.sma2 = bt.indicators.SMA(self.datas[1].close, period=20)

def next(self):
for i, d in enumerate(self.datas):
print(f{d._name}: close={d.close[0]:.2f})

数据重采样（分钟线转日线）

python

加载分钟数据

datamin = bt.feeds.GenericCSVData(dataname=1mindata.csv, timeframe=bt.TimeFrame.Minutes)
cerebro.adddata(data_min)

重采样为日线

cerebro.resampledata(data_min, timeframe=bt.TimeFrame.Days)

---

策略类详解

策略参数

python
class MyStrategy(bt.Strategy):
# 定义可调参数（元组格式）
params = (
(fast_period, 5), # 快速均线周期
(slow_period, 20), # 慢速均线周期
(stake, 100), # 每次交易手数
)

def init(self):
self.fastma = bt.indicators.SMA(period=self.p.fastperiod)
self.slowma = bt.indicators.SMA(period=self.p.slowperiod)
# self.p 是 self.params 的简写

def next(self):
if self.fastma[0] > self.slowma[0]:
self.buy(size=self.p.stake)

参数可在运行时覆盖

cerebro.addstrategy(MyStrategy, fastperiod=10, slowperiod=30)

交易方法

python
class MyStrategy(bt.Strategy):
def next(self):
# 按数量买入
self.buy(size=100) # 买入100股
self.sell(size=100) # 卖出100股

# 调整到目标仓位
self.ordertargetsize(target=500) # 调整持仓为500股
self.ordertargetvalue(target=50000) # 调整持仓为5万元市值
self.ordertargetpercent(target=0.5) # 调整持仓为总资产的50%

# 限价单
self.buy(size=100, price=10.5, exectype=bt.Order.Limit)
# 止损单
self.sell(size=100, price=9.0, exectype=bt.Order.Stop)
# 止损限价单
self.buy(size=100, price=10.5, pricelimit=10.8, exectype=bt.Order.StopLimit)

# 撤单
order = self.buy(size=100)
self.cancel(order)

# 对其他股票下单
self.buy(data=self.datas[1], size=200) # 买入第二只股票

订单通知回调

python
class MyStrategy(bt.Strategy):
def notify_order(self, order):
订单状态变化时触发
if order.status in [order.Submitted, order.Accepted]:
return # 订单已提交/已接受，等待执行

if order.status in [order.Completed]:
if order.isbuy():
print(f买入执行: 价格={order.executed.price:.2f},
f数量={order.executed.size}, 手续费={order.executed.comm:.2f})
else:
print(f卖出执行: 价格={order.executed.price:.2f},
f数量={order.executed.size}, 手续费={order.executed.comm:.2f})

elif order.status in [order.Canceled, order.Margin, order.Rejected]:
print(f订单失败: 状态={order.getstatusname()})

def notify_trade(self, trade):
交易完成时触发（一买一卖构成完整交易）
if trade.isclosed:
print(f交易完成: 毛利润={trade.pnl:.2f}, 净利润={trade.pnlcomm:.2f})

获取数据与持仓

python
class MyStrategy(bt.Strategy):
def next(self):
# 当前K线数据
current_close = self.data.close[0] # 当前收盘价
prev_close = self.data.close[-1] # 前一根K线收盘价
current_volume = self.data.volume[0] # 当前成交量
current_date = self.data.datetime.date(0) #