Backtrader 策略回测初探
这篇介绍简略的回测流程，次要的内容如下：
- 回测函数介绍
- 单股回测
- 多股回测
回测函数
回测策略类很简洁，间接继承 bt.Strategy ，复写父类的办法，最初把回测策略类增加到大脑即可。
回测参数
回测类参数增加通过属性变量 params 记录，能够是元组模式，也能够是字典模式。
定义参数<br/>
# 元组模式，留神最初一个，逗号别删除params = ( ('maperiod', 20), )或# 字典模式params = {'maperiod': 20}应用<br/>
通过 self.p.maperiod 拜访提取。bt.ind.SMA(self.data, period=self.p.maperiod)传参<br/>
将策略类传入大脑时，传入参数cerebro.addstrategy(TestStrategy, maperiod=5)
函数<br/>
因为继承了策略类 bt.Strategy ，运行策略时会回调这些函数，须要搞的事件是在相应的函数调用交易逻辑即可。<br/>
先说下 lines 线对象概念，每个指标都是一条 line 线对象，贯通所有的回测日期。bar 概念是每个日期对应所有的指标。简略了解就是 excel 表的中的列和行，line 线对象相当于 excel 的列，bar 概念相当于 excel 表的行。
| 函数名 | 函数阐明 | 调用机会 |
|---|---|---|
__init__() | 初始化时调用，只调用一次 | 初始化 |
| next() | 每条 bar 调用一次，<br/>直到所有 bar 回测结束,<br/>无效的bar | 每条bar都回调 |
| prenext() | 不在next()函数调用的bar,<br/>就调用到这函数 | 不在next()回调的bar |
| notify_order() | 当执行 self.buy() 、self.order_target_percent()、<br/>self.sell()等时触发回调 | 订单状态发生变化回调 |
| notify_trade() | 交易扭转是 | 交易产生时回调 |
| log() | 日志打印函数，打印一些交易信息 | 本人调用 |
还有其余一些函数，这里不一一介绍了，不是很重要的。
指标简介
指标在代码层面上的示意模式就是以 line 线对象的形式存在，贯通整个回测周期，个别在测类类的 __init__ 函数外面构建好<br/>
在数据篇，有展现过在 feeds.Data 上间接扩大指标，上面介绍的是通过 backtrader 的指标对象来构建新的指标，新的指标也是 line 对象，贯通整个回测周期。
上面例子构建 20 日挪动平均线 SMA：<br/>
def __init__(self): self.sma = bt.ind.SMA(self.data, period=20)留神： 这种模式构建的指标是基于数据篇外面所说的第一个表数据的指标，并不是针对所有表的。而且会影响到 next 回调机会。20 日挪动平均线的指标，前 20 日个回测 bar 是有效的，不会在 next() 函数回调，但回调到 prenext() 函数。
下图所示：<br/>
对于指标在这里不细说，前面会写一篇具体点的介绍。
回测繁难配置
只是做一个简略回测测试，所以繁难配置下经纪商，setcash() 配置了一小指标资产 1亿 ，设置佣金 setcommission() 千分之一，addstrategy() 增加策略并设置自定义的参数。
# 设置资产cerebro.broker.setcash(100000000.0)# 设置佣金cerebro.broker.setcommission(commission=0.001)# 增加回测策略，设置自定义参数数值cerebro.addstrategy(SingleTestStrategy, maperiod=20)# 执行策略cerebro.run()# 画图cerebro.plot()这里不细说这配置了，前面再具体说一篇。
单股回测
上面进入主题，搞一个策略，回测下能不能赚钱，这里只是介绍应用办法，理论利用必定不能只靠一个指标。
策略
这策略很简略，当当日收盘价高于 20 日平均线时买买买！当当日收盘价低于 20 日挪动平均线时卖卖卖。
留神：
- 该策略是以收盘价下单，以下单后的下一日开盘价作为交割价。
- 最初一日不做任何的交易。所以在回测的最初一日的前一天必须下单卖出股票，以便在最初一根 bar 的开盘价做为交割价卖出，不然呈现将来函数。
交易罕用函数：
- self.order_target_percent(secu_data, target_pct, name=secu)
- self.order_target_value(secu_data, target_val, name=secu)
- self.buy(secu_data, order_amount, name=secu)
- self.sell(secu_data, order_amount, name=secu)
代码胜过一言半语，间接上残缺代码：<br/>
import datetimeimport backtrader as btimport pandas as pdimport stock_db as sdbclass SingleTestStrategy(bt.Strategy): params = ( ('maperiod', 20), ) def __init__(self): self.order = None self.sma = bt.ind.SMA(self.data, period=self.p.maperiod) pass def downcast(self, amount, lot): return abs(amount // lot * lot) # 能够不要，但如果你数据未对齐，须要在这里测验 def prenext(self): print('prenext 执行 ', self.datetime.date(), self.getdatabyname('300015')._name , self.getdatabyname('300015').close[0]) pass def next(self): # 查看是否有指令执行，如果有则不执行这bar if self.order: return # 回测如果是最初一天，则不进行交易 if pd.Timestamp(self.data.datetime.date(0)) == end_date: return if not self.position: # 没有持仓 # 执行买入条件判断：收盘价格上涨冲破20日均线； # 不要在股票剔除日前一天进行买入 if self.datas[0].close > self.sma and pd.Timestamp(self.data.datetime.date(1)) < end_date: # 永远不要满仓买入某只股票 order_value = self.broker.getvalue() * 0.98 order_amount = self.downcast(order_value / self.datas[0].close[0], 100) self.order = self.buy(self.datas[0], order_amount, name=self.datas[0]._name) else: # 执行卖出条件判断：收盘价格跌破20日均线，或者股票剔除 if self.datas[0].close < self.sma or pd.Timestamp(self.data.datetime.date(1)) >= end_date: # 执行卖出 self.order = self.order_target_percent(self.datas[0], 0, name=self.datas[0]._name) self.log(f'卖{self.datas[0]._name},price:{self.datas[0].close[0]:.2f},pct: 0') pass 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, order.Canceled, order.Margin]: if order.isbuy(): self.log( f"买入{order.info['name']}, 成交量{order.executed.size}，成交价{order.executed.price:.2f} 订单状态：{order.status}") self.log('买入后以后资产：%.2f 元' % self.broker.getvalue()) elif order.issell(): self.log( f"卖出{order.info['name']}, 成交量{order.executed.size}，成交价{order.executed.price:.2f} 订单状态：{order.status}") self.log('卖出后以后资产：%.2f 元' % self.broker.getvalue()) self.bar_executed = len(self) # Write down: no pending order self.order = None def log(self, txt, dt=None): """ 输入日期 :param txt: :param dt: :return: """ dt = dt or self.datetime.date(0) # 当初的日期 print('%s , %s' % (dt.isoformat(), txt)) pass def notify_trade(self, trade): '''可选，打印交易信息''' pass# 开始查问工夫start_query = '2019-01-01'end_query = '2022-09-01'# 开始回测工夫from_date = datetime.datetime(2022, 1, 1)to_date = datetime.datetime(2022, 10, 10)cerebro = bt.Cerebro()# 增加几个股票数据codes = [ '300015', # '300347', # '300760', # '603127', # '600438']# 增加多个股票回测数据end_date = 0for code in codes: data = sdb.stock_daily(code, start_query, end_query) data.index.names = ['datetime'] data_feed = bt.feeds.PandasData(dataname=data, fromdate=from_date, todate=to_date) cerebro.adddata(data_feed, name=code) end_date = data.index[-1] # 股票剔除日 print('增加股票数据：code: %s' % code)cerebro.broker.setcash(100000000.0)cerebro.broker.setcommission(commission=0.001)cerebro.addstrategy(SingleTestStrategy, maperiod=20)cerebro.run()cerebro.plot()if __name__ == '__main__': pass后果：
em em ... 一个亿的资产，亏了靠近 1000w， 策略失败!!!!
来看看回测图，backtrader 的回测图确实有点丑哈，前面会有重构可视化篇的。<br/>
最下面是资产剖析图，行情数据区域的绿色三角形是买入，红色三角形是卖出。
多股回测
单股回测，下面曾经介绍了，那如何多个股同时回测呢？在这个问题上，咱们首先要解决的是多个股的指标计算并存储起来。
策略逻辑和下面的雷同。 计算均线的时候用了dict循环计算每只股票的指标。
- self.getdatanames()按程序返回所有股票的名称list
- self.getdatabyname(secu_name):返回该股票的data
所以，在给大脑塞数据时，须要指定 feedData 的 name ,对立用股票代码赋值，这样不便前面的索引。
间接上图说下整个流程逻辑：
代码只是再单股回测的根底下增加多股指标和多股持仓交易判断，策略和单股雷同。<br/>
代码如下：<br/>
import datetimeimport backtrader as btimport pandas as pdimport stock_db as sdbclass MultiTestStrategy(bt.Strategy): params = ( ('maperiod', 20), ) def prenext(self): pass def downcast(self, amount, lot): return abs(amount // lot * lot) def __init__(self): # 初始化交易指令 self.order = None self.buy_list = [] # 增加挪动平均线指标，循环计算每个股票的指标 self.sma = {x: bt.ind.SMA(self.getdatabyname(x), period=self.p.maperiod) for x in self.getdatanames()} def next(self): if self.order: # 查看是否有指令期待执行 return # 如果是最初一天，不进行交易 if pd.Timestamp(self.datas[0].datetime.date(0)) == end_dates[self.datas[0]._name]: return # 是否持仓 if len(self.buy_list) < 2: # 没有持仓 # 没有购买的票 for secu in set(self.getdatanames()) - set(self.buy_list): data = self.getdatabyname(secu) # 如果冲破 20 日均线买买买，不要在最初一根bar的前一天买 if data.close > self.sma[secu] and pd.Timestamp(data.datetime.date(1)) < end_dates[secu]: # 买买买 order_value = self.broker.getvalue() * 0.48 order_amount = self.downcast(order_value / data.close[0], 100) self.order = self.buy(data, size=order_amount, name=secu) self.log(f"买{secu}, price:{data.close[0]:.2f}, amout: {order_amount}") self.buy_list.append(secu) elif self.position: now_lst = [] for secu in self.buy_list: data = self.getdatabyname(secu) # 执行卖出条件判断：收盘价格跌破20日均线，或者股票最初一根bar 的前一天之剔除日 if data.close[0] < self.sma[secu] or pd.Timestamp(data.datetime.date(1)) >= end_dates[secu]: # 卖卖卖 self.order = self.order_target_percent(data, 0, name=secu) self.log(f"卖{secu}, price:{data.close[0]:.2f}, pct: 0") continue now_lst.append(secu) self.buy_list = now_lst 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, order.Canceled, order.Margin]: if order.isbuy(): self.log(f"""买入{order.info['name']}, 成交量{order.executed.size}，成交价{order.executed.price:.2f}""") self.log( f'资产：{self.broker.getvalue():.2f} 持仓：{[(x, self.getpositionbyname(x).size) for x in self.buy_list]}') elif order.issell(): self.log(f"""卖出{order.info['name']}, 成交量{order.executed.size}，成交价{order.executed.price:.2f}""") self.log( f'资产：{self.broker.getvalue():.2f} 持仓：{[(x, self.getpositionbyname(x).size) for x in self.buy_list]}') self.bar_executed = len(self) # Write down: no pending order self.order = None def log(self, txt, dt=None): """ 输入日期 :param txt: :param dt: :return: """ dt = dt or self.datetime.date(0) # 当初的日期 print('%s , %s' % (dt.isoformat(), txt))# 开始查问工夫start_query = '2019-01-01'end_query = '2022-09-01'# 开始回测工夫from_date = datetime.datetime(2022, 1, 1)to_date = datetime.datetime(2022, 10, 10)cerebro = bt.Cerebro()# 增加几个股票数据codes = [ '300015', '300347', # '300760', # '603127', # '600438']# 增加多个股票回测数据end_dates = {}end_date = 0for code in codes: data = sdb.stock_daily(code, start_query, end_query) data.index.names = ['datetime'] data_feed = bt.feeds.PandasData(dataname=data, fromdate=from_date, todate=to_date) cerebro.adddata(data_feed, name=code) end_dates[code] = data.index[-1] # 股票剔除日 end_date = data.index[-1] # 股票剔除日 print('增加股票数据：code: %s' % code)cerebro.broker.setcash(100000000.0)cerebro.broker.setcommission(commission=0.001)cerebro.addstrategy(MultiTestStrategy, maperiod=20)cerebro.run()# 获取回测完结后的总资金portvalue = cerebro.broker.getvalue()# 打印后果print(f'完结资金: {round(portvalue, 2)}')cerebro.plot()if __name__ == '__main__': pass看下日志：
enen ，竟然赚钱了，小赚靠近 1000w.
看下 backtracder 的交易点：
前面日期如同没触发交易逻辑，这前面再看看是咋回事。
写于 2022 年 10 月 23 日 10:27:29
本文由mdnice多平台公布