Hướng dẫn tải dữ liệu lịch sử độ sâu L2 hợp đồng vĩnh viễn BTC và phát lại tăng dần với Pandas

Tôi là Quang, lập trình viên độc lập với 7 năm xây dựng hệ thống cho SME tài chính tại TP.HCM. Tháng 4 vừa qua, ngay trước đợt go-live hệ thống RAG nội bộ của một quỹ phòng hộ crypto ở Singapore, họ giao cho tôi nhiệm vụ gấp: xây engine backtest cho chiến lược market-making trên hợp đồng vĩnh viễn BTC. Yêu cầu cứng là phải có dữ liệu độ sâu L2 (Level 2) từ 12 tháng trở lại, tần suất tick-by-tick, khả năng replay lại nhiều lần với tốc độ nén thời gian. Bài viết này là tổng hợp những gì tôi đã đổ mồ hôi trong 3 tuần đó, kèm theo mã nguồn thật tôi đã chạy thành công trên máy cá nhân (i7-12700H, 32GB RAM).

1. Vì sao dữ liệu L2 lại là "vũ khí bí mật"?

Khác với candle OHLCV thông thường chỉ cho biết giá đóng cửa, dữ liệu L2 phản ánh toàn bộ sổ lệnh: 400 cấp giá mua/bán mỗi bên trên OKX, hoặc 5000 cấp trên Binance Futures. Với nhà market-making, đây là thứ không thể thiếu để mô phỏng trượt giá (slippage), tính thanh khoản và dự đoán lệnh lớn sắp xuất hiện. Nguồn dữ liệu lịch sử L2 tốt nhất hiện tại gồm:

• Tardis.dev: lưu trữ tick-by-tick từ nhiều sàn, gói cá nhân $50/tháng.
• Kaiko: doanh nghiệp, giá cao, thường từ $1.000/tháng trở lên.
• REST + WebSocket trực tiếp từ sàn: miễn phí nhưng phải tự thu thập và xử lý.

Trong bài này tôi chọn hướng thứ 3 - miễn phí, kiểm soát 100%, phù hợp indie developer như tôi.

2. Chuẩn bị môi trường

pip install pandas==2.2.2 requests==2.31.0 websocket-client==1.7.0 pyarrow==15.0.0 openai==1.30.1 pyarrow tqdm

Tôi khuyên dùng môi trường ảo venv riêng để tránh xung đột với các dự án khác. Toàn bộ dữ liệu sẽ được lưu dưới định dạng Parquet - nén gấp 8 lần CSV và Pandas đọc cực nhanh.

3. Tải snapshot L2 từ REST API

Đầu tiên cần một hàm để kéo snapshot L2 độ sâu 400 cấp từ OKX - sàn có tỷ lệ phí thấp và lịch sử API ổn định nhất:

import requests
import pandas as pd
import time
import os
from datetime import datetime, timezone

BASE_REST = "https://www.okx.com"
INST_ID = "BTC-USDT-SWAP"  # Hop dong vinh vien BTC USDT
SAVE_DIR = "./data/l2_snapshots"
os.makedirs(SAVE_DIR, exist_ok=True)

def fetch_l2_snapshot(inst_id=INST_ID, depth=400):
    """Tai mot snapshot L2 hien tai tu OKX."""
    url = f"{BASE_REST}/api/v5/market/books"
    params = {"instId": inst_id, "sz": depth}
    r = requests.get(url, params=params, timeout=10)
    r.raise_for_status()
    data = r.json()["data"][0]
    bids = pd.DataFrame(data["bids"], columns=["price", "qty", "orders", "liq"])
    asks = pd.DataFrame(data["asks"], columns=["price", "qty", "orders", "liq"])
    bids["side"] = "bid"
    asks["side"] = "ask"
    df = pd.concat([bids, asks], ignore_index=True)
    df["ts"] = pd.to_datetime(int(data["ts"]), unit="ms", utc=True)
    df["inst_id"] = inst_id
    return df

Vi du: tai 1 snapshot va luu
df = fetch_l2_snapshot()
print(f"Da tai {len(df)} cap gia, tong {df['qty'].astype(float).sum():.4f} BTC")
Luu parquet, ten file co timestamp de tranh trung
fname = f"{SAVE_DIR}/okx_{datetime.now(timezone.utc):%Y%m%d_%H%M%S}.parquet"
df.to_parquet(fname, compression="zstd")
print(f"Da luu: {fname}")

Kết quả chạy thực tế trên máy tôi: tải 1 snapshot mất 180-220ms, file Parquet nặng 38KB. Tôi đã thu thập 21.000 snapshot trong 30 ngày, tổng 780MB - hoàn toàn vừa ổ SSD 1TB.

4. Kết nối WebSocket và lưu trữ incremental

Snapshot một lần chỉ chụp "bức tranh" tại một thời điểm. Để có tick-by-tick thật sự, cần WebSocket. Đoạn mã dưới đây ghi nhận mọi thay đổi L2 và ghi vào file Parquet theo lô 500 bản ghi để tránh I/O quá tải:

import websocket
import json
import threading

BATCH = []
BATCH_SIZE = 500
BATCH_FILE = f"{SAVE_DIR}/ws_stream.parquet"

def on_message(ws, msg):
    global BATCH
    payload = json.loads(msg)
    if "data" not in payload:
        return
    for d in payload["data"]:
        for side, levels in [("bid", d["bids"]), ("ask", d["asks"])]:
            for price, qty, orders, liq in levels:
                BATCH.append({
                    "ts": pd.to_datetime(int(d["ts"]), unit="ms", utc=True),
                    "side": side,
                    "price": float(price),
                    "qty": float(qty),
                    "orders": int(orders),
                    "liq": float(liq) if liq else 0.0,
                })
    if len(BATCH) >= BATCH_SIZE:
        flush_batch()

def flush_batch():
    global BATCH
    if not BATCH:
        return
    new_df = pd.DataFrame(BATCH)
    if os.path.exists(BATCH_FILE):
        old = pd.read_parquet(BATCH_FILE)
        new_df = pd.concat([old, new_df], ignore_index=True)
    new_df.to_parquet(BATCH_FILE, compression="zstd")
    BATCH = []
    print(f"[{datetime.now()}] Da flush, tong ban ghi: {len(new_df)}")

def on_open(ws):
    sub = {"op": "subscribe", "args": [{"channel": "books-l2-tbt", "instId": INST_ID}]}
    ws.send(json.dumps(sub))
    print("WebSocket da mo")

def run_ws():
    ws = websocket.WebSocketApp(
        "wss://ws.okx.com:8443/ws/v5/public",
        on_message=on_message,
        on_open=on_open,
    )
    ws.run_forever()

Chay nen (nen 12h lien tuc)
threading.Thread(target=run_ws, daemon=True).start()
print("Dang thu thap du lieu... nhan Ctrl+C de dung")
try:
    while True:
        time.sleep(60)
        flush_batch()  # flush dinh ky
except KeyboardInterrupt:
    flush_batch()
    print("Da dung va luu du lieu cuoi cung")

Sau 12 giờ chạy liên tục tôi nhận được khoảng 2,1 triệu bản ghi, kích thước 410MB. Tốc độ ghi trung bình 48 bản ghi/giây, hoàn toàn nằm trong khả năng xử lý của Pandas.

5. Phát lại tăng dần với Pandas

Đây là phần hay nhất: từ file Parquet, tôi xây dựng iterator phát lại dữ liệu theo từng tick, hỗ trợ tua nhanh thời gian để backtest. Ý tưởng là thay vì lặp từng dòng trong dataframe khổng lồ (rất chậm), tôi dùng generator kết hợp itertuples:

import pyarrow.parquet as pq

class L2Replay:
    """Phat lai du lieu L2 theo tick, ho tro tua nhanh thoi gian."""
    def __init__(self, parquet_path, speed_factor=100):
        self.path = parquet_path
        self.speed = speed_factor  # 100 = nhanh gap 100 lan
        self._table = pq.read_table(parquet_path)
        self._df = self._table.to_pandas().sort_values("ts").reset_index(drop=True)

    def __iter__(self):
        prev_ts = None
        for row in self._df.itertuples(index=False):
            cur_ts = row.ts
            if prev_ts is not None:
                delta = (cur_ts - prev_ts).total_seconds() / self.speed
                if delta > 0:
                    time.sleep(delta)
            prev_ts = cur_ts
            yield {
                "ts": cur_ts,
                "side": row.side,
                "price": row.price,
                "qty": row.qty,
            }

    def stats(self):
        df = self._df
        print(f"Tong ban ghi: {len(df):,}")
        print(f"Khoang thoi gian: {df['ts'].min()} den {df['ts'].max()}")
        print(f"Trung binh {len(df) / max(1, (df['ts'].max() - df['ts'].min()).total_seconds()):.1f} ban ghi/giay")

Su dung
replay = L2Replay("data/l2_snapshots/ws_stream.parquet", speed_factor=500)
replay.stats()
for i, tick in enumerate(replay):
    if i % 100000 == 0:
        print(f"Da xu ly {i:,} ticks...")
    if i >= 1000000:  # gioi han test
        break

Trong thử nghiệm của tôi, replay 1 triệu tick với speed_factor=500 mất 38 giây trên máy tính cá nhân - tức gần 27.000 tick/giây. Đủ nhanh để chạy backtest cho hơn 100 chiến lược trong vài giờ.

6. Tự động phát hiện mẫu bất thường với HolySheep AI

Sau khi có dữ liệu, tôi cần một AI giúp phân loại các đợt biến động lớn và đề xuất ngưỡng. Thay vì gọi trực tiếp OpenAI hay Anthropic (giá cao, thanh toán rắc rối), tôi dùng

Tài nguyên liên quan

• 📚 Hướng dẫn AI API
• 💰 Xem giá
• 📖 Tài liệu nhà phát triển
• 🚀 Đăng ký miễn phí

Bài viết liên quan

• Tardis L2 Order Book depth_snapshot: Giải mã trường dữ liệu
• Kaiko vs Tardis pricing breakdown 2026: institutional vs dev
• ETH Order Book bids/asks 字段含义与回测数据清洗方案