Build Your Own Event-Based Backtester in Python
Table of Contents
When testing an investment strategy, a common way is called backtesting. Backtesting is when you run the algorithm on historic data as if you were trading at that moment in time and had no knowledge of the future. Although backtesters exist in Python, this flexible framework can be modified to parse more than just tick data– giving you a leg up in your testing. In this post, we’re going introduce a simple event-based backtester in Python which utilizes the multiprocessing library.
Basics
The ideal algorithm would perform well in a backtest because that indicates that– at some point in time– the algorithm worked. There are many pitfalls that people run into when making a backtester. Many of the issues have to do with your choice of data, but the design can be a problem as well. Some common problems in finance lingo are 1. Look Ahead Bias: Your backtester somehow has more (immediate) future information than it should. This can happen in “vector” based backtesters. 2. Survival Bias: Your algorithm’s stock universe is missing delisted (failed/bankrupt) stocks and is choosing stocks that exist in the present. Similar to look ahead bias in that your algorithm also “knows” part of the future.
Both of these are examples of what the machine learning community would call “leakage”. And in fact, all the tenets of machine learning are valid for building a trading algorithm. However, in this case, your training and test set are refered to as “in sample” and “out of sample” and refer to non-overlapping periods of time. Your algorithm could have nothing to do with machine learning at all, and could be as straightforward as storing prices over time and watching the “momentum” of the stocks.
Event-Based Backtesting
We avoid the problem of look ahead bias by making an event based backtester. This type of backtester receives a data feed, or “events”, which trigger the algorithm to respond in real time (or at least, as the data comes in). This is the most accurate and fool-proof way to avoid look-ahead bias, because literally, your algorithm will not have the data while it is making decisions.
Design
Because we will be reading each data point one at a time, we need the algorithm to be as fast as possible. Vector-based backtesters are usually the fastest kind (however it’s also easy to introduce errors), but we will try to use the multiprocessing module in Python to alleviate any performance concerns. As its name suggests, this module allows multiple pieces of Python code to run in different processes and communicate with each other.
We utilize what is called a Producer-Consumer pattern for parallelism. One process, the Producer, will read from a data source and fill a queue with parsed information. Meanwhile, the other process, the Consumer, will take that data and use it for the backtesting process.
The multiprocessing Module
The multiprocessing module allows us to run python code in multiple processes. The central class is called a Process which takes in a target function to run and arguments for the function. When a processes is started (via the aptly named .start() function), an identical copy of the current process is made– that is– all objects are transfered over in their current state to the new process. We can use a Queue class to communicate between each process. The boilerplate code is found in the Backtester class:
1
2
3
4
5
6
7
from multiprocessing import Process, Queue
q= Queue()
p = Process(target=DataSource.process, args=((q,))
p.start()
p.join # Halts the main process until the started process completes
The DataSource Class
The Producer class for the backtester is called the DataSource. The source is able to connect to anything, including a real time feed. One must just implement the get_data() method. In this basic example, the ‘POISON’ message can be sent from the source to kill the process. This is also a common pattern known as the poison pill.
Here is a basic interface that would need to be extended:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
class DataSource:
'''
Data source for the backtester. Must implement a "get_data" function
which streams data from the data source.
Each data point should be of the form (Timestamp, Ticker, Price).
'''
def __init__(self):
self._logger = logging.getLogger(__name__)
@classmethod
def process(cls, queue, source = None):
source = cls() if source is None else source
while True:
data = source.get_data()
if data is not None:
queue.put(data)
if data == 'POISON':
break
We use a simple format for the data which is centered around pricing updates, but you can add a more complicated approach.
The Controller Class
In our backtester, the Controller class acts as the Consumer and runs the backtest itself. The Controller class takes a Portfolio and an Algorithm. During each loop it follows the same pattern: 1. Check the queue for new data points. 2. Process the data points by adding them to the algorithm’s data bank (if implemented) and updating the stock in the Portfolio. 3. Allow the Algorithm to any generate orders. 4. Execute orders by calling the OrdersAPI class.
Now, steps 3 and 4 could technically by done asyncronously as well for even more of a performance boost and a more valid simulation of high frequency trading. However, this backtester is envisioned more for a day trader rather than HFT.
The main loop of the class described above is
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@classmethod
def backtest(cls, queue, controller = None):
controller = cls() if controller is None else controller
try:
while True:
if not queue.empty():
o = queue.get()
controller._logger.debug(o)
if o == 'POISON':
# Poison Pill!
break
timestamp = o[0]
ticker = o[1]
price = o[2]
# Update pricing
controller.process_pricing(ticker = ticker, price = price)
# Generate Orders
orders = controller._algorithm \
.generate_orders(timestamp, controller._portfolio)
# Process orders
if len(orders) > 0:
# Randomize the order execution
final_orders = [orders[k] for k in np.random.choice(len(orders),
replace=False,
size=len(orders))]
for order in final_orders:
controller.process_order(order)
controller._logger.info(controller._portfolio.value_summary(timestamp))
except Exception as e:
print(e)
finally:
controller._logger.info(controller._portfolio.value_summary(None))
The OrderApi Class
The Controller class takes an OrderApi class that simulates a trade occuring. This way, we can include things like slippage (the stock price changing from when we generated the sale), fees, and simply failed trades. It adds another level of realism to the simulation. This can be augmented as much or as little as one likes.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
class OrderApi:
def __init__(self):
self._slippage_std = .01
self._prob_of_failure = .0001
self._fee = .02
self._fixed_fee = 10
self._calculate_fee = lambda x : self._fee*abs(x) + self._fixed_fee
def process_order(self, order):
slippage = np.random.normal(0, self._slippage_std, size=1)[0]
if np.random.choice([False, True], p=[self._prob_of_failure, 1 -self._prob_of_failure],size=1)[0]:
trade_fee = self._fee*order[1]*(1+slippage)*order[2]
return (order[0], order[1]*(1+slippage), order[2], self._calculate_fee(trade_fee))
The Algorithm Class
The Algorithm class is your chance to do whatever you want. The only requirement is you implement a generate_orders() and update() method. Otherwise, whatever you do “under the covers” like storing information, etc., is up to you. Here’s an interface:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class Algorithm:
'''
Must implement a "generate_orders" function which returns a list of orders
and an update function.
Each order is a tuple of the form
( Stock Ticker str, Current Price float, Order Amount in shares float)
'''
def update(self, stock, price):
# Update
pass
def generate_orders(self, timestamp, portfolio):
# Make some orders based off the data
return orders
Example Implementations
To see full examples of these classes, see the check out the repo on github. I will briefly go through a few implementations here to give you a feel for writing them.
DataSource
You can implement the DataSource to pull data from Yahoo via pandas. The data source must put tuples of the form (timestamp, ticker, price) into the queue. Here’s one such way to do this: we pull the data and then place them one by one into a list. Whenever get_data() is called, we pop the top entry of the list off and return it. When our list is empty, we send the dreaded poison pill.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def set_source(self, source, tickers, start, end):
prices = pd.DataFrame()
counter = 0.
for ticker in tickers:
try:
self._logger.info('Loading ticker %s' % (counter / len(tickers)))
prices[ticker] = DataReader(ticker, source, start, end).loc[:, 'Close']
except Exception as e:
self._logger.error(e)
pass
counter+=1
events = []
for row in prices.iterrows():
timestamp=row[0]
series = row[1]
vals = series.values
indx = series.index
for k in np.random.choice(len(vals),replace=False, size=len(vals)): # Shuffle!
if np.isfinite(vals[k]):
events.append((timestamp, indx[k], vals[k]))
self._source = events
self._logger.info('Loaded data!')
def get_data(self):
try:
return self._source.pop(0)
except IndexError as e:
return 'POISON'
Algorithm
Here is an example of a basic Algorithm class. Disclaimer: this strategy is for educational purposes only.
What we will do is keep a moving average of every stock, and when the price of a stock exceeds a percentage of its average, we will buy it. If it drops, we will sell it. For fun, the strategy will randomly liquidate holdings as well.
First, the update method will keep track of a dictionary of stocks:
1
2
3
4
5
6
7
8
def update(self, stock, price):
if stock in self._averages:
self.add_price(stock, price)
else:
length = self._price_window
self._averages[stock] = {'History' : np.zeros(length), 'Index' : 0, 'Length' : length}
data = self._averages[stock]['History']
data[0] = price
Next, we can implement a _determine_if_trading() function to first decide if the algorithm will trade at all. This way, we can set limits on the number of trades we perform to limit the amount of fees we incur. We will call this function in our generate_orders method. You can see that there’s a few criteria, either: 1. We have enough data to compute the moving averages (i.e. _updates = # of updates = price_window for the first set of trades). 2. The required number of days have passed since our last trade. 3. The past trend of the portfolio is much greater than the current value, i.e. the portfolio is doing worse than its recent trend. 4. The cash balance is too high.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
def _determine_if_trading(self, date, portfolio_value, cash_balance):
time_delay_met = True
trade = False
override = False
self._updates += 1
if self._last_date is not None:
if (date - self._last_date).days <= self._minimum_wait_between_trades:
# Make orders based on previous day
return False
if self._updates == self._price_window+1:
trade = True
if (np.mean(self._trend)-portfolio_value)/portfolio_value > 0.05:
override = True
if cash_balance > portfolio_value*.03:
override = True
return trade or override
Now we’ll write a simple generate orders method. It’ll first call the _determine_if_trading() method to see if we will trade, otherwise we return an empty set of orders. Then, we grab the stocks that have seen enough days to have a valid moving average. After calculating the average, we select stocks that are either on the rise (greater than their moving average) or falling (lower than their moving average) and buy/sell them. We buy based on the proportion of our available cash and when we sell, we liquidate, because why not?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def generate_orders(self, timestamp, portfolio):
orders = []
cash_balance = portfolio.balance
portfolio_value = portfolio.get_total_value()
self.add_trend_value(portfolio_value)
if not self._determine_if_trading(timestamp,portfolio_value,cash_balance):
return orders
valid_stocks = [stock for stock in self._averages if portfolio.get_update_count(stock) > self._price_window]
if len(valid_stocks) == 0:
return orders
for stock in np.random.choice(valid_stocks, replace=False, size=len(valid_stocks)):
amt = cash_balance / len(valid_stocks) # Spend available cash
relative_change = (self.get_window_average(stock=stock) - self.get_price(stock))/self.get_price(stock)
if abs(relative_change) > .03:
# Positive is buy, negative is sell
order_type = np.sign(relative_change)
if order_type > 0 and np.random.uniform(0,1,size=1)[0] < .9:
amt = np.round(amt/self.get_price(stock),0)
else:
amt = - portfolio.get_shares(stock) # Liquidate! Why not?
if abs(amt) < .01:
# Stop small trades
continue
orders.append((stock, self.get_price(stock), amt))
self._last_trade = self._updates
self._last_date = timestamp
return orders
Simulation
Now that we’ve went over an implementation. Let’s try it out!
The Backtester Class
To set up a simulation, use the Backtester class. The backtester class exposes setter methods to control settings of the simulation. There is also a default run setup which starts with an empty portfolio with cash in it and tests it on historical data using the default implementation of the DataSource (more info below).
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
class Backtester:
def __init__(self):
self._logger = logging.getLogger(__name__)
self._settings = {}
self._default_settings = {
'Portfolio' : Portfolio(),
'Algorithm' : Algorithm(),
'Source' : 'yahoo',
'Start_Day' : dt.datetime(2016,1,1),
'End_Day' : dt.datetime.today(),
'Tickers' : ['AAPL','GOGL','MSFT','AA','APB']
}
def set_portfolio(self, portfolio):
self._settings['Portfolio'] = portfolio
def set_algorithm(self, algorithm):
self._settings['Algorithm'] = algorithm
def set_source(self, source):
self._settings['Source'] = source
def set_start_date(self, date):
self._settings['Start_Day'] = date
def set_end_date(self, date):
self._settings['End_Day'] = date
def set_stock_universe(self, stocks):
self._settings['Tickers'] = stocks
You run the class by instantiating it and calling the “backtest” method, like so:
1
2
b = Backtester()
b.backtest()
The code by default will log much of the information of to a “run.log” file in the directory which you called the program from. This includes ticker updates, values of trades, and updates on your portfolio’s value. An example output looks like:
Loaded data!
(Timestamp('2016-01-04 00:00:00'), 'APB', 9.8900000000000006)
(Timestamp('2016-01-04 00:00:00'), 'GOGL', 1.01)
(Timestamp('2016-01-04 00:00:00'), 'AAPL', 105.349998)
(Timestamp('2016-01-04 00:00:00'), 'AA', 9.7100000000000009)
(Timestamp('2016-01-04 00:00:00'), 'MSFT', 54.799999)
...
Trade on GOGL for 317460.0 shares at 0.627397882997 with fee 89.6694927745
Trade on AAPL for 2127.0 shares at 94.8422810289 with fee 90.6918126994
Trade on MSFT for 4048.0 shares at 48.9216529629 with fee 89.2139404775
2016-02-08 00:00:00 : Stock value: 599992.013619, Cash: 400792.309876, Total 1000784.32349
Extensions
Now, you should be ready to implement your own algorithms/data sources and test the results. This framework is very open ended and could be extended in many ways. However, when you implement your own backtester, you can add data points that other packages do not allow. For example, your data feed could also include twitter data that you’d parse and use (somehow?) to trade. It could use a news feed to determine catastrophic events. The possibilities are endless, and this is why trading is difficult.
Disclaimer
This backtester code is provided for educational purposes ONLY!