Bybit WebSocketをPythonで購読する完全ガイド
ticker・kline・orderbook・publicTradeと
asyncio自作クライアント

# 本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
import time

from pybit.unified_trading import WebSocket

# channel_type は購読するcategoryに対応する
#   linear / inverse / spot / option / private
ws = WebSocket(testnet=False, channel_type="linear")


def on_ticker(msg):
    d = msg["data"]
    # linearのtickerは差分配信。存在しないキーがあるのでgetで受ける（後述）
    print(f"{msg['ts']} {d['symbol']} last={d.get('lastPrice')} "
          f"bid={d.get('bid1Price')} ask={d.get('ask1Price')}")


ws.ticker_stream(symbol="BTCUSDT", callback=on_ticker)

# pybitは受信を別スレッドで回すため、メインスレッドは生かしておく必要がある
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    ws.exit()

"""ticker_state.py — tickerのsnapshot/deltaをマージして常に完全な状態を保つ
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import threading
import time
from dataclasses import dataclass, field

from pybit.unified_trading import WebSocket


@dataclass
class TickerStore:
    """symbol -> 最新の完全なtickerフィールド。deltaをマージして保持する。"""

    _state: dict[str, dict] = field(default_factory=dict)
    _lock: threading.Lock = field(default_factory=threading.Lock)
    _updated_at: dict[str, float] = field(default_factory=dict)

    def on_message(self, msg: dict) -> None:
        data = msg.get("data") or {}
        symbol = data.get("symbol")
        if not symbol:
            return

        with self._lock:
            if msg.get("type") == "snapshot":
                # snapshotは全フィールドが揃っているので置き換える
                self._state[symbol] = dict(data)
            else:
                # deltaは変化したフィールドのみ。既存状態に上書きする
                self._state.setdefault(symbol, {}).update(data)
            self._updated_at[symbol] = time.time()

    def get(self, symbol: str) -> dict | None:
        with self._lock:
            s = self._state.get(symbol)
            return dict(s) if s else None

    def price(self, symbol: str) -> float | None:
        s = self.get(symbol)
        return float(s["lastPrice"]) if s and s.get("lastPrice") else None

    def age(self, symbol: str) -> float:
        """最後に更新されてからの経過秒。断線検知に使う。"""
        with self._lock:
            t = self._updated_at.get(symbol)
        return float("inf") if t is None else time.time() - t


if __name__ == "__main__":
    store = TickerStore()
    ws = WebSocket(testnet=False, channel_type="linear")

    for sym in ("BTCUSDT", "ETHUSDT", "SOLUSDT"):
        ws.ticker_stream(symbol=sym, callback=store.on_message)

    try:
        while True:
            time.sleep(2)
            for sym in ("BTCUSDT", "ETHUSDT", "SOLUSDT"):
                s = store.get(sym)
                if not s:
                    continue
                # マージ済みなので、どのフィールドも安全に読める
                print(f"{sym:<9} last={s['lastPrice']:>12} "
                      f"funding={s.get('fundingRate', '-'):>10} "
                      f"OI={s.get('openInterest', '-'):>14} "
                      f"({store.age(sym):.1f}秒前)")
            print("-" * 70)
    except KeyboardInterrupt:
        ws.exit()

"""kline_confirmed.py — 確定足のみを処理してシグナル判定に渡す
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import time
from collections import deque
from datetime import datetime, timezone

from pybit.unified_trading import WebSocket

JST = timezone.utc   # 表示をJSTにする場合は timezone(timedelta(hours=9))


class BarAggregator:
    """確定足だけをdequeに溜め、確定のたびにコールバックを呼ぶ。"""

    def __init__(self, maxlen: int = 500):
        self.bars: deque[dict] = deque(maxlen=maxlen)
        self._last_start: int | None = None

    def on_message(self, msg: dict) -> None:
        for k in msg.get("data", []):
            if not k.get("confirm"):
                continue                      # 未確定足は無視

            start = int(k["start"])
            if start == self._last_start:
                continue                      # 同じ足の重複配信を弾く
            self._last_start = start

            bar = {
                "start": start,
                "time": datetime.fromtimestamp(start / 1000, JST),
                "open": float(k["open"]),
                "high": float(k["high"]),
                "low": float(k["low"]),
                "close": float(k["close"]),
                "volume": float(k["volume"]),
                "turnover": float(k["turnover"]),
            }
            self.bars.append(bar)
            self.on_bar_closed(bar)

    def on_bar_closed(self, bar: dict) -> None:
        """ここに戦略のロジックを書く。1本の確定足につき必ず1回だけ呼ばれる。"""
        closes = [b["close"] for b in self.bars]
        if len(closes) < 20:
            print(f"{bar['time']:%H:%M} close={bar['close']} (ウォームアップ中 "
                  f"{len(closes)}/20)")
            return

        sma20 = sum(closes[-20:]) / 20
        signal = "LONG" if bar["close"] > sma20 else "FLAT"
        print(f"{bar['time']:%H:%M} close={bar['close']:>10.2f} "
              f"sma20={sma20:>10.2f} -> {signal}")


if __name__ == "__main__":
    agg = BarAggregator()
    ws = WebSocket(testnet=False, channel_type="linear")

    # interval: 1,3,5,15,30,60,120,240,360,720,D,W,M
    ws.kline_stream(interval=1, symbol="BTCUSDT", callback=agg.on_message)

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        ws.exit()

"""tick_bars.py — publicTradeから任意秒数のバーと出来高デルタを生成する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import threading
import time
from dataclasses import dataclass, field

from pybit.unified_trading import WebSocket


@dataclass
class Bar:
    start_ms: int
    open: float
    high: float
    low: float
    close: float
    buy_vol: float = 0.0     # テイカー買いの出来高
    sell_vol: float = 0.0    # テイカー売りの出来高
    trades: int = 0

    @property
    def volume(self) -> float:
        return self.buy_vol + self.sell_vol

    @property
    def delta(self) -> float:
        """出来高デルタ。プラスなら買いが攻めている。"""
        return self.buy_vol - self.sell_vol

    @property
    def delta_ratio(self) -> float:
        return self.delta / self.volume if self.volume else 0.0


class TimeBarBuilder:
    """指定秒数のバーをティックから組み立てる。"""

    def __init__(self, seconds: int = 15, on_close=None):
        self.window_ms = seconds * 1000
        self.on_close = on_close
        self.current: Bar | None = None
        self._seen: set[str] = set()          # 約定IDで重複排除
        self._lock = threading.Lock()

    def on_message(self, msg: dict) -> None:
        with self._lock:
            for t in msg.get("data", []):
                self._add(t)

    def _add(self, t: dict) -> None:
        trade_id = t.get("i")
        if trade_id:
            if trade_id in self._seen:
                return                        # 再接続直後の重複配信に備える
            self._seen.add(trade_id)
            if len(self._seen) > 20_000:
                self._seen.clear()

        ts, price, qty = int(t["T"]), float(t["p"]), float(t["v"])
        bucket = ts - (ts % self.window_ms)

        if self.current is None or bucket != self.current.start_ms:
            if self.current is not None and self.on_close:
                self.on_close(self.current)
            self.current = Bar(start_ms=bucket, open=price, high=price,
                               low=price, close=price)

        b = self.current
        b.high = max(b.high, price)
        b.low = min(b.low, price)
        b.close = price
        b.trades += 1
        if t["S"] == "Buy":                   # Sはテイカー側の方向
            b.buy_vol += qty
        else:
            b.sell_vol += qty


def print_bar(b: Bar) -> None:
    print(f"{time.strftime('%H:%M:%S', time.localtime(b.start_ms / 1000))} "
          f"O={b.open:>10.2f} H={b.high:>10.2f} L={b.low:>10.2f} C={b.close:>10.2f} "
          f"vol={b.volume:>8.3f} delta={b.delta:>+8.3f} "
          f"({b.delta_ratio:>+6.1%}) trades={b.trades:>4d}")


if __name__ == "__main__":
    builder = TimeBarBuilder(seconds=15, on_close=print_bar)
    ws = WebSocket(testnet=False, channel_type="linear")
    ws.trade_stream(symbol="BTCUSDT", callback=builder.on_message)

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        ws.exit()

"""ws_client.py — asyncio + websockets によるBybit public WSクライアント
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  pip install websockets orjson
"""
from __future__ import annotations

import asyncio
import contextlib
import random
import time
from collections.abc import Awaitable, Callable

import orjson
import websockets

PUBLIC_URL = "wss://stream.bybit.com/v5/public/{channel}"
TESTNET_URL = "wss://stream-testnet.bybit.com/v5/public/{channel}"

PING_INTERVAL = 20.0        # 公式推奨は20秒
Handler = Callable[[dict], Awaitable[None] | None]


class BybitPublicWS:
    def __init__(self, channel: str = "linear", testnet: bool = False):
        base = TESTNET_URL if testnet else PUBLIC_URL
        self.url = base.format(channel=channel)
        self._handlers: dict[str, list[Handler]] = {}
        self._topics: set[str] = set()
        self._ws: websockets.WebSocketClientProtocol | None = None
        self.last_msg_at: float = 0.0
        self.reconnects: int = 0

    # ---- 購読登録（接続前でも後でも呼べる） ----------------------------
    def subscribe(self, topic: str, handler: Handler) -> None:
        self._topics.add(topic)
        self._handlers.setdefault(topic, []).append(handler)

    async def _send_subscribe(self, topics: list[str]) -> None:
        if not topics or self._ws is None:
            return
        # 現物は1リクエスト10argsまで。安全側に倒して10件ずつ送る
        for i in range(0, len(topics), 10):
            chunk = topics[i:i + 10]
            await self._ws.send(orjson.dumps({"op": "subscribe", "args": chunk}))
            await asyncio.sleep(0.05)

    # ---- ping ---------------------------------------------------------
    async def _ping_loop(self) -> None:
        while True:
            await asyncio.sleep(PING_INTERVAL)
            if self._ws is None:
                return
            with contextlib.suppress(Exception):
                await self._ws.send(orjson.dumps({"op": "ping"}))

    # ---- 受信 ---------------------------------------------------------
    async def _recv_loop(self) -> None:
        assert self._ws is not None
        async for raw in self._ws:
            self.last_msg_at = time.time()
            msg = orjson.loads(raw)

            # 制御メッセージ（pong / subscribe結果）
            if "op" in msg:
                if msg["op"] == "subscribe" and not msg.get("success", True):
                    print(f"[ws] 購読失敗: {msg}")
                continue

            topic = msg.get("topic")
            if not topic:
                continue
            for h in self._handlers.get(topic, []):
                result = h(msg)
                if asyncio.iscoroutine(result):
                    await result

    # ---- 本体（自動再接続つき） -----------------------------------------
    async def run(self) -> None:
        backoff = 1.0
        while True:
            try:
                async with websockets.connect(
                    self.url,
                    ping_interval=None,        # Bybit独自のping/pongを使うので無効化
                    max_queue=2048,
                    open_timeout=10,
                    close_timeout=5,
                ) as ws:
                    self._ws = ws
                    self.last_msg_at = time.time()
                    print(f"[ws] 接続 {self.url}")

                    await self._send_subscribe(sorted(self._topics))
                    ping_task = asyncio.create_task(self._ping_loop())
                    try:
                        await self._recv_loop()
                    finally:
                        ping_task.cancel()
                        with contextlib.suppress(asyncio.CancelledError):
                            await ping_task

                backoff = 1.0                  # 正常終了したらバックオフをリセット

            except asyncio.CancelledError:
                raise
            except Exception as e:
                self.reconnects += 1
                # 指数バックオフ + ジッタ。上限30秒。
                # これを入れないと5分500接続の上限に即到達する
                wait = min(backoff, 30.0) * (0.5 + random.random())
                print(f"[ws] 切断 ({type(e).__name__}: {e}) -> {wait:.1f}秒後に再接続 "
                      f"(通算{self.reconnects}回)")
                await asyncio.sleep(wait)
                backoff = min(backoff * 2, 30.0)
            finally:
                self._ws = None


# ---- 使用例 -----------------------------------------------------------
async def main() -> None:
    ws = BybitPublicWS(channel="linear")

    async def on_trade(msg: dict) -> None:
        for t in msg["data"]:
            print(f"trade {t['s']} {t['S']:<4} {t['p']} x {t['v']}")

    def on_ticker(msg: dict) -> None:          # 同期関数も渡せる
        d = msg["data"]
        if "lastPrice" in d:
            print(f"tick  {d['symbol']} {d['lastPrice']}")

    ws.subscribe("publicTrade.BTCUSDT", on_trade)
    ws.subscribe("tickers.BTCUSDT", on_ticker)

    await ws.run()


if __name__ == "__main__":
    with contextlib.suppress(KeyboardInterrupt):
        asyncio.run(main())

"""multi_symbol.py — 数百銘柄を複数接続に分散して購読する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import asyncio

from pybit.unified_trading import HTTP

from ws_client import BybitPublicWS

MAX_TOPICS_PER_CONN = 100      # 1接続あたりの購読数（控えめに設定する）


def liquid_symbols(min_turnover: float = 20_000_000, limit: int = 200) -> list[str]:
    """24h売買代金でフィルタしたUSDT無期限の銘柄リストを返す。"""
    http = HTTP(testnet=False)
    rows = http.get_tickers(category="linear")["result"]["list"]
    rows = [r for r in rows
            if r["symbol"].endswith("USDT") and float(r["turnover24h"]) >= min_turnover]
    rows.sort(key=lambda r: float(r["turnover24h"]), reverse=True)
    return [r["symbol"] for r in rows[:limit]]


class MultiSymbolFeed:
    """トピックを複数接続に分割して購読する。"""

    def __init__(self, channel: str = "linear"):
        self.channel = channel
        self.clients: list[BybitPublicWS] = []
        self.prices: dict[str, float] = {}

    def _on_ticker(self, msg: dict) -> None:
        d = msg["data"]
        if "lastPrice" in d:                    # deltaでlastPriceが来た時だけ更新
            self.prices[d["symbol"]] = float(d["lastPrice"])

    async def run(self, symbols: list[str]) -> None:
        topics = [f"tickers.{s}" for s in symbols]

        for i in range(0, len(topics), MAX_TOPICS_PER_CONN):
            chunk = topics[i:i + MAX_TOPICS_PER_CONN]
            client = BybitPublicWS(channel=self.channel)
            for t in chunk:
                client.subscribe(t, self._on_ticker)
            self.clients.append(client)

        print(f"{len(symbols)}銘柄を{len(self.clients)}接続で購読します")

        # 接続の確立を少しずつずらす（5分500接続の制限に配慮）
        tasks = []
        for c in self.clients:
            tasks.append(asyncio.create_task(c.run()))
            await asyncio.sleep(0.3)

        tasks.append(asyncio.create_task(self._report()))
        await asyncio.gather(*tasks)

    async def _report(self) -> None:
        while True:
            await asyncio.sleep(10)
            recon = sum(c.reconnects for c in self.clients)
            print(f"[状況] 価格取得済み {len(self.prices)}銘柄 / 再接続通算 {recon}回")


if __name__ == "__main__":
    syms = liquid_symbols()
    asyncio.run(MultiSymbolFeed().run(syms))

"""watchdog.py — 受信停止（沈黙）の検知と、RESTによる欠損足の穴埋め
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import asyncio
import time

from pybit.unified_trading import HTTP

from ws_client import BybitPublicWS


class GapFiller:
    """WebSocketが止まっていた区間の確定足をRESTで補完する。"""

    def __init__(self, symbol: str, interval: str = "1", category: str = "linear"):
        self.http = HTTP(testnet=False)
        self.symbol = symbol
        self.interval = interval
        self.category = category
        self.last_bar_start: int | None = None

    def note_bar(self, start_ms: int) -> None:
        self.last_bar_start = start_ms

    def fill(self) -> list[list[str]]:
        """last_bar_start以降の足をRESTで取得する。"""
        if self.last_bar_start is None:
            return []
        res = self.http.get_kline(
            category=self.category, symbol=self.symbol,
            interval=self.interval,
            start=self.last_bar_start + 1, limit=1000,
        )
        if res["retCode"] != 0:
            print(f"[gapfill] 失敗: {res['retMsg']}")
            return []
        # V5のklineは新しい順で返るので、時系列に直す
        rows = list(reversed(res["result"]["list"]))
        if rows:
            print(f"[gapfill] {len(rows)}本を補完しました")
            self.last_bar_start = int(rows[-1][0])
        return rows


async def watchdog(ws: BybitPublicWS, filler: GapFiller,
                  silence_sec: float = 60.0, check_sec: float = 10.0) -> None:
    """一定時間データが来なければ警告し、復帰時にRESTで穴埋めする。"""
    was_silent = False

    while True:
        await asyncio.sleep(check_sec)

        if ws.last_msg_at == 0:
            continue
        silence = time.time() - ws.last_msg_at

        if silence > silence_sec:
            if not was_silent:
                print(f"[警告] {silence:.0f}秒間データが来ていません。"
                      f"エントリーを停止してください")
                was_silent = True
            # ここで記事24のサーキットブレーカーをREDUCE_ONLYに落とすのが実務的
        elif was_silent:
            print(f"[復帰] データ受信を再開しました（沈黙 {silence:.0f}秒）")
            was_silent = False
            filler.fill()          # 欠損区間を埋める


async def main() -> None:
    ws = BybitPublicWS(channel="linear")
    filler = GapFiller("BTCUSDT", interval="1")

    def on_kline(msg: dict) -> None:
        for k in msg["data"]:
            if k.get("confirm"):
                filler.note_bar(int(k["start"]))
                print(f"bar close {k['start']} {k['close']}")

    ws.subscribe("kline.1.BTCUSDT", on_kline)

    await asyncio.gather(ws.run(), watchdog(ws, filler))


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

"""latency.py — WebSocket受信遅延の分布を測る（VPSリージョン選定の判断材料）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

注意: ローカル時計がズレていると測定値もズレる。
      先に記事01の check_clock_skew() でズレを確認しておくこと。
"""
from __future__ import annotations

import asyncio
import statistics
import time

from ws_client import BybitPublicWS


class LatencyMonitor:
    def __init__(self, window: int = 2000):
        self.samples: list[float] = []
        self.window = window

    def on_message(self, msg: dict) -> None:
        # msg["ts"] はBybit側の送信時刻(ms)
        sent = msg.get("ts")
        if not sent:
            return
        delay_ms = time.time() * 1000 - int(sent)
        self.samples.append(delay_ms)
        if len(self.samples) > self.window:
            del self.samples[: len(self.samples) - self.window]

    def report(self) -> str:
        if len(self.samples) < 10:
            return "サンプル収集中..."
        s = sorted(self.samples)
        p = lambda q: s[min(int(len(s) * q), len(s) - 1)]
        return (f"n={len(s):5d}  p50={statistics.median(s):7.1f}ms  "
                f"p90={p(0.90):7.1f}ms  p99={p(0.99):7.1f}ms  max={s[-1]:7.1f}ms")


async def main() -> None:
    mon = LatencyMonitor()
    ws = BybitPublicWS(channel="linear")
    ws.subscribe("publicTrade.BTCUSDT", mon.on_message)
    ws.subscribe("tickers.BTCUSDT", mon.on_message)

    async def printer() -> None:
        while True:
            await asyncio.sleep(10)
            print(f"[遅延] {mon.report()}")

    await asyncio.gather(ws.run(), printer())


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

"""queue_pipeline.py — 受信と処理を分離し、遅れたら古いデータを捨てる
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import asyncio
import time

from ws_client import BybitPublicWS

QUEUE_MAX = 10_000


class Pipeline:
    def __init__(self):
        self.q: asyncio.Queue[dict] = asyncio.Queue(maxsize=QUEUE_MAX)
        self.dropped = 0
        self.processed = 0

    def enqueue(self, msg: dict) -> None:
        """コールバックはこれだけ。絶対にブロックしない。"""
        try:
            self.q.put_nowait(msg)
        except asyncio.QueueFull:
            # 溢れたら「古いものを捨てて新しいものを入れる」。
            # 相場データは鮮度が命なので、古いデータを守る意味がない。
            self.dropped += 1
            try:
                self.q.get_nowait()
                self.q.put_nowait(msg)
            except (asyncio.QueueEmpty, asyncio.QueueFull):
                pass

    async def worker(self, name: str) -> None:
        """重い処理はここ。複数起動して並列化してもよい。"""
        while True:
            msg = await self.q.get()
            try:
                await self._handle(msg)
                self.processed += 1
            except Exception as e:
                print(f"[{name}] 処理エラー: {type(e).__name__}: {e}")
            finally:
                self.q.task_done()

    async def _handle(self, msg: dict) -> None:
        # ここでDB書き込み・指標計算・発注判断などを行う
        await asyncio.sleep(0)          # 実処理の代わり

    async def stats(self) -> None:
        while True:
            await asyncio.sleep(10)
            print(f"[pipeline] 処理={self.processed} 破棄={self.dropped} "
                  f"待ち={self.q.qsize()}")
            if self.q.qsize() > QUEUE_MAX * 0.5:
                print("  [警告] キューが溜まっています。処理が追いついていません")


async def main() -> None:
    pipe = Pipeline()
    ws = BybitPublicWS(channel="linear")

    for sym in ("BTCUSDT", "ETHUSDT", "SOLUSDT"):
        ws.subscribe(f"publicTrade.{sym}", pipe.enqueue)
        ws.subscribe(f"tickers.{sym}", pipe.enqueue)

    await asyncio.gather(
        ws.run(),
        pipe.worker("w1"),
        pipe.worker("w2"),
        pipe.stats(),
    )


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