Bot Integration with Bybit API

Bybit API Bot Integration

Bybit provides REST and WebSocket APIs for trading, position monitoring, and market data retrieval. Documentation is current for Bybit V5 API — a unified endpoint that combines spot, linear, inverse, and option under a single interface.

Authentication

Bybit API uses HMAC-SHA256 signature. For private endpoints:

import hmac
import hashlib
import time
import httpx

class BybitClient:
    BASE_URL = "https://api.bybit.com"

    def __init__(self, api_key: str, api_secret: str, testnet: bool = False):
        self.api_key = api_key
        self.api_secret = api_secret
        if testnet:
            self.BASE_URL = "https://api-testnet.bybit.com"

    def _sign(self, params: str, timestamp: int) -> str:
        sign_str = f"{timestamp}{self.api_key}5000{params}"
        return hmac.new(
            self.api_secret.encode('utf-8'),
            sign_str.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()

    async def get_wallet_balance(self, account_type: str = "UNIFIED") -> dict:
        timestamp = int(time.time() * 1000)
        params = f"accountType={account_type}"
        signature = self._sign(params, timestamp)

        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.BASE_URL}/v5/account/wallet-balance",
                params={"accountType": account_type},
                headers={
                    "X-BAPI-API-KEY": self.api_key,
                    "X-BAPI-TIMESTAMP": str(timestamp),
                    "X-BAPI-RECV-WINDOW": "5000",
                    "X-BAPI-SIGN": signature
                }
            )
        return response.json()

Order Placement

async def place_order(
    self,
    category: str,     # 'spot', 'linear', 'inverse'
    symbol: str,       # 'BTCUSDT'
    side: str,         # 'Buy' or 'Sell' (uppercase!)
    order_type: str,   # 'Market' or 'Limit'
    qty: str,          # quantity as string
    price: str = None, # for limit orders
    time_in_force: str = "GTC"
) -> dict:
    payload = {
        "category": category,
        "symbol": symbol,
        "side": side,
        "orderType": order_type,
        "qty": qty,
        "timeInForce": time_in_force
    }
    if price:
        payload["price"] = price

    timestamp = int(time.time() * 1000)
    body = json.dumps(payload)
    signature = self._sign(body, timestamp)

    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{self.BASE_URL}/v5/order/create",
            content=body,
            headers={
                "X-BAPI-API-KEY": self.api_key,
                "X-BAPI-TIMESTAMP": str(timestamp),
                "X-BAPI-RECV-WINDOW": "5000",
                "X-BAPI-SIGN": signature,
                "Content-Type": "application/json"
            }
        )
    return response.json()

WebSocket for Real-time Data

import asyncio
import websockets
import json

class BybitWebSocket:
    WS_URL = "wss://stream.bybit.com/v5/public/linear"

    async def subscribe_orderbook(self, symbol: str, depth: int = 50):
        async with websockets.connect(self.WS_URL) as ws:
            # Subscribe
            await ws.send(json.dumps({
                "op": "subscribe",
                "args": [f"orderbook.{depth}.{symbol}"]
            }))

            async for message in ws:
                data = json.loads(message)
                if data.get("topic", "").startswith("orderbook"):
                    await self.process_orderbook(data)

    async def subscribe_private(self, api_key: str, api_secret: str):
        """Private WebSocket for order notifications"""
        ws_url = "wss://stream.bybit.com/v5/private"

        async with websockets.connect(ws_url) as ws:
            # Authentication
            expires = int((time.time() + 10) * 1000)
            sign = hmac.new(
                api_secret.encode(),
                f"GET/realtime{expires}".encode(),
                hashlib.sha256
            ).hexdigest()

            await ws.send(json.dumps({
                "op": "auth",
                "args": [api_key, expires, sign]
            }))

            # Subscribe to order executions
            await ws.send(json.dumps({
                "op": "subscribe",
                "args": ["order", "execution", "position"]
            }))

            async for message in ws:
                data = json.loads(message)
                await self.handle_private_event(data)

Rate Limits

Bybit limits the number of requests. For V5 API:

• REST: 120 requests/sec per IP (global), 10-600 requests/sec per endpoint
• WebSocket: up to 480 subscriptions per connection

import asyncio
from collections import deque

class RateLimiter:
    def __init__(self, max_requests: int, window_seconds: float):
        self.max_requests = max_requests
        self.window = window_seconds
        self.requests = deque()

    async def acquire(self):
        now = time.monotonic()
        # Remove outdated records
        while self.requests and self.requests[0] < now - self.window:
            self.requests.popleft()

        if len(self.requests) >= self.max_requests:
            sleep_time = self.requests[0] + self.window - now
            await asyncio.sleep(sleep_time)

        self.requests.append(time.monotonic())

Error Handling

Bybit returns retCode: 0 on success, non-zero on error:

def check_response(self, response: dict, operation: str):
    ret_code = response.get("retCode", -1)
    if ret_code != 0:
        error_msg = response.get("retMsg", "Unknown error")
        raise BybitAPIError(f"{operation} failed [{ret_code}]: {error_msg}")
    return response.get("result", {})

Common error codes: 10001 — invalid API key, 10006 — rate limit exceeded, 110007 — insufficient balance, 130021 — order not found. For a production bot — implement a handler for all codes with retry logic for temporary errors.