ในโลกของการเทรดอัตโนมัติด้วย Bot การเชื่อมต่อกับ API ของตลาดซื้อขายเป็นหัวใจสำคัญ แต่ปัญหาที่พบบ่อยที่สุดคือการหลุด Connection (Connection Drop) และการซิงโครไนซ์ข้อมูลที่ไม่ตรงกัน บทความนี้จะอธิบายวิธีการแก้ปัญหาอย่างละเอียด พร้อมโค้ดตัวอย่างที่ใช้งานได้จริง โดยใช้ Python เป็นภาษาหลักในการเขียน
ปัญหาหลักที่พบบ่อย
- Connection Drop กะทันหัน — เกิดจากเครือข่ายไม่เสถียร, Server ของ Exchange มีปัญหา หรือ Firewall บล็อก
- ข้อมูลไม่ตรงกัน (Data Mismatch) — Order book, Trade history หรือ Balance ไม่สอดคล้องกับข้อมูลจริง
- Rate Limit Exceeded — เรียก API บ่อยเกินไปจนโดนจำกัด
- Authentication Failure — Token หมดอายุ หรือ Signature ไม่ถูกต้อง
โครงสร้างพื้นฐานของ WebSocket Reconnection
ก่อนจะเข้าสู่การแก้ปัญหา มาดูโครงสร้างพื้นฐานของ WebSocket Client ที่รองรับการ Reconnect อัตโนมัติกันก่อน
import asyncio
import websockets
import json
import time
from typing import Optional, Callable, Dict, Any
from dataclasses import dataclass, field
from enum import Enum
import logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("ExchangeAPI")
class ConnectionState(Enum):
DISCONNECTED = "disconnected"
CONNECTING = "connecting"
CONNECTED = "connected"
RECONNECTING = "reconnecting"
ERROR = "error"
@dataclass
class ReconnectConfig:
max_retries: int = 10
base_delay: float = 1.0
max_delay: float = 60.0
exponential_base: float = 2.0
jitter: bool = True
@dataclass
class ExchangeWebSocketClient:
uri: str
api_key: str
api_secret: str
reconnect_config: ReconnectConfig = field(default_factory=ReconnectConfig)
state: ConnectionState = ConnectionState.DISCONNECTED
ws: Optional[websockets.WebSocketClientProtocol] = None
retry_count: int = 0
last_message_time: float = field(default_factory=time.time)
subscriptions: list = field(default_factory=list)
message_handler: Optional[Callable] = None
def calculate_delay(self) -> float:
"""คำนวณ delay สำหรับ reconnect แบบ exponential backoff"""
delay = self.reconnect_config.base_delay * (
self.reconnect_config.exponential_base ** self.retry_count
)
delay = min(delay, self.reconnect_config.max_delay)
if self.reconnect_config.jitter:
import random
delay = delay * (0.5 + random.random() * 0.5)
return delay
async def connect(self) -> bool:
"""เชื่อมต่อไปยัง WebSocket Server"""
try:
self.state = ConnectionState.CONNECTING
self.ws = await websockets.connect(
self.uri,
ping_interval=20,
ping_timeout=10,
close_timeout=5
)
await self.authenticate()
self.state = ConnectionState.CONNECTED
self.retry_count = 0
self.last_message_time = time.time()
logger.info(f"เชื่อมต่อสำเร็จ: {self.uri}")
return True
except Exception as e:
logger.error(f"เชื่อมต่อล้มเหลว: {e}")
self.state = ConnectionState.ERROR
return False
async def authenticate(self):
"""ยืนยันตัวตนกับ Exchange"""
timestamp = int(time.time() * 1000)
# สร้าง signature ตามมาตรฐานของแต่ละ Exchange
message = f"{timestamp}{self.api_key}"
import hmac
import hashlib
signature = hmac.new(
self.api_secret.encode(),
message.encode(),
hashlib.sha256
).hexdigest()
auth_payload = {
"type": "auth",
"apiKey": self.api_key,
"timestamp": timestamp,
"signature": signature
}
await self.ws.send(json.dumps(auth_payload))
response = await asyncio.wait_for(self.ws.recv(), timeout=10)
result = json.loads(response)
if result.get("status") != "success":
raise Exception(f"ยืนยันตัวตนล้มเหลว: {result}")
logger.info("ยืนยันตัวตนสำเร็จ")
async def reconnect(self) -> bool:
""" reconnect แบบ exponential backoff"""
if self.retry_count >= self.reconnect_config.max_retries:
logger.error("เกินจำนวนครั้งสูงสุดที่อนุญาตให้ reconnect")
return False
delay = self.calculate_delay()
self.retry_count += 1
self.state = ConnectionState.RECONNECTING
logger.info(f"กำลัง reconnect ครั้งที่ {self.retry_count} ในอีก {delay:.2f} วินาที")
await asyncio.sleep(delay)
# ปิด connection เดิมถ้ายังคงเปิดอยู่
if self.ws:
try:
await self.ws.close()
except:
pass
return await self.connect()
async def listen(self):
"""ฟังข้อความจาก WebSocket และจัดการ reconnect อัตโนมัติ"""
while True:
try:
if self.state != ConnectionState.CONNECTED:
success = await self.reconnect()
if not success:
logger.error("reconnect ล้มเหลว หยุดการทำงาน")
break
# resubscribe หลังจาก reconnect สำเร็จ
await self.resubscribe()
async for message in self.ws:
self.last_message_time = time.time()
data = json.loads(message)
if data.get("type") == "pong":
continue
if self.message_handler:
await self.message_handler(data)
except websockets.exceptions.ConnectionClosed as e:
logger.warning(f"Connection หลุด: {e.code} - {e.reason}")
self.state = ConnectionState.DISCONNECTED
except asyncio.TimeoutError:
logger.warning("หมดเวลารอข้อความ")
# ตรวจสอบ heartbeat
if time.time() - self.last_message_time > 60:
logger.warning("ไม่ได้รับข้อความนานเกินไป พยายาม reconnect")
self.state = ConnectionState.DISCONNECTED
except Exception as e:
logger.error(f"เกิดข้อผิดพลาด: {e}")
self.state = ConnectionState.ERROR
async def resubscribe(self):
"""สมัครรับข้อมูลใหม่หลัง reconnect"""
logger.info(f"สมัครรับข้อมูลใหม่: {len(self.subscriptions)} channels")
for sub in self.subscriptions:
await self.send_subscription(sub)
async def send_subscription(self, subscription: Dict[str, Any]):
"""ส่งคำขอสมัครรับข้อมูล"""
if self.ws and self.state == ConnectionState.CONNECTED:
await self.ws.send(json.dumps(subscription))
logger.info(f"ส่ง subscription: {subscription.get('channel')}")
async def subscribe(self, channel: str, symbol: str = None):
"""สมัครรับข้อมูลจาก channel"""
subscription = {
"type": "subscribe",
"channel": channel
}
if symbol:
subscription["symbol"] = symbol
self.subscriptions.append(subscription)
if self.state == ConnectionState.CONNECTED:
await self.send_subscription(subscription)
async def close(self):
"""ปิด connection อย่างปลอดภัย"""
self.state = ConnectionState.DISCONNECTED
if self.ws:
await self.ws.close()
logger.info("ปิด connection แล้ว")การซิงโครไนซ์ข้อมูล Order Book
การซิงโครไนซ์ Order Book เป็นสิ่งสำคัญมาก เพราะหลังจาก reconnect ข้อมูลที่เรามีอาจไม่ตรงกับข้อมูลจริงบน Server ต้องมีกลไกในการดึงข้อมูลใหม่ทั้งหมดและอัปเดตอย่างถูกต้อง
import asyncio
from collections import OrderedDict
from dataclasses import dataclass
from typing import Dict, List, Tuple, Optional
import time
import threading
import logging
logger = logging.getLogger("OrderBookSync")
@dataclass
class OrderBookEntry:
price: float
quantity: float
timestamp: float
def is_empty(self) -> bool:
return self.quantity <= 0
class OrderBook:
"""โครงสร้าง Order Book ที่รองรับการอัปเดตแบบ Delta Update"""
def __init__(self, symbol: str, max_levels: int = 20):
self.symbol = symbol
self.max_levels = max_levels
# bids: ราคาซื้อ (เรียงจากมากไปน้อย)
self.bids: OrderedDict[float, OrderBookEntry] = OrderedDict()
# asks: ราคาขาย (เรียงจากน้อยไปมาก)
self.asks: OrderedDict[float, OrderBookEntry] = OrderedDict()
self.last_update_id: int = 0
self.last_sync_time: float = 0
self.version: int = 0
self._lock = threading.Lock()
def _update_side(self, side: Dict, entries: Dict[float, OrderBookEntry]):
"""อัปเดตข้อมูลด้านใดด้านหนึ่ง"""
for price_str, qty in side.items():
price = float(price_str)
qty = float(qty)
if qty == 0:
entries.pop(price, None)
else:
entries[price] = OrderBookEntry(
price=price,
quantity=qty,
timestamp=time.time()
)
def apply_snapshot(self, bids: List, asks: List, update_id: int):
"""นำข้อมูล Snapshot มาใช้"""
with self._lock:
self.bids.clear()
self.asks.clear()
# เรียง bids จากมากไปน้อย
sorted_bids = sorted(bids, key=lambda x: float(x[0]), reverse=True)
for price, qty, *_ in sorted_bids[:self.max_levels]:
self.bids[float(price)] = OrderBookEntry(
price=float(price),
quantity=float(qty),
timestamp=time.time()
)
# เรียง asks จากน้อยไปมาก
sorted_asks = sorted(asks, key=lambda x: float(x[0]))
for price, qty, *_ in sorted_asks[:self.max_levels]:
self.asks[float(price)] = OrderBookEntry(
price=float(price),
quantity=float(qty),
timestamp=time.time()
)
self.last_update_id = update_id
self.last_sync_time = time.time()
self.version += 1
logger.info(f"Order Book synced: {self.symbol}, ID={update_id}, "
f"Bids={len(self.bids)}, Asks={len(self.asks)}")
def apply_update(self, update: Dict, expected_update_id: int) -> bool:
"""นำข้อมูล Delta Update มาใช้ ตรวจสอบลำดับก่อนอัปเดต"""
with self._lock:
update_id = update.get("u") or update.get("updateId") or update.get("lastUpdateId")
# ตรวจสอบว่า update_id ถูกต้องตามลำดับ
if update_id <= self.last_update_id:
logger.warning(f"Update ID {update_id} เก่ากว่า last_update_id {self.last_update_id}")
return False
# บาง Exchange ใช้ first_update_id และ final_update_id
first_id = update.get("f") or update.get("firstUpdateId")
final_id = update.get("F") or update.get("finalUpdateId")
if first_id and final_id:
if first_id > self.last_update_id + 1:
logger.warning(f"Gap detected: {self.last_update_id} -> {first_id}")
return False
if final_id < self.last_update_id:
logger.warning(f"Final ID {final_id} เก่ากว่า last_update_id")
return False
bids = update.get("b") or update.get("bids") or []
asks = update.get("a") or update.get("asks") or []
self._update_side(bids, self.bids)
self._update_side(asks, self.asks)
self.last_update_id = update_id
self.last_sync_time = time.time()
self.version += 1
return True
def get_best_bid(self) -> Optional[Tuple[float, float]]:
"""ราคาซื้อสูงสุด"""
with self._lock:
if not self.bids:
return None
first = next(iter(self.bids.items()))
return (first[0], first[1].quantity)
def get_best_ask(self) -> Optional[Tuple[float, float]]:
"""ราคาขายต่ำสุด"""
with self._lock:
if not self.asks:
return None
first = next(iter(self.asks.items()))
return (first[0], first[1].quantity)
def get_spread(self) -> Optional[float]:
"""คำนวณ Spread"""
best_bid = self.get_best_bid()
best_ask = self.get_best_ask()
if best_bid and best_ask:
return best_ask[0] - best_bid[0]
return None
def get_mid_price(self) -> Optional[float]:
"""ราคากลาง"""
best_bid = self.get_best_bid()
best_ask = self.get_best_ask()
if best_bid and best_ask:
return (best_bid[0] + best_ask[0]) / 2
return None
def get_depth(self, levels: int = None) -> Dict:
"""ดึงข้อมูลความลึกของ Order Book"""
levels = levels or self.max_levels
with self._lock:
return {
"symbol": self.symbol,
"timestamp": time.time(),
"version": self.version,
"last_update_id": self.last_update_id,
"bids": [
{"price": price, "qty": entry.quantity}
for price, entry in list(self.bids.items())[:levels]
],
"asks": [
{"price": price, "qty": entry.quantity}
for price, entry in list(self.asks.items())[:levels]
],
"spread": self.get_spread(),
"mid_price": self.get_mid_price()
}
class OrderBookManager:
"""จัดการ Order Book หลายตัวพร้อมกัน"""
def __init__(self):
self.order_books: Dict[str, OrderBook] = {}
self._lock = threading.Lock()
def get_or_create(self, symbol: str, max_levels: int = 20) -> OrderBook:
with self._lock:
if symbol not in self.order_books:
self.order_books[symbol] = OrderBook(symbol, max_levels)
return self.order_books[symbol]
async def full_sync(self, symbol: str, rest_client) -> bool:
"""ดึงข้อมูล Order Book ทั้งหมดจาก REST API"""
try:
ob = self.get_or_create(symbol)
# ดึง snapshot จาก REST API
snapshot = await rest_client.get_order_book(symbol, limit=20)
# หา update_id ล่าสุดจาก WebSocket buffer
pending_updates = await self._get_pending_updates(symbol)
if pending_updates:
# กรองเฉพาะ updates ที่ใหม่กว่า snapshot
valid_updates = [
u for u in pending_updates
if u.get("u", 0) > snapshot.get("lastUpdateId", 0)
]
if valid_updates:
# เรียงลำดับ updates
valid_updates.sort(key=lambda x: x.get("u", 0))
# หา update_id สุดท้ายที่ต้องรอ
last_update_id = valid_updates[-1].get("u", 0)
# รอจนกว่าจะได้รับ update ที่ update_id >= last_update_id + 1
await self._wait_for_update(symbol, last_update_id + 1)
# ประมวลผล updates ทั้งหมด
for update in valid_updates:
ob.apply_update(update, ob.last_update_id + 1)
# ประยุกต์ snapshot
ob.apply_snapshot(
snapshot.get("bids", []),
snapshot.get("asks", []),
snapshot.get("lastUpdateId", 0)
)
logger.info(f"Full sync สำเร็จสำหรับ {symbol}")
return True
except Exception as e:
logger.error(f"Full sync ล้มเหลวสำหรับ {symbol}: {e}")
return False
async def _get_pending_updates(self, symbol: str) -> List[Dict]:
# ดึง updates ที่รออยู่ใน buffer
return []
async def _wait_for_update(self, symbol: str, min_update_id: int, timeout: float = 10):
# รอจนกว่าจะได้รับ update ที่ต้องการ
start = time.time()
while time.time() - start < timeout:
ob = self.get_or_create(symbol)
if ob.last_update_id >= min_update_id:
return
await asyncio.sleep(0.01)
raise TimeoutError(f"รอ update {min_update_id} หมดเวลา")การจัดการ REST API พร้อม Retry และ Rate Limit
import aiohttp
import asyncio
from typing import Dict, Any, Optional
from enum import Enum
import time
import math
class RateLimitType(Enum):
REQUEST_WEIGHT = "request_weight"
ORDERS = "orders"
TRADES = "trades"
class RateLimiter:
"""จัดการ Rate Limit อย่างชาญฉลาด"""
def __init__(self):
self.limits: Dict[RateLimitType, Dict] = {
RateLimitType.REQUEST_WEIGHT: {"capacity": 1200, "refill_rate": 10, "current": 1200, "last_refill": time.time()},
RateLimitType.ORDERS: {"capacity": 50, "refill_rate": 1, "current": 50, "last_refill": time.time()},
RateLimitType.TRADES: {"capacity": 200, "refill_rate": 5, "current": 200, "last_refill": time.time()},
}
self.weights: Dict[str, int] = {
"GET /api/v3/order": 1,
"GET /api/v3/openOrders": 3,
"POST /api/v3/order": 1,
"DELETE /api/v3/order": 1,
}
self._lock = asyncio.Lock()
def _refill(self, limit_type: RateLimitType):
"""เติม token ตามเวลาที่ผ่านไป"""
now = time.time()
limit = self.limits[limit_type]
elapsed = now - limit["last_refill"]
tokens_to_add = elapsed * limit["refill_rate"]
limit["current"] = min(limit["capacity"], limit["current"] + tokens_to_add)
limit["last_refill"] = now
async def acquire(self, endpoint: str, weight: int = 1, limit_type: RateLimitType = RateLimitType.REQUEST_WEIGHT):
"""รอจนกว่าจะมี quota เพียงพอ"""
async with self._lock:
self._refill(limit_type)
limit = self.limits[limit_type]
# คำนวณ weight จาก endpoint ถ้าไม่ได้ระบุ
actual_weight = self.weights.get(endpoint, weight)
while limit["current"] < actual_weight:
wait_time = (actual_weight - limit["current"]) / limit["refill_rate"]
await asyncio.sleep(wait_time)
self._refill(limit_type)
limit["current"] -= actual_weight
def get_remaining(self, limit_type: RateLimitType) -> float:
self._refill(limit_type)
return self.limits[limit_type]["current"]
def set_limit(self, limit_type: RateLimitType, capacity: float, refill_rate: float):
"""ปรับ limit ตามข้อมูลจาก response header"""
self.limits[limit_type]["capacity"] = capacity
self.limits[limit_type]["refill_rate"] = refill_rate
class ExchangeRESTClient:
"""REST Client พร้อมระบบ Retry และ Rate Limit"""
def __init__(self, base_url: str, api_key: str, api_secret: str):
self.base_url = base_url.rstrip("/")
self.api_key = api_key
self.api_secret = api_secret
self.session: Optional[aiohttp.ClientSession] = None
self.rate_limiter = RateLimiter()
self.max_retries = 3
self.retry_delay = 1.0
async def __aenter__(self):
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
def _create_signature(self, params: Dict) -> str:
"""สร้าง signature สำหรับ request"""
import hmac
import hashlib
query_string = "&".join([f"{k}={v}" for k, v in sorted(params.items())])
signature = hmac.new(
self.api_secret.encode(),
query_string.encode(),
hashlib.sha256
).hexdigest()
return signature
async def _request(self, method: str, endpoint: str, params: Dict = None, signed: bool = False) -> Dict:
"""ส่ง request พร้อม retry logic"""
url = f"{self.base_url}{endpoint}"
headers = {"X-MBX-APIKEY": self.api_key}
for attempt in range(self.max_retries):
try:
# รอ rate limit
weight = self.rate_limiter.weights.get(f"{method} {endpoint}", 1)
await self.rate_limiter.acquire(f"{method} {endpoint}", weight)
# เพิ่ม timestamp และ signature สำหรับ signed request
request_params = params