บทนำ: ทำไมการดึงข้อมูล Crypto ถึงล้มเหลวบ่อยครั้ง
ผมเคยเจอปัญหานี้ในโปรเจกต์จริง: ระบบ Statistical Arbitrage ที่พัฒนามา 3 เดือน พังทั้งระบบเพราะ API rate limit ของ exchange ต่างประเทศ ทำให้ข้อมูลหายระหว่างทาง และสุดท้ายฟีเจอร์ที่สร้างมาก็กลายเป็น Garbage In Garbage Out
บทความนี้จะสอนคุณวิธีสร้าง data pipeline ที่ robust สำหรับ crypto arbitrage, การทำ feature engineering ที่เหมาะสมกับโมเดล statistical arbitrage และวิธีใช้ AI API ในการประมวลผลข้อมูลอย่างมีประสิทธิภาพ
Statistical Arbitrage คืออะไร และทำไมต้องเริ่มจาก Data
Statistical Arbitrage คือกลยุทธ์ที่หากำไรจากความผิดปกติชั่วคราวของราคาระหว่างสินทรัพย์ที่มีความสัมพันธ์กัน เช่น:
- Bitcoin บน Binance vs Coinbase
- ETH/USDT บน spot vs futures
- สเปรดระหว่าง DEX และ CEX
หัวใจสำคัญคือ คุณภาพข้อมูล ถ้าข้อมูลมี delay, missing values หรือ outlier ที่ไม่ได้จัดการ ทุกการคำนวณ spread จะผิดเพี้ยน และโมเดลจะพยากรณ์ผิดทิศ
การตั้งค่า Environment และการติดตั้ง Dependencies
# สร้าง virtual environment แยกสำหรับโปรเจกต์นี้
python -m venv crypto_arb_env
source crypto_arb_env/bin/activate # Linux/Mac
crypto_arb_env\Scripts\activate # Windows
ติดตั้ง dependencies ที่จำเป็น
pip install requests aiohttp pandas numpy ccxt python-dotenv
pip install ta scikit-learn xgboost
ติดตั้ง HolySheep AI SDK สำหรับ AI processing
pip install holysheep-ai
สร้างไฟล์ .env สำหรับเก็บ API keys
cat > .env << 'EOF'
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
BINANCE_API_KEY=your_binance_key
BINANCE_SECRET=your_binance_secret
EOF
Data Pipeline: การดึงข้อมูลจากหลาย Exchange
สำหรับ Statistical Arbitrage ที่ดี คุณต้องการข้อมูลจากหลาย exchange พร้อมกัน และต้องจัดการ latency อย่างเหมาะสม
import asyncio
import aiohttp
import ccxt
import pandas as pd
import numpy as np
from datetime import datetime
from typing import Dict, List, Optional
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class CryptoDataCollector:
"""
คลาสสำหรับรวบรวมข้อมูลจากหลาย exchange
ออกแบบมาเพื่อรองรับ Statistical Arbitrage
"""
def __init__(self):
# รองรับ exchange หลักๆ สำหรับ arbitrage
self.exchanges = {
'binance': ccxt.binance({'enableRateLimit': True}),
'coinbase': ccxt.coinbaseprime({'enableRateLimit': True}),
'kraken': ccxt.kraken({'enableRateLimit': True}),
'bybit': ccxt.bybit({'enableRateLimit': True}),
}
self.price_data = {}
async def fetch_ticker_async(self, session: aiohttp.ClientSession,
exchange_name: str, symbol: str) -> Optional[Dict]:
"""ดึงข้อมูล ticker แบบ async พร้อมจัดการ error"""
try:
exchange = self.exchanges[exchange_name]
ticker = await exchange.fetch_ticker(symbol)
return {
'exchange': exchange_name,
'symbol': symbol,
'bid': ticker['bid'],
'ask': ticker['ask'],
'last': ticker['last'],
'volume': ticker['quoteVolume'],
'timestamp': ticker['timestamp'],
'datetime': ticker['datetime'],
'spread': ticker['ask'] - ticker['bid'],
'spread_pct': (ticker['ask'] - ticker['bid']) / ticker['last'] * 100
}
except ccxt.RateLimitExceeded:
logger.warning(f"Rate limit exceeded for {exchange_name}")
await asyncio.sleep(1) # Wait before retry
return None
except ccxt.NetworkError as e:
logger.error(f"Network error {exchange_name}: {e}")
return None
except Exception as e:
logger.error(f"Unexpected error {exchange_name}: {e}")
return None
async def collect_arbitrage_data(self, symbol: str = 'BTC/USDT') -> pd.DataFrame:
"""
ดึงข้อมูลจากทุก exchange และคำนวณ spread
สำหรับ arbitrage opportunity detection
"""
async with aiohttp.ClientSession() as session:
tasks = [
self.fetch_ticker_async(session, exchange, symbol)
for exchange in self.exchanges.keys()
]
results = await asyncio.gather(*tasks)
# กรองเอาผลลัพธ์ที่ไม่ None
valid_data = [r for r in results if r is not None]
if not valid_data:
raise ConnectionError(f"ไม่สามารถดึงข้อมูล {symbol} จาก exchange ใดๆ")
df = pd.DataFrame(valid_data)
# คำนวณ Arbitrage Spread
df['max_bid'] = df['bid'].max()
df['min_ask'] = df['ask'].min()
df['max_ask'] = df['ask'].max()
df['min_bid'] = df['bid'].min()
# คำนวณ gross profit สำหรับ buy low sell high
df['buy_on'] = df.apply(
lambda x: df.loc[df['ask'].idxmin(), 'exchange'] if x['ask'] == df['ask'].min() else None,
axis=1
)
df['sell_on'] = df.apply(
lambda x: df.loc[df['bid'].idxmax(), 'exchange'] if x['bid'] == df['bid'].max() else None,
axis=1
)
# คำนวณ spread ระหว่าง exchange
df['cross_exchange_spread'] = df['max_bid'] - df['min_ask']
df['cross_exchange_spread_pct'] = (
(df['max_bid'] - df['min_ask']) / df['min_ask'] * 100
)
self.price_data[symbol] = df
return df
ทดสอบการทำงาน
async def test_collector():
collector = CryptoDataCollector()
# ดึงข้อมูลจากทุก exchange
df = await collector.collect_arbitrage_data('BTC/USDT')
print("=== Arbitrage Opportunities ===")
print(df[['exchange', 'bid', 'ask', 'cross_exchange_spread_pct']].to_string())
# แสดง opportunity ที่ดีที่สุด
best = df.loc[df['cross_exchange_spread_pct'].idxmax()]
print(f"\nBest Opportunity:")
print(f"Buy on: {best['buy_on']} @ {df['ask'].min()}")
print(f"Sell on: {best['sell_on']} @ {df['bid'].max()}")
print(f"Spread: {best['cross_exchange_spread_pct']:.4f}%")
Run test
asyncio.run(test_collector())
Feature Engineering สำหรับ Statistical Arbitrage Model
หลังจากได้ข้อมูล raw แล้ว ต้องสร้าง features ที่เหมาะสมกับ statistical arbitrage strategy
import pandas as pd
import numpy as np
from ta.momentum import RSIIndicator, StochasticOscillator
from ta.volatility import BollingerBands, AverageTrueRange
from ta.trend import MACD, SMA, EMA
from sklearn.preprocessing import StandardScaler
from typing import Tuple
class ArbitrageFeatureEngineer:
"""
สร้าง features สำหรับ Statistical Arbitrage Model
"""
def __init__(self, lookback_periods: dict = None):
self.lookback_periods = lookback_periods or {
'short': 5,
'medium': 15,
'long': 60
}
self.scaler = StandardScaler()
self.feature_columns = []
def create_spread_features(self, df: pd.DataFrame) -> pd.DataFrame:
"""สร้าง features จาก spread ระหว่าง exchange"""
# Spread พื้นฐาน
df['exchange_spread'] = df['ask'] - df['bid']
df['exchange_spread_pct'] = df['spread_pct']
# Z-score ของ spread (ต่ำกว่า 0 = มี opportunity)
for period in self.lookback_periods.values():
if len(df) >= period:
mean_spread = df['spread'].rolling(window=period).mean()
std_spread = df['spread'].rolling(window=period).std()
df[f'spread_zscore_{period}'] = (df['spread'] - mean_spread) / std_spread
# Cross-exchange spread features
if 'cross_exchange_spread' in df.columns:
for period in self.lookback_periods.values():
if len(df) >= period:
mean_cross = df['cross_exchange_spread'].rolling(window=period).mean()
std_cross = df['cross_exchange_spread'].rolling(window=period).std()
df[f'cross_spread_zscore_{period}'] = (
df['cross_exchange_spread'] - mean_cross
) / std_cross
# Historical percentile
df[f'cross_spread_percentile_{period}'] = (
df['cross_exchange_spread'].rolling(window=period)
.apply(lambda x: pd.Series(x).rank(pct=True).iloc[-1])
)
return df
def create_volatility_features(self, df: pd.DataFrame) -> pd.DataFrame:
"""สร้าง features ที่เกี่ยวกับ volatility"""
# Bollinger Bands
bb = BollingerBands(close=df['last'], window=20, window_dev=2)
df['bb_high'] = bb.bollinger_hband()
df['bb_low'] = bb.bollinger_lband()
df['bb_position'] = (df['last'] - df['bb_low']) / (df['bb_high'] - df['bb_low'])
# ATR
atr = AverageTrueRange(high=df.get('high', df['last']),
low=df.get('low', df['last']),
close=df['last'], window=14)
df['atr'] = atr.average_true_range()
df['atr_pct'] = df['atr'] / df['last'] * 100
# Rolling volatility
for period in [5, 15, 60]:
if len(df) >= period:
df[f'volatility_{period}'] = df['last'].pct_change().rolling(window=period).std()
return df
def create_momentum_features(self, df: pd.DataFrame) -> pd.DataFrame:
"""สร้าง momentum indicators"""
# RSI
rsi = RSIIndicator(close=df['last'], window=14)
df['rsi'] = rsi.rsi()
# Stochastic
stoch = StochasticOscillator(high=df.get('high', df['last']),
low=df.get('low', df['last']),
close=df['last'], window=14)
df['stoch_k'] = stoch.stoch()
df['stoch_d'] = stoch.stoch_signal()
# MACD
macd = MACD(close=df['last'])
df['macd'] = macd.macd()
df['macd_signal'] = macd.macd_signal()
df['macd_diff'] = macd.macd_diff()
# Price momentum
for period in [5, 15, 60]:
if len(df) >= period:
df[f'momentum_{period}'] = df['last'].pct_change(period)
return df
def create_liquidity_features(self, df: pd.DataFrame) -> pd.DataFrame:
"""สร้าง features ที่เกี่ยวกับ liquidity"""
# Volume-based
df['volume_change'] = df['volume'].pct_change()
df['volume_ma_ratio'] = df['volume'] / df['volume'].rolling(20).mean()
# Order book imbalance proxy (ถ้ามีข้อมูล)
if 'bid_volume' in df.columns and 'ask_volume' in df.columns:
df['book_imbalance'] = (
df['bid_volume'] - df['ask_volume']
) / (df['bid_volume'] + df['ask_volume'])
# Spread as liquidity indicator
df['spread_to_volume'] = df['spread'] / df['volume']
return df
def create_historical_features(self, df: pd.DataFrame) -> pd.DataFrame:
"""สร้าง lag features และ rolling statistics"""
# Lag features
for lag in [1, 2, 3, 5, 10]:
df[f'spread_lag_{lag}'] = df['spread'].shift(lag)
df[f'price_lag_{lag}'] = df['last'].shift(lag)
# Rolling statistics
for period in [10, 30, 60]:
if len(df) >= period:
df[f'spread_ma_{period}'] = df['spread'].rolling(window=period).mean()
df[f'spread_std_{period}'] = df['spread'].rolling(window=period).std()
df[f'price_ma_{period}'] = df['last'].rolling(window=period).mean()
df[f'price_std_{period}'] = df['last'].rolling(window=period).std()
# Mean reversion signals
df[f'spread_deviation_{period}'] = (
df['spread'] - df[f'spread_ma_{period}']
) / df[f'spread_std_{period}']
return df
def engineer_features(self, df: pd.DataFrame) -> pd.DataFrame:
"""รวมทุก feature engineering methods"""
df = df.copy()
df = self.create_spread_features(df)
df = self.create_volatility_features(df)
df = self.create_momentum_features(df)
df = self.create_liquidity_features(df)
df = self.create_historical_features(df)
# กำจัด infinite values และ NaN
df = df.replace([np.inf, -np.inf], np.nan)
df = df.fillna(method='ffill').fillna(method='bfill')
# เก็บ feature columns
self.feature_columns = [
col for col in df.columns
if col not in ['exchange', 'symbol', 'datetime', 'buy_on', 'sell_on']
and df[col].dtype in [np.float64, np.int64]
]
return df
def get_features_for_model(self, df: pd.DataFrame) -> Tuple[np.ndarray, List[str]]:
"""ส่งคืน features ที่พร้อมสำหรับ model training"""
X = df[self.feature_columns].values
return X, self.feature_columns
ตัวอย่างการใช้งาน
def example_feature_engineering():
# สร้าง sample data
dates = pd.date_range(start='2024-01-01', periods=500, freq='1min')
sample_data = pd.DataFrame({
'timestamp': dates,
'bid': 42000 + np.random.randn(500).cumsum() * 10,
'ask': 42000 + np.random.randn(500).cumsum() * 10 + 5,
'last': 42000 + np.random.randn(500).cumsum() * 10 + 2.5,
'volume': np.random.rand(500) * 1000000 + 500000,
'high': 42000 + np.random.randn(500).cumsum() * 10 + 20,
'low': 42000 + np.random.randn(500).cumsum() * 10 - 20,
'cross_exchange_spread': np.abs(np.random.randn(500)) * 20
})
sample_data['spread'] = sample_data['ask'] - sample_data['bid']
# Engineer features
engineer = ArbitrageFeatureEngineer()
df_features = engineer.engineer_features(sample_data)
print(f"สร้าง features สำเร็จ: {len(engineer.feature_columns)} features")
print(f"Feature names: {engineer.feature_columns[:10]}...")
# Get features for model
X, feature_names = engineer.get_features_for_model(df_features)
print(f"Feature matrix shape: {X.shape}")
return df_features, engineer
example_feature_engineering()
การใช้ AI API สำหรับ Signal Generation และ Pattern Recognition
นอกจาก features ทางสถิติแล้ว คุณยังสามารถใช้ AI ในการวิเคราะห์ patterns และสร้าง signals เพิ่มเติมได้
import os
import requests
from typing import Dict, List, Optional
import json
from datetime import datetime
class HolySheepArbitrageAI:
"""
ใช้ HolySheep AI API สำหรับ arbitrage signal analysis
Base URL: https://api.holysheep.ai/v1
"""
def __init__(self, api_key: str = None):
self.api_key = api_key or os.getenv('HOLYSHEEP_API_KEY')
self.base_url = 'https://api.holysheep.ai/v1'
self.headers = {
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json'
}
def analyze_arbitrage_opportunity(self, spread_data: Dict) -> Dict:
"""
ใช้ AI วิเคราะห์ arbitrage opportunity
เหมาะสำหรับการตัดสินใจ position sizing และ timing
"""
prompt = f"""
วิเคราะห์ Arbitrage Opportunity ต่อไปนี้:
Symbol: {spread_data.get('symbol', 'BTC/USDT')}
Current Spread: ${spread_data.get('spread', 0):.2f}
Spread %: {spread_data.get('spread_pct', 0):.4f}%
Historical Mean: ${spread_data.get('historical_mean', 0):.2f}
Z-Score: {spread_data.get('zscore', 0):.2f}
Volatility: {spread_data.get('volatility', 0):.4f}
Volume: ${spread_data.get('volume', 0):,.0f}
Timestamp: {spread_data.get('timestamp', 'N/A')}
กรุณาให้คำแนะนำ:
1. ควรเข้า position หรือไม่ (Confidence: 0-100%)
2. Position size ที่แนะนำ (เป็น % ของ capital)
3. Stop loss level
4. Take profit level
5. Risk/Reward ratio
ตอบเป็น JSON format พร้อม explanation
"""
try:
response = requests.post(
f'{self.base_url}/chat/completions',
headers=self.headers,
json={
'model': 'gpt-4.1', # $8/MTok - เหมาะสำหรับ analysis
'messages': [
{'role': 'system', 'content': 'คุณเป็นผู้เชี่ยวชาญ Crypto Arbitrage Trading'},
{'role': 'user', 'content': prompt}
],
'temperature': 0.3,
'response_format': {'type': 'json_object'}
},
timeout=30
)
if response.status_code == 200:
result = response.json()
return {
'success': True,
'analysis': json.loads(result['choices'][0]['message']['content']),
'model_used': 'gpt-4.1',
'cost': self._estimate_cost(result)
}
elif response.status_code == 401:
raise PermissionError('API Key ไม่ถูกต้อง กรุณาตรวจสอบ HolySheep API Key')
elif response.status_code == 429:
raise ConnectionError('Rate limit exceeded - กรุณารอและลองใหม่')
else:
raise Exception(f'API Error: {response.status_code}')
except requests.exceptions.Timeout:
raise ConnectionError('Request timeout - AI service ตอบสนองช้า')
except requests.exceptions.ConnectionError:
raise ConnectionError('ไม่สามารถเชื่อมต่อ AI service')
def generate_market_regime_analysis(self, historical_features: List[Dict]) -> Dict:
"""
วิเคราะห์ market regime (trending, ranging, volatile)
เพื่อปรับ strategy parameters
"""
# เตรียมข้อมูลสรุป
summary = {
'avg_volatility': sum(f.get('volatility', 0) for f in historical_features) / len(historical_features),
'avg_spread': sum(f.get('spread', 0) for f in historical_features) / len(historical_features),
'opportunity_count': sum(1 for f in historical_features if f.get('spread_pct', 0) > 0.1),
'period_length': len(historical_features)
}
prompt = f"""
วิเคราะห์ Market Regime จากข้อมูล {summary['period_length']} ช่วงเวลา:
Average Volatility: {summary['avg_volatility']:.4f}
Average Spread: ${summary['avg_spread']:.2f}
High Spread Opportunity Count: {summary['opportunity_count']}
ระบุว่าตลาดอยู่ใน regime ไหน:
- High Volatility / Trending
- Low Volatility / Ranging
- Mean Reverting
- Breakout
และให้คำแนะนำ strategy adjustment:
"""
try:
response = requests.post(
f'{self.base_url}/chat/completions',
headers=self.headers,
json={
'model': 'deepseek-v3.2', # $0.42/MTok - เหมาะสำหรับ routine analysis
'messages': [
{'role': 'user', 'content': prompt}
],
'temperature': 0.2
},
timeout=20
)
if response.status_code == 200:
result = response.json()
return {
'success': True,
'regime_analysis': result['choices'][0]['message']['content'],
'model_used': 'deepseek-v3.2'
}
else:
raise Exception(f'API Error: {response.status_code}')
except Exception as e:
return {'success': False, 'error': str(e)}
def _estimate_cost(self, response: Dict) -> float:
"""ประมาณค่าใช้จ่ายจาก response"""
usage = response.get('usage', {})
tokens = usage.get('total_tokens', 0)
# GPT-4.1: $8 per 1M tokens
return tokens * 8 / 1_000_000
ตัวอย่างการใช้งาน
def example_ai_analysis():
# ตัวอย่าง spread data
spread_data = {
'symbol': 'BTC/USDT',
'spread': 15.50,
'spread_pct': 0.037,
'historical_mean': 8.25,
'zscore': 2.1,
'volatility': 0.02,
'volume': 2500000,
'timestamp': datetime.now().isoformat()
}
# เรียกใช้ AI analysis
ai = HolySheepArbitrageAI()
try:
result = ai.analyze_arbitrage_opportunity(spread_data)
print("=== AI Analysis Result ===")
print(json.dumps(result['analysis'], indent=2, ensure_ascii=False))
except PermissionError as e:
print(f"Error: {e}")
except ConnectionError as e:
print(f"Connection Error: {e}")
หมายเหตุ: ต้องตั้งค่า HOLYSHEEP_API_KEY ก่อนใช้งานจริง
example_ai_analysis()
สร้าง Arbitrage Strategy แบบ Complete System
import asyncio
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class ArbitrageSignal:
"""โครงสร้างข้อมูลสำหรับ Arbitrage Signal"""
timestamp: datetime
symbol: str
buy_exchange: str
sell_exchange: str
buy_price: float
sell_price: float
spread_pct: float
zscore: float
confidence: float
position_size_pct: float
stop_loss_pct: float
take_profit_pct: float
def to_dict(self) -> Dict:
return {
'timestamp': self.timestamp.isoformat(),
'symbol': self.symbol,
'buy_exchange': self.buy_exchange,
'sell_exchange': self.sell_exchange,
'buy_price': self.buy_price,
'sell_price': self.sell_price,
'spread_pct': self.spread_pct,
'zscore': self.zscore,
'confidence': self.confidence,
'position_size_pct': self.position_size_pct,
'stop_loss_pct': self.stop_loss_pct,
'take_profit_pct': self.take_profit_pct
}
class StatisticalArbitrageStrategy:
"""
Complete Statistical Arbitrage Strategy
ใช้หลักการ Mean Reversion + Z-Score
"""
def __init__(
self,
entry_threshold: float = -2.0, # Z-score threshold สำหรับเข้า position
exit_threshold: float = 0.0, # Z-score threshold สำหรับออก
lookback_period: int = 60, # ช่วงเวลาสำหรับคำนวณ mean/std
max_position_pct: float = 0.1, # ขนาด position สูงสุด (10% ของ capital)
stop_loss_pct: float = 0.5, # Stop loss (0.5%)
take_profit_pct: float = 1.0, # Take profit (1%)
min_spread_pct: float = 0.05 # Spread ขั้นต่ำ (0.05%)
):
self.entry_threshold = entry_threshold
self.exit_threshold = exit_threshold
self.lookback_period = lookback_period
self.max_position_pct = max_position_pct
self.stop_loss_pct = stop_loss_pct
self.take_profit_pct = take_profit_pct
self.min_spread_pct = min_spread_pct
self.price_history: Dict[str, pd.DataFrame] = {}
self.current_positions: List[ArbitrageSignal] = []
self.trade_history: List[Dict] = []
def _calculate_zscore(self, spread: float, history: pd.Series) -> float:
"""คำนวณ Z-score ของ spread"""
if len(history) < self.lookback_period:
return 0.0
recent = history.tail(self.lookback_period)
mean = recent.mean()
std = recent.std()
if std == 0:
return 0.0
return (spread - mean) / std
def _determine_position_size(self, zscore: float, confidence: float = 1.0) -> float:
"""
กำหนดขนาด position ตาม Z-score
- ยิ่ง Z-score ต่ำ (oversold) ยิ่ง position ใหญ่
- แต่ไม่เกิน max_position_pct
"""
# Z-score -2.0 = 100% ของ max position
# Z-score
แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง