Thiết kế schema thống nhất dữ liệu thị trường đa sàn: Tardis, Binance, OKX, Bybit

2h sáng tháng 3/2026, tôi ngồi trước terminal với 4 tab WebSocket chạy song song, mỗi sàn lại trả về một cấu trúc JSON khác nhau: Binance gửi "p" cho giá, OKX gửi "px", Bybit gọi nó là "p" nhưng định dạng timestamp lại là "T" còn Binance là "T" đơn vị milisecond còn OKX lại là "ts" đơn vị milisecond… Tôi đã dành 2 đêm liền để gỡ lỗi chỉ vì không chuẩn hoá schema từ đầu.

Bài viết này ghi lại kinh nghiệm thực chiến khi tôi xây dựng hệ thống thu thập tick dữ liệu cho dự án cá nhân: một dashboard phân tích định lượng tích hợp AI, hỗ trợ truy vấn lịch sử giá bằng tiếng Việt. Mục tiêu là ánh xạ 4 nguồn dữ liệu — Tardis (historical tick replay), Binance Spot, OKX, Bybit — về cùng một schema chuẩn, đủ sạch để nhét vào TimescaleDB và gọi bằng LLM.

1. Vì sao phải chuẩn hoá schema?

Mỗi sàn có một lịch sử thiết kế API khác nhau. Nếu bạn viết code thuần cho từng sàn, bạn sẽ có 4 nhánh xử lý lỗi, 4 bộ test, và mỗi lần sàn nâng cấp là bạn sửa 4 chỗ. Schema thống nhất giải quyết tận gốc vấn đề này: code của bạn chỉ biết một kiểu dữ liệu, và 4 adapter chịu trách nhiệm "dịch" ngữ nghĩa.

• Giảm 70-80% boilerplate ở tầng persistence và analytics.
• Cho phép LLM truy vấn dữ liệu mà không cần biết sàn nào là sàn nào.
• Dễ dàng replay dữ liệu lịch sử từ Tardis để backtest đồng nhất với dữ liệu live.

2. Bảng ánh xạ trường giữa Tardis, Binance, OKX, Bybit

Lưu ý quan trọng: Binance trả về giá/khối lượng dưới dạng chuỗi để giữ chính xác số thập phân. OKX cũng vậy. Bybit thì trả số. Tardis lại trả số thực Python gốc. Bạn không được ép kiểu float rồi nhân lên — hãy dùng Decimal hoặc scale về integer (ví dụ price × 1e8).

3. Schema thống nhất bằng Pydantic v2

Tôi chọn Pydantic v2 vì tốc độ validate nhanh (~3-5 microseconds/bản ghi theo benchmark nội bộ của tôi) và tích hợp tốt với JSON serialization cho TimescaleDB.

# unified_schema.py — chạy được với: pip install pydantic>=2.6
from decimal import Decimal
from enum import Enum
from typing import Optional, Literal
from pydantic import BaseModel, Field, field_validator
import time

class Side(str, Enum):
    BUY = "buy"
    SELL = "sell"

SUPPORTED_EXCHANGES = Literal["binance", "okx", "bybit", "tardis:binance-futures"]

class UnifiedTrade(BaseModel):
    """Schema thống nhất cho một tick giao dịch từ bất kỳ sàn nào."""
    exchange: SUPPORTED_EXCHANGES
    symbol: str
    trade_id: str
    price: Decimal = Field(..., decimal_places=8, max_digits=20)
    quantity: Decimal = Field(..., decimal_places=8, max_digits=20)
    side: Side
    timestamp_ms: int = Field(..., ge=0)
    received_at_ms: int
    is_buyer_maker: Optional[bool] = None
    raw: Optional[dict] = None

    @field_validator("symbol")
    @classmethod
    def normalize_symbol(cls, v: str) -> str:
        # Chuẩn hoá về dạng BTCUSDT không phân biệt
        return v.replace("-", "").replace("/", "").upper()

    @field_validator("received_at_ms", mode="before")
    @classmethod
    def set_received_at(cls, v):
        return v if v else int(time.time_ns() // 1_000_000)

Khởi tạo một bản ghi mẫu
sample = UnifiedTrade(
    exchange="binance",
    symbol="BTCUSDT",
    trade_id="12345",
    price=Decimal("50000.10"),
    quantity=Decimal("0.005"),
    side=Side.BUY,
    timestamp_ms=1700000000000,
    is_buyer_maker=False,
)
print(sample.model_dump_json())

4. Adapter — bộ ánh xạ cho từng sàn

Đây là phần "xương sống". Mỗi adapter nhận raw message từ sàn và trả về UnifiedTrade. Tôi viết dạng class đơn giản để dễ test.

# adapters.py — bộ chuyển đổi raw → UnifiedTrade
from decimal import Decimal
from unified_schema import UnifiedTrade, Side

class BinanceAdapter:
    exchange = "binance"

    @staticmethod
    def parse(msg: dict) -> UnifiedTrade:
        # msg có dạng {"e":"trade","s":"BTCUSDT","t":12345,"p":"50000.10",
        #              "q":"0.005","T":1700000000000,"m":false}
        is_buyer_maker = bool(msg["m"])
        return UnifiedTrade(
            exchange="binance",
            symbol=msg["s"],
            trade_id=str(msg["t"]),
            price=Decimal(msg["p"]),
            quantity=Decimal(msg["q"]),
            side=Side.SELL if is_buyer_maker else Side.BUY,
            timestamp_ms=int(msg["T"]),
            is_buyer_maker=is_buyer_maker,
            raw=msg,
        )

class OKXAdapter:
    exchange = "okx"

    @staticmethod
    def parse(msg: dict) -> UnifiedTrade:
        # OKX gói data trong {"arg":..., "data":[ {...} ]}
        d = msg["data"][0]
        side = Side(d["side"])
        return UnifiedTrade(
            exchange="okx",
            symbol=d["instId"],          # "BTC-USDT" → sẽ được normalize
            trade_id=str(d["tradeId"]),
            price=Decimal(d["px"]),
            quantity=Decimal(d["sz"]),
            side=side,
            timestamp_ms=int(d["ts"]),
            # OKX không trực tiếp cho is_buyer_maker, suy ra:
            is_buyer_maker=(side == Side.SELL),
            raw=msg,
        )

class BybitAdapter:
    exchange = "bybit"

    @staticmethod
    def parse(msg: dict) -> UnifiedTrade:
        # Bybit V5: {"topic":"publicTrade.BTCUSDT","data":[ {...} ]}
        d = msg["data"][0]
        side = Side(d["S"].lower())
        return UnifiedTrade(
            exchange="bybit",
            symbol=d["s"],
            trade_id=str(d["i"]),
            price=Decimal(str(d["p"])),
            quantity=Decimal(str(d["v"])),
            side=side,
            timestamp_ms=int(d["T"]),
            is_buyer_maker=(side == Side.SELL),
            raw=msg,
        )

class TardisAdapter:
    """Tardis replay tick — dữ liệu đã được normalize một phần."""
    exchange = "tardis:binance-futures"

    @staticmethod
    def parse(msg: dict) -> UnifiedTrade:
        return UnifiedTrade(
            exchange="tardis:binance-futures",
            symbol=msg["symbol"].upper(),
            trade_id=str(msg["id"]),
            price=Decimal(str(msg["price"])),
            quantity=Decimal(str(msg["amount"])),
            side=Side(msg["side"]),
            timestamp_ms=int(msg["ts"]),
            # Tardis bổ sung local_realtime dùng cho latency analysis
            received_at_ms=int(msg.get("local_realtime", msg["ts"])),
            raw=msg,
        )

Demo chạy nhanh: cùng 1 lệnh, test cả 4 adapter
import json
binance_msg = {"e":"trade","s":"BTCUSDT","t":12345,"p":"50000.10","q":"0.005","T":1700000000000,"m":False}
print(BinanceAdapter.parse(binance_msg).model_dump_json(indent=2))

5. So sánh chi phí: Tardis thuê bao vs WebSocket thô + AI xử lý

Khi tôi bắt đầu dự án, tôi đã đứng trước lựa chọn: trả tiền cho Tardis để có dữ liệu tick lịch sử chuẩn (hữu ích cho backtest), hay tự lưu trữ dữ liệu live từ WebSocket (miễn phí nhưng tốn công). Bảng dưới tổng hợp chi phí thực tế tôi ghi nhận trong 1 tháng vận hành.