Từ kinh nghiệm 3 năm giao dịch tự động và từng burn $12,000 do lỗi API rate limit, tôi chia sẻ toàn bộ hành trình xây dựng bot giao dịch Bybit từ con số 0. Bài viết này dành cho người hoàn toàn chưa biết gì về API — bạn sẽ có bot chạy thực tế sau khi đọc xong.
Mục lục
- Giới thiệu: Tại sao API Bybit?
- Chuẩn bị: Tài khoản và API Key
- Python cho người mới: Cài đặt môi trường
- Kết nối Bybit: Code đầu tiên
- Chiến lược Arbitrage thực chiến
- Lỗi thường gặp và cách khắc phục
- Vì sao nên dùng HolySheep cho AI
Giới thiệu: Tại sao API Bybit?
Bybit là sàn giao dịch phái sinh lớn thứ 2 thế giới với khối lượng giao dịch $50 tỷ/ngày. API của họ nổi tiếng vì:
- Tốc độ phản hồi trung bình 20-50ms
- Hỗ trợ đầy đủ perpetual futures (hợp đồng không đáo hạn)
- Phí giao dịch chỉ 0.01% (maker) — thấp nhất thị trường
- Tài liệu API chi tiết, có cả testnet miễn phí
Với người mới, tôi khuyên bạn luôn test trên testnet trước. Tài khoản testnet hoàn toàn miễn phí, được cấp 100 USDT ảo mỗi ngày. Không có gì đáng xấu hổ hơn việc đặt sai lệnh và mất tiền thật trong lần thử nghiệm đầu tiên.
Chuẩn bị: Tạo tài khoản và API Key Bybit
Bước 1: Đăng ký tài khoản Bybit
Đăng ký tài khoản Bybit (sử dụng link giới thiệu để giảm phí 10%):
- Truy cập bybit.com → Đăng ký → Xác minh KYC cơ bản
- Vào mục Account & Security → API Management
- Tạo API key mới với quyền: Read-Only (ban đầu) hoặc Trade (khi đã quen)
Bước 2: Lấy thông tin cần thiết
Sau khi tạo API key, bạn sẽ nhận được 3 thông tin quan trọng:
API Key: bybit_test_api_key_abc123xyz
API Secret: super_secret_key_do_not_share
Testnet URL: https://api-testnet.bybit.com
LƯU Ý: API Secret chỉ hiển thị 1 LẦN DUY NHẤT
Hãy copy và lưu vào nơi an toàn ngay lập tức
Bước 3: Cài đặt môi trường Python
Tôi khuyên dùng Python 3.10+ vì sự ổn định và tương thích thư viện. Đây là code setup hoàn chỉnh:
# Cài đặt môi trường (chạy trong Terminal/Command Prompt)
Windows
python -m venv venv
venv\Scripts\activate
macOS/Linux
python3 -m venv venv
source venv/bin/activate
Cài đặt các thư viện cần thiết
pip install requests pyjwt cryptography pandas numpy
Kiểm tra cài đặt thành công
python -c "import requests; print('Setup thành công!')"
Python cho người mới: Kết nối API Bybit
Đây là phần quan trọng nhất. Tôi sẽ giải thích từng dòng code để bạn hiểu chứ không chỉ copy-paste.
1. Import thư viện và cấu hình
# file: bybit_connection.py
import requests
import time
import hashlib
import hmac
from urllib.parse import urlencode
====== CẤU HÌNH API CỦA BẠN ======
BYBIT_API_KEY = "your_bybit_api_key_here"
BYBIT_API_SECRET = "your_bybit_secret_here"
TESTNET_BASE_URL = "https://api-testnet.bybit.com"
Hàm tạo chữ ký điện tử (signature)
ĐÂY LÀ PHẦN QUAN TRỌNG NHẤT - KHÔNG THAY ĐỔI
def generate_signature(param_str, secret):
"""
Tạo chữ ký HMAC-SHA256 để xác thực request
param_str: chuỗi tham số đã mã hóa
secret: API Secret của bạn
"""
signature = hmac.new(
bytes(secret, 'utf-8'),
bytes(param_str, 'utf-8'),
hashlib.sha256
).hexdigest()
return signature
Hàm gửi request đến Bybit
def send_request(method, endpoint, params=None):
"""
Gửi request HTTP đến Bybit API
method: GET hoặc POST
endpoint: đường dẫn API (ví dụ: /v5/market/tickers)
params: tham số truyền vào
"""
url = f"{TESTNET_BASE_URL}{endpoint}"
timestamp = str(int(time.time() * 1000))
# Tạo chuỗi query cho request
if params:
params['api_key'] = BYBIT_API_KEY
params['timestamp'] = timestamp
params['sign'] = generate_signature(
urlencode(sorted(params.items())),
BYBIT_API_SECRET
)
else:
params = {
'api_key': BYBIT_API_KEY,
'timestamp': timestamp
}
params['sign'] = generate_signature(
urlencode(sorted(params.items())),
BYBIT_API_SECRET
)
if method == "GET":
response = requests.get(url, params=params)
else:
response = requests.post(url, data=params)
return response.json()
print("✅ Kết nối API Bybit đã được cấu hình!")
2. Lấy dữ liệu thị trường thời gian thực
# file: market_data.py
Tiếp tục từ file trên
def get_perpetual_tickers():
"""
Lấy danh sách tất cả perpetual futures đang giao dịch
Endpoint: /v5/market/tickers
"""
params = {
'category': 'linear', # linear = perpetual futures
}
result = send_request("GET", "/v5/market/tickers", params)
if result.get('retCode') == 0:
print("✅ Lấy dữ liệu thành công!")
return result['result']['list']
else:
print(f"❌ Lỗi: {result.get('retMsg')}")
return None
def get_ticker_price(symbol):
"""
Lấy giá hiện tại của một cặp giao dịch cụ thể
Ví dụ: symbol = "BTCUSDT"
"""
params = {
'category': 'linear',
'symbol': symbol
}
result = send_request("GET", "/v5/market/tickers", params)
if result.get('retCode') == 0 and result['result']['list']:
ticker = result['result']['list'][0]
return {
'symbol': ticker['symbol'],
'lastPrice': float(ticker['lastPrice']),
'markPrice': float(ticker['markPrice']),
'indexPrice': float(ticker['indexPrice']),
'fundingRate': float(ticker['fundingRate']) * 100, # Chuyển sang %
'volume24h': float(ticker['volume24h']),
}
return None
====== TEST THỰC TẾ ======
if __name__ == "__main__":
# Lấy giá BTCUSDT
btc_data = get_ticker_price("BTCUSDT")
print(f"\n📊 Giá BTCUSDT hiện tại:")
print(f" Giá giao dịch: ${btc_data['lastPrice']:,.2f}")
print(f" Giá mark: ${btc_data['markPrice']:,.2f}")
print(f" Funding rate: {btc_data['fundingRate']:.4f}%")
print(f" Volume 24h: ${btc_data['volume24h']:,.0f}")
Chiến lược Arbitrage thực chiến
Sau 3 năm backtest và giao dịch thực tế, tôi chia sẻ 3 chiến lược arbitrage phù hợp với người mới. Chiến lược 1 và 2 có thể triển khai ngay với code bên dưới.
Chiến lược 1: Spot-Futures Arbitrage (Độ rủi ro thấp)
Mua BTC trên spot, bán future cùng giá. Chênh lệch = funding rate trả hàng 8 giờ. Chiến lược này gần như không rủi ro vì bạn nắm giữ tài sản thực.
# file: spot_futures_arbitrage.py
class SpotFuturesArbitrage:
def __init__(self, capital=1000):
"""
capital: Vốn đầu tư ban đầu (USD)
"""
self.capital = capital
self.min_profit_threshold = 0.02 # 2% chênh lệch mới vào lệnh
self.positions = {}
def scan_opportunity(self):
"""
Quét cơ hội arbitrage trên tất cả cặp perpetual
"""
tickers = get_perpetual_tickers()
opportunities = []
for ticker in tickers:
symbol = ticker['symbol']
last_price = float(ticker['lastPrice'])
mark_price = float(ticker['markPrice'])
funding_rate = float(ticker['fundingRate']) * 100
# Tính chênh lệch giá
price_diff = abs(last_price - mark_price) / mark_price * 100
# Tính lợi nhuận kỳ vọng 24h (3 funding payments)
expected_daily_profit = funding_rate * 3
# Chỉ quan tâm cặp có funding rate dương (người LONG trả)
if funding_rate > 0 and expected_daily_profit > self.min_profit_threshold:
opportunities.append({
'symbol': symbol,
'last_price': last_price,
'mark_price': mark_price,
'funding_rate': funding_rate,
'expected_daily_profit': expected_daily_profit,
'price_diff': price_diff,
'score': expected_daily_profit / (price_diff + 0.01) # Risk/reward
})
# Sắp xếp theo điểm số
opportunities.sort(key=lambda x: x['score'], reverse=True)
return opportunities
def execute_arbitrage(self, symbol, position_size):
"""
Thực hiện arbitrage:
1. Mua spot (giữ tài sản)
2. Short perpetual (nhận funding)
position_size: Số lượng USD muốn đầu tư
"""
print(f"\n🚀 Thực hiện arbitrage cho {symbol}")
# Trong thực tế, đây là nơi gọi API để đặt lệnh
# Ví dụ:
# spot_order = self.place_spot_order(symbol, "BUY", position_size)
# futures_order = self.place_futures_order(symbol, "SELL", position_size)
self.positions[symbol] = {
'size': position_size,
'entry_time': time.time(),
'expected_profit': self.scan_opportunity()[0]['expected_daily_profit']
}
print(f" ✅ Đã mở position: ${position_size}")
print(f" 📈 Lợi nhuận kỳ vọng/ngày: {self.positions[symbol]['expected_profit']:.2f}%")
return self.positions[symbol]
def get_current_pnl(self, symbol):
"""
Tính PnL hiện tại của một position
"""
if symbol not in self.positions:
return 0
position = self.positions[symbol]
current_price = get_ticker_price(symbol)['lastPrice']
# Tính funding đã nhận được
hours_elapsed = (time.time() - position['entry_time']) / 3600
funding_payments = hours_elapsed / 8 # Mỗi 8 giờ có 1 funding
funding_earned = position['size'] * (position['expected_profit'] / 100) * funding_payments
return funding_earned
====== CHẠY DEMO ======
if __name__ == "__main__":
bot = SpotFuturesArbitrage(capital=5000)
print("🔍 Đang quét cơ hội arbitrage...")
opps = bot.scan_opportunity()
if opps:
print(f"\n🔥 Tìm thấy {len(opps)} cơ hội:")
for i, opp in enumerate(opps[:5]):
print(f" {i+1}. {opp['symbol']}: {opp['expected_daily_profit']:.2f}%")
# Tự động vào lệnh nếu có cơ hội tốt
best = opps[0]
print(f"\n💡 Cơ hội tốt nhất: {best['symbol']}")
print(f" Funding rate: {best['funding_rate']:.4f}%")
print(f" Expected daily: {best['expected_daily_profit']:.2f}%")
Chiến lược 2: Funding Rate Arbitrage (Độ rủi ro trung bình)
Arbitrage giữa funding rate khác nhau trên nhiều sàn. Khi funding rate sàn A cao hơn sàn B, bạn long sàn A và short sàn B.
# file: funding_arbitrage.py
import asyncio
import aiohttp
from datetime import datetime
class FundingArbitrageBot:
def __init__(self):
self.positions = {}
self.trade_history = []
# Ngưỡng cắt lỗ/chốt lời
self.profit_target = 0.5 # 0.5% profit
self.max_loss = 0.2 # 0.2% loss
async def fetch_bybit_funding(self, symbol):
"""Lấy funding rate từ Bybit"""
# Sử dụng code đã viết ở trên
pass
async def fetch_okx_funding(self, symbol):
"""Lấy funding rate từ OKX (sàn khác)"""
# Tương tự như Bybit nhưng đổi endpoint
pass
async def compare_funding_rates(self, symbol):
"""
So sánh funding rate giữa các sàn
Cơ hội: Sàn A có funding = 0.05%, Sàn B = 0.01%
→ Short A (nhận funding), Long B (trả funding)
→ Lợi nhuận = 0.04% mỗi 8 giờ
"""
bybit_funding = await self.fetch_bybit_funding(symbol)
okx_funding = await self.fetch_okx_funding(symbol)
if bybit_funding and okx_funding:
diff = bybit_funding - okx_funding
return {
'symbol': symbol,
'bybit': bybit_funding,
'okx': okx_funding,
'spread': diff,
'annualized': diff * 3 * 365, # 3 funding/ngày
'action': 'short_bybit_long_okx' if diff > 0 else 'short_okx_long_bybit'
}
return None
async def run_scanner(self, symbols):
"""
Quét cơ hội liên tục
symbols: danh sách cặp giao dịch
"""
print(f"🔄 Bắt đầu quét lúc {datetime.now().strftime('%H:%M:%S')}")
tasks = [self.compare_funding_rates(s) for s in symbols]
results = await asyncio.gather(*tasks)
valid_opps = [r for r in results if r and abs(r['spread']) > 0.01]
if valid_opps:
print(f"\n✅ Tìm thấy {len(valid_opps)} cơ hội:")
for opp in sorted(valid_opps, key=lambda x: x['annualized'], reverse=True):
print(f" {opp['symbol']}: Spread {opp['spread']:.4f}% "
f"(Annualized: {opp['annualized']:.1f}%)")
else:
print("❌ Không có cơ hội nào vượt ngưỡng")
return valid_opps
====== CHẠY DEMO ======
async def main():
bot = FundingArbitrageBot()
# Danh sách top perpetual contracts
symbols = ['BTCUSDT', 'ETHUSDT', 'SOLUSDT', 'BNBUSDT', 'XRPUSDT']
# Chạy quét 1 lần
opps = await bot.run_scanner(symbols)
# Hoặc chạy liên tục mỗi phút
# while True:
# await bot.run_scanner(symbols)
# await asyncio.sleep(60)
if __name__ == "__main__":
asyncio.run(main())
Chiến lược 3: Statistical Arbitrage (Độ rủi ro cao)
Sử dụng AI/ML để dự đoán chênh lệch giá sẽ hội tụ về mean. Đây là chiến lược tôi dùng kết hợp HolySheep AI để train model với chi phí cực thấp.
Lỗi thường gặp và cách khắc phục
Qua 3 năm phát triển bot giao dịch, tôi đã gặp và fix hàng trăm lỗi. Đây là những lỗi phổ biến nhất với giải pháp đã test.
Lỗi 1: "Invalid signature" — Lỗi xác thực phổ biến nhất
❌ LỖI THƯỜNG GẶP
Request URL: https://api-testnet.bybit.com/v5/market/tickers?api_key=xxx×tamp=xxx&sign=abc
Response: {"retCode":10003,"retMsg":"invalid request"}
✅ CÁCH KHẮC PHỤC
Vấn đề 1: Thứ tự tham số không đúng
Bybit yêu cầu tham số phải được sắp xếp theo alphabetical order
Code SAI:
def generate_signature_wrong(params, secret):
param_str = urlencode(params) # Thứ tự ngẫu nhiên
return hmac.new(bytes(secret, 'utf-8'),
bytes(param_str, 'utf-8'),
hashlib.sha256).hexdigest()
Code ĐÚNG:
def generate_signature_correct(params, secret):
sorted_params = sorted(params.items()) # Sắp xếp A-Z
param_str = urlencode(sorted_params)
return hmac.new(bytes(secret, 'utf-8'),
bytes(param_str, 'utf-8'),
hashlib.sha256).hexdigest()
Vấn đề 2: Timestamp không đúng timezone
Phải dùng UTC timestamp
import datetime
def get_correct_timestamp():
# ❌ Sai: Dùng local time
# return str(int(time.time() * 1000))
# ✅ Đúng: Dùng UTC và chuyển sang milliseconds
utc_now = datetime.datetime.now(datetime.timezone.utc)
timestamp = int(utc_now.timestamp() * 1000)
return str(timestamp)
Vấn đề 3: Recv window quá ngắn
Mạng chậm → request bị reject vì hết hạn
def send_request_with_recv_window(method, endpoint, params=None):
url = f"{TESTNET_BASE_URL}{endpoint}"
if params is None:
params = {}
params['api_key'] = BYBIT_API_KEY
params['timestamp'] = get_correct_timestamp()
params['recv_window'] = 60000 # ⬅️ TĂNG LÊN 60 giây (mặc định chỉ 5s)
param_str = urlencode(sorted(params.items()))
params['sign'] = hmac.new(
bytes(BYBIT_API_SECRET, 'utf-8'),
bytes(param_str, 'utf-8'),
hashlib.sha256
).hexdigest()
response = requests.get(url, params=params, timeout=30)
return response.json()
Lỗi 2: "Rate limit exceeded" — Quá nhiều request
❌ LỖI THƯỜNG GẶP
Response: {"retCode":10029,"retMsg":"Too many requests"}
Giới hạn Bybit API:
- Public endpoints: 600 requests/10 giây
- Private endpoints: 600 requests/10 giây
- Order placement: 50 requests/giây
✅ CÁCH KHẮC PHỤC
import time
from functools import wraps
from collections import defaultdict
class RateLimiter:
def __init__(self, max_calls=50, period=10):
self.max_calls = max_calls
self.period = period
self.calls = defaultdict(list)
def wait_if_needed(self, endpoint_type="private"):
"""Chờ nếu vượt rate limit"""
now = time.time()
recent_calls = self.calls[endpoint_type]
# Xóa các request cũ (quá period giây)
self.calls[endpoint_type] = [
t for t in recent_calls
if now - t < self.period
]
if len(self.calls[endpoint_type]) >= self.max_calls:
# Tính thời gian chờ
oldest = min(self.calls[endpoint_type])
wait_time = self.period - (now - oldest) + 0.1
print(f"⏳ Rate limit: Chờ {wait_time:.1f}s...")
time.sleep(wait_time)
self.calls[endpoint_type].append(now)
Cách sử dụng
limiter = RateLimiter(max_calls=50, period=10)
def throttled_request(method, endpoint, params=None):
limiter.wait_if_needed("private" if "api_key" in str(params) else "public")
# Retry logic cho lỗi rate limit
for attempt in range(3):
response = send_request(method, endpoint, params)
if response.get('retCode') == 10029:
wait = 2 ** attempt # Exponential backoff: 1s, 2s, 4s
print(f"⚠️ Rate limit hit, retry sau {wait}s...")
time.sleep(wait)
continue
return response
return {"retCode": -1, "retMsg": "Max retries exceeded"}
Ngoài ra, BATCH các request thay vì gọi riêng lẻ
def get_multiple_prices(symbols):
"""Lấy giá nhiều cặp trong 1 request thay vì nhiều request"""
params = {
'category': 'linear',
'limit': 200 # Max 200 kết quả mỗi request
}
result = send_request("GET", "/v5/market/tickers", params)
if result.get('retCode') == 0:
all_tickers = result['result']['list']
return {
t['symbol']: float(t['lastPrice'])
for t in all_tickers
if t['symbol'] in symbols
}
return {}
Lỗi 3: "Position not found" khi đóng lệnh
❌ LỖI THƯỜNG GẶP
Khi bot restart, nó quên position đang mở
Hoặc đóng lệnh trùng lặp
✅ CÁCH KHẮC PHỤC
class PositionManager:
def __init__(self, db_path="positions.db"):
self.positions = {} # symbol -> position data
self.db_path = db_path
self.load_positions()
def load_positions(self):
"""Load positions từ file khi khởi động"""
import json
try:
with open(self.db_path, 'r') as f:
saved = json.load(f)
self.positions = saved
print(f"📂 Đã load {len(saved)} positions từ file")
except FileNotFoundError:
print("📂 Chưa có file, bắt đầu mới")
def save_positions(self):
"""Lưu positions ra file"""
import json
with open(self.db_path, 'w') as f:
json.dump(self.positions, f, indent=2)
def sync_with_exchange(self):
"""Đồng bộ với positions thực trên sàn"""
# Gọi API lấy danh sách positions đang mở
result = send_request("GET", "/v5/position/list", {
'category': 'linear'
})
if result.get('retCode') == 0:
exchange_positions = result['result']['list']
# Tạo dict từ positions trên sàn
active = {
p['symbol']: {
'size': float(p['size']),
'side': p['side'],
'entry_price': float(p['avgPrice']),
'unrealized_pnl': float(p['unrealisedPnl'])
}
for p in exchange_positions
if float(p['size']) != 0
}
# Cập nhật local positions
self.positions = active
self.save_positions()
print(f"🔄 Đã đồng bộ: {len(active)} positions đang mở")
return active
return {}
def close_position(self, symbol, size=None):
"""Đóng position an toàn"""
if symbol not in self.positions:
print(f"⚠️ Không tìm thấy position {symbol}")
return None
position = self.positions[symbol]
# Xác định size đóng
close_size = size if size else abs(position['size'])
# Xác định direction (ngược với position hiện tại)
side = "Sell" if position['side'] == "Buy" else "Buy"
# Gửi lệnh đóng
result = send_request("POST", "/v5/order/create", {
'category': 'linear',
'symbol': symbol,
'side': side,
'orderType': "Market",
'qty': str(close_size)
})
if result.get('retCode') == 0:
del self.positions[symbol]
self.save_positions()
print(f"✅ Đã đóng position {symbol}")
return result
print(f"❌ Lỗi đóng position: {result.get('retMsg')}")
return result
def is_position_open(self, symbol):
"""Kiểm tra position có đang mở không"""
return symbol in self.positions
Sử dụng
manager = PositionManager()
Khi khởi động bot
manager.sync_with_exchange()
Khi cần đóng lệnh
if manager.is_position_open("BTCUSDT"):
manager.close_position("BTCUSDT")
Lỗi 4: Xử lý số thập phân sai dẫn đến order bị reject
❌ LỖI THƯỜNG GẶP
Response: {"retCode":10001,"retMsg":"invalid qty"}
Nguyên nhân: Mỗi cặp giao dịch có độ chính xác khác nhau
BTCUSDT: qty phải là số nguyên (không có decimal)
ETHUSDT: qty có tối đa 3 chữ số thập phân
✅ CÁCH KHẮC PHỤC
def get_symbol_info(symbol):
"""Lấy thông tin precision của symbol"""
result = send_request("GET", "/v5/market/instruments-info", {
'category': 'linear',
'symbol': symbol
})
if result.get('retCode') == 0:
info = result['result']['list'][0]
return {
'symbol': symbol,
'price_precision': int(info['pricePrecision']), # VD: 2 = 2 chữ số thập phân
'qty_precision': int(info['qtyStep']), # VD: 0.001
'min_qty': float(info['minOrderQty']),
'max_qty': float(info['maxOrderQty'])
}
return None
def round_to_step(value, step):
"""Làm tròn số theo bước nhảy quy định"""
return round(value / step) * step
def validate_order_qty(symbol, qty):
"""Validate và format số lượng order"""
info = get_symbol_info(symbol)
if not info:
print(f"❌ Không lấy được thông tin {symbol}")
return None
# Kiểm tra min/max
if qty < info['min_qty']:
print(f"⚠️ Qty {qty} nhỏ hơn min {info['min_qty']}")