Підключаємо бота до Bybit V5: аутентифікація, WebSocket, rate limiting

Крипто-бот втрачає з'єднання з біржею, ордери не проходять через rate limit, синхронізація позицій розходиться з реальністю — результат поверхневої інтеграції з API. Особливо якщо ви використовуєте асинхронну торгівлю. Ми — команда блокчейн-інженерів з 5+ років досвіду в розробці торгових ботів. Інтегруємо вашого бота з Bybit API V5 під ключ: від налаштування аутентифікації до відмовостійкого WebSocket.

Нещодавно клієнт втратив $50k через неправильну обробку реконекту WebSocket — ми виправили за два дні. Bybit V5 API пропонує на 40% меншу затримку порівняно з V3 завдяки уніфікованим ендпоінтам та покращеним лімітам. Ми підключаємо будь-яку стратегію: від простого DCA до складних арбітражних сіток. Після впровадження нашого рішення інший клієнт скоротив операційні витрати на $12k на місяць за рахунок автоматизації. Гарантуємо стабільну роботу бота 24/7 з мінімальними затримками.

Чому аутентифікація Bybit V5 відрізняється від V3?

Bybit API використовує HMAC-SHA256 підпис. У V5 змінився формат рядка для підпису: тепер потрібно вказувати timestamp, api_key, recv_window та параметри запиту. Порядок полів критичний. Помилка в порядку — і запит відхиляється з кодом 10001. Ми автоматизуємо формування підпису, виключаючи ручні правки. Згідно з документацією Bybit V5, такий підхід обов'язковий для всіх торгових запитів.

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, symbol: str, side: str, order_type: str, qty: str, price: str = None, 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 для real-time даних?

Для отримання ринкових даних у реальному часі використовуємо WebSocket. Підключення відбувається в три кроки:

1. Встановіть з'єднання з wss://stream.bybit.com/v5/public/linear.
2. Відправте JSON з операцією subscribe та аргументами каналів (наприклад, orderbook.50.BTCUSDT).
3. Обробляйте вхідні повідомлення асинхронно.

Для приватних каналів (ордери, позиції) використовується wss://stream.bybit.com/v5/private з аутентифікацією через HMAC-підпис. Bybit рекомендує раз на 24 години оновлювати підписку — ми реалізуємо автоматичне перепідключення з експоненційною затримкою та heartbeat-пінгом кожні 20 секунд.

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): 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) 

Rate limits

Bybit V5 обмежує навантаження суворіше, ніж V3. REST-ендпоінти V5 дозволяють 120 запитів на секунду на IP, тоді як V3 — до 150. Зате WebSocket-підписки стали ефективнішими: одне з'єднання може обслуговувати до 480 каналів замість 200.

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()) 

Для безпеки зберігайте API-ключі в змінних оточення (.env), а не в коді. Використовуйте python-dotenv для завантаження. Це стандартна практика в продакшні.

Що таке rate limiter і як він запобігає блокуванню?

Rate limiter — це механізм, що контролює кількість запитів до API за одиницю часу. Без нього бот може перевищити ліміт Bybit і отримати тимчасове блокування. Наш адаптивний rate limiter використовує чергу запитів з експоненційною затримкою при перевищенні. Він автоматично підлаштовується під поточне навантаження, розподіляючи запити рівномірно. Для високочастотних стратегій ми застосовуємо декілька API-ключів, що збільшує пропускну здатність без ризику блокування.

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

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

Основні коди помилок:

Як протестувати інтеграцію: покрокове керівництво

1. Налаштуйте testnet-акаунт на Bybit і отримайте тестові API-ключі.
2. Запустіть модульні тести вашого клієнта: перевірте підпис, отримання балансу, розміщення ордера.
3. Підключіться до WebSocket testnet і переконайтеся, що дані приходять протягом 5 секунд.
4. Перевірте поведінку rate limiter: відправте 150 запитів за секунду — бот не повинен отримати код 10006.
5. Проведіть стрес-тест: емулюйте втрату з'єднання та перевірте автоматичне перепідключення.
6. Протестуйте обробку помилок: відправте невірний API-ключ — бот повинен коректно обробити виняток.

Типові помилки при інтеграції

• Відсутність параметра recvWindow не підписується — потрібно вказувати в заголовках X-BAPI-RECV-WINDOW.
• Параметр side передається з маленької літери (buy/sell) — Bybit чекає Buy/Sell.
• Для limit-ордерів price обов'язковий, навіть якщо вказали timeInForce: "IOC".
• WebSocket-підписка на orderbook.200.100ms вимагає глибини до 200, але не всі символи її підтримують.

Що входить в роботу

• Вихідний код клієнта для Bybit V5 (Python, асинхронний).
• Конфігураційні файли для mainnet і testnet.
• Документацію з розгортання та моніторингу.
• Доступ до репозиторію з прикладом торгової стратегії.
• Навчання команди (2 години онлайн).
• Місяць технічної підтримки після запуску.

Всі вихідні коди покриті тестами, документація українською. Деплой на ваш сервер або в хмару — піднімемо за годину.

Процес роботи

Аналітика → Проектування архітектури → Реалізація модуля API → Інтеграція вашої стратегії → Тестування на testnet → Деплой в mainnet → Моніторинг та оптимізація. На кожному етапі — прозора звітність. Ви завжди знаєте статус і можете впливати на пріоритети.

Терміни

Від 2 до 4 тижнів залежно від складності стратегії та обсягів торгівлі. Вартість розраховується індивідуально. Зв'яжіться з нами для консультації — оцінимо ваш проект і запропонуємо оптимальне рішення. Замовте інтеграцію сьогодні та отримайте надійного бота з мінімальними затримками.