Khi thị trường crypto ngày càng cạnh tranh, việc kết nối OKX API v5 một cách ổn định và hiệu quả trở thành yếu tố sống còn cho các nhà phát triển trading bot, data analyst và platform giao dịch. Bài viết này sẽ hướng dẫn chi tiết cách implement WebSocket multi-subscription, xử lý rate limiting thông minh và xây dựng hệ thống auto-reconnect bất khả chiến bại.

Kết luận ngắn: Nếu bạn cần tốc độ phản hồi dưới 50ms, chi phí API rẻ hơn 85% so với provider phương Tây, hãy cân nhắc đăng ký HolySheep AI — nền tảng hỗ trợ WeChat/Alipay với tín dụng miễn phí khi bắt đầu.

OKX API v5 có gì mới?

Phiên bản v5 của OKX mang đến nhiều cải tiến đáng chú ý:

So sánh HolySheep với OKX API chính thức và đối thủ

Tiêu chí OKX API chính thức HolySheep AI AWS API Gateway
Giá tham chiếu Miễn phí (có giới hạn) GPT-4.1: $8/MTok
Claude Sonnet 4.5: $15/MTok
DeepSeek V3.2: $0.42/MTok		 $3.50/MTok + data transfer
Độ trễ trung bình 80-150ms (AP region) <50ms (optimized routing) 100-200ms
Thanh toán Card quốc tế, Wire WeChat, Alipay, USDT Card quốc tế, AWS credits
Phương thức REST + WebSocket REST, Streaming, Webhook REST only
Độ phủ mô hình N/A (Exchange API) OpenAI, Anthropic, Google, DeepSeek Chỉ OpenAI
Nhóm phù hợp Dev crypto trading Dev cần chi phí thấp, thị trường AP Enterprise scaling

Phù hợp / không phù hợp với ai

✅ Nên dùng OKX API v5 khi:

❌ Không nên dùng khi:

Giá và ROI

Với chiến lược multi-subscription hiệu quả trên OKX API v5, bạn có thể giảm đáng kể số lượng connections cần thiết:

Package Giá Tính năng Phù hợp
OKX Free Tier $0 5 requests/2s, 20 subscription channels Học tập, testing
OKX VIP 1 $50/tháng 20 requests/2s, 100 channels Retail traders
HolySheep AI Từ $8/MTok <50ms, WeChat/Alipay, free credits Dev thị trường AP

Implement WebSocket Multi-Subscription OKX v5

Dưới đây là code implementation hoàn chỉnh với error handling và auto-reconnect:

// OKX WebSocket v5 Multi-Subscription với Auto-Reconnect
// File: okx_websocket_client.py

import asyncio
import json
import time
import hmac
import hashlib
import base64
import websockets
from typing import Dict, List, Optional, Callable
from datetime import datetime
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class OKXv5WebSocketClient:
    """
    OKX API v5 WebSocket Client với:
    - Multi-channel subscription trên 1 connection
    - Automatic reconnection với exponential backoff
    - Rate limit awareness
    - Heartbeat management
    """
    
    # OKX WebSocket endpoints
    WS_PUBLIC_URL = "wss://ws.okx.com:8443/ws/v5/public"
    WS_PRIVATE_URL = "wss://ws.okx.com:8443/ws/v5/private"
    
    # Rate limits OKX v5
    MAX_SUBSCRIPTIONS_PER_CONNECTION = 100
    MAX_MESSAGES_PER_SECOND = 50
    PING_INTERVAL = 20  # seconds
    PING_TIMEOUT = 10   # seconds
    
    def __init__(
        self,
        api_key: str = "",
        secret_key: str = "",
        passphrase: str = "",
        testnet: bool = False
    ):
        self.api_key = api_key
        self.secret_key = secret_key
        self.passphrase = passphrase
        
        # Connection state
        self.ws: Optional[websockets.WebSocketClientProtocol] = None
        self.is_connected = False
        self.subscriptions: Dict[str, dict] = {}
        
        # Reconnection config
        self.max_reconnect_attempts = 10
        self.base_reconnect_delay = 1  # seconds
        self.max_reconnect_delay = 60  # seconds
        
        # Message handlers
        self.handlers: Dict[str, List[Callable]] = {}
        
        # Rate limiting
        self.message_timestamps: List[float] = []
        self.rate_limit_lock = asyncio.Lock()
        
    def _get_sign(self, timestamp: str, method: str, path: str, body: str = "") -> str:
        """Generate HMAC SHA256 signature for OKX authentication"""
        message = timestamp + method + path + body
        mac = hmac.new(
            self.secret_key.encode('utf-8'),
            message.encode('utf-8'),
            hashlib.sha256
        )
        return base64.b64encode(mac.digest()).decode('utf-8')
    
    def _get_auth_params(self) -> dict:
        """Generate authentication parameters for private channels"""
        timestamp = datetime.utcnow().isoformat() + 'Z'
        signature = self._get_sign(timestamp, "GET", "/users/self/verify")
        
        return {
            "op": "login",
            "args": [
                {
                    "apiKey": self.api_key,
                    "passphrase": self.passphrase,
                    "timestamp": timestamp,
                    "sign": signature
                }
            ]
        }
    
    async def _check_rate_limit(self):
        """Check and enforce rate limiting"""
        async with self.rate_limit_lock:
            current_time = time.time()
            # Remove timestamps older than 1 second
            self.message_timestamps = [
                ts for ts in self.message_timestamps 
                if current_time - ts < 1
            ]
            
            if len(self.message_timestamps) >= self.MAX_MESSAGES_PER_SECOND:
                sleep_time = 1 - (current_time - self.message_timestamps[0])
                if sleep_time > 0:
                    logger.warning(f"Rate limit reached, sleeping {sleep_time:.2f}s")
                    await asyncio.sleep(sleep_time)
            
            self.message_timestamps.append(current_time)
    
    async def connect(self, private: bool = False) -> bool:
        """Establish WebSocket connection"""
        url = self.WS_PRIVATE_URL if private else self.WS_PUBLIC_URL
        
        try:
            self.ws = await websockets.connect(
                url,
                ping_interval=self.PING_INTERVAL,
                ping_timeout=self.PING_TIMEOUT
            )
            self.is_connected = True
            logger.info(f"Connected to OKX WebSocket: {url}")
            
            # Authenticate if private channel
            if private and self.api_key:
                await self._authenticate()
            
            return True
            
        except Exception as e:
            logger.error(f"Connection failed: {e}")
            self.is_connected = False
            return False
    
    async def _authenticate(self):
        """Send authentication login"""
        auth_params = self._get_auth_params()
        await self.ws.send(json.dumps(auth_params))
        
        response = await asyncio.wait_for(
            self.ws.recv(),
            timeout=10
        )
        
        data = json.loads(response)
        if data.get("event") == "login" and data.get("code") == "0":
            logger.info("Authentication successful")
        else:
            logger.error(f"Authentication failed: {data}")
            raise Exception("OKX authentication failed")
    
    async def subscribe(
        self, 
        channel: str, 
        inst_id: str = None,
        inst_type: str = "SPOT",
        uly: str = None,
        exp_time: str = None
    ):
        """
        Subscribe to a channel using OKX v5 format
        Supported channels: ticker, trades, books, candles, mark-price
        """
        if len(self.subscriptions) >= self.MAX_SUBSCRIPTIONS_PER_CONNECTION:
            logger.warning("Max subscriptions reached, consider reconnecting")
            return False
        
        # Build subscription args
        args = [{"channel": channel, "instType": inst_type}]
        
        if inst_id:
            args[0]["instId"] = inst_id
        if uly:
            args[0]["uly"] = uly
        if exp_time:
            args[0]["expTime"] = exp_time
        
        subscribe_msg = {
            "op": "subscribe",
            "args": args
        }
        
        await self._check_rate_limit()
        await self.ws.send(json.dumps(subscribe_msg))
        
        response = await asyncio.wait_for(self.ws.recv(), timeout=5)
        data = json.loads(response)
        
        if data.get("event") == "subscribe" and data.get("code") == "0":
            sub_key = f"{channel}:{inst_id or uly}"
            self.subscriptions[sub_key] = args[0]
            logger.info(f"Subscribed to {channel} for {inst_id or uly}")
            return True
        else:
            logger.error(f"Subscription failed: {data}")
            return False
    
    async def unsubscribe(self, channel: str, inst_id: str = None):
        """Unsubscribe from a channel"""
        args = [{"channel": channel, "instId": inst_id}] if inst_id else [{"channel": channel}]
        
        unsubscribe_msg = {
            "op": "unsubscribe",
            "args": args
        }
        
        await self._check_rate_limit()
        await self.ws.send(json.dumps(unsubscribe_msg))
        
        response = await asyncio.wait_for(self.ws.recv(), timeout=5)
        data = json.loads(response)
        
        if data.get("event") == "unsubscribe":
            sub_key = f"{channel}:{inst_id}"
            self.subscriptions.pop(sub_key, None)
            logger.info(f"Unsubscribed from {channel} for {inst_id}")
            return True
        return False
    
    def register_handler(self, channel: str, handler: Callable):
        """Register callback handler for a channel type"""
        if channel not in self.handlers:
            self.handlers[channel] = []
        self.handlers[channel].append(handler)
    
    async def listen(self):
        """Main listen loop with automatic reconnection"""
        reconnect_attempts = 0
        
        while True:
            try:
                if not self.is_connected:
                    if reconnect_attempts >= self.max_reconnect_attempts:
                        logger.error("Max reconnect attempts reached")
                        break
                    
                    delay = min(
                        self.base_reconnect_delay * (2 ** reconnect_attempts),
                        self.max_reconnect_delay
                    )
                    logger.info(f"Reconnecting in {delay}s (attempt {reconnect_attempts + 1})")
                    await asyncio.sleep(delay)
                    
                    if await self.connect():
                        # Re-subscribe all previous subscriptions
                        for sub_key, params in self.subscriptions.items():
                            channel = params["channel"]
                            inst_id = params.get("instId")
                            inst_type = params.get("instType")
                            await self.subscribe(channel, inst_id, inst_type)
                        
                        reconnect_attempts = 0
                    else:
                        reconnect_attempts += 1
                
                async for message in self.ws:
                    try:
                        data = json.loads(message)
                        await self._handle_message(data)
                    except json.JSONDecodeError:
                        logger.warning(f"Invalid JSON: {message}")
                        
            except websockets.ConnectionClosed as e:
                logger.warning(f"Connection closed: {e}")
                self.is_connected = False
                reconnect_attempts += 1
                
            except Exception as e:
                logger.error(f"Listen error: {e}")
                self.is_connected = False
                reconnect_attempts += 1
    
    async def _handle_message(self, data: dict):
        """Route incoming messages to appropriate handlers"""
        # Handle ping-pong
        if data.get("event") == "pong":
            return
        
        # Handle subscription confirmations
        if data.get("event") == "subscribe":
            return
        
        # Route data messages
        if "data" in data:
            arg = data.get("arg", {})
            channel = arg.get("channel")
            
            if channel in self.handlers:
                for handler in self.handlers[channel]:
                    await handler(data["data"])
    
    async def close(self):
        """Gracefully close connection"""
        if self.ws:
            await self.ws.close()
        self.is_connected = False
        logger.info("WebSocket connection closed")



========== USAGE EXAMPLE ==========


async def handle_ticker(tickers: List[dict]):
    """Handle incoming ticker data"""
    for ticker in tickers:
        logger.info(
            f"Ticker: {ticker['instId']} - "
            f"Last: {ticker['last']} - "
            f"Bid: {ticker['bidPx']} - "
            f"Ask: {ticker['askPx']}"
        )

async def handle_trades(trades: List[dict]):
    """Handle incoming trade data"""
    for trade in trades:
        logger.info(
            f"Trade: {trade['instId']} - "
            f"Side: {trade['side']} - "
            f"Price: {trade['px']} - "
            f"Size: {trade['sz']}"
        )

async def main():
    # Initialize client
    client = OKXv5WebSocketClient()
    
    # Register handlers
    client.register_handler("ticker", handle_ticker)
    client.register_handler("trades", handle_trades)
    
    # Connect and subscribe
    await client.connect(private=False)
    
    # Subscribe to multiple channels on single connection (OKX v5 feature)
    await client.subscribe("ticker", inst_id="BTC-USDT", inst_type="SPOT")
    await client.subscribe("ticker", inst_id="ETH-USDT", inst_type="SPOT")
    await client.subscribe("trades", inst_id="BTC-USDT", inst_type="SPOT")
    await client.subscribe("books", inst_id="BTC-USDT", inst_type="SPOT")
    
    # Start listening
    await client.listen()

if __name__ == "__main__":
    asyncio.run(main())

Implement Rate Limiter với Token Bucket Algorithm

OKX API v5 có rate limit phức tạp. Dưới đây là implementation rate limiter thông minh:

// OKX v5 Rate Limiter với Token Bucket + Exponential Backoff
// File: rate_limiter.ts

interface RateLimitConfig {
  requestsPerSecond: number;
  burstCapacity: number;
  endpointMultipliers: Record;
}

interface TokenBucket {
  tokens: number;
  lastRefill: number;
}

class OKXv5RateLimiter {
  private buckets: Map = new Map();
  private config: RateLimitConfig;
  private queue: Array<{
    endpoint: string;
    priority: number;
    resolve: (value: any) => void;
    reject: (error: Error) => void;
  }> = [];
  private processing = false;
  
  // OKX v5 specific rate limits
  private readonly LIMITS = {
    // REST API limits (requests per second by endpoint type)
    GET: { requestsPerSecond: 20, burstCapacity: 20 },
    GET_MARKET: { requestsPerSecond: 20, burstCapacity: 40 },
    GET_ACCOUNT: { requestsPerSecond: 10, burstCapacity: 10 },
    POST: { requestsPerSecond: 10, burstCapacity: 10 },
    POST_ORDER: { requestsPerSecond: 5, burstCapacity: 5 },
    
    // WebSocket limits
    WS_MESSAGES: { requestsPerSecond: 50, burstCapacity: 100 },
    WS_SUBSCRIPTIONS: { requestsPerSecond: 10, burstCapacity: 20 },
    
    // Error code penalties
    ERROR_RATE_LIMIT: { penaltySeconds: 5, backoffMultiplier: 2 },
    ERROR_429: { penaltySeconds: 30, backoffMultiplier: 3 },
    ERROR_5XX: { penaltySeconds: 10, backoffMultiplier: 2 },
  };
  
  private backoffState: Map = new Map();
  
  constructor() {
    this.config = {
      requestsPerSecond: 20,
      burstCapacity: 20,
      endpointMultipliers: {
        '/api/v5/market': 1.5,
        '/api/v5/account': 0.5,
        '/api/v5/trade': 0.3,
      }
    };
    
    // Initialize buckets
    Object.keys(this.LIMITS).forEach(key => {
      this.buckets.set(key, {
        tokens: this.LIMITS[key as keyof typeof this.LIMITS].burstCapacity,
        lastRefill: Date.now()
      });
    });
  }
  
  private getBucket(endpointType: string): TokenBucket {
    if (!this.buckets.has(endpointType)) {
      this.buckets.set(endpointType, {
        tokens: 10,
        lastRefill: Date.now()
      });
    }
    return this.buckets.get(endpointType)!;
  }
  
  private refillBucket(bucket: TokenBucket, limit: { requestsPerSecond: number; burstCapacity: number }): void {
    const now = Date.now();
    const timePassed = (now - bucket.lastRefill) / 1000;
    const tokensToAdd = timePassed * limit.requestsPerSecond;
    
    bucket.tokens = Math.min(bucket.burstCapacity, bucket.tokens + tokensToAdd);
    bucket.lastRefill = now;
  }
  
  private calculateEndpointType(method: string, path: string): string {
    if (path.includes('/market')) return 'GET_MARKET';
    if (path.includes('/account')) return 'GET_ACCOUNT';
    if (path.includes('/trade') && method === 'POST') return 'POST_ORDER';
    return method;
  }
  
  private checkBackoff(endpointType: string): boolean {
    const state = this.backoffState.get(endpointType);
    if (!state) return true;
    
    if (Date.now() < state.nextRetry) {
      return false;
    }
    
    // Reset backoff after successful request
    this.backoffState.delete(endpointType);
    return true;
  }
  
  private applyBackoff(endpointType: string, errorCode: string): void {
    let penalty = this.LIMITS.ERROR_5XX.penaltySeconds;
    let multiplier = this.LIMITS.ERROR_5XX.backoffMultiplier;
    
    if (errorCode === '429' || errorCode.includes('rate_limit')) {
      penalty = this.LIMITS.ERROR_429.penaltySeconds;
      multiplier = this.LIMITS.ERROR_429.backoffMultiplier;
    } else if (errorCode.startsWith('5')) {
      penalty = this.LIMITS.ERROR_RATE_LIMIT.penaltySeconds;
      multiplier = this.LIMITS.ERROR_RATE_LIMIT.backoffMultiplier;
    }
    
    const currentState = this.backoffState.get(endpointType) || { attempts: 0, nextRetry: 0 };
    const newAttempts = currentState.attempts + 1;
    const backoffTime = penalty * Math.pow(multiplier, newAttempts - 1);
    
    this.backoffState.set(endpointType, {
      attempts: newAttempts,
      nextRetry: Date.now() + (backoffTime * 1000)
    });
    
    console.warn(
      Rate limit applied to ${endpointType}:  +
      Attempt ${newAttempts}, retry in ${backoffTime}s
    );
  }
  
  async acquire(
    method: string,
    path: string,
    priority: number = 5
  ): Promise<{ waitTime: number; endpointType: string }> {
    const endpointType = this.calculateEndpointType(method, path);
    const limit = this.LIMITS[endpointType as keyof typeof this.LIMITS] || this.LIMITS.GET;
    const bucket = this.getBucket(endpointType);
    
    // Check backoff
    if (!this.checkBackoff(endpointType)) {
      const state = this.backoffState.get(endpointType)!;
      const waitTime = (state.nextRetry - Date.now()) / 1000;
      throw new Error(
        Backoff active for ${endpointType}, retry in ${waitTime.toFixed(1)}s
      );
    }
    
    // Refill bucket
    this.refillBucket(bucket, limit);
    
    // Calculate wait time if no tokens available
    if (bucket.tokens < 1) {
      const waitTime = (1 - bucket.tokens) / limit.requestsPerSecond;
      return { waitTime, endpointType };
    }
    
    // Consume token
    bucket.tokens -= 1;
    return { waitTime: 0, endpointType };
  }
  
  async withRateLimit(
    fn: () => Promise,
    method: string,
    path: string
  ): Promise {
    let retries = 0;
    const maxRetries = 5;
    
    while (retries < maxRetries) {
      try {
        const { waitTime, endpointType } = await this.acquire(method, path);
        
        if (waitTime > 0) {
          console.log(Rate limit: waiting ${waitTime.toFixed(3)}s);
          await this.delay(waitTime);
        }
        
        const result = await fn();
        
        // Reset backoff on success
        this.backoffState.delete(endpointType);
        return result;
        
      } catch (error: any) {
        const errorCode = error.response?.data?.code || error.code;
        
        if (errorCode === '429' || errorCode?.includes('rate_limit')) {
          retries++;
          const waitTime = retries * 2; // Exponential backoff
          console.warn(Rate limit hit (${retries}/${maxRetries}), waiting ${waitTime}s);
          await this.delay(waitTime);
        } else if (errorCode?.startsWith('5') || errorCode === '4001') {
          retries++;
          await this.delay(Math.min(1000 * Math.pow(2, retries), 30000));
        } else {
          throw error;
        }
      }
    }
    
    throw new Error(Max retries exceeded for ${method} ${path});
  }
  
  private delay(ms: number): Promise {
    return new Promise(resolve => setTimeout(resolve, ms * 1000));
  }
  
  getStatus(): Record {
    const status: Record = {};
    
    this.buckets.forEach((bucket, key) => {
      const limit = this.LIMITS[key as keyof typeof this.LIMITS];
      status[key] = {
        tokens: Math.round(bucket.tokens * 100) / 100,
        backoff: this.backoffState.has(key)
      };
    });
    
    return status;
  }
}

// ========== INTEGRATION EXAMPLE ==========

class OKXAPIClient {
  private baseURL = 'https://www.okx.com';
  private rateLimiter = new OKXv5RateLimiter();
  private httpClient: AxiosInstance;
  
  constructor(apiKey: string, secretKey: string, passphrase: string) {
    this.httpClient = axios.create({
      baseURL: this.baseURL,
      timeout: 30000,
    });
    
    // Add request interceptor for auth and rate limiting
    this.httpClient.interceptors.request.use(async (config) => {
      const timestamp = new Date().toISOString();
      const method = config.method?.toUpperCase() || 'GET';
      const path = config.url || '';
      
      // Rate limit
      await this.rateLimiter.withRateLimit(
        async () => {},
        method,
        path
      );
      
      // Sign request
      const signature = this.generateSignature(
        timestamp,
        method,
        path,
        config.data || ''
      );
      
      config.headers['OK-ACCESS-KEY'] = apiKey;
      config.headers['OK-ACCESS-SIGN'] = signature;
      config.headers['OK-ACCESS-TIMESTAMP'] = timestamp;
      config.headers['OK-ACCESS-PASSPHRASE'] = passphrase;
      config.headers['Content-Type'] = 'application/json';
      
      return config;
    });
    
    // Add response interceptor for error handling
    this.httpClient.interceptors.response.use(
      response => response,
      async (error) => {
        if (error.response?.status === 429) {
          console.error('429 Rate Limit Exceeded');
          // Rate limiter will handle retry
        }
        return Promise.reject(error);
      }
    );
  }
  
  private generateSignature(
    timestamp: string,
    method: string,
    path: string,
    body: string
  ): string {
    const message = timestamp + method + path + body;
    const hmac = crypto.createHmac('sha256', this.secretKey);
    return hmac.update(message).digest('base64');
  }
  
  // Example API calls
  async getTicker(instId: string) {
    return this.rateLimiter.withRateLimit(
      () => this.httpClient.get(/api/v5/market/ticker?instId=${instId}),
      'GET',
      '/api/v5/market'
    );
  }
  
  async getOrderBook(instId: string, depth: number = 25) {
    return this.rateLimiter.withRateLimit(
      () => this.httpClient.get(
        /api/v5/market/books?instId=${instId}&sz=${depth}
      ),
      'GET',
      '/api/v5/market'
    );
  }
  
  async placeOrder(order: OrderParams) {
    return this.rateLimiter.withRateLimit(
      () => this.httpClient.post('/api/v5/trade/order', order),
      'POST',
      '/api/v5/trade'
    );
  }
}

// Usage
const client = new OKXAPIClient(
  process.env.OKX_API_KEY!,
  process.env.OKX_SECRET_KEY!,
  process.env.OKX_PASSPHRASE!
);

// Safe concurrent requests with rate limiting
async function fetchMultipleTickers(symbols: string[]) {
  const promises = symbols.map(symbol => client.getTicker(symbol));
  return Promise.all(promises);
}

// Monitor rate limiter status
setInterval(() => {
  console.log('Rate limiter status:', client.rateLimiter.getStatus());
}, 60000);

Lỗi thường gặp và cách khắc phục

Lỗi 1: WebSocket Connection Timeout sau 60 giây không có message

Mô tả lỗi: Kết nối WebSocket bị đóng đột ngột sau khi idle quá lâu, log hiển thị ConnectionClosed: code=1006

Nguyên nhân: OKX có cơ chế timeout cho connection không active. Nếu không có message trong 60s, server sẽ disconnect.

# Fix: Implement ping-pong heartbeat mechanism

Thêm vào OKXv5WebSocketClient


class OKXv5WebSocketClient:
    PING_INTERVAL = 20  # OKX yêu cầu ping mỗi 20-30 giây
    PING_TIMEOUT = 5    # Timeout cho pong response
    
    async def _heartbeat_loop(self):
        """Gửi ping định kỳ để giữ connection alive"""
        while self.is_connected:
            try:
                ping_msg = json.dumps({"op": "ping", "args": [int(time.time() * 1000)]})
                await asyncio.wait_for(
                    self.ws.send(ping_msg),
                    timeout=5
                )
                logger.debug("Ping sent")
                
                # Chờ pong với timeout ngắn
                try:
                    response = await asyncio.wait_for(
                        self.ws.recv(),
                        timeout=self.PING_TIMEOUT
                    )
                    data = json.loads(response)
                    if data.get("event") == "pong":
                        logger.debug("Pong received")
                except asyncio.TimeoutError:
                    logger.warning("Pong timeout - connection may be dead")
                    self.is_connected = False
                    break
                    
            except Exception as e:
                logger.error(f"Heartbeat error: {e}")
                self.is_connected = False
                break
            
            await asyncio.sleep(self.PING_INTERVAL)
    
    async def connect(self, private: bool = False) -> bool:
        # ... existing code ...
        
        # Start heartbeat task
        self.heartbeat_task = asyncio.create_task(self._heartbeat_loop())
        
        return True

Lỗi 2: Subscription bị Reject với error code "1"

Mô tả lỗi: Subscribe thành công nhưng không nhận được data, response trả về {"event": "error", "code": "1", "msg": "subscription failed"}

Nguyên nhân: Channel hoặc inst_id không đúng format, hoặc subscription type không phù hợp với endpoint.

# Fix: Validate subscription parameters trước khi gửi

VALID_INST_TYPES = ["SPOT", "MARGIN", "SWAP", "FUTURES", "OPTION"]
VALID_CHANNELS = [
    "ticker", "trade", "books", "books-l2-tbt", 
    "candle1m", "candle5m", "candle15m", "candle1H",
    "mark-price", "funding-rate", "index", "oracle"
]

def validate_subscription(channel: str, inst_id: str, inst_type: str) -> tuple[bool, str]:
    """
    Validate subscription parameters theo OKX v5 spec
    Returns: (is_valid, error_message)
    """
    # Check inst_type
    if inst_type not in VALID_INST_TYPES:
        return False, f"Invalid inst_type '{inst_type}'. Valid: {VALID_INST_TYPES}"
    
    # Check channel format
    if not channel or channel not in VALID_CHANNELS:
        return False, f"Invalid channel '{channel}'. Valid: {VALID_CHANNELS}"
    
    # Validate inst_id format theo inst_type
    if inst_type == "SPOT":
        # Format: XXX-XXX (VD: BTC-USDT)
        if not re.match(r'^[A-Z]+-[A-Z]+$', inst_id):
            return False, f"SPOT inst_id must be XXX-XXX format, got: {inst_id}"
    elif inst_type in ["SWAP", "FUTURES"]:
        # Format: XXX-USD-XXXX (VD: BTC-USD-210625)
        if not re

Tài nguyên liên quan
Bài viết liên quan