Backtrader（二十二）- 混合时间粒度回测

简述

当要使用不同时间粒度的行情数据时，backtrader有两种处理方式，第一种是读入不同的行情bar数据，第二种是将细粒度数据合并成粗粒度数据（resample）

读入不同的行情bar数据

规则

• 最小粒度的数据必须被第一个加入到Cerebro实例中
• 数据必须按照日期时间升序排列
• 大粒度数据的使用有可能使得策略的最小周期变大
• 若使用不同时间粒度的指标进行运算，则要采用cerebro.run(runonce=False) 若策略中使用了不同时间粒度的指标进行运算，不能用cerebro.run()
• init方法中不同粒度的指标执行混合运算时，大粒度的指标后要加括号()

使用resample合成大时间粒度数据

用法：

cerebro.resampledata(data, timeframe=bt.TimeFrame.Minutes, compression=1)
data: bt数据源
timeframe：用于指定合成结果(Ticks, MicroSeconds, Seconds, Minutes, Days, Weeks, Months, Years, NoTimeFrame)
compression：设置bar的条数进一步压缩合并，如resmaple初次合成为周数据后，每根bar是周线。如设置compression=2，则进一步将两条周数据bar压缩成 一根bar为两周数据。

注意：

如果采用了resmaple方法，则数据不会预加载preload。

1、日线合并成周线

'''重采样合成大粒度数据 日线合成周线'''

import datetime
import backtrader as bt
from feed import feed, feed_weekly
from logger import lg
import backtrader.indicators as btind


class SmaCross(bt.Strategy):

    def __init__(self):
        # 5日移动平均线指标
        self.move_average = bt.ind.MovingAverageSimple(
            self.data.close,
            period=5
        )
        # 2周线移动平均
        self.move_average2 = bt.ind.MovingAverageSimple(
            self.datas[1].close,
            period=2
        )
        # 日线交叉信号指标
        self.crossover = bt.ind.CrossOver(
            self.data.close,
            self.move_average
        )

    def log(self, txt, dt=None):
        dt = dt or self.data.datetime.date(0)
        lg.info(
            '%s, %s' % (dt.isoformat(), txt)
        )

    def notify_order(self, order):
        if order.status in [order.Submitted, order.Accepted]:
            return

        if order.status in [order.Completed]:
            if order.isbuy():
                self.log(
                    f'Buy {order.executed.price}'
                )
            elif order.issell():
                self.log(
                    f'Sell {order.executed.price}'
                )

    def notify_trade(self, trade):
        if trade.isclosed:
            self.log(
                f'''
            毛收益 {trade.pnl},
            扣佣后收益 {trade.pnlcomm},
            佣金 {trade.commission}
            '''
            )

    def next(self):
        if not self.position:
            if self.crossover > 0 and self.move_average[0] > self.move_average2[0]:
                self.buy()

        elif self.crossover < 0 and self.move_average[0] < self.move_average2[0]:
            self.close()

if __name__ == '__main__':
    cerebro = bt.Cerebro()
    cerebro.adddata(feed)

    cerebro.addstrategy(SmaCross)

    cerebro.resampledata(
        feed,
        timeframe=bt.TimeFrame.Weeks,
        compression=1
    )

    cerebro.broker.setcash(10000.0)
    cerebro.broker.setcommission(0.001)
    cerebro.broker.set_slippage_fixed(0.05)
    lg.error(f'市值 {cerebro.broker.getvalue()}')
    cerebro.run(stdstats=False, runonce=False)
    lg.error(f'市值 {cerebro.broker.getvalue()}')
    cerebro.plot()

2、5分钟线合并成10分钟线

'''重采样合成大粒度数据 5分钟线合成10分钟线'''
from datetime import datetime
import backtrader as bt



class TestStrategy(bt.Strategy):
    def next(self):
        print(
            f'''
        {self.data0.datetime.datetime(0)},
        {len(self.data0)},
        {self.data0.open[0]},
        {self.data0.high[0]},
        {self.data0.low[0]},
        {self.data0.close[0]},
        '''
        )

if __name__ == '__main__':
    cerebro = bt.Cerebro(stdstats=False)
    cerebro.addstrategy(TestStrategy)

    feed = bt.feeds.BacktraderCSVData(
        dataname='min.csv',
        timeframe=bt.TimeFrame.Minutes,
        todate=datetime(2016, 1, 3)
    )

    cerebro.resampledata(
        feed,
        timeframe=bt.TimeFrame.Minutes,
        compression=2  # 5min整合10分钟的关键
    )

    cerebro.run()

3、Tick数据合成 1分钟数据

'''Tick数据合成 1分钟数据'''

from datetime import datetime
import backtrader as bt


class TestStrategy(bt.Strategy):
    def next(self):
        print(
            f'''
        {self.data0.datetime.datetime(0)},
        {len(self.data0)},
        {self.data0.open[0]},
        {self.data0.high[0]},
        {self.data0.low[0]},
        {self.data0.close[0]},
        '''
        )

if __name__ == '__main__':
    cerebro = bt.Cerebro(stdstats=False)
    cerebro.addstrategy(TestStrategy)

    feed = bt.feeds.GenericCSVData(
        dataname='tick.csv',
        dtformat='%Y-%m-%dT%H:%M:%S.%f',
        timeframe=bt.TimeFrame.Ticks
    )

    cerebro.resampledata(
        feed,
        timeframe=bt.TimeFrame.Minutes,
        compression=1
    )

    cerebro.run()