Backtrader: Creating Analyzers

We have covered using Backtrader’s analyzers in an earlier post. At that time, we looked at using the built-in TradeAnalyzer and SQN to provide some meaningful feedback as to how our strategy performed. In this post  we shall go a step further and create our own analyzer. Backtrader might not be the first thing that comes to mind when thinking about statistical analysis. However, there are some positives to consider that make it a good choice. Namely, you can use a platform you are familiar with. No need to spend hours running through tutorials for other frameworks “better suited” to statistical analysis. If you can leverage your Backtrader skills to do the same in less time why not? Another benefit is that analyzers can provide immediate feedback in real-time or as soon as the Backtest ends. No need to export your final data into another framework for further analysis.

Creating an Analyzer

We are going to create a simple analyzer to count the number of times price action hits a pivot indicators p, s1, s2, r1 and r2 lines. Knowing how frequently the pivot is hit might be useful in helping to find an edge. Alternatively, it may also be useful in determining stop placement if you know that s2 is only hit x% of the time. These are just examples and the code in the post is intended to spark ideas only. You may want to expand on the analysis in the code or look at another area which interests you.

Pivot Points Analyzer

import backtrader as bt
from backtrader import Analyzer
from backtrader.utils import AutoOrderedDict
from backtrader.indicators import PivotPoint

class pivotPointAnalyzer(Analyzer):
    '''
    Analyzer to return some basic statistics showing:
        - Total days/periods analyzed
        - Percentage p was touched
        - Percentage s1 was touched
        - Percentage s2 was touched
        - Percentage r1 was touched
        - Percentage r2 was touched
        - Percentage s1 and r1 were touched on the same day / period
        - Percentage s1 and r2 were touched on the same day / period
        - Percentage s2 and r1 were touched on the same day / period
        - Percentage s2 and r1 were touched on the same day / period
        - Percentage p was touched without touching s1
        - Percentage p was touched without touching r1
        - Percentage s1 was touched without touching r1
        - Percentage r1 was touched without touching s1
    '''

    def __init__(self):
        #data1 is the resampled data
        self.pivots = PivotPoint()
        #We don't want to print an analyzer
        self.pivots.autoplot = False


    def create_analysis(self):
        hit_desc = ('Mesures the frequency in percent that each pivot\n      '
            'was hit over the course of the time period analyzed\n')

        db_desc = ('Mesures the frequency in percent that a pair of pivots\n      '
            'were hit on the same day over the course of the time period analyzed\n')

        xy_desc = ('Mesures the frequency in percent that one pivot (x)\n      '
            'was hit without hitting another pivot (y) on the same day\n      '
            'over the course of the time period analyzed\n')

        self.rets = AutoOrderedDict()
        self.counts = AutoOrderedDict()
        self.counts.total = 0

        self.counts['Hit']['R1'] = 0
        self.counts['Hit']['R2'] = 0
        self.counts['Hit']['P'] = 0
        self.counts['Hit']['S1'] = 0
        self.counts['Hit']['S2'] = 0

        self.counts['DB']['S1 & R1'] = 0
        self.counts['DB']['S1 & R2'] = 0
        self.counts['DB']['S2 & R1'] = 0
        self.counts['DB']['S2 & R2'] = 0

        self.counts['XY']['x=P y=R2'] = 0
        self.counts['XY']['x=P y=R1'] = 0
        self.counts['XY']['x=P y=S2'] = 0
        self.counts['XY']['x=P y=S1'] = 0
        self.counts['XY']['x=R1 y=S2'] = 0
        self.counts['XY']['x=R1 y=S1'] = 0
        self.counts['XY']['x=R1 y=P'] = 0
        self.counts['XY']['x=R2 y=S2'] = 0
        self.counts['XY']['x=R2 y=S1'] = 0
        self.counts['XY']['x=R2 y=P'] = 0
        self.counts['XY']['x=S1 y=R2'] = 0
        self.counts['XY']['x=S1 y=R1'] = 0
        self.counts['XY']['x=S1 y=P'] = 0
        self.counts['XY']['x=S2 y=R2'] = 0
        self.counts['XY']['x=S2 y=R1'] = 0
        self.counts['XY']['x=S2 y=P'] = 0

        self.rets['Hit']['Description'] = hit_desc
        self.rets['Hit']['R1'] = 0
        self.rets['Hit']['R2'] = 0
        self.rets['Hit']['P'] = 0
        self.rets['Hit']['S1'] = 0
        self.rets['Hit']['S2'] = 0

        self.rets['DB']['Description'] = db_desc
        self.rets['DB']['S1 & R1'] = 0
        self.rets['DB']['S1 & R2'] = 0
        self.rets['DB']['S2 & R1'] = 0
        self.rets['DB']['S2 & R2'] = 0

        self.rets['XY']['Description'] = xy_desc
        self.rets['XY']['x=P y=R2'] = 0
        self.rets['XY']['x=P y=R1'] = 0
        self.rets['XY']['x=P y=S2'] = 0
        self.rets['XY']['x=P y=S1'] = 0
        self.rets['XY']['x=R1 y=S2'] = 0
        self.rets['XY']['x=R1 y=S1'] = 0
        self.rets['XY']['x=R1 y=P'] = 0
        self.rets['XY']['x=R2 y=S2'] = 0
        self.rets['XY']['x=R2 y=S1'] = 0
        self.rets['XY']['x=R2 y=P'] = 0
        self.rets['XY']['x=S1 y=R2'] = 0
        self.rets['XY']['x=S1 y=R1'] = 0
        self.rets['XY']['x=S1 y=P'] = 0
        self.rets['XY']['x=S2 y=R2'] = 0
        self.rets['XY']['x=S2 y=R1'] = 0
        self.rets['XY']['x=S2 y=P'] = 0

    def next(self):
        r2 = self.pivots.lines.r2[-1]
        r1 = self.pivots.lines.r1[-1]
        p = self.pivots.lines.p[-1]
        s1 = self.pivots.lines.s1[-1]
        s2 = self.pivots.lines.s2[-1]

        o = self.data.open[0]
        h = self.data.high[0]
        l = self.data.low[0]
        c = self.data.close[0]

        pivots = [
            ['R2', r2],
            ['R1', r1],
            ['P', p],
            ['S1', s1],
            ['S2', s2]]

        for piv in pivots:
            #if piv[0] == 'r2':
                #print('h: {} L {}, piv {}'.format(h,l, piv[1]))
            if h > piv[1] and l < piv[1]:
                #print('h: {} L {}, piv {}'.format(h,l, piv[1]))
                #Pivot touched
                self.counts['Hit'][piv[0]] +=1

        db_pivots = [
            ['S1 & R1', s1, r1],
            ['S1 & R2', s1, r2],
            ['S2 & R1', s2, r1],
            ['S2 & R2', s2, r2]
        ]
        #DB
        for piv in db_pivots:
            db_conditions = [
                h > piv[1],
                h > piv[2],
                l < piv[1],
                l < piv[2],
            ]
            if all(db_conditions):
                self.counts['DB'][piv[0]] +=1

        #X without touching Y
        # Can probably build this in a nice progrmatic way.
        xy_pivots = [
        ['x=P y=R2', p, r2, 'r'],
        ['x=P y=R1', p, r1, 'r'],
        ['x=P y=S2', p, s2, 's'],
        ['x=P y=S1', p, s1,'s'],
        ['x=R1 y=S2', r1, s2, 's'],
        ['x=R1 y=S1', r1, s1, 's'],
        ['x=R1 y=P', r1, p, 'p'],
        ['x=R2 y=S2', r2, s2, 's'],
        ['x=R2 y=S1', r2, s1, 's'],
        ['x=R2 y=P', r2, p, 'p'],
        ['x=S1 y=R2', s1, r2, 'r'],
        ['x=S1 y=R1', s1, r1, 'r'],
        ['x=S1 y=P', s1, p, 'p'],
        ['x=S2 y=R2', s2, r2, 'r'],
        ['x=S2 y=R1', s2, r1, 'r'],
        ['x=S2 y=P', s2, p, 'p']
        ]
        for piv in xy_pivots:
            if piv[3] == 'r':
                db_conditions = [
                h > piv[1],
                l < piv[1],
                h < piv[2]
                ]
            elif piv[3] == 's':
                db_conditions = [
                h > piv[1],
                l < piv[1],
                l > piv[2]
                ]
            elif piv[3] == 'p':
                db_conditions = [
                h > piv[1],
                l < piv[1],
                (h > piv[2] and l > piv[2]) or (h < piv[2] and l < piv[2])
                ]

            if all(db_conditions):
                self.counts['XY'][piv[0]] +=1


    def stop(self):
        self.counts.total = len(self.data)
        #ITERATE OVER COUNTS SO THE DESCRIPTION DOES NOT CAUSE AN ERROR
        for key, value in self.counts['Hit'].items():
            try:
                perc = round((value / self.counts.total) * 100,2)
                self.rets['Hit'][key] = str(perc) + '%'
            except ZeroDivisionError:
                self.rets['Hit'][key] = '0%'

        #ITERATE OVER COUNTS SO THE DESCRIPTION DOES NOT CAUSE AN ERROR
        for key, value in self.counts['DB'].items():
            try:
                perc = round((value / self.counts.total) * 100,2)
                self.rets['DB'][key] = str(perc) + '%'
            except ZeroDivisionError:
                self.rets['DB'][key] = '0%'

        #ITERATE OVER COUNTS SO THE DESCRIPTION DOES NOT CAUSE AN ERROR
        for key, value in self.counts['XY'].items():
            try:

                perc = round((value / self.counts.total) * 100,2)
                self.rets['XY'][key] = str(perc) + '%'
            except ZeroDivisionError:
                self.rets['XY'][key] = '0%'

        self.rets._close()  # . notation cannot create more keys

Code Commentary

First things first. To create an analyzer we need to inherit from Backtraders Analyzerclass. Following this, the __init_() is fairly straightforward. We just import Backtrader’s PivotPointindicator and turn off plotting. Next we move onto the more interesting create_analysis() method. Here we setup the metrics that need to be tracked by the analyzer. Backtrader uses an AutoOrderedDict()for storing the metrics that you want to track. An ordered dictionary allows the analyzers inherited print() method to print the metrics in a fixed and defined order. If you have ever tried printing values from a normal dictionary, you may have noticed that the order in which the keys are printed can seem a bit random. (It is likely not random, but I certainly do not know the logic). Speaking of the print() method, there will be an example of how to call it later. Backtrader’s built-in analyzers use a naming convention for the dictionary that is used to store metrics to be printing. It is called self.rets. You will notice in the code example, I have one dictionary which follows this convention and one which does not. (counts and rets). The reasoning is because I am primarily interested in the percentage/frequency that the pivots are hit rather than the absolute number of times. As such, did not want the counts to be included in the rets dictionary and printed at the same time.

next()

Analyzers have a next() method just like indicators and strategies. As such this makes it easy to take your knowledge developed so far and apply it to an analyzer. The next() method for this analyzer checks to see if the high of the data is above the pivot AND the low of the data is below it. This signifies that the pivot was broken during the bar.

stop()

The stop() method is called at the end of the backtesting. This is also available in strategies and indicators. An example of this was shown in the post Backtrader: Live trading shutdown. In the code example above, we use stop() to build the final percentages from the count dict. If you wanted to access the analyzer during a run. I.e Live or access values during a strategies next() method, the analyzer could be improved to calculate the percentages every bar.

Adding the analyzer

You might have noticed in the code example above, no code has been provided to setup cerebro or define a strategy. This is because I want to keep the analyzer code modularized in a separate location for reuse. For an overview on how to create python modules for your Backtrader projects see this post: Backtrader: Making modular code. The analyzer will be imported into a strategy when required. Ok – So now lets assume you have this Analyzer stored in a module. To use the analyzer you would need to create a script like below:

'''
Helper script to test developed analyzers.
'''

import backtrader as bt
from datetime import datetime
from extensions.analyzers import pivotPointAnalyzer
from extensions.misc.datafeeds import OandaCSVMidData

#Create an instance of cerebro
cerebro = bt.Cerebro()


data = OandaCSVMidData(dataname='data/fx/GBP_USD-2005-2017-D1.csv',
                            timeframe=bt.TimeFrame.Days,
                            compression=1)

cerebro.adddata(data)

cerebro.addanalyzer(pivotPointAnalyzer)

# Run over everything and get the list of strategy objects
stratList = cerebro.run()

# get the first item in the list
strat = stratList[0]

for x in strat.analyzers:
    x.print()

cerebro.plot(style='candlestick')

The code above will not run out of the box. It is for example purposes only. Here is why:

• The code assumes the analyzer is saved in an sub module analyzers which is part of the extensions module. You could emulate this, create your own module structure or simply copy and paste the analyzer into the main script.
• The code uses my own data saved from Oanda. You will need to replace this part with your own data and data setup.

Adding the analyzer and getting the results is pretty easy. Once imported, you add it with a single line cerebro.addanalyzer(pivotPointAnalyzer). Then to access the results you can print them after the run using the lines:

cerebro.addanalyzer(pivotPointAnalyzer)

# Run over everything and get the list of strategy objects
stratList = cerebro.run()

# get the first item in the list
strat = stratList[0]

for x in strat.analyzers:
    x.print()

And that is all there is to it. Once setup and ran you should see some output that looks like this: