LSTMを使用したビットコインオプションのインプライド・ボラティリティ予測
はじめに
暗号資産市場は価格変動が大きく、オプション取引におけるボラティリティは重要な指標となります。本記事では、ビットコイン（BTC）のオプション取引におけるインプライド・ボラティリティ（IV）を、LSTM（長短期記憶）モデルを用いて予測する実装について解説します。
目次
- なぜBTCオプションのIVを予測するのか
- 全体の処理フロー
- コードのポイント解説
- モデルの評価結果
- 予測結果の可視化
- まとめと今後の展望
1. なぜBTCオプションのIVを予測するのか
オプション取引において、インプライド・ボラティリティ（IV）は価格計算に不可欠な要素です。しかし、市場環境によって激しく変動するため、その予測は困難です。過去のオプション取引データや基礎資産の価格情報などを時系列データとして学習させることで、将来のIVを予測することができれば、以下のような価値が生まれます：
- オプション取引のリスク管理の向上
- より精緻な取引戦略の構築
- プライシングの精度向上
2. 全体の処理フロー
2.1 データ読み込み
- MongoDBからオプション価格などの時系列データを取得
- 必要なカラムの抽出とデータ型の変換
2.2 前処理
- 銘柄（symbol）のパース
- 「BTC-日付-ストライク価格-コール/プット」の情報を抽出
- ストライク価格でのフィルタリング（20,000～105,000）
- シンボルごとのDataFrame管理
- 各銘柄ごとにデータを整理
- 欠損値や異常値のチェック
- 特徴量のスケーリング
- MinMaxScalerを使用して0～1の範囲に正規化
- LSTMの学習安定化のため
2.3 LSTM用データセット作成
- ウィンドウサイズ：24ステップ
- 入力：過去24ステップの時系列データ
- 出力：次の1ステップのask1Iv
- データ形状：
- X: (サンプル数, 24, 特徴量数)
- y: (サンプル数,)
2.4 学習と評価用データの分割
- 訓練データ：80%
- テストデータ：20%
3. コードのポイント解説
3.1 parse_symbol関数
def parse_symbol(symbol: str) -> Tuple[str, str, float, str]:
splitted = symbol.split('-')
ticker = splitted[0] # BTC
expiry = splitted[1] # 28MAR25など
strike = float(splitted[2]) # 90000.0など
option_type = splitted[3] # C or P
return ticker, expiry, strike, option_type
3.2 process_option_data関数
def process_option_data(df: pd.DataFrame) -> Dict[str, pd.DataFrame]:
df['date'] = pd.to_datetime(df['date'])
symbol_groups = {}
for symbol in df['symbol'].unique():
symbol_df = df[df['symbol'] == symbol].copy()
symbol_df = symbol_df.sort_values('date')
symbol_df.set_index('date', inplace=True)
symbol_groups[symbol] = symbol_df
return symbol_groups
3.3 LSTM用データセット作成関数
def create_lstm_dataset(
data: np.ndarray, window_size: int = 24
) -> Tuple[np.ndarray, np.ndarray]:
X, y = [], []
ask1_iv_idx = FEATURE_INDICES['ask1Iv']
for i in range(len(data) - window_size):
X.append(data[i : i + window_size])
next_iv = data[i + window_size, ask1_iv_idx]
y.append(next_iv)
return np.array(X), np.array(y)
3.4 モデルの構築と学習
model = Sequential()
model.add(LSTM(32, input_shape=(X_train.shape[1], X_train.shape[2]), return_sequences=False))
model.add(Dense(1, activation='linear'))
model.compile(loss='mean_squared_error', optimizer='adam', metrics=['mae'])
history = model.fit(
X_train, y_train,
epochs=20,
batch_size=32,
validation_split=0.05,
verbose=1
)
3.5 完全なソースコード
以下に、実装の完全なソースコードを示します：
import os
import sys
from typing import Tuple, Dict
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense
from sklearn.preprocessing import MinMaxScaler
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_squared_error, mean_absolute_error
# Get the absolute path of the current directory
current_dir = os.path.dirname(os.path.abspath(__file__))
parent_dir = os.path.dirname(current_dir)
sys.path.append(parent_dir)
from mongodb.data_loader_mongo import MongoDataLoader
from common.constants import *
from option_pricing import simulate_option_prices
# 特徴量の定義
FEATURE_COLS = [
'ask1Price', 'bid1Price', 'ask1Iv', 'bid1Iv',
'markIv', 'underlyingPrice', 'delta', 'gamma',
'vega', 'theta', 'openInterest', 'markPrice'
]
# 特徴量のインデックスを定数として定義
FEATURE_INDICES = {col: idx for idx, col in enumerate(FEATURE_COLS)}
def parse_symbol(symbol: str) -> Tuple[str, str, float, str]:
splitted = symbol.split('-')
ticker = splitted[0]
expiry = splitted[1]
strike = float(splitted[2])
option_type = splitted[3] # "C" or "P"
return ticker, expiry, strike, option_type
def process_option_data(df: pd.DataFrame) -> Dict[str, pd.DataFrame]:
"""Process option data and organize it by symbol into time series"""
df['date'] = pd.to_datetime(df['date'])
symbol_groups = {}
for symbol in df['symbol'].unique():
symbol_df = df[df['symbol'] == symbol].copy()
symbol_df = symbol_df.sort_values('date')
symbol_df.set_index('date', inplace=True)
symbol_groups[symbol] = symbol_df
return symbol_groups
def create_lstm_dataset(
data: np.ndarray, window_size: int = 24
) -> Tuple[np.ndarray, np.ndarray]:
"""LSTM用の入力(X), 出力(y)を作成する関数"""
X, y = [], []
ask1_iv_idx = FEATURE_INDICES['ask1Iv']
for i in range(len(data) - window_size):
X.append(data[i : i + window_size])
next_iv = data[i + window_size, ask1_iv_idx]
y.append(next_iv)
return np.array(X), np.array(y)
def main():
# データ読み込み
db = MongoDataLoader()
df = db.load_data(OPTION_TICKER)
# ストライク価格でフィルタリング
df['strike'] = df['symbol'].apply(lambda s: parse_symbol(s)[2])
df = df[(df['strike'] >= 20000) & (df['strike'] <= 105000)]
# シンボルごとに時系列データ作成
symbol_timeseries = process_option_data(df)
# LSTM学習データ作成
all_X, all_y = [], []
window_size = 24
for symbol, ts_df in symbol_timeseries.items():
if len(ts_df) < 50:
continue
missing_cols = [col for col in FEATURE_COLS if col not in ts_df.columns]
if missing_cols:
continue
for col in FEATURE_COLS:
ts_df[col] = ts_df[col].astype(float)
features_data = ts_df[FEATURE_COLS].values
scaler = MinMaxScaler()
features_data_scaled = scaler.fit_transform(features_data)
X, y = create_lstm_dataset(features_data_scaled, window_size=window_size)
if len(X) > 0:
all_X.append(X)
all_y.append(y)
if len(all_X) == 0:
print("有効な学習データがありません。")
return
X_all = np.concatenate(all_X, axis=0)
y_all = np.concatenate(all_y, axis=0)
# データ分割
X_train, X_test, y_train, y_test = train_test_split(
X_all, y_all, test_size=0.2, shuffle=True, random_state=42
)
# モデル構築 & 学習
model = Sequential()
model.add(LSTM(32, input_shape=(X_train.shape[1], X_train.shape[2]), return_sequences=False))
model.add(Dense(1, activation='linear'))
model.compile(loss='mean_squared_error', optimizer='adam', metrics=['mae'])
history = model.fit(
X_train, y_train,
epochs=20,
batch_size=32,
validation_split=0.05,
verbose=1
)
# モデル評価
y_pred = model.predict(X_test).flatten()
mse = mean_squared_error(y_test, y_pred)
rmse = np.sqrt(mse)
mae = mean_absolute_error(y_test, y_pred)
print(f"Test MSE : {mse:.4f}")
print(f"Test RMSE: {rmse:.4f}")
print(f"Test MAE : {mae:.4f}")
# 結果の可視化
import matplotlib.pyplot as plt
target_symbol = "BTC-28MAR25-95000-C"
if target_symbol not in symbol_timeseries:
print(f"{target_symbol} のデータがありません。")
return
ts_df = symbol_timeseries[target_symbol].copy()
for col in FEATURE_COLS:
if col not in ts_df.columns:
print(f"{target_symbol} に {col} が存在しないため可視化できません。")
return
features_data = ts_df[FEATURE_COLS].astype(float).values
scaler_target = MinMaxScaler()
features_data_scaled = scaler_target.fit_transform(features_data)
X_target, y_target = create_lstm_dataset(features_data_scaled, window_size=24)
if len(X_target) == 0:
print(f"{target_symbol} の時系列が短いため可視化できません。")
return
y_pred_target = model.predict(X_target).flatten()
time_index = ts_df.index[window_size:]
plt.figure(figsize=(12, 6))
plt.plot(time_index, y_target, label='Actual ask1Iv')
plt.plot(time_index, y_pred_target, label='Predicted ask1Iv')
plt.title(f"Prediction vs Actual for {target_symbol}")
plt.xlabel("Date")
plt.ylabel("ask1Iv (scaled)")
plt.legend()
plt.grid()
plt.show()
if __name__ == "__main__":
main()
4. モデルの評価結果
評価指標の結果は以下の通りです：
| 指標 | 値 |
|---|---|
| MSE | 0.0068 |
| RMSE | 0.0823 |
| MAE | 0.0380 |
これらの値は、スケーリング後（0～1の範囲）の誤差を示しています。実際のIVの変動幅と比較すると、比較的良好な予測精度が得られていることがわかります。
5. 予測結果の可視化
以下は、BTC-28MAR25-90000-Cにおける実測値と予測値の比較グラフです：
グラフの特徴：
- 青線：実測値（Actual ask1Iv）
- オレンジ線：予測値（Predicted ask1Iv）
- 横軸：日付（Date）
- 縦軸：スケーリングされたIV（ask1Iv）
観察結果：
- 初期のデータポイントでは若干の乖離が見られる
- その後は実測値を良好に追随
- 全体的な傾向を捉えることに成功
6. まとめと今後の展望
6.1 成果
- LSTMモデルによるIV予測の基本的な実装に成功
- 評価指標から見る予測精度は良好
- 実用に向けた基盤を構築
6.2 改善案
-
特徴量の拡充
- 出来高データの追加
- オンチェーンデータの活用
- 他のマーケット要因の考慮
-
モデルの最適化
- LSTMユニット数の調整
- 層の深さの最適化
- ドロップアウトの導入
-
データ処理の改善
- スケーリング処理の一貫性確保
- 異常値処理の精緻化
- リアルタイムデータへの対応
おわりに
本実装は、BTCオプション取引におけるIV予測の基礎的なアプローチを示したものです。実際の取引現場での活用に向けては、上記の改善案を考慮しながら、さらなる精度向上とモデルの堅牢性強化が必要となるでしょう。