บทนำ: ทำไมข้อมูล Tick-by-Tick ถึงสำคัญ
ในโลกของ Algorithmic Trading และ Quantitative Research ข้อมูลการซื้อขายรายวินาที (Tick Data) คือ "ทองคำ" ของการวิเคราะห์ตลาด ผมเคยทำโปรเจกต์สร้างระบบ Backtest สำหรับสกุลเงินดิจิทัล และพบว่าข้อมูล OHLCV ธรรมดาไม่เพียงพอต่อการตรวจจับ Order Flow หรือ Liquidity Sniper ที่แม่นยำ
OKX Exchange เป็นหนึ่งใน Exchange ที่มี API ให้บริการอย่างครบวงจร วันนี้ผมจะพาทุกท่านไปสร้าง Pipeline สำหรับเก็บข้อมูล Tick-by-Tick ผ่าน WebSocket และ Archive เป็น CSV สำหรับการวิเคราะห์ในอนาคต
สถาปัตยกรรมระบบโดยรวม
ระบบที่เราจะสร้างประกอบด้วย 3 ส่วนหลัก:
- Ingestion Layer: WebSocket Client สำหรับเชื่อมต่อ OKX Public API
- Processing Layer: Python Script สำหรับ Parse, Validate และ Aggregate ข้อมูล
- Storage Layer: Local File System สำหรับเก็บ CSV แบบ Daily Partition
การตั้งค่า Environment และ Dependencies
ก่อนเริ่มต้น ติดตั้ง Package ที่จำเป็น:
pip install okx-sdk websockets pandas numpy python-dotenv asyncio
สร้างไฟล์
.env สำหรับเก็บ Configuration:
# OKX API Configuration
OKX_WS_ENDPOINT=wss://ws.okx.com:8443/ws/v5/public
OKX_ENABLE_SANDBOX=false
Storage Configuration
DATA_OUTPUT_DIR=./tick_data
LOG_LEVEL=INFO
WebSocket Client สำหรับ OKX Trade Channel
OKX ใช้ WebSocket Protocol สำหรับการรับข้อมูล Real-time โดย Trade Channel จะส่งข้อมูลทุกครั้งที่มี Order Match เกิดขึ้น:
import asyncio
import json
import websockets
import pandas as pd
from datetime import datetime
from pathlib import Path
from typing import List, Dict
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class OKXTradeCollector:
"""
WebSocket Client สำหรับเก็บข้อมูล Tick-by-Tick จาก OKX
รองรับหลาย Symbols พร้อมกัน
"""
def __init__(self, symbols: List[str], output_dir: str = "./tick_data"):
self.ws_endpoint = "wss://ws.okx.com:8443/ws/v5/public"
self.symbols = [s.upper() for s in symbols]
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
self.trade_buffer = []
self.buffer_size = 1000 # Flush ทุก 1000 records
def _build_subscribe_message(self) -> dict:
"""สร้าง Subscribe Message สำหรับ OKX WebSocket"""
args = []
for symbol in self.symbols:
# OKX ใช้ format: {instType}-{instId}
# สำหรับ SPOT: BTC-USDT
inst_id = symbol.replace("/", "-")
args.append({
"channel": "trades",
"instId": symbol
})
return {
"op": "subscribe",
"args": args
}
async def _write_to_csv(self, data: List[Dict]):
"""เขียนข้อมูลลง CSV แบบ Daily Partition"""
if not data:
return
df = pd.DataFrame(data)
# แปลง Timestamp
df['ts'] = pd.to_datetime(df['ts'], unit='ms')
date_str = df['ts'].dt.strftime('%Y%m%d').iloc[0]
# เพิ่ม Metadata columns
df['date'] = date_str
df['collected_at'] = datetime.now().isoformat()
# ตรวจสอบว่า Symbol เป็น Spot หรือ Futures
symbol = df['instId'].iloc[0]
if '-' in symbol:
inst_type = 'SPOT' if len(symbol.split('-')) == 2 else 'FUTURES'
else:
inst_type = 'UNKNOWN'
df['inst_type'] = inst_type
# เขียนลง CSV
output_file = self.output_dir / f"trades_{symbol}_{date_str}.csv"
mode = 'a' if output_file.exists() else 'w'
header = mode == 'w'
df.to_csv(output_file, mode=mode, header=header, index=False)
logger.info(f"Written {len(data)} trades to {output_file.name}")
async def _process_message(self, message: str):
"""Process ข้อมูลที่ได้รับจาก WebSocket"""
try:
data = json.loads(message)
# Handle subscribe confirmation
if 'event' in data:
logger.info(f"Subscription event: {data}")
return
# Handle trade data
if 'data' in data:
for trade in data['data']:
processed_trade = {
'instId': trade['instId'],
'tradeId': trade['tradeId'],
'px': float(trade['px']),
'sz': float(trade['sz']),
'side': trade['side'],
'ts': int(trade['ts']),
'tickPx': float(trade.get('tickPx', trade['px'])),
'markPx': float(trade.get('markPx', 0))
}
self.trade_buffer.append(processed_trade)
# Flush buffer เมื่อถึง threshold
if len(self.trade_buffer) >= self.buffer_size:
await self._write_to_csv(self.trade_buffer)
self.trade_buffer = []
except json.JSONDecodeError as e:
logger.error(f"JSON decode error: {e}")
except Exception as e:
logger.error(f"Process error: {e}")
async def start(self, duration_seconds: int = None):
"""เริ่มเชื่อมต่อ WebSocket และเก็บข้อมูล"""
logger.info(f"Starting OKX Trade Collector for: {self.symbols}")
subscribe_msg = self._build_subscribe_message()
try:
async with websockets.connect(self.ws_endpoint) as ws:
# Subscribe to channels
await ws.send(json.dumps(subscribe_msg))
logger.info("Subscription message sent")
# Receive messages
start_time = asyncio.get_event_loop().time()
while True:
if duration_seconds:
elapsed = asyncio.get_event_loop().time() - start_time
if elapsed >= duration_seconds:
logger.info(f"Duration {duration_seconds}s reached, stopping...")
break
message = await asyncio.wait_for(ws.recv(), timeout=30.0)
await self._process_message(message)
except websockets.exceptions.ConnectionClosed as e:
logger.warning(f"Connection closed: {e}")
except asyncio.TimeoutError:
logger.warning("Connection timeout, attempting reconnect...")
finally:
# Flush remaining buffer
if self.trade_buffer:
await self._write_to_csv(self.trade_buffer)
ตัวอย่างการใช้งาน
async def main():
collector = OKXTradeCollector(
symbols=["BTC-USDT", "ETH-USDT", "SOL-USDT"],
output_dir="./tick_data"
)
# เก็บข้อมูล 1 ชั่วโมง (3600 วินาที)
await collector.start(duration_seconds=3600)
if __name__ == "__main__":
asyncio.run(main())
Batch Download Historical Data ผ่าน REST API
นอกจาก Real-time WebSocket แล้ว OKX ยังมี REST API สำหรับดาวน์โหลด Historical Trades ซึ่งเหมาะสำหรับการ Backfill ข้อมูลย้อนหลัง:
import requests
import pandas as pd
import time
from datetime import datetime, timedelta
from typing import List, Optional
from concurrent.futures import ThreadPoolExecutor, as_completed
class OKXHistoricalTradeDownloader:
"""
Download Historical Trade Data จาก OKX REST API
รองรับการดาวน์โหลดแบบ Batch พร้อม Rate Limiting
"""
BASE_URL = "https://www.okx.com"
def __init__(self, output_dir: str = "./historical_trades"):
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
self.session = requests.Session()
self.session.headers.update({
'Content-Type': 'application/json',
'User-Agent': 'OKX-TickData-Downloader/1.0'
})
def _rate_limit(self, calls_per_second: int = 10):
"""Simple Rate Limiter"""
time.sleep(1 / calls_per_second)
def get_historical_trades(
self,
instId: str,
after: Optional[int] = None,
before: Optional[int] = None,
limit: int = 100
) -> List[dict]:
"""
ดึงข้อมูล Trade History จาก OKX API
Args:
instId: Instrument ID เช่น BTC-USDT
after: Cursor สำหรับดึงข้อมูลหลัง timestamp นี้ (milliseconds)
before: Cursor สำหรับดึงข้อมูลก่อน timestamp นี้
limit: จำนวน records ต่อ request (max 100)
Returns:
List of trade records
"""
endpoint = f"{self.BASE_URL}/api/v5/market/trades"
params = {
'instId': instId,
'limit': min(limit, 100)
}
if after:
params['after'] = after
if before:
params['before'] = before
response = self.session.get(endpoint, params=params)
response.raise_for_status()
data = response.json()
if data.get('code') != '0':
raise Exception(f"API Error: {data.get('msg')}")
return data.get('data', [])
def download_date_range(
self,
instId: str,
start_date: datetime,
end_date: datetime,
progress_callback=None
) -> str:
"""
ดาวน์โหลดข้อมูลในช่วงวันที่กำหนด
Returns:
Path ของไฟล์ CSV ที่ดาวน์โหลดสำเร็จ
"""
all_trades = []
# Convert dates to milliseconds
current_ts = int(end_date.timestamp() * 1000)
end_ts = int(start_date.timestamp() * 1000)
print(f"Downloading {instId} from {start_date} to {end_date}")
while current_ts > end_ts:
self._rate_limit(10) # OKX rate limit
trades = self.get_historical_trades(
instId=instId,
after=current_ts,
limit=100
)
if not trades:
break
all_trades.extend(trades)
# ใช้ timestamp ของ record ล่าสุดเป็น cursor สำหรับ request ถัดไป
last_ts = int(trades[-1]['ts'])
current_ts = last_ts - 1 # ขอข้อมูลก่อน record ล่าสุด
if progress_callback:
progress_callback(len(all_trades))
print(f" Downloaded {len(all_trades)} trades... (last ts: {last_ts})")
# Convert to DataFrame and save
df = pd.DataFrame(all_trades)
# Parse timestamp
df['datetime'] = pd.to_datetime(df['ts'], unit='ms')
df['date'] = df['datetime'].dt.strftime('%Y%m%d')
# Convert numeric columns
df['px'] = df['px'].astype(float)
df['sz'] = df['sz'].astype(float)
df['tradeSz'] = df['tradeSz'].astype(float)
# Sort by timestamp
df = df.sort_values('ts').reset_index(drop=True)
# Save to CSV
date_str = start_date.strftime('%Y%m%d')
output_file = self.output_dir / f"trades_{instId}_{date_str}.csv"
df.to_csv(output_file, index=False)
print(f"Saved {len(df)} records to {output_file}")
return str(output_file)
def batch_download(
self,
symbols: List[str],
days_back: int = 7
):
"""
ดาวน์โหลดข้อมูลหลาย Symbols พร้อมกัน
"""
end_date = datetime.now()
start_date = end_date - timedelta(days=days_back)
results = []
with ThreadPoolExecutor(max_workers=3) as executor:
futures = {
executor.submit(
self.download_date_range,
symbol,
start_date,
end_date
): symbol
for symbol in symbols
}
for future in as_completed(futures):
symbol = futures[future]
try:
result = future.result()
results.append(result)
print(f"✓ {symbol} completed: {result}")
except Exception as e:
print(f"✗ {symbol} failed: {e}")
return results
ตัวอย่างการใช้งาน
if __name__ == "__main__":
downloader = OKXHistoricalTradeDownloader(output_dir="./historical_trades")
symbols = ["BTC-USDT", "ETH-USDT", "SOL-USDT"]
# ดาวน์โหลดย้อนหลัง 7 วัน
results = downloader.batch_download(symbols, days_back=7)
Data Schema และ Field Description
ข้อมูล Tick-by-Tick ที่ได้จาก OKX ประกอบด้วย Field สำคัญดังนี้:
| Field | Type | Description | Example |
|---|
| tradeId | string | Unique Trade ID จาก Exchange | 496823348674851858 |
| instId | string | Instrument ID | BTC-USDT |
| px | float | ราคาที่เกิดการ Match | 42150.50 |
| sz | float | ขนาดของ Order ที่ถูก Match | 0.001 |
| side | string | ฝั่งของ Order ที่ถูก Match (buy/sell) | buy |
| ts | integer | Timestamp เป็น milliseconds | 1703894400000 |
| tickPx | float | ราคา Tick ล่าสุด | 42150.50 |
| indexPx | float | ราคา Index (สำหรับ Derivatives) | 42148.23 |
| markPx | float | ราคา Mark (สำหรับ Derivatives) | 42149.10 |
การใช้ Tick Data ร่วมกับ AI Models สำหรับ Trading Analysis
ข้อมูล Tick-by-Tick สามารถนำไปใช้กับ AI Models ได้หลากหลาย เช่น:
- Order Flow Prediction: ทำนาย Direction ของราคาจาก Order Flow patterns
- Market Making Strategy: คำนวณ Optimal Bid-Ask Spread
- Liquidity Analysis: วิเคราะห์ VPIN (Volume-Synchronized Probability of Informed Trading)
- Anomaly Detection: ตรวจจับ Whale Activities หรือ Manipulations
สำหรับการประมวลผลข้อมูลจำนวนมากด้วย AI Models ผมแนะนำให้ใช้
HolySheep AI ซึ่งมีความเร็วในการประมวลผลต่ำกว่า 50ms และรองรับ Models หลากหลาย เช่น GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash และ DeepSeek V3.2
ตัวอย่างการใช้ HolySheep AI สำหรับ Order Flow Analysis
หลังจากเก็บข้อมูล Tick แล้ว คุณสามารถใช้ HolySheep API สำหรับวิเคราะห์ด้วย LLM ได้ทันที:
import requests
import pandas as pd
from pathlib import Path
HolySheep AI Configuration
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
def analyze_order_flow_with_ai(csv_file: str, api_key: str) -> dict:
"""
ใช้ AI วิเคราะห์ Order Flow จากไฟล์ CSV
"""
# อ่านข้อมูล Tick
df = pd.read_csv(csv_file)
# คำนวณ Order Flow Metrics
buy_volume = df[df['side'] == 'buy']['sz'].sum()
sell_volume = df[df['side'] == 'sell']['sz'].sum()
total_volume = buy_volume + sell_volume
buy_ratio = buy_volume / total_volume if total_volume > 0 else 0.5
# สร้าง Summary สำหรับ AI
summary = f"""
Order Flow Analysis Summary:
- Total Trades: {len(df)}
- Buy Volume: {buy_volume:.4f}
- Sell Volume: {sell_volume:.4f}
- Buy Ratio: {buy_ratio:.2%}
- Price Range: {df['px'].min():.2f} - {df['px'].max():.2f}
- Unique Timestamps: {df['ts'].nunique()}
"""
# เรียก HolySheep AI API
response = requests.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [
{
"role": "system",
"content": """You are an expert in cryptocurrency order flow analysis.
Analyze the provided order flow metrics and provide insights about:
1. Market sentiment (bullish/bearish/neutral)
2. Potential whale activities
3. Recommendations for trading decisions
"""
},
{
"role": "user",
"content": f"Analyze this order flow data:\n{summary}"
}
],
"temperature": 0.3,
"max_tokens": 500
}
)
result = response.json()
return {
"metrics": {
"buy_volume": buy_volume,
"sell_volume": sell_volume,
"buy_ratio": buy_ratio
},
"ai_analysis": result.get("choices", [{}])[0].get("message", {}).get("content", ""),
"usage": result.get("usage", {})
}
def batch_analyze_trading_days(
data_dir: str = "./tick_data",
symbol: str = "BTC-USDT"
):
"""
วิเคราะห์ Order Flow หลายวันพร้อมกัน
"""
import os
csv_files = [
os.path.join(data_dir, f)
for f in os.listdir(data_dir)
if symbol in f and f.endswith('.csv')
]
results = []
for csv_file in csv_files:
print(f"Analyzing {csv_file}...")
analysis = analyze_order_flow_with_ai(
csv_file,
HOLYSHEEP_API_KEY
)
results.append({
"file": Path(csv_file).name,
"buy_ratio": analysis["metrics"]["buy_ratio"],
"analysis": analysis["ai_analysis"]
})
print(f" Buy Ratio: {analysis['metrics']['buy_ratio']:.2%}")
# สร้างรายงานสรุป
summary_df = pd.DataFrame(results)
summary_df.to_csv(f"./order_flow_summary_{symbol}.csv", index=False)
return results
ตัวอย่างการใช้งาน
if __name__ == "__main__":
result = analyze_order_flow_with_ai(
"./tick_data/trades_BTC-USDT_20240130.csv",
HOLYSHEEP_API_KEY
)
print("AI Analysis:")
print(result["ai_analysis"])
print(f"\nToken Usage: {result['usage']}")
Best Practices สำหรับ Production Deployment
1. Database Storage แทน File System
สำหรับข้อมูลจำนวนมาก แนะนำให้ใช้ TimescaleDB หรือ InfluxDB แทน CSV:
from sqlalchemy import create_engine, Column, BigInteger, Float, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime
Base = declarative_base()
class TradeRecord(Base):
__tablename__ = 'trades'
id = Column(BigInteger, primary_key=True, autoincrement=True)
trade_id = Column(String(50), unique=True, index=True)
inst_id = Column(String(20), index=True)
price = Column(Float)
size = Column(Float)
side = Column(String(10))
timestamp = Column(BigInteger, index=True)
created_at = Column(DateTime, default=datetime.utcnow)
# Composite index for time-range queries
__table_args__ = (
Index('idx_inst_timestamp', 'inst_id', 'timestamp'),
)
class TradeDatabaseManager:
def __init__(self, connection_string: str):
self.engine = create_engine(connection_string)
Base.metadata.create_all(self.engine)
self.Session = sessionmaker(bind=self.engine)
def bulk_insert(self, trades: List[dict], batch_size: int = 1000):
session = self.Session()
try:
for i in range(0, len(trades), batch_size):
batch = trades[i:i + batch_size]
records = [
TradeRecord(
trade_id=t['tradeId'],
inst_id=t['instId'],
price=float(t['px']),
size=float(t['sz']),
side=t['side'],
timestamp=int(t['ts'])
)
for t in batch
]
session.bulk_save_objects(records)
session.commit()
logger.info(f"Inserted {len(trades)} trades")
except Exception as e:
session.rollback()
logger.error(f"Insert failed: {e}")
raise
finally:
session.close()
def query_range(
self,
inst_id: str,
start_ts: int,
end_ts: int
) -> pd.DataFrame:
session = self.Session()
try:
result = session.query(TradeRecord).filter(
TradeRecord.inst_id == inst_id,
TradeRecord.timestamp >= start_ts,
TradeRecord.timestamp <= end_ts
).order_by(TradeRecord.timestamp).all()
return pd.DataFrame([{
'tradeId': r.trade_id,
'instId': r.inst_id,
'px': r.price,
'sz': r.size,
'side': r.side,
'ts': r.timestamp
} for r in result])
finally:
session.close()
2. High Availability Setup
สำหรับ Production แนะนำให้ใช้ Redis สำหรับ Reconnection และ State Management:
import redis.asyncio as redis
import json
import asyncio
class RedisStateManager:
def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis = redis.from_url(redis_url, decode_responses=True)
async def save_last_cursor(self, symbol: str, ts: int):
"""เก็บ Cursor ล่าสุดสำหรับ Reconnection"""
key = f"okx:cursor:{symbol}"
await self.redis.set(key, str(ts))
await self.redis.expire(key, 86400 * 7) # 7 days TTL
async def get_last_cursor(self, symbol: str) -> int:
"""ดึง Cursor ล่าสุด"""
key = f"okx:cursor:{symbol}"
cursor = await self.redis.get(key)
return int(cursor) if cursor else None
async def increment_counter(self, metric: str):
"""นับจำนวน Trades ที่รับได้"""
key = f"okx:counter:{metric}"
return await self.redis.incr(key)
async def health_check(self) -> bool:
"""ตรวจสอบ Connection"""
try:
await self.redis.ping()
return True
except:
return False
ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
1. WebSocket Connection Closed Unexpectedly (Error Code 1006)
ปัญหานี้เกิดจาก Server ปิด Connection โดยไม่มี Close Frame ซึ่งมักเกิดจาก:
# ❌ วิธีที่ไม่ถูกต้อง - ไม่มี Reconnection Logic
async def connect_websocket():
async with websockets.connect(WSS_URL) as ws:
while True:
msg = await ws.recv()
process(msg)
# ✅ วิธีที่ถูกต้อง - Implement Reconnection with Exponential Backoff
import asyncio
class RobustWebSocketClient:
def __init__(self, url, max_retries=10, base_delay=1):
self.url = url
self.max_retries = max_retries
self.base_delay = base_delay
async def connect_with_retry(self):
retries = 0
while retries < self.max_retries:
try:
async with websockets.connect(self.url) as ws:
logger.info(f"Connected successfully")
retries = 0 # Reset on successful connection
while True:
msg = await asyncio.wait_for(ws.recv(), timeout=30)
await self.process(msg)
except websockets.exceptions.ConnectionClosed as e:
retries += 1
delay = min(self.base_delay * (2 ** retries), 60)
logger.warning(
f"Connection closed: {e.code} - "
f"Reconnecting in {delay}s (attempt {retries}/{self.max_retries})"
)
await asyncio.sleep(delay
แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง