Backtrader - 指标使用

以下内容皆参考 Backtrader 官网

在评估股票的时候，我们常常会用一些指标来辅助，今天来介绍的是移动平均线，像是 5日线、月线、季线...等。在 backtrader 里，移动平均线也和 Open, High, Low, Close...等一样的，我们可以在某个时间点，取出对应的值来进行评估。

程序码如下:

class TestStrategy(bt.Strategy):
    #...略
    
    def __init__(self):
        #...略
        
        # period 就是我们要取几日去做平均
        self.sma = bt.indicators.SimpleMovingAverage(self.data, period=5)

如果想要懒一点，self.data 预设就是第一组数据，除非你不是用第一组，不然可以省略

假设我们今天要做一个策略如下:

• 价格涨破 5日线就买入
• 价格跌破 5日线就卖出
• 一买一卖

程序码如下:

class TestStrategy(bt.Strategy):
    #...略
    
    def next(self):
        self.log("收盘价: %.2f" % self.dataclose[0])
        
        if self.order:
            return
            
        if not self.position:
            if self.dataclose[0] > self.sma[0]:
                self.log("买入 %.2f" % self.dataclose[0])
                self.order = self.buy()
        else:
            if self.dataclose[0] < self.sma[0]:
                self.log("卖出 %.2f" % self.dataclose[0])
                self.order = self.sell()

这边如果你有下去跑，而且输出结果来看的话，会发现它起始的日期不是 2020-01-02，因为移动平均线是前几是的收盘价的平圴，所以它会累积到足够的数据才会开始执行回测的策略。

再来，配合昨天的参数化，把移动平均线参数化

class TestStrategy(bt.Strategy):
    params = (
        ( "maperiod", 5 ),
        ( "defaultSize", 1000 ),
        ( "printLog", False ),
    )
    
    #...略
    
    def __init__(self):
        self.dataclose[0] = self.data.close
        self.sizer.setsizing(self.p.defaultSize)
        self.sma = bt.indicators.SimpleMovingAverage(period = self.p.maperiod)
        
        self.order = None

有了参数化之後，我们还可以进行参数的最佳化，以移动平均来说，要用 5日、月线或者是季线来当我们的指标呢？这里就可以指定多组参数，让 backtrader 去执行回测，我们就可以看到不同值结果。

cerebro.optstrategy(
    TestStrategy,
    maperiod=range(10, 31)
)

因为我们有多组的参数去跑回测，所以我们关注的结果就是每个参数的结果，要取得这个结果，可以利用策略里的 stop 事件，当回测跑完之後就会去执行这一个程序

class TestStrategy(bt.Strategy):
    #...略
    def stop(self):
        self.log("MA Period %2d 结束收益 %.2f" % (self.p.maperiod, self.broker.getvalue()), doPrint = True)

完整程序码如下:

'''
- 低於 均线 买入
- 高於 均线 卖出
- 一买一卖
'''
import backtrader as bt

class TestStrategy(bt.Strategy):

    params = (
        ('maperiod', 15),
        ('defaultSize', 1000),
        ('printlog', False),
    )

    def log(self, txt, dt = None, doPrint = False):
        ''' Log function for this strategy '''
        if self.params.printlog or doPrint:
            dt = dt or self.datas[0].datetime.date(0)
            print("%s %s" % (dt.isoformat(), txt))

    def __init__(self):
        self.dataclose = self.data.close
        self.sizer.setsizing(self.p.defaultSize)
        self.sma = bt.indicators.SimpleMovingAverage(period = self.p.maperiod)

        # pendding order 
        self.order = None


    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("买入 价格: %.2f 成本: %.2f 手续费: %.2f" % (order.executed.price, order.executed.value, order.executed.comm))
            elif order.issell():
                self.log("卖出 价格: %.2f 成本: %.2f 手续费: %.2f" % (order.executed.price, order.executed.value, order.executed.comm))

            self.bar_executed = len(self)

        elif order.status in [order.Canceled, order.Margin, order.Rejected]:
            self.log("交易取消/余额不足/拒绝交易")

        self.order = None

    def notify_trade(self, trade):
        if not trade.isclosed:
            return

        self.log("交易收益: 毛利 %.2f 净利 %.2f" % (trade.pnl, trade.pnlcomm))

    def next(self):
        self.log("收盘价: %.2f" % self.dataclose[0])

        if self.order:
            return

        if not self.position:

            if self.dataclose[0] > self.sma[0]:
                self.log("买入 %.2f" % self.dataclose[0])
                self.order = self.buy()

        else:

            if self.dataclose[0] < self.sma[0]:
                self.log("卖出 %.2f" % self.dataclose[0])
                self.order = self.sell()

    def stop(self):
        self.log('MA Period %2d  结束收益 %.2f' % (self.params.maperiod, self.broker.getvalue()), doPrint = True)
        
        

cerebro = bt.Cerebro()

# set strategy
cerebro.optstrategy(
    TestStrategy,
    maperiod = range(10, 31)
)

# set cash 
cerebro.broker.setcash(10**6)

# set commission
cerebro.broker.setcommission(commission=0.1425/100)

data = bt.feeds.PandasData(dataname=df, timeframe=bt.TimeFrame.Minutes)

cerebro.resampledata(data, timeframe=bt.TimeFrame.Days)

cerebro.run()