How to Build a Backtesting Framework with Python

In this article, we will try to create a backtesting framework with Python that includes the following features:

• Modularity – We want to make it modular so that components can be freely combined and replaced.
• Extensibility – The code should be easy to extend.
• Support for single-asset and multi-asset strategies
• Access to historical equity data and multiple data providers
• Include trading costs and commissions
• Performance metrics

To achieve this goal, we need several key components, including:

• Data management: Responsible for importing, storing, and retrieving OHLCV data, as well as any alternative data sources used to generate signals.
• Signal generation: Contains the logic for analyzing data and generating buy and sell signals based on predefined strategies or indicators.
• Execution engine: Simulates trade execution based on signals, taking commissions and slippage into account, and optionally considering the bid-ask spread.
• Performance evaluation: Calculate key performance indicators, such as returns, volatility, Sharpe ratio, drawdown, and more, to evaluate the effectiveness of the strategy.
• Utilities: Includes logging, configuration management, and other support functions.

Here are the Python libraries we will use:

• Poetry
• OpenBB Platform – This will provide us with seamless access to market data from multiple data providers. You can read more information here.
• Pandas
• Numpy
• Matplotlib
• Ruff, Black, MyPy – My preferred code linting tools (optional). You can read more related information here.

Creating a Data Handler with OpenBB

Creating a data handler on the OpenBB Platform is very simple. The platform handles the challenges of different API specifications, various data providers, messy outputs, data validation, and more.

This means we no longer need to create custom classes for data validation and processing. You can easily access multiple data providers, hundreds of data points, different asset classes, and more. The platform also ensures that the returned data conforms to standards and is high quality.

Here, I will focus on stock assets and limit the data to daily candlesticks. Of course, you can expand and modify these settings as needed. I will also allow users to change the data provider, ticker symbols, and the start and end dates for the data.

One thing I particularly like about the OpenBB Platform is that some of its endpoints allow multiple ticker symbols to be passed. This means we have already taken an important step toward supporting multi-asset trading by simply passing a comma-separated list of symbols.

Here is the code:

"""Data-processing utilities for loading and preparing market data."""

from typing import Optional

import pandas as pd
from openbb import obb


class DataHandler:
"""Load and prepare market data for a backtest."""

def __init__(
self,
symbol: str,
start_date: Optional[str] = None,
end_date: Optional[str] = None,
provider: str = "fmp",
):
"""Initialize the data handler."""
self.symbol = symbol.upper()
self.start_date = start_date
self.end_date = end_date
self.provider = provider

def load_data(self) -> pd.DataFrame | dict[str, pd.DataFrame]:
"""Load historical equity data."""
data = obb.equity.price.historical(
symbol=self.symbol,
start_date=self.start_date,
end_date=self.end_date,
provider=self.provider,
).to_df()

if "," in self.symbol:
data = data.reset_index().set_index("symbol")
return {symbol: data.loc[symbol] for symbol in self.symbol.split(",")}

return data

def load_data_from_csv(self, file_path) -> pd.DataFrame:
"""Load market data from a CSV file."""
return pd.read_csv(file_path, index_col="date", parse_dates=True)

Note that when multiple ticker symbols are passed, it returns a dictionary containing Pandas data frames. I also added a function that loads data from a custom CSV file and uses the date column as the index. You can further expand and modify this functionality according to your needs.

To retrieve some data, we simply need to initialize this class and then call it like thisload_datamethod:

data = DataHandler("AAPL").load_data()
data.head()

Creating the Strategy Handler

The next step is to create a module for handling strategies. By this, I mean building a module that can generate signals based on the strategy's requirements and attach them to the data, so the executor can use these signals during backtesting.

What I want to implement is something similar to a strategy base class that developers can inherit from, modify, or use to build their own custom strategies. I also want it to work seamlessly when handling multiple assets, allowing the same signal logic to be applied to multiple assets.

Here is a code example:

class Strategy:

def __init__(self, indicators: dict, signal_logic: Any):
"""Initialize the strategy with indicators and signal logic."""
self.indicators = indicators
self.signal_logic = signal_logic

def generate_signals(
self, data: pd.DataFrame | dict[str, pd.DataFrame]
) -> pd.DataFrame | dict[str, pd.DataFrame]:
"""Generate trading signals from the configured strategy."""
if isinstance(data, dict):
for _, asset_data in data.items():
self._apply_strategy(asset_data)
else:
self._apply_strategy(data)
return data

def _apply_strategy(self, df: pd.DataFrame) -> None:
"""Apply the strategy to one data frame."""
for name, indicator in self.indicators.items():
df[name] = indicator(df)

df["signal"] = df.apply(lambda row: self.signal_logic(row), axis=1)
df["positions"] = df["signal"].diff().fillna(0)

The idea is to accept a dictionary of indicators that need to be calculated, along with the logic for generating signals. These signals can be -1 to indicate a sell and +1 to indicate a buy. It also tracks our current position status.

Currently, the way it is coded is that the Lambda function we pass to it is applied to the data frame.

Example:

strategy = Strategy(
indicators={
"sma_20": lambda row: row["close"].rolling(window=20).mean(),
"sma_60": lambda row: row["close"].rolling(window=60).mean(),
},
signal_logic=lambda row: 1 if row["sma_20"] > row["sma_60"] else -1,
)
data = strategy.generate_signals(data)
data.tail()

In the example above, I created a slow and a fast moving average and defined my trading logic based on them: go long when the fast moving average crosses above the slow moving average, and go short otherwise.

Now that we have methods for obtaining data and generating trading signals, all that is missing is a method to actually run the backtest. This is the most complex part.

Creating the Main Backtesting Logic

The main backtester logic will consist of several parts. The main components we need to include are as follows:

• Trade executor
• Commission calculator
• Performance metrics calculator
• Portfolio manager
• The link that ties all these components together

We first define the class and set some basic variables that we want it to handle:

class Backtester:

def __init__(
self,
initial_capital: float = 10000.0,
commission_pct: float = 0.001,
commission_fixed: float = 1.0,
):
"""Initialize the backtester with initial capital and commission fees."""
self.initial_capital: float = initial_capital
self.commission_pct: float = commission_pct
self.commission_fixed: float = commission_fixed
self.assets_data: Dict = {}
self.portfolio_history: Dict = {}
self.daily_portfolio_values: List[float] = []

Now, we will define the trade executor:

def execute_trade(self, asset: str, signal: int, price: float) -> None:
if signal > 0 and self.assets_data[asset]["cash"] > 0: # Buy
trade_value = self.assets_data[asset]["cash"]
commission = self.calculate_commission(trade_value)
shares_to_buy = (trade_value - commission) / price
self.assets_data[asset]["positions"] += shares_to_buy
self.assets_data[asset]["cash"] -= trade_value
elif signal < 0 and self.assets_data[asset]["positions"] > 0: # Sell
trade_value = self.assets_data[asset]["positions"] * price
commission = self.calculate_commission(trade_value)
self.assets_data[asset]["cash"] += trade_value - commission
self.assets_data[asset]["positions"] = 0

The trade executor will buy the asset when the signal is greater than 0 and sell the asset when the signal is less than 0. It will also ensure that we have enough cash to make the purchase and that we are in a position to sell. In addition, it will calculate the number of shares we can purchase and account for trading commissions.

To calculate the commission, we need to perform the following steps:

def calculate_commission(self, trade_value: float) -> float:
return max(trade_value * self.commission_pct, self.commission_fixed)

Now, we need to track the position, value, and history of the asset we are trading:

def update_portfolio(self, asset: str, price: float) -> None:
self.assets_data[asset]["position_value"] = (
self.assets_data[asset]["positions"] * price
)
self.assets_data[asset]["total_value"] = (
self.assets_data[asset]["cash"] + self.assets_data[asset]["position_value"]
)
self.portfolio_history[asset].append(self.assets_data[asset]["total_value"])

Finally, we can run the backtester by using these methods, as shown below:

def backtest(self, data: pd.DataFrame | dict[str, pd.DataFrame]):
if isinstance(data, pd.DataFrame): # Single asset
data = {
"SINGLE_ASSET": data
}
for asset in data:
self.assets_data[asset] = {
"cash": self.initial_capital / len(data),
"positions": 0,
"position_value": 0,
"total_value": 0,
}
self.portfolio_history[asset] = []

for date, row in data[asset].iterrows():
self.execute_trade(asset, row["signal"], row["close"])
self.update_portfolio(asset, row["close"])
if len(self.daily_portfolio_values) < len(data[asset]):
self.daily_portfolio_values.append(
self.assets_data[asset]["total_value"]
)
else:
self.daily_portfolio_values[
len(self.portfolio_history[asset]) - 1
] += self.assets_data[asset]["total_value"]

Now, I will add a method to calculate some metrics, and these functions can be extended by using third-party libraries, for example. I will do the same for the plotting functionality. The specific code can be found in the code repository.

def calculate_performance(self, plot: bool = True) -> None:
if not self.daily_portfolio_values:
print("No portfolio history to calculate performance.")
return

portfolio_values = pd.Series(self.daily_portfolio_values)
daily_returns = portfolio_values.pct_change().dropna()

total_return = calculate_total_return(
portfolio_values.iloc[-1], self.initial_capital
)
annualized_return = calculate_annualized_return(
total_return, len(portfolio_values)
)
annualized_volatility = calculate_annualized_volatility(daily_returns)
sharpe_ratio = calculate_sharpe_ratio(annualized_return, annualized_volatility)
sortino_ratio = calculate_sortino_ratio(daily_returns, annualized_return)
max_drawdown = calculate_maximum_drawdown(portfolio_values)

print(f"Final Portfolio Value: {portfolio_values.iloc[-1]:.2f}")
print(f"Total Return: {total_return * 100:.2f}%")
print(f"Annualized Return: {annualized_return * 100:.2f}%")
print(f"Annualized Volatility: {annualized_volatility * 100:.2f}%")
print(f"Sharpe Ratio: {sharpe_ratio:.2f}")
print(f"Sortino Ratio: {sortino_ratio:.2f}")
print(f"Maximum Drawdown: {max_drawdown * 100:.2f}%")

if plot:
self.plot_performance(portfolio_values, daily_returns)

def plot_performance(self, portfolio_values: Dict, daily_returns: pd.DataFrame):
plt.figure(figsize=(10, 6))

plt.subplot(2, 1, 1)
plt.plot(portfolio_values, label="Portfolio Value")
plt.title("Portfolio Value Over Time")
plt.legend()

plt.subplot(2, 1, 2)
plt.plot(daily_returns, label="Daily Returns", color="orange")
plt.title("Daily Returns Over Time")
plt.legend()

plt.tight_layout()
plt.show()

Final Portfolio Value: Final Portfolio Value
Total Return: Total Return
Annualized Return: Annualized Return
Annualized Volatility: Annualized Volatility
Sharpe Ratio: Sharpe Ratio
Sortino Ratio: Sortino Ratio
Maximum Drawdown: Maximum Drawdown

Now that the backtester is ready, let’s try it with several different strategies.

How to Backtest a Crossover Strategy in Python?

The goal of this strategy is to create a very basic crossover strategy in which we use a fast simple moving average (SMA) and a slow simple moving average. We buy when the fast line crosses above the slow line and sell otherwise.

We can run it with Apple stock. Here is how:

from backtester.data_handler import DataHandler
from backtester.backtester import Backtester
from backtester.strategies import Strategy

symbol = "AAPL,MSFT"
start_date = "2023-01-01"
end_date = "2023-12-31"

data = DataHandler(
symbol=symbol, start_date=start_date, end_date=end_date
).load_data()

strategy = Strategy(
indicators={
"sma_20": lambda row: row["close"].rolling(window=20).mean(),
"sma_60": lambda row: row["close"].rolling(window=60).mean(),
},
signal_logic=lambda row: 1 if row["sma_20"] > row["sma_60"] else -1,
)
data = strategy.generate_signals(data)

backtester = Backtester()
backtester.backtest(data)
backtester.calculate_performance()

Output:

Final Portfolio Value: 11804.58
Total Return: 18.05%
Annualized Return: 18.20%
Annualized volatility: 13.06%
Sharpe ratio: 1.39
Sortino ratio: 2.06
Maximum drawdown: -12.07%

Looks pretty good!

How to backtest a mean reversion strategy with Python?

First, let's plan the strategy logic:

The goal of this strategy is to sell the asset when its trading price is more than three standard deviations above the rolling average, and to buy the asset when its trading price is more than three standard deviations below the rolling average.

For it to work properly, note the following:

• A rolling average is required
• The standard deviation must be calculated from the rolling average
• The upper and lower bounds must be calculated

Because our strategy class applies calculations in the specified order, we can easily chain these calculations together in logical order and create signals based on them.

Let's start by defining the basic backtesting parameters:

symbol = "HE"
start_date = "2022-01-01"
end_date = "2022-12-31"

Now, we need to get the data and chain the operations together:

data = DataHandler(symbol=symbol, start_date=start_date, end_date=end_date).load_data()

strategy = Strategy(
indicators={
"sma_50": lambda row: row["close"].rolling(window=50).mean(),
"std_3": lambda row: row["close"].rolling(window=50).std() * 3,
"std_3_upper": lambda row: row["sma_50"] + row["std_3"],
"std_3_lower": lambda row: row["sma_50"] - row["std_3"],
},
signal_logic=lambda row: (
1
if row["close"] < row["std_3_lower"]
else -1 if row["close"] > row["std_3_upper"] else 0
),
)
data = strategy.generate_signals(data)

backtester = Backtester()
backtester.backtest(data)
backtester.calculate_performance()

Final portfolio value: 10725.54
Total return: 7.26%
Annualized return: 7.29%
Annualized volatility: 18.32%
Sharpe ratio: 0.40
Sortino ratio: 0.53
Maximum drawdown: -23.37%

How to Backtest a Pairs Trading Strategy with Python?

Backtesting a pairs trading strategy with Python is a more complex example, but our backtester should be able to execute this strategy. The complexity lies in the fact that we need to put the data for two assets in the same data frame. First, let's define the strategy.

The assets we will trade are Roku (ROKU) and Netflix (NFLX), because they are cointegrated according to our previous articles and analysis.

If one stock gains relative to the other by 5% or more over the past five days, we will enter a position (buy). We will sell the one with the higher price and buy the one with the lower price until the spread reverses. Let's start setting up and quickly process the data:

import pandas as pd

symbol = "NFLX,ROKU"
start_date = "2023-01-01"

data = DataHandler(
symbol=symbol,
start_date=start_date,
).load_data()

data = pd.merge(
data["NFLX"].reset_index(),
data["ROKU"].reset_index(),
left_index=True,
right_index=True,
suffixes=("_NFLX", "_ROKU"),
)

data = data.rename(columns={"close_ROKU": "close"})
data.head()

Now, we just need to formulate the trading logic, and we can run the backtester:

strategy = Strategy(
indicators={
"day_5_lookback_NFLX": lambda row: row["close_NFLX"].shift(5),
"day_5_lookback_ROKU": lambda row: row["close"].shift(5),
},
signal_logic=lambda row: (
1
if row["close_NFLX"] > row["day_5_lookback_NFLX"] * 1.05
else -1 if row["close_NFLX"] < row["day_5_lookback_NFLX"] * 0.95 else 0
),
)
data = strategy.generate_signals(data)

backtester = Backtester()
backtester.backtest(data)
backtester.calculate_performance()

Final portfolio value: 14387.50
Total return: 43.88%
Annualized return: 34.80%
Annualized volatility: 55.77%
Sharpe ratio: 0.62
Sortino ratio: 0.74
Maximum drawdown: -39.86%