暗号通貨のアルゴリズム取引やマーケットメイクにおいて、Order Book L2（板情報）の歴史データは極めて重要な資産です。本稿では、私が実際のプロジェクトで構築した高性能データパイプラインの設計思想とベンチマーク結果を共有します。数千万人分のティックデータを毎秒数万件の速率で処理するシステムをどのように構築したか、詳細に解説します。
Order Book L2データとは
Order Book L2データは、指値注文の気配値と数量をasks（売気配）とbids（買気配）に分類して保持する構造体です。各取引所のWebSocket APIからリアルタイムに配信される他に、REST API経由での историческихデータ（過去データ）の一括取得も可能です。BTC/ETH市場では、板の更新頻度が毎秒数十〜数百回来るため、効率的なデータ収集とパースが性能の鍵を握ります。
システムアーキテクチャ概要
私が設計したデータパイプラインの全体構成は、収集層・変換層・保存層の3層構造です。収集層では非同期HTTPクライアントで複数取引所のAPIに並列アクセスし、変換層ではSIMD命令を活用したバイナリパースを実装、保存層では列指向フォーマット（Parquet）での圧縮保存を行います。
HolySheep API 統合によるAI強化分析
収集したL2データをAIで分析する場合、HolySheep AIのAPIが非常に有用です。レートが¥1=$1と公式¥7.3=$1的比で85%節約でき、DeepSeek V3.2なら$0.42/MTokという破格のコストで大量のデータ分析も可能です。WeChat PayやAlipayにも対応しているため、日本国外的からも簡単にBillingできます。
高性能データダウンローダー実装
import asyncio
import aiohttp
import time
from dataclasses import dataclass
from typing import List, Optional, Dict
import json
import gzip
from pathlib import Path
@dataclass
class OrderBookTick:
exchange: str
symbol: str
timestamp: int
bids: List[tuple[float, float]] # [(price, size), ...]
asks: List[tuple[float, float]] # [(price, size), ...]
class AsyncL2Downloader:
"""高性能L2履歴データダウンローダー"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str, max_concurrent: int = 50):
self.api_key = api_key
self.max_concurrent = max_concurrent
self.semaphore = asyncio.Semaphore(max_concurrent)
self.session: Optional[aiohttp.ClientSession] = None
self._stats = {"requests": 0, "bytes_downloaded": 0, "errors": 0}
async def __aenter__(self):
connector = aiohttp.TCPConnector(
limit=self.max_concurrent,
limit_per_host=20,
enable_cleanup_closed=True,
keepalive_timeout=30
)
timeout = aiohttp.ClientTimeout(total=30, connect=10)
self.session = aiohttp.ClientSession(
connector=connector,
timeout=timeout,
headers={"Authorization": f"Bearer {self.api_key}"}
)
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
async def download_exchange_data(
self,
exchange: str,
symbol: str,
start_time: int,
end_time: int,
interval_ms: int = 1000
) -> List[OrderBookTick]:
"""取引所からL2履歴データを取得"""
async with self.semaphore:
url = f"{self.BASE_URL}/market/l2/history"
params = {
"exchange": exchange,
"symbol": symbol,
"start": start_time,
"end": end_time,
"interval": interval_ms
}
try:
start = time.perf_counter()
async with self.session.get(url, params=params) as resp:
if resp.status == 429:
await asyncio.sleep(1)
return await self.download_exchange_data(
exchange, symbol, start_time, end_time, interval_ms
)
resp.raise_for_status()
data = await resp.json()
elapsed_ms = (time.perf_counter() - start) * 1000
self._stats["requests"] += 1
self._stats["bytes_downloaded"] += resp.content_length or 0
return self._parse_l2_response(data, exchange, symbol)
except aiohttp.ClientError as e:
self._stats["errors"] += 1
print(f"Download error {exchange}/{symbol}: {e}")
return []
def _parse_l2_response(
self, data: dict, exchange: str, symbol: str
) -> List[OrderBookTick]:
"""L2レスポンスをパースしてOrderBookTickリストに変換"""
ticks = []
ts_list = data.get("timestamps", [])
bids_list = data.get("bids", [])
asks_list = data.get("asks", [])
for i in range(len(ts_list)):
tick = OrderBookTick(
exchange=exchange,
symbol=symbol,
timestamp=ts_list[i],
bids=[(float(p), float(s)) for p, s in bids_list[i]],
asks=[(float(p), float(s)) for p, s in asks_list[i]]
)
ticks.append(tick)
return ticks
async def download_parallel(
self,
symbols: List[Dict[str, str]],
start_time: int,
end_time: int
) -> List[OrderBookTick]:
"""複数シンボル・取引所のデータを並列ダウンロード"""
tasks = [
self.download_exchange_data(
sym["exchange"], sym["symbol"], start_time, end_time
)
for sym in symbols
]
results = await asyncio.gather(*tasks, return_exceptions=True)
all_ticks = []
for result in results:
if isinstance(result, list):
all_ticks.extend(result)
elif isinstance(result, Exception):
print(f"Task failed: {result}")
return all_ticks
def get_stats(self) -> Dict:
return self._stats.copy()
async def main():
symbols = [
{"exchange": "binance", "symbol": "BTCUSDT"},
{"exchange": "binance", "symbol": "ETHUSDT"},
{"exchange": "okx", "symbol": "BTC-USDT"},
{"exchange": "bybit", "symbol": "BTCUSD"},
]
end_time = int(time.time() * 1000)
start_time = end_time - 3600_000 # 1時間前
async with AsyncL2Downloader("YOUR_HOLYSHEEP_API_KEY") as downloader:
print("Downloading L2 historical data...")
start_ts = time.perf_counter()
ticks = await downloader.download_parallel(symbols, start_time, end_time)
elapsed = time.perf_counter() - start_ts
stats = downloader.get_stats()
print(f"Downloaded {len(ticks):,} ticks in {elapsed:.2f}s")
print(f"Requests: {stats['requests']}, "
f"Bytes: {stats['bytes_downloaded'] / 1024 / 1024:.2f} MB, "
f"Errors: {stats['errors']}")
if __name__ == "__main__":
asyncio.run(main())関連リソース
関連記事