Event-Driven Backtesting with Python - Part IV

Event-Driven Backtesting with Python - Part IV

The discussion of the event-driven backtesting implementation has previously considered the event-loop, the event class hierarchy and the data handling component. In this article a Strategy class hierarchy will be outlined. Strategy objects take market data as input and produce trading signal events as output.

A Strategy object encapsulates all calculations on market data that generate advisory signals to a Portfolio object. At this stage in the event-driven backtester development there is no concept of an indicator or filter, such as those found in technical trading. These are also good candidates for creating a class hierarchy but are beyond the scope of this article.

The strategy hierarchy is relatively simple as it consists of an abstract base class with a single pure virtual method for generating SignalEvent objects. In order to create the Strategy hierarchy it is necessary to import NumPy, pandas, the Queue object, abstract base class tools and the SignalEvent:

# strategy.py

import datetime
import numpy as np
import pandas as pd
import Queue

from abc import ABCMeta, abstractmethod

from event import SignalEvent

The Strategy abstract base class simply defines a pure virtual calculate_signals method. In derived classes this is used to handle the generation of SignalEvent objects based on market data updates:

# strategy.py

class Strategy(object):
    Strategy is an abstract base class providing an interface for
    all subsequent (inherited) strategy handling objects.

    The goal of a (derived) Strategy object is to generate Signal
    objects for particular symbols based on the inputs of Bars 
    (OLHCVI) generated by a DataHandler object.

    This is designed to work both with historic and live data as
    the Strategy object is agnostic to the data source,
    since it obtains the bar tuples from a queue object.

    __metaclass__ = ABCMeta

    def calculate_signals(self):
        Provides the mechanisms to calculate the list of signals.
        raise NotImplementedError("Should implement calculate_signals()")

The definition of the Strategy ABC is straightforward. Our first example of subclassing the Strategy object makes use of a buy and hold strategy to create the BuyAndHoldStrategy class. This simply goes long in a particular security on a certain date and keeps it within the portfolio. Thus only one signal per security is ever generated.

The constructor (__init__) requires the bars market data handler and the events event queue object:

# strategy.py

class BuyAndHoldStrategy(Strategy):
    This is an extremely simple strategy that goes LONG all of the 
    symbols as soon as a bar is received. It will never exit a position.

    It is primarily used as a testing mechanism for the Strategy class
    as well as a benchmark upon which to compare other strategies.

    def __init__(self, bars, events):
        Initialises the buy and hold strategy.

        bars - The DataHandler object that provides bar information
        events - The Event Queue object.
        self.bars = bars
        self.symbol_list = self.bars.symbol_list
        self.events = events

        # Once buy & hold signal is given, these are set to True
        self.bought = self._calculate_initial_bought()

On initialisation of the BuyAndHoldStrategy the bought dictionary member has a set of keys for each symbol that are all set to False. Once the asset has been "longed" then this is set to True. Essentially this allows the Strategy to know whether it is "in the market" or not:

# strategy.py

    def _calculate_initial_bought(self):
        Adds keys to the bought dictionary for all symbols
        and sets them to False.
        bought = {}
        for s in self.symbol_list:
            bought[s] = False
        return bought

The calculate_signals pure virtual method is implemented concretely in this class. The method loops over all symbols in the symbol list and retrieves the latest bar from the bars data handler. It then checks whether that symbol has been "bought" (i.e. whether we're in the market for this symbol or not) and if not creates a single SignalEvent object. This is then placed on the events queue and the bought dictionary is correctly updated to True for this particular symbol key:

# strategy.py

    def calculate_signals(self, event):
        For "Buy and Hold" we generate a single signal per symbol
        and then no additional signals. This means we are 
        constantly long the market from the date of strategy

        event - A MarketEvent object. 
        if event.type == 'MARKET':
            for s in self.symbol_list:
                bars = self.bars.get_latest_bars(s, N=1)
                if bars is not None and bars != []:
                    if self.bought[s] == False:
                        # (Symbol, Datetime, Type = LONG, SHORT or EXIT)
                        signal = SignalEvent(bars[0][0], bars[0][1], 'LONG')
                        self.bought[s] = True

This is clearly a simple strategy but it is sufficient to demonstrate the nature of an event-driven strategy hierarchy. In subsequent articles we will consider more sophisticated strategies such as a pairs trade. In the next article we will consider how to create a Portfolio hierarchy that keeps track of our positions with a profit and loss ("PnL").