暗号通貨取引所の板情報（Order Book）は、リアルタイムの市場深度と流動性を把握する上で不可欠なデータです。私は以前、アルゴリズム取引システムの開発において、1秒間に数百件の注文更新を処理する必要があり、Binance WebSocket APIを活用した高頻度データ収集アーキテクチャを実装しました。本記事では、その際に構築したアーキテクチャの詳細と、HolySheep AIを活用したAI統合アプローチについて解説します。
ユースケース：ECサイトのAIカスタマーサービス最適化
具体的なユースケースとして、ECサイトのAIチャットボットがあります。ユーザーが「在庫状況は？」と質問した際、リアルタイムの板情報を解析することで、「現在A社は ¥1,200 で 0.5 BTC を売却可能、B社は ¥1,195 で 1.2 BTC を売却可能」といった具体的な情報を基に、より精度の高い回答が可能になります。
アーキテクチャ概要
高頻度データ収集システムは、以下の3層構造で設計します。
- WebSocket接続層：Binance公式WebSocket APIとのリアルタイム接続
- バッファリング層：メモリ内バッファによるバースト制御
- 処理・配信層：収集データの加工と下流システムへの配信
実装：WebSocket接続とデータ収集
まずは、基本となるWebSocket接続クラスを作成します。
import websocket
import json
import threading
import time
from collections import deque
from datetime import datetime
class BinanceOrderBookCollector:
"""
Binance WebSocket API から板情報を高頻度で収集するクラス
"""
def __init__(self, symbol='btcusdt', depth=100,
buffer_size=10000, on_update_callback=None):
self.symbol = symbol.lower()
self.depth = depth
self.buffer_size = buffer_size
self.on_update_callback = on_update_callback
# WebSocket URL
self.ws_url = f"wss://stream.binance.com:9443/ws/{self.symbol}@depth{depth}@100ms"
# データバッファ
self.order_book_buffer = deque(maxlen=buffer_size)
self.last_update_id = None
# スレッド管理
self.ws = None
self.running = False
self.reconnect_delay = 1
self.max_reconnect_delay = 60
# 接続統計
self.stats = {
'messages_received': 0,
'messages_per_second': 0,
'last_message_time': None,
'connection_errors': 0
}
self._stats_lock = threading.Lock()
def start(self):
"""WebSocket接続を開始"""
self.running = True
self.connection_thread = threading.Thread(target=self._run_websocket, daemon=True)
self.connection_thread.start()
print(f"[{datetime.now()}] Binance {self.symbol.upper()} 接続開始")
def _run_websocket(self):
"""WebSocket接続のメインループ"""
while self.running:
try:
self.ws = websocket.WebSocketApp(
self.ws_url,
on_message=self._on_message,
on_error=self._on_error,
on_close=self._on_close,
on_open=self._on_open
)
self.ws.run_forever(ping_interval=30, ping_timeout=10)
except Exception as e:
print(f"[{datetime.now()}] WebSocketエラー: {e}")
with self._stats_lock:
self.stats['connection_errors'] += 1
if self.running:
time.sleep(self.reconnect_delay)
# 指数バックオフ
self.reconnect_delay = min(self.reconnect_delay * 2,
self.max_reconnect_delay)
def _on_open(self, ws):
print(f"[{datetime.now()}] WebSocket接続確立")
self.reconnect_delay = 1 # リセット
def _on_message(self, ws, message):
try:
data = json.loads(message)
# 統計更新
with self._stats_lock:
self.stats['messages_received'] += 1
self.stats['last_message_time'] = time.time()
now = time.time()
if self.stats.get('_last_calc_time'):
elapsed = now - self.stats['_last_calc_time']
if elapsed >= 1.0:
self.stats['messages_per_second'] = \
self.stats['messages_received'] / elapsed
self.stats['_last_calc_time'] = now
else:
self.stats['_last_calc_time'] = now
# バッファに追加
order_book = self._parse_order_book(data)
self.order_book_buffer.append({
'timestamp': datetime.now().isoformat(),
'data': order_book
})
# コールバック実行
if self.on_update_callback:
self.on_update_callback(order_book)
except json.JSONDecodeError as e:
print(f"JSON解析エラー: {e}")
def _parse_order_book(self, data):
"""板情報データをパース"""
return {
'last_update_id': data.get('u') or data.get('lastUpdateId'),
'bids': [[float(p), float(q)] for p, q in data.get('b', data.get('bids', []))],
'asks': [[float(p), float(q)] for p, q in data.get('a', data.get('asks', []))],
'symbol': self.symbol.upper()
}
def _on_error(self, ws, error):
print(f"[{datetime.now()}] WebSocketエラー: {error}")
with self._stats_lock:
self.stats['connection_errors'] += 1
def _on_close(self, ws, close_status_code, close_msg):
print(f"[{datetime.now()}] WebSocket切断: {close_status_code}")
def get_latest_order_book(self):
"""最新の手形帳を取得"""
if self.order_book_buffer:
return self.order_book_buffer[-1]['data']
return None
def get_statistics(self):
"""接続統計を取得"""
with self._stats_lock:
return self.stats.copy()
def stop(self):
"""接続を停止"""
self.running = False
if self.ws:
self.ws.close()
print(f"[{datetime.now()}] 接続停止")
使用例
def on_order_book_update(order_book):
"""板情報更新時のコールバック"""
best_bid = order_book['bids'][0] if order_book['bids'] else None
best_ask = order_book['asks'][0] if order_book['asks'] else None
if best_bid and best_ask:
spread = best_ask[0] - best_bid[0]
spread_pct = (spread / best_bid[0]) * 100
print(f"spread: {spread:.2f} ({spread_pct:.4f}%)")
collector = BinanceOrderBookCollector(symbol='btcusdt', depth=100)
collector.start()
AI統合：BinanceデータとHolySheep APIの連携
収集した板情報をAIで分析し、自然言語で市場状況を解説するシステムを構築します。HolySheep AI APIを使用すると、レートが¥1=$1