ในโลกของสัญญาซื้อขายแลกเปลี่ยนถาวร (Perpetual Futures) การเข้าถึงข้อมูล tick-by-tick ที่แม่นยำเป็นรากฐานของกลยุทธ์ market making และ arbitrage บทความนี้จะพาคุณสำรวจเชิงลึกเกี่ยวกับ Tardis Data API วิธีดึงข้อมูลจาก OKX และ Bybit พร้อมเทคนิคการจัดการ funding rates และ order book depth snapshots ที่ใช้งานจริงใน production
Tardis Data API คืออะไร และทำไมถึงสำคัญ
Tardis เป็น data aggregator ที่รวบรวม raw market data จาก exchanges หลายตัวผ่าน normalized API เดียว ข้อดีคือไม่ต้องจัดการ connection หลายตัว และได้ข้อมูลที่ cleaned แล้ว สำหรับ OKX และ Bybit ซึ่งเป็น top-tier derivatives exchanges นั้น Tardis ให้ความแม่นยำของ timestamp ที่ <1ms และรองรับทั้ง trade ticks, funding rate updates, และ depth snapshots
การตั้งค่า Environment และ Dependencies
# requirements.txt
tardis-client==2.1.0
websockets==12.0
pandas==2.0.3
numpy==1.24.3
pyarrow==14.0.1
aiokafka==0.10.0
redis==5.0.1
สำหรับ HolySheep AI ในการประมวลผลข้อมูลเพิ่มเติม
httpx==0.27.0
orjson==3.9.12# config.py
import os
from dataclasses import dataclass
from typing import List
@dataclass
class ExchangeConfig:
exchange: str
symbols: List[str]
channels: List[str]
api_key: str
api_secret: str
@dataclass
class TardisConfig:
api_key: str
base_url: str = "wss://tardis.dev:9000"
reconnect_delay: float = 5.0
max_reconnect_attempts: int = 10
heartbeat_interval: int = 30
@dataclass
class HolySheepConfig:
api_key: str = "YOUR_HOLYSHEEP_API_KEY"
base_url: str = "https://api.holysheep.ai/v1"
timeout: int = 30
max_retries: int = 3
OKX Perpetual Config
OKX_CONFIG = ExchangeConfig(
exchange="okx",
symbols=["BTC-USDT-SWAP", "ETH-USDT-SWAP"],
channels=["trades", "funding_rate", "booksL2"],
api_key=os.getenv("OKX_API_KEY", ""),
api_secret=os.getenv("OKX_API_SECRET", "")
)
Bybit Perpetual Config
BYBIT_CONFIG = ExchangeConfig(
exchange="bybit",
symbols=["BTCUSDT", "ETHUSDT"],
channels=["trades", "funding", "orderbook"],
api_key=os.getenv("BYBIT_API_KEY", ""),
api_secret=os.getenv("BYBIT_API_SECRET", "")
)
HolySheep AI Config สำหรับ AI-powered analysis
HOLYSHEEP_CONFIG = HolySheepConfig(
api_key="YOUR_HOLYSHEEP_API_KEY", # รับได้จาก https://www.holysheep.ai/register
base_url="https://api.holysheep.ai/v1",
timeout=30,
max_retries=3
)Client หลักสำหรับเชื่อมต่อ Tardis Data API
# tardis_client.py
import asyncio
import json
import logging
from datetime import datetime
from typing import Dict, Callable, Optional, List
from dataclasses import dataclass, field
import pandas as pd
import aiohttp
import websockets
from websockets.exceptions import ConnectionClosed
from config import TardisConfig, OKX_CONFIG, BYBIT_CONFIG
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class TickData:
exchange: str
symbol: str
timestamp: datetime
price: float
size: float
side: str
trade_id: str
@dataclass
class FundingRateData:
exchange: str
symbol: str
timestamp: datetime
funding_rate: float
next_funding_time: datetime
@dataclass
class DepthSnapshot:
exchange: str
symbol: str
timestamp: datetime
bids: List[tuple] # [(price, size), ...]
asks: List[tuple] # [(price, size), ...]
class TardisDataClient:
def __init__(self, config: TardisConfig):
self.config = config
self.ws: Optional[websockets.WebSocketClientProtocol] = None
self.subscriptions: Dict[str, set] = {}
self.handlers: Dict[str, Callable] = {}
self._running = False
self._reconnect_count = 0
self._last_ping = None
async def connect(self):
"""เชื่อมต่อ WebSocket กับ Tardis API"""
auth_url = f"{self.config.base_url}?auth={self.config.api_key}"
try:
self.ws = await websockets.connect(auth_url)
self._running = True
self._reconnect_count = 0
logger.info(f"เชื่อมต่อ Tardis สำเร็จ: {self.config.base_url}")
except Exception as e:
logger.error(f"ไม่สามารถเชื่อมต่อ: {e}")
raise
async def subscribe(self, exchange: str, symbols: List[str], channels: List[str]):
"""สมัครรับข้อมูลจาก exchange เฉพาะ"""
for symbol in symbols:
for channel in channels:
sub_id = f"{exchange}:{symbol}:{channel}"
subscribe_msg = {
"type": "subscribe",
"exchange": exchange,
"channel": channel,
"symbol": symbol
}
await self.ws.send(json.dumps(subscribe_msg))
logger.info(f"สมัครรับ: {sub_id}")
if exchange not in self.subscriptions:
self.subscriptions[exchange] = set()
self.subscriptions[exchange].add(sub_id)
async def receive_loop(self):
"""Loop หลักสำหรับรับข้อมูล tick-by-tick"""
buffer = []
buffer_size = 1000
try:
async for message in self.ws:
try:
data = json.loads(message)
parsed = self._parse_message(data)
if parsed:
buffer.append(parsed)
# Batch processing เพื่อลด CPU overhead
if len(buffer) >= buffer_size:
await self._process_buffer(buffer)
buffer = []
except json.JSONDecodeError as e:
logger.warning(f"JSON decode error: {e}")
continue
except ConnectionClosed as e:
logger.error(f"Connection closed: {e}")
await self._handle_reconnect()
def _parse_message(self, data: dict) -> Optional[dict]:
"""Parse ข้อมูลตามประเภท message"""
msg_type = data.get("type", "")
if msg_type == "trade":
return TickData(
exchange=data["exchange"],
symbol=data["symbol"],
timestamp=datetime.fromisoformat(data["timestamp"].replace("Z", "+00:00")),
price=float(data["price"]),
size=float(data["size"]),
side=data["side"],
trade_id=data["id"]
)
elif msg_type in ["funding", "funding_rate"]:
return FundingRateData(
exchange=data["exchange"],
symbol=data["symbol"],
timestamp=datetime.fromisoformat(data["timestamp"].replace("Z", "+00:00")),
funding_rate=float(data["fundingRate"]),
next_funding_time=datetime.fromisoformat(data["nextFundingTime"].replace("Z", "+00:00"))
)
elif msg_type in ["book", "bookSnapshot", "orderbook"]:
return DepthSnapshot(
exchange=data["exchange"],
symbol=data["symbol"],
timestamp=datetime.fromisoformat(data["timestamp"].replace("Z", "+00:00")),
bids=[(float(p), float(s)) for p, s in data.get("bids", [])],
asks=[(float(p), float(s)) for p, s in data.get("asks", [])]
)
return None
async def _process_buffer(self, buffer: List):
"""Process batch ของ ticks"""
# แบ่งตามประเภท
trades = [t for t in buffer if isinstance(t, TickData)]
fundings = [f for f in buffer if isinstance(f, FundingRateData)]
depths = [d for d in buffer if isinstance(d, DepthSnapshot)]
if trades:
await self._process_trades(trades)
if fundings:
await self._process_fundings(fundings)
if depths:
await self._process_depths(depths)
async def _process_trades(self, trades: List[TickData]):
logger.debug(f"จำนวน trades: {len(trades)}")
# Implement trade processing logic
pass
async def _process_fundings(self, fundings: List[FundingRateData]):
logger.debug(f"จำนวน funding updates: {len(fundings)}")
# Implement funding processing logic
pass
async def _process_depths(self, depths: List[DepthSnapshot]):
logger.debug(f"จำนวน depth snapshots: {len(depths)}")
# Implement depth processing logic
pass
async def _handle_reconnect(self):
"""จัดการการ reconnect เมื่อ connection หลุด"""
if self._reconnect_count >= self.config.max_reconnect_attempts:
logger.error("เกินจำนวนครั้งที่กำหนดสำหรับ reconnect")
return
self._reconnect_count += 1
delay = self.config.reconnect_delay * (2 ** (self._reconnect_count - 1))
logger.info(f"พยายาม reconnect ใน {delay} วินาที (ครั้งที่ {self._reconnect_count})")
await asyncio.sleep(delay)
await self.connect()
# Resubscribe to all previous subscriptions
for exchange, subs in self.subscriptions.items():
for sub in subs:
parts = sub.split(":")
if len(parts) == 3:
await self.subscribe(parts[0], [parts[1]], [parts[2]])การจัดการ Funding Rate อย่างมีประสิทธิภาพ
Funding rate เป็นกลไกสำคัญที่ทำให้ราคา perpetual futures อยู่ใกล้ spot price มากที่สุด ใน OKX และ Bybit funding rate จะถูกคำนวณทุก 8 ชั่วโมง แต่การอัปเดตจะเกิดขึ้นบ่อยกว่านั้นมาก การติดตาม funding rate changes อย่างใกล้ชิดช่วยให้เข้าใจ sentiment ของตลาดและหาจังหวะ arbitrage ที่ดี
# funding_rate_tracker.py
import asyncio
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from collections import defaultdict
import pandas as pd
from dataclasses import dataclass, asdict
from tardis_client import FundingRateData, TardisDataClient, TardisConfig
@dataclass
class FundingRateSnapshot:
exchange: str
symbol: str
current_rate: float
predicted_rate: float
timestamp: datetime
hours_to_funding: float
annualised_rate: float
premium_index: float
interest_rate: float
class FundingRateTracker:
"""Track และวิเคราะห์ funding rate ข้าม exchanges"""
def __init__(self, storage_enabled: bool = True):
self.history: Dict[str, List[FundingRateData]] = defaultdict(list)
self.snapshots: Dict[str, List[FundingRateSnapshot]] = defaultdict(list)
self.rate_cache: Dict[str, float] = {}
self._window_size = 72 # 3 วันที่ผ่านมา
self._storage_enabled = storage_enabled
self._save_interval = 300 # 5 นาที
def process_funding_update(self, funding: FundingRateData):
"""Process funding rate update และคำนวณ derived metrics"""
key = f"{funding.exchange}:{funding.symbol}"
# เก็บ history
self.history[key].append(funding)
# รักษา window size
if len(self.history[key]) > self._window_size:
self.history[key] = self.history[key][-self._window_size:]
# คำนวณ snapshot
snapshot = self._calculate_snapshot(funding)
self.snapshots[key].append(snapshot)
# Cache ค่าปัจจุบัน
self.rate_cache[key] = funding.funding_rate
# ตรวจจับ funding rate spike
if self._detect_spike(key):
self._alert_funding_spike(key, snapshot)
def _calculate_snapshot(self, funding: FundingRateData) -> FundingRateSnapshot:
"""คำนวณ metrics ที่ derived จาก funding rate"""
hours_to_funding = (funding.next_funding_time - datetime.now(funding.timestamp.tzinfo)).total_seconds() / 3600
# Annualized rate (compounded 3 times daily)
annualised = ((1 + funding.funding_rate) ** (3 * 365)) - 1 if funding.funding_rate else 0
# Predicted rate จาก moving average
key = f"{funding.exchange}:{funding.symbol}"
predicted = self._predict_rate(key)
return FundingRateSnapshot(
exchange=funding.exchange,
symbol=funding.symbol,
current_rate=funding.funding_rate,
predicted_rate=predicted,
timestamp=funding.timestamp,
hours_to_funding=hours_to_funding,
annualised_rate=annualised,
premium_index=funding.funding_rate * 3, # Approximate
interest_rate=0.0001 # Bybit/OKX standard
)
def _predict_rate(self, key: str) -> float:
"""Predict funding rate จาก historical data ใช้ EMA"""
if len(self.history[key]) < 3:
return self.rate_cache.get(key, 0.0)
rates = [h.funding_rate for h in self.history[key][-24:]] # 24 ชั่วโมง
if not rates:
return 0.0
# EMA with alpha = 0.3
alpha = 0.3
ema = rates[0]
for rate in rates[1:]:
ema = alpha * rate + (1 - alpha) * ema
return ema
def _detect_spike(self, key: str) -> bool:
"""ตรวจจับ funding rate spike (เปลี่ยนแปลงเกิน 50% จาก average)"""
if len(self.history[key]) < 6:
return False
recent = self.history[key][-6:]
avg = sum(h.funding_rate for h in recent) / len(recent)
current = self.history[key][-1].funding_rate
if abs(avg) < 0.0001: # Near zero
return abs(current) > 0.001
change = abs((current - avg) / avg)
return change > 0.5
def _alert_funding_spike(self, key: str, snapshot: FundingRateSnapshot):
"""ส่ง alert เมื่อตรวจพบ funding spike"""
print(f"[ALERT] Funding Rate Spike Detected!")
print(f" Exchange: {snapshot.exchange}")
print(f" Symbol: {snapshot.symbol}")
print(f" Current: {snapshot.current_rate:.6f}")
print(f" Predicted: {snapshot.predicted_rate:.6f}")
print(f" Annualized: {snapshot.annualised_rate:.2%}")
def get_arbitrage_opportunity(self) -> List[Dict]:
"""หา arbitrage opportunity ระหว่าง OKX และ Bybit"""
opportunities = []
for symbol in ["BTC-USDT-SWAP", "ETH-USDT-SWAP"]:
okx_key = f"okx:{symbol}"
bybit_key = f"bybit:{symbol.replace('-USDT-SWAP', 'USDT')}"
okx_rate = self.rate_cache.get(okx_key)
bybit_rate = self.rate_cache.get(bybit_key)
if okx_rate is not None and bybit_rate is not None:
diff = abs(okx_rate - bybit_rate)
if diff > 0.0001: # มากกว่า 0.01%
opportunities.append({
"symbol": symbol,
"okx_rate": okx_rate,
"bybit_rate": bybit_rate,
"diff": diff,
"direction": "long_okx_short_bybit" if okx_rate > bybit_rate else "long_bybit_short_okx",
"annualised_spread": diff * 3 * 365
})
return opportunities
def calculate_funding_impact(self, position_size: float, exchange: str, symbol: str) -> Dict:
"""คำนวณผลกระทบของ funding rate ต่อ position"""
key = f"{exchange}:{symbol}"
rate = self.rate_cache.get(key, 0.0)
# Funding per 8 hours
funding_per_period = position_size * rate
# Daily funding
daily_funding = funding_per_period * 3
# Monthly funding
monthly_funding = daily_funding * 30
return {
"position_size": position_size,
"rate": rate,
"per_period": funding_per_period,
"daily": daily_funding,
"monthly": monthly_funding,
"annualised": monthly_funding * 12
}
async def run_funding_tracker():
"""Run funding rate tracker เป็น standalone process"""
config = TardisConfig(api_key=os.getenv("TARDIS_API_KEY"))
client = TardisDataClient(config)
tracker = FundingRateTracker()
await client.connect()
# Subscribe to funding channels
await client.subscribe("okx", ["BTC-USDT-SWAP", "ETH-USDT-SWAP"], ["funding_rate"])
await client.subscribe("bybit", ["BTCUSDT", "ETHUSDT"], ["funding"])
# Handler for funding updates
async def handle_funding(funding: FundingRateData):
tracker.process_funding_update(funding)
# Check for arbitrage
opp = tracker.get_arbitrage_opportunity()
if opp:
print(f"[ARBITRAGE] พบโอกาส: {opp}")
# Run with periodic reporting
while True:
await asyncio.sleep(60) # Report every minute
print(f"[STATUS] Tracking {len(tracker.rate_cache)} symbols")
print(f"[CACHE] Current rates: {tracker.rate_cache}")
import osการจัดการ Order Book Depth Snapshot
Depth snapshot จาก order book เป็นข้อมูลสำคัญสำหรับการคำนวณ slippage, market impact, และ liquidity analysis OKX และ Bybit มีรูปแบบการส่ง depth data ที่แตกต่างกัน ต้อง handle อย่างถูกต้องเพื่อให้ได้ข้อมูลที่ consistent
# depth_analyzer.py
import asyncio
from datetime import datetime, timedelta
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass, field
from collections import deque
import numpy as np
from tardis_client import DepthSnapshot, TardisDataClient, TardisConfig
@dataclass
class OrderBookLevel:
price: float
size: float
orders: int = 1
@dataclass
class ProcessedDepth:
exchange: str
symbol: str
timestamp: datetime
best_bid: float
best_ask: float
spread: float
spread_pct: float
mid_price: float
imbalance: float # Bid/Ask size ratio
depth_5: float # Combined size in top 5 levels
depth_20: float # Combined size in top 20 levels
weighted_mid: float
def to_dict(self) -> dict:
return {
"exchange": self.exchange,
"symbol": self.symbol,
"timestamp": self.timestamp.isoformat(),
"best_bid": self.best_bid,
"best_ask": self.best_ask,
"spread": self.spread,
"spread_pct": self.spread_pct,
"mid_price": self.mid_price,
"imbalance": self.imbalance,
"depth_5": self.depth_5,
"depth_20": self.depth_20,
"weighted_mid": self.weighted_mid
}
class DepthAnalyzer:
"""วิเคราะห์ order book depth สำหรับ OKX และ Bybit"""
# ค่าคงที่สำหรับแต่ละ exchange
OKX_LEVEL_PRECISION = {
"BTC-USDT-SWAP": 0.1,
"ETH-USDT-SWAP": 0.01
}
BYBIT_LEVEL_PRECISION = {
"BTCUSDT": 0.01,
"ETHUSDT": 0.01
}
def __init__(self, symbol: str):
self.symbol = symbol
self.bids: List[OrderBookLevel] = []
self.asks: List[OrderBookLevel] = []
self.last_update: Optional[datetime] = None
self.sequence: int = 0
# Rolling history สำหรับ analysis
self.history: deque = deque(maxlen=1000)
self.imbalance_history: deque = deque(maxlen=100)
def update_depth(self, snapshot: DepthSnapshot):
"""Update order book จาก depth snapshot"""
self.bids = [OrderBookLevel(price=p, size=s) for p, s in snapshot.bids]
self.asks = [OrderBookLevel(price=p, size=s) for p, s in snapshot.asks]
self.last_update = snapshot.timestamp
self.sequence += 1
# Calculate processed depth
processed = self._process_depth(snapshot)
self.history.append(processed)
# Track imbalance
self.imbalance_history.append(processed.imbalance)
def _process_depth(self, snapshot: DepthSnapshot) -> ProcessedDepth:
"""Process raw depth เป็น derived metrics"""
bids = [(p, s) for p, s in snapshot.bids[:20]]
asks = [(p, s) for p, s in snapshot.asks[:20]]
if not bids or not asks:
return None
best_bid = bids[0][0]
best_ask = asks[0][0]
spread = best_ask - best_bid
spread_pct = spread / ((best_ask + best_bid) / 2) if best_bid and best_ask else 0
mid_price = (best_bid + best_ask) / 2
# Calculate imbalance (bid/ask ratio)
bid_size = sum(s for _, s in bids[:5])
ask_size = sum(s for _, s in asks[:5])
imbalance = (bid_size - ask_size) / (bid_size + ask_size) if (bid_size + ask_size) > 0 else 0
# Depth calculations
depth_5 = sum(s for _, s in bids[:5]) + sum(s for _, s in asks[:5])
depth_20 = sum(s for _, s in bids) + sum(s for _, s in asks)
# Volume-weighted mid price
total_bid_vol = sum(s for _, s in bids[:5])
total_ask_vol = sum(s for _, s in asks[:5])
bid_prices = [p for p, _ in bids[:5]]
ask_prices = [p for p, _ in asks[:5]]
if total_bid_vol + total_ask_vol > 0:
weighted_mid = (
sum(p * s for p, s in bids[:5]) + sum(p * s for p, s in asks[:5])
) / (total_bid_vol + total_ask_vol)
else:
weighted_mid = mid_price
return ProcessedDepth(
exchange=snapshot.exchange,
symbol=snapshot.symbol,
timestamp=snapshot.timestamp,
best_bid=best_bid,
best_ask=best_ask,
spread=spread,
spread_pct=spread_pct,
mid_price=mid_price,
imbalance=imbalance,
depth_5=depth_5,
depth_20=depth_20,
weighted_mid=weighted_mid
)
def calculate_slippage(self, side: str, size: float) -> Tuple[float, float]:
"""คำนวณ slippage สำหรับ order ขนาด size"""
if side == "buy":
levels = self.asks
else:
levels = self.bids
remaining_size = size
total_cost = 0
filled_size = 0
for level in levels:
fill_size = min(remaining_size, level.size)
total_cost += fill_size * level.price
filled_size += fill_size
remaining_size -= fill_size
if remaining_size <= 0:
break
if filled_size == 0:
return 0, float('inf')
avg_price = total_cost / filled_size
slippage = abs(avg_price - self.mid_price) if self.mid_price else 0
return slippage, slippage / self.mid_price if self.mid_price else 0