《量化交易入门（三十三）BIAS指标实现和回测》配套源码

import backtrader as bt import yfinance as yf # 定义BIAS指标 class BIAS(bt.Indicator): lines = ('bias',) params = (('period', 20),) def __init__(self): self.lines.bias = (self.data.close - bt.indicators.SMA(period=self.params.period)) / bt.indicators.SMA(period=self.params.period) * 100 # 定义交易策略 class BIASStrategy(bt.Strategy): params = (('bias_upper', 10), ('bias_lower', -10),) def __init__(self): self.bias = BIAS(self.data) self.order = None def next(self): if not self.position: if self.bias[0] > self.params.bias_upper: pass # 不进行操作 elif self.bias[0] < self.params.bias_lower: commission_info = self.broker.getcommissioninfo(self.data) cash = self.broker.get_cash() size = int(cash / (self.data.close[0] * (1 + commission_info.p.commission))) self.order = self.buy(size=size) print(f'BUY: {size} shares') else: if self.order and self.order.isbuy() and self.bias[0] > self.params.bias_upper: self.close() self.order = None # 重置self.order为None print(f'SELL: {self.position.size} shares') elif self.order and self.order.issell() and self.bias[0] < self.params.bias_lower: self.close() self.order = None # 重置self.order为None print(f'BUY: {self.position.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(BIASStrategy) # 设置佣金为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()