ก่อนจะลงลึกเรื่อง Tardis ขอเริ่มด้วยข้อมูลต้นทุน AI ปี 2026 ที่ผู้เขียนใช้งานจริงในไปป์ไลน์วิจัยเชิงปริมาณ เพราะเมื่อคุณเริ่มยิงโมเดلเพื่อคลาสสิฟายเซนติเมนต์ข่าว ตรวจจับ spoofing หรือสร้างฟีเจอร์เรจิเมต ค่าใช้จ่ายต่อเดือนคือปัจจัยที่ส่งผลต่อ PnL โดยตรง
| โมเดل | ราคา output ($/MTok) | ต้นทุน 10M tokens/เดือน | ส่วนต่างเทียบ Claude |
|---|---|---|---|
| Claude Sonnet 4.5 | 15.00 | $150.00 | — |
| GPT-4.1 | 8.00 | $80.00 | -$70.00 |
| Gemini 2.5 Flash | 2.50 | $25.00 | -$125.00 |
| DeepSeek V3.2 | 0.42 | $4.20 | -$145.80 |
ส่วนต่างระหว่าง Claude Sonnet 4.5 กับ DeepSeek V3.2 อยู่ที่ประมาณ $145.80 ต่อเดือน หรือคิดเป็น 35 เท่า ถ้าคุณรันหลายร้อนเพื่อ enrich ข้อมูล Tardis ตัวเลขนี้จะกลายเป็นต้นทุนคงที่ที่กัดกำไรโดยตรง ส่วนตัวผู้เขียนใช้ HolySheep AI ในเรท ¥1=$1 (ประหยัด 85%+) รองรับ WeChat/Alipay และ latency <50ms ทำให้ต้นทุนต่อเดือนเหลือหลักสิบเซนต์
Tardis L2 Order Book คืออะไร แล้วทำไมต้อง Level-3
Tardis เป็นดาต้าเซ็นเตอร์ที่เก็บข้อมูล tick-by-tick ของคริปโตเดอริเวทีฟ เช่น Binance, Bybit, OKX, Coinbase ทั้งแบบ normalized L2 และ L3 ความแตกต่างที่ส่งผลต่อความแม่นยำของแบ็คเทสต์โดยตรง:
- L2 = สมุดคำสั่งที่รวมยอดต่อ price level เช่น bid ที่ 100,000 มี 3.5 BTC รวมกัน ไม่มี order_id เหมาะกับการดูความลึกของตลาด
- L3 = เห็นทุก order_id ว่าเกิด แก้ไข ยกเลิก หรือถูกฟิลด์เมื่อใด จำเป็นต่อการแบ็คเทสต์ market making เพราะคุณต้องรู้ว่าคุณอยู่ในคิวตำแหน่งที่เท่าไหร่
ไม่ใช่ทุก exchange ที่ Tardis มี L3 บางตลาดให้แค่ L2 ดังนั้นการรีโหลด L3 จาก L2 deltas ด้วยโมเดล FIFO จึงเป็นเทคนิคที่เทรดเดอร์มืออาชีพใช้กัน เพื่อจำลอง queue position ของ limit order ที่คุณวาง
ขั้นที่ 1 ดึง L2 สแนปช็อตและเดลต้าจาก Tardis
Tardis ให้บริการสอง endpoint หลักคือ /snapshots/ สำหรับ full book state ณ เวลาใดเวลาหนึ่ง และ /incremental_book_L2/ สำหรับ delta ที่ใช้ apply ต่อจาก snapshot ตัวอย่างโค้ดดึงข้อมูล BTCUSDT ของ Binance ย้อนหลัง:
import requests, pandas as pd, io
TARDIS_KEY = "YOUR_TARDIS_API_KEY"
SYMBOL = "binance-futures.BTCUSDT"
DATE = "2024-06-15"
def fetch_tardis(path: str) -> pd.DataFrame:
url = f"https://api.tardis.dev/v1{path}"
headers = {"Authorization": f"Bearer {TARDIS_KEY}"}
r = requests.get(url, headers=headers, stream=True, timeout=60)
r.raise_for_status()
return pd.read_csv(io.StringIO(r.text))
snapshot_df = fetch_tardis(
f"/data-feeds/{SYMBOL}/snapshots/{DATE}.csv.gz"
)
incremental_df = fetch_tardis(
f"/data-feeds/{SYMBOL}/incremental_book_L2/{DATE}.csv.gz"
)
print(snapshot_df.head())
print(incremental_df.columns.tolist())
คาดหวังคอลัมน์: ts_recv, ts_send, local_timestamp, side, price, amount
โค้ดข้างบนจะคืน DataFrame ของสแนปช็อตที่ /snapshots/ ตามด้วยทุก delta ของวันนั้น ให้สังเกตว่า Tardis แยก timestamp เป็น ts_recv (เวลาที่ Tardis รับจาก exchange gateway) และ ts_send (เวลาที่ exchange ส่งออก) ค่าความคลาดเคลื่อนระหว่างสองฟิลด์นี้คือ latency จริงของ exchange ที่คุณควรบันทึกไว้ใน log แบ็คเทสต์ด้วย
ขั้นที่ 2 รีโหลด L3 Order Book ด้วยโมเดล FIFO
หลักการคือเราจะ treat ทุก price level เป็น virtual order queue เมื่อมี delta ที่ลด amount ลง เราจะถือว่า order ที่อยู่หัวคิวถูก fill ก่อน แล้วเพิ่ม order ใหม่ที่ท้ายคิว เทคนิคนี้ตรงกับ behavior ของคู่ BTCUSDT บน Binance/OKX ที่ใช้ price-time priority
from collections import deque, defaultdict
from typing import Tuple
class VirtualL3Book:
def __init__(self):
self.bids: dict[float, deque] = defaultdict(deque) # price -> queue of (oid, size)
self.asks: dict[float, deque] = defaultdict(deque)
self.oid_counter = 0
def apply_delta(self, side: str, price: float, amount: float):
book = self.bids if side == "buy" else self.asks
q = book[price]
delta_remaining = amount
while delta_remaining != 0 and q:
oid, size = q[0]
if size > delta_remaining:
q[0] = (oid, size - delta_remaining)
delta_remaining = 0
else:
q.popleft()
delta_remaining -= size
if delta_remaining > 0:
self.oid_counter += 1
q.append((f"v{self.oid_counter}", delta_remaining))
def place_limit(self, side: str, price: float, size: float) -> str:
book = self.bids if side == "buy" else self.asks
self.oid_counter += 1
oid = f"me_{self.oid_counter}"
book[price].append((oid, size))
return oid
def queue_position(self, oid: str, side: str) -> Tuple[int, float]:
book = self.bids if side == "buy" else self.asks
for price, q in book.items():
ahead = 0.0
for o, s in q:
if o == oid:
return int(ahead), ahead
ahead += s
return -1, -1.0
book = VirtualL3Book()
apply ทุก delta จาก incremental_df
for _, row in incremental_df.iterrows():
book.apply_delta(row["side"], float(row["price"]), float(row["amount"]))
my_oid = book.place_limit("buy", 65000.0, 0.5)
ahead_size, _ = book.queue_position(my_oid, "buy")
print(f"คิวหน้าฉันมี {ahead_size:.3f} BTC")
ค่า queue_position คือปริมาณที่ต้องถูก fill ไปก่อนคุณจะได้ fill ซึ่งเป็นหัวใจของการแบ็คเทสต์ market making ถ้า ahead_size สูงเทียบกับปริมาณเฉลี่ยต่อนาที โอกาส fill ของคุณต่ำ และควรขยับสเปรดให้กว้างขึ้น
ขั้นที่ 3 แบ็คเทสต์กลยุทธ์ Market Making แบบ Avellaneda-Stoikov
เมื่อได้ L3 แล้ว เราจะรัน strategy แบบคลาสสิก โดยคำนวณ reservation price และ optimal spread ทุก tick แล้วเช็คว่า order เราถูก fill หรือไม่จาก delta ที่วิ่งผ่านมา
import numpy as np
from dataclasses import dataclass, field
@dataclass
class MMConfig:
risk_aversion: float = 0.1 # gamma
volatility: float = 0.02 # sigma ต่อ tick
time_horizon: float = 300.0 # T หน่วยเป็น tick
order_size: float = 0.01 # BTC ต่อ order
inventory_limit: float = 1.0 # BTC
@dataclass
class MMState:
cash: float = 0.0
inventory: float = 0.0
pnl_history: list = field(default_factory=list)
my_active: dict = field(default_factory=dict)
def avellaneda_spread(s: MMState, cfg: MMConfig) -> tuple[float, float]:
mid = s.cash and 0 # ใส่ mid price จาก L2 top
reservation = mid - s.inventory * cfg.risk_aversion * (cfg.volatility ** 2) * cfg.time_horizon
spread = (cfg.risk_aversion * (cfg.volatility ** 2) * cfg.time_horizon) + \
(2 / cfg.risk_aversion) * np.log(1 + cfg.risk_aversion / 0.1)
return reservation - spread / 2, reservation + spread / 2
รันย้อนหลังผ่าน incremental_df
state = MMState()
cfg = MMConfig()
for _, row in incremental_df.iterrows():
mid = float(row["price"]) # ใช้ mid จาก delta (ปรับด้วย top of book จริง)
bid_px, ask_px = avellaneda_spread(state, cfg)
if abs(state.inventory) < cfg.inventory_limit:
state.my_active["bid"] = (bid_px, cfg.order_size)
state.my_active["ask"] = (ask_px, cfg.order_size)
# simulate fill: ถ้า delta ทำลาย price level ของเรา = ถูก fill
if row["side"] == "buy" and row["amount"] > 0:
# market มี buyer ใหม่ = มีคนยิง bid ผ่าน ask ของเรา
if float(row["price"]) >= state.my_active.get("ask", (0, 0))[0]:
state.cash += float(row["price"]) * cfg.order_size
state.inventory -= cfg.order_size
elif row["side"] == "sell" and row["amount"]
แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง