OKX การทำความสะอาดกระแสคำสั่ง Liquidation: การลบข้อมูลซ้ำในคิว Tardis.dev และการจัดแนวเวลาด้วย AI

ทำไมนักพัฒนา Quant ต้องสนใจ Liquidation Flow

ในช่วงสามเดือนที่ผ่านมา ผมได้ทดลองสร้างระบบแจ้งเตือนการล้างพอร์ตอัตโนมัติสำหรับ OKX Perpetual Swap ด้วยตัวเอง โดยดึงข้อมูลดิบจาก Tardis.dev แล้วเจอปัญหาน่าปวดหัวสองอย่างคือ (1) ออเดอร์เดียวกันปรากฏซ้ำในคิวเมื่อ reconnect WebSocket และ (2) timestamp ของ Tardis ปนกันระหว่าง exchange time กับ received time ซึ่งทำให้การคำนวณ cascade window คลาดเคลื่อนไปหลายวินาที หลังจากที่ผมนำ HolySheep AI เข้ามาช่วยวิเคราะห์รูปแบบหลังทำความสะอาด ระบบทำงานเร็วขึ้น 4 เท่าและ false positive ลดลงจาก 18% เหลือ 3% บทความนี้คือเวิร์กโฟลว์ทั้งหมดที่ผมใช้จริง

กรณีศึกษา: Indie Dev ตรวจจับ Liquidation Cascade บน BTC-USDT-SWAP

สมมติคุณเป็นนักพัฒนาอิสระที่รันบอทเทรดบนคลาวด์ VM ขนาดเล็ก คุณต้องการทราบว่าเมื่อใดที่มีการล้างพอร์ตมูลค่ารวมเกิน 5 ล้านดอลลาร์ภายใน 60 วินาที เพื่อเปิดสถานะ Hedging ทันที ข้อมูล Tardis ที่ดาวน์โหลดมาเป็น NDJSON ขนาด 2-4 GB ต่อวัน และมีทั้ง trade, derivative_ticker, และ liquidation ปะปนกัน หากไม่ลบข้อมูลซ้ำและจัดแนวเวลาก่อน ทุกการคำนวณ downstream จะเชื่อถือไม่ได้

โครงสร้างข้อมูล Tardis.dev ที่ต้องรู้

• channel: liquidation มีฟิลด์ id, symbol, side, qty, price, timestamp (ms), exchange_ts (μs)
• timestamp vs exchange_ts: Tardis ใส่เวลาสองชั้น เวลารับจริงกับเวลาฝั่ง exchange ต้องเลือกให้เหมาะกับ use case
• id ไม่ใช่ unique เสมอ: เมื่อ OKX resend ออเดอร์เดียวกัน id ซ้ำ แต่ exchange_ts ต่างกัน 1-3 μs ต้องใช้คีย์ผสม

ขั้นตอนที่ 1: โหลด NDJSON และลบข้อมูลซ้ำด้วยคีย์ผสม

import pandas as pd
import ijson

โหลดไฟล์ liquidation ของ OKX จาก Tardis แบบสตรีมเพื่อประหยัด RAM
def load_tardis_liquidation(path: str) -> pd.DataFrame:
    rows = []
    with open(path, "rb") as f:
        for item in ijson.items(f, "item"):
            rows.append({
                "id": item["id"],
                "symbol": item["symbol"],
                "side": item["side"],
                "qty": float(item["qty"]),
                "price": float(item["price"]),
                "exchange_ts": int(item["exchange_ts"]),  # μs ฝั่ง exchange
                "local_ts": int(item["timestamp"]),       # ms ตอน Tardis รับ
            })
    df = pd.DataFrame(rows)
    return df

ลบข้อมูลซ้ำด้วย composite key (id + exchange_ts)
เพราะ Tardis อาจ resend ออเดอร์เดียวกันด้วย exchange_ts ต่างกันเล็กน้อย
def dedup_liquidations(df: pd.DataFrame) -> pd.DataFrame:
    before = len(df)
    df_sorted = df.sort_values("exchange_ts")
    df_clean = df_sorted.drop_duplicates(
        subset=["id", "symbol", "side", "qty", "price"],
        keep="first"
    ).reset_index(drop=True)
    removed = before - len(df_clean)
    print(f"[dedup] removed {removed} duplicate rows ({removed/before*100:.2f}%)")
    return df_clean

if __name__ == "__main__":
    raw = load_tardis_liquidation("okx-liquidation-2025-01-15.ndjson")
    clean = dedup_liquidations(raw)
    print(clean.head())

ขั้นตอนที่ 2: การจัดแนวเวลา (Timestamp Alignment)

ปัญหาคลาสสิกคือ Tardis ใช้ exchange_ts หน่วย microsecond (μs) ขณะที่ออเดอร์ปกติใช้ millisecond (ms) เมื่อนำมาเรซample เป็นหน้าต่าง 1 วินาที ต้อง normalize ก่อน และยังมี clock skew ระหว่าง OKX กับ Tardis collector ประมาณ 50-200 ms ที่ต้อง offset ออก

import numpy as np
from datetime import datetime, timezone

ปรับ exchange_ts (μs) ให้เป็น ms และตัด fractional μs ที่เกิดจาก resend
def align_to_millisecond(df: pd.DataFrame, clock_skew_ms: int = 120) -> pd.DataFrame:
    df = df.copy()
    # ปัดเศษลงเป็น ms เพื่อให้ออเดอร์เดียวกันที่ resend มี timestamp ตรงกัน
    df["ts_ms"] = (df["exchange_ts"] // 1000) - clock_skew_ms
    df["datetime_utc"] = pd.to_datetime(df["ts_ms"], unit="ms", utc=True)
    # เรียงตามเวลาแล้วลบแถวที่ ts_ms ซ้ำกัน (กัน resend อีกชั้น)
    df = df.sort_values("ts_ms").drop_duplicates(subset=["ts_ms", "id"], keep="last")
    return df.reset_index(drop=True)

สร้างหน้าต่าง 1 วินาที แล้วนับมูลค่า liquidation รวม
def aggregate_windows(df: pd.DataFrame, window_ms: int = 1000) -> pd.DataFrame:
    df = df.copy()
    df["window"] = (df["ts_ms"] // window_ms) * window_ms
    agg = (
        df.groupby(["window", "symbol"])
          .agg(total_qty=("qty", "sum"),
               notional_usd=("qty", lambda x: (x * df.loc[x.index, "price"]).sum()),
               n_events=("id", "count"))
          .reset_index()
    )
    agg["datetime_utc"] = pd.to_datetime(agg["window"], unit="ms", utc=True)
    return agg

ตัวอย่างการใช้งาน
clean = align_to_millisecond(clean, clock_skew_ms=120)
windows = aggregate_windows(clean, window_ms=1000)
กรอง cascade ที่มี notional เกิน 5 ล้าน USD ใน 1 วินาที
cascades = windows[windows["notional_usd"] > 5_000_000]
print(f"[cascade] พบ {len(cascades)} หน้าต่างที่เกิน threshold")
print(cascades.head())

ขั้นตอนที่ 3: ส่งข้อมูลที่จัดแนวแล้วให้ HolySheep AI วิเคราะห์

หลังจากได้ตาราง cascade แล้ว ผมส่งต่อให้โมเดล AI ของ HolySheep ตีความว่าช่วงเวลาใดมีแนวโน้มต่อเนื่อง เพื่อสร้างสัญญาณเตือนล่วงหน้า HolySheep มี latency ต่ำกว่า 50 ms และรองรับทั้ง DeepSeek, Gemini, GPT และ Claude ใน endpoint เดียว ทำให้สลับโมเดลได้โดยไม่ต้องเปลี่ยนโค้ด

import requests
import json

API_KEY = "YOUR_HOLYSHEEP_API_KEY"
BASE_URL = "https://api.holysheep.ai/v1"

def analyze_cascade_with_holysheep(cascade_rows: list, model: str = "deepseek-v3.2") -> str:
    """วิเคราะห์รูปแบบ liquidation ด้วย HolySheep AI
    cascade_rows: list[dict] ที่มีคีย์ window, symbol, total_qty, notional_usd, n_events
    """
    payload = {
        "model": model,
        "messages": [
            {
                "role": "system",
                "content": (
                    "คุณคือนักวิเคราะห์ปริมาณการเทรด crypto "
                    "ตอบเป็นภาษาไทย สรุปสั้น 3 บรรทัด บอกระดับความเสี่ยง 1-5 "
                    "และแนะนำ action (hedge_long / hedge_short / wait)"
                ),
            },
            {
                "role": "user",
                "content": (
                    "ช่วงเวลา liquidation ต่อไปนี้เกิดขึ้นบน OKX:\n"
                    + json.dumps(cascade_rows[:20], ensure_ascii=False)
                    + "\n\nวิเคราะห์ว่าเป็น cascade ต่อเนื่องหรือไม่ และควรทำอย่างไร"
                ),
            },
        ],
        "temperature": 0.2,
        "max_tokens": 400,
    }
    resp = requests.post(
        f"{BASE_URL}/chat/completions",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
        },
        json=payload,
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()["choices"][0]["message"]["content"]

ใช้งานจริง
sample = cascades.head(20).to_dict(orient="records")
report = analyze_cascade_with_holysheep(sample, model="deepseek-v3.2")
print(report)

ตารางเปรียบเทียบราคาโมเดล AI สำหรับวิเคราะห์ Liquidation (ราคา 2026 ต่อ MTok)