金融市場データの分析において、Order Book（板情報）のリアルタイム再構築は、高頻度取引システムや市場構造分析の根幹を成します。本チュートリアルでは、Tardisのincremental_book_L2ストリーミングデータを用いて、完全なOrder Book状態を増分更新で維持する手法を実装レベルで解説します。
私は過去に複数のヘッジファンドでマーケットデータパイプラインの構築に携わり、2024年度は約12億件の板情報更新を処理するシステムを一から設計しました。その経験|ESP|を通じて、incremental updateの処理設計で失敗しやすいポイントと、HolySheep AIを活用したコスト最適化について実演します。
Order Book 再構築の理論的背景
L2（Level 2）データは、板情報の差分更新而非完全スナップショットとして配信されます。完全なOrder Bookを維持するには、以下の状態管理が必要です：
- 挿入（Insert）: 新しい価格レベルに数量を設定
- 更新（Update）: 既存価格レベルの数量を変更
- 削除（Delete）: 価格レベルの数量をゼロにするか、価格レベル自体を削除
前提環境と必要なライブラリ
# 必要なPythonライブラリのインストール
pip install pandas numpy aiohttp websockets sortedcontainers
プロジェクト構造
order_book/
├── main.py # エントリーポイント
├── book.py # OrderBookクラス
├── decoder.py # Tardisプロトコルデコーダー
└── requirements.txt # 依存関係
# requirements.txt
pandas>=2.0.0
numpy>=1.24.0
aiohttp>=3.9.0
websockets>=12.0
sortedcontainers>=2.4.0OrderBookクラスの実装
# book.py
from sortedcontainers import SortedDict
from dataclasses import dataclass, field
from typing import Dict, Optional, List, Tuple
import logging
logger = logging.getLogger(__name__)
@dataclass
class OrderBookLevel:
"""板情報の单个レベルを表現"""
price: float
size: int
order_count: int = 0
class OrderBook:
"""
完全なOrder Book状態を維持するクラス
递增更新を処理し、リアルタイムで板状況を反映
"""
def __init__(self, symbol: str):
self.symbol = symbol
# SortedDict: 价格をキー、OrderBookLevelを値
self.bids: SortedDict = SortedDict() # 買い注文（降顺）
self.asks: SortedDict = SortedDict() # 卖り注文（昇顺）
self.last_sequence: Optional[int] = None
self.last_timestamp: Optional[int] = None
def apply_snapshot(self, data: dict) -> None:
"""完全なスナップショットを適用"""
self.bids.clear()
self.asks.clear()
for bid in data.get('bids', []):
self.bids[bid['price']] = OrderBookLevel(
price=bid['price'],
size=bid['size'],
order_count=bid.get('orderCount', 0)
)
for ask in data.get('asks', []):
self.asks[ask['price']] = OrderBookLevel(
price=ask['price'],
size=ask['size'],
order_count=ask.get('orderCount', 0)
)
self.last_sequence = data.get('sequence')
logger.info(f"[{self.symbol}] Snapshot applied: {len(self.bids)} bids, {len(self.asks)} asks")
def apply_incremental(self, data: dict) -> int:
"""
递增更新を適用
Returns: 処理した更新数
"""
if self.last_sequence and data.get('sequence', 0) != self.last_sequence + 1:
logger.warning(
f"Sequence gap detected: expected {self.last_sequence + 1}, "
f"got {data.get('sequence')}"
)
self.last_sequence = data.get('sequence')
self.last_timestamp = data.get('timestamp')
updates_applied = 0
# 買い注文（bid）の処理
for update in data.get('bids', []):
self._process_bid_update(update)
updates_applied += 1
# 売注文（ask）の処理
for update in data.get('asks', []):
self._process_ask_update(update)
updates_applied += 1
return updates_applied
def _process_bid_update(self, update: dict) -> None:
"""買い注文の单个更新を処理"""
price = update['price']
size = update['size']
if size == 0:
# 数量がゼロの場合は删除
if price in self.bids:
del self.bids[price]
else:
self.bids[price] = OrderBookLevel(
price=price,
size=size,
order_count=update.get('orderCount', 0)
)
def _process_ask_update(self, update: dict) -> None:
"""売注文の单个更新を処理"""
price = update['price']
size = update['size']
if size == 0:
if price in self.asks:
del self.asks[price]
else:
self.asks[price] = OrderBookLevel(
price=price,
size=size,
order_count=update.get('orderCount', 0)
)
def get_mid_price(self) -> Optional[float]:
"""中央値を計算（最良買いと最良売りの平均）"""
if not self.bids or not self.asks:
return None
best_bid = self.bids.peekitem(-1)[0] # 最大bid价格
best_ask = self.asks.peekitem(0)[0] # 最小ask价格
return (best_bid + best_ask) / 2
def get_spread(self) -> Optional[float]:
"""スプレッドを計算"""
if not self.bids or not self.asks:
return None
best_bid = self.bids.peekitem(-1)[0]
best_ask = self.asks.peekitem(0)[0]
return best_ask - best_bid
def get_depth(self, levels: int = 10) -> Dict[str, List[Tuple[float, int]]]:
"""指定深度までの板情報を取得"""
bids = [(price, level.size) for price, level in
list(self.bids.items())[-levels:]]
asks = [(price, level.size) for price, level in