Setting Up Real-Time WebSockets Telemetry Pipeline for Bybit Spot

{
  "topic": "orderbook.50.BTCUSDT",
  "type": "snapshot",
  "ts": 1683724800123,
  "data": {
    "s": "BTCUSDT",
    "b": [
      ["27450.50", "1.245"],
      ["27450.00", "5.102"]
    ],
    "a": [
      ["27451.00", "0.550"],
      ["27451.50", "2.114"]
    ],
    "u": 1054923,
    "seq": 45032192
  }
}

pip install websockets asyncio

import sys
import json
import asyncio
import logging
import websockets

# Configure diagnostic logging output
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger("BybitTelemetry")

class BybitSpotTelemetryPipeline:
    def __init__(self, target_symbol="BTCUSDT"):
        self.uri = "wss://stream.bybit.com/v5/public/spot"
        self.symbol = target_symbol.upper()
        self.is_active = True
        self.connection_backoff = 1
        
        # Bounded in-memory queue to prevent RAM overflow under heavy market load
        self.data_stream_queue = asyncio.Queue(maxsize=10000)

    async def launch_pipeline(self):
        """Launches parallel non-blocking tasks for networking and message parsing."""
        logger.info(f"Initializing telemetry framework for symbol: {self.symbol}")
        
        await asyncio.gather(
            self._network_socket_manager(),
            self._telemetry_data_processor()
        )

    async def _network_socket_manager(self):
        """Maintains persistent WebSocket connection and handles unexpected disconnects."""
        while self.is_active:
            try:
                logger.info(f"Connecting to outbound cluster: {self.uri}")
                async with websockets.connect(self.uri, ping_interval=None) as ws:
                    self.connection_backoff = 1  # Reset exponential backoff on successful connect
                    
                    # Spawn continuous background ping task
                    heartbeat_task = asyncio.create_task(self._send_heartbeat_ping(ws))
                    
                    # Define topic subscription payload
                    subscription_payload = {
                        "op": "subscribe",
                        "args": [
                            f"orderbook.50.{self.symbol}",
                            f"publicTrade.{self.symbol}"
                        ]
                    }
                    await ws.send(json.dumps(subscription_payload))
                    logger.info("Subscription arguments submitted successfully.")

                    # Continuous network socket read loop
                    async for raw_message in ws:
                        if self.data_stream_queue.full():
                            logger.warning("Queue buffer full! Discarding oldest telemetry frame.")
                            self.data_stream_queue.get_nowait()
                        await self.data_stream_queue.put(raw_message)

                    # Cancel heartbeat task when connection closes
                    heartbeat_task.cancel()

            except (websockets.exceptions.ConnectionClosed, Exception) as error:
                logger.error(f"Network anomaly encountered: {str(error)}")
                logger.info(f"Re-establishing bridge connection in {self.connection_backoff}s...")
                await asyncio.sleep(self.connection_backoff)
                self.connection_backoff = min(self.connection_backoff * 2, 60)

    async def _send_heartbeat_ping(self, websocket_client):
        """Transmits periodic ping requests to keep connection active."""
        try:
            while True:
                await asyncio.sleep(20)  # Bybit requires ping every 20-30 seconds
                ping_frame = {"op": "ping"}
                await websocket_client.send(json.dumps(ping_frame))
                logger.debug("Keep-alive heartbeat sent.")
        except asyncio.CancelledError:
            pass

    async def _telemetry_data_processor(self):
        """Asynchronously extracts and processes incoming JSON payload strings."""
        while self.is_active:
            raw_data = await self.data_stream_queue.get()
            try:
                parsed_json = json.loads(raw_data)
                
                # Check for operational responses (e.g. subscription confirmations)
                if "op" in parsed_json and parsed_json.get("success"):
                    logger.info(f"Operation confirmed: {parsed_json.get('ret_msg')}")
                    self.data_stream_queue.task_done()
                    continue
                
                # Route incoming payloads to specific data handlers
                topic = parsed_json.get("topic", "")
                if "orderbook" in topic:
                    self._handle_order_book_update(parsed_json)
                elif "publicTrade" in topic:
                    self._handle_trade_update(parsed_json)

            except json.JSONDecodeError:
                logger.error("Malformed frame detected. Could not decode JSON string.")
            finally:
                self.data_stream_queue.task_done()

    def _handle_order_book_update(self, payload):
        """Parses and logs top bid/ask levels from order book updates."""
        data_block = payload.get("data", {})
        sequence_id = data_block.get("seq", 0)
        top_bid = data_block.get("b", [["0.0", "0.0"]])[0]
        top_ask = data_block.get("a", [["0.0", "0.0"]])[0]
        
        logger.info(
            f"[ORDERBOOK] Seq: {sequence_id} | "
            f"Best Bid: {top_bid[0]} (Vol: {top_bid[1]}) | "
            f"Best Ask: {top_ask[0]} (Vol: {top_ask[1]})"
        )

    def _handle_trade_update(self, payload):
        """Parses and logs individual market order execution records."""
        trades_list = payload.get("data", [])
        for individual_trade in trades_list:
            logger.info(
                f"[TRADE] Price: {individual_trade.get('p')} | "
                f"Volume: {individual_trade.get('v')} | "
                f"Side: {individual_trade.get('S')}"
            )

if __name__ == "__main__":
    pipeline = BybitSpotTelemetryPipeline(target_symbol="BTCUSDT")
    try:
        asyncio.run(pipeline.launch_pipeline())
    except KeyboardInterrupt:
        logger.info("Pipeline execution suspended by developer.")

import random

def calculate_backoff_delay(attempt: int, base_delay: float = 1.0, max_delay: float = 60.0) -> float:
    """Calculates exponential backoff delay with random jitter for robust auto-reconnection."""
    jitter = random.uniform(0.1, 0.5)
    exponential_delay = base_delay * (2 ** attempt)
    return min(exponential_delay + jitter, max_delay)