暗号通貨トレーディングBotを運用していて、Bybitの永続契約（Perpetual Futures）のリアルタイム価格データ取得に課題を感じていませんか？本稿では、私自身のAlgorithmic Tradingプロジェクトで直面した問題を解決した実体験をもとに、Bybit WebSocket APIの効率的な接続方法から、データ整形、そしてHolySheep AIを活用したAI駆動型分析パイプラインの構築まで、包括的に解説します。

ユースケース：私のBTC/USDT自動売買Botの実体験

私は2024年初頭、BTC/USDTのスキャルピングBotをPythonで構築していました。当初、REST APIのPolling方式で価格を取得していましたが、約200-500msのレイテンシとAPIリクエスト制限（1秒間に10回まで）に直面。高頻度な裁定取引には完全不向きでした。

解決策としてWebSocketリアルタイムストリーミングに移行した結果、レイテンシは50ms未満に低下し、1秒間に100回以上の価格更新を処理できるようになりました。この過程でBybit WebSocket APIの構造を深く理解し、HolySheep AIの高性能APIサービスと組み合わせたリアルタイム分析システムを構築しました。

Bybit 永続契約 WebSocket APIの基礎

接続エンドポイント

BybitのWebSocket APIは Public（認証不要）とPrivate（認証必要）の2系統があります。永続契約のリアルタイム行情はPublicエンドポイントを利用します。

# Bybit WebSocket接続エンドポイント

メインネット

BYBIT_WS_URL = "wss://stream.bybit.com/v5/public/linear"


テストネット

BYBIT_TEST_WS_URL = "wss://stream-testnet.bybit.com/v5/public/linear"


対応チャンネル


- orderbook.50: 50深度の板情報


- orderbook.200: 200深度の板情報


- trade: リアルタイム取引履歴


- publicTrade: 公開取引履歴


- ticker: リアルタイムティッカー情報


- kline_1: 1分足ローソク足


- liquidation: 強制清算イベント

リアルタイム行情取得の実装

以下は、私の一軍Botで実際に運用しているWebSocket接続クラスです。Bybit公式ドキュメントの情報をもとに、reconnection処理とデータ整形を実装しています。

import json
import asyncio
import websockets
from datetime import datetime
from typing import Dict, Callable, Optional, List
import threading
import queue
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class BybitWebSocketClient:
    """
    Bybit 永続契約 WebSocket リアルタイム行情クライアント
    対応チャンネル: orderbook, trade, ticker, kline
    """
    
    def __init__(self, base_url: str = "wss://stream.bybit.com/v5/public/linear"):
        self.base_url = base_url
        self.websocket = None
        self.running = False
        self.subscriptions: List[Dict] = []
        self.message_queue = queue.Queue(maxsize=10000)
        self._reconnect_delay = 1
        self._max_reconnect_delay = 60
        
    async def connect(self):
        """WebSocket接続確立"""
        try:
            self.websocket = await websockets.connect(
                self.base_url,
                ping_interval=20,
                ping_timeout=10,
                max_size=10 * 1024 * 1024  # 10MB
            )
            self.running = True
            self._reconnect_delay = 1
            logger.info("Bybit WebSocket接続成功")
            
            # 購読開始
            if self.subscriptions:
                await self._send_subscribe()
                
        except Exception as e:
            logger.error(f"WebSocket接続失敗: {e}")
            raise
            
    async def _send_subscribe(self):
        """購読リクエスト送信"""
        subscribe_msg = {
            "op": "subscribe",
            "args": self.subscriptions
        }
        await self.websocket.send(json.dumps(subscribe_msg))
        logger.info(f"購読開始: {self.subscriptions}")
        
    async def subscribe(self, channels: List[str], symbols: List[str]):
        """
        チャンネルの購読開始
        
        Args:
            channels: チャンネルリスト (例: ["orderbook.50", "trade", "ticker"])
            symbols: 銘柄リスト (例: ["BTCUSDT", "ETHUSDT"])
        """
        for channel in channels:
            for symbol in symbols:
                self.subscriptions.append(f"{channel}.{symbol}")
                
        if self.websocket and self.running:
            await self._send_subscribe()
            
    async def listen(self, callback: Optional[Callable] = None):
        """メッセージ受信ループ"""
        async for message in self.websocket:
            if not self.running:
                break
                
            try:
                data = json.loads(message)
                await self._handle_message(data, callback)
            except json.JSONDecodeError as e:
                logger.error(f"JSON解析エラー: {e}")
            except Exception as e:
                logger.error(f"メッセージ処理エラー: {e}")
                
    async def _handle_message(self, data: Dict, callback: Optional[Callable]):
        """メッセージ種類別の処理"""
        if data.get("op") == "subscribe":
            logger.info(f"購読確認: {data.get('success')}")
            return
            
        if data.get("topic"):
            topic = data["topic"]
            
            if "orderbook" in topic:
                await self._handle_orderbook(data, callback)
            elif "trade" in topic:
                await self._handle_trade(data, callback)
            elif "ticker" in topic:
                await self._handle_ticker(data, callback)
            elif "kline" in topic:
                await self._handle_kline(data, callback)
                
    async def _handle_orderbook(self, data: Dict, callback: Optional[Callable]):
        """板情報処理"""
        orderbook_data = {
            "symbol": data.get("topic", "").split(".")[-1],
            "timestamp": datetime.now().isoformat(),
            "data_type": "orderbook",
            "bids": data["data"].get("b", []),  # [価格, 数量]
            "asks": data["data"].get("a", []),
            "update_id": data["data"].get("u")
        }
        
        if callback:
            await callback(orderbook_data)
        else:
            self.message_queue.put(orderbook_data)
            
    async def _handle_trade(self, data: Dict, callback: Optional[Callable]):
        """リアルタイム取引処理"""
        for trade in data["data"]:
            trade_data = {
                "symbol": data.get("topic", "").split(".")[-1],
                "timestamp": datetime.now().isoformat(),
                "data_type": "trade",
                "trade_id": trade["i"],
                "price": float(trade["p"]),
                "quantity": float(trade["v"]),
                "side": trade["S"],  # Buy/Sell
                "trade_time": trade["T"]
            }
            
            if callback:
                await callback(trade_data)
                
    async def _handle_ticker(self, data: Dict, callback: Optional[Callable]):
        """ティッカー情報処理"""
        ticker = data["data"]
        ticker_data = {
            "symbol": ticker["symbol"],
            "timestamp": datetime.now().isoformat(),
            "data_type": "ticker",
            "last_price": float(ticker["lastPrice"]),
            "mark_price": float(ticker["markPrice"]),
            "index_price": float(ticker["indexPrice"]),
            "funding_rate": float(ticker["fundingRate"]),
            "next_funding_time": ticker["nextFundingTime"],
            "volume_24h": float(ticker["volume24h"]),
            "turnover_24h": float(ticker["turnover24h"])
        }
        
        if callback:
            await callback(ticker_data)
            
    async def _handle_kline(self, data: Dict, callback: Optional[Callable]):
        """ローソク足処理"""
        kline = data["data"]
        kline_data = {
            "symbol": kline["symbol"],
            "timestamp": datetime.now().isoformat(),
            "data_type": "kline",
            "interval": data["topic"].split("_")[1],
            "open": float(kline["open"]),
            "high": float(kline["high"]),
            "low": float(kline["low"]),
            "close": float(kline["close"]),
            "volume": float(kline["volume"]),
            "confirm": kline["confirm"]
        }
        
        if callback:
            await callback(kline_data)
            
    async def auto_reconnect(self):
        """自動再接続（指数バックオフ）"""
        while self.running:
            try:
                await self.connect()
                await self.listen()
            except (websockets.exceptions.ConnectionClosed, 
                    ConnectionError, OSError) as e:
                logger.warning(f"切断検出、再接続まで{self._reconnect_delay}秒待機")
                await asyncio.sleep(self._reconnect_delay)
                self._reconnect_delay = min(
                    self._reconnect_delay * 2, 
                    self._max_reconnect_delay
                )
                
    def start_background(self, loop: asyncio.AbstractEventLoop):
        """バックグラウンドでWebSocket接続開始"""
        asyncio.ensure_future(self.auto_reconnect(), loop=loop)



=====  использование例 =====

async def on_market_data(data: dict):
    """リアルタイム行情受領時のコールバック"""
    if data["data_type"] == "trade":
        print(f"約定: {data['symbol']} @ {data['price']} x {data['quantity']}")
        
    elif data["data_type"] == "ticker":
        print(f"ティッカー: {data['symbol']} | "
              f"現在値: {data['last_price']} | "
              f"マーク価格: {data['mark_price']} | "
              f"資金調達率: {data['funding_rate']*100:.4f}%")


async def main():
    client = BybitWebSocketClient()
    
    # 購読設定
    await client.connect()
    await client.subscribe(
        channels=["orderbook.50", "trade", "ticker"],
        symbols=["BTCUSDT", "ETHUSDT", "SOLUSDT"]
    )
    
    # リスン開始
    await client.listen(callback=on_market_data)


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

HolySheep AI × Bybit WebSocket：AI分析パイプラインの構築

リアルタイム行情をWebSocketで受信できたら、次はHolySheep AIの高速・低遅延APIを活用したAI分析パイプラインを構築します。以下は、受領した取引データを基に感情分析とシグナル生成を行うシステムです。

import aiohttp
import asyncio
import json
from datetime import datetime
from typing import List, Dict, Optional
from dataclasses import dataclass, asdict
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


@dataclass
class HolySheepConfig:
    """HolySheep AI 設定"""
    api_key: str = "YOUR_HOLYSHEEP_API_KEY"  # 実際のキーに置換
    base_url: str = "https://api.holysheep.ai/v1"  # HolySheep公式エンドポイント
    model: str = "gpt-4.1"  # 2026年価格: $8/MTok（レート¥1=$1）
    
    @property
    def headers(self) -> Dict[str, str]:
        return {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }


@dataclass
class TradingSignal:
    """取引シグナル"""
    symbol: str
    timestamp: str
    signal_type: str  # "BUY", "SELL", "HOLD"
    confidence: float  # 0.0 - 1.0
    reasoning: str
    price: float
    volume_24h: float
    funding_rate: float


class HolySheepAnalyzer:
    """
    HolySheep AI APIを使用した市場分析クラス
    特徴:
    - ¥1=$1の為替レート（公式比85%節約）
    - <50msレイテンシ
    - WeChat Pay / Alipay対応
    """
    
    def __init__(self, config: HolySheepConfig):
        self.config = config
        self.session: Optional[aiohttp.ClientSession] = None
        
    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self
        
    async def __aexit__(self, *args):
        if self.session:
            await self.session.close()
            
    async def analyze_market_sentiment(
        self, 
        symbol: str,
        recent_trades: List[Dict],
        ticker: Dict
    ) -> Optional[TradingSignal]:
        """
        市場感情分析と取引シグナル生成
        
        Args:
            symbol: 銘柄名
            recent_trades: 直近の取引履歴（WebSocketから取得）
            ticker: 現在のティッカー情報
            
        Returns:
            TradingSignal: 取引シグナル
        """
        # プロンプト構築
        trades_summary = self._format_trades(recent_trades)
        
        prompt = f"""あなたは暗号通貨トレーディングExpertです。
以下の{symbol}市場データに基づき、取引シグナルを生成してください。

【ティッカー情報】
- 現在価格: ${ticker['last_price']:,.2f}
- マーク価格: ${ticker['mark_price']:,.2f}
- 24時間volume: {ticker['volume_24h']:,.2f}
- 資金調達率: {ticker['funding_rate']*100:.4f}%

【直近取引（最新20件）】
{trades_summary}

出力形式（JSONのみ）:
{{
  "signal_type": "BUY"または"SELL"または"HOLD",
  "confidence": 0.0-1.0,
  "reasoning": "分析根拠（50文字以内）"
}}"""
        
        try:
            response = await self._call_holysheep(prompt)
            
            if response and response.get("signal_type"):
                return TradingSignal(
                    symbol=symbol,
                    timestamp=datetime.now().isoformat(),
                    signal_type=response["signal_type"],
                    confidence=float(response["confidence"]),
                    reasoning=response.get("reasoning", ""),
                    price=ticker["last_price"],
                    volume_24h=ticker["volume_24h"],
                    funding_rate=ticker["funding_rate"]
                )
                
        except Exception as e:
            logger.error(f"分析エラー ({symbol}): {e}")
            
        return None
        
    async def _call_holysheep(self, prompt: str) -> Optional[Dict]:
        """HolySheep AI API呼び出し"""
        payload = {
            "model": self.config.model,
            "messages": [
                {"role": "system", "content": "あなたは暗号通貨市場分析Expertです。"},
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.3,  # 低temperatureで一貫性確保
            "max_tokens": 200
        }
        
        async with self.session.post(
            f"{self.config.base_url}/chat/completions",
            headers=self.config.headers,
            json=payload
        ) as resp:
            if resp.status == 200:
                result = await resp.json()
                content = result["choices"][0]["message"]["content"]
                return json.loads(content)
            else:
                error_text = await resp.text()
                logger.error(f"APIエラー {resp.status}: {error_text}")
                return None
                
    def _format_trades(self, trades: List[Dict]) -> str:
        """取引履歴の整形"""
        lines = []
        for trade in trades[-20:]:
            side = "▲買い" if trade.get("side") == "Buy" else "▼売り"
            lines.append(
                f"{side} ${trade.get('price', 0):,.2f} x {trade.get('quantity', 0):.4f}"
            )
        return "\n".join(lines)
        
        

===== 統合システム例 =====

class TradingBotWithAI:
    """WebSocket + HolySheep AI 統合トレーディングBot"""
    
    def __init__(self, holysheep_config: HolySheepConfig):
        self.analyzer = HolySheepAnalyzer(holysheep_config)
        self.recent_trades: Dict[str, List[Dict]] = {}  # symbol -> trades
        self.current_tickers: Dict[str, Dict] = {}
        self.trade_buffer_size = 100
        
    async def process_trade(self, trade_data: Dict):
        """取引データの処理"""
        symbol = trade_data["symbol"]
        
        if symbol not in self.recent_trades:
            self.recent_trades[symbol] = []
            
        self.recent_trades[symbol].append(trade_data)
        
        # バッファサイズ制限
        if len(self.recent_trades[symbol]) > self.trade_buffer_size:
            self.recent_trades[symbol] = self.recent_trades[symbol][-self.trade_buffer_size:]
            
        # 一定件数ごとにAI分析実行
        if len(self.recent_trades[symbol]) % 20 == 0:
            await self._run_analysis(symbol)
            
    async def process_ticker(self, ticker_data: Dict):
        """ティッカーデータの処理"""
        self.current_tickers[ticker_data["symbol"]] = ticker_data
        
    async def _run_analysis(self, symbol: str):
        """AI分析実行"""
        if symbol not in self.current_tickers:
            return
            
        async with self.analyzer as analyzer:
            signal = await analyzer.analyze_market_sentiment(
                symbol=symbol,
                recent_trades=self.recent_trades.get(symbol, []),
                ticker=self.current_tickers[symbol]
            )
            
            if signal:
                logger.info(
                    f"シグナル検出 [{symbol}]: {signal.signal_type} "
                    f"(信頼度: {signal.confidence*100:.1f}%) - {signal.reasoning}"
                )
                
                # シグナルに応じた取引執行（Botロジック）
                await self._execute_signal(signal)
                
    async def _execute_signal(self, signal: TradingSignal):
        """シグナルに基づく取引執行"""
        # 手数料計算（Bybit USDT永続契約: メイカー -0.025%, テイカー 0.075%）
        if signal.confidence >= 0.8:
            logger.info(f"高信頼度シグナル執行準備: {signal.signal_type}")
            # 実際の注文執行処理はBybit Private APIを使用
        elif signal.confidence >= 0.6:
            logger.info(f"中信頼度シグナル監視対象: {signal.signal_type}")



===== 使用例 =====

async def integrated_trading_example():
    """統合Botの使用例"""
    
    # HolySheep設定（¥1=$1レートでコスト効率最大化）
    holysheep_config = HolySheepConfig(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        model="gpt-4.1"  # $8/MTok
    )
    
    bot = TradingBotWithAI(holysheep_config)
    
    # WebSocketクライアントと連携
    ws_client = BybitWebSocketClient()
    
    async def on_market_data(data: Dict):
        if data["data_type"] == "trade":
            await bot.process_trade(data)
        elif data["data_type"] == "ticker":
            await bot.process_ticker(data)
            
    await ws_client.connect()
    await ws_client.subscribe(
        channels=["trade", "ticker"],
        symbols=["BTCUSDT", "ETHUSDT"]
    )
    
    await ws_client.listen(callback=on_market_data)



===== 成本計算例（HolySheep ¥1=$1レート）=====

def calculate_monthly_cost():
    """月間コスト計算（HolySheep ¥1=$1）"""
    
    # 月間Token消費見積もり
    monthly_input_tokens = 10_000_000  # 10M input tokens
    monthly_output_tokens = 1_000_000   # 1M output tokens
    
    # 2026年 HolySheep価格
    prices = {
        "gpt-4.1": {"input": 8, "output": 8},      # $8/MTok
        "claude-sonnet-4.5": {"input": 15, "output": 15},  # $15/MTok
        "gemini-2.5-flash": {"input": 2.50, "output": 2.50},  # $2.50/MTok
        "deepseek-v3.2": {"input": 0.42, "output": 0.42}  # $0.42/MTok
    }
    
    print("=== HolySheep AI 月間コスト（¥1=$1レート）===\n")
    
    for model, price in prices.items():
        input_cost = (monthly_input_tokens / 1_000_000) * price["input"]
        output_cost = (monthly_output_tokens / 1_000_000) * price["output"]
        total = input_cost + output_cost
        
        print(f"{model}:")
        print(f"  Input: ${input_cost:.2f} (¥{input_cost:.2f})")
        print(f"  Output: ${output_cost:.2f} (¥{output_cost:.2f})")
        print(f"  合計: ${total:.2f} (¥{total:.2f})")
        print()


if __name__ == "__main__":
    # コスト計算例
    calculate_monthly_cost()
    
    # 統合Bot実行（コメント解除して使用）
    # asyncio.run(integrated_trading_example())

Bybit WebSocket 行情接入 サービス比較

暗号通貨市場の行情取得サービスには複数の選択肢があります。以下に主要な選択肢とHolySheep AIを組み合わせた場合の優位性を比較します。

項目 Bybit公式WebSocket 第三者aggregator Bybit REST API Polling Bybit WS + HolySheep AI
レイテンシ <50ms 50-200ms 200-500ms <50ms + AI分析追加
データ信頼性 最高（公式） 中〜高 高 最高（公式 + AI検証）
APIコスト 無料 $29-499/月 無料 行情:無料 + AI:¥1=$1
AI分析統合 ✗ △（制限あり） △ ✓（ネイティブ）
対応モデル - 限定 - GPT-4.1/Claude/Gemini/DeepSeek
決済方法 カード/暗号 カードのみ カード/暗号 カード + WeChat Pay/Alipay
再接続処理 手動実装 自動 N/A 自動（指数バックオフ）
個人開発者向 ★★★★★ ★★☆☆☆ ★★★☆☆ ★★★★★

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

✓ 向いている人

✗ 向いていない人

価格とROI

HolySheep AIの2026年価格は、暗号通貨Bot開発における明確なコスト優位性を提供します。

モデル Input ($/MTok) Output ($/MTok) 1M Tokensコスト 適用場面
DeepSeek V3.2 $0.42 $0.42 ¥420相当 高频分析・コスト重視
Gemini 2.5 Flash $2.50 $2.50 ¥2,500相当 バランス型、一般分析
GPT-4.1 $8.00 $8.00 ¥8,000相当 高精度判断・シグナル
Claude Sonnet 4.5 $15.00 $15.00 ¥15,000相当 複雑分析・長文生成

ROI計算事例（私の場合）：

HolySheepを選ぶ理由

私がHolySheep AIを主要なAI Providerとして採用した理由は以下の5点です：

  1. ¥1=$1の為替レート：HolySheepは ¥1=$1 の固定レートを提供。公式レート（¥7.3=$1）との比較で85%のコスト削減を実現这是我选择HolySheep的主要原因。
  2. <50ms 超低レイテンシ：暗号通貨市場は秒単位の動きが重要。HolySheepのAPI応答速度は私のBotの意思決定ループに完全適合します。
  3. WeChat Pay / Alipay対応：日本の銀行振込や海外カードに制約されない決済手段は、国際的なプロジェクトにおいて柔軟性を提供します。
  4. 登録で無料クレジット：新規登録者は即座にAPIを試せる無料クレジットが付与され、実商用前に性能検証できます。
  5. マルチモデル対応：GPT-4.1、Claude Sonnet 4.5、Gemini 2.5 Flash、DeepSeek V3.2を单一プラットフォーム에서利用でき、タスクに応じて最適なモデルを選択できます。

よくあるエラーと対処法

エラー1：WebSocket切断と再接続の無限ループ

# ❌ 误った実装例：再接続ロジックなし
async def listen(self):
    async for message in self.websocket:  # 切断時に例外発生
        process(message)


✅ 正しい実装：指数バックオフ付き再接続

async def auto_reconnect(self):
    reconnect_delay = 1
    max_delay = 60
    
    while self.running:
        try:
            await self.connect()
            await self.listen()
        except websockets.exceptions.ConnectionClosed as e:
            logger.warning(f"切断: {e.code} {e.reason}")
            await asyncio.sleep(reconnect_delay)
            reconnect_delay = min(reconnect_delay * 2, max_delay)
            # 再接続試行...

エラー2：API Key認証失敗（401 Unauthorized）

# ❌ 误り：Key形式不正确
headers = {
    "Authorization": "YOUR_HOLYSHEEP_API_KEY"  # Bearer なし
}


✅ 正しい：Bear erプレフィックス必须

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


確認ポイント：


1. API Keyが"sk-hs-"から始まることを確認


2. base_urlが"https://api.holysheep.ai/v1"であることを確認


3. api.openai.com や api.anthropic.com を使用していないことを確認

エラー3：レート制限（429 Too Many Requests）

# ❌ 误り：制限なくリクエスト送信
async def analyze_trades(trades):
    for trade in trades:
        await analyzer.analyze(trade)  # 1秒間に数百リクエスト


✅ 正しい：Semaphoreで同時リクエスト制御

import asyncio

class RateLimitedAnalyzer:
    def __init__(self, max_concurrent: int = 5, requests_per_second: int = 10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.rate_limiter = asyncio.Semaphore(requests_per_second)
        
    async def analyze(self, data):
        async with self.rate_limiter:
            async with self.semaphore:
                result = await self._call_api(data)
                await asyncio.sleep(0.1)  # 追加ディレイ
                return result

エラー4：メモリリーク（過去の購読データ蓄積）

# ❌ 误り：データ蓄積のみで清理なし
self.recent_trades[symbol].append(trade_data)

→ 长时间运行时内存持续增长



✅ 正しい：バッファサイズ制限

MAX_BUFFER = 100
if symbol not in self.recent_trades:
    self.recent_trades[symbol] = []
    
self.recent_trades[symbol].append(trade_data)


バッファサイズ超過時に古いデータを削除

if len(self.recent_trades[symbol]) > MAX_BUFFER:
    self.recent_trades[symbol] = self.recent_trades[symbol][-MAX_BUFFER:]
    

或いはcollections.dequeを使用（自動サイズ制限）

from collections import deque
self.recent_trades = defaultdict(lambda: deque(maxlen=100))

エラー5：板情報のデータ不整合

# ❌ 误り：snapshotとupdateの混同

Bybitではorderbook.50はsnapshot、orderbook.200はupdateを返す


これを混同するとデータが不整合



✅ 正しい：数据类型別の处理

async def _handle_orderbook(self, data: Dict):
    topic = data.get("topic", "")
    orderbook_type = data.get("type")  # "snapshot" or "delta"
    
    if "orderbook.50" in topic:
        # 全量snapshot：完全置换
        self.orderbooks[symbol] = {
            "bids": {p: q for p, q in data["data"]["b"]},
            "asks": {p: q for p, q in data["data"]["a"]}
        }
    elif "orderbook.200" in topic:
        # 差分update：部分更新
        if symbol in self.orderbooks:
            for price, qty in data["data"]["b"]:
                if float(qty) == 0:
                    self.orderbooks[symbol]["bids"].pop(price, None)
                else:
                    self.orderbooks[symbol]["bids"][price] = qty
            # asksも同様の処理

実装チェックリスト