暗号資産市場のボラティリティ予測は、取引戦略の根幹を担う重要な課題です。本稿では、板情報（Order Book）から抽出した微細なシグナルを大規模言語モデル（LLM）と組み合わせた予測システムを、アーキテクチャ設計から本番実装まで体系的に解説します。私は以前、板情報ベースの機械学習モデルを構築する際に遅延問題とコスト増加に直面しましたが、HolySheep AIの低レイテンシAPIを活用することで这些问题を解決し、実用的な予測精度を達成しました。
1. システムアーキテクチャ概要
本システムの核心は、Order Bookの変化量を時系列特徴量として抽出し、それを自然言語プロンプトに変換してLLMに分析させるハイブリッド構成にあります。従来の数値ベース機械学習では捉えきれない「市場の不確実性」や「投資家の心理」をLLMの推論能力で補完することで、短期ボラティリティ予測の精度を显著に向上させます。
┌─────────────────────────────────────────────────────────────────┐
│ Crypto Volatility Prediction System │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ Order Book │───▶│ Feature │───▶│ Prompt │ │
│ │ WebSocket │ │ Extraction │ │ Constructor │ │
│ │ (Binance) │ │ Pipeline │ │ │ │
│ └──────────────┘ └──────────────┘ └──────────────────┘ │
│ │ │
│ ┌──────────────┐ ┌──────────────┐ ▼ │
│ │ Trade │───▶│ Time Series │◀──── LLM Analysis │
│ │ WebSocket │ │ Processor │ (HolySheep API) │
│ └──────────────┘ └──────────────┘ │
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ Prediction │◀───│ Signal │◀───│ Risk │ │
│ │ Engine │ │ Aggregator │ │ Calculator │ │
│ └──────────────┘ └──────────────┘ └──────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘2. Order Book 特徴量抽出の実装
Order Bookからの特徴量抽出は、予測精度に直結する重要工程です。私は以下の指標に着目し、リアルタイム処理可能な軽量パイプラインを構築しました。
- 板の傾斜（Bid-Ask Spread Ratio）: 買い注文と売り注文の 量バランス
- 注文密集度（Order Density）: 特定価格帯に集中する注文量
- 変化速度（Order Flow Velocity）: 時間あたりの板変化率
- 大口注文の存在（Whale Indicator）: 平均注文サイズとの乖離
import asyncio
import json
import numpy as np
from collections import deque
from datetime import datetime
from typing import Dict, List, Optional
import aiohttp
class OrderBookAnalyzer:
"""Order Book からボラティリティ予測用特徴量を抽出"""
def __init__(self, symbol: str, window_size: int = 100):
self.symbol = symbol
self.window_size = window_size
self.bid_history = deque(maxlen=window_size)
self.ask_history = deque(maxlen=window_size)
self.trade_history = deque(maxlen=window_size)
self.last_update_id = None
async def fetch_order_book(self, session: aiohttp.ClientSession) -> Dict:
"""Binance APIからOrder Bookを取得"""
url = f"https://api.binance.com/api/v3/depth"
params = {"symbol": f"{self.symbol.upper()}USDT", "limit": 20}
async with session.get(url, params=params) as response:
if response.status != 200:
raise ConnectionError(f"API Error: {response.status}")
return await response.json()
def extract_features(self, order_book: Dict) -> Dict[str, float]:
"""特徴量抽出"""
bids = [[float(p), float(q)] for p, q in order_book.get("bids", [])]
asks = [[float(p), float(q)] for p, q in order_book.get("asks", [])]
if not bids or not asks:
return {}
# 基本指標
best_bid = bids[0][0]
best_ask = asks[0][0]
spread = best_ask - best_bid
spread_ratio = spread / best_bid if best_bid > 0 else 0
# 板の傾斜（加重平均価格差）
bid_volume = sum(q for _, q in bids)
ask_volume = sum(q for _, q in asks)
volume_imbalance = (bid_volume - ask_volume) / (bid_volume + ask_volume)
# 注文密集度（最良気配付近の注文量比率）
depth_price = best_bid * 0.005 # 0.5%范围内的注文
near_bid_volume = sum(q for p, q in bids if p >= best_bid - depth_price)
near_ask_volume = sum(q for p, q in asks if p <= best_ask + depth_price)
density_ratio = (near_bid_volume + near_ask_volume) / (bid_volume + ask_volume)
# 価格Impact推定
mid_price = (best_bid + best_ask) / 2
impact_estimates = []
cumulative_volume = 0
for i in range(min(5, len(bids))):
cumulative_volume += bids[i][1]
price_impact = (mid_price - bids[i][0]) / mid_price
impact_estimates.append(price_impact * cumulative_volume)
return {
"timestamp": datetime.utcnow().isoformat(),
"symbol": self.symbol,
"spread": spread,
"spread_ratio": spread_ratio,
"bid_volume": bid_volume,
"ask_volume": ask_volume,
"volume_imbalance": volume_imbalance,
"density_ratio": density_ratio,
"mid_price": mid_price,
"price_impact_vwap": np.mean(impact_estimates) if impact_estimates else 0
}
def calculate_volatility_signal(self, features_list: List[Dict]) -> Dict:
"""時系列特徴量からボラティリティシグナルを生成"""
if len(features_list) < 10:
return {"signal": "insufficient_data", "confidence": 0}
spreads = [f["spread_ratio"] for f in features_list]
imbalances = [f["volume_imbalance"] for f in features_list]
# 移動平均と標準偏差
spread_ma = np.mean(spreads)
spread_std = np.std(spreads)
imbalance_ma = np.mean(imbalances)
imbalance_std = np.std(imbalances)
# シグナル強度（z-scoreベース）
current = features_list[-1]
spread_zscore = (current["spread_ratio"] - spread_ma) / spread_std if spread_std > 0 else 0
imbalance_zscore = (current["volume_imbalance"] - imbalance_ma) / imbalance_std if imbalance_std > 0 else 0
# 複合シグナル（-1 to 1）
signal = np.clip((spread_zscore * 0.6 + imbalance_zscore * 0.4) / 3, -1, 1)
confidence = min(len(features_list) / 100, 1.0) * 0.9 + 0.1