量化专题——backtrader多周期回测 01

在backtrader官网文档中，介绍了两种多周期回测方式，第一种是读入多周期的K线数据，第二种是对数据进行resample。本文将对第一种方式进行介绍。

简述

有些策略会使用到多周期的数据，典型的应用为：

• 使用周线（大周期）数据判断趋势

• 使用日线（小周期）数据判断买卖点

这就需要同时读入多周期数据进行回测，backtrader 内置了对多周期策略回测的支持。

backtrader多周期回测规则

回测时需要遵循以下规则：

• 小周期数据必须被第一个加入到Cerebro实例中

在backtrader中，第一个被添加的数据将被作为时钟数据，因此需要将小周期数据首先添加到系统中，以使得小力度的时间都能被遍历到。

• 数据必须按照日期时间做好对齐，这样backtrader才能正确地使用数据

backtrader在回测过程中，不会对时间进行重新排序，只能按照整理好的数据顺序依次处理，因此需要将不同周期的数据都做好对齐。很幸运，从TuShare及BaoStock下载下来的不同周期的数据均做好对齐，可以直接使用，注意数据需要按时间升序组织。

• 大周期数据的使用会使得策略的最小周期变大。

如果要计算出技术指标的有效值，至少需要若干根K线数据，例如：计算5日均线，则至少需要前5根日K线，才能计算出第一个有效值，也就是至少经过5日，这里的5日就是backtrader里所指的最小周期。不同指标可能具有不同的最小周期，单周期回测时，策略的最小周期就是所有指标最小周期的最大值。而多周期回测中，最小周期则变得复杂。通常，要计算出大周期技术指标的有效值，策略的最小周期会变大，也就是回测开始后，要经过更多的K线来保证大周期技术指标能够计算出第一个有效值。

示例

为了演示多周期策略回测，本文使用以下方案：

策略将先读入日线数据，再读入周线数据。回测股票为000001平安银行，回测周期为2018年1月1日至2019年12月31日。

# 加载数据
def load_data(data_info):
    fromdate = datetime.datetime(2018, 1, 1)
    todate = datetime.datetime(2019, 12, 31)
    datapath = './stk_data/' + data_info[0] +'/' + data_info[1] + '.csv'
    return bt.feeds.GenericCSVData(
        dataname = datapath,
        fromdate = fromdate,
        todate = todate + datetime.timedelta(days=1),
        nullvalue = 0.0,
        dtformat = ('%Y-%m-%d'),
        datetime = 0,
        open = 1,
        high = 2,
        low = 3,
        close = 4,
        volume = 5,
        openinterest = -1
        )

    for stk_code in ['sz.000001']:
        data_list = list(map(load_data, [('d', stk_code), ('w', stk_code)]))
        for data in data_list:
            cerebro.adddata(data)

平安银行的日线及周线数据分别保存在目录 ‘./stk_data/d/sz.000001.csv’ 和 ‘./stk_data/w/sz.000001.csv’ 中，通过load_data方法先后读入日线和K线数据，保存在data_list之后，依次添加到cerebro中，等待系统回测。

• 买入条件：日MACD金叉、周RSI小于50；卖出条件：价格较最高收盘价回撤5%卖出。

• 在策略类的init方法中，定义所需的技术指标：

def __init__(self):
      # 存储不同数据的技术指标
      self.inds = dict()
      # 存储特定股票的订单，key为股票的代码
      self.orders = dict()
      # 遍历所有数据
      for i, d in enumerate(self.datas):
          self.orders[d._name] = None
          # 为每个数据定义字典，存储技术指标
          self.inds[d] = dict()
          # 判断d是否为日线数据
          if 0 == i % 2:
              self.inds[d]['crossup'] = btind.CrossUp(btind.MACD(d).macd, btind.MACD(d).signal)
          # d为周线数据
          else:
              self.inds[d]['rsi'] = btind.RSI_Safe(d)

定义字典self.inds，来存储不同数据的技术指标。

定义self.orders，来存储特定股票的订单，key为股票的代码。

然后遍历所有的数据，将数据的订单先置为空，并且为每个数据创建字典，来存储技术指标。由于系统是先添加日线数据，再添加周线数据，因此当i % 2等于0时，d为日线数据，那么就计算MACD金叉指标；当i % 2不等于0时，d为周线数据，那么计算RSI指标。

在策略类的next方法中，定义买入卖出条件：

def next(self):
       #print(self.datetime.date())
       for i, d in enumerate(self.datas):
           # 如果处理周线数据则跳过买卖条件，因为已在日线数据判断处理过
           if 1 == i % 2:
               continue
           pos = self.getposition(d)
           # 不在场内，则可以买入
           if not len(pos):
               # 达到买入条件
               if self.inds[d]['crossup'][0] and self.inds[self.datas[i + 1]]['rsi'][0] < 50:
                   # 买入手数，如果是多只股票回测，这里需要修改
                   stake = int(self.broker.cash // (d.close[0] * 100)) * 100
                   # 买买买
                   self.buy(data = d, size = stake)
           elif not self.orders[d._name]:
               # 下保护点卖单
               self.orders[d._name] = self.close(data = d, exectype= bt.Order.StopTrail,
                           trailamount=self.p.trailamount,
                           trailpercent=self.p.trailpercent)

这里依然只对i % 2等于0的数据（即日线数据）进行条件判断，然后使用self.inds[self.datas[i + 1]][‘rsi’]的方式对周线技术指标进行访问。

当达到买入条件后下买单，订单将在第二天以开盘价成交。在计算买入仓位大小时，保证资金得到最大程度的使用。需要注意，如果是要对多只股票同时进行回测，这里需要做资金分配。

买单成交当天，下StopTrail卖单，当股价较最高收盘价回撤5%卖出（具体参加笔记20和笔记31）。这里使用close方法而不是sell方法，如果使用sell方法，股票将以1股1股的卖出，使用close则是全部卖出。

输出结果为：

018-05-15 BUY sz.000001 EXECUTED, Price: 11.18
2018-05-25 SELL sz.000001 EXECUTED, Price: 10.56
2018-06-19 BUY sz.000001 EXECUTED, Price: 10.05
2018-06-25 SELL sz.000001 EXECUTED, Price: 9.41
2018-07-11 BUY sz.000001 EXECUTED, Price: 8.76
2018-08-02 SELL sz.000001 EXECUTED, Price: 8.95
2019-01-08 BUY sz.000001 EXECUTED, Price: 9.73
2019-03-08 SELL sz.000001 EXECUTED, Price: 12.43
2019-06-11 BUY sz.000001 EXECUTED, Price: 12.34
2019-06-25 SELL sz.000001 EXECUTED, Price: 13.11
Final Portfolio Value: 1226710.85

两年间共有5笔交易，收益率为22.7%

总结

backtrader可以通过加载不同周期的数据来实现多周期策略回测。

加载不同周期数据进行策略回测时，需要先加载小周期数据，后加载大周期数据。

在策略实现时，通过数据索引的取余运算来区分小、大周期数据。

多周期策略回测程序v1代码：

# 多周期
# 买入条件：日MACD金叉、周RSI小于50
# 卖出条件：价格较最高收盘价回撤5%卖出
from __future__ import (absolute_import, division, print_function,
                        unicode_literals)

import backtrader as bt
import backtrader.feeds as btfeeds
import backtrader.indicators as btind
import datetime
import pandas as pd
from collections import defaultdict

class RSIMACDMultiTF(bt.Strategy):
    params = (
        ('trailamount', 0.0),
        ('trailpercent', 0.05),
    )

    def __init__(self):
        # 存储不同数据的技术指标
        self.inds = dict()
        # 存储特定股票的订单，key为股票的代码
        self.orders = dict()
        # 遍历所有数据
        for i, d in enumerate(self.datas):
            self.orders[d._name] = None
            # 为每个数据定义字典，存储技术指标
            self.inds[d] = dict()
            # 判断d是否为日线数据
            if 0 == i % 2:
                self.inds[d]['crossup'] = btind.CrossUp(btind.MACD(d).macd, btind.MACD(d).signal)
            # d为周线数据
            else:
                self.inds[d]['rsi'] = btind.RSI_Safe(d)

    def next(self):
        #print(self.datetime.date())
        for i, d in enumerate(self.datas):
            # 如果处理周线数据则跳过买卖条件，因为已在日线数据判断处理过
            if 1 == i % 2:
                continue
            pos = self.getposition(d)
            # 不在场内，则可以买入
            if not len(pos):
                # 达到买入条件
                if self.inds[d]['crossup'][0] and self.inds[self.datas[i + 1]]['rsi'][0] < 50:
                    # 买入手数，如果是多只股票回测，这里需要修改
                    stake = int(self.broker.cash // (d.close[0] * 100)) * 100
                    # 买买买
                    self.buy(data = d, size = stake)
            elif not self.orders[d._name]:
                # 下保护点卖单
                self.orders[d._name] = self.close(data = d, exectype= bt.Order.StopTrail,
                            trailamount=self.p.trailamount,
                            trailpercent=self.p.trailpercent)

    def notify_order(self, order):

        if order.status in [order.Completed]:
            if order.isbuy():
                print('{} BUY {} EXECUTED, Price: {:.2f}'.format(self.datetime.date(), order.data._name, order.executed.price))
            else:  # Sell
                self.orders[order.data._name] = None
                print('{} SELL {} EXECUTED, Price: {:.2f}'.format(self.datetime.date(), order.data._name, order.executed.price))

# 加载数据
def load_data(data_info):
    fromdate = datetime.datetime(2018, 1, 1)
    todate = datetime.datetime(2019, 12, 31)
    datapath = './stk_data/' + data_info[0] +'/' + data_info[1] + '.csv'
    return bt.feeds.GenericCSVData(
        dataname = datapath,
        fromdate = fromdate,
        todate = todate + datetime.timedelta(days=1),
        nullvalue = 0.0,
        dtformat = ('%Y-%m-%d'),
        datetime = 0,
        open = 1,
        high = 2,
        low = 3,
        close = 4,
        volume = 5,
        openinterest = -1
        )

def runstrat():
    cerebro = bt.Cerebro()
    cerebro.broker.setcash(1000000.0)
    cerebro.addstrategy(RSIMACDMultiTF)

    for stk_code in ['sz.000001']:
        data_list = list(map(load_data, [('d', stk_code), ('w', stk_code)]))
        for data in data_list:
            cerebro.adddata(data)
    cerebro.addwriter(bt.WriterFile, out = 'log.csv', csv = True)
    cerebro.run()
    print('Final Portfolio Value: %.2f' % cerebro.broker.getvalue())
    # Plot the result绘制结果
    cerebro.plot(start=datetime.date(2018, 1, 1), end=datetime.date(2019, 12, 31),
            volume = False, style = 'candle',
            barup = 'red', bardown = 'green')

if __name__ == '__main__':
    runstrat()

---The End---