Handling Bybit API Errors and Systemic Rate-Limits in Python

import httpx
import time
import hmac
import hashlib
import logging
import json
import asyncio

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

class BybitAPIException(Exception):
    """Custom exception raised when Bybit returns a non-zero retCode or HTTP error."""
    def __init__(self, status_code: int, ret_code: int, message: str, response_body: dict = None):
        self.status_code = status_code
        self.ret_code = ret_code
        self.message = message
        self.response_body = response_body
        super().__init__(f"Bybit Exception [HTTP {status_code}] Code {ret_code}: {message}")

class BybitTransportException(Exception):
    """Custom exception raised for network transport disruptions (timeouts, DNS failures)."""
    pass

class ResilientBybitClient:
    """Production-grade asynchronous Bybit V5 API REST Client for Python developers."""
    def __init__(self, api_key: str, api_secret: str, base_url: str = "https://api.bybit.com"):
        self.api_key = api_key
        self.api_secret = api_secret
        self.base_url = base_url
        self.client = httpx.AsyncClient(timeout=httpx.Timeout(5.0, read=10.0))

    def _generate_signature(self, timestamp: str, recv_window: str, payload: str) -> str:
        """Generates HMAC-SHA256 signature required for Bybit V5 private endpoints."""
        param_str = timestamp + self.api_key + recv_window + payload
        return hmac.new(
            self.api_secret.encode("utf-8"),
            param_str.encode("utf-8"),
            hashlib.sha256
        ).hexdigest()

    async def send_request(self, method: str, endpoint: str, params: dict = None) -> dict:
        """Dispatches HTTP request with authentication headers and robust response parsing."""
        url = f"{self.base_url}{endpoint}"
        timestamp = str(int(time.time() * 1000))
        recv_window = "5000"
        
        serialized_payload = ""
        if method == "POST" and params:
            serialized_payload = json.dumps(params)
        elif method == "GET" and params:
            serialized_payload = "&".join([f"{k}={v}" for k, v in sorted(params.items())])

        signature = self._generate_signature(timestamp, recv_window, serialized_payload)
        
        headers = {
            "X-Bapi-Api-Key": self.api_key,
            "X-Bapi-Timestamp": timestamp,
            "X-Bapi-Sign": signature,
            "X-Bapi-Recv-Window": recv_window,
            "Content-Type": "application/json"
        }

        try:
            if method == "POST":
                response = await self.client.post(url, json=params, headers=headers)
            else:
                response = await self.client.get(url, params=params, headers=headers)
                
            return self._parse_response(response)

        except httpx.TimeoutException as te:
            logging.error(f"Network timeout encountered during {method} {endpoint}: {te}")
            raise BybitTransportException("Timeout communicating with Bybit servers.") from te
        except httpx.NetworkError as ne:
            logging.error(f"Network transport disruption encountered: {ne}")
            raise BybitTransportException("Network connection lost or refused.") from ne

    def _parse_response(self, response: httpx.Response) -> dict:
        """Parses HTTP status codes and Bybit V5 retCode execution signals."""
        status_code = response.status_code
        
        try:
            data = response.json()
        except ValueError as e:
            if status_code >= 500:
                raise BybitAPIException(status_code, -500, "Severe Exchange Gateway Degradation (HTML returned).", response.text) from e
            raise BybitAPIException(status_code, -1, "Invalid JSON payload received from exchange.", response.text) from e

        ret_code = data.get("retCode")
        ret_msg = data.get("retMsg", "No error message provided.")

        # Check HTTP Gateway restrictions
        if status_code in [429, 403] and ret_code != 0:
            raise BybitAPIException(status_code, ret_code, f"IP-Level Threshold Restriction: {ret_msg}", data)

        # Check Business Execution Layer
        if ret_code != 0:
            raise BybitAPIException(status_code, ret_code, ret_msg, data)

        return data.get("result", {})

import asyncio
import random
import logging
from functools import wraps

def retry_on_rate_limit(max_retries: int = 5, base_delay: float = 0.5, max_delay: float = 10.0):
    """
    Python Decorator for handling Bybit V5 Rate Limits (retCode 10006 / HTTP 429)
    and transient HTTP gateway timeouts with Exponential Backoff + Random Jitter.
    """
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            retries = 0
            while True:
                try:
                    return await func(*args, **kwargs)
                except BybitAPIException as e:
                    # retCode 10006 represents 'Too many visits' / Endpoint Rate Limit
                    if e.ret_code == 10006 or e.status_code == 429:
                        retries += 1
                        if retries > max_retries:
                            logging.critical("Max retries breached for rate limit. Escalating exception.")
                            raise e
                        
                        # Calculate exponential backoff with full jitter
                        delay = min(max_delay, base_delay * (2 ** (retries - 1)))
                        jittered_delay = delay + random.uniform(0, 0.5 * delay)
                        
                        logging.warning(
                            f"Rate limit hit during {func.__name__}. "
                            f"Attempt {retries}/{max_retries}. Retrying in {jittered_delay:.2f}s..."
                        )
                        await asyncio.sleep(jittered_delay)
                        continue
                    
                    # If it's a structural server error (502, 503, 504), perform a fast retry
                    if e.status_code in [502, 503, 504]:
                        retries += 1
                        if retries > max_retries:
                            raise e
                        await asyncio.sleep(base_delay * retries)
                        continue
                    
                    # For all other business logic errors (e.g. Insufficient Balance 3100115), fail immediately
                    raise e
                except BybitTransportException:
                    retries += 1
                    if retries > max_retries:
                        raise
                    await asyncio.sleep(base_delay * retries)
                    continue
        return wrapper
    return decorator

import websockets
import json
import asyncio
import logging

class ResilientBybitWS:
    """
    Production Asynchronous WebSocket Manager for Bybit V5 API.
    Handles continuous heartbeat ping-pong, atomic reconnection, and channel resubscription.
    """
    def __init__(self, uri: str = "wss://stream.bybit.com/v5/public/linear"):
        self.uri = uri
        self.ws = None
        self.is_running = False
        self.subscriptions = []

    async def connect(self):
        logging.info(f"Establishing persistent connection to Bybit WebSocket: {self.uri}")
        self.ws = await websockets.connect(self.uri)
        self.is_running = True
        # Re-subscribe to channels if recovering from a disconnect
        if self.subscriptions:
            await self._subscribe(self.subscriptions)

    async def monitor_connection(self):
        """Monitors socket state and triggers automatic reconnection on disconnection."""
        while self.is_running:
            try:
                if not self.ws or self.ws.closed:
                    logging.warning("WebSocket disruption detected! Initiating reconnection loop...")
                    await self.connect()
                await asyncio.sleep(5)
            except Exception as e:
                logging.error(f"Error during WebSocket connection health check: {e}")
                await asyncio.sleep(5)

    async def start(self, channels: list):
        self.subscriptions = channels
        await self.connect()
        
        # Concurrently fire heartbeat, read loop, and connection health monitor
        await asyncio.gather(
            self._heartbeat_loop(),
            self._read_loop(),
            self.monitor_connection()
        )

    async def _subscribe(self, channels: list):
        payload = {
            "op": "subscribe",
            "args": channels
        }
        await self.ws.send(json.dumps(payload))
        logging.info(f"Subscription payload transmitted for: {channels}")

    async def _heartbeat_loop(self):
        """Sends ping payload every 20 seconds to prevent Bybit idle socket pruning."""
        while self.is_running:
            try:
                if self.ws and not self.ws.closed:
                    await self.ws.send(json.dumps({"op": "ping"}))
                    logging.debug("WebSocket Ping frame transmitted.")
            except Exception as e:
                logging.error(f"Failed sending heartbeat ping: {e}")
            await asyncio.sleep(20)

    async def _read_loop(self):
        """Receives incoming data frames and handles ping-pong control frames."""
        while self.is_running:
            try:
                if self.ws and not self.ws.closed:
                    message = await self.ws.recv()
                    data = json.loads(message)
                    
                    # Intercept pong response
                    if data.get("op") == "pong" or data.get("ret_msg") == "pong":
                        logging.debug("WebSocket Pong frame received safely.")
                        continue
                    
                    # Pass market data payload to strategy processor
                    await self.handle_market_data(data)
            except websockets.exceptions.ConnectionClosed:
                logging.error("WebSocket connection closed by Bybit edge server.")
                await asyncio.sleep(1)
            except Exception as e:
                logging.error(f"Unexpected error processing inbound WS message: {e}")
                await asyncio.sleep(0.1)

    async def handle_market_data(self, data: dict):
        """Custom handler for processing orderbook tickers and trade execution updates."""
        if "topic" in data:
            logging.info(f"Stream Update [{data['topic']}]: Received market payload.")

import logging
import sys

def setup_bot_logging(log_filename: str = "bybit_trading_bot.log"):
    """
    Configures a dual-output structured logger for beginner traders.
    Outputs clean format to terminal stdout and records detailed tracebacks in file.
    """
    logger = logging.getLogger("BybitBot")
    logger.setLevel(logging.DEBUG)

    # Console Handler for real-time monitoring
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setLevel(logging.INFO)
    console_format = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", "%H:%M:%S")
    console_handler.setFormatter(console_format)

    # File Handler for audit trail and post-mortem analysis
    file_handler = logging.FileHandler(log_filename)
    file_handler.setLevel(logging.DEBUG)
    file_format = logging.Formatter("%(asctime)s [%(levelname)s] [%(filename)s:%(lineno)d] %(message)s")
    file_handler.setFormatter(file_format)

    logger.addHandler(console_handler)
    logger.addHandler(file_handler)
    
    return logger