การมาตรฐานข้อมูลจากตลาดคริปโต: WebSocket กับ REST API เลือกอะไรดี

ผมเคยเจอสถานการณ์ที่ทำให้ปวดหัวมาก ตอนพัฒนาระบบ Trading Bot ที่ดึงข้อมูลราคาจากหลายตลาดพร้อมกัน ใช้ REST API ดึงข้อมูลทุก 2 วินาที แล้วอยู่ดีๆ ก็เจอ ConnectionError: timeout ตอนตลาดเป็นขาขึ้น พอเปลี่ยนไปใช้ WebSocket ก็กลับมาเจอ 401 Unauthorized เพราะ Token หมดอายุ จบวันขาดทุนไป 200 ดอลลาร์เพราะคำสั่งซื้อไม่ทันเวลา บทความนี้จะสอนวิธีแก้ปัญหาจริงที่พบบ่อยที่สุด 3 กรณี พร้อมโค้ดที่รันได้ทันที

ทำไมต้องมาตรฐานข้อมูลตลาดคริปโต

แต่ละตลาดมีรูปแบบ API ไม่เหมือนกัน Binance ใช้ symbol, Coinbase ใช้ product_id, Kraken ใช้ pair ถ้าไม่มีมาตรฐาน โค้ดจะกระจัดกระจาย แก้ยาก และผิดพลาดง่าย การสร้าง Layer กลางเพื่อ Normalize ข้อมูลให้เป็นรูปแบบเดียวกันจะช่วยประหยัดเวลาพัฒนาได้มาก และยังทำให้ระบบทำงานเสถียรขึ้นเมื่อเชื่อมต่อหลายตลาดพร้อมกัน

REST API กับ WebSocket: อะไรเหมาะกับอะไร

REST API — ดึงข้อมูลตามคำขอ

REST API เหมาะกับการดึงข้อมูลที่ไม่เร่งด่วน เช่น ดึงยอด Balance, ประวัติ Order, รายละเอียดคู่เทรด ข้อดีคือเข้าใจง่าย ใช้ HTTP Method มาตรฐาน GET POST DELETE ดีบักง่าย ผลลัพธ์ Cache ได้ ข้อเสียคือต้อง Poll ตลอด ถ้าดึงบ่อยจะเปลือง Request และมี Latency ระหว่าง Poll แต่ละครั้ง

WebSocket — รับข้อมูลแบบเรียลไทม์

WebSocket เหมาะกับการรับข้อมูลราคาแบบเรียลไทม์ รับ Trade Updates ทันทีที่เกิดขึ้น รับ Order Book Updates ทุกครั้งที่มีการเปลี่ยนแปลง ข้อดีคือ Latency ต่ำมาก ไม่ต้อง Poll ข้อเสียคือต้องจัดการ Reconnection เอง ยากกว่าในการ Debug และ Connection อาจหลุดเมื่อ Network มีปัญหา

ตัวอย่างโค้ด: REST API

โค้ดด้านล่างเป็นตัวอย่างการดึงข้อมูลราคาจาก Binance ผ่าน REST API ด้วย Python พร้อมการจัดการ Error เบื้องต้น

import requests
import time
from datetime import datetime

BASE_URL = "https://api.binance.com"
HEADERS = {
    "X-MBX-APIKEY": "YOUR_BINANCE_API_KEY"
}

def get_ticker(symbol: str) -> dict:
    """ดึงข้อมูลราคาล่าสุดของคู่เทรด"""
    endpoint = "/api/v3/ticker/24hr"
    params = {"symbol": symbol}
    
    try:
        response = requests.get(
            f"{BASE_URL}{endpoint}",
            params=params,
            headers=HEADERS,
            timeout=5
        )
        response.raise_for_status()
        data = response.json()
        
        return {
            "symbol": data["symbol"],
            "price": float(data["lastPrice"]),
            "volume_24h": float(data["volume"]),
            "timestamp": datetime.now().isoformat()
        }
    except requests.exceptions.Timeout:
        print(f"Connection timeout สำหรับ {symbol}")
        return None
    except requests.exceptions.HTTPError as e:
        print(f"HTTP error {e.response.status_code}: {e}")
        return None

def normalize_ticker(data: dict) -> dict:
    """แปลงข้อมูลให้เป็นมาตรฐานกลาง"""
    return {
        "exchange": "binance",
        "pair": data["symbol"],
        "last_price": data["price"],
        "volume_24h_usdt": data["volume_24h"],
        "fetched_at": data["timestamp"]
    }

ทดสอบการใช้งาน
if __name__ == "__main__":
    symbols = ["BTCUSDT", "ETHUSDT", "BNBUSDT"]
    
    for symbol in symbols:
        raw_data = get_ticker(symbol)
        if raw_data:
            normalized = normalize_ticker(raw_data)
            print(f"{normalized['pair']}: ${normalized['last_price']:,.2f}")
        
        time.sleep(0.2)  # รอเพื่อไม่ให้เกิน Rate Limit

ตัวอย่างโค้ด: WebSocket

โค้ดด้านล่างใช้ WebSocket เพื่อรับข้อมูลราคาแบบเรียลไทม์ มีระบบ Reconnection อัตโนมัติเมื่อ Connection หลุด

import websockets
import asyncio
import json
from datetime import datetime

BINANCE_WS_URL = "wss://stream.binance.com:9443/ws"

class WebSocketClient:
    def __init__(self):
        self.connection = None
        self.reconnect_delay = 1
        self.max_reconnect_delay = 60
        self.running = False
        
    async def connect(self, symbols: list):
        """เชื่อมต่อ WebSocket สำหรับหลายคู่เทรด"""
        streams = [f"{s.lower()}@ticker" for s in symbols]
        subscribe_msg = {
            "method": "SUBSCRIBE",
            "params": streams,
            "id": 1
        }
        
        while self.running:
            try:
                async with websockets.connect(BINANCE_WS_URL) as ws:
                    self.connection = ws
                    self.reconnect_delay = 1  # Reset delay
                    
                    await ws.send(json.dumps(subscribe_msg))
                    print(f"เชื่อมต่อสำเร็จ รอรับข้อมูล...")
                    
                    async for message in ws:
                        data = json.loads(message)
                        normalized = self.normalize_websocket_data(data)
                        if normalized:
                            print(f"{normalized['pair']}: ${normalized['last_price']:,.2f} | Vol: ${normalized['volume_24h_usdt']:,.0f}")
                            
            except websockets.exceptions.ConnectionClosed:
                print(f"Connection หลุด รอเชื่อมต่อใหม่ใน {self.reconnect_delay} วินาที...")
                await asyncio.sleep(self.reconnect_delay)
                self.reconnect_delay = min(self.reconnect_delay * 2, self.max_reconnect_delay)
                
            except Exception as e:
                print(f"เกิดข้อผิดพลาด: {type(e).__name__}: {e}")
                await asyncio.sleep(self.reconnect_delay)

    def normalize_websocket_data(self, data: dict) -> dict:
        """แปลง WebSocket data ให้เป็นมาตรฐานกลาง"""
        if "e" not in data:  # ไม่ใช่ Ticker event
            return None
            
        return {
            "exchange": "binance",
            "pair": data["s"],
            "last_price": float(data["c"]),
            "volume_24h_usdt": float(data["q"]),
            "price_change_pct": float(data["P"]),
            "fetched_at": datetime.now().isoformat()
        }
    
    async def start(self, symbols: list):
        """เริ่ม WebSocket client"""
        self.running = True
        await self.connect(symbols)
    
    def stop(self):
        """หยุด WebSocket client"""
        self.running = False

ทดสอบการใช้งาน
if __name__ == "__main__":
    client = WebSocketClient()
    
    try:
        asyncio.run(client.start(["btcusdt", "ethusdt", "bnbusdt"]))
    except KeyboardInterrupt:
        print("\nกำลังหยุด...")
        client.stop()

ตัวอย่างโค้ด: Hybrid Approach สำหรับระบบ Production

ระบบจริงควรใช้ทั้ง REST และ WebSocket ร่วมกัน REST สำหรับคำสั่งที่ต้องยืนยัน (Order, Balance) WebSocket สำหรับข้อมูลราคาและ Order Book

import requests
import websockets
import asyncio
import json
from enum import Enum
from typing import Optional, Callable
from dataclasses import dataclass
from datetime import datetime

class DataSource(Enum):
    REST_API = "rest"
    WEBSOCKET = "ws"

@dataclass
class MarketData:
    pair: str
    last_price: float
    volume_24h: float
    source: DataSource
    timestamp: str

class CryptoDataAggregator:
    """รวมข้อมูลจากหลายตลาดผ่าน REST และ WebSocket"""
    
    def __init__(self):
        self.ticker_cache = {}
        self.rest_session = requests.Session()
        self.rest_session.headers.update({"X-MBX-APIKEY": "YOUR_BINANCE_API_KEY"})
        self.ws_clients = {}
        self.callbacks = []
        
    def register_callback(self, func: Callable):
        """ลงทะเบียน callback สำหรับรับข้อมูลเรียลไทม์"""
        self.callbacks.append(func)
        
    async def start_websocket_feed(self, symbols: list):
        """เริ่มรับข้อมูล WebSocket สำหรับหลายคู่เทรด"""
        url = "wss://stream.binance.com:9443/ws"
        streams = "/".join([f"{s.lower()}@ticker" for s in symbols])
        
        async with websockets.connect(f"{url}/{streams}") as ws:
            async for msg in ws:
                data = json.loads(msg)
                if "e" in data:
                    ticker = MarketData(
                        pair=data["s"],
                        last_price=float(data["c"]),
                        volume_24h=float(data["q"]),
                        source=DataSource.WEBSOCKET,
                        timestamp=datetime.now().isoformat()
                    )
                    self.ticker_cache[ticker.pair] = ticker
                    
                    # แจ้ง callback ทุกตัว
                    for cb in self.callbacks:
                        await cb(ticker)
                        
    def get_cached_price(self, pair: str) -> Optional[float]:
        """ดึงราคาล่าสุดจาก cache"""
        if pair in self.ticker_cache:
            return self.ticker_cache[pair].last_price
        return None
        
    def fetch_order_book(self, symbol: str, limit: int = 20) -> dict:
        """ดึง Order Book ผ่าน REST API"""
        url = "https://api.binance.com/api/v3/depth"
        params = {"symbol": symbol, "limit": limit}
        
        try:
            response = self.rest_session.get(url, params=params, timeout=5)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 401:
                raise PermissionError("API Key ไม่ถูกต้องหรือหมดอายุ")
            raise
            
    def normalize_order_book(self, data: dict, pair: str) -> dict:
        """แปลง Order Book ให้เป็นมาตรฐานกลาง"""
        return {
            "pair": pair,
            "bids": [[float(p), float(q)] for p, q in data.get("bids", [])],
            "asks": [[float(p), float(q)] for p, q in data.get("asks", [])],
            "timestamp": datetime.now().isoformat()
        }

ตัวอย่างการใช้งาน
async def on_ticker_update(ticker: MarketData):
    print(f"[WS] {ticker.pair}: ${ticker.last_price:,.2f}")

async def main():
    aggregator = CryptoDataAggregator()
    aggregator.register_callback(on_ticker_update)
    
    # เริ่ม WebSocket ใน background
    ws_task = asyncio.create_task(
        aggregator.start_websocket_feed(["BTCUSDT", "ETHUSDT"])
    )
    
    # ทำงานอื่นขณะรอข้อมูล
    await asyncio.sleep(5)
    
    # ดึง Order Book ผ่าน REST
    try:
        ob = aggregator.fetch_order_book("BTCUSDT", limit=10)
        normalized_ob = aggregator.normalize_order_book(ob, "BTCUSDT")
        print(f"Best Bid: {normalized_ob['bids'][0]}")
        print(f"Best Ask: {normalized_ob['asks'][0]}")
    except PermissionError as e:
        print(f"เกิดข้อผิดพลาด: {e}")
        
    ws_task.cancel()

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

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

กรณีที่ 1: ConnectionError: timeout

สาเหตุ: เกิดจาก Server ไม่ตอบสนองภายในเวลาที่กำหนด หรือ Network มีปัญหา มักเกิดเวลาตลาดมีความผันผวนสูง Server รับโหลดไม่ไหว

วิธีแก้: เพิ่ม timeout และ retry logic พร้อม Exponential Backoff

import time
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

def create_session_with_retry(retries=3, backoff_factor=0.5):
    """สร้าง requests Session พร้อมระบบ Retry อัตโนมัติ"""
    session = requests.Session()
    
    retry_strategy = Retry(
        total=retries,
        backoff_factor=backoff_factor,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET", "POST"]
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    
    return session

การใช้งาน
session = create_session_with_retry(retries=5, backoff_factor=1)

def safe_request(url, **kwargs):
    """ส่ง request พร้อมจัดการ timeout และ retry"""
    kwargs.setdefault("timeout", (3, 10))  # (connect, read) timeout
    
    for attempt in range(5):
        try:
            response = session.get(url, **kwargs)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.Timeout:
            wait = 2 ** attempt  # 2, 4, 8, 16, 32 วินาที
            print(f"Timeout ครั้งที่ {attempt+1} รอ {wait} วินาที...")
            time.sleep(wait)
        except requests.exceptions.ConnectionError as e:
            print(f"Connection error: {e}")
            time.sleep(2 ** attempt)
            
    raise Exception("Request ล้มเหลวหลังจากลอง 5 ครั้ง")

กรณีที่ 2: 401 Unauthorized

สาเหตุ: API Key หมดอายุ, Key ไม่ถูกต้อง, หมดโควต้า, หรือ Signature ไม่ถูกต้อง มักเกิดเวลาใช้งานนานเกินไปโดยไม่ Refresh Token

วิธีแก้: ตรวจสอบ Response Header และ Implement Token Refresh

import hmac
import hashlib
import time
from datetime import datetime, timedelta

class BinanceAuth:
    """จัดการ Authentication สำหรับ Binance API"""
    
    def __init__(self, api_key: str, api_secret: str):
        self.api_key = api_key
        self.api_secret = api_secret
        self.token_expires_at = None
        self.access_token = None
        
    def generate_signature(self, params: dict) -> str:
        """สร้าง HMAC SHA256 Signature"""
        query_string = "&".join([f"{k}={v}" for k, v in sorted(params.items())])
        signature = hmac.new(
            self.api_secret.encode("utf-8"),
            query_string.encode("utf-8"),
            hashlib.sha256
        ).hexdigest()
        return signature
    
    def get_auth_headers(self) -> dict:
        """สร้าง Headers สำหรับ Signed Request"""
        return {
            "X-MBX-APIKEY": self.api_key,
            "Content-Type": "application/x-www-form-urlencoded"
        }
    
    def is_token_valid(self) -> bool:
        """ตรวจสอบว่า Token ยังไม่หมดอายุหรือไม่"""
        if not self.token_expires_at:
            return False
        return datetime.now() < self.token_expires_at - timedelta(minutes=5)
    
    def make_signed_request(self, method: str, endpoint: str, params: dict = None):
        """ส่ง Signed Request พร้อมจัดการ Authentication"""
        params = params or {}
        params["timestamp"] = int(time.time() * 1000)
        params["recvWindow"] = 5000
        
        params["signature"] = self.generate_signature(params)
        
        headers = self.get_auth_headers()
        url = f"https://api.binance.com{endpoint}"
        
        if method == "GET":
            response = requests.get(url, params=params, headers=headers)
        else:
            response = requests.post(url, data=params, headers=headers)
            
        if response.status_code == 401:
            # Token หมดอายุ ลอง refresh
            if not self.is_token_valid():
                raise PermissionError("API Key หมดอายุ กรุณาสร้างใหม่")
            raise PermissionError("Signature ไม่ถูกต้อง ตรวจสอบ API Secret")
            
        response.raise_for_status()
        return response.json()

การใช้งาน
auth = BinanceAuth("YOUR_API_KEY", "YOUR_API_SECRET")

try:
    balance = auth.make_signed_request("GET", "/api/v3/account")
    print(f"ยอด USDT: {balance['balances'][0]['free']}")
except PermissionError as e:
    print(f"เกิดข้อผิดพลาด: {e}")

กรณีที่ 3: WebSocket Connection หลุดเอง

สาเหตุ: Server ปิด Connection, Network มีปัญหา, Firewall บล็อก, หรือเกินเวลา Keep-Alive

วิธีแก้: Implement Heartbeat และ Auto Reconnection

import asyncio
import websockets
import json
from datetime import datetime

class RobustWebSocket:
    """WebSocket Client พร้อมระบบ Auto Reconnection และ Heartbeat"""
    
    HEARTBEAT_INTERVAL = 30  # วินาที
    MAX_RECONNECT_DELAY = 120
    PING_TIMEOUT = 10
    
    def __init__(self, url: str):
        self.url = url
        self.ws = None
        self.reconnect_delay = 1
        self.last_ping = None
        self.is_running = False
        self.subscriptions = set()
        
    async def connect(self):
        """เชื่อมต่อ WebSocket พร้อม Heartbeat"""
        try:
            self.ws = await websockets.connect(
                self.url,
                ping_interval=self.HEARTBEAT_INTERVAL,
                ping_timeout=self.PING_TIMEOUT
            )
            self.reconnect_delay = 1
            self.last_ping = datetime.now()
            print("WebSocket เชื่อมต่อสำเร็จ")
            
            # Re-subscribe to previous streams
            if self.subscriptions:
                await self.subscribe(list(self.subscriptions))
                
            return True
            
        except Exception as e:
            print(f"เชื่อมต่อไม่ได้: {e}")
            return False
    
    async def subscribe(self, streams: list):
        """ส่งคำสั่ง Subscribe"""
        if not self.ws:
            return False
            
        subscribe_msg = {
            "method": "SUBSCRIBE",
            "params": streams,
            "id": int(datetime.now().timestamp())
        }
        
        await self.ws.send(json.dumps(subscribe_msg))
        self.subscriptions.update(streams)
        print(f"Subscribe: {streams}")
        return True
    
    async def listen(self, callback):
        """ฟังข้อมูลที่เข้ามาพร้อมจัดการ Reconnection"""
        self.is_running = True
        
        while self.is_running:
            try:
                if not self.ws or self.ws.closed:
                    if not await self.connect():
                        await self._wait_before_reconnect()
                        continue
                
                async for message in self.ws:
                    try:
                        data = json.loads(message)
                        await callback(data)
                    except json.JSONDecodeError:
                        print("รับข้อมูลที่ parse ไม่ได้")
                        
            except websockets.exceptions.ConnectionClosed as e:
                print(f"Connection หลุด: {e}")
                await self._wait_before_reconnect()
                
            except Exception as e:
                print(f"เกิดข้อผิดพลาด: {e}")
                await self._wait_before_reconnect()
    
    async def _wait_before_reconnect(self):
        """รอก่อนเชื่อมต่อใหม่ พร้อม Exponential Backoff"""
        print(f"รอเชื่อมต่อใหม่ใน {self.reconnect_delay} วินาที...")
        await asyncio.sleep(self.reconnect_delay)
        self.reconnect_delay = min(self.reconnect_delay * 2, self.MAX_RECONNECT_DELAY)
    
    def stop(self):
        """หยุด WebSocket"""
        self.is_running = False
        if self.ws:
            asyncio.create_task(self.ws.close())

การใช้งาน
async def on_message(data):
    if "e" in data:
        print(f"[{data['s']}] ${float(data['c']):,.2f}")

async def main():
    ws = RobustWebSocket("wss://stream.binance.com:9443/ws")
    await ws.subscribe(["btcusdt@ticker", "ethusdt@ticker"])
    await ws.listen(on_message)

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

เปรียบเทียบ REST API กับ WebSocket

คุณสมบัติ	REST API	WebSocket
ความเร็วในการตอบสนอง	100-500ms ต่อ Request	<50ms แบบเรียลไทม์
ความถี่ในการอัปเดต	ถี่สุด 1-2 วินาทีต่อครั้ง	ทุกครั้งที่มีการเปลี่ยนแปลง
การใช้ Bandwidth	สูง — ส่ง Header ทุกครั้ง	ต่ำ — เชื่อมต่อค้าง
ความซับซ้อนในการ Implement	ง่าย	ยากกว่า (ต้องจัดการ Reconnect)
เหมาะกับงาน	ดึงข้อมูลที่ไม่เร่งด่วน, คำสั่งซื้อขาย	ราคาแบบเรียลไทม์, Order Book

เหมาะกับใคร / ไม่เหมาะกับใคร

เหมาะกับ REST API:

• ผู้เริ่มต้นพัฒนาระบบเทรดอัตโนมัติ
• ต้องการดึงข้อมูลที่ไม่เร่งด่วน เช่น ประวัติราคา, ยอด Balance
• ต้องการ Debug ง่าย ใช้ HTTP มาตรฐาน
• ระบบที่ไม่ต้องการ Latency ต่ำมาก

เหมาะกับ WebSocket:

• ระบบ Scalping หรือ High-Frequency Trading
• ต้องการราคาล่าสุดทันทีที่เปลี่ยน
• ระบบที่ต้องรับ Order Book Updates ตลอดเวลา
• พัฒนาโดยมีประสบการณ์จัดการ Connection

ไม่เหมาะกับ WebSocket:

• ระบบง่ายๆ ที่ดึงข้อมูลไม่กี่ครั้งต่อนาที
• ผู้ที่ไม่ถนัดจัดการ Error ของ Connection
• Environment ที่มี Firewall จำกัดการเชื่อมต่อ WebSocket

ราคาและ ROI

การใช้ API ของตลาดคริปโตโดยตรงมีค่าใช้จ่ายเป็นค่าไฟฟ้าและเวลาพัฒนา หากต้