บทนำ: ทำไมต้องดึงข้อมูล Tick จาก OKX
สำหรับเทรดเดอร์และนักพัฒนาโบรกเกอร์ที่ต้องการทดสอบกลยุทธ์การเทรดสัญญา perpetuity ของ OKX การได้ข้อมูล Tick คุณภาพสูงเป็นพื้นฐานที่สำคัญที่สุด ข้อมูล Tick ประกอบด้วยราคา ปริมาณการซื้อขาย และเวลาที่แม่นยำถึงมิลลิวินาที ซึ่งช่วยให้การทดสอบระบบ (Backtest) มีความใกล้เคียงกับสภาพตลาดจริงมากที่สุด Tardis API เป็นบริการที่ให้บริการข้อมูลตลาดคริปโตแบบ Historical Replay ที่สามารถดึงข้อมูล Tick จาก OKX ได้อย่างครบถ้วน
บทความนี้จะพาคุณสร้าง Pipeline สำหรับดาวน์โหลดข้อมูล OKX Perpetual Futures Tick และประมวลผลด้วย Python เพื่อนำไปใช้ในการ Backtest กลยุทธ์ โดยเราจะใช้
HolySheep AI เพื่อวิเคราะห์ข้อมูลและสร้างรายงานด้วย AI ที่มีความเร็วสูงและต้นทุนต่ำ
ราคา AI API 2026: เปรียบเทียบต้นทุนสำหรับ 10 ล้าน Tokens/เดือน
ก่อนเริ่มสร้าง Pipeline เรามาดูต้นทุนของ AI API ต่างๆ สำหรับการประมวลผลข้อมูลและสร้างรายงานการ Backtest
| โมเดล | ราคา ($/M Tokens) | ต้นทุน 10M Tokens | ความเร็วโดยประมาณ | เหมาะกับงาน |
|---|
| DeepSeek V3.2 | $0.42 | $4.20 | <50ms | วิเคราะห์ข้อมูล, สร้างสคริปต์ |
| Gemini 2.5 Flash | $2.50 | $25.00 | <100ms | รายงานสรุป, การจัดการข้อมูล |
| GPT-4.1 | $8.00 | $80.00 | <200ms | งานวิเคราะห์เชิงลึก |
| Claude Sonnet 4.5 | $15.00 | $150.00 | <150ms | การเขียนโค้ดซับซ้อน |
จากตารางจะเห็นได้ว่า
HolySheep AI มีราคาที่ประหยัดมาก โดย DeepSeek V3.2 มีราคาเพียง $0.42/MTok ซึ่งถูกกว่า Claude Sonnet 4.5 ถึง 35 เท่า และรองรับทั้ง WeChat และ Alipay พร้อมเครดิตฟรีเมื่อลงทะเบียน
การติดตั้งและตั้งค่า Python Environment
# สร้าง Virtual Environment สำหรับ Backtest Pipeline
python -m venv backtest_env
source backtest_env/bin/activate # Windows: backtest_env\Scripts\activate
ติดตั้ง Dependencies ที่จำเป็น
pip install requests pandas numpy \
tardis-client \
websocket-client \
python-dotenv \
pyarrow \
sqlalchemy
สร้างไฟล์ .env สำหรับเก็บ API Keys
cat > .env << 'EOF'
TARDIS_API_KEY=your_tardis_api_key_here
HOLYSHEEP_API_KEY=your_holysheep_api_key_here
OKX_API_KEY=your_okx_api_key_here
EOF
ดาวน์โหลดข้อมูล OKX Tick ผ่าน Tardis API
import os
import json
import asyncio
from datetime import datetime, timedelta
from tardis_client import TardisClient, TardisRealtime, MessageType
class OKXDataDownloader:
"""Downloader สำหรับข้อมูล OKX Perpetual Futures Tick"""
def __init__(self, api_key: str):
self.api_key = api_key
self.tardis_client = TardisClient(api_key)
self.exchange = "okx"
self.data_type = "trade" # ดึงเฉพาะ Trade Data (Tick)
def download_trades(
self,
symbol: str,
from_timestamp: datetime,
to_timestamp: datetime,
output_path: str = "./data"
) -> str:
"""
ดาวน์โหลดข้อมูล Trade จาก OKX
Args:
symbol: คู่เทรด เช่น "BTC-USDT-SWAP"
from_timestamp: เวลาเริ่มต้น
to_timestamp: เวลาสิ้นสุด
output_path: โฟลเดอร์สำหรับบันทึกข้อมูล
Returns:
ที่อยู่ไฟล์ข้อมูลที่บันทึก
"""
os.makedirs(output_path, exist_ok=True)
# แปลง timestamp เป็นรูปแบบ ISO 8601
from_str = from_timestamp.isoformat()
to_str = to_timestamp.isoformat()
print(f"เริ่มดาวน์โหลด {symbol} จาก {from_str} ถึง {to_str}")
# ดึงข้อมูล Trade จาก Tardis
trades_data = []
for trade in self.tardis_client.trades(
exchange=self.exchange,
symbol=symbol,
from_timestamp=from_str,
to_timestamp=to_str
):
trade_dict = {
"timestamp": trade.timestamp,
"symbol": trade.symbol,
"side": trade.side, # buy หรือ sell
"price": float(trade.price),
"amount": float(trade.amount),
"trade_id": trade.id
}
trades_data.append(trade_dict)
# บันทึกเป็น JSON Lines format
filename = f"{symbol.replace('-', '_')}_{from_timestamp.strftime('%Y%m%d')}_{to_timestamp.strftime('%Y%m%d')}.jsonl"
filepath = os.path.join(output_path, filename)
with open(filepath, 'w', encoding='utf-8') as f:
for trade in trades_data:
f.write(json.dumps(trade, ensure_ascii=False) + '\n')
print(f"ดาวน์โหลดเสร็จสิ้น: {len(trades_data)} records -> {filepath}")
return filepath
ตัวอย่างการใช้งาน
if __name__ == "__main__":
from dotenv import load_dotenv
load_dotenv()
downloader = OKXDataDownloader(api_key=os.getenv("TARDIS_API_KEY"))
# ดาวน์โหลดข้อมูลย้อนหลัง 1 ชั่วโมง
to_time = datetime.now()
from_time = to_time - timedelta(hours=1)
data_file = downloader.download_trades(
symbol="BTC-USDT-SWAP",
from_timestamp=from_time,
to_timestamp=to_time
)
สร้าง Backtest Pipeline แบบครบวงจร
import pandas as pd
import numpy as np
from dataclasses import dataclass
from typing import List, Dict, Callable
from datetime import datetime
import json
@dataclass
class OHLCV:
"""โครงสร้างข้อมูล OHLCV สำหรับการ Backtest"""
timestamp: datetime
open: float
high: float
low: float
close: float
volume: float
@dataclass
class BacktestResult:
"""ผลลัพธ์การ Backtest"""
total_trades: int
winning_trades: int
losing_trades: int
total_profit: float
max_drawdown: float
sharpe_ratio: float
trades: List[Dict]
class BacktestEngine:
"""Engine สำหรับรัน Backtest จากข้อมูล Tick"""
def __init__(self, initial_balance: float = 10000.0):
self.initial_balance = initial_balance
self.balance = initial_balance
self.position = 0.0
self.trades: List[Dict] = []
self.equity_curve: List[float] = []
def load_tick_data(self, filepath: str, timeframe: str = "1T") -> pd.DataFrame:
"""
แปลงข้อมูล Tick เป็น OHLCV ตาม timeframe ที่กำหนด
Args:
filepath: ที่อยู่ไฟล์ JSONL
timeframe: Timeframe เช่น "1T" (1 นาที), "5T", "1H"
Returns:
DataFrame ที่มี OHLCV
"""
# อ่านข้อมูล Tick
trades = []
with open(filepath, 'r', encoding='utf-8') as f:
for line in f:
trades.append(json.loads(line))
df = pd.DataFrame(trades)
df['timestamp'] = pd.to_datetime(df['timestamp'])
df.set_index('timestamp', inplace=True)
df = df.sort_index()
# Resample เป็น OHLCV
ohlcv = df['price'].resample(timeframe).ohlc()
ohlcv['volume'] = df['amount'].resample(timeframe).sum()
ohlcv = ohlcv.dropna()
print(f"แปลงข้อมูลเสร็จ: {len(ohlcv)} candles")
return ohlcv
def add_strategy(
self,
name: str,
entry_condition: Callable[[pd.DataFrame, int], bool],
exit_condition: Callable[[pd.DataFrame, int], bool]
):
"""
เพิ่มกลยุทธ์การเทรด
Args:
name: ชื่อกลยุทธ์
entry_condition: ฟังก์ชันตรวจสอบเงื่อนไขเข้า Position
exit_condition: ฟังก์ชันตรวจสอบเงื่อนไขออก Position
"""
self.strategy_name = name
self.entry_condition = entry_condition
self.exit_condition = exit_condition
def run(self, ohlcv_data: pd.DataFrame) -> BacktestResult:
"""รัน Backtest"""
print(f"เริ่ม Backtest: {self.strategy_name}")
self.balance = self.initial_balance
self.position = 0.0
self.trades = []
self.equity_curve = [self.initial_balance]
entry_price = 0.0
for i in range(len(ohlcv_data)):
current_price = ohlcv_data.iloc[i]['close']
# เข้า Position
if self.position == 0 and self.entry_condition(ohlcv_data, i):
entry_price = current_price
self.position = self.balance / current_price
self.trades.append({
"entry_time": ohlcv_data.index[i],
"entry_price": entry_price,
"type": "LONG"
})
# ออก Position
elif self.position > 0 and self.exit_condition(ohlcv_data, i):
exit_price = current_price
pnl = (exit_price - entry_price) * self.position
self.balance += pnl
self.trades[-1].update({
"exit_time": ohlcv_data.index[i],
"exit_price": exit_price,
"pnl": pnl,
"balance": self.balance
})
self.position = 0.0
# บันทึก Equity Curve
equity = self.balance + (self.position * current_price if self.position > 0 else 0)
self.equity_curve.append(equity)
return self._calculate_results()
def _calculate_results(self) -> BacktestResult:
"""คำนวณผลลัพธ์การ Backtest"""
if not self.trades:
return BacktestResult(0, 0, 0, 0, 0, 0, [])
winning_trades = [t for t in self.trades if t.get('pnl', 0) > 0]
losing_trades = [t for t in self.trades if t.get('pnl', 0) <= 0]
total_profit = sum(t.get('pnl', 0) for t in self.trades)
# คำนวณ Max Drawdown
equity = np.array(self.equity_curve)
running_max = np.maximum.accumulate(equity)
drawdown = (equity - running_max) / running_max
max_drawdown = abs(drawdown.min())
# คำนวณ Sharpe Ratio (annualized)
returns = np.diff(equity) / equity[:-1]
sharpe_ratio = (returns.mean() / returns.std()) * np.sqrt(252 * 24) if returns.std() > 0 else 0
return BacktestResult(
total_trades=len(self.trades),
winning_trades=len(winning_trades),
losing_trades=len(losing_trades),
total_profit=total_profit,
max_drawdown=max_drawdown,
sharpe_ratio=sharpe_ratio,
trades=self.trades
)
ตัวอย่างกลยุทธ์ Simple Moving Average Crossover
def sma_crossover_strategy(data: pd.DataFrame, i: int, short_period: int = 10, long_period: int = 30) -> bool:
"""กลยุทธ์ SMA Crossover"""
if i < long_period:
return False
short_sma = data['close'].iloc[i-short_period:i].mean()
long_sma = data['close'].iloc[i-long_period:i].mean()
prev_short_sma = data['close'].iloc[i-short_period-1:i-1].mean()
prev_long_sma = data['close'].iloc[i-long_period-1:i-1].mean()
# Golden Cross - เข้า LONG
return prev_short_sma <= prev_long_sma and short_sma > long_sma
def sma_crossunder_exit(data: pd.DataFrame, i: int, short_period: int = 10, long_period: int = 30) -> bool:
"""เงื่อนไขออก - Death Cross"""
if i < long_period:
return False
short_sma = data['close'].iloc[i-short_period:i].mean()
long_sma = data['close'].iloc[i-long_period:i].mean()
prev_short_sma = data['close'].iloc[i-short_period-1:i-1].mean()
prev_long_sma = data['close'].iloc[i-long_period-1:i-1].mean()
# Death Cross - ออก
return prev_short_sma >= prev_long_sma and short_sma < long_sma
รัน Backtest
engine = BacktestEngine(initial_balance=10000.0)
engine.add_strategy(
name="SMA Crossover 10/30",
entry_condition= lambda df, i: sma_crossover_strategy(df, i),
exit_condition= lambda df, i: sma_crossunder_exit(df, i)
)
โหลดข้อมูลและรัน
data = engine.load_tick_data("./data/BTC_USDT_SWAP_20260101_20260102.jsonl", "1T")
results = engine.run(data)
print(f"Total Trades: {results.total_trades}")
print(f"Win Rate: {results.winning_trades / results.total_trades * 100:.2f}%")
print(f"Total Profit: ${results.total_profit:.2f}")
print(f"Max Drawdown: {results.max_drawdown * 100:.2f}%")
print(f"Sharpe Ratio: {results.sharpe_ratio:.2f}")
ใช้ AI วิเคราะห์ผลลัพธ์ Backtest ด้วย HolySheep
import os
import requests
from datetime import datetime
class HolySheepAnalyzer:
"""ใช้ HolySheep AI วิเคราะห์ผลลัพธ์ Backtest"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1" # Base URL ของ HolySheep
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def analyze_backtest_results(self, backtest_result, symbol: str, strategy: str) -> str:
"""
วิเคราะห์ผลลัพธ์ Backtest ด้วย DeepSeek V3.2
Args:
backtest_result: ผลลัพธ์จาก BacktestEngine
symbol: คู่เทรด
strategy: ชื่อกลยุทธ์
Returns:
รายงานการวิเคราะห์จาก AI
"""
prompt = f"""คุณเป็นผู้เชี่ยวชาญด้าน Quantitative Trading วิเคราะห์ผลการ Backtest ต่อไปนี้:
สินทรัพย์: {symbol}
กลยุทธ์: {strategy}
วันที่: {datetime.now().strftime('%Y-%m-%d')}
ผลการ Backtest:
- จำนวน Trades ทั้งหมด: {backtest_result.total_trades}
- Trades ที่กำไร: {backtest_result.winning_trades}
- Trades ที่ขาดทุน: {backtest_result.losing_trades}
- อัตราชนะ: {backtest_result.winning_trades / backtest_result.total_trades * 100:.2f}%
- กำไรรวม: ${backtest_result.total_profit:.2f}
- Max Drawdown: {backtest_result.max_drawdown * 100:.2f}%
- Sharpe Ratio: {backtest_result.sharpe_ratio:.2f}
กรุณาวิเคราะห์:
1. ประสิทธิภาพของกลยุทธ์นี้
2. จุดแข็งและจุดอ่อน
3. คำแนะนำในการปรับปรุง
4. ความเสี่ยงที่ต้องระวัง
5. สรุปแนะนำว่าควรใช้กลยุทธ์นี้หรือไม่
ตอบเป็นภาษาไทย มีความละเอียดและเป็นประโยชน์สำหรับเทรดเดอร์"""
payload = {
"model": "deepseek-v3.2", # ใช้ DeepSeek V3.2 ราคา $0.42/MTok
"messages": [
{"role": "user", "content": prompt}
],
"temperature": 0.7,
"max_tokens": 2000
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload
)
if response.status_code == 200:
result = response.json()
return result['choices'][0]['message']['content']
else:
raise Exception(f"API Error: {response.status_code} - {response.text}")
ตัวอย่างการใช้งาน
if __name__ == "__main__":
analyzer = HolySheepAnalyzer(api_key=os.getenv("HOLYSHEEP_API_KEY"))
# วิเคราะห์ผลลัพธ์
analysis = analyzer.analyze_backtest_results(
backtest_result=results,
symbol="BTC-USDT-SWAP",
strategy="SMA Crossover 10/30"
)
print("=" * 60)
print("รายงานการวิเคราะห์จาก HolySheep AI")
print("=" * 60)
print(analysis)
สร้าง Telegram Bot แจ้งเตือนสถานะ Pipeline
import requests
from datetime import datetime
class PipelineNotifier:
"""ส่งการแจ้งเตือนสถานะ Pipeline ผ่าน Telegram"""
def __init__(self, bot_token: str, chat_id: str):
self.bot_token = bot_token
self.chat_id = chat_id
self.api_url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
def send_status(self, status: str, message: str):
"""ส่งข้อความสถานะ"""
payload = {
"chat_id": self.chat_id,
"text": f"📊 *Pipeline Status*\n\n"
f"🕐 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
f"📌 Status: {status}\n"
f"📝 Message: {message}",
"parse_mode": "Markdown"
}
response = requests.post(self.api_url, json=payload)
return response.status_code == 200
def send_backtest_summary(self, results, symbol: str, strategy: str):
"""ส่งสรุปผล Backtest"""
win_rate = results.winning_trades / results.total_trades * 100 if results.total_trades > 0 else 0
message = f"""
*📈 Backtest Summary*
*สินทรัพย์:* {symbol}
*กลยุทธ์:* {strategy}
*ผลการเทรด:*
├ จำนวน Trades: {results.total_trades}
├ Win Rate: {win_rate:.1f}%
├ กำไร: ${results.total_profit:.2f}
├ Max Drawdown: {results.max_drawdown * 100:.1f}%
└ Sharpe Ratio: {results.sharpe_ratio:.2f}
🕐 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
"""
payload = {
"chat_id": self.chat_id,
"text": message,
"parse_mode": "Markdown"
}
return requests.post(self.api_url, json=payload).status_code == 200
ใช้งาน
notifier = PipelineNotifier(
bot_token="YOUR_TELEGRAM_BOT_TOKEN",
chat_id="YOUR_CHAT_ID"
)
ส่งสถานะ
notifier.send_status("✅", "Backtest Pipeline รันเสร็จสมบูรณ์")
notifier.send_backtest_summary(results, "BTC-USDT-SWAP", "SMA Crossover 10/30")
รวมทุกอย่างเป็น Automated Pipeline
import schedule
import time
from datetime import datetime, timedelta
def run_daily_backtest_pipeline():
"""รัน Pipeline ทุกวันอัตโนมัติ"""
print(f"[{datetime.now()}] เริ่ม Daily Backtest Pipeline")
try:
notifier = PipelineNotifier(
bot_token=os.getenv("TELEGRAM_BOT_TOKEN"),
chat_id=os.getenv("TELEGRAM_CHAT_ID")
)
notifier.send_status("⏳", "กำลังดาวน์โหลดข้อมูล...")
# 1. ดาวน์โหลดข้อมูล
downloader = OKXDataDownloader(api_key=os.getenv("TARDIS_API_KEY"))
to_time = datetime.now()
from_time = to_time - timedelta(days=7)
symbols = ["BTC-USDT-SWAP", "ETH-USDT-SWAP", "SOL-USDT-SWAP"]
all_results = {}
for symbol in symbols:
try:
data_file = downloader.download_trades(symbol, from_time, to_time)
# 2. รัน Backtest
engine = BacktestEngine(initial_balance=10000.0)
engine.add_strategy(
name="SMA Crossover 10/30",
entry_condition=lambda df, i: sma_crossover_strategy(df, i),
exit_condition=lambda df, i: sma_crossunder_exit(df, i)
)
data = engine.load_tick_data(data_file, "1T")
results = engine.run(data)
all_results[symbol] = results
except Exception as e:
print(f"Error processing {symbol}: {e}")
continue
# 3. วิเคราะห์ด้วย AI
notifier.send_status("🤖", "กำลังวิเคราะห์ด้วย HolySheep AI...")
analyzer = HolySheepAnalyzer(api_key=os.getenv("HOLYSHEEP_API_KEY"))
for symbol, results in all_results.items():
analysis = analyzer.analyze_backtest_results(
results, symbol, "SMA Crossover 10/30"
)
print(f"\n--- {symbol} Analysis ---")
print(analysis)
# 4. ส่งสรุปไป Telegram
for symbol, results in all_results.items():
notifier.send_backtest_summary(results, symbol, "SMA Crossover 10/30")
notifier.send_status("✅", "Pipeline เสร็จสมบูรณ์!")
except Exception as e:
print(f"Pipeline Error: {e}")
notifier.send_status("❌", f"Pipeline Error: {e}")
ตั้งเวลารันทุกวันเวลา 08:00 น.
schedule.every().day.at("08:00").do(run_daily_backtest_pipeline)
print("เริ่มต้น Pipeline Scheduler...")
while True:
schedule.run_pending()
time.sleep(60)
เหมาะกับใคร / ไม่เหมาะกับใคร
| เหมาะกับ | ไม่เหมาะกับ |
|---|
แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง
🔥 ลอง HolySheep AI
เกตเวย์ AI API โดยตรง รองรับ Claude, GPT-5, Gemini, DeepSeek — หนึ่งคีย์ ไม่ต้อง VPN
👉 สมัครฟรี →