บทนำ
การพัฒนาระบบเทรดอัตโนมัติหรือบอทที่เชื่อมต่อกับ Exchange หลายตัวพร้อมกันเป็นโจทย์ที่ซับซ้อน โดยเฉพาะอย่างยิ่งเมื่อต้องรับมือกับความแตกต่างของรูปแบบข้อมูล (Data Format) ระหว่าง OKX และ Binance ซึ่งเป็นสอง Exchange ที่ได้รับความนิยมสูงสุดในตลาดคริปโต
จากประสบการณ์ตรงในการสร้างระบบ Cross-Exchange Arbitrage ที่เชื่อมต่อทั้ง OKX และ Binance พร้อมกัน ผมพบว่าการทำความเข้าใจความแตกต่างเหล่านี้ตั้งแต่แรกจะช่วยประหยัดเวลาได้มาก และในบทความนี้ผมจะแชร์แนวทางการสร้าง Unified Data Layer ที่ทำให้การ switch ระหว่าง Exchange ทำได้อย่างราบรื่น
สำหรับนักพัฒนาที่ต้องการประมวลผลข้อมูลจาก API หลายตัวอย่างมีประสิทธิภาพ ผมแนะนำให้ลองใช้
HolySheep AI ซึ่งมีความหน่วงต่ำกว่า 50 มิลลิวินาทีและรองรับการเชื่อมต่อผ่าน OpenAI-compatible API ได้ทันที
ภาพรวมการเปรียบเทียบ Data Format
ทั้ง OKX และ Binance ต่างก็มี REST API และ WebSocket สำหรับดึงข้อมูล แต่รูปแบบของ response นั้นแตกต่างกันอย่างมีนัยสำคัญ
1. รูปแบบ Ticker/Price Data
สำหรับข้อมูลราคาปัจจุบัน ความแตกต่างหลักอยู่ที่การตั้งชื่อฟิลด์และโครงสร้าง JSON
import requests
import json
class ExchangeTickerFetcher:
"""Class สำหรับดึงข้อมูล Ticker จาก Exchange ต่างๆ"""
def __init__(self):
self.exchange_configs = {
'binance': {
'base_url': 'https://api.binance.com',
'symbols_endpoint': '/api/v3/ticker/24hr'
},
'okx': {
'base_url': 'https://www.okx.com',
'symbols_endpoint': '/api/v5/market/ticker'
}
}
def get_binance_ticker(self, symbol: str) -> dict:
"""ดึงข้อมูล Ticker จาก Binance
ตัวอย่าง Response:
{
"symbol": "BTCUSDT",
"lastPrice": "43250.00",
"priceChange": "1250.00",
"priceChangePercent": "2.98",
"highPrice": "44000.00",
"lowPrice": "42000.00",
"volume": "12345.67",
"quoteVolume": "531234567.89"
}
"""
url = f"{self.exchange_configs['binance']['base_url']}{self.exchange_configs['binance']['symbols_endpoint']}"
params = {'symbol': symbol.upper()}
response = requests.get(url, params=params)
response.raise_for_status()
raw_data = response.json()
# ปรับรูปแบบให้เป็น Standard Format
return {
'exchange': 'binance',
'symbol': raw_data['symbol'],
'price': float(raw_data['lastPrice']),
'price_change_24h': float(raw_data['priceChange']),
'price_change_percent_24h': float(raw_data['priceChangePercent']),
'high_24h': float(raw_data['highPrice']),
'low_24h': float(raw_data['lowPrice']),
'volume_24h': float(raw_data['volume']),
'quote_volume_24h': float(raw_data['quoteVolume'])
}
def get_okx_ticker(self, inst_id: str) -> dict:
"""ดึงข้อมูล Ticker จาก OKX
ตัวอย่าง Response:
{
"instId": "BTC-USDT",
"last": "43255.50",
"lastSz": "0.5",
"askPx": "43256.00",
"askSz": "10",
"bidPx": "43255.00",
"bidSz": "8",
"open24h": "42000.00",
"high24h": "44000.00",
"low24h": "42000.00",
"volCcy24h": "531234567.89",
"vol24h": "12345.67",
"ts": "1703123456789"
}
"""
url = f"{self.exchange_configs['okx']['base_url']}{self.exchange_configs['okx']['symbols_endpoint']}"
params = {'instId': inst_id.upper()}
response = requests.get(url, params=params)
response.raise_for_status()
raw_data = response.json()['data'][0]
# ปรับรูปแบบให้เป็น Standard Format
return {
'exchange': 'okx',
'symbol': inst_id.upper(),
'price': float(raw_data['last']),
'bid': float(raw_data['bidPx']),
'ask': float(raw_data['askPx']),
'price_change_24h': float(raw_data['last']) - float(raw_data['open24h']),
'price_change_percent_24h': ((float(raw_data['last']) - float(raw_data['open24h'])) / float(raw_data['open24h'])) * 100,
'high_24h': float(raw_data['high24h']),
'low_24h': float(raw_data['low24h']),
'volume_24h': float(raw_data['vol24h']),
'quote_volume_24h': float(raw_data['volCcy24h']),
'timestamp': int(raw_data['ts'])
}
ตัวอย่างการใช้งาน
fetcher = ExchangeTickerFetcher()
binance_btc = fetcher.get_binance_ticker('BTCUSDT')
okx_btc = fetcher.get_okx_ticker('BTC-USDT')
print(f"Binance BTC: ${binance_btc['price']:,.2f}")
print(f"OKX BTC: ${okx_btc['price']:,.2f}")
print(f"Spread: ${abs(binance_btc['price'] - okx_btc['price']):,.2f}")
ความแตกต่างหลักที่ต้องระวัง:
- **Binance** ใช้ symbol แบบ concatenated (เช่น BTCUSDT) ส่วน **OKX** ใช้ hyphen separator (เช่น BTC-USDT)
- **Binance** มี priceChangePercent ในตัว แต่ **OKX** ต้องคำนวณเองจาก last และ open24h
- **OKX** มี bid/ask price ใน response เดียวกัน ส่วน **Binance** ต้องเรียก endpoint อื่น
2. รูปแบบ Order Book Data
ข้อมูล Order Book เป็นสิ่งสำคัญสำหรับการคำนวณความลึกของตลาดและ Slippage
import asyncio
import aiohttp
from typing import List, Dict
from dataclasses import dataclass
from collections import defaultdict
@dataclass
class OrderBookLevel:
"""โครงสร้างข้อมูลระดับราคาใน Order Book"""
price: float
quantity: float
total: float = 0.0
class UnifiedOrderBook:
"""Class สำหรับดึงและ normalize Order Book จาก Exchange ต่างๆ"""
@staticmethod
async def fetch_binance_orderbook(symbol: str, limit: int = 100) -> Dict:
"""ดึง Order Book จาก Binance
Response Format:
{
"lastUpdateId": 123456789,
"bids": [["43250.00", "5.5"], ...],
"asks": [["43251.00", "3.2"], ...]
}
"""
url = "https://api.binance.com/api/v3/depth"
params = {'symbol': symbol.upper(), 'limit': limit}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as resp:
data = await resp.json()
return {
'exchange': 'binance',
'symbol': symbol.upper(),
'last_update_id': data['lastUpdateId'],
'bids': [OrderBookLevel(float(b[0]), float(b[1])) for b in data['bids']],
'asks': [OrderBookLevel(float(a[0]), float(a[1])) for a in data['asks']]
}
@staticmethod
async def fetch_okx_orderbook(inst_id: str, sz: int = 100) -> Dict:
"""ดึง Order Book จาก OKX
Response Format:
{
"data": [{
"instId": "BTC-USDT",
"asks": [["43251.00", "3.2", "0"], ...],
"bids": [["43250.00", "5.5", "0"], ...],
"ts": "1703123456789"
}]
}
"""
url = "https://www.okx.com/api/v5/market/books"
params = {'instId': inst_id.upper(), 'sz': sz}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as resp:
data = await resp.json()
book_data = data['data'][0]
# OKX ส่ง timestamp มาด้วย ต้อง sync กับ Binance
return {
'exchange': 'okx',
'symbol': inst_id.upper(),
'timestamp': int(book_data['ts']),
'bids': [OrderBookLevel(float(b[0]), float(b[1])) for b in book_data['bids']],
'asks': [OrderBookLevel(float(a[0]), float(a[1])) for a in book_data['asks']]
}
@staticmethod
def calculate_mid_price(orderbook: Dict) -> float:
"""คำนวณ mid price จาก Order Book"""
best_bid = orderbook['bids'][0].price
best_ask = orderbook['asks'][0].price
return (best_bid + best_ask) / 2
@staticmethod
def calculate_slippage(orderbook: Dict, quantity: float) -> float:
"""คำนวณ slippage สำหรับ quantity ที่กำหนด"""
cumulative_qty = 0.0
weighted_sum = 0.0
for level in orderbook['asks']:
if cumulative_qty + level.quantity >= quantity:
remaining = quantity - cumulative_qty
weighted_sum += remaining * level.price
break
cumulative_qty += level.quantity
weighted_sum += level.quantity * level.price
avg_price = weighted_sum / quantity
mid_price = UnifiedOrderBook.calculate_mid_price(orderbook)
return ((avg_price - mid_price) / mid_price) * 100
async def compare_orderbooks():
"""เปรียบเทียบ Order Book ระหว่าง Exchange"""
symbol_map = {'BTCUSDT': 'BTC-USDT'}
symbol = 'BTCUSDT'
inst_id = 'BTC-USDT'
# ดึงข้อมูลพร้อมกันจากทั้งสอง Exchange
binance_book, okx_book = await asyncio.gather(
UnifiedOrderBook.fetch_binance_orderbook(symbol),
UnifiedOrderBook.fetch_okx_orderbook(inst_id)
)
# คำนวณ mid price
binance_mid = UnifiedOrderBook.calculate_mid_price(binance_book)
okx_mid = UnifiedOrderBook.calculate_mid_price(okx_book)
# คำนวณ slippage สำหรับ 1 BTC
binance_slip = UnifiedOrderBook.calculate_slippage(binance_book, 1.0)
okx_slip = UnifiedOrderBook.calculate_slippage(okx_book, 1.0)
print(f"Binance Mid Price: ${binance_mid:,.2f} | Slippage: {binance_slip:.4f}%")
print(f"OKX Mid Price: ${okx_mid:,.2f} | Slippage: {okx_slip:.4f}%")
print(f"Price Difference: ${abs(binance_mid - okx_mid):,.2f}")
รันการเปรียบเทียบ
asyncio.run(compare_orderbooks())
ข้อแตกต่างสำคัญเรื่อง Order Book:
- **Binance** ใช้
limit สำหรับจำนวนระดับ ส่วน **OKX** ใช้
sz
- **Binance** มี
lastUpdateId สำหรับ order book checksum ส่วน **OKX** ใช้
ts (timestamp)
- **OKX** มีข้อมูล order count ต่อระดับราคา (element ที่ 3) ซึ่ง **Binance** ไม่มี
3. รูปแบบ Trade/History Data
import time
from typing import Optional
from datetime import datetime
class TradeDataNormalizer:
"""Normalize ข้อมูล Trade จาก Exchange ต่างๆ ให้เป็นรูปแบบเดียวกัน"""
@staticmethod
def normalize_binance_trade(trade: dict) -> dict:
"""Normalize Binance trade response
Binance Response:
[
[
1499040000000, // Trade time
"0.016", // Price
"3.700", // Quantity
"0", // Is buyer maker
false // Is best match
]
]
"""
return {
'exchange': 'binance',
'trade_id': None,
'price': float(trade[1]),
'quantity': float(trade[2]),
'quote_quantity': float(trade[1]) * float(trade[2]),
'time': datetime.fromtimestamp(trade[0] / 1000),
'is_buyer_maker': trade[4] == '1' or trade[4] == True,
'is_best_match': trade[5]
}
@staticmethod
def normalize_okx_trade(trade: dict) -> dict:
"""Normalize OKX trade response
OKX Response:
{
"instId": "BTC-USDT",
"tradeId": "123456",
"px": "43250.00",
"sz": "0.5",
"side": "buy",
"ts": "1703123456789"
}
"""
return {
'exchange': 'okx',
'trade_id': trade['tradeId'],
'price': float(trade['px']),
'quantity': float(trade['sz']),
'quote_quantity': float(trade['px']) * float(trade['sz']),
'time': datetime.fromtimestamp(int(trade['ts']) / 1000),
'is_buyer_maker': trade['side'] == 'sell', # sell = maker
'is_best_match': True
}
@staticmethod
def calculate_volume_profile(trades: list, intervals: int = 10) -> dict:
"""คำนวณ volume profile จาก trade history
แบ่ง trades ออกเป็น intervals ส่วนตามราคา
"""
if not trades:
return {}
prices = [t['price'] for t in trades]
volumes = [t['quantity'] for t in trades]
min_price = min(prices)
max_price = max(prices)
price_range = max_price - min_price
if price_range == 0:
return {'equal': sum(volumes)}
bucket_size = price_range / intervals
buckets = defaultdict(float)
for price, volume in zip(prices, volumes):
bucket_idx = min(int((price - min_price) / bucket_size), intervals - 1)
bucket_key = f"{min_price + bucket_idx * bucket_size:.2f}"
buckets[bucket_key] += volume
return dict(buckets)
ตัวอย่างการใช้งาน
normalizer = TradeDataNormalizer()
ดึง recent trades จาก Binance
binance_trades_url = "https://api.binance.com/api/v3/trades"
binance_trades = requests.get(binance_trades_url, params={'symbol': 'BTCUSDT', 'limit': 100}).json()
normalized_binance = [normalizer.normalize_binance_trade(t) for t in binance_trades[:50]]
ดึง recent trades จาก OKX
okx_trades_url = "https://www.okx.com/api/v5/market/trades"
okx_trades = requests.get(okx_trades_url, params={'instId': 'BTC-USDT', 'limit': 100}).json()
normalized_okx = [normalizer.normalize_okx_trade(t) for t in okx_trades['data'][:50]]
คำนวณ volume profile
binance_profile = normalizer.calculate_volume_profile(normalized_binance)
okx_profile = normalizer.calculate_volume_profile(normalized_okx)
print(f"Binance Total Volume: {sum(t['quantity'] for t in normalized_binance):.4f} BTC")
print(f"OKX Total Volume: {sum(t['quantity'] for t in normalized_okx):.4f} BTC")
print(f"Binance Volume Profile (Top 5):")
for price, vol in sorted(binance_profile.items(), key=lambda x: x[1], reverse=True)[:5]:
print(f" ${price}: {vol:.4f} BTC")
4. การสร้าง Unified Data Layer
from abc import ABC, abstractmethod
from typing import Union, Optional
from enum import Enum
class Exchange(Enum):
BINANCE = "binance"
OKX = "okx"
ALL = "all"
class ExchangeAdapter(ABC):
"""Abstract Base Class สำหรับ Exchange Adapter"""
@abstractmethod
def get_ticker(self, symbol: str) -> dict:
pass
@abstractmethod
def get_orderbook(self, symbol: str, depth: int = 20) -> dict:
pass
@abstractmethod
def get_recent_trades(self, symbol: str, limit: int = 100) -> list:
pass
@abstractmethod
def normalize_symbol(self, symbol: str) -> str:
"""แปลง symbol format ระหว่าง Exchange"""
pass
class BinanceAdapter(ExchangeAdapter):
"""Adapter สำหรับ Binance"""
BASE_URL = "https://api.binance.com"
def __init__(self, api_key: str = None, api_secret: str = None):
self.api_key = api_key
self.api_secret = api_secret
def normalize_symbol(self, symbol: str) -> str:
# BTC-USDT -> BTCUSDT หรือ BTCUSDT -> BTCUSDT
return symbol.replace('-', '').replace('_', '').upper()
def get_ticker(self, symbol: str) -> dict:
url = f"{self.BASE_URL}/api/v3/ticker/24hr"
response = requests.get(url, params={'symbol': self.normalize_symbol(symbol)})
data = response.json()
return {
'exchange': 'binance',
'symbol': data['symbol'],
'price': float(data['lastPrice']),
'bid': float(data['lastPrice']) * 0.999, # ประมาณ
'ask': float(data['lastPrice']) * 1.001, # ประมาณ
'volume': float(data['volume']),
'quote_volume': float(data['quoteVolume']),
'timestamp': data['closeTime']
}
def get_orderbook(self, symbol: str, depth: int = 20) -> dict:
url = f"{self.BASE_URL}/api/v3/depth"
response = requests.get(url, params={
'symbol': self.normalize_symbol(symbol),
'limit': depth
})
data = response.json()
return {
'exchange': 'binance',
'symbol': self.normalize_symbol(symbol),
'bids': [[float(p), float(q)] for p, q in data['bids']],
'asks': [[float(p), float(q)] for p, q in data['asks']],
'last_update_id': data['lastUpdateId']
}
def get_recent_trades(self, symbol: str, limit: int = 100) -> list:
url = f"{self.BASE_URL}/api/v3/trades"
response = requests.get(url, params={
'symbol': self.normalize_symbol(symbol),
'limit': limit
})
return [{
'price': float(t['price']),
'quantity': float(t['qty']),
'time': datetime.fromtimestamp(t['time'] / 1000),
'is_buyer_maker': t['isBuyerMaker']
} for t in response.json()]
class OKXAdapter(ExchangeAdapter):
"""Adapter สำหรับ OKX"""
BASE_URL = "https://www.okx.com"
def __init__(self, api_key: str = None, api_secret: str = None, passphrase: str = None):
self.api_key = api_key
self.api_secret = api_secret
self.passphrase = passphrase
def normalize_symbol(self, symbol: str) -> str:
# BTCUSDT -> BTC-USDT
if '-' not in symbol and '_' not in symbol:
# พยายามหา separator
for i, c in enumerate(symbol):
if
แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง