ในโลกของ High-Frequency Trading หรือ HFT โดยเฉพาะอย่างยิ่ง Market Making Strategy ที่ต้องการความแม่นยำในการวาง Bid/Ask และ Dynamic Hedging การเลือกใช้ Exchange API ที่เหมาะสมคือหัวใจสำคัญของความสำเร็จ บทความนี้จะพาคุณเจาะลึกการ Integrate OKX Futures API เพื่อสร้างระบบ Market Making ที่พร้อมใช้งานจริงในระดับ Production
ภาพรวมสถาปัตยกรรมระบบ Market Making
Market Making Strategy ที่มีประสิทธิภาพต้องอาศัย 3 ส่วนหลักที่ทำงานร่วมกันอย่างลงตัว ส่วนแรกคือ Data Ingestion Layer ที่รับ Market Data ผ่าน WebSocket แบบ Real-time ส่วนที่สองคือ Strategy Engine ที่คำนวณ Bid/Ask Prices และ Position Management ส่วนสุดท้ายคือ Execution Layer ที่ส่ง Orders ไปยัง Exchange ด้วยความเร็วสูงสุด
┌─────────────────────────────────────────────────────────────────┐
│ MARKET MAKING ARCHITECTURE │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ WebSocket ┌───────────────────────┐ │
│ │ OKX Futures │ ──────────────► │ Market Data Buffer │ │
│ │ API │ wss://... │ (Order Book Cache) │ │
│ └──────────────┘ └───────────┬───────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ ┌───────────────────────┐ │
│ │ Position │ ◄────────────── │ Strategy Engine │ │
│ │ Manager │ Risk Check │ (Bid/Ask Calculator) │ │
│ └──────┬───────┘ └───────────┬───────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌──────────────┐ ┌───────────────────────┐ │
│ │ Risk │ │ Order Executor │ │
│ │ Monitor │ │ (REST API + WS) │ │
│ └──────────────┘ └───────────┬───────────┘ │
│ │ │
└───────────────────────────────────────────────┼───────────────────┘
│
▼
┌───────────────────────┐
│ HolySheep AI │
│ (Auxiliary Services) │
│ - Sentiment Analysis │
│ - Price Prediction │
│ - Risk Assessment │
└───────────────────────┘
การเชื่อมต่อ WebSocket สำหรับ Real-time Market Data
OKX Futures API มี WebSocket Endpoint ที่รองรับการรับ Order Book, Trade Data และ Ticker Information แบบ Real-time โดยมี Latency เฉลี่ยอยู่ที่ประมาณ 2-5ms จาก Exchange ไปยัง Server ของคุณ สิ่งสำคัญคือต้องใช้ WebSocket แทน REST API สำหรับ Market Data เพราะ REST API มี Latency สูงกว่าถึง 5-10 เท่าในบางกรณี
import asyncio
import json
import hmac
import hashlib
import time
from typing import Callable, Dict, List, Optional
from dataclasses import dataclass, field
from collections import deque
import threading
@dataclass
class OrderBookEntry:
price: float
quantity: float
orders: int = 1
@dataclass
class OrderBook:
bids: List[OrderBookEntry] = field(default_factory=list)
asks: List[OrderBookEntry] = field(default_factory=list)
timestamp: int = 0
def get_mid_price(self) -> float:
if not self.bids or not self.asks:
return 0.0
return (self.bids[0].price + self.asks[0].price) / 2
def get_spread(self) -> float:
if not self.bids or not self.asks:
return 0.0
return self.asks[0].price - self.bids[0].price
def get_spread_bps(self) -> float:
mid = self.get_mid_price()
if mid == 0:
return 0.0
return (self.get_spread() / mid) * 10000
class OKXFuturesWebSocket:
"""
High-performance WebSocket client สำหรับ OKX Futures API
รองรับการรับ Order Book, Trade และ Ticker Data
"""
WS_URL = "wss://ws.okx.com:8443/ws/v5/public"
def __init__(
self,
api_key: str = "",
api_secret: str = "",
passphrase: str = "",
testnet: bool = False
):
self.ws_url = self.WS_URL
self.api_key = api_key
self.api_secret = api_secret
self.passphrase = passphrase
self.order_books: Dict[str, OrderBook] = {}
self.recent_trades: Dict[str, deque] = {}
self.tickers: Dict[str, Dict] = {}
self._ws = None
self._running = False
self._subscriptions: List[Dict] = []
self._reconnect_delay = 1
self._max_reconnect_delay = 60
self._callbacks: Dict[str, List[Callable]] = {
'orderbook': [],
'trade': [],
'ticker': []
}
self._lock = threading.RLock()
self._last_ping_time = 0
def subscribe(self, channel: str, inst_id: str):
"""สมัครรับข้อมูลจากช่องทางที่ต้องการ"""
sub = {
"op": "subscribe",
"args": [{
"channel": channel,
"instId": inst_id
}]
}
with self._lock:
self._subscriptions.append(sub)
if self._ws and self._running:
self._send_json(sub)
def _send_json(self, data: dict):
"""ส่งข้อมูลผ่าน WebSocket"""
if self._ws:
self._ws.send(json.dumps(data))
async def connect(self):
"""เชื่อมต่อ WebSocket"""
import websockets
self._running = True
while self._running:
try:
async with websockets.connect(
self.ws_url,
ping_interval=20,
ping_timeout=10
) as ws:
self._ws = ws
# Resubscribe to all channels
for sub in self._subscriptions:
await ws.send(json.dumps(sub))
# Reset reconnect delay on successful connection
self._reconnect_delay = 1
# Start ping task
ping_task = asyncio.create_task(self._ping_loop(ws))
async for message in ws:
await self._handle_message(message)
except Exception as e:
print(f"WebSocket connection error: {e}")
await asyncio.sleep(self._reconnect_delay)
self._reconnect_delay = min(
self._reconnect_delay * 2,
self._max_reconnect_delay
)
async def _ping_loop(self, ws):
"""รักษา connection ให้ active ด้วย ping/pong"""
while self._running:
try:
await asyncio.sleep(20)
await ws.send(json.dumps({"op": "ping"}))
self._last_ping_time = time.time()
except Exception:
break
async def _handle_message(self, message: str):
"""ประมวลผลข้อความจาก WebSocket"""
try:
data = json.loads(message)
# Handle subscription confirmation
if "event" in data:
if data["event"] == "subscribe":
print(f"Subscribed: {data}")
return
# Handle data messages
if "data" in data:
arg = data.get("arg", {})
channel = arg.get("channel", "")
if channel == "books":
await self._handle_orderbook(data["data"])
elif channel == "trades":
await self._handle_trade(data["data"])
elif channel == "tickers":
await self._handle_ticker(data["data"])
except Exception as e:
print(f"Error handling message: {e}")
async def _handle_orderbook(self, data: List[Dict]):
"""ประมวลผล Order Book Data"""
for entry in data:
inst_id = entry.get("instId", "")
bids = []
asks = []
for bid in entry.get("bids", []):
bids.append(OrderBookEntry(
price=float(bid[0]),
quantity=float(bid[1]),
orders=int(bid[2]) if len(bid) > 2 else 1
))
for ask in entry.get("asks", []):
asks.append(OrderBookEntry(
price=float(ask[0]),
quantity=float(ask[1]),
orders=int(ask[2]) if len(ask) > 2 else 1
))
self.order_books[inst_id] = OrderBook(
bids=bids,
asks=asks,
timestamp=int(time.time() * 1000)
)
# Trigger callbacks
for callback in self._callbacks['orderbook']:
await callback(inst_id, self.order_books[inst_id])
async def _handle_trade(self, data: List[Dict]):
"""ประมวลผล Trade Data"""
for entry in data:
inst_id = entry.get("instId", "")
if inst_id not in self.recent_trades:
self.recent_trades[inst_id] = deque(maxlen=1000)
trade = {
"inst_id": inst_id,
"price": float(entry["px"]),
"quantity": float(entry["sz"]),
"side": entry["side"],
"timestamp": int(entry["ts"]),
"trade_id": entry.get("tradeId", "")
}
self.recent_trades[inst_id].append(trade)
for callback in self._callbacks['trade']:
await callback(trade)
async def _handle_ticker(self, data: List[Dict]):
"""ประมวลผล Ticker Data"""
for entry in data:
inst_id = entry.get("instId", "")
self.tickers[inst_id] = {
"last_price": float(entry.get("last", 0)),
"bid_price": float(entry.get("bidPx", 0)),
"ask_price": float(entry.get("askPx", 0)),
"volume_24h": float(entry.get("vol24h", 0)),
"timestamp": int(entry.get("ts", 0))
}
for callback in self._callbacks['ticker']:
await callback(inst_id, self.tickers[inst_id])
def on_orderbook(self, callback: Callable):
"""Register callback สำหรับ Order Book updates"""
self._callbacks['orderbook'].append(callback)
def on_trade(self, callback: Callable):
"""Register callback สำหรับ Trade updates"""
self._callbacks['trade'].append(callback)
def disconnect(self):
"""ยกเลิกการเชื่อมต่อ"""
self._running = False
if self._ws:
self._ws.close()
ตัวอย่างการใช้งาน
async def main():
ws = OKXFuturesWebSocket()
async def on_orderbook_update(inst_id: str, orderbook: OrderBook):
print(f"{inst_id}: Mid Price = {orderbook.get_mid_price():.4f}, "
f"Spread = {orderbook.get_spread_bps():.2f} bps")
ws.on_orderbook(on_orderbook_update)
ws.subscribe("books", "BTC-USD-SWAP")
ws.subscribe("books", "ETH-USD-SWAP")
await ws.connect()
if __name__ == "__main__":
asyncio.run(main())
Market Making Strategy Implementation
หัวใจของ Market Making Strategy คือการคำนวณ Bid/Ask Prices ที่เหมาะสม โดยคำนึงถึงหลายปัจจัย ได้แก่ Current Market Spread, Inventory Risk, Order Book Imbalance และ Volatility ระบบที่ดีต้องสามารถปรับ Spread แบบ Dynamic ตามสภาพตลาด และมีการจัดการ Position ที่เหมาะสมเพื่อไม่ให้เกิดความเสี่ยงจากการกลับตัวของราคา
import numpy as np
from typing import Dict, Tuple, Optional
from dataclasses import dataclass
from enum import Enum
import math
class PositionSide(Enum):
LONG = "long"
SHORT = "short"
FLAT = "flat"
@dataclass
class MarketMakingConfig:
# Spread Configuration
base_spread_bps: float = 5.0 # Base spread in basis points
min_spread_bps: float = 2.0 # Minimum spread
max_spread_bps: float = 50.0 # Maximum spread
# Inventory Management
max_position_size: float = 100000.0 # Maximum position in USD
inventory_target: float = 0.0 # Target inventory (0 = neutral)
inventory_penalty: float = 0.0001 # Penalty for inventory deviation
# Risk Parameters
max_orders_per_side: int = 5 # Maximum orders per side
order_size_usd: float = 1000.0 # Size per order in USD
max_daily_loss: float = 10000.0 # Maximum daily loss
# Volatility Adjustment
volatility_window: int = 100 # Window for volatility calculation
volatility_multiplier: float = 1.5 # Multiplier for volatility adjustment
# Time-based Adjustment
quote_refresh_ms: int = 100 # How often to requote
class MarketMakingStrategy:
"""
Market Making Strategy พร้อม Dynamic Spread Adjustment
และ Inventory Risk Management
"""
def __init__(
self,
config: MarketMakingConfig,
ws_client: 'OKXFuturesWebSocket'
):
self.config = config
self.ws = ws_client
# State
self.current_position: float = 0.0
self.position_side: PositionSide = PositionSide.FLAT
self.daily_pnl: float = 0.0
self.trade_count: int = 0
# Price history for volatility calculation
self.price_history: Dict[str, list] = {}
# Active orders tracking
self.active_bids: Dict[str, dict] = {}
self.active_asks: Dict[str, dict] = {}
# Lock for thread safety
import threading
self._lock = threading.Lock()
def calculate_adjusted_spread(
self,
inst_id: str,
orderbook: 'OrderBook'
) -> float:
"""
คำนวณ Spread ที่ปรับตามสภาพตลาด
- Inventory risk: ถ้ามี Position เยอะ ต้องการ Spread สูงขึ้น
- Volatility: ถ้าตลาดผันผวน Spread ต้องกว้างขึ้น
- Order book imbalance: ถ้า Bids มากกว่า Asks อย่างมาก Spread ต้องกว้างขึ้น
"""
mid_price = orderbook.get_mid_price()
if mid_price == 0:
return self.config.base_spread_bps
# 1. Base spread from config
spread_bps = self.config.base_spread_bps
# 2. Inventory adjustment
inventory_ratio = self.current_position / self.config.max_position_size
inventory_adjustment = abs(inventory_ratio) * self.config.inventory_penalty * 10000
spread_bps += inventory_adjustment
# 3. Volatility adjustment
if inst_id in self.price_history:
prices = self.price_history[inst_id]
if len(prices) >= 10:
returns = np.diff(prices) / prices[:-1]
volatility = np.std(returns) * math.sqrt(1440) * 10000 # Annualized bps
volatility_adjustment = volatility * self.config.volatility_multiplier
spread_bps += volatility_adjustment
# 4. Order book imbalance adjustment
if orderbook.bids and orderbook.asks:
bid_volume = sum(b.quantity for b in orderbook.bids[:10])
ask_volume = sum(a.quantity for a in orderbook.asks[:10])
if bid_volume + ask_volume > 0:
imbalance = (bid_volume - ask_volume) / (bid_volume + ask_volume)
imbalance_adjustment = abs(imbalance) * 10 # Up to 10 bps
spread_bps += imbalance_adjustment
# 5. Apply min/max bounds
spread_bps = max(
self.config.min_spread_bps,
min(spread_bps, self.config.max_spread_bps)
)
return spread_bps
def calculate_quote_prices(
self,
inst_id: str,
orderbook: 'OrderBook'
) -> Tuple[Optional[float], Optional[float]]:
"""
คำนวณ Bid และ Ask Prices สำหรับการวาง Order
สูตร:
Bid = Mid Price * (1 - Spread/2)
Ask = Mid Price * (1 + Spread/2)
"""
mid_price = orderbook.get_mid_price()
if mid_price == 0:
return None, None
spread_bps = self.calculate_adjusted_spread(inst_id, orderbook)
half_spread = spread_bps / 2 / 10000 # Convert bps to decimal
bid_price = mid_price * (1 - half_spread)
ask_price = mid_price * (1 + half_spread)
# Round to tick size (OKX uses 0.1 for BTC-USD-SWAP)
tick_size = 0.1
bid_price = math.floor(bid_price / tick_size) * tick_size
ask_price = math.ceil(ask_price / tick_size) * tick_size
# Ensure bid < ask
if bid_price >= ask_price:
bid_price = ask_price - tick_size
return bid_price, ask_price
def calculate_order_size(self, inst_id: str) -> float:
"""
คำนวณขนาด Order โดยคำนึงถึง:
- Available position limit
- Current inventory
- Risk appetite
"""
remaining_capacity = self.config.max_position_size - abs(self.current_position)
# Base size
size = self.config.order_size_usd
# Reduce size if near position limit
if remaining_capacity < size:
size = max(remaining_capacity * 0.5, 0)
# Reduce size if daily loss is high
loss_ratio = abs(self.daily_pnl) / self.config.max_daily_loss
if loss_ratio > 0.5:
size *= (1 - loss_ratio)
return size
def should_quote(self, inst_id: str) -> bool:
"""
ตัดสินใจว่าควรวาง Quote หรือไม่
"""
# Check daily loss limit
if self.daily_pnl < -self.config.max_daily_loss:
return False
# Check position limit
if abs(self.current_position) >= self.config.max_position_size:
return False
# Check if market is liquid enough
orderbook = self.ws.order_books.get(inst_id)
if not orderbook or not orderbook.bids or not orderbook.asks:
return False
# Check spread is reasonable
if orderbook.get_spread_bps() < self.config.min_spread_bps * 0.5:
return False
return True
def update_position(self, trade: dict):
"""
Update position หลังจาก Trade เกิดขึ้น
"""
with self._lock:
inst_id = trade['inst_id']
price = trade['price']
quantity = trade['quantity']
side = trade['side']
# Calculate position change
if side == 'buy':
position_change = quantity
else:
position_change = -quantity
self.current_position += position_change
self.position_side = PositionSide.LONG if self.current_position > 0 else \
PositionSide.SHORT if self.current_position < 0 else PositionSide.FLAT
# Update PnL
# สมมติว่า trades มีการ close position ที่ทำให้เกิด PnL
# ในระบบจริงต้อง track entry price ด้วย
self.daily_pnl += position_change * price * 0.0001 # Simplified fee impact
self.trade_count += 1
async def generate_quotes(self, inst_id: str) -> Dict:
"""
สร้าง Quote สำหรับ Market Making
Returns:
dict with 'bid' and 'ask' keys containing order details
"""
if not self.should_quote(inst_id):
return {'bid': None, 'ask': None}
orderbook = self.ws.order_books.get(inst_id)
if not orderbook:
return {'bid': None, 'ask': None}
bid_price, ask_price = self.calculate_quote_prices(inst_id, orderbook)
size = self.calculate_order_size(inst_id)
if size <= 0:
return {'bid': None, 'ask': None}
return {
'bid': {
'inst_id': inst_id,
'side': 'buy',
'price': bid_price,
'size': size,
'type': 'limit'
},
'ask': {
'inst_id': inst_id,
'side': 'sell',
'price': ask_price,
'size': size,
'type': 'limit'
}
}
ตัวอย่างการใช้งาน Strategy
async def run_market_maker():
from okx_futures_ws import OKXFuturesWebSocket, OrderBook
ws = OKXFuturesWebSocket()
config = MarketMakingConfig(
base_spread_bps=5.0,
min_spread_bps=2.0,
max_spread_bps=30.0,
max_position_size=100000,
order_size_usd=5000,
max_daily_loss=5000
)
strategy = MarketMakingStrategy(config, ws)
async def on_orderbook(inst_id: str, orderbook: OrderBook):
# Update price history
if inst_id not in strategy.price_history:
strategy.price_history[inst_id] = []
strategy.price_history[inst_id].append(orderbook.get_mid_price())
# Keep only recent prices
if len(strategy.price_history[inst_id]) > strategy.config.volatility_window:
strategy.price_history[inst_id] = \
strategy.price_history[inst_id][-strategy.config.volatility_window:]
# Generate quotes
quotes = await strategy.generate_quotes(inst_id)
if quotes['bid']:
print(f"Quote: Bid @ {quotes['bid']['price']:.2f}, "
f"Ask @ {quotes['ask']['price']:.2f}")
print(f"Position: {strategy.current_position:.2f}, "
f"Daily PnL: ${strategy.daily_pnl:.2f}")
ws.on_orderbook(on_orderbook)
ws.subscribe("books", "BTC-USD-SWAP")
await ws.connect()
if __name__ == "__main__":
asyncio.run(run_market_maker())
การเพิ่มประสิทธิภาพ Latency และ Throughput
สำหรับ Market Making ในระดับ Production ที่ต้องการความเร็วในการตอบสนองต่อตลาดภายใน 10ms การ Optimize Latency คือสิ่งจำเป็น ในส่วนนี้จะแนะนำเทคนิคต่างๆ ที่ช่วยลด Latency ได้อย่างมีนัยสำคัญ โดยเริ่มจากการเลือก Data Center Location ที่ใกล้กับ OKX Server มากที่สุด ไปจนถึงการใช้เทคนิค Zero-Copy ในการประมวลผลข้อมูล
การจัดการ Concurrency และ Thread Safety
ระบบ Market Making ต้องจัดการหลาย Tasks พร้อมกัน ได้แก่ การรับ Market Data, การคำนวณ Strategy, การส่ง Orders และการ Monitor Risks การใช้ Asyncio อย่างถูกต้องจะช่วยให้ระบบทำงานได้อย่างมีประสิทธิภาพโดยไม่มี Blocking Operations
import asyncio
import uvloop
from typing import Dict, List, Optional
from dataclasses import dataclass, field
from collections import defaultdict
import time
import statistics
@dataclass
class PerformanceMetrics:
"""Metrics สำหรับวัดประสิทธิภาพระบบ"""
latencies: List[float] = field(default_factory=list)
throughput: int = 0
errors: int = 0
start_time: float = field(default_factory=time.time)
def record_latency(self, latency_ms: float):
self.latencies.append(latency_ms)
def get_stats(self) -> Dict:
if not self.latencies:
return {
'p50': 0, 'p95': 0, 'p99': 0,
'avg': 0, 'throughput': 0
}
sorted_latencies = sorted(self.latencies)
n = len(sorted_latencies)
return {
'p50': sorted_latencies[int(n * 0.50)],
'p95': sorted_latencies[int(n * 0.95)],
'p99': sorted_latencies[int(n * 0.99)],
'avg': statistics.mean(self.latencies),
'min': min(sorted_latencies),
'max': max(sorted_latencies),
'throughput': self.throughput / (time.time() - self.start_time)
}
class OrderExecutor:
"""
High-Performance Order Executor พร้อม:
- Rate limiting
- Order management
- Retry logic with exponential backoff
- Concurrent order processing
"""
def __init__(
self,
api_key: str,
api_secret: str,
passphrase: str,
testnet: bool = False
):
self.api_key = api_key
self.api_secret = api_secret
self.passphrase = passphrase
self.base_url = "https://www.okx.com" if not testnet else "https://www.okx.com"
# Rate limiting
self.requests_per_second = 100
self.order_per_second = 20
self._request_timestamps: List[float] = []
self._order_timestamps: List[float] = []
# Order tracking
self.pending_orders: Dict[str, dict] = {}
self.filled_orders: Dict[str, dict] = {}
self.order_lock = asyncio.Lock()
# Metrics
self.metrics = Performance