Pythonでバックテストフレームワークを構築する方法

この記事では、以下の機能を備えたバックテストフレームワークをPythonで作成してみます：

• モジュール化 – モジュール化して、自由に組み合わせたり、置き換えたりできるようにしたいと考えています。
• 拡張性 – コードは容易に拡張できる必要があります。
• 単一資産および複数資産の戦略をサポート
• 過去の株式データおよび複数のデータプロバイダーへのアクセス
• 取引手数料およびコミッションを含む
• パフォーマンス指標を備える

この目標を達成するために、以下を含むいくつかの重要なコンポーネントが必要です：

• データ管理: OHLCVデータ、およびシグナル生成に使用するあらゆる代替データソースのインポート、保存、取得を担当します。
• シグナル生成: データを分析し、事前定義された戦略または指標に基づいて売買シグナルを生成するロジックが含まれます。
• 執行エンジン: シグナルに基づく取引の執行をシミュレートし、コミッション、スリッページ、さらにオプションで売買スプレッドを考慮します。
• パフォーマンス評価：回収率、ボラティリティ、シャープレシオ、ドローダウンなどの主要なパフォーマンス指標を計算し、戦略の有効性を評価します。
• ユーティリティ：ログ記録、設定管理、その他のサポート機能が含まれます。

以下は、使用するPythonライブラリです：

• Poetry
• OpenBB Platform – これにより、複数のデータプロバイダーの市場データにシームレスにアクセスできます。詳しくはこちらをご覧ください。
• Pandas
• Numpy
• Matplotlib
• Ruff, Black, MyPy – 私が個人的に好んで使用しているコードチェックツール（任意）です。詳しくはこちらをご覧ください。

OpenBBを使用したデータプロセッサの作成

OpenBBプラットフォームでデータプロセッサを作成するのは非常に簡単です。このプラットフォームは、異なるAPI仕様、さまざまなデータプロバイダー、整理されていない出力、データ検証などの難題に対応してくれます。

これにより、データの検証と処理のためにカスタムクラスを作成する必要がなくなります。複数のデータプロバイダー、数百のデータポイント、さまざまな資産クラスなどに簡単にアクセスできます。さらに、このプラットフォームは返されるデータが標準に準拠し、品質が保証されるようにします。

ここでは株式資産に焦点を当て、データを日次ローソク足に限定します。もちろん、必要に応じてこれらの設定を拡張・変更できます。また、ユーザーがデータプロバイダー、取引コード、データの開始日と終了日を変更できるようにします。

OpenBBプラットフォームで特に気に入っている点の一つは、一部のエンドポイントで複数の銘柄コードを渡せることです。つまり、カンマ区切りのシンボルリストを渡すだけで、マルチアセット取引をサポートするための重要な一歩をすでに踏み出せるということです。

以下がコードです：

"""市場データを読み込み、処理するためのデータ処理モジュール。"""

from typing import Optional

import pandas as pd
from openbb import obb


class DataHandler:
"""バックテスト用の市場データを読み込み、処理するクラス。"""

def __init__(
self,
symbol: str,
start_date: Optional[str] = None,
end_date: Optional[str] = None,
provider: str = "fmp",
):
"""データハンドラーを初期化する。"""
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]:
"""株価データを読み込む。"""
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:
"""CSVファイルからデータを読み込む。"""
return pd.read_csv(file_path, index_col="date", parse_dates=True)

複数の取引コードを渡すと、Pandasデータフレームを含む辞書が返される点に注意してください。また、カスタムCSVファイルからデータを読み込み、日付列をインデックスとして使用する関数も追加しました。必要に応じて、この機能をさらに拡張・変更できます。

データを取得するには、このクラスを初期化して、次のように呼び出すだけですload_data方法だけで実現できます：

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

戦略プロセッサを作成する

次のステップは、戦略を処理するためのモジュールを作成することです。つまり、戦略の要件に基づいてシグナルを生成し、それをデータに付加できるモジュールを構築します。これにより、実行器はバックテストでこれらのシグナルを使用できるようになります。

私が実装したいのは、戦略の基底クラスのようなもので、開発者がそれを継承して変更したり、独自のカスタム戦略を構築したりできるものです。また、複数の資産を処理する場合にもシームレスに動作し、同じシグナルロジックを複数の資産に適用できるようにしたいと考えています。

以下はコード例です：

class Strategy:

def __init__(self, indicators: dict, signal_logic: Any):
"""指標とシグナルロジックで戦略を初期化する。"""
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]:
"""戦略の指標とシグナルロジックから売買シグナルを生成する。"""
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:
"""1つのデータフレームに戦略を適用する。"""
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)

その仕組みは、計算が必要な指標の辞書と、シグナルを生成するロジックを受け取るというものです。これらのシグナルは、売りを表す -1、買いを表す +1 とすることができます。また、現在のポジション状態も追跡します。

現在のコーディング方法では、渡した Lambda 関数がデータフレームに適用されます。

例：

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()

上の例では、遅い移動平均線と速い移動平均線を作成し、それに基づいて取引ロジックを定義しました。速い移動平均線が遅い移動平均線を上抜けたときにロングし、逆の場合はショートします。

これでデータを取得して取引シグナルを生成する方法が整いましたが、実際にバックテストを実行する方法だけが不足しています。これが最も複雑な部分です。

主要なバックテストロジックを作成する

主要なバックテスターのロジックはいくつかの部分で構成されます。含める必要がある主な部分は次のとおりです：

• 取引実行器
• 手数料計算器
• パフォーマンス指標計算器
• ポートフォリオマネージャー
• これらすべての部分をつなぎ合わせる連携部分

まずクラスを定義し、処理させたい基本変数をいくつか設定します：

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] = []

次に、取引実行器を定義します：

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

取引執行器は、シグナルが0を上回ったときに資産を購入し、シグナルが0を下回ったときに資産を売却します。また、購入に十分な現金があり、売却可能なポジションにあることも確認します。さらに、購入できる株式数を計算し、取引手数料も考慮します。

手数料を計算するには、次の手順を実行する必要があります：

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

ここで、取引する資産の保有状況、その価値、および履歴を追跡する必要があります：

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"])

最後に、これらのメソッドを使用して、次のようにバックテスターを実行できます：

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"]

ここで、いくつかの指標を計算するメソッドを追加します。また、サードパーティーライブラリなどを使用して、これらの機能を拡張することもできます。描画機能についても同様に処理します。具体的なコードはコードリポジトリで確認できます。

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()

最終ポートフォリオ価値：最終ポートフォリオ価値
総リターン：総リターン
年率リターン：年率リターン
年率ボラティリティ：年率ボラティリティ
シャープレシオ：シャープレシオ
ソルティノレシオ：ソルティノレシオ
最大ドローダウン：最大ドローダウン

バックテスターの準備が整ったので、いくつかの異なる戦略を試してみましょう。

Pythonでクロスオーバー戦略をバックテストする方法

この戦略の目的は、非常に基本的なクロスオーバー戦略を作成することです。ここでは、速い単純移動平均線（SMA）と遅い単純移動平均線を使用します。速い線が遅い線を上回ったときに買い、逆の場合に売却します。

Apple株で実行してみましょう。具体的な方法は次のとおりです：

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()

出力結果：

最終ポートフォリオ価値：11804.58
総リターン：18.05%
年率リターン：18.20%
年率ボラティリティ：13.06%
シャープレシオ：1.39
ソルティノレシオ：2.06
最大ドローダウン：-12.07%

なかなか良さそうです！

Pythonで平均回帰戦略をバックテストする方法

まず、戦略のロジックを計画しましょう：

この戦略の目標は、資産の取引価格が移動平均を三標準偏差上回ったときにその資産を売却し、資産の取引価格が移動平均を三標準偏差下回ったときにその資産を購入することです。

正常に機能させるには、以下の点に注意する必要があります：

• 移動平均が必要です
• 移動平均から標準偏差を計算する必要があります
• 上下限を計算する必要があります

私たちの戦略クラスは、指定された順序で計算を適用するため、ロジックの順序に従ってこれらの計算を簡単に連結し、これらの計算に基づいてシグナルを作成できます。

基本的なバックテストパラメータの定義から始めましょう：

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

次に、データを取得して操作を連結する必要があります：

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()

最終的なポートフォリオ価値：10725.54
総リターン：7.26%
年率リターン：7.29%
年率ボラティリティ：18.32%
シャープレシオ：0.40
ソルティノレシオ：0.53
最大ドローダウン：-23.37%

Pythonを使ってペアトレーディング戦略をバックテストする方法

Pythonを使ってペアトレーディング戦略をバックテストするのは、より複雑な例ですが、私たちのバックテスターはこの戦略を実行できるはずです。複雑なのは、2つの資産のデータを同じデータフレームに入れる必要がある点です。まず、この戦略を定義しましょう。

取引する資産はRoku (ROKU)とNetflix (NFLX)です。以前の記事と分析によると、これらには共和分関係があるためです。

一方の株式が過去5日間で、もう一方の株式に対して5%以上上昇した場合、ポジションに入ります（買い）。価格が高い方を売り、価格が低い方を買い、スプレッドが反転するまで保有します。設定を始め、データを手早く処理しましょう：

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()

あとは取引ロジックを策定すれば、バックテスターを実行できます：

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()

最終ポートフォリオ価値：14387.50
総リターン：43.88%
年率リターン：34.80%
年率ボラティリティ：55.77%
シャープレシオ：0.62
ソルティノレシオ：0.74
最大ドローダウン：-39.86%