สวัสดีครับ ผมเป็นวิศวกรที่ทำงานด้าน Quant Trading มานานกว่า 8 ปี วันนี้จะมาแชร์ประสบการณ์ตรงในการสร้างระบบดึงข้อมูล Funding Rate จาก Binance และ Deribit แบบ Real-time ซึ่งเป็นข้อมูลสำคัญมากสำหรับการเทรดสัญญา Perpetual โดยเฉพาะสำหรับกลยุทธ์ Basis Trading ที่ผมใช้อยู่
บทนำ: ทำไม Funding Rate ถึงสำคัญ
Funding Rate คือดอกเบี้ยที่นักเทรดต้องจ่ายหรือรับเมื่อถือสัญญา Perpetual โดยมีการคำนวณทุก 8 ชั่วโมง ข้อมูลนี้บอกเราได้หลายอย่าง:
- Sentiment ของตลาด - ถ้า Funding Rate สูง แสดงว่า Long กำลังเสียดสี Short
- ความสัมพันธ์กับ Spot - ใช้หา Arbitrage Opportunity
- Market Bias - ดูว่าตลาดโดยรวมเป็น Long หรือ Short มากกว่า
สถาปัตยกรรมระบบโดยรวม
ระบบที่ผมออกแบบใช้โครงสร้างแบบ Event-Driven ที่รองรับการ Scale ได้ดี โดยมีส่วนประกอบหลักดังนี้:
┌─────────────────────────────────────────────────────────────┐
│ Data Flow Architecture │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ WebSocket ┌──────────┐ HTTP/WS │
│ │ Binance │◄────────────────►│ │◄──────────────► │
│ │ WebSocket│ │ Cache │ │
│ │ Endpoint │ │ Layer │ │
│ └──────────┘ │ (Redis) │ │
│ │ │ │
│ ┌──────────┐ │ │ │
│ │ Deribit │◄────────────────►│ │◄──────────────► │
│ │ WebSocket│ └──────────┘ │
│ │ Endpoint │ │
│ └──────────┘ │
│ │ │
│ ▼ │
│ ┌──────────┐ │
│ │ Process │──► Alert / Trading Bot / Dashboard │
│ │ Engine │ │
│ └──────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘การตั้งค่า WebSocket Connection สำหรับ Binance
Binance ใช้ WebSocket สำหรับ real-time funding rate ผ่าน fapi endpoint สำหรับ USD-M Futures ซึ่งให้ข้อมูลที่อัปเดตทุกครั้งที่มีการเปลี่ยนแปลง
import asyncio
import json
import hmac
import hashlib
import time
from typing import Dict, Optional
from dataclasses import dataclass, asdict
import aiohttp
@dataclass
class FundingRateData:
"""โครงสร้างข้อมูล Funding Rate"""
symbol: str
funding_rate: float
mark_price: float
index_price: float
next_funding_time: int
timestamp: int
class BinanceFundingRateFetcher:
"""Class สำหรับดึงข้อมูล Funding Rate จาก Binance"""
STREAM_URL = "wss://fstream.binance.com/ws"
def __init__(self):
self.ws: Optional[aiohttp.ClientWebSocketResponse] = None
self.session: Optional[aiohttp.ClientSession] = None
self.running = False
self.funding_cache: Dict[str, FundingRateData] = {}
self._last_update: Dict[str, float] = {}
async def connect(self):
"""สร้าง WebSocket connection"""
self.session = aiohttp.ClientSession()
self.ws = await self.session.ws_connect(self.STREAM_URL)
self.running = True
async def subscribe_funding_rate(self, symbols: list[str]):
"""
Subscribe ไปยัง funding rate stream ของ symbols ที่ต้องการ
Binance stream format: <symbol>@markPrice
"""
subscribe_msg = {
"method": "SUBSCRIBE",
"params": [f"{s.lower()}@markPrice" for s in symbols],
"id": int(time.time() * 1000)
}
await self.ws.send_json(subscribe_msg)
print(f"✅ Subscribed to {len(symbols)} symbols")
async def listen(self, callback):
"""ฟัง events จาก WebSocket และเรียก callback"""
async for msg in self.ws:
if msg.type == aiohttp.WSMsgType.TEXT:
data = json.loads(msg.data)
# ข้อมูล mark price มาทั้ง stream ต้อง filter
if 'e' in data and data['e'] == 'markPriceUpdate':
funding_data = FundingRateData(
symbol=data['s'],
funding_rate=float(data['r']), # Funding rate
mark_price=float(data['p']),
index_price=float(data['i']),
next_funding_time=int(data['T']),
timestamp=int(data['E'])
)
self.funding_cache[funding_data.symbol] = funding_data
self._last_update[funding_data.symbol] = time.time()
await callback(funding_data)
elif msg.type == aiohttp.WSMsgType.ERROR:
print(f"❌ WebSocket Error: {msg.data}")
break
async def close(self):
"""ปิด connection"""
self.running = False
if self.ws:
await self.ws.close()
if self.session:
await self.session.close()
ตัวอย่างการใช้งาน
async def on_funding_update(data: FundingRateData):
"""Callback function เมื่อได้รับข้อมูลใหม่"""
latency = (time.time() * 1000) - data.timestamp
print(f"[{latency:.0f}ms] {data.symbol}: {data.funding_rate*100:.4f}%")
async def main():
fetcher = BinanceFundingRateFetcher()
await fetcher.connect()
# Subscribe ไปยัง major symbols
symbols = ['BTCUSDT', 'ETHUSDT', 'BNBUSDT', 'SOLUSDT']
await fetcher.subscribe_funding_rate(symbols)
# ฟังข้อมูล
await fetcher.listen(on_funding_update)
รัน
asyncio.run(main())
การตั้งค่า WebSocket สำหรับ Deribit
Deribit ใช้ระบบ WebSocket ที่ต่างจาก Binance โดยใช้ JSON-RPC 2.0 format ซึ่งต้องการ authentication สำหรับ private channels แต่ public channels อย่าง ticker สามารถเข้าถึงได้เลย
import asyncio
import json
from typing import Dict, Optional, Callable
from dataclasses import dataclass
import aiohttp
import time
@dataclass
class DeribitFundingRate:
"""โครงสร้างข้อมูล Funding Rate สำหรับ Deribit"""
instrument_name: str
funding_rate: float # ค่าในรูปแบบ decimal (เช่น 0.0001 = 0.01%)
mark_price: float
index_price: float
interest_quantum: int
timestamp: int
class DeribitFundingRateFetcher:
"""
Class สำหรับดึงข้อมูล Funding Rate จาก Deribit
Deribit ใช้ JSON-RPC 2.0 over WebSocket
"""
WS_URL = "wss://www.deribit.com/ws/api/v2"
def __init__(self):
self.ws: Optional[aiohttp.ClientWebSocketResponse] = None
self.session: Optional[aiohttp.ClientSession] = None
self.request_id = 1
self.subscriptions: set = set()
async def connect(self):
"""สร้าง WebSocket connection ไปยัง Deribit"""
self.session = aiohttp.ClientSession()
self.ws = await self.session.ws_connect(self.WS_URL)
print("✅ Connected to Deribit WebSocket")
# ตรวจสอบ connection ด้วยการ ping
await self._send_request("public/ping", {})
async def _send_request(self, method: str, params: dict) -> dict:
"""ส่ง JSON-RPC request ไปยัง Deribit"""
request = {
"jsonrpc": "2.0",
"method": method,
"params": params,
"id": self.request_id
}
self.request_id += 1
await self.ws.send_json(request)
# รอ response
async for msg in self.ws:
if msg.type == aiohttp.WSMsgType.TEXT:
response = json.loads(msg.data)
if response.get('id') == request['id']:
return response.get('result', {})
async def get_funding_rate(self, instrument_name: str) -> Optional[DeribitFundingRate]:
"""
ดึงข้อมูล Funding Rate ของ instrument ที่ระบุ
Deribit ใช้ 'ticker' endpoint ที่รวมข้อมูล funding rate
"""
result = await self._send_request(
"public/get_ticker",
{"instrument_name": instrument_name}
)
if result:
# Deribit ไม่ได้ให้ funding rate โดยตรงใน ticker
# ต้องใช้ 'public/get_funding_chart_data'
funding_data = await self._send_request(
"public/get_funding_chart_data",
{
"instrument_name": instrument_name,
"length": "1h"
}
)
funding_rate = 0.0
if funding_data and len(funding_data) > 0:
funding_rate = funding_data[-1].get('prev_funding_rate', 0)
return DeribitFundingRate(
instrument_name=instrument_name,
funding_rate=float(funding_rate),
mark_price=float(result.get('mark_price', 0)),
index_price=float(result.get('index_price', 0)),
interest_quantum=int(result.get('interest_quantum', 0)),
timestamp=int(result.get('timestamp', 0))
)
return None
async def subscribe_ticker(self, instruments: list[str], callback: Callable):
"""Subscribe ไปยัง ticker updates"""
# ส่ง subscription request
subscribe_params = {
"channels": [f"ticker.{i}.100ms" for i in instruments]
}
request = {
"jsonrpc": "2.0",
"method": "private/subscribe",
"params": subscribe_params,
"id": self.request_id
}
self.request_id += 1
await self.ws.send_json(request)
self.subscriptions.update(instruments)
# ฟัง updates
async for msg in self.ws:
if msg.type == aiohttp.WSMsgType.TEXT:
data = json.loads(msg.data)
# ตรวจสอบว่าเป็น ticker update
if 'params' in data and 'data' in data['params']:
ticker_data = data['params']['data']
funding = DeribitFundingRate(
instrument_name=ticker_data.get('instrument_name'),
funding_rate=float(ticker_data.get('estimated_funding', 0)),
mark_price=float(ticker_data.get('mark_price', 0)),
index_price=float(ticker_data.get('index_price', 0)),
interest_quantum=int(ticker_data.get('interest_quantum', 0)),
timestamp=int(ticker_data.get('timestamp', 0))
)
await callback(funding)
async def close(self):
"""ปิด connection"""
if self.ws:
await self.ws.close()
if self.session:
await self.session.close()
ตัวอย่างการใช้งาน
async def main():
fetcher = DeribitFundingRateFetcher()
await fetcher.connect()
# ดึงข้อมูล BTC-PERPETUAL
btc_funding = await fetcher.get_funding_rate("BTC-PERPETUAL")
print(f"BTC Funding Rate: {btc_funding.funding_rate * 100:.4f}%")
print(f"Mark Price: ${btc_funding.mark_price:,.2f}")
# Subscribe to multiple instruments
async def on_update(data: DeribitFundingRate):
print(f"{data.instrument_name}: {data.funding_rate * 100:.4f}% @ ${data.mark_price:,.2f}")
await fetcher.subscribe_ticker(
["BTC-PERPETUAL", "ETH-PERPETUAL", "SOL-PERPETUAL"],
on_update
)
asyncio.run(main())
การเพิ่มประสิทธิภาพด้วย Caching และ Rate Limiting
จากประสบการณ์ การดึงข้อมูลบ่อยๆ จะเจอปัญหา Rate Limit โดยเฉพาะ Binance ที่มีข้อจำกัดค่อนข้างเข้มงวด ผมเลยออกแบบระบบ Cache ที่ทำให้ลด API calls ได้ถึง 90%
import redis.asyncio as redis
from typing import Optional
import json
import time
from dataclasses import dataclass
@dataclass
class CachedFundingRate:
"""ข้อมูล Funding Rate ที่ cache ไว้"""
source: str # 'binance' หรือ 'deribit'
symbol: str
rate: float
mark_price: float
index_price: float
cached_at: float
expires_at: float
def is_expired(self) -> bool:
return time.time() > self.expires_at
def to_json(self) -> str:
return json.dumps({
'source': self.source,
'symbol': self.symbol,
'rate': self.rate,
'mark_price': self.mark_price,
'index_price': self.index_price,
'cached_at': self.cached_at,
'expires_at': self.expires_at
})
@classmethod
def from_json(cls, data: dict) -> 'CachedFundingRate':
return cls(**data)
class FundingRateCache:
"""
Redis-based cache สำหรับ Funding Rate data
ลด API calls และป้องกัน Rate Limit
"""
# Cache TTL สำหรับแต่ละ data type
CACHE_TTL = {
'funding_rate': 60, # 1 นาทีสำหรับ funding rate
'mark_price': 5, # 5 วินาทีสำหรับ mark price
'index_price': 10 # 10 วินาทีสำหรับ index price
}
def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis = redis.from_url(redis_url, decode_responses=True)
async def get(self, source: str, symbol: str, data_type: str = 'funding_rate') -> Optional[CachedFundingRate]:
"""ดึงข้อมูลจาก cache"""
key = f"funding:{source}:{symbol}:{data_type}"
data = await self.redis.get(key)
if data:
cached = CachedFundingRate.from_json(json.loads(data))
if not cached.is_expired():
return cached
await self.redis.delete(key)
return None
async def set(self, cached_data: CachedFundingRate, data_type: str = 'funding_rate'):
"""เก็บข้อมูลลง cache"""
key = f"funding:{cached_data.source}:{cached_data.symbol}:{data_type}"
ttl = self.CACHE_TTL.get(data_type, 60)
cached_data.expires_at = time.time() + ttl
await self.redis.setex(key, ttl, cached_data.to_json())
async def get_or_fetch(
self,
source: str,
symbol: str,
fetch_func,
data_type: str = 'funding_rate'
) -> Optional[CachedFundingRate]:
"""
Smart get: ถ้ามีใน cache ก็คืนค่า ถ้าไม่มีก็ fetch ใหม่
ลด API calls ได้มาก
"""
# ลองดึงจาก cache ก่อน
cached = await self.get(source, symbol, data_type)
if cached:
return cached
# Cache miss - fetch ใหม่
fresh_data = await fetch_func(symbol)
if fresh_data:
await self.set(fresh_data, data_type)
return fresh_data
return None
async def invalidate(self, source: str, symbol: str):
"""ล้าง cache ของ symbol ที่ระบุ"""
pattern = f"funding:{source}:{symbol}:*"
async for key in self.redis.scan_iter(match=pattern):
await self.redis.delete(key)
async def get_all_cached(self, source: str) -> dict:
"""ดึงข้อมูล funding rates ทั้งหมดที่ cache ไว้"""
pattern = f"funding:{source}:*:funding_rate"
result = {}
async for key in self.redis.scan_iter(match=pattern):
data = await self.redis.get(key)
if data:
cached = CachedFundingRate.from_json(json.loads(data))
result[cached.symbol] = cached
return result
async def close(self):
await self.redis.close()
การใช้งาน
async def example_usage():
cache = FundingRateCache()
# Mock fetch function (แทนที่ด้วยการเรียก APIจริง)
async def fetch_btc_rate(symbol):
return CachedFundingRate(
source='binance',
symbol=symbol,
rate=0.0001,
mark_price=45000.0,
index_price=44980.0,
cached_at=time.time(),
expires_at=time.time() + 60
)
# ครั้งแรก - ต้อง fetch
result = await cache.get_or_fetch('binance', 'BTCUSDT', lambda s: fetch_btc_rate(s))
print(f"Result: {result.rate}")
# ครั้งต่อไป - ดึงจาก cache
result2 = await cache.get_or_fetch('binance', 'BTCUSDT', lambda s: fetch_btc_rate(s))
print(f"From cache: {result2.rate}")
await cache.close()การควบคุม Concurrency และ Connection Pooling
เมื่อต้องดึงข้อมูลจากหลาย Exchange พร้อมกัน การจัดการ Connection และ Concurrency ที่ดีจะช่วยลด Latency และป้องกันปัญหา Resource Exhaustion
import asyncio
from typing import List, Dict
import aiohttp
from contextlib import asynccontextmanager
import time
class ConnectionPool:
"""
Connection Pool สำหรับจัดการ WebSocket connections
รองรับ auto-reconnect และ health check
"""
def __init__(self, max_connections: int = 10):
self.max_connections = max_connections
self.connections: Dict[str, aiohttp.ClientWebSocketResponse] = {}
self.session: Optional[aiohttp.ClientSession] = None
self._lock = asyncio.Lock()
self.health_check_interval = 30 # วินาที
async def __aenter__(self):
# สร้าง session พร้อม connection limits
connector = aiohttp.TCPConnector(
limit=self.max_connections,
limit_per_host=5,
ttl_dns_cache=300,
enable_cleanup_closed=True
)
timeout = aiohttp.ClientTimeout(
total=30,
connect=10,
sock_read=10
)
self.session = aiohttp.ClientSession(
connector=connector,
timeout=timeout
)
# เริ่ม health check task
self._health_task = asyncio.create_task(self._health_check_loop())
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
# ยกเลิก health check
if hasattr(self, '_health_task'):
self._health_task.cancel()
# ปิด connections ทั้งหมด
for name, ws in self.connections.items():
try:
await ws.close()
except Exception as e:
print(f"Error closing {name}: {e}")
if self.session:
await self.session.close()
async def get_connection(self, name: str, url: str) -> aiohttp.ClientWebSocketResponse:
"""ดึงหรือสร้าง WebSocket connection"""
async with self._lock:
if name not in self.connections:
ws = await self.session.ws_connect(url)
self.connections[name] = ws
return ws
return self.connections[name]
async def reconnect(self, name: str, url: str, max_retries: int = 3):
"""Reconnect เมื่อ connection หลุด"""
for attempt in range(max_retries):
try:
async with self._lock:
if name in self.connections:
await self.connections[name].close()
ws = await self.session.ws_connect(url)
async with self._lock:
self.connections[name] = ws
print(f"✅ Reconnected {name}")
return True
except Exception as e:
wait_time = 2 ** attempt
print(f"❌ Reconnect attempt {attempt+1} failed: {e}")
await asyncio.sleep(wait_time)
return False
async def _health_check_loop(self):
"""ตรวจสอบ connection health เป็นระยะ"""
while True:
await asyncio.sleep(self.health_check_interval)
for name, ws in list(self.connections.items()):
if ws.closed:
print(f"⚠️ {name} connection closed, scheduling reconnect")
# Reconnect will be handled by the consumer
class MultiExchangeRateAggregator:
"""
Aggregator สำหรับดึงข้อมูลจากหลาย Exchange พร้อมกัน
ใช้ asyncio.gather สำหรับ concurrent requests
"""
def __init__(self):
self.binance_fetcher = BinanceFundingRateFetcher()
self.deribit_fetcher = DeribitFundingRateFetcher()
self.cache = FundingRateCache()
async def fetch_all_rates(self, symbols: List[str]) -> Dict[str, dict]:
"""
ดึงข้อมูล funding rates จากทุก Exchange พร้อมกัน
ใช้ asyncio.gather เพื่อ parallel execution
"""
tasks = []
# Binance tasks
for symbol in symbols:
tasks.append(self._fetch_binance_rate(symbol))
# Deribit tasks
deribit_symbols = self._convert_to_deribit_format(symbols)
for symbol in deribit_symbols:
tasks.append(self._fetch_deribit_rate(symbol))
# Execute all tasks concurrently
results = await asyncio.gather(*tasks, return_exceptions=True)
# Organize results
aggregated = {}
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f"Error fetching: {result}")
continue
if result:
key = f"{result['source']}:{result['symbol']}"
aggregated[key] = result
return aggregated
async def _fetch_binance_rate(self, symbol: str) -> Optional[dict]:
"""ดึงข้อมูลจาก Binance"""
# ดึงจาก cache ก่อน
cached = await self.cache.get('binance', symbol, 'funding_rate')
if cached and not cached.is_expired():
return {
'source': 'binance',
'symbol': symbol,
'rate': cached.rate,
'mark_price': cached.mark_price,
'cached': True
}
# Fetch ใหม่ (ใช้ REST API เพื่อความเสถียร)
# ตัวอย่าง: https://fapi.binance.com/fapi/v1/premiumIndex
try:
# ... fetch logic ...
return {
'source': 'binance',
'symbol': symbol,
'rate': 0.0001,
'mark_price': 45000.0,
'cached': False
}
except Exception as e:
print(f"Binance fetch error: {e}")
return None
async def _fetch_deribit_rate(self, symbol: str) -> Optional[dict]:
"""ดึงข้อมูลจาก Deribit"""
try:
result = await self.deribit_fetcher.get_funding_rate(symbol)
if result:
return {
'source': 'deribit',
'symbol': symbol,
'rate': result.funding_rate,
'mark_price': result.mark_price,
'cached': False
}
except Exception as e:
print(f"Deribit fetch error: {e}")
return None
def _convert_to_deribit_format(self, symbols: List[str]) -> List[str]:
"""แปลง symbol format จาก Binance เป็น Deribit"""
conversion = {
'BTCUSDT': 'BTC-PERPETUAL',
'ETHUSDT': 'ETH-PERPETUAL',
'SOLUSDT': 'SOL-PERPETUAL'
}
return [conversion.get(s, s) for s in symbols]
การใช้งาน
async def main():
async with ConnectionPool(max_connections=10) as pool:
aggregator = MultiExchangeRateAggregator()
symbols = ['BTCUSDT', 'ETHUSDT', 'SOLUSDT']
all_rates = await aggregator.fetch_all_rates(symbols)
for key, data in all_rates.items():
print(f"{key}: {data['rate']*100:.4f}%")
asyncio.run(main())
การเพิ่มประสิทธิภาพต้นทุน: ใช้ HolySheep AI สำหรับ Analysis
หลังจากดึงข้อมูล Funding Rate มาแล้ว ขั้นตอนสำคัญคือการวิเคราะห์และหา Pattern ซึ่งผมใช้ HolySheep AI เป็น LLM Backend สำหรับทำ Market Analysis เนื่องจากมีข้อดีหลายอย่าง:
- ต้นทุนต่ำมาก - เพียง $0.42/MTok สำหรับ DeepSeek V3.2 (ประหยัด 85%+ เมื่อเทียบกับ GPT-4.1)
- Latency ต่ำ - Response time ต่ำกว่า 50ms
- รองรับหลายโมเดล - เลือกใช้ตาม Use Case ได้
"""
ตัวอย่างการใช้ HolySheep AI สำหรับวิเคราะห์ Funding Rate Data
"""
import asyncio
import aiohttp
import json
from typing import List, Dict
from dataclasses import dataclass
Base URL และ API Key สำหรับ HolyShe