ในฐานะวิศวกรที่ทำงานด้าน Quantitative Trading มากว่า 5 ปี ผมเคยเจอปัญหาเดิมๆ ซ้ำแล้วซ้ำเล่า — การทำ Backtest ด้วยข้อมูล Tick-level บน Exchange อย่าง OKX ต้องแลกกับค่าใช้จ่าย API ที่สูงลิบ และ Data Pipeline ที่ซับซ้อน บทความนี้จะพาคุณสร้างระบบ Backtest ที่ครบวงจร โดยใช้ Tardis API สำหรับดึงข้อมูล และ Parquet สำหรับจัดเก็บ Local พร้อมเปรียบเทียบว่าทำไมการใช้ HolySheep AI สำหรับ Signal Generation ถึงประหยัดกว่า 85%
สถาปัตยกรรมระบบโดยรวม
ระบบที่เราจะสร้างประกอบด้วย 3 ส่วนหลัก:
- Data Layer: Tardis API สำหรับดึง Tick Data จาก OKX
- Storage Layer: Parquet Files บน Local Storage พร้อม Schema แบบ Optimized
- Computation Layer: Python Workers สำหรับ Backtest Engine และ AI Signal Generation
การติดตั้งและ Setup Environment
เริ่มต้นด้วยการติดตั้ง Dependencies ที่จำเป็นทั้งหมด:
# requirements.txt
tardis-dev==2.1.0
pandas==2.2.0
pyarrow==15.0.0
numpy==1.26.3
asyncio==3.4.3
httpx==0.27.0
openpyxl==3.1.2 # สำหรับ Excel export
AI Integration
anthropic==0.21.0 # สำหรับ Claude API
openai==1.12.0 # สำหรับ GPT API
pip install -r requirements.txt
สร้าง directory structure
mkdir -p data/tick_data data/parquet logs models configs
ตอนที่ 1: Tardis API Integration สำหรับ OKX Tick Data
Tardis เป็นบริการที่รวบรวม Historical Data จาก Exchange หลายตัว รวมถึง OKX โดยให้ API ที่รองรับการดึงข้อมูลระดับ Tick ตามเวลาจริง ค่าใช้จ่ายเริ่มต้นอยู่ที่ $0.0008 ต่อ 1,000 messages ซึ่งถ้าคุณทำ Backtest หลายสิบครั้งต่อวัน ค่าใช้จ่ายจะพุ่งสูงอย่างรวดเร็ว
Configuration และ Environment Setup
import os
from dataclasses import dataclass
from typing import Optional
import asyncio
from datetime import datetime, timedelta
@dataclass
class OKXTardisConfig:
"""Configuration สำหรับ OKX Tardis API"""
api_key: str = os.getenv("TARDIS_API_KEY", "")
exchange: str = "okx"
channels: list = None
# การตั้งค่า Performance
batch_size: int = 10000
max_concurrent_requests: int = 5
request_timeout: int = 30
# การตั้งค่า Storage
output_dir: str = "./data/tick_data"
def __post_init__(self):
if self.channels is None:
self.channels = ["trades", "books"]
os.makedirs(self.output_dir, exist_ok=True)
การตั้งค่า Default
config = OKXTardisConfig(
api_key="your_tardis_api_key_here",
batch_size=10000,
max_concurrent_requests=5
)
Core Data Fetcher ด้วย Asyncio
นี่คือหัวใจของระบบ — Class ที่จัดการการดึงข้อมูลจาก Tardis API แบบ Asynchronous พร้อมรองรับ Concurrent Requests สำหรับการ Optimize เวลา:
import json
import aiohttp
import asyncio
from pathlib import Path
from typing import AsyncGenerator, Dict, Any
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class OKXTardisFetcher:
"""
Async Fetcher สำหรับ OKX Tick Data จาก Tardis API
รองรับ concurrent requests และ automatic retry
"""
BASE_URL = "https://api.tardis.dev/v1"
def __init__(self, config: OKXTardisConfig):
self.config = config
self.session: Optional[aiohttp.ClientSession] = None
self.semaphore = asyncio.Semaphore(config.max_concurrent_requests)
async def __aenter__(self):
timeout = aiohttp.ClientTimeout(total=self.config.request_timeout)
connector = aiohttp.TCPConnector(limit=100, limit_per_host=50)
self.session = aiohttp.ClientSession(
timeout=timeout,
connector=connector
)
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
async def fetch_trades(
self,
symbol: str,
start_date: datetime,
end_date: datetime,
symbol_suffix: str = "-SWAP"
) -> AsyncGenerator[Dict[str, Any], None]:
"""
ดึงข้อมูล Trade ตามช่วงเวลาที่กำหนด
Args:
symbol: เช่น "BTC-USDT-SWAP"
start_date: วันเริ่มต้น
end_date: วันสิ้นสุด
symbol_suffix: suffix ของ symbol (default: "-SWAP")
"""
full_symbol = f"{symbol}{symbol_suffix}"
current_date = start_date
while current_date <= end_date:
async with self.semaphore:
next_date = min(
current_date + timedelta(days=1),
end_date
)
url = (
f"{self.BASE_URL}/historical/okx/trades"
f"?symbol={full_symbol}"
f"&from={current_date.isoformat()}"
f"&to={next_date.isoformat()}"
f"&format=json"
f"&limit={self.config.batch_size}"
)
headers = {
"Authorization": f"Bearer {self.config.api_key}",
"Accept": "application/x-json-stream"
}
retry_count = 0
max_retries = 3
while retry_count < max_retries:
try:
async with self.session.get(
url,
headers=headers
) as response:
if response.status == 200:
async for line in response.content:
if line.strip():
try:
data = json.loads(line)
yield data
except json.JSONDecodeError:
continue
break
elif response.status == 429:
# Rate limited - wait and retry
wait_time = 2 ** retry_count
logger.warning(
f"Rate limited, waiting {wait_time}s"
)
await asyncio.sleep(wait_time)
retry_count += 1
else:
logger.error(
f"API Error {response.status}: "
f"{await response.text()}"
)
break
except aiohttp.ClientError as e:
logger.error(f"Connection error: {e}")
await asyncio.sleep(2 ** retry_count)
retry_count += 1
current_date = next_date
async def get_trades_count(
self,
symbol: str,
start_date: datetime,
end_date: datetime
) -> int:
"""นับจำนวน trades ทั้งหมด (สำหรับ estimate cost)"""
count = 0
async for _ in self.fetch_trades(symbol, start_date, end_date):
count += 1
return count
ตัวอย่างการใช้งาน
async def main():
config = OKXTardisConfig(api_key="your_tardis_key")
async with OKXTardisFetcher(config) as fetcher:
# ดึงข้อมูล BTC-USDT-SWAP 1 สัปดาห์
start = datetime(2024, 12, 1)
end = datetime(2024, 12, 7)
count = 0
async for trade in fetcher.fetch_trades(
"BTC-USDT",
start,
end,
symbol_suffix="-SWAP"
):
count += 1
if count % 100000 == 0:
logger.info(f"Fetched {count:,} trades")
logger.info(f"Total: {count:,} trades")
รัน: asyncio.run(main())
ตอนที่ 2: Parquet Storage และ Schema Optimization
หลังจากดึงข้อมูลมาแล้ว สิ่งสำคัญคือการจัดเก็บให้เหมาะกับการ Query ในภายหลัง Parquet เป็น Format ที่เหมาะมากเพราะรองรับ Columnar Storage และ Compression ในตัว ช่วยประหยัดพื้นที่ได้ถึง 70% เมื่อเทียบกับ CSV
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from pathlib import Path
from typing import Optional
from dataclasses import dataclass
import numpy as np
@dataclass
class TradeSchema:
"""
Optimized Schema สำหรับ Tick Data
ออกแบบให้ Query ได้เร็วด้วย Dictionary Encoding
"""
# Schema Definition
SCHEMA = pa.schema([
pa.field('timestamp', pa.timestamp('ms'), nullable=False),
pa.field('symbol', pa.string(), nullable=False),
pa.field('side', pa.dictionary(pa.string(), pa.int16()), nullable=False),
pa.field('price', pa.decimal128(18, 8), nullable=False),
pa.field('size', pa.decimal128(18, 8), nullable=False),
pa.field('trade_id', pa.string(), nullable=False),
pa.field('is_auction', pa.bool_(), nullable=True),
pa.field('mark_price', pa.decimal128(18, 8), nullable=True),
pa.field('underlying_index_price', pa.decimal128(18, 8), nullable=True),
])
# Partition columns
PARTITION_COLS = ['year', 'month', 'day']
@staticmethod
def get_parquet_options() -> pq.ParquetWriter:
"""Compression options สำหรับ Parquet"""
return pq.ParquetFile.write_options(
compression='snappy', # เร็ว, กิน CPU ต่ำ
use_dictionary=True,
write_statistics=True,
data_page_size=1024 * 1024 # 1MB pages
)
class TickDataParquetWriter:
"""
Writer สำหรับแปลง Raw Tick Data → Optimized Parquet
รองรับ Streaming และ Batch Processing
"""
def __init__(
self,
output_dir: str = "./data/parquet",
buffer_size: int = 100_000
):
self.output_dir = Path(output_dir)
self.buffer_size = buffer_size
self.buffer: list = []
self.symbol_stats: Dict[str, dict] = {}
def _normalize_trade(self, trade: Dict) -> Dict:
"""Normalize trade data ให้ตรงกับ Schema"""
return {
'timestamp': pd.to_datetime(
trade.get('timestamp', trade.get('localTimestamp')),
unit='ms',
utc=True
),
'symbol': trade['symbol'],
'side': trade['side'].upper(),
'price': float(trade['price']),
'size': float(trade['size']),
'trade_id': str(trade.get('id', trade.get('tradeId', ''))),
'is_auction': trade.get('isAuction', False),
'mark_price': float(trade.get('markPrice')) if trade.get('markPrice') else None,
'underlying_index_price': (
float(trade.get('underlyingIndexPrice'))
if trade.get('underlyingIndexPrice')
else None
),
# Partition columns
'year': pd.to_datetime(
trade.get('timestamp', trade.get('localTimestamp')),
unit='ms',
utc=True
).year,
'month': pd.to_datetime(
trade.get('timestamp', trade.get('localTimestamp')),
unit='ms',
utc=True
).month,
'day': pd.to_datetime(
trade.get('timestamp', trade.get('localTimestamp')),
unit='ms',
utc=True
).day,
}
def add_trade(self, trade: Dict):
"""เพิ่ม trade เข้า buffer"""
normalized = self._normalize_trade(trade)
self.buffer.append(normalized)
# Update statistics
symbol = normalized['symbol']
if symbol not in self.symbol_stats:
self.symbol_stats[symbol] = {
'count': 0,
'min_time': normalized['timestamp'],
'max_time': normalized['timestamp']
}
self.symbol_stats[symbol]['count'] += 1
self.symbol_stats[symbol]['max_time'] = normalized['timestamp']
# Auto-flush เมื่อ buffer เต็ม
if len(self.buffer) >= self.buffer_size:
self.flush()
def flush(self):
"""เขียน buffer ลง Parquet file"""
if not self.buffer:
return
df = pd.DataFrame(self.buffer)
# Extract symbol สำหรับ filename
first_symbol = df['symbol'].iloc[0]
# สร้าง filename ตาม date range
min_date = df['timestamp'].min()
max_date = df['timestamp'].max()
filename = (
f"trades_{first_symbol.replace('/', '-')}_"
f"{min_date.strftime('%Y%m%d_%H%M%S')}_"
f"{max_date.strftime('%Y%m%d_%H%M%S')}.parquet"
)
filepath = self.output_dir / filename
# Convert to PyArrow Table
table = pa.Table.from_pandas(
df,
schema=TradeSchema.SCHEMA,
preserve_index=False
)
# Write with partitioning
pq.write_table(
table,
filepath,
compression='snappy',
use_dictionary=True,
write_statistics=True
)
print(f"✓ Written {len(self.buffer):,} trades → {filepath}")
print(f" File size: {filepath.stat().st_size / 1024 / 1024:.2f} MB")
# Clear buffer
self.buffer = []
def finalize(self):
"""Flush remaining และสร้าง manifest"""
self.flush()
# สร้าง Statistics manifest
manifest_path = self.output_dir / "manifest.json"
import json
with open(manifest_path, 'w') as f:
json.dump({
'symbols': self.symbol_stats,
'created_at': pd.Timestamp.now().isoformat()
}, f, indent=2, default=str)
print(f"✓ Manifest created: {manifest_path}")
return self.symbol_stats
ตอนที่ 3: Backtest Engine พร้อม AI Signal Generation
สำหรับการทำ Backtest จริง ผมแนะนำให้ใช้ HolySheep AI สำหรับ Signal Generation เพราะค่าใช้จ่ายถูกกว่า API อื่นๆ มาก และ Latency ต่ำกว่า 50ms ทำให้เหมาะกับงานที่ต้องประมวลผลจำนวนมาก
import httpx
import json
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
import asyncio
import pandas as pd
@dataclass
class BacktestSignal:
"""Signal ที่ได้จาก AI"""
timestamp: pd.Timestamp
symbol: str
action: str # 'BUY' | 'SELL' | 'HOLD'
confidence: float
reasoning: str
price: float
class HolySheepSignalGenerator:
"""
AI Signal Generator โดยใช้ HolySheep API
ราคาถูกกว่า 85% เมื่อเทียบกับ OpenAI/Claude
Pricing Reference (2026):
- GPT-4.1: $8/MTok
- Claude Sonnet 4.5: $15/MTok
- Gemini 2.5 Flash: $2.50/MTok
- DeepSeek V3.2: $0.42/MTok ← HolySheep's best value
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str, model: str = "deepseek-v3.2"):
self.api_key = api_key
self.model = model
self.client = httpx.Client(timeout=30.0)
def generate_signals(
self,
market_data: List[Dict[str, Any]],
context_window: int = 50
) -> List[BacktestSignal]:
"""
Generate trading signals จาก Market Data
Args:
market_data: List ของ tick data (price, volume, etc.)
context_window: จำนวน candles ที่ให้ AI วิเคราะห์
"""
# Prepare context
recent_data = market_data[-context_window:]
# Build prompt
system_prompt = """คุณเป็นนักวิเคราะห์ Quantitative Trading
วิเคราะห์ข้อมูลตลาดและให้สัญญาณซื้อขาย
ตอบกลับเฉพาะ JSON format:
{
"action": "BUY|SELL|HOLD",
"confidence": 0.0-1.0,
"reasoning": "เหตุผลสั้นๆ"
}"""
user_prompt = f"""ข้อมูลตลาดล่าสุด:
{json.dumps(recent_data[:10], indent=2)}
สถิติราคา:
- ราคาปัจจุบัน: {recent_data[-1].get('price')}
- สูงสุด: {max(d.get('price', 0) for d in recent_data)}
- ต่ำสุด: {min(d.get('price', 0) for d in recent_data)}
- Volume รวม: {sum(d.get('size', 0) for d in recent_data)}
ให้สัญญาณซื้อขาย:"""
# Call HolySheep API
response = self.client.post(
f"{self.BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": self.model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
"temperature": 0.3,
"max_tokens": 200
}
)
if response.status_code != 200:
raise Exception(f"API Error: {response.status_code} - {response.text}")
result = response.json()
content = result['choices'][0]['message']['content']
# Parse response
try:
signal_data = json.loads(content)
return BacktestSignal(
timestamp=pd.Timestamp.now(),
symbol=recent_data[-1].get('symbol', 'UNKNOWN'),
action=signal_data['action'],
confidence=signal_data['confidence'],
reasoning=signal_data['reasoning'],
price=recent_data[-1].get('price', 0)
)
except json.JSONDecodeError:
# Fallback to HOLD if parsing fails
return BacktestSignal(
timestamp=pd.Timestamp.now(),
symbol=recent_data[-1].get('symbol', 'UNKNOWN'),
action='HOLD',
confidence=0.5,
reasoning='Parse error - default to HOLD',
price=recent_data[-1].get('price', 0)
)
def batch_generate(
self,
market_data_chunks: List[List[Dict]],
context_window: int = 50
) -> List[List[BacktestSignal]]:
"""
Batch generate signals สำหรับหลาย time periods
ใช้ Async เพื่อความเร็ว
"""
async def async_batch():
async with httpx.AsyncClient(timeout=60.0) as client:
tasks = []
for chunk in market_data_chunks:
task = self._async_generate(
client, chunk, context_window
)
tasks.append(task)
return await asyncio.gather(*tasks)
return asyncio.run(async_batch())
async def _async_generate(
self,
client: httpx.AsyncClient,
market_data: List[Dict],
context_window: int
) -> List[BacktestSignal]:
recent_data = market_data[-context_window:]
# ... (similar logic to generate_signals but async)
# สำหรับบทความนี้ขอย่อเหลือส่วนนี้
return [] # Placeholder
def close(self):
self.client.close()
ตัวอย่างการใช้งาน
def run_backtest_with_ai():
# Initialize
api_key = "YOUR_HOLYSHEEP_API_KEY" # รับจาก https://www.holysheep.ai/register
generator = HolySheepSignalGenerator(api_key, model="deepseek-v3.2")
# Mock market data (แทนที่ด้วยข้อมูลจริงจาก Parquet)
sample_data = [
{'timestamp': '2024-12-01T10:00:00', 'price': 97000, 'size': 0.5, 'symbol': 'BTC-USDT-SWAP'},
{'timestamp': '2024-12-01T10:00:01', 'price': 97050, 'size': 0.3, 'symbol': 'BTC-USDT-SWAP'},
{'timestamp': '2024-12-01T10:00:02', 'price': 97100, 'size': 0.8, 'symbol': 'BTC-USDT-SWAP'},
# ... เพิ่มข้อมูลจริงต่อไป
] * 20
# Generate signal
signal = generator.generate_signals(sample_data, context_window=50)
print(f"Signal: {signal.action}")
print(f"Confidence: {signal.confidence:.2%}")
print(f"Reasoning: {signal.reasoning}")
generator.close()
ตอนที่ 4: Data Pipeline Orchestration
Pipeline เต็มรูปแบบที่รวมทุกอย่างเข้าด้วยกัน พร้อม Progress Tracking และ Error Handling:
import asyncio
from pathlib import Path
from datetime import datetime, timedelta
from typing import Optional
import logging
from concurrent.futures import ProcessPoolExecutor
import multiprocessing as mp
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class OKXBacktestPipeline:
"""
Complete Pipeline สำหรับ OKX Tick Data Backtest
1. Fetch from Tardis API
2. Store as Parquet
3. Run Backtest with AI Signals
"""
def __init__(
self,
tardis_key: str,
holysheep_key: str,
data_dir: str = "./data",
workers: int = None
):
self.tardis_config = OKXTardisConfig(
api_key=tardis_key,
output_dir=f"{data_dir}/tick_data"
)
self.holysheep = HolySheepSignalGenerator(holysheep_key)
self.writer = TickDataParquetWriter(
output_dir=f"{data_dir}/parquet",
buffer_size=100_000
)
self.workers = workers or max(1, mp.cpu_count() - 1)
async def fetch_and_store(
self,
symbol: str,
start_date: datetime,
end_date: datetime,
progress_callback: Optional[callable] = None
) -> dict:
"""
Fetch data จาก Tardis และ Store ลง Parquet
"""
logger.info(f"Starting fetch for {symbol} from {start_date} to {end_date}")
async with OKXTardisFetcher(self.tardis_config) as fetcher:
trade_count = 0
async for trade in fetcher.fetch_trades(
symbol, start_date, end_date
):
self.writer.add_trade(trade)
trade_count += 1
if progress_callback and trade_count % 50000 == 0:
progress_callback(trade_count)
stats = self.writer.finalize()
logger.info(f"Completed: {trade_count:,} trades fetched")
return stats
def run_backtest(
self,
symbol: str,
start_date: datetime,
end_date: datetime,
initial_capital: float = 100_000,
commission: float = 0.0004
) -> dict:
"""
Run full backtest พร้อม AI signal generation
"""
parquet_dir = Path(f"./data/parquet")
# Load Parquet files for the date range
pf = pq.ParquetDataset(parquet_dir)
table = pf.read()
df = table.to_pandas()
# Filter by symbol and date
df = df[df['symbol'].str.contains(symbol)]
df = df[
(df['timestamp'] >= start_date) &
(df['timestamp'] <= end_date)
].sort_values('timestamp')
logger.info(f"Loaded {len(df):,} trades for backtest")
# Resample to candles for AI analysis
df['price'] = df['price'].astype(float)
df.set_index('timestamp', inplace=True)
# Backtest logic here...
# (Simplified for article brevity)
return {
'total_trades': len(df),
'symbol': symbol,
'start_date': start_date,
'end_date': end_date,
'initial_capital': initial_capital,
'final_capital': initial_capital, # Calculate actual
'return': 0.0 # Calculate actual
}
def estimate_cost(self, start_date: datetime, end_date: datetime) -> dict:
"""
Estimate API cost ก่อน fetch จริง
"""
days = (end_date - start_date).days + 1
# Tardis pricing: ~$0.0008 per 1000 messages
# OKX BTC-SWAP: ~500,000 trades/day
estimated_trades = days * 500_000
tardis_cost = (estimated_trades / 1000) * 0.0008
# HolySheep AI (DeepSeek V3.2)
# ~100 tokens per signal
ai_calls = estimated_trades // 50 # 1 signal per 50 trades
ai_cost = (ai_calls * 100 / 1_000_000) * 0.42
return {
'estimated_trades': estimated_trades,
'tardis_cost_usd': tardis_cost,
'ai_cost_usd': ai_cost,
'total_cost_usd': tardis_cost + ai_cost,
'cost_per_million_trades': (
(tardis_cost + ai_cost) /