การพัฒนาระบบเทรดอัตโนมัติบน Bybit API ต้องเผชิญกับความท้าทายสำคัญด้าน Rate Limiting ซึ่งหากไม่จัดการอย่างเหมาะสม ระบบของคุณจะถูกบล็อกชั่วคราวหรือถาวร บทความนี้จะอธิบายกลยุทธ์ Throttling ที่ใช้งานได้จริง พร้อมตัวอย่างโค้ด Python และวิธีแก้ปัญหาเฉพาะหน้า

Bybit API Rate Limits เข้าใจตารางขีดจำกัด

Bybit กำหนด Rate Limits แตกต่างกันตามประเภท Endpoint โดยมีรายละเอียดดังนี้

ประเภท Endpoint Rate Limit ระยะเวลา หมายเหตุ
Public REST (แชร์) 600 requests 1 นาที รวมทุก Public endpoint
Private REST 600 requests 1 นาที ต่อ API Key
Order (ส่งคำสั่ง) 50 requests 1 วินาที หรือ 200 คำสั่ง/10 วินาที
WebSocket (เริ่มต้น) 5 messages 1 วินาที ถ้าเกินจะถูกตัด connection
WebSocket (อัปเกรด) 240 messages 1 นาที ต้องส่ง ping ทุก 30 วินาที

กรณีการใช้งานจริง: เมื่อใดที่ต้องใช้ Throttling

1. ระบบ AI Customer Relationship Management สำหรับ E-commerce

เมื่อ AI CRM ต้องดึงข้อมูลราคาจาก Bybit เพื่อวิเคราะห์แนวโน้มตลาดสำหรับการตั้งราคาสินค้า การเรียก API พร้อมกันหลายร้อยครั้งจะทำให้เกิน Rate Limit ทันที โซลูชันคือการใช้ Queue-based Throttling ที่จัดลำดับความสำคัญของ request ตามความเร่งด่วน

2. Enterprise RAG System Deployment

องค์กรที่ใช้ RAG (Retrieval-Augmented Generation) เพื่อสร้าง Chatbot ที่ตอบคำถามเกี่ยวกับสินทรัพย์ดิจิทัล ต้องเรียก Bybit API เพื่อดึงข้อมูลราคาปัจจุบัน การใช้ Token Bucket Algorithm จะช่วยรักษา Throughput ที่เสถียรโดยไม่ถูกบล็อก

3. Independent Developer Project

นักพัฒนาอิสระที่สร้าง Dashboard สำหรับติดตามพอร์ตโฟลิโอ มักประสบปัญหาเมื่อต้องดึงข้อมูลหลายสินทรัพย์พร้อมกัน Simple Throttling ด้วย asyncio.Semaphore เป็นทางออกที่เรียบง่ายแต่มีประสิทธิภาพ

Token Bucket Algorithm: กลยุทธ์ Throttling ที่ได้รับความนิยมสูงสุด

Token Bucket Algorithm ทำงานโดยมี "ถัง" ที่บรรจุ Token จำนวนหนึ่ง แต่ละ Request ต้องใช้ Token หนึ่งใบ และ Token จะถูกเติมใหม่ตามอัตราที่กำหนด วิธีนี้เหมาะกับการรับมือกับ Traffic ที่ไม่สม่ำเสมอ

import time
import asyncio
from threading import Lock

class TokenBucketThrottler:
    """
    Token Bucket Algorithm สำหรับ Bybit API Rate Limiting
    - capacity: จำนวน Token สูงสุดในถัง
    - refill_rate: จำนวน Token ที่เติมต่อวินาที
    """
    
    def __init__(self, capacity: int, refill_rate: float):
        self.capacity = capacity
        self.refill_rate = refill_rate
        self.tokens = capacity
        self.last_refill = time.time()
        self.lock = Lock()
    
    def _refill(self):
        """เติม Token ตามเวลาที่ผ่านไป"""
        now = time.time()
        elapsed = now - self.last_refill
        new_tokens = elapsed * self.refill_rate
        self.tokens = min(self.capacity, self.tokens + new_tokens)
        self.last_refill = now
    
    async def acquire(self, tokens: int = 1):
        """
        รอจนกว่าจะมี Token เพียงพอ
        tokens: จำนวน Token ที่ต้องการใช้
        """
        while True:
            with self.lock:
                self._refill()
                if self.tokens >= tokens:
                    self.tokens -= tokens
                    return
            
            # รอก่อนลองใหม่ (sleep 50ms เพื่อลด CPU usage)
            await asyncio.sleep(0.05)


ตัวอย่างการใช้งานสำหรับ Bybit Private API (600 req/min)

bybit_throttler = TokenBucketThrottler(
    capacity=600,
    refill_rate=10  # 600/60 = 10 tokens/วินาที
)


สำหรับ Order Endpoint (50 req/sec)

order_throttler = TokenBucketThrottler(
    capacity=50,
    refill_rate=50
)

async def call_bybit_api(endpoint: str, params: dict):
    """เรียก Bybit API พร้อม Throttling"""
    await bybit_throttler.acquire()
    # เรียก API จริงที่นี่
    print(f"[{time.strftime('%H:%M:%S')}] Calling {endpoint}")
    return {"success": True, "data": {}}

Async Queue-based Throttling: เหมาะกับ High-frequency Trading

สำหรับระบบที่ต้องจัดการ Request จำนวนมากพร้อมกัน การใช้ Asyncio Queue ร่วมกับ Semaphore จะช่วยควบคุม Throughput ได้อย่างแม่นยำ

import asyncio
import aiohttp
from collections import deque
from dataclasses import dataclass, field
from typing import Callable, Any, Optional
import time

@dataclass
class SlidingWindowThrottler:
    """
    Sliding Window Rate Limiter สำหรับ Bybit API
    ใช้ sliding window แบบละเอียดกว่า token bucket
    """
    max_requests: int
    window_seconds: float
    _timestamps: deque = field(default_factory=deque)
    _lock: asyncio.Lock = field(default_factory=asyncio.Lock)
    
    async def acquire(self):
        """รอจนกว่าจะสามารถส่ง request ได้"""
        async with self._lock:
            now = time.time()
            cutoff = now - self.window_seconds
            
            # ลบ timestamps ที่เก่ากว่า window
            while self._timestamps and self._timestamps[0] < cutoff:
                self._timestamps.popleft()
            
            if len(self._timestamps) >= self.max_requests:
                # คำนวณเวลารอที่เหลือ
                sleep_time = self._timestamps[0] + self.window_seconds - now + 0.01
                if sleep_time > 0:
                    await asyncio.sleep(sleep_time)
                    return await self.acquire()  # ลองใหม่
            
            self._timestamps.append(time.time())
    
    @property
    def current_usage(self) -> int:
        """ดูจำนวน request ที่ใช้ไปใน window ปัจจุบัน"""
        now = time.time()
        cutoff = now - self.window_seconds
        return sum(1 for ts in self._timestamps if ts >= cutoff)

class BybitAPIClient:
    """Client สำหรับเรียก Bybit API พร้อม Throttling"""
    
    def __init__(self, api_key: str, api_secret: str):
        self.api_key = api_key
        self.api_secret = api_secret
        self.base_url = "https://api.bybit.com"
        
        # Throttlers สำหรับ endpoint ต่างๆ
        self.public_throttler = SlidingWindowThrottler(600, 60)  # 600/min
        self.order_throttler = SlidingWindowThrottler(50, 1)      # 50/sec
        self.private_throttler = SlidingWindowThrottler(600, 60)  # 600/min
    
    async def get_symbol_price(self, symbol: str) -> dict:
        """ดึงราคาสินทรัพย์ (Public endpoint)"""
        await self.public_throttler.acquire()
        
        url = f"{self.base_url}/v5/market/tickers"
        params = {"category": "spot", "symbol": symbol}
        
        async with aiohttp.ClientSession() as session:
            async with session.get(url, params=params) as resp:
                return await resp.json()
    
    async def place_order(self, symbol: str, side: str, qty: float) -> dict:
        """ส่งคำสั่งซื้อขาย (Order endpoint - มี limit เข้มงวดกว่า)"""
        await self.order_throttler.acquire()
        
        url = f"{self.base_url}/v5/order/create"
        payload = {
            "category": "spot",
            "symbol": symbol,
            "side": side,
            "orderType": "Market",
            "qty": str(qty)
        }
        
        headers = self._generate_signature(payload)
        async with aiohttp.ClientSession() as session:
            async with session.post(url, json=payload, headers=headers) as resp:
                return await resp.json()
    
    def _generate_signature(self, payload: dict) -> dict:
        """สร้าง HMAC signature สำหรับ Private endpoint"""
        # ต้องใช้ Bybit signature generation จริง
        import hmac
        import hashlib
        
        timestamp = str(int(time.time() * 1000))
        param_str = json.dumps(payload)
        signature_str = timestamp + self.api_key + param_str
        signature = hmac.new(
            self.api_secret.encode(),
            signature_str.encode(),
            hashlib.sha256
        ).hexdigest()
        
        return {
            "X-BAPI-API-KEY": self.api_key,
            "X-BAPI-SIGN": signature,
            "X-BAPI-TIMESTAMP": timestamp,
            "X-BAPI-SIGN-TYPE": "2"
        }


การใช้งาน

async def main():
    client = BybitAPIClient(
        api_key="YOUR_BYBIT_API_KEY",
        api_secret="YOUR_BYBIT_API_SECRET"
    )
    
    # ดึงราคาหลายสินทรัพย์พร้อมกัน
    symbols = ["BTCUSDT", "ETHUSDT", "SOLUSDT"]
    
    tasks = [client.get_symbol_price(s) for s in symbols]
    results = await asyncio.gather(*tasks)
    
    for symbol, result in zip(symbols, results):
        print(f"{symbol}: {result}")

asyncio.run(main())

Retry Strategy กับ Exponential Backoff

เมื่อเจอ Rate Limit Error (HTTP 429) ต้องมี Retry Strategy ที่ฉลาด ไม่ใช่แค่รอคร่าวๆ แต่ต้องใช้ Exponential Backoff ที่เหมาะสมกับ Bybit

import asyncio
import aiohttp
from typing import Optional
import random

class BybitRetryClient:
    """Client ที่จัดการ Retry อย่างฉลาดเมื่อเจอ Rate Limit"""
    
    def __init__(
        self,
        max_retries: int = 5,
        base_delay: float = 1.0,
        max_delay: float = 60.0,
        jitter: bool = True
    ):
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.jitter = jitter
    
    async def _calculate_delay(self, attempt: int, retry_after: Optional[int] = None) -> float:
        """คำนวณ delay ด้วย Exponential Backoff"""
        
        # ถ้า server แนะนำ retry_after ให้ใช้ค่านั้นก่อน
        if retry_after:
            return retry_after
        
        # Exponential backoff: 2^attempt * base_delay
        delay = self.base_delay * (2 ** attempt)
        delay = min(delay, self.max_delay)
        
        # เพิ่ม jitter เพื่อป้องกัน thundering herd
        if self.jitter:
            delay *= (0.5 + random.random() * 0.5)
        
        return delay
    
    async def request_with_retry(
        self,
        method: str,
        url: str,
        **kwargs
    ) -> dict:
        """เรียก API พร้อม Retry Logic"""
        
        for attempt in range(self.max_retries):
            try:
                async with aiohttp.ClientSession() as session:
                    async with session.request(method, url, **kwargs) as resp:
                        
                        if resp.status == 200:
                            return await resp.json()
                        
                        # ตรวจสอบ Rate Limit Error
                        if resp.status == 429:
                            # ลองอ่าน Retry-After header
                            retry_after = resp.headers.get('Retry-After')
                            retry_after_val = int(retry_after) if retry_after else None
                            
                            # อ่าน response body เพื่อดูรายละเอียด
                            error_body = await resp.json()
                            print(f"[Rate Limit] Attempt {attempt + 1} failed: {error_body}")
                            
                            delay = await self._calculate_delay(attempt, retry_after_val)
                            print(f"[Retry] Waiting {delay:.2f}s before retry...")
                            await asyncio.sleep(delay)
                            continue
                        
                        # HTTP Error อื่นๆ
                        error_text = await resp.text()
                        print(f"[HTTP Error {resp.status}] {error_text}")
                        raise aiohttp.ClientError(f"HTTP {resp.status}: {error_text}")
                        
            except aiohttp.ClientError as e:
                print(f"[Connection Error] Attempt {attempt + 1}: {e}")
                if attempt == self.max_retries - 1:
                    raise
                
                delay = await self._calculate_delay(attempt)
                await asyncio.sleep(delay)
        
        raise Exception(f"Max retries ({self.max_retries}) exceeded")

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

กรณีที่ 1: HTTP 10009 - Order rate limit exceeded

สาเหตุ: เรียก Order endpoint เกิน 50 คำสั่ง/วินาที หรือ 200 คำสั่ง/10 วินาที

วิธีแก้ไข: 
# ปัญหา: ส่งคำสั่งพร้อมกันหลายคำ
async def bad_example():
    tasks = [place_order(symbol, qty) for symbol in symbols]  # ผิด!
    await asyncio.gather(*tasks)


แก้ไข: ใช้ Rate Limiter แยกสำหรับ Order endpoint

from aiohttp import ClientSession
import asyncio

class OrderRateLimiter:
    def __init__(self):
        self.orders_per_second = 0
        self.orders_per_10sec = 0
        self.last_10sec_reset = time.time()
        self.lock = asyncio.Lock()
    
    async def acquire(self):
        async with self.lock:
            now = time.time()
            
            # Reset counter ทุก 10 วินาที
            if now - self.last_10sec_reset >= 10:
                self.orders_per_10sec = 0
                self.last_10sec_reset = now
            
            # ตรวจสอบทั้งสองเงื่อนไข
            if self.orders_per_second >= 50 or self.orders_per_10sec >= 200:
                wait_time = max(
                    1 - (now - self.last_10sec_reset) if self.orders_per_second >= 50 else 0,
                    10 - (now - self.last_10sec_reset) if self.orders_per_10sec >= 200 else 0
                ) + 0.1
                await asyncio.sleep(wait_time)
                return await self.acquire()
            
            self.orders_per_second += 1
            self.orders_per_10sec += 1
    
    async def reset_per_second(self):
        """เรียกทุกวินาทีเพื่อ reset counter"""
        while True:
            await asyncio.sleep(1)
            async with self.lock:
                self.orders_per_second = 0


การใช้งาน

order_limiter = OrderRateLimiter()


Start background task สำหรับ reset per second

asyncio.create_task(order_limiter.reset_per_second())

async def safe_place_order(client, symbol: str, qty: float):
    await order_limiter.acquire()
    return await client.place_order(symbol, "Buy", qty)

กรณีที่ 2: WebSocket Disconnection บ่อย

สาเหตุ: ส่ง Subscribe message เกิน 5 ข้อความ/วินาที หรือไม่ส่ง Ping เพียงพอ

วิธีแก้ไข: 
import websockets
import asyncio
import json

class BybitWebSocketClient:
    """WebSocket Client ที่จัดการ Rate Limit อย่างถูกต้อง"""
    
    def __init__(self):
        self.ws = None
        self.ping_interval = 20  # Bybit ต้องการ ping ทุก 20-30 วินาที
        self.message_timestamps = []
        self.msg_limit = 5  # 5 messages/sec
        self.msg_lock = asyncio.Lock()
    
    async def _check_message_rate(self):
        """ตรวจสอบว่าส่ง message เร็วเกินไปหรือไม่"""
        async with self.msg_lock:
            now = asyncio.get_event_loop().time()
            
            # ลบ timestamps เก่ากว่า 1 วินาที
            self.message_timestamps = [
                ts for ts in self.message_timestamps 
                if now - ts < 1
            ]
            
            if len(self.message_timestamps) >= self.msg_limit:
                sleep_time = 1 - (now - self.message_timestamps[0]) + 0.05
                await asyncio.sleep(sleep_time)
            
            self.message_timestamps.append(now)
    
    async def subscribe(self, channels: list):
        """Subscribe ไปยังหลาย channels พร้อมกัน"""
        
        # รวม channels เป็น single subscribe message
        for channel in channels:
            await self._check_message_rate()
            
            subscribe_msg = {
                "op": "subscribe",
                "args": [channel]
            }
            
            await self.ws.send(json.dumps(subscribe_msg))
            await asyncio.sleep(0.2)  # รอเล็กน้อยระหว่างแต่ละ subscribe
        
        print(f"Subscribed to {len(channels)} channels")
    
    async def connect(self, url: str):
        """เชื่อมต่อ WebSocketพร้อม Auto-ping"""
        
        self.ws = await websockets.connect(url)
        
        async def ping_loop():
            """ส่ง ping ทุก 25 วินาที"""
            while True:
                await asyncio.sleep(25)
                if self.ws and self.ws.open:
                    await self._check_message_rate()
                    await self.ws.send(json.dumps({"op": "ping"}))
        
        asyncio.create_task(ping_loop())
        return self
    
    async def receive(self):
        """รับ messages"""
        async for msg in self.ws:
            data = json.loads(msg)
            yield data


การใช้งาน

async def main():
    async with BybitWebSocketClient() as client:
        await client.connect("wss://stream.bybit.com/v5/public/spot")
        
        # Subscribe หลาย channels พร้อมกัน
        await client.subscribe([
            "orderbook.50.BTCUSDT",
            "orderbook.50.ETHUSDT", 
            "publicTrade.BTCUSDT"
        ])
        
        async for data in client.receive():
            print(data)

asyncio.run(main())

กรณีที่ 3: 10002 - Signature verification failed หลัง Retry

สาเหตุ: ใช้ timestamp เดิมในการสร้าง signature หลังจาก retry

วิธีแก้ไข: 
import time
import hmac
import hashlib
from typing import Optional

class BybitAuthenticator:
    """Authenticator ที่จัดการ signature อย่างถูกต้องสำหรับ Retry"""
    
    def __init__(self, api_key: str, api_secret: str):
        self.api_key = api_key
        self.api_secret = api_secret
    
    def generate_auth_headers(
        self, 
        params: dict, 
        timestamp: Optional[str] = None,
        recv_window: str = "5000"
    ) -> dict:
        """
        สร้าง headers สำหรับ Bybit API
        - timestamp ต้องสร้างใหม่ทุกครั้ง (ไม่ใช้ค่าคงที่)
        - recv_window ควรลดลงเมื่อ retry
        """
        
        # สร้าง timestamp ใหม่เสมอ
        current_timestamp = timestamp or str(int(time.time() * 1000))
        
        # ลด recv_window สำหรับ retry เพื่อลดโอกาส signature ล้าสมัย
        param_str = json.dumps(params, separators=(',', ':'))
        
        # Signature string
        sign_str = current_timestamp + self.api_key + recv_window + param_str
        
        signature = hmac.new(
            self.api_secret.encode('utf-8'),
            sign_str.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()
        
        return {
            'X-BAPI-API-KEY': self.api_key,
            'X-BAPI-SIGN': signature,
            'X-BAPI-TIMESTAMP': current_timestamp,
            'X-BAPI-RECV-WINDOW': recv_window,
            'X-BAPI-SIGN-TYPE': '2',
            'Content-Type': 'application/json'
        }


การใช้งานใน Retry Logic

class RetryableBybitClient:
    def __init__(self, api_key: str, api_secret: str):
        self.auth = BybitAuthenticator(api_key, api_secret)
    
    async def place_order_with_retry(
        self, 
        symbol: str, 
        side: str, 
        qty: float,
        max_retries: int = 3
    ):
        for attempt in range(max_retries):
            # ลด recv_window ทุกครั้งที่ retry (ชดเชยเวลาที่ผ่านไป)
            recv_window = str(5000 - (attempt * 500))  # 5000, 4500, 4000
            
            params = {
                "category": "spot",
                "symbol": symbol,
                "side": side,
                "orderType": "Market",
                "qty": str(qty)
            }
            
            headers = self.auth.generate_auth_headers(params, recv_window=recv_window)
            
            async with aiohttp.ClientSession() as session:
                async with session.post(
                    "https://api.bybit.com/v5/order/create",
                    json=params,
                    headers=headers
                ) as resp:
                    data = await resp.json()
                    
                    if data.get('retCode') == 0:
                        return data
                    
                    # ถ้าเป็น signature error ให้ retry
                    if data.get('retCode') == 10002:
                        print(f"Signature error on attempt {attempt + 1}, retrying...")
                        continue
                    
                    raise Exception(f"API Error: {data}")

เปรียบเทียบ Throttling Strategies

Strategy ข้อดี ข้อเสีย เหมาะกับ
Token Bucket รองรับ Burst traffic ได้ดี, ใช้ง่าย ต้องจัดการ Lock สำหรับ thread safety การดึงข้อมูล Public API
Sliding Window แม่นยำกว่า, กระจายโหลดดี ใช้ memory มากกว่า High-frequency trading
Queue-based ควบคุม Throughput ได้แม่นยำ เพิ่ม Latency Batch processing
Exponential Backoff ฉลาดในการ Retry, ลด thundering herd อาจ retry ช้าเกินไป Error recovery

Best Practices สำหรับ Production