Bybit API WebSocket 实时行情接入：初心者向け完全ガイド

暗号資産の取引において、リアルタイムの市場データ（リアルタイムマーケットデータ）は生命線です。価格が一瞬で変動する市場では、0.1秒の遅延が大きな損失をもたらす可能性があります。本記事では、Bybit APIのWebSocket接続を使用して、リアルタイム行情データを取得する方法を完全に初心者向けに解説します。プログラミングの経験がない方も、このガイド读完後には自分のパソコンでリアルタイムのBTC/USDT価格を取得できるようになります。

💡 筆者の実践経験：私は以前、レート制限（Rate Limit）に起因する接続切断に何度も苦しみました。本記事のエラー対処セクションでは、私の失敗経験から生まれた具体的な解決策を共有します。

WebSocketとは？为什么重要？

традиционная APIリクエスト）では、あなたが「今のBTC価格は？」とサーバーに尋ねると、サーバーが答える形式です。しかし、これは毎秒ごとに同じ質問を何度も送る必要があり、非効率的です。

WebSocketは、サーバーとクライアントの間に持続的な接続通道を開き、サーバーが新しいデータをお届けします你不要.request. BTC価格が变动するたびに、あなたのプログラムに自动的推送されます。これにより：

• レイテンシ削減：每度接続を確立する必要がないため、50ms〜100ms高速化
• サーバー负荷軽減：不必要的リクエストが削減され、API制限に引っかかりにくくなる
• リアルタイム性：市场价格の变动をほぼ延迟なしで取得可能

Bybit WebSocketの全体構成

BybitのWebSocket接続は、用途別に2つの入り口を提供します：

• パブリック通道：取引_SYMBOL.market ticker._symbol.market_depth（板情報）など、認証不要
• プライベート通道：自分の.balance（残高）、.order（注文）など、APIキーが必要

本記事は主にパブリック通道を使用し、リアルタイム行情データの取得に聚焦します。

必要な準備

1. Bybitアカウントの作成

まず、Bybitのアカウントが必要です持っていない場合は、Bybit公式サイトから作成してください。デモ取引（Testnet）もあるので、最初はリスクなく練習できます。

2. Python環境の準備

このガイドではPythonを使用します。Pythonをダウンロードしていない場合は、公式サイトから最新版をインストールしてください。インストール時、「Add Python to PATH」に必ずチェックを入れてください。

3. 必要なライブラリのインストール

コマンドプロンプト（Windows）またはターミナル（Mac/Linux）を開き、以下のコマンドを実行してください：

# WebSocket接続用のライブラリをインストール
pip install websocket-client

データ分析用のライブラリ（任意）
pip install pandas
pip install numpy

📸 スクリーンショットポイント：コマンドプロンプトで上のコマンドを実行すると、「Successfully installed websocket-client」と表示されるはずです。こんな風に緑色の「Successfully」メッセージが出たら成功です。

ステップ1：WebSocket基本接続を実装する

まず、最もシンプルなWebSocket接続を作成しましょう。Bybitのデモ環境（Testnet）に接続します。

import json
import time
from websocket import create_connection

Bybit Testnet WebSocket エンドポイント
WS_URL = "wss://stream-testnet.bybit.com/v5/public/spot"

def connect_websocket():
    """WebSocket接続の基本形"""
    try:
        # 接続を作成
        ws = create_connection(WS_URL)
        print("✅ 接続成功！")
        print(f"接続先: {WS_URL}")
        return ws
    except Exception as e:
        print(f"❌ 接続失敗: {e}")
        return None

接続テスト
ws = connect_websocket()
if ws:
    print("👍 WebSocket接続確立完毕！")
    ws.close()
    print("🔌 接続を切断しました")

このコードを実行すると、以下のような出力が表示されます：

✅ 接続成功！
接続先: wss://stream-testnet.bybit.com/v5/public/spot
👍 WebSocket接続確立完毕！
🔌 接続を切断しました

📸 スクリーンショットポイント：初次接続時、Windows Defenderがポップアップ表示される場合があります。「アクセスを許可する」を選択してください。

ステップ2：BTC/USDTリアルタイム価格を取得する

接続できました！次は、実際の価格データを.subscribe. Bybit WebSocketでは、特定のトピックをサブスクライブする必要があります.

import json
import time
from websocket import create_connection

Bybit Testnet WebSocket エンドポイント
WS_URL = "wss://stream-testnet.bybit.com/v5/public/spot"

def get_btc_price():
    """BTC/USDTリアルタイム価格を取得"""
    
    ws = create_connection(WS_URL)
    
    # サブスクライブリクエストを作成
    subscribe_msg = {
        "op": "subscribe",
        "args": ["tickers.BTCUSDT"]  # BTC/USDTのティッカー情報
    }
    
    # JSONに変換して送信
    ws.send(json.dumps(subscribe_msg))
    print("📡 サブスクリプション要求を送信しました...")
    
    # 最初の5件のメッセージを受信
    for i in range(5):
        result = ws.recv()
        data = json.loads(result)
        print(f"\n--- メッセージ {i+1} ---")
        print(json.dumps(data, indent=2, ensure_ascii=False))
        time.sleep(1)  # 1秒間隔
    
    ws.close()
    print("\n🔌 接続を切断しました")

実行
get_btc_price()

実行结果は以下のようになります：

📡 サブスクリプション要求を送信しました...

--- メッセージ 1 ---
{
  "topic": "tickers.BTCUSDT",
  "type": "snapshot",
  "data": {
    "symbol": "BTCUSDT",
    "lastPrice": "62543.55",
    "highPrice24h": "63120.00",
    "lowPrice24h": "61890.12",
    "volume24h": "12543.21",
    "turnover24h": "782123456.78"
  },
  "ts": 1707234567890,
  "datetime": "2024-02-06T12:34:27.890Z"
}

--- メッセージ 2 ---
{
  "topic": "tickers.BTCUSDT",
  "type": "delta",
  "data": {
    "lastPrice": "62545.12",
    "volume24h": "12544.55"
  },
  "ts": 1707234568890
}

💡 ポイント：「type」が「snapshot」の場合は全データ、「delta」の場合は変動分のみです постоянный.Updatesでは주로「delta」がきます.

ステップ3：板情報（Market Depth）を取得する

取引Botには、板情報（指の注文 목록）も重要です。板情報は、今は买单と卖单の注文一覧を提供します。

import json
import time
from websocket import create_connection

WS_URL = "wss://stream-testnet.bybit.com/v5/public/spot"

def get_order_book():
    """板情報をリアルタイムで取得"""
    
    ws = create_connection(WS_URL)
    
    # 板情報のサブスクリプション（ Depth50 = 板50段階）
    subscribe_msg = {
        "op": "subscribe", 
        "args": ["orderbook.50.BTCUSDT"]
    }
    
    ws.send(json.dumps(subscribe_msg))
    print("📡 板情報のサブスクリプションを開始...")
    
    # 3秒間のデータを取得
    start_time = time.time()
    while time.time() - start_time < 3:
        result = ws.recv()
        data = json.loads(result)
        
        if "data" in data:
            orderbook = data["data"]
            print(f"\nBID（買い）: 価格 {orderbook['b'][0][0]}, 量 {orderbook['b'][0][1]}")
            print(f"ASK（売り）: 価格 {orderbook['a'][0][0]}, 量 {orderbook['a'][0][1]}")
    
    ws.close()

get_order_book()

结果：

📡 板情報のサブスクリプションを開始...

BID（買い）: 価格 62540.12, 量 0.52341
ASK（売り）: 価格 62541.55, 量 0.31289

BID（買い）: 価格 62540.50, 量 0.61423
ASK（売り）: 価格 62541.22, 量 0.29845

📸 スクリーンショットポイント：Bybit取引画面と照らし合わせると、板の见本積みと実際の価格が一致することが确认できます。

ステップ4：複数トピックを同時にSubscribe

実際のBot开发では、複数の通貨ペアを同時に监视する必要があります。以下は、複数のトピックを1つの接続で监视する方法です。

import json
import time
from websocket import create_connection

WS_URL = "wss://stream-testnet.bybit.com/v5/public/spot"

def monitor_multiple_symbols():
    """複数の通貨ペアを同時に監視"""
    
    ws = create_connection(WS_URL)
    
    # 複数のシンボルをサブスクライブ
    symbols = ["BTCUSDT", "ETHUSDT", "SOLUSDT"]
    subscribe_msg = {
        "op": "subscribe",
        "args": [f"tickers.{sym}" for sym in symbols]
    }
    
    ws.send(json.dumps(subscribe_msg))
    print(f"📡 {len(symbols)}個の通貨ペアをサブスクリプション中...")
    
    # 5秒間データを収集
    start_time = time.time()
    prices = {}
    
    while time.time() - start_time < 5:
        result = ws.recv()
        data = json.loads(result)
        
        if "data" in data:
            symbol = data["data"]["symbol"]
            price = data["data"]["lastPrice"]
            prices[symbol] = price
            print(f"{symbol}: ${price}")
    
    print("\n📊 収集した価格データ:")
    for sym, price in prices.items():
        print(f"  {sym}: ${price}")
    
    ws.close()

monitor_multiple_symbols()

ステップ5：エラー處理と再接続机制の実装

長時間運行するBotでは、接続断などのエラーが発生します。以下のコードは、自動的に再接続する仕組みを実装しています。

import json
import time
from websocket import create_connection, WebSocketConnectionClosedException
import threading

WS_URL = "wss://stream-testnet.bybit.com/v5/public/spot"
MAX_RECONNECT_ATTEMPTS = 5
RECONNECT_DELAY = 3  # 秒

class BybitWebSocketClient:
    def __init__(self, url=WS_URL):
        self.url = url
        self.ws = None
        self.running = False
        self.reconnect_count = 0
    
    def connect(self):
        """接続を確立"""
        try:
            self.ws = create_connection(self.url, timeout=10)
            self.reconnect_count = 0
            print("✅ 接続確立")
            return True
        except Exception as e:
            print(f"❌ 接続エラー: {e}")
            return False
    
    def subscribe(self, topics):
        """トピックをサブスクライブ"""
        msg = {"op": "subscribe", "args": topics}
        self.ws.send(json.dumps(msg))
        print(f"📡 サブスクライブ: {topics}")
    
    def receive_messages(self):
        """メッセージ受信用ループ"""
        self.running = True
        
        while self.running:
            try:
                msg = self.ws.recv()
                data = json.loads(msg)
                self.process_message(data)
                
            except WebSocketConnectionClosedException:
                print("⚠️ 接続が切断されました")
                self.handle_disconnect()
                
            except Exception as e:
                print(f"❌ エラー: {e}")
                time.sleep(1)
    
    def process_message(self, data):
        """受信したメッセージを処理"""
        if "topic" in data and "data" in data:
            topic = data["topic"]
            price = data["data"].get("lastPrice", "N/A")
            print(f"[{topic}] 現在価格: ${price}")
    
    def handle_disconnect(self):
        """切断時の処理"""
        if self.reconnect_count < MAX_RECONNECT_ATTEMPTS:
            self.reconnect_count += 1
            print(f"🔄 再接続試行 {self.reconnect_count}/{MAX_RECONNECT_ATTEMPTS}")
            time.sleep(RECONNECT_DELAY)
            if self.connect():
                self.subscribe(["tickers.BTCUSDT"])
        else:
            print("❌ 最大再接続回数に達しました。終了します。")
            self.running = False
    
    def start(self):
        """Botを開始"""
        if self.connect():
            self.subscribe(["tickers.BTCUSDT"])
            self.receive_messages()
    
    def stop(self):
        """Botを停止"""
        self.running = False
        if self.ws:
            self.ws.close()

使用例
if __name__ == "__main__":
    client = BybitWebSocketClient()
    try:
        client.start()
    except KeyboardInterrupt:
        print("\n🛑 停止要求を受け付けました")
        client.stop()

💡 筆者の実践経験：この再接続机制は、私が初めて作ったBotが夜里に切断され、次の日に気がついて6时间分のデータを見逃していた的痛苦経験から必须だと実感しました。今は必ず実装しています。

ステップ6：実践的なBotの例 — 価格アラートBot

ここまでの知识を組み合わせた実践例として、指定価格になったら通知するBotを作成します。

import json
import time
from websocket import create_connection
from datetime import datetime

WS_URL = "wss://stream-testnet.bybit.com/v5/public/spot"

class PriceAlertBot:
    def __init__(self, symbol="BTCUSDT", alert_price=63000, direction="above"):
        self.symbol = symbol
        self.alert_price = float(alert_price)
        self.direction = direction  # "above" または "below"
        self.alert_triggered = False
    
    def start(self):
        """Botを開始"""
        ws = create_connection(WS_URL)
        
        subscribe_msg = {
            "op": "subscribe",
            "args": [f"tickers.{self.symbol}"]
        }
        ws.send(json.dumps(subscribe_msg))
        print(f"🔔 {self.symbol} 价格アラート Bot 起動")
        print(f"   監視条件: {self.direction} ${self.alert_price}")
        print("-" * 40)
        
        while not self.alert_triggered:
            try:
                result = ws.recv()
                data = json.loads(result)
                
                if "data" in data:
                    current_price = float(data["data"]["lastPrice"])
                    self.check_alert(current_price)
                    
            except Exception as e:
                print(f"エラー: {e}")
                break
        
        ws.close()
    
    def check_alert(self, current_price):
        """アラート条件をチェック"""
        now = datetime.now().strftime("%H:%M:%S")
        
        if self.direction == "above" and current_price >= self.alert_price:
            print(f"\n🚨 【アラート！】{now}")
            print(f"   {self.symbol} が ${self.alert_price} を突破！")
            print(f"   現在価格: ${current_price}")
            self.alert_triggered = True
            
        elif self.direction == "below" and current_price <= self.alert_price:
            print(f"\n🚨 【アラート！】{now}")
            print(f"   {self.symbol} が ${self.alert_price} を下落！")
            print(f"   現在価格: ${current_price}")
            self.alert_triggered = True
        else:
            print(f"[{now}] 監視中: ${current_price}", end="\r")

使用例
if __name__ == "__main__":
    # BTCが63000ドルを超えたらアラート
    bot = PriceAlertBot(symbol="BTCUSDT", alert_price=63000, direction="above")
    bot.start()

実行結果（価格がアラート条件に達した場合）：

🔔 BTCUSDT 价格アラート Bot 起動
   監視条件: above $63000
----------------------------------------
[14:30:15] 監視中: $62543.21
[14:30:16] 監視中: $62548.90
[14:30:17] 監視中: $62560.55
[14:30:18] 監視中: $62612.30
...

🚨 【アラート！】14:35:42
   BTCUSDT が $63000 を突破！
   現在価格: $63015.55

Bybit WebSocketの主要なトピック一覧

Bybitでは 다양한数据类型が用意されています。以下は主なトピックです：

トピック名	説明	更新頻度
tickers.{symbol}	最新のティッカー情報（価格、出来高など）	リアルタイム
orderbook.50.{symbol}	板情報（50レベル）	リアルタイム
orderbook.500.{symbol}	板情報（500レベル）	リアルタイム
publicTrade.{symbol}	公共取引履歴	約100ms
kline.1.{symbol}	1分足チャート	約1秒
kline.60.{symbol}	1時間足チャート	約1秒

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

✅ 向いている人

• 暗号資産トレーダー：リアルタイムの市場データを活用した自動取引戦略を構築したい人
• アプリ開発者：暗号資産関連のサービスやダッシュボードを作成したい人
• データ分析に興味がある人：市場データの収集と分析を始めたい人
• Python基础知识がある人：最低限のプログラミング経験がある方

❌ 向いていない人

• プログラミング完全初心者：変数や関数の概念が不明な方は、まずPython基础を勉强してください
• 高频取引（HFT）を目指す人：Bybitの免费WebSocketは毫秒レベルの遅延があり、本番環境のHFTには专用インフラが必要です
• リスクを許容できない人：自動取引には资金loss的风险が伴います

価格とROI

Bybit WebSocket API本身は無料です。ただし、以下のコストを考慮する必要があります：

項目	Bybit公式	HolySheep AI統合
WebSocket接続	無料	無料
APIリクエスト制限	6回/秒（パブリック）	拡張可能
開発サポート	ドキュメントのみ	日本語サポート対応
追加AI機能	ー	sentiment分析など

もしあなたのBot開発でAI分析機能（市場感情分析、價格予測など）が必要であれば、HolySheep AIのAPIを使用することで、GPT-4.1が$8/1Mトークン、Claude Sonnet 4.5が$15/1Mトークンという业界最安水準の価格でAI機能を統合できます。公式汇率（¥7.3/$1）相比、HolySheepなら¥1=$1兑换率は85%节约になり、大量データ处理应用中显著なコスト削减が可能です。

HolySheepを選ぶ理由

WebSocketでリアルタイムデータを取得できたら、次はそのデータを活用した高度な分析が必要ませんか？HolySheep AIなら、こんなことができます：

• 市場感情分析：リアルタイム行情データをAIに投入し、現在の市場心理を数值化
• 价格予測辅助：複数の指標を統合分析し、取引判断をサポート
• 超低コスト：DeepSeek V3.2なら$0.42/1Mトークンで大量処理が可能
• 日本語対応：完全日本語のインターフェースとサポート

💰 コスト比較の實際：1日10万件の行情データ分析和仮定した場合、HolySheepのDeepSeek V3.2なら约$0.042/日。公式汇率で计算すると同等功能が约$0.30/日，因此在HolySheepなら85%节约できます。

よくあるエラーと対処法

エラー1：WebSocketConnectionClosedException — 接続が突然切れる

# ❌ エラーの例
Traceback (most recent call last):
  File "bot.py", line 23, in receive_messages
    msg = self.ws.recv()
  File "...\websocket.py", line 310, in recv
    raise WebSocketConnectionClosedException("Connection is already closed.")

✅ 解決方法：再接続ロジックを追加
import time
from websocket import create_connection, WebSocketConnectionClosedException

def safe_receive(ws, max_retries=3):
    for attempt in range(max_retries):
        try:
            return ws.recv()
        except WebSocketConnectionClosedException:
            print(f"接続切断、{attempt+1}回目再試行...")
            time.sleep(2)
            ws = create_connection(WS_URL)
            ws.send(json.dumps({"op": "subscribe", "args": ["tickers.BTCUSDT"]}))
    raise Exception("最大再試行回数を超過")

エラー2：Subscribe後にデータが来ない

# ❌ よくある原因：サブスクライブ完了を待たずにデータ受信を始める
ws.send(json.dumps({"op": "subscribe", "args": ["tickers.BTCUSDT"]}))
msg = ws.recv()  # ここでハングする可能性がある

✅ 解決方法：ACK応答を待ってから受信開始
ws.send(json.dumps({"op": "subscribe", "args": ["tickers.BTCUSDT"]}))

サブスクライブのACKを確認
ack = ws.recv()
ack_data = json.loads(ack)
if ack_data.get("success"):
    print("✅ サブスクライブ成功！")
    # ここから実際のデータを受信
else:
    print(f"❌ サブスクライブ失敗: {ack_data}")

エラー3：Invalid JSON formatエラー

# ❌ エラーの例
json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

✅ 解決方法：空の応答をスキップ
def safe_json_parse(data):
    try:
        return json.loads(data)
    except json.JSONDecodeError:
        if data == "" or data is None:
            return None  # 空データはスキップ
        print(f"⚠️ JSON解析エラー: {data[:100]}...")
        return None

使用例
while True:
    raw_data = ws.recv()
    parsed_data = safe_json_parse(raw_data)
    if parsed_data:
        process_message(parsed_data)

エラー4：Too Many Requests — レート制限

# ❌ エラーの例
{"ret_code":10002,"ret_msg":"err_ratelimit","result":null}

✅ 解決方法：リクエスト間に延迟を入れる
import time
import random

def throttled_subscribe(ws, topics):
    for topic in topics:
        ws.send(json.dumps({"op": "subscribe", "args": [topic]}))
        # 次のリクエストまで0.2〜0.5秒待機
        delay = 0.2 + random.uniform(0, 0.3)
        time.sleep(delay)
    print(f"✅ {len(topics)}件のトピックをレート制限内でサブスクライブ完了")

エラー5：証明書の問題（SSL/TLS Error）

# ❌ エラーの例
ssl.SSLCertVerificationError: CERTIFICATE_VERIFY_FAILED

✅ 解決方法：SSL検証をスキップ（開発環境のみ）
import ssl

開発/テスト用
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE

ws = create_connection(WS_URL, ssl=ssl_context)
print("⚠️ SSL検証を無効化（開発環境のみ使用）")

本番環境では正しく証明書を設定すること！
ssl_context = ssl.create_default_context()
ws = create_connection(WS_URL, ssl=ssl_context)

次のステップ

このガイド看完後、以下のコンテンツに挑戦してみてください：

• 複数の通貨ペアを監視：同時にBTC、ETH、SOLの価格を監視するBot
• 、板情報 бота：板の歪みを検出してアービトラージ機会を探す
• 、AI統合：HolySheep AIAPIを使って市場感情を分析

リアルタイムデータの取り扱いに慣れてきたら、次は実践的な取引Bot开发に進みましょう。市場は待ってくれません。今すぐHolySheep AIに登録して%、最初のステップを踏み出してください！

---

📚 参考リソース：

• Bybit公式WebSocketドキュメント
• HolySheep AI APIドキュメント
• Bybit公式SDK（GitHub）

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