Інтеграція бота з API Bybit

Інтеграція бота з API Bybit

Bybit надає REST та WebSocket API для торгівлі, моніторингу позицій та отримання даних про ринок. Документація актуальна для Bybit V5 API — уніфікованого endpoint'а, який об'єднує spot, linear, inverse та option під одним інтерфейсом.

Аутентифікація

Bybit API використовує HMAC-SHA256 підпис. Для приватних 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()

Розміщення ордерів

async def place_order(
    self,
    category: str,     # 'spot', 'linear', 'inverse'
    symbol: str,       # 'BTCUSDT'
    side: str,         # 'Buy' або 'Sell' (з великої букви!)
    order_type: str,   # 'Market' або 'Limit'
    qty: str,          # кількість як рядок
    price: str = None, # для limit ордерів
    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 для даних у реальному часі

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:
            # Підписка
            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):
        """Приватний WebSocket для сповіщень про ордери"""
        ws_url = "wss://stream.bybit.com/v5/private"

        async with websockets.connect(ws_url) as ws:
            # Аутентифікація
            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]
            }))

            # Підписка на виконання ордерів
            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)

Обмеження швидкості

Bybit обмежує кількість запитів. Для V5 API:

• REST: 120 запитів/сек на IP (глобально), 10-600 запитів/сек на endpoint
• WebSocket: до 480 підписок на з'єднання

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()
        # Видаляємо застарілі записи
        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())

Обробка помилок

Bybit повертає retCode: 0 при успіху, ненульове значення при помилці:

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", {})

Частіші коди помилок: 10001 — невірний API key, 10006 — перевищено обмеження швидкості, 110007 — недостатньо балансу, 130021 — ордер не знайдено. Для production бота — реалізуйте обробник для всіх кодів з логікою повторних спроб для тимчасових помилок.