《量化交易入门（二十九）布林带指标实现和回测》配套源码

import backtrader as bt import yfinance as yf # 定义布林带交易策略 class BollingerBandsStrategy(bt.Strategy): params = ( ('period', 20), ('devfactor', 2) ) def __init__(self): self.bbands = bt.indicators.BollingerBands(period=self.params.period, devfactor=self.params.devfactor) self.order = None def next(self): if self.order: return if self.data.close[0] > self.bbands.lines.top[0]: if self.position.size == 0: commission_info = self.broker.getcommissioninfo(self.data) cash = self.broker.get_cash() - commission_info.getsize(1, self.data.close[0]) size = cash // self.data.close[0] self.buy(size=size) print(f'BUY: {size} shares') elif self.data.close[0] < self.bbands.lines.bot[0]: if self.position.size > 0: size = self.position.size self.close(size=size) print(f'SELL: {size} shares') 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 at {self.data.num2date(order.executed.dt).date()}, Price: {order.executed.price:.2f}, Cost: {order.executed.value:.2f}, Comm: {order.executed.comm:.2f}') elif order.issell(): cost = order.executed.value profit = order.executed.value - order.created.size * order.created.price profit_percent = (profit / cost) * 100 print(f'SELL executed at {self.data.num2date(order.executed.dt).date()}, Price: {order.executed.price:.2f}, Cost: {cost:.2f}, Profit: {profit:.2f}, Profit %: {profit_percent:.2f}%') elif order.status in [order.Canceled, order.Margin, order.Rejected]: print('Order Canceled/Margin/Rejected') # 创建Cerebro引擎 cerebro = bt.Cerebro() # 设置初始资金 cerebro.broker.setcash(100000.0) # 下载苹果股票数据 data = yf.download('AAPL', '2020-01-01', '2023-12-30') data = data.dropna() # 将数据添加到Cerebro引擎中 data = bt.feeds.PandasData(dataname=data) cerebro.adddata(data) # 添加MACD策略 cerebro.addstrategy(BollingerBandsStrategy) # 设置佣金为0.1% cerebro.broker.setcommission(commission=0.001) # 添加分析指标 cerebro.addanalyzer(bt.analyzers.Returns, _name='returns') cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe') cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown') # 运行回测 print('Starting Portfolio Value: %.2f' % cerebro.broker.getvalue()) results = cerebro.run() print('Final Portfolio Value: %.2f' % cerebro.broker.getvalue()) # 获取回测结果 strat = results[0] returns = strat.analyzers.returns.get_analysis() sharpe = strat.analyzers.sharpe.get_analysis() drawdown = strat.analyzers.drawdown.get_analysis() # 打印回测指标 print('Annualized Return: %.2f%%' % (returns['rnorm100'])) print('Sharpe Ratio: %.2f' % (sharpe['sharperatio'])) print('Max Drawdown: %.2f%%' % (drawdown['max']['drawdown'])) print('Max Drawdown Period: %s' % (drawdown['max']['len'])) # 绘制回测结果 cerebro.plot()