Creating a Backtesting Framework in Python
Basic Layout of the Backtesting Framework
Object-oriented classes:
- Engine: it is the main class that will be used to run our backtest.
- Strategy: this base class will serve as the building block for implementing the logic of our trading strategies.
- Order: When buying or selling, we first create an order object. If the order is filled, we create a trade object.
- Trade: The corresponding trade object will be created whenever an order is filled.
You might wonder why we are creating both Order and Trade classes. The reason is simple: it is standard practice for backtesting engines to assume that an order is created on close and filled on the next open. Doing so is good practice and will avoid look-ahead bias!
“””When buying or selling, we first create an order object. If the order is filled, we create a trade object.”””
“””Trade objects are created when an order is filled.”””
Implementing the Engine Class
import pandas as pd
from tqdm import tqdm
class Engine():
def __init__(self, initial_cash=100_000):
self.strategy = None
self.cash = initial_cash
self.data = None
self.current_idx = None
def add_data(self, data:pd.DataFrame):
# Add OHLC data to the engine
self.data = data
def add_strategy(self, strategy):
# Add a strategy to the engine
self.strategy = strategy
def run(self):
# We need to preprocess a few things before running the backtest
self.strategy.data = self.data
for idx in tqdm(self.data.index):
self.current_idx = idx
self.strategy.current_idx = self.current_idx
# fill orders from previus period
self._fill_orders()
# Run the strategy on the current bar
self.strategy.on_bar()
print(idx)
def _fill_orders(self):
# Fill orders from the previous period
pass
Although incomplete, we now have something we can execute, so let’s go ahead and do that!
import yfinance as yf
data = yf.Ticker('AAPL').history(start='2020-01-01', end='2022-12-31', interval='1d')
e = Engine()
e.add_data(data)
e.add_strategy(Strategy())
e.run()
Implementing the Strategy Class
class Strategy():
def __init__(self):
self.current_idx = None
self.data = None
self.orders = []
self.trades = []
def buy(self,ticker,size=1):
self.orders.append(
Order(
ticker = ticker,
side = 'buy',
size = size,
idx = self.current_idx
))
def sell(self,ticker,size=1):
self.orders.append(
Order(
ticker = ticker,
side = 'sell',
size = -size,
idx = self.current_idx
))
@property
def position_size(self):
return sum([t.size for t in self.trades])
def on_bar(self):
"""This method will be overriden by our strategies.
"""
pass
Implementing the Order and Trade Classes
class Order():
def __init__(self, ticker, size, side, idx):
self.ticker = ticker
self.side = side
self.size = size
self.type = 'market'
self.idx = idx
class Trade():
def __init__(self, ticker,side,size,price,type,idx):
self.ticker = ticker
self.side = side
self.price = price
self.size = size
self.type = type
self.idx = idx
def __repr__(self):
# Trade class string representation dunder method
return f'<Trade: {self.idx} {self.ticker} {self.size}@{self.price}>'
Refining the Engine Class
We’ll now return to our Engine class and implement the _fill_orders() method.
Whenever there are orders to be filled, we will also check for the following conditions:
- If we’re buying, our cash balance has to be large enough to cover the order.
- If we are selling, we must have enough shares to cover the order.
In other words, our backtesting engine will be restricted to long-only strategies. I introduced this restriction to show how to incorporate such a feature, but it is by no means a requirement. If you want to test long-short strategies, you can safely comment out these lines of code. Even better, you could add an “allow_short_trading” boolean parameter to the engine class and let the end-user decide.
def _fill_orders(self):
"""this method fills buy and sell orders, creating new trade objects and adjusting the strategy's cash balance.
Conditions for filling an order:
- If we're buying, our cash balance has to be large enough to cover the order.
- If we are selling, we have to have enough shares to cover the order.
"""
for order in self.strategy.orders:
can_fill = False
if order.side == 'buy' and self.cash >= self.data.loc[self.current_idx]['Open'] * order.size:
can_fill = True
elif order.side == 'sell' and self.strategy.position_size >= order.size:
can_fill = True
if can_fill:
t = Trade(
ticker = order.ticker,
side = order.side,
price= self.data.loc[self.current_idx]['Open'],
size = order.size,
type = order.type,
idx = self.current_idx)
self.strategy.trades.append(t)
self.cash -= t.price * t.size
self.strategy.orders = []
試試看
class BuyAndSellSwitch(Strategy):
def on_bar(self):
if self.position_size == 0:
self.buy('AAPL', 1)
print(self.current_idx,"buy")
else:
self.sell('AAPL', 1)
print(self.current_idx,"sell")
data = yf.Ticker('AAPL').history(start='2022-12-01', end='2022-12-31', interval='1d')
e = Engine()
e.add_data(data)
e.add_strategy(BuyAndSellSwitch())
e.run()
# We can check the list of trades that were executed:
# e.strategy.trades
Calculating the Total Return
# strategy.py
def _get_stats(self):
metrics = {}
total_return =100 * ((self.data.loc[self.current_idx]['Close'] * self.strategy.position_size + self.cash) / self.initial_cash -1)
metrics['total_return'] = total_return
return metrics
Sanity check: comparing results with Backtesting.py
Implementing Limit Orders
This feature impacts quite a few parts of our backtesting engine. Let’s try not to break everything!
1) Add buy_limit and sell_limit methods to our Strategy class
# strategy.py
def buy_limit(self,ticker,limit_price, size=1):
self.orders.append(
Order(
ticker = ticker,
side = 'buy',
size = size,
limit_price=limit_price,
order_type='limit',
idx = self.current_idx
))
def sell_limit(self,ticker,limit_price, size=1):
self.orders.append(
Order(
ticker = ticker,
side = 'sell',
size = -size,
limit_price=limit_price,
order_type='limit',
idx = self.current_idx
))
2) Add limit_price and order_type attributes to the Order Class
# order.py
class Order():
def __init__(self, ticker, size, side, idx, limit_price=None, order_type='market'):
...
self.type = order_type
self.limit_price = limit_price
3) Update the _fill_orders() method
# engine.py
def _fill_orders(self):
for order in self.strategy.orders:
# FOR NOW, SET FILL PRICE TO EQUAL OPEN PRICE. THIS HOLDS TRUE FOR MARKET ORDERS
fill_price = self.data.loc[self.current_idx]['Open']
can_fill = False
if order.side == 'buy' and self.cash >= self.data.loc[self.current_idx]['Open'] * order.size:
if order.type == 'limit':
# LIMIT BUY ORDERS ONLY GET FILLED IF THE LIMIT PRICE IS GREATER THAN OR EQUAL TO THE LOW PRICE
if order.limit_price >= self.data.loc[self.current_idx]['Low']:
fill_price = order.limit_price
can_fill = True
print(self.current_idx, 'Buy Filled. ', "limit",order.limit_price," / low", self.data.loc[self.current_idx]['Low'])
else:
print(self.current_idx,'Buy NOT filled. ', "limit",order.limit_price," / low", self.data.loc[self.current_idx]['Low'])
else:
can_fill = True
elif order.side == 'sell' and self.strategy.position_size >= order.size:
if order.type == 'limit':
#LIMIT SELL ORDERS ONLY GET FILLED IF THE LIMIT PRICE IS LESS THAN OR EQUAL TO THE HIGH PRICE
if order.limit_price <= self.data.loc[self.current_idx]['High']:
fill_price = order.limit_price
can_fill = True
print(self.current_idx,'Sell filled. ', "limit",order.limit_price," / high", self.data.loc[self.current_idx]['High'])
else:
print(self.current_idx,'Sell NOT filled. ', "limit",order.limit_price," / high", self.data.loc[self.current_idx]['High'])
else:
can_fill = True
if can_fill:
t = Trade(
ticker = order.ticker,
side = order.side,
price= fill_price,
size = order.size,
type = order.type,
idx = self.current_idx)
self.strategy.trades.append(t)
self.cash -= t.price * t.size
self.strategy.orders = []
4) Add a close property in the Strategy class to retrieve the latest close
# strategy.py
@property
def close(self):
return self.data.loc[self.current_idx]['Close']
5) Testing the feature
# main.py
import yfinance as yf
class BuyAndSellSwitch(Strategy):
def on_bar(self):
if self.position_size == 0:
limit_price = self.close * 0.995
self.buy_limit('AAPL', size=100,limit_price=limit_price)
print(self.current_idx,"buy")
else:
limit_price = self.close * 1.005
self.sell_limit('AAPL', size=100,limit_price=limit_price)
print(self.current_idx,"sell")
data = yf.Ticker('AAPL').history(start='2022-12-01', end='2022-12-31', interval='1d')
e = Engine()
e.add_data(data)
e.add_strategy(BuyAndSellSwitch())
e.run()
5) Testing the feature
# main.py
import yfinance as yf
class BuyAndSellSwitch(Strategy):
def on_bar(self):
if self.position_size == 0:
limit_price = self.close * 0.995
self.buy_limit('AAPL', size=100,limit_price=limit_price)
print(self.current_idx,"buy")
else:
limit_price = self.close * 1.005
self.sell_limit('AAPL', size=100,limit_price=limit_price)
print(self.current_idx,"sell")
data = yf.Ticker('AAPL').history(start='2022-12-01', end='2022-12-31', interval='1d')
e = Engine()
e.add_data(data)
e.add_strategy(BuyAndSellSwitch())
e.run()
Adding Output Metrics
After each iteration in the run() method of the Engine class, add the cash holdings to the cash_series:
# engine.py
class Engine():
def __init__(self, initial_cash=100_000):
...
self.cash_series = {}
self.stock_series = {}
1) Buy & Hold Benchmark
# engine.py
def _get_stats(self):
...
# Buy & hold benchmark
portfolio_bh = self.initial_cash / self.data.loc[self.data.index[0]]['Open'] * self.data.Close
...
2) Exposure to the Asset [%]
# engine.py
def _get_stats(self):
...
# Create a dataframe with the cash and stock holdings at the end of each bar
portfolio = pd.DataFrame({'stock':self.stock_series, 'cash':self.cash_series})
# Add a third column with the total assets under managemet
portfolio['total_aum'] = portfolio['stock'] + portfolio['cash']
# Caclulate the total exposure to the asset as a percentage of our total holdings
metrics['exposure_pct'] = ((portfolio['stock'] / portfolio['total_aum']) * 100).mean()
...
3) Annualized Returns
# Calculate annualized returns p = portfolio.total_aum metrics['returns_annualized'] = ((p.iloc[-1] / p.iloc[0]) ** (1 / ((p.index[-1] - p.index[0]).days / 365)) - 1) * 100 p_bh = portfolio_bh metrics['returns_bh_annualized'] = ((p_bh.iloc[-1] / p_bh.iloc[0]) ** (1 / ((p_bh.index[-1] - p_bh.index[0]).days / 365)) - 1) * 100
4) Annualized Volatility
For this calculation, I’m assuming we’re using daily data of an asset that trades only during working days (i.e., stocks). If you’re trading cryptocurrencies, use 365 instead of 252.
# Annual Volatility # self.trading_days = 252 # for stock self.trading_days = 365 # for crypto metrics['volatility_ann'] = p.pct_change().std() * np.sqrt(self.trading_days) * 100 metrics['volatility_bh_ann'] = p_bh.pct_change().std() * np.sqrt(self.trading_days) * 100
5) Sharpe Ratio
# Sharpe ratio
self.risk_free_rate = 0
metrics['sharpe_ratio'] = (metrics['returns_annualized'] - self.risk_free_rate) / metrics['volatility_ann']
metrics['sharpe_ratio_bh'] = (metrics['returns_bh_annualized'] - self.risk_free_rate) / metrics['volatility_bh_ann']
6) Maximum Drawdown
def _get_stats(self):
...
metrics["max_drawdown"] = self.get_max_drawdown(self.data["Close"])
return metrics
def get_max_drawdown(self, close):
# Maximum Drawdown
roll_max = close.cummax()
daily_drawdown = close / roll_max - 1.0
max_daily_drawdown = daily_drawdown.cummin()
return max_daily_drawdown.min() * 100
Implementing an SMA Crossover Strategy
Reference:
https://www.qmr.ai/building-a-backtesting-framework-in-python-from-scratch/
https://www.qmr.ai/building-a-backtesting-framework-in-python-part-ii/
https://www.qmr.ai/is-backtesting-accurate/
Note:
There are two types of backtester implementation:
1. for loop
2. event-driven
There are series of articles about event-driven backtester implementation here: https://www.quantstart.com/articles/Event-Driven-Backtesting-with-Python-Part-I/
