In the previous article we discussed the type of events that the system is prepared to handle. In this article we’ll build on that and focus on the Strategy component of the engine.
In line with the original series on Quanstart, I’ll use the classic Buy & Hold strategy for illustration purposes, but will also explain how to implement a more involved strategy.
The buy & hold strategy can be summarised as: buy an asset and don’t sell it. Thus, in terms of signals, in such strategy we’ll only be generating BUY signals, one for each asset. A more advanced buy & hold strategy can employ filters of all types, i.e. buy only assets that have certain characteristics, but to keep it simple, we’ll naively generate BUY signals for all the assets passed from the command line.
The final code looks like:
# trading_engine/strategy.py
from abc import ABC, abstractmethod
from events import SignalEvent
class Strategy(ABC):
"""
Strategy is an abstract base class providing an interface for all inherited strategy handling objects.
The goal of a derived Strategy class object is to generate Signal objects for particular symbols based
on the input Bars (OHLCVI) generated by the DataHandler object.
This is designed to work with both live and historical data, since the Strategy object obtains the bar
tuples from a queue object.
"""
@abstractmethod
def calculate_signals(self):
"""
Provides the mechanics to calculate the list of signals.
"""
raise NotImplementedError("Should implement calculate_signals()")
class BuyAndHoldStrategy(Strategy):
"""
Simple buy & hold strategy: go long on all symbols and never exit a position.
It's a useful benchmark for other strategies.
"""
def __init__(self, bars, events):
"""
Initialise the buy & hold strategy object.
:param bars: the DataHandler object that provides bar information
:param events: the Event queue object
"""
self.bars = bars
self.symbol_list = self.bars.symbol_list
self.events = events
self.bought = self._calculate_initial_bought()
def _calculate_initial_bought(self):
"""
Add keys to the dictionary that tracks bought symbols for all symbols in the list and sets them to False
"""
bought = {}
for s in self.symbol_list:
bought[s] = False
return bought
def calculate_signals(self, event):
"""
For a buy & hold strategy the signal is simply a 'LONG' or 'BUY' for each symbol in the list.
:param event: a MarketEvent
"""
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] is False:
# (Symbol, datetime, type = "LONG" || "SHORT")
signal = SignalEvent(bars[0][0], bars[0][1], 'LONG')
self.events.put(signal)
self.bought[s] = True
The key here is the calculate_signal() method. It takes in a MarketEvent,
goes through each symbol in the list, generates a BUY SignalEvent for each,
and adds them to the event queue.
In a more complex strategy, you will very likely also generate SELL signals. For example in a Statistical Arbitrage strategy you will need to buy and sell combinations of assets, almost simultaneously, based on what the statistical model suggests. I have implemented one myself and the pseudo-code should look something like this:
# [...]
class StatArbStrategy(Strategy):
"""
Statistical arbitrage strategy based on some model.
"""
def __init__(self, bars, events):
"""
Initialises the statistical arbitrage strategy.
:param bars: the DataHandler object that provides bar information
:param events: the Event queue object
"""
self.bars = bars
self.symbol_list = self.bars.symbol_list
self.events = events
self.exposure = self._calculate_initial_exposure()
def _calculate_initial_exposure(self):
"""
Set initial market exposue to 0/False.
:return: transacted dict with symbols as keys and values as the type of
exposure, i.e. LONG or SHORT -> dict
"""
transacted = {}
for s in self.symbol_list:
transacted[s] = False
return transacted
def calculate_signals(self, event):
"""
Compute the entry and exit signals in line with strategy.
:param 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 != []:
# this is a good point to inject the statistical model
model = super_duper_money_printing_strat()
if model.trade is True:
# (Symbol, datetime, type = LONG; SHORT; or EXIT)
signal = SignalEvent(bars[0][0], bars[0][1], model.trade_direction)
self.events.put(signal)
self.exposure[s] = model.trade_direction
The key line is:
model = super_duper_money_printing_strat()
Which is a placeholder for where you could inject your actual model’s parameters.
That’s it for signals.
In the next article we will cover the Portfolio component. It is a central part of the system, so we will spend extra time on it. But as usual, we’ll look at tests for the current component before moving on.
Strategy tests#
# trading_engine/tests/test_strategies.py
import queue
import pytest
from data import HistoricCsvDataHandler
from strategy import BuyAndHoldStrategy
from events import MarketEvent, SignalEvent
from utils import data_dir_setup, asset_class_selector, cli_parser
@pytest.fixture
def strategy_setup():
events = queue.Queue()
args = ["--asset_class", "stocks", "--ticker", "AMD INTC"]
parsed_args = cli_parser(args)
asset_class = asset_class_selector(parsed_args)
data_dir = data_dir_setup(asset_class)
symbols = ["AMD", "INTC"]
bars = HistoricCsvDataHandler(events, data_dir, symbols)
return events, bars
class TestBuyAndHoldStrategy:
def test_calculate_signals(self, strategy_setup):
events, bars = strategy_setup
bars.update_bars()
strategy = BuyAndHoldStrategy(bars, events)
market_event = MarketEvent()
strategy.calculate_signals(market_event) # triggers a SignalEvent
# test that the event queue has been updated with a SignalEvent
assert events.empty() is False
events.get(False)
assert type(events.get(False)) == SignalEvent
Feel free to share the article on socials: