backtrader

name backtrader description Backtrader 开源量化回测框架 - 支持多数据源、多策略、多周期回测与实盘交易，纯Python实现。 version 1.1.0 homepage https://github.com/mementum/backtrader metadata {"clawdbot":{"emoji":"🔄","requires":{"bins":"[Truncated]"}}} Backtrader（开源量化回测框架） Backtrader 是一个强大的开源Python量化回测框架，支持多数据源、多策略、多周期回测与实盘交易。纯Python实现，无外部依赖，架构清晰且易于扩展。 文档： https://www.backtrader.com/docu/ 安装 pip install backtrader # 如需绘图 pip install backtrader[plotting] # 或者 pip install matplotlib 核心概念 Backtrader 使用面向对象的事件驱动架构： Cerebro ：策略引擎，负责协调数据、策略和经纪商 Strategy ：策略类，编写交易逻辑的地方 Data Feed ：数据源，支持CSV、Pandas和在线数据 Broker ：经纪商模拟，管理资金和订单 Indicator ：技术指标，内置100+常用指标 Analyzer ：分析器，计算策略绩效指标 Observer ：观察器，记录策略运行时状态 最简示例 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(): .2 f} ' ) cerebro.run() print ( f'最终资金: {cerebro.broker.getvalue(): .2 f} ' ) # 绘制结果 cerebro.plot() 数据源 从Pandas DataFrame加载 import backtrader as bt import pandas as pd # 从CSV读取数据 df = pd.read_csv( 'stock_data.csv' , parse_dates=[ 'date' ], index_col= 'date' ) # DataFrame必须包含列: open, high, low, close, volume（小写列名） data = bt.feeds.PandasData(dataname=df) cerebro.adddata(data) 从CSV文件加载 # 通用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) 多股票 / 多周期 # 加载多只股票数据 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 ]: .2 f} ' ) 数据重采样（分钟线转日线） # 加载分钟数据 data_min = bt.feeds.GenericCSVData(dataname= '1min_data.csv' , timeframe=bt.TimeFrame.Minutes) cerebro.adddata(data_min) # 重采样为日线 cerebro.resampledata(data_min, timeframe=bt.TimeFrame.Days) 策略类详解 策略参数 class MyStrategy (bt.Strategy): # 定义可调参数（元组格式） params = ( ( 'fast_period' , 5 ), # 快速均线周期 ( 'slow_period' , 20 ), # 慢速均线周期 ( 'stake' , 100 ), # 每次交易手数 ) def __init__ ( self ): self .fast_ma = bt.indicators.SMA(period= self .p.fast_period) self .slow_ma = bt.indicators.SMA(period= self .p.slow_period) # self.p 是 self.params 的简写 def next ( self ): if self .fast_ma[ 0 ] > self .slow_ma[ 0 ]: self .buy(size= self .p.stake) # 参数可在运行时覆盖 cerebro.addstrategy(MyStrategy, fast_period= 10 , slow_period= 30 ) 交易方法 class MyStrategy (bt.Strategy): def next ( self ): # 按数量买入 self .buy(size= 100 ) # 买入100股 self .sell(size= 100 ) # 卖出100股 # 调整到目标仓位 self .order_target_size(target= 500 ) # 调整持仓为500股 self .order_target_value(target= 50000 ) # 调整持仓为5万元市值 self .order_target_percent(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 ) # 买入第二只股票 订单通知回调 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'Buy executed: price= {order.executed.price: .2 f} , ' f'size= {order.executed.size} , commission= {order.executed.comm: .2 f} ' ) else : print ( f'Sell executed: price= {order.executed.price: .2 f} , ' f'size= {order.executed.size} , commission= {order.executed.comm: .2 f} ' ) elif order.status in [order.Canceled, order.Margin, order.Rejected]: print ( f'Order failed: status= {order.getstatusname()} ' ) def notify_trade ( self, trade ): """交易完成时触发（一买一卖构成完整交易）""" if trade.isclosed: print ( f'Trade completed: gross P&L= {trade.pnl: .2 f} , net P&L= {trade.pnlcomm: .2 f} ' ) 获取数据与持仓 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 ) # 当前日期 # 持仓信息 position = self .getposition( self .data) print ( f'Position size: {position.size} ' ) print ( f'Average price: {position.price: .2 f} ' ) # 账户信息 cash = self .broker.getcash() # 可用资金 value = self .broker.getvalue() # 总资产 print ( f'Available cash: {cash: .2 f} , Total value: {value: .2 f} ' ) 内置技术指标 class MyStrategy (bt.Strategy): def __init__ ( self ): # 均线 self .sma = bt.indicators.SimpleMovingAverage( self .data.close, period= 20 ) self .ema = bt.indicators.ExponentialMovingAverage( self .data.close, period= 20 ) self .wma = bt.indicators.WeightedMovingAverage( self .data.close, period= 20 ) # MACD self .macd = bt.indicators.MACD( self .data.close) # self.macd.macd = DIF线, self.macd.signal = DEA线, self.macd.histo = MACD柱 # RSI self .rsi = bt.indicators.RSI( self .data.close, period= 14 ) # Bollinger Bands self .boll = bt.indicators.BollingerBands( self .data.close, period= 20 , devfactor= 2.0 ) # self.boll.mid = 中轨, self.boll.top = 上轨, self.boll.bot = 下轨 # KDJ (Stochastic Oscillator) self .stoch = bt.indicators.Stochastic( self .data, period= 14 ) # ATR (Average True Range) self .atr = bt.indicators.ATR( self .data, period= 14 ) # Crossover signals self .crossover = bt.indicators.CrossOver( self .sma, self .ema) # crossover > 0 表示金叉, < 0 表示死叉 券商/经纪商设置 cerebro = bt.Cerebro() # 设置初始资金 cerebro.broker.setcash( 1000000.0 ) # 设置手续费 cerebro.broker.setcommission(commission= 0.001 ) # 0.1% # 设置手续费 by percentage cerebro.broker.setcommission( commission= 0.0003 , # 0.03% margin= None , # 保证金（期货用） mult= 1.0 # 合约乘数（期货用） ) # Set slippage cerebro.broker.set_slippage_perc(perc= 0.001 ) # 百分比滑点 cerebro.broker.set_slippage_fixed(fixed= 0.02 ) # 固定滑点 # Set trade size per order cerebro.addsizer(bt.sizers.FixedSize, stake= 100 ) # 固定100股 cerebro.addsizer(bt.sizers.PercentSizer, percents= 95 ) # 总资产的95% 分析器 cerebro = bt.Cerebro() cerebro.addstrategy(MyStrategy) # 添加分析器 cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name= 'sharpe' ) # 夏普比率 cerebro.addanalyzer(bt.analyzers.DrawDown, _name= 'drawdown' ) # 最大回撤 cerebro.addanalyzer(bt.analyzers.Returns, _name= 'returns' ) # 收益率 cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name= 'trades' ) # 交易统计 cerebro.addanalyzer(bt.analyzers.SQN, _name= 'sqn' ) # 系统质量数 cerebro.addanalyzer(bt.analyzers.AnnualReturn, _name= 'annual' ) # 年化收益 results = cerebro.run() strat = results[ 0 ] # 获取分析结果 print ( f"Sharpe Ratio: {strat.analyzers.sharpe.get_analysis()[ 'sharperatio' ]: .2 f} " ) print ( f"Max Drawdown: {strat.analyzers.drawdown.get_analysis()[ 'max' ][ 'drawdown' ]: .2 f} %" ) print ( f"Total Return: {strat.analyzers.returns.get_analysis()[ 'rtot' ]: .4 f} " ) # 交易统计 trade_analysis = strat.analyzers.trades.get_analysis() print ( f"Total trades: {trade_analysis[ 'total' ][ 'total' ]} " ) print ( f"Winning trades: {trade_analysis[ 'won' ][ 'total' ]} " ) print ( f"Losing trades: {trade_analysis[ 'lost' ][ 'total' ]} " ) 参数优化 # Use optstrategy for parameter grid search cerebro = bt.Cerebro() cerebro.optstrategy( MyStrategy, fast_period= range ( 5 , 15 ), # Fast MA: 5 to 14 slow_period= range ( 20 , 40 , 5 ) # Slow MA: 20, 25, 30, 35 ) data = bt.feeds.PandasData(dataname=df) cerebro.adddata(data) cerebro.broker.setcash( 100000 ) cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name= 'sharpe' ) # 运行优化（自动遍历所有参数组合） results = cerebro.run(maxcpus= 4 ) # 多核并行 # 提取最优参数 best_sharpe = - 999 best_params = None for result in results: for strat in result: sharpe = strat.analyzers.sharpe.get_analysis().get( 'sharperatio' , 0 ) if sharpe and sharpe > best_sharpe: best_sharpe = sharpe best_params = strat.params print ( f'Best params: fast= {best_params.fast_period} , slow= {best_params.slow_period} ' ) print ( f'Best Sharpe: {best_sharpe: .2 f} ' ) 进阶示例 MACD + 布林带组合策略 import backtrader as bt class MACDBollStrategy (bt.Strategy): """MACD金叉 + 布林带下轨支撑组合买入策略""" params = ( ( 'macd_fast' , 12 ), ( 'macd_slow' , 26 ), ( 'macd_signal' , 9 ), ( 'boll_period' , 20 ), ( 'boll_dev' , 2.0 ), ( 'stake' , 100 ), ) def __init__ ( self ): self .macd = bt.indicators.MACD( self .data.close,