""" Bybit V5 public WebSocket provider — backup for Binance. Streams: - publicTrade.{symbol} → TradeEvent - tickers.{symbol} → QuoteEvent (best bid/ask from tickers) Docs: https://bybit-exchange.github.io/docs/v5/ws/connect No API key needed for public market data. """ from __future__ import annotations import asyncio import json import logging from datetime import datetime, timezone from typing import AsyncIterator import websockets from websockets.exceptions import ConnectionClosed from app.config import settings from app.domain.events import ( Event, QuoteEvent, TradeEvent, ) from app.providers import MarketDataProvider logger = logging.getLogger(__name__) def _ms_to_dt(ms: int | float | str | None) -> datetime | None: """Convert millisecond epoch to UTC datetime.""" if ms is None: return None try: return datetime.fromtimestamp(int(ms) / 1000.0, tz=timezone.utc) except (ValueError, TypeError, OSError): return None class BybitProvider(MarketDataProvider): """ Bybit V5 public WebSocket (spot market). Connects to the spot public channel and subscribes to publicTrade + tickers for each symbol. """ name = "bybit" def __init__(self) -> None: self._ws: websockets.WebSocketClientProtocol | None = None self._symbols: list[str] = [] self._connected = False self._reconnect_count = 0 self._base_url = settings.bybit_ws_url async def connect(self) -> None: """Establish WebSocket connection.""" logger.info("bybit.connecting", extra={"url": self._base_url}) self._ws = await websockets.connect( self._base_url, ping_interval=20, ping_timeout=10, close_timeout=5, ) self._connected = True logger.info("bybit.connected") async def subscribe(self, symbols: list[str]) -> None: """Subscribe to publicTrade + tickers for each symbol.""" if not self._ws: raise RuntimeError("Not connected. Call connect() first.") self._symbols = [s.upper() for s in symbols] args = [] for sym in self._symbols: args.append(f"publicTrade.{sym}") args.append(f"tickers.{sym}") subscribe_msg = { "op": "subscribe", "args": args, } await self._ws.send(json.dumps(subscribe_msg)) logger.info( "bybit.subscribed", extra={"symbols": self._symbols, "channels": len(args)}, ) async def stream(self) -> AsyncIterator[Event]: """Yield domain events. Handles reconnect automatically.""" backoff = settings.reconnect_base_delay while True: try: if not self._connected or not self._ws: await self._reconnect(backoff) try: raw = await asyncio.wait_for( self._ws.recv(), # type: ignore timeout=settings.heartbeat_timeout, ) except asyncio.TimeoutError: logger.warning( "bybit.heartbeat_timeout", extra={"timeout": settings.heartbeat_timeout}, ) self._connected = False continue # Reset backoff on successful message backoff = settings.reconnect_base_delay data = json.loads(raw) # Handle pong (Bybit sends {"op":"pong",...}) if data.get("op") in ("pong", "subscribe"): if data.get("success") is False: logger.warning("bybit.subscribe_failed", extra={"msg": data}) continue event = self._parse(data) if event: yield event except ConnectionClosed as e: logger.warning( "bybit.connection_closed", extra={"code": e.code, "reason": str(e.reason)}, ) self._connected = False backoff = min(backoff * 2, settings.reconnect_max_delay) except Exception as e: logger.error("bybit.stream_error", extra={"error": str(e)}) self._connected = False backoff = min(backoff * 2, settings.reconnect_max_delay) async def _reconnect(self, delay: float) -> None: """Reconnect with delay, then resubscribe.""" self._reconnect_count += 1 logger.info( "bybit.reconnecting", extra={"delay": delay, "attempt": self._reconnect_count}, ) await asyncio.sleep(delay) try: if self._ws: await self._ws.close() except Exception: pass await self.connect() if self._symbols: await self.subscribe(self._symbols) def _parse(self, data: dict) -> Event | None: """Parse raw Bybit V5 message into domain events.""" topic = data.get("topic", "") event_data = data.get("data") if not topic or event_data is None: return None if topic.startswith("publicTrade."): return self._parse_trades(event_data) elif topic.startswith("tickers."): return self._parse_ticker(event_data) return None def _parse_trades(self, data: list | dict) -> Event | None: """ Bybit publicTrade payload (V5): {"data": [{"s":"BTCUSDT","S":"Buy","v":"0.001","p":"70000.5","T":1672515782136,"i":"..."}]} We take the last trade in the batch. """ if isinstance(data, list): if not data: return None trade = data[-1] # latest in batch else: trade = data return TradeEvent( provider=self.name, symbol=trade.get("s", "").upper(), price=float(trade.get("p", 0)), size=float(trade.get("v", 0)), ts_exchange=_ms_to_dt(trade.get("T")), side=trade.get("S", "").lower() if trade.get("S") else None, trade_id=str(trade.get("i", "")), ) def _parse_ticker(self, data: dict) -> QuoteEvent | None: """ Bybit tickers (V5 spot): {"data": {"symbol":"BTCUSDT","bid1Price":"70000.5","bid1Size":"1.5", "ask1Price":"70001.0","ask1Size":"2.0",...}} """ if isinstance(data, list): data = data[0] if data else {} bid = data.get("bid1Price") or data.get("bidPrice") ask = data.get("ask1Price") or data.get("askPrice") bid_size = data.get("bid1Size") or data.get("bidSize") ask_size = data.get("ask1Size") or data.get("askSize") if not bid or not ask: return None return QuoteEvent( provider=self.name, symbol=data.get("symbol", "").upper(), bid=float(bid), ask=float(ask), bid_size=float(bid_size or 0), ask_size=float(ask_size or 0), ts_exchange=_ms_to_dt(data.get("ts")), ) async def close(self) -> None: """Close the WebSocket connection.""" self._connected = False if self._ws: try: await self._ws.close() except Exception: pass logger.info( "bybit.closed", extra={"reconnect_count": self._reconnect_count}, )