Your crypto bot loses connection to the exchange, orders fail due to rate limits, position sync drifts from reality — the result of a shallow API integration. Especially if you use async trading. We are a team of blockchain engineers with 5+ years of experience in trading bot development. We integrate your bot with Bybit API V5 end-to-end: from authentication setup to fault-tolerant WebSocket.
Recently, a client lost $50k due to incorrect WebSocket reconnect handling — we fixed it in two days. Bybit V5 API offers 40% lower latency compared to V3 thanks to unified endpoints and improved limits. We connect any strategy: from simple DCA to complex arbitrage grids. After deploying our solution, another client cut operational costs by $12k per month through automation. We guarantee stable bot operation 24/7 with minimal latency.
Why Bybit V5 Authentication Differs from V3
Bybit API uses HMAC-SHA256 signing. In V5, the signature string format changed: now you must include timestamp, api_key, recv_window, and request parameters. The field order is critical. An ordering mistake — and the request is rejected with code 10001. We automate signature generation, eliminating manual edits. According to Bybit V5 documentation, this approach is mandatory for all trading requests.
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()
Placing Orders
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()
How to Set Up WebSocket for Real-Time Data?
For real-time market data, we use WebSocket. The connection involves three steps:
- Connect to
wss://stream.bybit.com/v5/public/linear. - Send a JSON with operation
subscribeand channel arguments (e.g.,orderbook.50.BTCUSDT). - Process incoming messages asynchronously.
For private channels (orders, positions), use wss://stream.bybit.com/v5/private with HMAC-signed authentication. Bybit recommends refreshing subscriptions every 24 hours — we implement automatic reconnection with exponential backoff and heartbeat pings every 20 seconds.
Example WebSocket Implementation
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 enforces stricter limits than V3. REST endpoints in V5 allow 120 requests per second per IP, while V3 allowed up to 150. However, WebSocket subscriptions became more efficient: a single connection can serve up to 480 channels instead of 200.
| Method | Limit | Comment |
|---|---|---|
| REST (global) | 120 req/s per IP | Across all endpoints |
| REST (per endpoint) | 10-600 req/s | Depends on type |
| WebSocket | 480 subscriptions per connection | 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()
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())
For security, store API keys in environment variables (.env), not in code. Use python-dotenv for loading. This is standard practice in production.
What Is a Rate Limiter and How Does It Prevent Blocking?
A rate limiter controls the number of requests to an API per unit time. Without it, your bot may exceed Bybit's limits and get temporarily blocked. Our adaptive rate limiter uses a request queue with exponential backoff on overage. It automatically adjusts to current load, distributing requests evenly. For high-frequency strategies, we use multiple API keys to increase throughput without risking a block.
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:
| Code | Meaning | Action |
|---|---|---|
| 10001 | Invalid API key | Check key and permissions |
| 10006 | Rate limit exceeded | Wait or reduce frequency |
| 110007 | Insufficient balance | Adjust order size |
| 130021 | Order not found | Verify orderId |
How to Test the Integration: Step-by-Step Guide
- Set up a testnet account on Bybit and obtain test API keys.
- Run unit tests for your client: check signing, balance retrieval, order placement.
- Connect to WebSocket testnet and verify data arrives within 5 seconds.
- Test the rate limiter: send 150 requests per second — the bot should not get code 10006.
- Run a stress test: simulate connection loss and check automatic reconnection.
- Test error handling: send an invalid API key — the bot should handle the exception correctly.
Common Integration Mistakes
- Missing
recvWindowparameter not signed — must includeX-BAPI-RECV-WINDOWin headers. -
sideparameter sent in lowercase (buy/sell) — Bybit expectsBuy/Sell. - For limit orders,
priceis mandatory even iftimeInForce: "IOC"is set. - WebSocket subscription to
orderbook.200.100msrequires depth up to 200, but not all symbols support it.
What's Included in the Work
- Source code of the Bybit V5 client (Python, async).
- Configuration files for mainnet and testnet.
- Documentation for deployment and monitoring.
- Access to a repository with a sample trading strategy.
- Team training (2 hours online).
- One month of technical support post-launch.
All source code is covered by tests, documentation is in Russian. Deployment to your server or cloud — we set it up within an hour.
Process
Analysis → Architecture design → API module implementation → Integration of your strategy → Testnet testing → Mainnet deployment → Monitoring and optimization. At each stage — transparent reporting. You always know the status and can influence priorities.
Timelines
From 2 to 4 weeks depending on strategy complexity and trading volumes. The cost is calculated individually. Contact us for a consultation — we will evaluate your project and offer the optimal solution. Order the integration today and get a reliable bot with minimal latency.