こんにちは、HolySheep AIの技術チームです。暗号資産市場のボラティリティ予測は、トレーディング戦略のリスク管理において極めて重要な役割を果たします。本稿では、Tardisから取得した高頻度BTC/USD取引データを活用し、伝統的なモデルと機械学習(ML)手法の双方を用いたボラティリティ予測モデルの構築方法を詳細に解説します。

前提条件と環境構築

本研究ではPython 3.10以上を使用し、以下のライブラリが必要です。HolySheep AIのAPIを活用することで、モデル構築过程中的なデータ取得から予測推論まで、低コストで高速に処理できます。

pip install pandas numpy scipy statsmodels scikit-learn arch
pip install requests python-dotenv matplotlib seaborn
pip install tardis-client  # Tardis APIクライアント

TardisからのBTC/USD高頻度データ取得

Tardisは、主要暗号資産取引所のリアルタイム・ исторических данныхを提供するプラットフォームです。HolySheep AIでは、このデータを前処理した状態でAPI経由でもアクセス可能です。

import requests
import pandas as pd
from datetime import datetime, timedelta

HolySheep AI経由でTardisデータ取得

BASE_URL = "https://api.holysheep.ai/v1" API_KEY = "YOUR_HOLYSHEEP_API_KEY" def fetch_btc_historical_data(symbol: str, start: str, end: str) -> pd.DataFrame: """BTC/USDの約定履歴を取得""" headers = { "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json" } payload = { "model": "tardis-btc-data", "symbol": symbol, "start_date": start, "end_date": end, "interval": "1min" } response = requests.post( f"{BASE_URL}/data/historical", headers=headers, json=payload, timeout=30 ) if response.status_code == 401: raise ConnectionError("APIキーが無効です。") elif response.status_code == 429: raise ConnectionError("レートリミットに達しました。1分後に再試行してください。") data = response.json() df = pd.DataFrame(data['trades']) df['timestamp'] = pd.to_datetime(df['timestamp']) df.set_index('timestamp', inplace=True) return df

実データ取得

btc_data = fetch_btc_historical_data( symbol="BTC-USD", start="2024-01-01", end="2024-12-31" ) print(f"取得レコード数: {len(btc_data):,}") print(f"データ期間: {btc_data.index.min()} ~ {btc_data.index.max()}")

私の場合、Tardisから2024年1年間の1分足を全て取得すると約52万件のレコードになりました。HolySheepのDeepSeek V3.2モデル($0.42/MTok)を利用すれば、このデータに対する特徴量生成を$0.02以下で処理できます。

ボラティリティ特徴量の設計

高精度な予測モデルには、適切な特徴量設計が不可欠です。以下に、我々が構築した特徴量エンジニアリングパイプラインを示します。

import numpy as np

def engineer_volatility_features(df: pd.DataFrame, windows: list = [5, 15, 60]) -> pd.DataFrame:
    """ボラティリティ予測用の特徴量を生成"""
    features = pd.DataFrame(index=df.index)
    
    # 対数収益率
    features['log_return'] = np.log(df['price'] / df['price'].shift(1))
    
    # 実現ボラティリティ(複数の 윈도우)
    for w in windows:
        features[f'rv_{w}min'] = (
            features['log_return']**2
        ).rolling(window=w).sum() * np.sqrt(1440 / w) * 100
        
        features[f'rv_std_{w}min'] = (
            features['log_return']
        ).rolling(window=w).std() * np.sqrt(1440) * 100
    
    # Realized Quarticity(ボラティリティの非対称性検出)
    features['rq'] = (
        features['log_return']**4
    ).rolling(window=60).sum() * (60 / 1440)
    
    #  торговый volume
    features['volume'] = df['volume']
    features['log_volume'] = np.log(df['volume'] + 1)
    
    #  VWAP近似
    features['vwap_proxy'] = (
        df['price'] * df['volume']
    ).rolling(window=10).sum() / df['volume'].rolling(window=10).sum()
    
    #  Bid/Ask Spread(板情報から計算)
    if 'bid' in df.columns and 'ask' in df.columns:
        features['spread_pct'] = (
            (df['ask'] - df['bid']) / ((df['ask'] + df['bid']) / 2)
        ) * 100
    
    return features.dropna()

特徴量生成

features_df = engineer_volatility_features(btc_data) print(f"生成特徴量数: {features_df.shape[1]}") print(features_df.describe())

GARCHモデルによるボラティリティ予測

GARCH(1,1)モデルの実装

GARCH(Generalized Autoregressive Conditional Heteroskedasticity)モデルは、金融データの時系列ボラティリティ予測において広く使用される伝統的な手法です。

from arch import arch_model
from sklearn.metrics import mean_squared_error, mean_absolute_error
import warnings
warnings.filterwarnings('ignore')

def fit_garch_model(returns: pd.Series, p: int = 1, q: int = 1):
    """GARCH(p,q)モデルを適合"""
    # 百分率に変換
    returns_pct = returns * 100
    
    model = arch_model(
        returns_pct,
        vol='Garch',
        p=p,
        q=q,
        dist='t'  # スチューデントt分布
    )
    
    result = model.fit(disp='off', show_warning=False)
    return result

def garch_forecast(model_result, horizon: int = 60):
    """GARCH予測を実行"""
    forecast = model_result.forecast(horizon=horizon)
    return np.sqrt(forecast.variance.values[-1, :]) / 100  # 百分率から小数へ

訓練データとテストデータに分割

train_size =