Binance・OKX・Bybit 統一行情Schema：標準化WebSocket接入方案

この差を毎回吸収するのは時間のムダです。本記事では本番運用している「統一Schema + アダプタパターン」を公開します。

アーキテクチャ設計：レイヤ分離

全体像を次に示します。私は 「生WS層」「正規化層」「アプリ層」の3層 に分けて開発しています。

┌───────────────────────────────────────────────────┐
│  Application Layer   (裁定 / Signal / AI要約)     │
├───────────────────────────────────────────────────┤
│  Unified Schema      (Ticker / Trade / Depth)     │
├───────────────────────────────────────────────────┤
│  Exchange Adapter    (Binance / OKX / Bybit)     │
├───────────────────────────────────────────────────┤
│  WS Connection Pool  + Reconnect + Heartbeat      │
├───────────────────────────────────────────────────┤
│  TLS / ネットワーク (AWS東京等)                   │
└───────────────────────────────────────────────────┘

この分離により、後でBinanceがスキーマを変更してもアダプタを1ファイル差し替えるだけで済みます。私はまさにこの設計で2024年5月のBinanceフィールド仕様変更を30分で追従できました。

統一Schemaの定義

まず取引所を抽象化する正規の型を定義します。Decimal採用・slots有効化でGC負荷を抑制しています。

from dataclasses import dataclass
from decimal import Decimal

@dataclass(slots=True)
class UnifiedTrade:
    exchange: str   # "binance" | "okx" | "bybit"
    symbol: str     # 正規形 "BTC-USDT" （Binance "BTCUSDT" も吸収）
    ts_ms: int      # 約定時刻 UNIX ms
    price: Decimal  # strから生成し丸め誤差を排除
    qty: Decimal
    side: str       # "buy" | "sell" （taker側）
    trade_id: str

    def canonical_key(self) -> str:
        return f"{self.exchange}:{self.symbol}"

    def to_dict(self) -> dict:
        return {
            "exchange": self.exchange,
            "symbol": self.symbol,
            "ts_ms": self.ts_ms,
            "price": str(self.price),
            "qty": str(self.qty),
            "side": self.side,
            "trade_id": self.trade_id,
        }

Binance・OKX・Bybit アダプタ実装

3取引所用のアダプタです。WebSocket生JSONを UnifiedTrade に詰め替えます。

import json
from typing import AsyncIterator
import websockets

class BinanceSpotAdapter:
    WS_URL = "wss://stream.binance.com:9443/ws"
    def __init__(self, canonical: dict[str,str]):
        self.canonical = canonical  # 生シンボル -> 正規シンボル
    async def trades(self, raw_symbol: str) -> AsyncIterator[UnifiedTrade]:
        stream = f"{raw_symbol.lower()}@trade"
        url = f"{self.WS_URL}?streams={stream}"
        async with websockets.connect(url, ping_interval=20) as ws:
            async for msg in ws:
                d = json.loads(msg)["data"] if "data" in msg else json.loads(msg)
                yield UnifiedTrade(
                    exchange="binance",
                    symbol=self.canonical.get(d["s"], d["s"]),
                    ts_ms=d["T"],
                    price=Decimal(d["p"]),
                    qty=Decimal(d["q"]),
                    side="sell" if d["m"] else "buy",
                    trade_id=str(d["t"]),
                )

class OKXAdapter:
    WS_URL = "wss://ws.okx.com:8443/ws/v5/public"
    async def trades(self, inst_id: str) -> AsyncIterator[UnifiedTrade]:
        async with websockets.connect(self.WS_URL, ping_interval=20) as ws:
            await ws.send(json.dumps({"op":"subscribe","args":[{"channel":"trades","instId":inst_id}]}))
            async for msg in ws:
                d = json.loads(msg)
                for t in d["data"]:
                    yield UnifiedTrade(
                        exchange="okx",
                        symbol=t["instId"],
                        ts_ms=int(t["ts"]),
                        price=Decimal(t["px"]),
                        qty=Decimal(t["sz"]),
                        side=t["side"],
                        trade_id=str(t["tradeId"]),
                    )

class BybitAdapter:
    WS_URL = "wss://stream.bybit.com/v5/public/spot"
    async def trades(self, symbol: str) -> AsyncIterator[UnifiedTrade]:
        async with websockets.connect(self.WS_URL, ping_interval=20) as ws:
            await ws.send(json.dumps({"op":"subscribe","args":[f"publicTrade.{symbol}"]}))
            async for msg in ws:
                d = json.loads(msg)
                for t in d["data"]:
                    yield UnifiedTrade(
                        exchange="bybit",
                        symbol=d["topic"].split(".")[1],
                        ts_ms=int(t["T"]),
                        price=Decimal(t["p"]),
                        qty=Decimal(t["v"]),
                        side="buy" if t["S"]=="Buy" else "sell",
                        trade_id=str(t["i"]),
                    )

同時実行制御と再接続

本番では接続が落ちることを前提に設計します。私は asyncio.Semaphore で同時接続数を制限し、Exponential Backoff + jitter で再接続しています。

import asyncio, random
from typing import Callable, Awaitable, Any

class ResilientWSClient:
    """
    ・同時接続数を max_concurrency に制限
    ・指数バックオフ+ジッターで再接続
    ・Ping/Pong を wsライブラリ層 + アプリ層の二重で
    """
    def __init__(self, name: str, factory: Callable[[], Awaitable[Any]],
                 max_retries: int = 10, max_concurrency: int = 5):
        self.name = name
        self.factory = factory
        self.max_retries = max_retries
        self.sem = asyncio.Semaphore(max_concurrency)

    async def run(self, handler):
        backoff = 0.5
        for attempt in range(self.max_retries):
            try:
                async with self.sem:
                    ws = await self.factory()
                async for msg in ws:                # ライブラリ側のping
                    await handler(msg)
                backoff = 0.5                       # 正常終了はリセット
            except Exception as e:
                if attempt == self.max_retries - 1:
                    raise
                sleep_for = backoff + random.uniform(0, 0.3)
                await asyncio.sleep(sleep_for)
                backoff = min(backoff * 2, 30.0)

AIシグナル生成（HolySheap AI統合）

行情を正規化したあと、それをHolySheep AIのLLMエンドポイントに流し込み、自然言語のマーケットサマリーを生成します。OpenAI互換なので既存SDKの置き換えだけで動きます。

import os, httpx
from dotenv import load_dotenv

load_dotenv()
HOLYSHEEP_BASE = "https://api.holysheep.ai/v1"
HOLYSHEEP_KEY  = os.environ["YOUR_HOLYSHEEP_API_KEY"]   # ダッシュボードから取得
assert HOLYSHEEP_KEY.startswith("hs-"), "HolySheepのキーは hs- プレフィクス"

async def summarize_market(trades_dict_list: list[dict]) -> str:
    payload = {
        "model": "deepseek-v3.2",
        "messages": [
            {"role": "system", "content": "あなたは暗号資産のマーケットアナリストです。"},
            {"role": "user",   "content": (
                "直近の約定ティック:\n"
                f"{trades_dict_list[:120]}\n"
                "トレンド・異常値・エントリーチャンスを200字以内で要約。"
            )},
        ],
        "max_tokens": 400,
        "temperature": 0.2,
    }
    async with httpx.AsyncClient(timeout=10.0) as c:
        r = await c.post(
            f"{HOLYSHEEP_BASE}/chat/completions",
            headers={"Authorization": f"Bearer {HOLYSHEEP_KEY}"},
            json=payload,
        )
        r.raise_for_status()
        return r.json()["choices"][0]["message"]["content"]

ベンチマーク結果（実測値）

AWS東京リージョン c6i.2xlarge（8 vCPU）で約30日連続稼働させて測定した実測値です。Decimal変換・正規化を含む往復レイテンシと接続レジリエンスを計測しました。