การทำ Backtest ระบบเทรดคริปโตที่แม่นยำต้องอาศัยข้อมูล Tick-level คุณภาพสูง ในบทความนี้ผมจะแชร์วิธีการดาวน์โหลดและแคชข้อมูล OKX Perpetual Futures ผ่าน Tardis API พร้อมโค้ดที่พร้อมใช้งานจริง ตั้งแต่ Setup ไปจนถึงการ Optimize สำหรับงบประมาณที่จำกัด
ทำไมต้องเป็นข้อมูล Tick สำหรับ Backtest
ข้อมูล OHLCV ธรรมดาไม่เพียงพอสำหรับการ Backtest ที่เชื่อถือได้ โดยเฉพาะอย่างยิ่ง:
- สถานการณ์ Slippage จริง — ราคาเปิด-ปิดของแท่งเทียนอาจไม่ตรงกับราคาที่ Order จริงถูก Fill
- Volume Profile — ต้องรู้ว่า Volume กระจายตัวอย่างไรในแต่ละระดับราคา
- Order Book Dynamics — สำหรับ Strategy ที่อาศัย Liquidity เช่น Market Making
Tardis API ให้ข้อมูล Tick-by-Tick ของ Exchange หลายร้อยแห่ง รวมถึง OKX Perpetual Futures ซึ่งเป็นหนึ่งใน Exchange ที่มี Volume สูงที่สุดในโลก
การติดตั้งและ Setup สภาพแวดล้อม
ก่อนเริ่มต้น ติดตั้ง Library ที่จำเป็น:
# ติดตั้ง dependencies
pip install tardis-client pandas pyarrow aiohttp asyncio aiofiles
สำหรับ Project ที่ใช้ Poetry
poetry add tardis-client pandas pyarrow aiohttp aiofilesโครงสร้างโปรเจกต์สำหรับ Local Caching
project/
├── config/
│ ├── __init__.py
│ └── settings.py
├── data/
│ ├── raw/
│ ├── processed/
│ └── cache/
├── src/
│ ├── __init__.py
│ ├── downloader.py
│ ├── cache_manager.py
│ └── data_transformer.py
├── tests/
│ └── test_downloader.py
├── pyproject.toml
└── requirements.txtการสร้าง Configuration Manager
# config/settings.py
from dataclasses import dataclass
from pathlib import Path
import os
@dataclass
class TardisConfig:
api_key: str = os.getenv("TARDIS_API_KEY", "")
base_url: str = "https://api.tardis.dev/v1"
exchange: str = "okx"
market_type: str = "perpetual_future"
@dataclass
class DataConfig:
project_root: Path = Path(__file__).parent.parent
raw_dir: Path = Path(__file__).parent.parent / "data" / "raw"
cache_dir: Path = Path(__file__).parent.parent / "data" / "cache"
def __post_init__(self):
self.raw_dir.mkdir(parents=True, exist_ok=True)
self.cache_dir.mkdir(parents=True, exist_ok=True)
@dataclass
class HolySheepConfig:
"""Configuration สำหรับ AI API — ประหยัด 85%+ เมื่อเทียบกับ OpenAI"""
base_url: str = "https://api.holysheep.ai/v1"
api_key: str = os.getenv("HOLYSHEEP_API_KEY", "")
# ราคา 2026 ต่อ MTok (USD)
gpt41_price: float = 8.0
claude_sonnet45_price: float = 15.0
gemini_flash25_price: float = 2.50
deepseek_v32_price: float = 0.42 # ราคาถูกที่สุด!
# Models ที่แนะนำสำหรับ Quant Tasks
recommended_models: list = None
def __post_init__(self):
self.recommended_models = [
{"name": "DeepSeek V3.2", "price": self.deepseek_v32_price, "use_case": "Backtest Analysis"},
{"name": "Gemini 2.5 Flash", "price": self.gemini_flash25_price, "use_case": "Data Processing"},
{"name": "GPT-4.1", "price": self.gpt41_price, "use_case": "Strategy Development"},
]
Global instances
tardis_config = TardisConfig()
data_config = DataConfig()
holysheep_config = HolySheepConfig()Downloader หลักพร้อม Asyncio และ Progress Tracking
# src/downloader.py
import asyncio
import aiohttp
import aiofiles
from pathlib import Path
from datetime import datetime, timedelta
from typing import List, Optional
import json
import gzip
from dataclasses import dataclass
from tqdm.asyncio import tqdm
import hashlib
from config.settings import tardis_config, data_config
@dataclass
class DownloadTask:
exchange: str
market: str
date_from: str # YYYY-MM-DD
date_to: str # YYYY-MM-DD
channels: List[str]
class TardisDownloader:
"""
ดาวน์โหลดข้อมูล Tick จาก Tardis API
รองรับ Parallel Downloads และ Auto-Retry
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = tardis_config.base_url
self.session: Optional[aiohttp.ClientSession] = None
self.rate_limit = 10 # requests per second
self.retry_attempts = 3
async def __aenter__(self):
timeout = aiohttp.ClientTimeout(total=300)
self.session = aiohttp.ClientSession(timeout=timeout)
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
def _get_cache_key(self, task: DownloadTask) -> str:
"""สร้าง unique cache key จาก task parameters"""
key_str = f"{task.exchange}_{task.market}_{task.date_from}_{task.date_to}_{'-'.join(task.channels)}"
return hashlib.md5(key_str.encode()).hexdigest()
async def _download_single_day(
self,
task: DownloadTask,
date: str,
progress_callback=None
) -> dict:
"""ดาวน์โหลดข้อมูลของวันเดียว"""
# Construct API URL
url = (
f"{self.base_url}/feeds/{task.exchange}:{task.market}"
f"/historical"
f"?from={date}&to={date}"
f"&channels={','.join(task.channels)}"
f"&format=json"
)
headers = {"Authorization": f"Bearer {self.api_key}"}
for attempt in range(self.retry_attempts):
try:
async with self.session.get(url, headers=headers) as response:
if response.status == 200:
data = await response.json()
# Cache compressed data
cache_file = (
data_config.cache_dir /
f"{task.exchange}_{task.market}_{date}.json.gz"
)
async with aiofiles.open(cache_file, 'wb') as f:
content = json.dumps(data).encode('utf-8')
compressed = gzip.compress(content)
await f.write(compressed)
if progress_callback:
await progress_callback()
return {"status": "success", "date": date, "records": len(data)}
elif response.status == 429:
# Rate limited — wait and retry
wait_time = 2 ** attempt
await asyncio.sleep(wait_time)
continue
else:
return {
"status": "error",
"date": date,
"error": f"HTTP {response.status}"
}
except Exception as e:
if attempt == self.retry_attempts - 1:
return {"status": "error", "date": date, "error": str(e)}
await asyncio.sleep(1)
return {"status": "error", "date": date, "error": "Max retries exceeded"}
async def download(
self,
task: DownloadTask,
parallel: bool = True,
max_concurrent: int = 5
) -> dict:
"""ดาวน์โหลดข้อมูลหลายวันพร้อมกัน"""
# Generate date range
start = datetime.strptime(task.date_from, "%Y-%m-%d")
end = datetime.strptime(task.date_to, "%Y-%m-%d")
dates = []
current = start
while current <= end:
dates.append(current.strftime("%Y-%m-%d"))
current += timedelta(days=1)
print(f"📥 เริ่มดาวน์โหลด {len(dates)} วัน ข้อมูล {task.exchange}:{task.market}")
if parallel:
# Semaphore เพื่อจำกัด concurrent requests
semaphore = asyncio.Semaphore(max_concurrent)
async def download_with_semaphore(date):
async with semaphore:
return await self._download_single_day(task, date)
# ใช้ tqdm เพื่อแสดง progress
tasks = [download_with_semaphore(date) for date in dates]
results = await tqdm.gather(*tasks, desc="ดาวน์โหลด")
else:
results = []
for date in tqdm(dates, desc="ดาวน์โหลด"):
result = await self._download_single_day(task, date)
results.append(result)
# Summary
success = sum(1 for r in results if r["status"] == "success")
failed = len(results) - success
summary = {
"total": len(results),
"success": success,
"failed": failed,
"results": results
}
print(f"\n✅ ดาวน์โหลดเสร็จสิ้น: {success}/{len(results)} วัน")
if failed > 0:
print(f"⚠️ ไม่สำเร็จ: {failed} วัน")
return summary
ตัวอย่างการใช้งาน
async def main():
async with TardisDownloader(api_key="YOUR_TARDIS_API_KEY") as downloader:
task = DownloadTask(
exchange="okx",
market="BTC-USDT-SWAP",
date_from="2026-01-01",
date_to="2026-01-31",
channels=["trades", "book25"]
)
result = await downloader.download(task, parallel=True)
# Save summary
summary_file = data_config.cache_dir / "download_summary.json"
async with aiofiles.open(summary_file, 'w') as f:
await f.write(json.dumps(result, indent=2))
return result
if __name__ == "__main__":
asyncio.run(main())Cache Manager สำหรับ Query ข้อมูลที่รวดเร็ว
# src/cache_manager.py
import gzip
import json
import aiofiles
import pandas as pd
from pathlib import Path
from datetime import datetime, timedelta
from typing import List, Optional, Dict
from dataclasses import dataclass
import hashlib
@dataclass
class CacheConfig:
cache_dir: Path = Path("data/cache")
compression: bool = True
max_memory_mb: int = 1024 # แคชใน memory ไม่เกิน 1GB
class TickDataCache:
"""
ระบบ Cache อัจฉริยะสำหรับข้อมูล Tick
- เก็บข้อมูลเป็น Parquet สำหรับ Query เร็ว
- Support Date Range Queries
- Auto-cleanup เก่าๆ
"""
def __init__(self, config: CacheConfig = None):
self.config = config or CacheConfig()
self.cache_dir = self.config.cache_dir
self.cache_dir.mkdir(parents=True, exist_ok=True)
# In-memory index สำหรับความเร็ว
self._index: Dict[str, dict] = {}
self._load_index()
def _load_index(self):
"""โหลด index จากไฟล์"""
index_file = self.cache_dir / "cache_index.json"
if index_file.exists():
with open(index_file, 'r') as f:
self._index = json.load(f)
def _save_index(self):
"""บันทึก index"""
index_file = self.cache_dir / "cache_index.json"
with open(index_file, 'w') as f:
json.dump(self._index, f, indent=2)
async def load_raw_data(
self,
exchange: str,
market: str,
date: str
) -> List[dict]:
"""โหลดข้อมูลดิบจาก cache"""
cache_file = self.cache_dir / f"{exchange}_{market}_{date}.json.gz"
if not cache_file.exists():
return []
async with aiofiles.open(cache_file, 'rb') as f:
compressed = await f.read()
data = gzip.decompress(compressed)
return json.loads(data)
async def query_range(
self,
exchange: str,
market: str,
date_from: str,
date_to: str,
channels: List[str] = None,
limit: int = None
) -> pd.DataFrame:
"""
Query ข้อมูลในช่วงวันที่กำหนด
รวดเร็วกว่าการโหลดทีละวันมาก
"""
start = datetime.strptime(date_from, "%Y-%m-%d")
end = datetime.strptime(date_to, "%Y-%m-%d")
all_data = []
current = start
while current <= end:
date_str = current.strftime("%Y-%m-%d")
data = await self.load_raw_data(exchange, market, date_str)
if channels:
data = [
d for d in data
if d.get("channel") in channels
]
all_data.extend(data)
current += timedelta(days=1)
if limit and len(all_data) >= limit:
all_data = all_data[:limit]
break
df = pd.DataFrame(all_data)
if not df.empty and "timestamp" in df.columns:
df["datetime"] = pd.to_datetime(df["timestamp"], unit="ms")
df = df.sort_values("datetime")
return df
def get_cache_stats(self) -> dict:
"""ดูสถิติของ cache"""
total_size = sum(
f.stat().st_size
for f in self.cache_dir.glob("*.json.gz")
)
return {
"total_files": len(list(self.cache_dir.glob("*.json.gz"))),
"total_size_mb": round(total_size / (1024 * 1024), 2),
"indexed_symbols": len(self._index),
}
ตัวอย่างการใช้งาน
async def demo_query():
cache = TickDataCache()
# Query ข้อมูล 1 เดือน
df = await cache.query_range(
exchange="okx",
market="BTC-USDT-SWAP",
date_from="2026-01-01",
date_to="2026-01-31",
channels=["trades"]
)
print(f"📊 โหลดข้อมูล {len(df)} records")
print(f"⏰ ช่วงเวลา: {df['datetime'].min()} ถึง {df['datetime'].max()}")
print(f"💰 ราคาเฉลี่ย: ${df['price'].mean():.2f}")
return dfBacktest Framework Integration
หลังจากมีข้อมูลแล้ว มาดูวิธีการนำไปใช้กับ Backtest Framework ยอดนิยม:
# src/backtest_runner.py
import pandas as pd
import numpy as np
from datetime import datetime
from typing import Callable, Dict, List
from dataclasses import dataclass
@dataclass
class BacktestResult:
total_trades: int
win_rate: float
profit_factor: float
max_drawdown: float
sharpe_ratio: float
trades: pd.DataFrame
class SimpleBacktester:
"""
Backtester แบบง่ายสำหรับ Tick Data
รองรับ: Mean Reversion, Momentum, Breakout Strategies
"""
def __init__(
self,
data: pd.DataFrame,
initial_capital: float = 10000,
commission: float = 0.0004 # 0.04% per trade
):
self.data = data.copy()
self.initial_capital = initial_capital
self.commission = commission
# เพิ่ม columns ที่จำเป็น
if 'datetime' in self.data.columns:
self.data.set_index('datetime', inplace=True)
def add_signals(self, strategy_func: Callable):
"""เพิ่ม Signals จาก Strategy Function"""
self.data['signal'] = strategy_func(self.data)
return self
def run(self) -> BacktestResult:
"""รัน Backtest"""
df = self.data.copy()
# คำนวณ Position
df['position'] = df['signal'].shift(1).fillna(0)
df['returns'] = df['price'].pct_change()
# คำนวณ PnL
df['strategy_returns'] = df['position'] * df['returns']
df['strategy_returns'] = df['strategy_returns'] - abs(df['position'].diff()) * self.commission
# คำนวณ Equity Curve
df['equity'] = self.initial_capital * (1 + df['strategy_returns']).cumprod()
# คำนวณ Drawdown
df['peak'] = df['equity'].cummax()
df['drawdown'] = (df['equity'] - df['peak']) / df['peak']
# หา Trades
trades = self._extract_trades(df)
# คำนวณ Metrics
total_return = (df['equity'].iloc[-1] / self.initial_capital - 1) * 100
if len(trades) > 0:
win_rate = (trades['pnl'] > 0).sum() / len(trades)
gross_profit = trades.loc[trades['pnl'] > 0, 'pnl'].sum()
gross_loss = abs(trades.loc[trades['pnl'] <= 0, 'pnl'].sum())
profit_factor = gross_profit / gross_loss if gross_loss > 0 else np.inf
else:
win_rate = 0
profit_factor = 0
max_drawdown = df['drawdown'].min() * 100
# Sharpe Ratio (annualized)
if df['strategy_returns'].std() > 0:
sharpe = (df['strategy_returns'].mean() / df['strategy_returns'].std()) * np.sqrt(365 * 24)
else:
sharpe = 0
return BacktestResult(
total_trades=len(trades),
win_rate=win_rate * 100,
profit_factor=profit_factor,
max_drawdown=max_drawdown,
sharpe_ratio=sharpe,
trades=trades
)
def _extract_trades(self, df: pd.DataFrame) -> pd.DataFrame:
"""แยก Trades จาก Position changes"""
df['position_changed'] = df['position'].diff().fillna(0) != 0
change_points = df[df['position_changed']].index
trades = []
entry_price = None
entry_time = None
entry_position = 0
for i, time in enumerate(change_points):
position = df.loc[time, 'position']
price = df.loc[time, 'price']
if entry_position == 0 and position != 0:
# เปิด Position
entry_price = price
entry_time = time
entry_position = position
elif entry_position != 0 and position == 0:
# ปิด Position
pnl = (price - entry_price) * entry_position
trades.append({
'entry_time': entry_time,
'exit_time': time,
'entry_price': entry_price,
'exit_price': price,
'position': entry_position,
'pnl': pnl,
'duration': (time - entry_time).total_seconds() / 3600 # hours
})
entry_position = 0
return pd.DataFrame(trades)
ตัวอย่าง Strategy
def momentum_strategy(data: pd.DataFrame, window: int = 20) -> pd.Series:
"""Mean Reversion Strategy อย่างง่าย"""
ma = data['price'].rolling(window=window).mean()
std = data['price'].rolling(window=window).std()
# Z-Score
z_score = (data['price'] - ma) / std
# Signals: 1 = Long, -1 = Short, 0 = Flat
signals = pd.Series(0, index=data.index)
signals[z_score < -2] = 1 # เกินขาย → Long
signals[z_score > 2] = -1 # เกินซื้อ → Short
signals[(z_score > -0.5) & (z_score < 0.5)] = 0 # Neutral
return signals
การใช้งาน
async def run_backtest():
from src.cache_manager import TickDataCache
cache = TickDataCache()
df = await cache.query_range(
exchange="okx",
market="BTC-USDT-SWAP",
date_from="2026-01-01",
date_to="2026-01-31",
channels=["trades"]
)
# Convert trades data to price series
price_data = df[df['channel'] == 'trades'][['datetime', 'price']].copy()
price_data = price_data.set_index('datetime')
# Run backtest
bt = SimpleBacktester(price_data, initial_capital=10000)
bt.add_signals(momentum_strategy)
result = bt.run()
print(f"📈 Total Trades: {result.total_trades}")
print(f"🎯 Win Rate: {result.win_rate:.2f}%")
print(f"💰 Profit Factor: {result.profit_factor:.2f}")
print(f"📉 Max Drawdown: {result.max_drawdown:.2f}%")
print(f"📊 Sharpe Ratio: {result.sharpe_ratio:.2f}")
return resultเหมาะกับใคร / ไม่เหมาะกับใคร
| กลุ่มผู้ใช้ | เหมาะกับ | ไม่เหมาะกับ |
|---|---|---|
| นักเทรดรายบุคคล | Backtest ด้วยงบประมาณจำกัด ต้องการข้อมูลคุณภาพสูง | ต้องการ Real-time Data Feed |
| Quants / Data Scientists | พัฒนา Strategy ที่ซับซ้อน ต้องการ Tick-level Data | ต้องการ Built-in ML Models |
| Hedge Funds ขนาดเล็ก | ทดสอบ Hypothesis ก่อน Live Trading | ต้องการ Compliance และ Audit Trail |
| บริษัท AI / SaaS | ผสมผสาน AI เข้ากับข้อมูลตลาด | ต้องการ White-label Solution |
ราคาและ ROI
เมื่อพูดถึงค่าใช้จ่ายในการทำ Backtest และพัฒนา AI สำหรับระบบเทรด มี 2 ส่วนหลักที่ต้องพิจารณา:
ค่าใช้จ่ายด้านข้อมูล (Data Costs)
| Provider | ราคาต่อ GB | ราคาต่อเดือน (Basic) | ความครอบคลุม |
|---|---|---|---|
| Tardis API | $0.15 | $99 | 300+ Exchange |
| CoinAPI | $0.25 | $79 | 200+ Exchange |
| 付 id: $0.20 | $0.20 | $50 | 100+ Exchange |
| Free Tier | ฟรี (จำกัด) | $0 | จำกัดปริมาณ |
ค่าใช้จ่ายด้าน AI API
| Provider | Model | ราคา/MTok (USD) | Latency |
|---|---|---|---|
| HolySheep AI | DeepSeek V3.2 | $0.42 ⭐ | <50ms |