รีวิวการออกแบบ Schema รวมข้อมูลหลายกระดาน: Binance / OKX / Bybit (Spot & Perpetual) — พร้อมใช้ HolySheep AI ช่วยเขียนโค้ด

ผมใช้เวลาประมาณ 6 สัปดาห์ในการทดลองสร้าง data aggregator สำหรับเก็บ orderbook + ticker จาก 3 กระดานหลัก ได้แก่ Binance, OKX และ Bybit ทั้งตลาด Spot และ Perpetual (USDT-margined) บทความนี้เป็นรีวิวจริงตั้งแต่ requirement, schema design, โค้ด mapping, ไปจนถึง benchmark ค่าหน่วง อัตราสำเร็จ และต้นทุนการรัน inference ผ่าน HolySheep AI เพื่อช่วย refactor โค้ด Parser ให้ทนทานขึ้น

1. ทำไมต้องมี Unified Schema

• ทุกกระดานส่ง payload คนละรูปแบบ เช่น Binance ใช้ e/U/u/b/a, OKX ใช้ arg.channel + data[], Bybit ใช้ topic + data โดยทั้งหมดมีความหมายเดียวกัน (best bid/ask + timestamp)
• ถ้าเขียน business logic แยกตามกระดาน จะเกิด code duplication สูง และแก้ยากเมื่อกระดานเปลี่ยน format
• Unified schema ช่วยให้ทีมทำ strategy / risk / dashboard ใช้ data type เดียวกันได้ทันที โดยไม่ต้องรู้ว่ามาจากกระดานไหน

2. Unified Schema (Canonical)

ผมเลือกใช้ Pydantic v2 เพราะ validate ได้ทั้ง runtime และ export เป็น JSON Schema ไปใช้กับ contract test ได้สะดวก

from pydantic import BaseModel, Field
from typing import List, Literal
from datetime import datetime

MarketType = Literal["spot", "perpetual_usdt"]
Exchange = Literal["binance", "okx", "bybit"]
Side = Literal["bid", "ask"]

class PriceLevel(BaseModel):
    price: float = Field(..., gt=0, description="ราคาต่อหน่วย")
    size: float = Field(..., ge=0, description="ขนาดคำสั่งซื้อขาย")
    order_count: int | None = Field(None, ge=0, description="จำนวน order รวม (ถ้ามี)")

class NormalizedOrderBook(BaseModel):
    exchange: Exchange
    market_type: MarketType
    symbol: str = Field(..., description="เช่น BTCUSDT หรือ BTC-USDT")
    ts_exchange: datetime
    ts_received: datetime
    seq: int | None = Field(None, description="last update id / sequence ของกระดาน")
    bids: List[PriceLevel]
    asks: List[PriceLevel]
    is_snapshot: bool = True

3. Field Mapping ต่อกระดาน

ผมลองรัน benchmark บนเครื่อง VM สิงคโปร์ (AWS ap-southeast-1) เชื่อมต่อ WebSocket 3 กระดานพร้อมกันเป็นเวลา 24 ชม. ได้ค่าดังนี้

หมายเหตุ: ค่าเฉลี่ยวัดจาก ts_exchange ถึงเวลาที่ parser สร้าง NormalizedOrderBook สำเร็จ (ไม่รวมเวลา network จาก local ไป exchange)

4. Parser ต่อกระดาน (Runnable)

ตัวอย่างนี้เป็นโค้ดจริงที่รันใน pipeline ของผม รับ raw message จาก WebSocket แล้วคืน NormalizedOrderBook

import json
from datetime import datetime, timezone

def parse_binance(raw: dict, market_type: str = "spot") -> NormalizedOrderBook:
    d = raw.get("data", raw)
    return NormalizedOrderBook(
        exchange="binance",
        market_type=market_type,
        symbol=d["s"],
        ts_exchange=datetime.fromtimestamp(d["E"] / 1000, tz=timezone.utc),
        ts_received=datetime.now(tz=timezone.utc),
        seq=d.get("u"),
        bids=[PriceLevel(price=float(p), size=float(q)) for p, q in d["b"][:20]],
        asks=[PriceLevel(price=float(p), size=float(q)) for p, q in d["a"][:20]],
        is_snapshot="e" not in raw or raw["e"] == "depthUpdate" and False,
    )

def parse_okx(raw: dict, market_type: str = "perpetual_usdt") -> NormalizedOrderBook:
    data = raw["data"][0]
    sym = raw["arg"]["instId"].replace("-", "")
    return NormalizedOrderBook(
        exchange="okx",
        market_type=market_type,
        symbol=sym,
        ts_exchange=datetime.fromtimestamp(int(data["ts"]) / 1000, tz=timezone.utc),
        ts_received=datetime.now(tz=timezone.utc),
        seq=None,
        bids=[PriceLevel(price=float(p), size=float(q)) for p, q, *_ in data["bids"][:20]],
        asks=[PriceLevel(price=float(p), size=float(q)) for p, q, *_ in data["asks"][:20]],
        is_snapshot=False,
    )

def parse_bybit(raw: dict, market_type: str = "perpetual_usdt") -> NormalizedOrderBook:
    d = raw["data"]
    return NormalizedOrderBook(
        exchange="bybit",
        market_type=market_type,
        symbol=d["s"],
        ts_exchange=datetime.fromtimestamp(raw["ts"] / 1000, tz=timezone.utc),
        ts_received=datetime.now(tz=timezone.utc),
        seq=d.get("seq"),
        bids=[PriceLevel(price=float(p), size=float(q)) for p, q in d["b"][:20]],
        asks=[PriceLevel(price=float(p), size=float(q)) for p, q in d["a"][:20]],
        is_snapshot=raw.get("type") == "snapshot",
    )

5. ใช้ HolySheep AI ช่วยเขียน Test + Refactor

หลังจากที่ parser ทำงานครบ 3 กระดาน ผมอยากได้ unit test ที่ครอบคลุม edge case เช่น field หาย, symbol mismatch, decimal precision ผมเลือกใช้ HolySheep AI เป็น LLM gateway เพราะราคาถูกกว่า OpenAI ตรง 85%+ (อัตรา ¥1=$1) และตอบเร็วกว่า 50ms ทดลองในงานจริงได้ลื่นกว่าการยิงตรงไป api.openai.com

import httpx, os

base_url ของ HolySheep เท่านั้น ห้ามใช้ api.openai.com
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY  = os.environ["HOLYSHEEP_API_KEY"]  # ใส่ YOUR_HOLYSHEEP_API_KEY

SYSTEM = "คุณคือ senior Python engineer เชี่ยวชาญ Pydantic v2 และ crypto exchange API"

def holy(prompt: str, model: str = "deepseek-v3.2") -> str:
    r = httpx.post(
        f"{BASE_URL}/chat/completions",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "model": model,
            "messages": [
                {"role": "system", "content": SYSTEM},
                {"role": "user",   "content": prompt},
            ],
            "temperature": 0.2,
        },
        timeout=30.0,
    )
    r.raise_for_status()
    return r.json()["choices"][0]["message"]["content"]

ขอเทสต์จาก parser ที่เขียนเสร็จแล้ว
unit_tests = holy("""
เขียน pytest unit test สำหรับฟังก์ชัน parse_binance, parse_okx, parse_bybit
ครอบคลุม: snapshot ปกติ, depth update, field หาย, price ติดลบ, asks

# ❌ ผิด
ts = datetime.fromisoformat(data["ts"])

✅ ถูก
ts_ms = int(data["ts"])
ts = datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc)

# ✅ normalize ตอน parse
symbol = raw["arg"]["instId"].replace("-", "")   # BTC-USDT -> BTCUSDT

# ✅ คำนวณบิลผ่าน HolySheep (อัตราคงที่)
cost_usd = (input_tokens / 1e6) * price_in_per_mtok
ไม่ต้องหักค่า FX อีก