私は普段、アルゴリズムトレードや暗号資産のシグナル開発工作中、WebSocket接続の不安定さに何度も頭を悩ませてきました。「ConnectionError: timeout」「401 Unauthorized」「pong timeout」「JSON decode error」——これらのエラーは原因がわかれば一瞬で解決できるものばかりですが、初めて遭遇すると数時間つぶしてしまうことも珍しくありません。

本記事では、OKX（欧易）のパブリックWebSocket APIを使ってBTC/USDの板情報（深度データ）をリアルタイムで取得・解析するPythonコードを、ゼロから丁寧に解説します。後半では筆者が実際に踏んだエラーとその解決策を3つ以上まとめ、最後にHolySheep AIを組み合わせた高度な活用術までお届けします。

前提条件と環境準備

動作確認环境は以下の通りです。

# 必要なライブラリのインストール
pip install websockets pandas requests


本記事の動作確認済みバージョン

python --version

Python 3.11.5

pip show websockets pandas requests

websockets==11.0.3


pandas==2.0.3


requests==2.31.0

OKX WebSocketの基本構造

OKXのパブリックWebSocketエンドポイントは wss://ws.okx.com:8443/ws/v5/public です。深度データを.subscribeするには、購読したいチャンネルと銘柄をJSON形式で送信します。

{
    "op": "subscribe",
    "args": [
        {
            "channel": "books5",    # 板情報（トップ5）
            "instId": "BTC-USDT"    # インスタンスID
        }
    ]
}

リアルタイム深度データ解析の完全コード

基本実装：WebSocket接続とパース

import json
import asyncio
import websockets
from datetime import datetime
from dataclasses import dataclass
from typing import Optional


@dataclass
class DepthLevel:
    """板情報の1レベルを表現"""
    price: float
    size: float
    side: str  # "ask" or "bid"


@dataclass
class DepthSnapshot:
    """深度データのスナップショット"""
    inst_id: str
    timestamp: datetime
    asks: list[DepthLevel]  # 売板（価格昇順）
    bids: list[DepthLevel]  # 買板（価格降順）
    mid_price: float
    spread: float
    spread_bps: float  # basis points

    @classmethod
    def from_okx_data(cls, inst_id: str, data: dict) -> "DepthSnapshot":
        asks = [
            DepthLevel(price=float(p), size=float(s), side="ask")
            for p, s in data.get("asks", [])
        ]
        bids = [
            DepthLevel(price=float(p), size=float(s), side="bid")
            for p, s in data.get("bids", [])
        ]
        best_ask = asks[0].price if asks else 0.0
        best_bid = bids[0].price if bids else 0.0
        mid = (best_ask + best_bid) / 2
        spread = best_ask - best_bid
        spread_bps = (spread / mid * 10000) if mid > 0 else 0.0

        return cls(
            inst_id=inst_id,
            timestamp=datetime.utcnow(),
            asks=asks,
            bids=bids,
            mid_price=mid,
            spread=spread,
            spread_bps=spread_bps
        )

    def best_five(self) -> dict:
        """トップ5の板情報を整形"""
        return {
            "inst_id": self.inst_id,
            "timestamp": self.timestamp.isoformat(),
            "mid_price": round(self.mid_price, 2),
            "spread": round(self.spread, 2),
            "spread_bps": round(self.spread_bps, 2),
            "top_asks": [
                {"price": round(a.price, 2), "size": round(a.size, 4)}
                for a in self.asks[:5]
            ],
            "top_bids": [
                {"price": round(b.price, 2), "size": round(b.size, 4)}
                for b in self.bids[:5]
            ],
        }


class OKXDepthClient:
    """OKX WebSocket深度データクライアント"""

    WS_URL = "wss://ws.okx.com:8443/ws/v5/public"
    PING_INTERVAL = 20  # OKX推奨のping間隔（秒）

    def __init__(self, inst_id: str = "BTC-USDT", depth_levels: int = 5):
        self.inst_id = inst_id
        self.depth_levels = depth_levels
        self.channel = f"books{depth_levels}"
        self.latest_snapshot: Optional[DepthSnapshot] = None
        self._running = False

    def _build_subscribe_message(self) -> dict:
        return {
            "op": "subscribe",
            "args": [{
                "channel": self.channel,
                "instId": self.inst_id
            }]
        }

    async def _heartbeat(self, ws: websockets.WebSocketClientProtocol):
        """20秒間隔でpingを送信（OKXのタイムアウト対策）"""
        while self._running:
            try:
                await asyncio.sleep(self.PING_INTERVAL)
                if self._running:
                    await ws.send("ping")
            except Exception:
                break

    async def connect_and_listen(self):
        """WebSocket接続＋メッセージlisten"""
        self._running = True
        reconnect_delay = 1
        max_delay = 60

        while self._running:
            try:
                async with websockets.connect(
                    self.WS_URL,
                    ping_interval=None  # 手動pingのため無効化
                ) as ws:
                    reconnect_delay = 1  # 正常接続時にリセット
                    await ws.send(json.dumps(self._build_subscribe_message()))
                    print(f"[OKX] 購読開始: {self.channel} / {self.inst_id}")

                    # heartbeat coroutine 並行起動
                    hb_task = asyncio.create_task(self._heartbeat(ws))

                    async for raw in ws:
                        if raw == "pong":
                            continue

                        try:
                            msg = json.loads(raw)
                            await self._handle_message(msg)
                        except json.JSONDecodeError as e:
                            print(f"[ERROR] JSON decode failed: {e} | raw: {raw[:100]}")
                        except Exception as e:
                            print(f"[ERROR] Unexpected: {type(e).__name__}: {e}")

                    hb_task.cancel()

            except websockets.ConnectionClosed as e:
                print(f"[WARN] 切断 → {e.code} {e.reason} | {reconnect_delay}s後に再接続")
            except OSError as e:
                print(f"[ERROR] ConnectionError: {e} | {reconnect_delay}s後に再接続")
            except Exception as e:
                print(f"[ERROR] {type(e).__name__}: {e} | {reconnect_delay}s後に再接続")

            await asyncio.sleep(reconnect_delay)
            reconnect_delay = min(reconnect_delay * 2, max_delay)

    async def _handle_message(self, msg: dict):
        """メッセージの種類に応じて処理分岐"""
        if msg.get("event") == "subscribe":
            print(f"[OKX] 購読成功: {msg.get('arg', {})}")
            return

        if msg.get("event") == "error":
            print(f"[ERROR] Subscribe error: {msg.get('msg')}")
            return

        data_list = msg.get("data", [])
        if not data_list:
            return

        for item in data_list:
            snapshot = DepthSnapshot.from_okx_data(self.inst_id, item)
            self.latest_snapshot = snapshot

            # 整形出力
            info = snapshot.best_five()
            print(f"\n{'='*50}")
            print(f"[{info['timestamp']}] BTC-USDT 深度更新")
            print(f"  中値: ${info['mid_price']:,.2f}")
            print(f"  スプレッド: ${info['spread']:.2f} ({info['spread_bps']:.1f} bps)")
            print(f"  --- 売板 (Ask) ---")
            for i, ask in enumerate(info['top_asks']):
                print(f"    [{i+1}] ${ask['price']:,.2f} × {ask['size']:.4f}")
            print(f"  --- 買板 (Bid) ---")
            for i, bid in enumerate(info['top_bids']):
                print(f"    [{i+1}] ${bid['price']:,.2f} × {bid['size']:.4f}")

    def stop(self):
        self._running = False



===== 実行エントリーポイント =====

async def main():
    client = OKXDepthClient(inst_id="BTC-USDT", depth_levels=5)
    try:
        await client.connect_and_listen()
    except KeyboardInterrupt:
        print("\n[INFO] ユーザー割込み → 終了")
        client.stop()


if __name__ == "__main__":
    asyncio.run(main())

応用：HolySheep AIで深度データをAI分析する

リアルタイムで取得した深度データをそのまま保存するだけでは意味がありません。異常値を検出하거나将来の価格変動を予測するにはHolySheep AIのAPIが非常に有効です。以下は深度データの快照を HolySheep に投げて分析コメントを自動生成もらう例です。

import requests
import asyncio


===== HolySheep AI API設定 =====


重要: 実際のAPIキーは https://www.holysheep.ai/register から取得してください

HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"


注意: 絶対に api.openai.com や api.anthropic.com は使用禁止


HolySheepは独自のエンドポイント https://api.holysheep.ai/v1 を使用します


API_PRICE_PER_MTOK = {
    "gpt-4.1": 8.00,          # $/MTok
    "claude-sonnet-4.5": 15.00,
    "gemini-2.5-flash": 2.50,
    "deepseek-v3.2": 0.42,    # HolySheep最安モデル
}

def analyze_depth_with_holysheep(snapshot_data: dict, api_key: str) -> str:
    """
    深度データの分析をHolySheep AIにリクエストする
    - 板の偏りを検出
    - 流動性スコア算出
    - 異常値の警告
    """
    prompt = f"""以下のBTC-USDT深度データを分析してください:

中値: ${snapshot_data['mid_price']:,.2f}
スプレッド: ${snapshot_data['spread']:.2f} ({snapshot_data['spread_bps']:.1f} bps)

売板(Ask):
{chr(10).join([f"  ${a['price']:,.2f} × {a['size']:.4f}" for a in snapshot_data['top_asks']])}

買板(Bid):
{chr(10).join([f"  ${b['price']:,.2f} × {b['size']:.4f}" for b in snapshot_data['top_bids']])}

分析項目:
1. 板の偏り（買い優勢か売り優勢か）
2. 流動性の評価（トップ5の合計サイズから判断）
3. 異常値の有無
4. 短期的な価格動向の示唆
"""

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }

    payload = {
        "model": "deepseek-v3.2",  # $0.42/MTok — コスト効率最高
        "messages": [
            {"role": "system", "content": "あなたは暗号資産の板読み 전문가です。簡潔に分析結果を報告してください。"},
            {"role": "user", "content": prompt}
        ],
        "max_tokens": 300,
        "temperature": 0.3,
    }

    response = requests.post(
        f"{HOLYSHEEP_BASE_URL}/chat/completions",
        headers=headers,
        json=payload,
        timeout=5
    )

    if response.status_code == 401:
        raise PermissionError("APIキーが無効です。https://www.holysheep.ai/register で確認してください")
    elif response.status_code != 200:
        raise RuntimeError(f"API Error {response.status_code}: {response.text}")

    result = response.json()
    return result["choices"][0]["message"]["content"]



===== 統合メイン処理 =====

async def main_integrated():
    client = OKXDepthClient(inst_id="BTC-USDT", depth_levels=5)
    analysis_count = 0
    max_analyses = 10  # コスト管理: 10回分析したら終了

    async def analyze_task():
        nonlocal analysis_count
        while client._running and analysis_count < max_analyses:
            await asyncio.sleep(10)  # 10秒ごとに分析
            if client.latest_snapshot:
                try:
                    analysis_count += 1
                    print(f"\n[HolySheep] 分析#{analysis_count} リクエスト中...")
                    result = analyze_depth_with_holysheep(
                        client.latest_snapshot.best_five(),
                        HOLYSHEEP_API_KEY
                    )
                    print(f"[HolySheep AI 分析結果]\n{result}")

                    # コスト計算
                    input_tokens_est = 400  # 概算
                    output_tokens_est = 80
                    cost = (input_tokens_est + output_tokens_est) / 1_000_000 * API_PRICE_PER_MTOK["deepseek-v3.2"]
                    print(f"[コスト] 1回あたり約 ${cost:.6f} (DeepSeek V3.2)")

                except PermissionError as e:
                    print(f"[ERROR] {e}")
                except Exception as e:
                    print(f"[ERROR] 分析失敗: {type(e).__name__}: {e}")

        print(f"[INFO] 分析上限({max_analyses})到達 →終了")

    try:
        await asyncio.gather(
            client.connect_and_listen(),
            analyze_task()
        )
    except KeyboardInterrupt:
        print("\n[INFO] 中断 → 終了")
        client.stop()


if __name__ == "__main__":
    print("注意: YOUR_HOLYSHEEP_API_KEY を実際のキーに置き換えてください")
    asyncio.run(main_integrated())

OKX WebSocket API チャンネル早見表

チャンネル名データ内容更新頻度用途
books5トップ5の板情報リアルタイム短期トレーディング、スプレッド監視
books50トップ50の板情報リアルタイム流動性分析、アルゴリズム注文
bbo-tbt最良買気配・最良売気配tick by tick高頻度取引、最速価格取得
tickersティッカー（全データ）リアルタイム価格監視、ろうそく足作成
candle1m1分足OHLCV1分ごとテクニカル分析

向いている人・向いていない人

向いている人向いていない人
暗号資産のbotトレードを自作したい人WebSocket経験ゼロで最短スタートしたい人
深度データから流動性分析したい人コードを書かずにサービスだけで実現したい人
HolySheep等のLLMでデータ分析したい人板情報ではなく裁定取引ツールを探している人
低コスト・高頻度の価格取得が必要な人本番環境ですぐに使いたい人（要自作 Merahool 化）

価格とROI

深度データ解析のコスト構造を整理します。WebSocketによるOKXからのデータ取得自体は無料ですが、深度データをLLMで分析する場合のHolySheep AIコストは以下の通りです。

モデルOutput価格 ($/MTok)1分析あたり概算コスト1日100分析の月額コスト
DeepSeek V3.2$0.42$0.0002〜$0.001約$0.6〜$3
Gemini 2.5 Flash$2.50$0.001〜$0.003約$3〜$9
Claude Sonnet 4.5$15.00$0.005〜$0.015約$15〜$45
GPT-4.1$8.00$0.003〜$0.010約$9〜$30

深度データの分析にはdeepseek-v3.2（$0.42/MTok）が最もコスト効率が高く、1日100分析でも月額$3以下に抑えられます。HolySheep AIなら登録無料でクレジットが付くため、個人開発や検証段階でのコストリスクはゼロです。

HolySheepを選ぶ理由

深度データ解析にLLMを活用する場合、なぜHolySheep AIなのか。私の実体験から3つ理由を挙げます。

よくあるエラーと対処法

1. 「ConnectionError: timeout」——接続がタイムアウトする

# ❌ よくある失敗例: websockets.connect() にタイムアウト設定がない
async with websockets.connect("wss://ws.okx.com:8443/ws/v5/public") as ws:
    ...


✅ 解決: timeout を明示的に設定 + 例外処理を追加

import asyncio
from websockets.exceptions import InvalidURI, WebSocketException

async def safe_connect(url: str):
    timeout = 30
    try:
        async with websockets.connect(
            url,
            open_timeout=timeout,
            close_timeout=10,
            max_size=10 * 1024 * 1024  # 10MB（深度データ量大対応）
        ) as ws:
            return ws
    except websockets.InvalidURI:
        raise ValueError(f"無効なURI: {url}")
    except asyncio.TimeoutError:
        print("[ERROR] 接続タイムアウト（30秒）→ ネットワークを確認")
        raise
    except Exception as e:
        print(f"[ERROR] 接続失敗: {type(e).__name__}: {e}")
        raise

原因: ファイアウォール・プロキシ・VPNの блокировка、またはOKXサーバーの一時的な過負荷。OKXのWebSocketは中国本土外からの接続の場合、ネットワーク経路によって不安定になることがあります。解決策: 接続先に wss://aws.okx.com/ws/v5/public（AWSリージョン）を試すか、タイムアウトを長めに設定してください。

2. 「401 Unauthorized」——API認証に失敗する

# ❌ よくある失敗: APIキーのプレースホルダーをそのまま送信
headers = {
    "Authorization": "Bearer YOUR_HOLYSHEEP_API_KEY",  # 置き換えていない
    "Content-Type": "application/json",
}


✅ 解決: 環境変数からAPIキーを安全に取得

import os
from pathlib import Path

def get_api_key() -> str:
    """優先順位: 環境変数 > .envファイル > 例外"""
    key = os.environ.get("HOLYSHEEP_API_KEY")
    if key and key != "YOUR_HOLYSHEEP_API_KEY":
        return key

    # .env ファイルからの読込（python-dotenv使用）
    try:
        from dotenv import load_dotenv
        env_path = Path(__file__).parent / ".env"
        load_dotenv(env_path)
        key = os.getenv("HOLYSHEEP_API_KEY", "")
        if key:
            return key
    except ImportError:
        pass

    raise PermissionError(
        "HolySheep APIキーが未設定です。\n"
        "1. https://www.holysheep.ai/register でAPIキーを取得\n"
        "2. 環境変数 HOLYSHEEP_API_KEY を設定、または.envファイルに記述してください"
    )


使用例

API_KEY = get_api_key()
headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}

原因: APIキーを設定していない、プレースホルダーのまま実行している、あるいはキーを取り消した的可能性。2026年現在のHolySheepはキーのローテーション機能を 지원하며、無効化されたキーでアクセスすると即座に401が返されます。解決策: まずダッシュボードでキーの状態を確認し、有効なキーを環境変数経由で参照してください。

3. 「pong timeout」——ping/pongのタイムアウトで切断される

# ❌ よくある失敗: pingを送信していない
async def bad_example():
    async with websockets.connect(url) as ws:
        while True:
            msg = await ws.recv()  # OKXは25秒後に切断する
            process(msg)


✅ 解決: asyncio.coroutineでpingを並列実行

async def heartbeat(ws, interval: int = 20):
    """OKX推奨: 20秒ごとにpingを送信"""
    while True:
        await asyncio.sleep(interval)
        try:
            await ws.send("ping")
        except websockets.ConnectionClosed:
            break

async def good_example():
    async with websockets.connect(url, ping_interval=None) as ws:
        hb = asyncio.create_task(heartbeat(ws))
        try:
            async for msg in ws:
                if msg == "pong":
                    continue
                process(json.loads(msg))
        finally:
            hb.cancel()
            try:
                await hb
            except asyncio.CancelledError:
                pass

原因: OKXのWebSocketサーバーは接続後25秒以上アクティビティがない場合、自動的に接続を切断します。pingを送信しなければ必ずこのエラーが発生します。解決策: 必ずasyncio.create_task()でheartbeatを並行実行し、接続切断時にhb.cancel()でクリーンアップしてください。

4. 「JSONDecodeError: Expecting value」——空メッセージのパースに失敗

# ❌ よくある失敗: 空文字やping/pongをそのままjson.loadsする
async for raw in ws:
    msg = json.loads(raw)  # raw="" や raw="ping" で落ちる


✅ 解決: パース前にバリデーションを追加

async for raw in ws:
    # 空メッセージ・ping/pongをスキップ
    if not raw or raw in ("ping", "pong"):
        continue

    try:
        msg = json.loads(raw)
    except json.JSONDecodeError as e:
        print(f"[WARN] 不正JSONをスキップ: {e} | データ: {repr(raw[:80])}")
        continue

    # 正常処理
    await handle(msg)

原因: WebSocket仕様ではping/pongフレームが文字列ではなく单独フレームで届く場合があり、接続ハンドシェイク時の空メッセージも存在します。解決策: JSONパース前に必ずif not rawのチェックを入れ、try/exceptで安全に触れてください。

5. 再接続時の「Duplicate subscription」——購読重複エラー

# ❌ よくある失敗: 再接続時にsubscribeメッセージを再送信してしまう
async def bad_reconnect(ws, inst_id):
    await ws.send(json.dumps({"op": "subscribe", "args": [...]}))  # 重複!


✅ 解決: 購読状態を管理し、再接続時は購読済みチェック

class SubscriptionManager:
    def __init__(self):
        self._subscribed: dict[str, bool] = {}

    def subscribe(self, channel: str, inst_id: str):
        key = f"{channel}:{inst_id}"
        if self._subscribed.get(key):
            print(f"[WARN] 既に購読中: {key} → 再購読をスキップ")
            return False
        self._subscribed[key] = True
        return True

    def on_disconnect(self):
        """切断時に購読状態をリセット（サーバーがクリアするため）"""
        self._subscribed.clear()

async def robust_connect():
    manager = SubscriptionManager()
    while True:
        try:
            async with websockets.connect(OKXDepthClient.WS_URL) as ws:
                # 購読は一度だけ
                if manager.subscribe("books5", "BTC-USDT"):
                    await ws.send(json.dumps({
                        "op": "subscribe",
                        "args": [{"channel": "books5", "instId": "BTC-USDT"}]
                    }))
                else:
                    # 購読済みだが、サーバーが切断した可能性がある
                    # → サーバーがクリアしているので再購読が必要
                    manager.on_disconnect()
                    await ws.send(json.dumps({
                        "op": "subscribe",
                        "args": [{"channel": "books5", "instId": "BTC-USDT"}]
                    }))

                async for msg in ws:
                    ...
        except Exception:
            await asyncio.sleep(5)

原因: WebSocket切断後、サーバーが購読状態をクリアする前にクライアントが再接続すると、同じop: subscribeを送信してもエラーになりません。しかし稀にサーバー側で購読が残っている場合、重複購読として処理されないことがあります。解決策: 購読状態をクライアント側で管理し、再接続時はsubscribeメッセージを必ず再送信してください（サーバー側で購読状態がクリアされている保障がないため）。

まとめと次のステップ

本記事では、OKX WebSocket APIからBTC/USDTの深度データをリアルタイムで取得・解析するPython実装を解説しました。ポイントをおさらいします。

コードを試す第一步として、まずは基本実装（1つ目のコードブロック）のみを実行してみてください。WebSocket接続→深度データ表示→Ctrl+Cで中断、の流れを体験すれば、以後の応用実装もスムーズに取り組めます。

👉 HolySheep AI に登録して無料クレジットを獲得

関連リソース

関連記事