Nếu bạn đang xây dựng bot giao dịch tự động hoặc tích hợp dữ liệu thị trường từ Bybit, việc hiểu rõ sự khác biệt giữa public API và private API là bước đầu tiên bắt buộc. Trong bài viết này, mình sẽ chia sẻ kinh nghiệm thực chiến khi tích hợp Bybit API cho nhiều dự án khác nhau — từ dashboard theo dõi portfolio đến hệ thống trading tự động hoàn chỉnh.
Public API vs Private API: Khác Biệt Cốt Lõi
Bybit phân chia API thành hai nhóm rõ ràng dựa trên mức độ truy cập:
Public API — Không Cần Xác Thực
Public endpoints cho phép truy cập dữ liệu công khai mà không cần API key. Đây là lựa chọn lý tưởng để:
- Lấy dữ liệu ticker, orderbook, klines
- Query thông tin tài sản và trading pairs
- Theo dõi price action theo thời gian thực
Private API — Bắt Buộc HMAC Authentication
Private endpoints yêu cầu xác thực đầy đủ để thực hiện các thao tác như:
- Đặt lệnh mua/bán
- Query account balance
- Quản lý withdrawal
- Access order history
Cách Thiết Lập Bybit API Key
Trước khi bắt đầu code, bạn cần tạo API key từ Bybit console:
- Đăng nhập Bybit → Account & Security → API Management
- Tạo API key mới với label识别 (ví dụ: "TradingBot")
- Chọn permissions phù hợp: Read-Only, Trade, or Withdrawal
- Lưu giữ API Key và Secret Key — secret key chỉ hiển thị một lần duy nhất
Authentication Flow Chi Tiết
Mình đã từng debug nhiều trường hợp authentication failed vì thiếu bước nào đó trong quy trình. Dưới đây là flow đầy đủ mà mình sử dụng cho mọi dự án:
Bước 1: Tạo Signature
Bybit sử dụng HMAC SHA256 để sign request. Công thức:
import hmac
import hashlib
import time
def create_signature(secret_key: str, timestamp: str, recv_window: str, message: str) -> str:
"""
Tạo signature cho Bybit API request
secret_key: API Secret Key từ Bybit
timestamp: Unix timestamp milliseconds (e.g., "1704067200000")
recv_window: Request timeout (e.g., "5000")
message: String nối từ params (param=value)
"""
pre_hash = timestamp + secret_key + recv_window + message
signature = hmac.new(
secret_key.encode('utf-8'),
pre_hash.encode('utf-8'),
hashlib.sha256
).hexdigest()
return signature
Ví dụ sử dụng
timestamp = str(int(time.time() * 1000))
recv_window = "5000"
params_string = "category=linear&symbol=BTCUSDT&limit=10"
secret_key = "YOUR_BYBIT_SECRET_KEY"
signature = create_signature(secret_key, timestamp, recv_window, params_string)
print(f"Signature: {signature}")
Bước 2: Build Request Headers
import requests
from typing import Dict, Optional
class BybitAPIClient:
BASE_URL = "https://api.bybit.com"
def __init__(self, api_key: str, api_secret: str):
self.api_key = api_key
self.api_secret = api_secret
self.recv_window = "5000"
def _get_headers(self, timestamp: str, signature: str) -> Dict[str, str]:
"""Build headers cho request có authentication"""
return {
"X-BAPI-API-KEY": self.api_key,
"X-BAPI-SIGN": signature,
"X-BAPI-SIGN-TYPE": "2", # HMAC SHA256
"X-BAPI-TIMESTAMP": timestamp,
"X-BAPI-RECV-WINDOW": self.recv_window,
"Content-Type": "application/json"
}
def _make_request(
self,
method: str,
endpoint: str,
params: Optional[Dict] = None,
requires_auth: bool = True
) -> Dict:
"""Generic request handler cho Bybit API"""
timestamp = str(int(time.time() * 1000))
url = f"{self.BASE_URL}{endpoint}"
if requires_auth:
# Build param string theo format: key1=value1&key2=value2
param_str = "&".join([f"{k}={v}" for k, v in sorted(params.items())]) if params else ""
# Tạo signature
signature = create_signature(self.api_secret, timestamp, self.recv_window, param_str)
headers = self._get_headers(timestamp, signature)
else:
headers = {"Content-Type": "application/json"}
if method == "GET":
response = requests.get(url, params=params, headers=headers)
else:
response = requests.post(url, json=params, headers=headers)
result = response.json()
# Error handling theo Bybit convention
if result.get("retCode") != 0:
raise BybitAPIError(
code=result.get("retCode"),
msg=result.get("retMsg")
)
return result.get("result", {})
Sử dụng
client = BybitAPIClient(
api_key="YOUR_API_KEY",
api_secret="YOUR_SECRET_KEY"
)
Ví Dụ Thực Tế: Public vs Private Endpoints
Ví dụ 1: Lấy Dữ Liệu Thị Trường (Public)
# === PUBLIC API: Không cần authentication ===
def get_ticker_info(symbol: str = "BTCUSDT"):
"""Lấy thông tin ticker - không cần API key"""
endpoint = "/v5/market/tickers"
params = {
"category": "linear", # perpetual futures
"symbol": symbol
}
# Không cần signature, không cần headers đặc biệt
url = "https://api.bybit.com/v5/market/tickers"
response = requests.get(url, params=params)
return response.json()
Sử dụng
ticker = get_ticker_info("BTCUSDT")
print(f"Giá BTC hiện tại: {ticker['result']['list'][0]['lastPrice']}")
def get_orderbook(symbol: str = "BTCUSDT", limit: int = 50):
"""Lấy orderbook depth - public endpoint"""
endpoint = "/v5/market/orderbook"
params = {
"category": "linear",
"symbol": symbol,
"limit": limit
}
url = "https://api.bybit.com/v5/market/orderbook"
response = requests.get(url, params=params)
return response.json()
Test
book = get_orderbook("BTCUSDT", 20)
print(f"Top 5 asks: {book['result']['a'][:5]}")
print(f"Top 5 bids: {book['result']['b'][:5]}")
Ví dụ 2: Query Balance (Private)
# === PRIVATE API: Bắt buộc authentication ===
def get_wallet_balance(client: BybitAPIClient, coin: str = "USDT"):
"""Lấy số dư ví - yêu cầu authentication"""
params = {
"accountType": "UNIFIED",
"coin": coin
}
return client._make_request(
method="GET",
endpoint="/v5/account/wallet-balance",
params=params,
requires_auth=True
)
Sử dụng với client đã khởi tạo
balance = get_wallet_balance(client, "USDT")
total_equity = balance['list'][0]['totalEquity']
print(f"Tổng equity USDT: {total_equity}")
def place_order(
client: BybitAPIClient,
symbol: str = "BTCUSDT",
side: str = "Buy",
order_type: str = "Market",
qty: float = 0.001
):
"""Đặt lệnh giao dịch - private endpoint"""
params = {
"category": "linear",
"symbol": symbol,
"side": side,
"orderType": order_type,
"qty": str(qty), # Bybit yêu cầu string
"timeInForce": "GTC"
}
return client._make_request(
method="POST",
endpoint="/v5/order/create",
params=params,
requires_auth=True
)
Đặt lệnh market buy 0.001 BTC
order = place_order(client, "BTCUSDT", "Buy", "Market", 0.001)
print(f"Order ID: {order['orderId']}")
So Sánh Authentication Requirements
| Yêu cầu | Public API | Private API |
|---|---|---|
| API Key | Không cần | Bắt buộc |
| Secret Key | Không cần | Bắt buộc |
| HMAC Signature | Không cần | Bắt buộc |
| Timestamp | Không cần | Bắt buộc |
| Recv Window | Không cần | Bắt buộc |
| Rate Limit | 600/min | 600/min |
| IP Whitelist | Không bắt buộc | Khuyến nghị |
Security Best Practices
Qua nhiều dự án, mình rút ra một số best practices quan trọng:
- Không hardcode API keys: Sử dụng environment variables hoặc secret manager
- IP Whitelist: Giới hạn IP truy cập API key để ngăn chặn truy cập trái phép
- Permission tối thiểu: Chỉ cấp quyền cần thiết (Read-Only cho dashboard, Trade cho bot)
- Kiểm tra signature: Luôn validate response signature nếu có
- Retry với exponential backoff: Xử lý rate limit gracefully
# === Environment Variables Setup ===
import os
from dotenv import load_dotenv
load_dotenv() # Load từ .env file
BYBIT_API_KEY = os.getenv("BYBIT_API_KEY")
BYBIT_API_SECRET = os.getenv("BYBIT_API_SECRET")
Kiểm tra trước khi sử dụng
if not BYBIT_API_KEY or not BYBIT_API_SECRET:
raise ValueError("Missing Bybit API credentials in environment variables")
=== Retry Logic với Exponential Backoff ===
import time
from functools import wraps
def retry_on_rate_limit(max_retries=3, base_delay=1):
"""Decorator để retry khi gặp rate limit"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except BybitAPIError as e:
if e.code == 10002: # Rate limit error
delay = base_delay * (2 ** attempt)
print(f"Rate limited. Retry in {delay}s...")
time.sleep(delay)
else:
raise
raise Exception(f"Max retries exceeded after {max_retries} attempts")
return wrapper
return decorator
@retry_on_rate_limit(max_retries=3, base_delay=2)
def get_server_time():
"""Lấy server time để sync"""
url = "https://api.bybit.com/v5/market/time"
response = requests.get(url)
return response.json()['result']['timeMs']
Lỗi Thường Gặp và Cách Khắc Phục
Lỗi 1: "10002 - signature not match"
Nguyên nhân: Signature không khớp với params đã gửi
# === SAI: Thứ tự params không đúng ===
def wrong_way():
params = {
"symbol": "BTCUSDT",
"category": "linear", # Thứ tự sai!
}
param_str = "category=linear&symbol=BTCUSDT" # Must be sorted!
=== ĐÚNG: Params phải được sort theo key ===
def correct_way():
params = {
"category": "linear",
"symbol": "BTCUSDT"
}
# Tự động sort và nối
param_str = "&".join([f"{k}={v}" for k, v in sorted(params.items())])
# Result: "category=linear&symbol=BTCUSDT"
=== Checklist debug signature ===
def debug_signature(api_secret, timestamp, recv_window, params):
"""Tool debug signature"""
# 1. Đảm bảo params đã sort
sorted_params = sorted(params.items())
# 2. Tạo param string chuẩn
param_str = "&".join([f"{k}={v}" for k, v in sorted_params])
# 3. Pre-hash format phải đúng
pre_hash = f"{timestamp}{api_secret}{recv_window}{param_str}"
print(f"Timestamp: {timestamp}")
print(f"Recv Window: {recv_window}")
print(f"Param String: {param_str}")
print(f"Pre-hash: {pre_hash}")
signature = hmac.new(
api_secret.encode(),
pre_hash.encode(),
hashlib.sha256
).hexdigest()
print(f"Signature: {signature}")
return signature
Lỗi 2: "10006 - timestamp expired"
Nguyên nhân: Timestamp request vượt quá recv_window (thường là 30 giây)
# === VẤN ĐỀ: Timestamp server client không sync ===
def common_mistake():
# Sai: Dùng time.time() nhưng server có độ trễ mạng
timestamp = str(int(time.time() * 1000))
# Server nhận request sau 5 giây
# Timestamp expired!
=== GIẢI PHÁP 1: Tăng recv_window ===
def solution_1():
client.recv_window = "30000" # 30 seconds thay vì 5 seconds
=== GIẢI PHÁP 2: Sync timestamp với server ===
def solution_2():
# Lấy server time trước
server_time_url = "https://api.bybit.com/v5/market/time"
response = requests.get(server_time_url)
server_time = int(response.json()['result']['timeMs'])
# Tính offset với local time
local_time = int(time.time() * 1000)
time_offset = server_time - local_time
# Sử dụng adjusted timestamp
adjusted_timestamp = str(int(time.time() * 1000) + time_offset)
return adjusted_timestamp
=== GIẢI PHÁP 3: Wrapper tự động sync ===
class SyncedBybitClient(BybitAPIClient):
def __init__(self, api_key, api_secret):
super().__init__(api_key, api_secret)
self._sync_time()
def _sync_time(self):
"""Sync timestamp với Bybit server"""
server_time_url = "https://api.bybit.com/v5/market/time"
response = requests.get(server_time_url)
self._server_offset = int(response.json()['result']['timeMs']) - int(time.time() * 1000)
def _get_synced_timestamp(self) -> str:
return str(int(time.time() * 1000) + self._server_offset)
Lỗi 3: "10003 - invalid API key"
Nguyên nhân: API key không hợp lệ, bị revoke, hoặc sai permission
# === CHECKLIST KHI GẶP LỖI NÀY ===
def troubleshoot_api_key():
"""
1. Kiểm tra API key còn active không
→ Vào Bybit → Account & Security → API Management
→ Xem trạng thái key (Active/Disabled)
2. Kiểm tra permission có đúng không
→ Read-Only key không thể place order
→ Cần Trade permission cho private endpoints
3. Kiểm tra IP whitelist
→ Nếu bật IP whitelist, server IP phải được whitelisted
→ Test bằng cách tạm tắt IP whitelist
4. Kiểm tra API key có bị rate limit không
→ 10002 vs 10003 có thể bị confuse
"""
pass
=== TEST KẾT NỐI ===
def test_connection(api_key: str, api_secret: str) -> dict:
"""Test kết nối với private endpoint đơn giản nhất"""
client = BybitAPIClient(api_key, api_secret)
try:
# Lấy thông tin tài khoản - endpoint đơn giản nhất
result = client._make_request(
method="GET",
endpoint="/v5/account/wallet-balance",
params={"accountType": "UNIFIED"},
requires_auth=True
)
return {"success": True, "data": result}
except BybitAPIError as e:
if e.code == 10003:
return {"success": False, "error": "Invalid API key or permission"}
elif e.code == 10002:
return {"success": False, "error": "Rate limit hit"}
else:
return {"success": False, "error": str(e)}
=== SỬ DỤNG ===
result = test_connection("YOUR_KEY", "YOUR_SECRET")
if result["success"]:
print("✅ Kết nối thành công!")
else:
print(f"❌ Lỗi: {result['error']}")
Lỗi 4: "10004 - missing required parameter"
Nguyên nhân: Thiếu parameter bắt buộc hoặc sai format
# === VẤN ĐỀ THƯỜNG GẶP ===
def common_param_issues():
# 1. Qty phải là STRING, không phải FLOAT
params = {"qty": 0.001} # SAI!
params = {"qty": "0.001"} # ĐÚNG
# 2. Symbol phải đúng format
params = {"symbol": "BTC/USDT"} # SAI!
params = {"symbol": "BTCUSDT"} # ĐÚNG
# 3. Category bắt buộc cho v5 API
params = {} # SAI - thiếu category
params = {"category": "linear"} # ĐÚNG - spot, linear, option
# 4. timeInForce bắt buộc cho limit order
params = {
"orderType": "Limit", # Limit order
# "timeInForce": "GTC" # SAI - thiếu timeInForce
}
=== VALIDATION HELPER ===
def validate_order_params(symbol: str, qty: float, price: float = None,
order_type: str = "Market") -> dict:
"""Validate và format order params theo Bybit requirements"""
params = {
"category": "linear",
"symbol": symbol.upper().replace("/", ""), # Normalize symbol
"side": "Buy", # Cần specify
"orderType": order_type,
"qty": str(qty), # MUST be string!
}
if order_type == "Limit":
if price is None:
raise ValueError("Limit order requires price")
params["price"] = str(price)
params["timeInForce"] = "GTC" # Required for limit
return params
=== SỬ DỤNG ===
try:
params = validate_order_params("btc/usdt", 0.001, 50000, "Limit")
print(f"Params hợp lệ: {params}")
except ValueError as e:
print(f"Validation failed: {e}")
Rate Limits và Best Practices
Bybit có rate limit khác nhau cho từng loại endpoint:
- Market Data: 600 requests/second (public)
- Trading: 600 requests/second (private)
- Position: 600 requests/second
- Account: 600 requests/second
# === RATE LIMIT HANDLER ===
import threading
import time
from collections import deque
class RateLimiter:
"""Token bucket rate limiter cho Bybit API"""
def __init__(self, max_requests: int = 600, window: float = 1.0):
self.max_requests = max_requests
self.window = window
self.requests = deque()
self.lock = threading.Lock()
def acquire(self):
"""Block cho đến khi có quota available"""
with self.lock:
now = time.time()
# Remove requests cũ
while self.requests and self.requests[0] < now - self.window:
self.requests.popleft()
if len(self.requests) >= self.max_requests:
# Phải đợi
sleep_time = self.window - (now - self.requests[0])
time.sleep(sleep_time)
return self.acquire() # Recursive
self.requests.append(now)
Sử dụng
limiter = RateLimiter(max_requests=100, window=1.0)
def rate_limited_request(method, endpoint, params, auth=True):
limiter.acquire() # Chờ nếu cần
return client._make_request(method, endpoint, params, auth)
=== BATCH REQUEST OPTIMIZATION ===
def get_multiple_tickers(symbols: list) -> list:
"""Lấy ticker cho nhiều symbol trong 1 request thay vì nhiều request"""
# Bybit cho phép query nhiều symbol 1 lúc
symbols_str = ",".join(symbols) # "BTCUSDT,ETHUSDT,SOLUSDT"
params = {
"category": "linear",
"symbol": symbols_str
}
# 1 request thay vì 3 requests
response = requests.get("https://api.bybit.com/v5/market/tickers", params=params)
return response.json()['result']['list']
Tổng Kết
Việc phân biệt rõ ràng giữa public API (không cần authentication) và private API (yêu cầu HMAC signature) là nền tảng quan trọng khi làm việc với Bybit. Mình đã tổng hợp lại những lỗi thường gặp nhất dựa trên kinh nghiệm thực chiến:
- Signature not match: Kiểm tra lại thứ tự sort params
- Timestamp expired: Sync time với server hoặc tăng recv_window
- Invalid API key: Verify permissions và IP whitelist
- Missing parameter: Validate format trước khi gửi
Code examples trong bài viết này sử dụng Bybit API v5 — phiên bản mới nhất với performance tốt hơn và rate limit hào phóng hơn so với v3.
Nếu bạn đang xây dựng hệ thống trading phức tạp với nhiều exchange, việc abstract hóa authentication layer sẽ giúp code sạch hơn rất nhiều. Mình sẽ chia sẻ thêm về chủ đề này trong các bài viết tiếp theo.
Chúc các bạn implement thành công! Nếu có câu hỏi hoặc cần hỗ trợ thêm, hãy để lại comment bên dưới.
Bài viết được cập nhật lần cuối: 2026. Thông tin rate limits và API endpoints có thể thay đổi theo chính sách Bybit.