Chào bạn đọc! Mình là Minh, chuyên gia về dữ liệu thị trường crypto và đã làm việc với API của hơn 10 sàn giao dịch lớn. Hôm nay mình sẽ chia sẻ chi tiết cách kết nối Bybit Futures API để lấy dữ liệu đối soát giao dịch (tick-by-tick trades) và khôi phục sổ lệnh (order book reconstruction) — hai nguồn dữ liệu quan trọng nhất cho việc phân tích kỹ thuật và xây dựng bot giao dịch.
Lưu ý quan trọng: Bài viết này tập trung vào public endpoints — không cần API key cũng có thể thử nghiệm được. Nếu bạn cần xử lý dữ liệu AI nhanh hơn (phân tích pattern, dự đoán xu hướng), mình khuyên dùng HolySheep AI với độ trễ dưới 50ms và chi phí chỉ từ $0.42/1M tokens.
Mục Lục
- Giới thiệu về dữ liệu Bybit
- Thiết lập môi trường
- Lấy dữ liệu lịch sử giao dịch
- Reconstruction Order Book
- Code mẫu hoàn chỉnh
- Lỗi thường gặp và cách khắc phục
- So sánh giải pháp
- Kết luận
1. Giới Thiệu Về Dữ Liệu Bybit Futures
Bybit là một trong những sàn perpetual futures lớn nhất thế giới với khối lượng giao dịch hàng tỷ USD mỗi ngày. API của họ cung cấp hai loại dữ liệu chính:
- Trade Data (逐笔成交): Dữ liệu từng giao dịch riêng lẻ — ai bán, ai mua, giá bao nhiêu, khối lượng bao nhiêu. Đây là dữ liệu "thô" nhất của thị trường.
- Order Book (订单簿): Sổ lệnh chờ thực hiện — cho biết có bao nhiêu lệnh mua/bán ở mỗi mức giá. Dùng để phân tích depth of market.
🤔 Gợi ý ảnh chụp màn hình: Chụp giao diện Bybit Futures, highlight vùng "Recent Trades" và "Order Book" để người đọc hình dung.
2. Thiết Lập Môi Trường Làm Việc
Trước khi bắt đầu, bạn cần cài đặt Python và thư viện cần thiết. Đây là bước quan trọng nhất cho người mới — đừng bỏ qua nhé!
2.1 Cài Đặt Python
Nếu bạn chưa có Python, hãy tải tại python.org. Mình khuyên dùng Python 3.9 trở lên.
2.2 Tạo Virtual Environment (Khuyến nghị)
# Tạo thư mục dự án
mkdir bybit_api_tutorial
cd bybit_api_tutorial
Tạo môi trường ảo (tránh xung đột thư viện)
python -m venv venv
Kích hoạt môi trường ảo
Windows:
venv\Scripts\activate
Mac/Linux:
source venv/bin/activate
Cài đặt các thư viện cần thiết
pip install requests pandas websocket-client asyncio aiohttp2.3 Cấu Trúc Thư Mục Dự Án
bybit_api_tutorial/
├── config.py # Cấu hình API
├── public_api.py # Các hàm lấy dữ liệu public
├── websocket_api.py # Kết nối real-time
├── orderbook.py # Xử lý order book
├── main.py # Chương trình chính
├── requirements.txt # Danh sách thư viện
└── data/ # Thư mục lưu dữ liệu
├── trades/
└── orderbook/3. Lấy Dữ Liệu Lịch Sử Giao Dịch (Trade History)
Đây là phần quan trọng nhất — mình sẽ hướng dẫn từng dòng code để bạn hiểu rõ.
3.1 Tạo File Cấu Hình
# config.py
import os
Cấu hình Bybit API (Public - không cần key)
BYBIT_BASE_URL = "https://api.bybit.com"
SYMBOL = "BTCUSDT" # Cặp giao dịch
CATEGORY = "linear" # linear = USDT perpetual futures
Cấu hình HolySheep AI (nếu cần xử lý AI)
Đăng ký tại: https://www.holysheep.ai/register
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
Thư mục lưu dữ liệu
DATA_DIR = "data"
os.makedirs(f"{DATA_DIR}/trades", exist_ok=True)
os.makedirs(f"{DATA_DIR}/orderbook", exist_ok=True)3.2 Hàm Lấy Dữ Liệu Trade History
# public_api.py
import requests
import pandas as pd
import time
from datetime import datetime
import config
def get_recent_trades(symbol: str = config.SYMBOL, limit: int = 100):
"""
Lấy dữ liệu giao dịch gần đây từ Bybit
Parameters:
symbol: Cặp giao dịch (VD: BTCUSDT)
limit: Số lượng giao dịch cần lấy (tối đa 1000)
Returns:
DataFrame chứa dữ liệu giao dịch
"""
endpoint = f"{config.BYBIT_BASE_URL}/v5/market/recent-trade"
params = {
"category": config.CATEGORY,
"symbol": symbol,
"limit": limit
}
try:
response = requests.get(endpoint, params=params, timeout=10)
response.raise_for_status()
data = response.json()
if data["retCode"] == 0:
trades = data["result"]["list"]
# Chuyển đổi sang DataFrame để dễ xử lý
df = pd.DataFrame(trades)
# Đổi tên cột cho dễ hiểu
df.columns = [
"trade_id", # ID giao dịch
"trade_time_ms", # Thời gian (milliseconds)
"symbol", # Cặp giao dịch
"price", # Giá giao dịch
"size", # Khối lượng
"side", # Hướng: Buy hoặc Sell
]
# Chuyển đổi thời gian
df["trade_time"] = pd.to_datetime(
df["trade_time_ms"].astype(float), unit="ms"
)
# Sắp xếp theo thời gian tăng dần
df = df.sort_values("trade_time").reset_index(drop=True)
# Chuyển đổi kiểu dữ liệu
df["price"] = df["price"].astype(float)
df["size"] = df["size"].astype(float)
print(f"✅ Đã lấy {len(df)} giao dịch gần nhất của {symbol}")
return df
else:
print(f"❌ Lỗi API: {data['retMsg']}")
return None
except requests.exceptions.RequestException as e:
print(f"❌ Lỗi kết nối: {e}")
return None
def get_historical_trades(symbol: str, start_time: int, end_time: int):
"""
Lấy dữ liệu giao dịch theo khoảng thời gian
Parameters:
symbol: Cặp giao dịch
start_time: Thời gian bắt đầu (milliseconds)
end_time: Thời gian kết thúc (milliseconds)
"""
endpoint = f"{config.BYBIT_BASE_URL}/v5/market/history-trade"
all_trades = []
current_time = start_time
while current_time < end_time:
params = {
"category": config.CATEGORY,
"symbol": symbol,
"startTime": current_time,
"limit": 1000
}
response = requests.get(endpoint, params=params, timeout=10)
data = response.json()
if data["retCode"] == 0:
trades = data["result"]["list"]
all_trades.extend(trades)
if len(trades) < 1000:
break
# Lấy thời gian của giao dịch cuối cùng + 1ms
current_time = int(trades[-1]["tradeTime"]) + 1
else:
print(f"❌ Lỗi: {data['retMsg']}")
break
print(f"✅ Đã lấy tổng cộng {len(all_trades)} giao dịch")
return all_trades
Test nhanh
if __name__ == "__main__":
df = get_recent_trades(limit=10)
if df is not None:
print(df.head())🤔 Gợi ý ảnh chụp màn hình: Chụp output khi chạy script, highlight các cột dữ liệu quan trọng.
4. Reconstruction Order Book (重建订单簿)
Đây là kỹ thuật nâng cao — ghép các giao dịch riêng lẻ thành snapshot của sổ lệnh tại bất kỳ thời điểm nào. Rất hữu ích cho backtesting!
4.1 Lấy Order Book Snapshot
# orderbook.py
import requests
import pandas as pd
from typing import Dict, List, Tuple
from dataclasses import dataclass
from datetime import datetime
@dataclass
class OrderBookLevel:
"""Một mức giá trong sổ lệnh"""
price: float
size: float
class OrderBook:
"""Class quản lý order book"""
def __init__(self):
self.bids: Dict[float, float] = {} # {price: size} cho lệnh mua
self.asks: Dict[float, float] = {} # {price: size} cho lệnh bán
def update_bid(self, price: float, size: float):
"""Cập nhật lệnh mua"""
if size == 0:
self.bids.pop(price, None)
else:
self.bids[price] = size
def update_ask(self, price: float, size: float):
"""Cập nhật lệnh bán"""
if size == 0:
self.asks.pop(price, None)
else:
self.asks[price] = size
def get_best_bid(self) -> Tuple[float, float]:
"""Lấy giá mua tốt nhất (bid cao nhất)"""
if not self.bids:
return 0, 0
best_price = max(self.bids.keys())
return best_price, self.bids[best_price]
def get_best_ask(self) -> Tuple[float, float]:
"""Lấy giá bán tốt nhất (ask thấp nhất)"""
if not self.asks:
return 0, 0
best_price = min(self.asks.keys())
return best_price, self.asks[best_price]
def get_spread(self) -> float:
"""Tính spread (chênh lệch giá mua-bán)"""
best_bid, _ = self.get_best_bid()
best_ask, _ = self.get_best_ask()
if best_bid and best_ask:
return best_ask - best_bid
return 0
def get_mid_price(self) -> float:
"""Giá giữa thị trường"""
best_bid, _ = self.get_best_bid()
best_ask, _ = self.get_best_ask()
if best_bid and best_ask:
return (best_bid + best_ask) / 2
return 0
def get_depth(self, levels: int = 10) -> pd.DataFrame:
"""Lấy độ sâu thị trường"""
# Lấy N mức giá tốt nhất
top_bids = sorted(self.bids.items(), reverse=True)[:levels]
top_asks = sorted(self.asks.items())[:levels]
# Tạo DataFrame
bids_df = pd.DataFrame(top_bids, columns=["bid_price", "bid_size"])
asks_df = pd.DataFrame(top_asks, columns=["ask_price", "ask_size"])
# Thêm cột cumulative
bids_df["bid_cumulative"] = bids_df["bid_size"].cumsum()
asks_df["ask_cumulative"] = asks_df["ask_size"].cumsum()
return pd.concat([bids_df, asks_df], axis=1)
def to_dict(self) -> dict:
"""Chuyển sang dictionary để lưu"""
return {
"timestamp": datetime.now().isoformat(),
"bids": dict(sorted(self.bids.items(), reverse=True)),
"asks": dict(sorted(self.asks.items())),
"best_bid": self.get_best_bid()[0],
"best_ask": self.get_best_ask()[0],
"spread": self.get_spread(),
"mid_price": self.get_mid_price()
}
def get_order_book_snapshot(symbol: str, limit: int = 50) -> OrderBook:
"""
Lấy snapshot order book hiện tại từ Bybit
Parameters:
symbol: Cặp giao dịch
limit: Số lượng mức giá (tối đa 200)
"""
endpoint = f"{config.BYBIT_BASE_URL}/v5/market/orderbook"
params = {
"category": config.CATEGORY,
"symbol": symbol,
"limit": limit
}
try:
response = requests.get(endpoint, params=params, timeout=10)
response.raise_for_status()
data = response.json()
if data["retCode"] == 0:
result = data["result"]
ob = OrderBook()
# Cập nhật bids
for price, size in zip(result["b"], result["bs"]):
ob.update_bid(float(price), float(size))
# Cập nhật asks
for price, size in zip(result["a"], result["as"]):
ob.update_ask(float(price), float(size))
print(f"✅ Đã lấy order book: Spread = {ob.get_spread():.2f}")
return ob
else:
print(f"❌ Lỗi API: {data['retMsg']}")
return None
except Exception as e:
print(f"❌ Lỗi: {e}")
return None4.2 Reconstruction Từ Trade Data
Đây là phần "magic" — mình sẽ hướng dẫn cách reconstruct order book từ các giao dịch. Kỹ thuật này dùng khi bạn có dữ liệu trade history và muốn biết trạng thái sổ lệnh tại thời điểm bất kỳ.
def reconstruct_order_book_from_trades(
trades: List[dict],
initial_book: OrderBook = None
) -> OrderBook:
"""
Khôi phục order book từ danh sách giao dịch
Nguyên lý: Mỗi giao dịch sẽ thay đổi order book:
- Nếu trade có side="Buy" → lệnh market mua → khớp với ask
- Nếu trade có side="Sell" → lệnh market bán → khớp với bid
Parameters:
trades: Danh sách giao dịch (đã sắp xếp theo thời gian)
initial_book: Order book ban đầu (nếu có)
Returns:
OrderBook đã được cập nhật
"""
if initial_book is None:
ob = OrderBook()
else:
ob = initial_book
for trade in trades:
price = float(trade["price"])
size = float(trade["size"])
side = trade["side"].lower()
# Xác định order book level bị ảnh hưởng
# Quan trọng: Đây là DEMO, thực tế cần xử lý phức tạp hơn
if side == "buy":
# Buyer taker → khớp với ask (giảm ask size)
if price in ob.asks:
new_size = ob.asks[price] - size
ob.update_ask(price, max(0, new_size))
else:
# Tạo bid mới (limit buy order)
ob.update_bid(price, size)
else: # side == "sell"
# Seller taker → khớp với bid (giảm bid size)
if price in ob.bids:
new_size = ob.bids[price] - size
ob.update_bid(price, max(0, new_size))
else:
# Tạo ask mới (limit sell order)
ob.update_ask(price, size)
return ob
def analyze_trade_flow(trades_df: pd.DataFrame) -> dict:
"""
Phân tích dòng tiền từ dữ liệu giao dịch
Returns thông tin quan trọng cho trading:
- Buy/Sell ratio
- Volume-weighted average price (VWAP)
- Tổng khối lượng buy/sell
"""
trades_df["value"] = trades_df["price"] * trades_df["size"]
buy_trades = trades_df[trades_df["side"] == "Buy"]
sell_trades = trades_df[trades_df["side"] == "Sell"]
analysis = {
"total_trades": len(trades_df),
"buy_count": len(buy_trades),
"sell_count": len(sell_trades),
"buy_ratio": len(buy_trades) / len(trades_df) if len(trades_df) > 0 else 0,
"buy_volume": buy_trades["size"].sum(),
"sell_volume": sell_trades["size"].sum(),
"total_volume": trades_df["size"].sum(),
"volume_buy_ratio": buy_trades["size"].sum() / trades_df["size"].sum() if trades_df["size"].sum() > 0 else 0,
"vwap": trades_df["value"].sum() / trades_df["size"].sum() if trades_df["size"].sum() > 0 else 0,
"price_range": {
"min": trades_df["price"].min(),
"max": trades_df["price"].max(),
"open": trades_df["price"].iloc[0],
"close": trades_df["price"].iloc[-1]
}
}
return analysis⚠️ Lưu ý quan trọng: Việc reconstruction order book từ trade data chỉ cho kết quả ước lượng. Để chính xác 100%, bạn cần có full order book history (dữ liệu mà sàn không public). Tuy nhiên, với mục đích backtesting, độ chính xác này là chấp nhận được.
5. Chương Trình Hoàn Chỉnh
Đây là script main.py kết hợp tất cả các phần trên — bạn có thể chạy ngay!
# main.py
import time
import json
import pandas as pd
from datetime import datetime, timedelta
import config
from public_api import get_recent_trades, get_historical_trades
from orderbook import get_order_book_snapshot, OrderBook, reconstruct_order_book_from_trades, analyze_trade_flow
def main():
"""Chương trình chính - Demo đầy đủ chức năng"""
print("=" * 60)
print("BYBIT FUTURES API - DEMO")
print("=" * 60)
# 1. Lấy dữ liệu trade gần đây
print("\n📊 Bước 1: Lấy dữ liệu giao dịch gần đây...")
trades_df = get_recent_trades(symbol="BTCUSDT", limit=100)
if trades_df is not None:
# Phân tích dòng tiền
analysis = analyze_trade_flow(trades_df)
print(f"\n📈 Phân tích:")
print(f" - Tổng giao dịch: {analysis['total_trades']}")
print(f" - Buy/Sell ratio: {analysis['buy_ratio']:.2%}")
print(f" - Khối lượng mua: {analysis['buy_volume']:.4f} BTC")
print(f" - Khối lượng bán: {analysis['sell_volume']:.4f} BTC")
print(f" - VWAP: ${analysis['vwap']:,.2f}")
print(f" - Biên độ giá: ${analysis['price_range']['min']:,.2f} - ${analysis['price_range']['max']:,.2f}")
# Lưu vào file
filename = f"{config.DATA_DIR}/trades/{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
trades_df.to_csv(filename, index=False)
print(f"💾 Đã lưu: {filename}")
# 2. Lấy order book snapshot
print("\n📋 Bước 2: Lấy Order Book...")
ob = get_order_book_snapshot(symbol="BTCUSDT", limit=20)
if ob is not None:
best_bid, bid_size = ob.get_best_bid()
best_ask, ask_size = ob.get_best_ask()
print(f"\n💹 Sổ lệnh BTCUSDT:")
print(f" Bid: ${best_bid:,.2f} (Size: {bid_size:.4f})")
print(f" Ask: ${best_ask:,.2f} (Size: {ask_size:.4f})")
print(f" Spread: ${ob.get_spread():.2f}")
print(f" Mid Price: ${ob.get_mid_price():,.2f}")
# Hiển thị depth
print("\n📊 Độ sâu thị trường (top 5):")
depth = ob.get_depth(levels=5)
print(depth.to_string(index=False))
# Lưu order book
ob_data = ob.to_dict()
filename = f"{config.DATA_DIR}/orderbook/{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
with open(filename, "w") as f:
json.dump(ob_data, f, indent=2)
print(f"💾 Đã lưu: {filename}")
# 3. Demo reconstruction (tùy chọn)
print("\n🔄 Bước 3: Demo Order Book Reconstruction...")
print("(Sử dụng 20 giao dịch đầu tiên để demo)")
if trades_df is not None:
# Lấy 20 giao dịch đầu
sample_trades = trades_df.head(20).to_dict("records")
# Bắt đầu với order book hiện tại
reconstructed = reconstruct_order_book_from_trades(sample_trades)
print(f"\n📊 Sau khi reconstruction:")
print(f" Số lượng bid levels: {len(reconstructed.bids)}")
print(f" Số lượng ask levels: {len(reconstructed.asks)}")
print("\n" + "=" * 60)
print("✅ Hoàn thành!")
print("=" * 60)
if __name__ == "__main__":
main()Cách chạy:
# Kích hoạt môi trường ảo (nếu chưa làm)
Windows:
venv\Scripts\activate
Linux/Mac:
source venv/bin/activate
Chạy chương trình
python main.py🤔 Gợi ý ảnh chụp màn hình: Chụp terminal output khi script chạy thành công, highlight các số liệu quan trọng.
6. Lỗi Thường Gặp Và Cách Khắc Phục
Qua quá trình làm việc với API, mình đã gặp nhiều lỗi phổ biến. Dưới đây là tổng hợp cách xử lý:
| Lỗi | Nguyên nhân | Cách khắc phục |
|---|---|---|
| retCode: 10001 - Permission denied | Dùng private endpoint mà không có API key hoặc key không có quyền |
|
| requests.exceptions.ReadTimeout | Server phản hồi chậm hoặc mạng không ổn định |
|
| JSON decode error | API trả về HTML (thường là trang lỗi 403/429) hoặc rate limit |
|
| Order book reconstruction không chính xác | Thiếu dữ liệu order book ban đầu hoặc trades không đầy đủ |
|
Mẹo Tối Ưu Hóa Khi Làm Việc Với Dữ Liệu Lớn
# Tip 1: Sử dụng async để lấy nhiều symbol cùng lúc
import asyncio
import aiohttp
async def fetch_multiple_symbols(symbols: list):
"""Lấy dữ liệu nhiều symbol song song"""
async with aiohttp.ClientSession() as session:
tasks = []
for symbol in symbols:
url = f"https://api.bybit.com/v5/market/recent-trade?category=linear&symbol={symbol}&limit=50"
tasks.append(fetch_trade(session, url))
results = await asyncio.gather(*tasks)
return results
Tip 2: Lưu data vào SQLite để truy vấn nhanh
import sqlite3
def save_to_sqlite(trades_df, table_name="trades"):
conn = sqlite3.connect("trades.db")
trades_df.to_sql(table_name, conn, if_exists="append", index=False)
conn.close()
Tip 3: Sử dụng Pandas efficiently
Đọc chunk thay vì toàn bộ
for chunk in pd.read_csv("large_file.csv", chunksize=10000):
process(chunk) # Xử lý từng phần7. So Sánh Giải Pháp - Bảng Giá
Nếu bạn cần xử lý dữ liệu