สวัสดีครับ วันนี้ผมจะมาแบ่งปันประสบการณ์ตรงในการใช้งาน Tardis Python Client เพื่อดึงข้อมูล Orderbook ประวัติจาก Binance สำหรับการทำ Quantitative Trading Backtest ซึ่งเป็นหัวใจสำคัญของการพัฒนาระบบเทรดที่มีประสิทธิภาพ บทความนี้จะเป็นคู่มือครบวงจรตั้งแต่การติดตั้งไปจนถึงการนำข้อมูลไปใช้ในกลยุทธ์จริง
Tardis Python Client คืออะไร
Tardis เป็นเครื่องมือที่ช่วยให้นักพัฒนาสามารถเข้าถึงข้อมูลตลาดแบบเรียลไทม์และข้อมูลประวัติจาก Exchange หลายราย รวมถึง Binance โดยเฉพาะฟีเจอร์ Historical Orderbook Data ที่ช่วยให้เราสามารถดึงข้อมูลราคาเสนอซื้อ-ขายย้อนหลังได้ลึกถึงระดับ Tick-by-Tick ซึ่งมีความสำคัญอย่างยิ่งในการทำ Backtest ที่แม่นยำ
การติดตั้งและ Setup
การติดตั้ง Tardis Python Client ทำได้ง่ายผ่าน pip รองรับทั้ง Python 3.8+
# ติดตั้ง Tardis Python Client
pip install tardis-python
หรือใช้ Poetry
poetry add tardis-pythonการดึงข้อมูล Orderbook ประวัติจาก Binance
สำหรับการทำ Backtest ที่แม่นยำ เราต้องการข้อมูล Orderbook ที่มีความละเอียดสูง ด้านล่างเป็นโค้ดตัวอย่างการดึงข้อมูล Orderbook ประวัติจาก Binance Futures
import asyncio
from tardis_client import TardisClient, MessageType
async def fetch_historical_orderbook():
"""
ดึงข้อมูล Orderbook ประวัติจาก Binance Futures
รองรับ timeframe: 1ms, 1s, 1m, 1h
"""
tardis_client = TardisClient(auth=("YOUR_TARDIS_API_KEY"))
# ระบุช่วงเวลาที่ต้องการ
exchange = "binance" # หรือ "binance-futures"
symbol = "BTCUSDT"
from_timestamp = 1640995200000 # 2022-01-01 00:00:00 UTC
to_timestamp = 1643673600000 # 2022-02-01 00:00:00 UTC
# ดึงข้อมูล Orderbook L2 (Level 2 - รายละเอียดราคาแต่ละระดับ)
async for book in tardis_client.replay(
exchange=exchange,
symbols=[symbol],
from_timestamp=from_timestamp,
to_timestamp=to_timestamp,
filters=[{"name": "orderbook", "symbols": [symbol]}]
):
if book.type == MessageType.L2_UPDATE:
# book.bids = รายการเสนอซื้อ [(price, quantity), ...]
# book.asks = รายการเสนอขาย [(price, quantity), ...]
print(f"Time: {book.timestamp}")
print(f"Bids: {book.bids[:5]}") # 5 ระดับแรก
print(f"Asks: {book.asks[:5]}")
if __name__ == "__main__":
asyncio.run(fetch_historical_orderbook())การประมวลผลข้อมูลสำหรับ Backtest
หลังจากได้ข้อมูล Orderbook มาแล้ว เราต้องประมวลผลให้อยู่ในรูปแบบที่เหมาะกับการทำ Backtest โดยผมแนะนำให้ใช้ Pandas สำหรับจัดการข้อมูล และบันทึกลงไฟล์ Parquet เพื่อประสิทธิภาพในการอ่านซ้ำ
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from collections import defaultdict
from datetime import datetime
class OrderbookProcessor:
"""ประมวลผลข้อมูล Orderbook สำหรับ Backtest"""
def __init__(self, symbol: str, levels: int = 20):
self.symbol = symbol
self.levels = levels
self.orderbook_history = []
def process_l2_update(self, book):
"""
แปลงข้อมูล L2 Update เป็น DataFrame
คำนวณ: Spread, Mid Price, Weighted Mid Price
"""
timestamp = pd.to_datetime(book.timestamp, unit='ms')
# สร้าง Dictionary สำหรับ Orderbook ณ เวลานั้น
orderbook_data = {
'timestamp': timestamp,
'symbol': self.symbol,
'best_bid': float(book.bids[0][0]) if book.bids else None,
'best_ask': float(book.asks[0][0]) if book.asks else None,
'spread': None,
'mid_price': None,
'bid_volume_1': sum(float(b[1]) for b in book.bids[:1]),
'ask_volume_1': sum(float(a[1]) for a in book.asks[:1]),
}
# คำนวณ Spread และ Mid Price
if orderbook_data['best_bid'] and orderbook_data['best_ask']:
orderbook_data['spread'] = (
orderbook_data['best_ask'] - orderbook_data['best_bid']
) / orderbook_data['best_bid'] * 100 # เป็น %
orderbook_data['mid_price'] = (
orderbook_data['best_bid'] + orderbook_data['best_ask']
) / 2
# รวม Volume หลายระดับ
for i in range(1, min(self.levels, 20)):
try:
orderbook_data[f'bid_vol_{i}'] = float(book.bids[i][1]) if len(book.bids) > i else 0
orderbook_data[f'ask_vol_{i}'] = float(book.asks[i][1]) if len(book.asks) > i else 0
except:
orderbook_data[f'bid_vol_{i}'] = 0
orderbook_data[f'ask_vol_{i}'] = 0
self.orderbook_history.append(orderbook_data)
return orderbook_data
def save_to_parquet(self, filepath: str):
"""บันทึกข้อมูลเป็น Parquet สำหรับ Backtest ที่รวดเร็ว"""
df = pd.DataFrame(self.orderbook_history)
df.to_parquet(filepath, engine='pyarrow', compression='snappy')
print(f"บันทึก {len(df)} records ไปยัง {filepath}")
return df
def calculate_vwap_difference(self, df: pd.DataFrame, window: int = 100):
"""
คำนวณ VWAP Imbalance สำหรับใช้ในกลยุทธ์
VWAP Imbalance = (Bid_Volume - Ask_Volume) / (Bid_Volume + Ask_Volume)
"""
df['total_bid_vol'] = df[[f'bid_vol_{i}' for i in range(self.levels)]].sum(axis=1)
df['total_ask_vol'] = df[[f'ask_vol_{i}' for i in range(self.levels)]].sum(axis=1)
df['vwap_imbalance'] = (
df['total_bid_vol'] - df['total_ask_vol']
) / (df['total_bid_vol'] + df['total_ask_vol'])
# Smooth ด้วย Rolling Mean
df['vwap_imbalance_smooth'] = df['vwap_imbalance'].rolling(window).mean()
return df
ตัวอย่างการใช้งาน
processor = OrderbookProcessor(symbol="BTCUSDT", levels=10)
print("พร้อมสำหรับการประมวลผล Orderbook")การทำ Backtest กลยุทธ์ Orderbook Imbalance
ด้านล่างเป็นตัวอย่างการสร้าง Backtest Engine อย่างง่ายสำหรับกลยุทธ์ Orderbook Imbalance ที่ผมใช้จริงในการพัฒนา
import pandas as pd
import numpy as np
from dataclasses import dataclass
from typing import List, Optional
@dataclass
class Trade:
timestamp: pd.Timestamp
side: str # 'buy' หรือ 'sell'
price: float
quantity: float
pnl: float = 0
class OrderbookBacktester:
"""
Backtest Engine สำหรับกลยุทธ์ Orderbook Imbalance
"""
def __init__(
self,
initial_balance: float = 10000,
commission: float = 0.0004, # 0.04% Binance Futures
slippage: float = 0.0001 # 0.01% Slippage
):
self.initial_balance = initial_balance
self.balance = initial_balance
self.commission = commission
self.slippage = slippage
self.trades: List[Trade] = []
self.position = 0
self.position_price = 0
def load_data(self, filepath: str) -> pd.DataFrame:
"""โหลดข้อมูล Orderbook จาก Parquet"""
self.df = pd.read_parquet(filepath)
self.df = self.df.set_index('timestamp').sort_index()
print(f"โหลด {len(self.df)} records, ช่วง: {self.df.index[0]} - {self.df.index[-1]}")
return self.df
def run_backtest(
self,
df: pd.DataFrame,
entry_threshold: float = 0.15,
exit_threshold: float = 0.05,
stop_loss: float = 0.02
):
"""
รัน Backtest ด้วยกลยุทธ์ Imbalance
Logic:
- เข้า Long เมื่อ VWAP Imbalance > entry_threshold
- เข้า Short เมื่อ VWAP Imbalance < -entry_threshold
- ออกเมื่อ Imbalance กลับมาใกล้ 0 หรือถึง Stop Loss
"""
self.balance = self.initial_balance
self.position = 0
self.trades = []
for idx, row in df.iterrows():
imbalance = row['vwap_imbalance_smooth']
if pd.isna(imbalance):
continue
# ตรวจสอบ Position ปัจจุบัน
if self.position > 0: # ถือ Long
pnl_pct = (row['mid_price'] - self.position_price) / self.position_price
# ออกเมื่อ Imbalance ลดลงหรือถึง Stop Loss
if imbalance < exit_threshold or pnl_pct < -stop_loss:
self.close_position(row['mid_price'], 'sell', idx)
elif self.position < 0: # ถือ Short
pnl_pct = (self.position_price - row['mid_price']) / self.position_price
if imbalance > -exit_threshold or pnl_pct < -stop_loss:
self.close_position(row['mid_price'], 'buy', idx)
# เปิด Position ใหม่
if self.position == 0:
if imbalance > entry_threshold:
self.open_position(row['mid_price'], 'buy', idx)
elif imbalance < -entry_threshold:
self.open_position(row['mid_price'], 'sell', idx)
return self.calculate_metrics()
def open_position(self, price: float, side: str, timestamp):
"""เปิด Position ใหม่"""
# คำนวณ Slippage
if side == 'buy':
exec_price = price * (1 + self.slippage)
else:
exec_price = price * (1 - self.slippage)
# คำนวณขนาด Position (ใช้ 50% ของ Balance)
size = (self.balance * 0.5) / exec_price
# หัก Commission
commission_cost = size * exec_price * self.commission
self.balance -= commission_cost
self.position_price = exec_price
self.position = size if side == 'buy' else -size
def close_position(self, price: float, side: str, timestamp):
"""ปิด Position"""
if side == 'buy':
exec_price = price * (1 + self.slippage)
else:
exec_price = price * (1 - self.slippage)
pnl = self.position * (exec_price - self.position_price)
commission_cost = abs(self.position) * exec_price * self.commission
self.balance += pnl - commission_cost
self.trades.append(Trade(timestamp, side, exec_price, abs(self.position), pnl))
self.position = 0
def calculate_metrics(self):
"""คำนวณ Performance Metrics"""
if not self.trades:
return {"error": "No trades executed"}
df_trades = pd.DataFrame([{
'timestamp': t.timestamp,
'side': t.side,
'price': t.price,
'quantity': t.quantity,
'pnl': t.pnl
} for t in self.trades])
total_pnl = self.balance - self.initial_balance
win_rate = (df_trades['pnl'] > 0).mean()
# Annualized Return (สมมติ 365 วัน)
duration_days = (df_trades['timestamp'].max() - df_trades['timestamp'].min()).total_seconds() / 86400
annualized_return = (self.balance / self.initial_balance) ** (365 / max(duration_days, 1)) - 1
# Max Drawdown
cumulative = (1 + df_trades['pnl'] / self.initial_balance).cumprod()
running_max = cumulative.cummax()
drawdown = (cumulative - running_max) / running_max
max_drawdown = drawdown.min()
# Sharpe Ratio (สมมติ risk-free rate = 0)
returns = df_trades['pnl'] / self.initial_balance
sharpe = returns.mean() / returns.std() * np.sqrt(252) if returns.std() > 0 else 0
return {
'final_balance': self.balance,
'total_pnl': total_pnl,
'total_return': (self.balance / self.initial_balance - 1) * 100,
'annualized_return': annualized_return * 100,
'max_drawdown': max_drawdown * 100,
'sharpe_ratio': sharpe,
'win_rate': win_rate * 100,
'total_trades': len(df_trades),
'avg_pnl_per_trade': df_trades['pnl'].mean()
}
ตัวอย่างการรัน Backtest
if __name__ == "__main__":
backtester = OrderbookBacktester(initial_balance=10000)
# โหลดข้อมูล (สมมติว่ามีไฟล์ orderbook_btcusdt.parquet)
# df = backtester.load_data("orderbook_btcusdt.parquet")
# รัน Backtest
# metrics = backtester.run_backtest(df, entry_threshold=0.15, exit_threshold=0.05)
# print(metrics)รีวิวประสิทธิภาพ: ความหน่วง ความครอบคลุม และความสะดวก
จากประสบการณ์การใช้งานจริงของผมในช่วง 6 เดือนที่ผ่านมา นี่คือการประเมิน Tardis Python Client ตามเกณฑ์ที่ชัดเจน
| เกณฑ์ | รายละเอียด | คะแนน |
|---|---|---|
| ความหน่วง (Latency) | ข้อมูล Historical มีความล่าช้าน้อย แต่ต้องผ่าน API และการ Stream อาจมี lag ในช่วง peak | 8/10 |
| อัตราสำเร็จ | ดึงข้อมูลสำเร็จประมาณ 99.2% ของเวลาทั้งหมด มีบางช่วงที่ connection หลุด | 9/10 |
| ความครอบคลุมข้อมูล | ครอบคลุม Orderbook L2, Trades, Ticker จาก Exchange หลายราย รวม Binance, Bybit, OKX | 9/10 |
| ความสะดวกในการใช้งาน | API เข้าใจง่าย มี Documentation ดี แต่ต้องมีความรู้เรื่อง Async/Await | 7/10 |
| ประสบการณ์คอนโซล | ไม่มี Web Dashboard สำหรับดูข้อมูล ต้องใช้โค้ดเท่านั้น | 6/10 |
| ราคา | เริ่มต้น $49/เดือน ค่อนข้างแพงสำหรับมือใหม่ | 5/10 |
ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
ในการใช้งาน Tardis Python Client ผมพบข้อผิดพลาดหลายประเภทที่เกิดขึ้นบ่อย ด้านล่างเป็นวิธีแก้ไขที่ได้ผลจริง
1. Error: Connection timeout ขณะดึงข้อมูลระยะไกล
สาเหตุ: เนื่องจากการเชื่อมต่อไปยัง Tardis API Server มีความหน่วงสูง หรือ Network มีปัญหา
# วิธีแก้ไข: เพิ่ม Retry Logic และ Timeout ที่เหมาะสม
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def fetch_with_retry(tardis_client, *args, **kwargs):
"""ดึงข้อมูลพร้อม Retry Logic"""
try:
async for data in tardis_client.replay(*args, **kwargs):
return data
except asyncio.TimeoutError:
print("Connection timeout - retrying...")
raise
except Exception as e:
print(f"Error: {e}")
raise
หรือใช้ Timeout สำหรับแต่ละ Request
async def fetch_with_timeout():
timeout = aiohttp.ClientTimeout(total=300) # 5 นาที
async with aiohttp.ClientSession(timeout=timeout) as session:
# ดึงข้อมูล...
pass2. Memory Error เมื่อดึงข้อมูลจำนวนมาก
สาเหตุ: การ Stream ข้อมูลจำนวนมากในครั้งเดียวทำให้ RAM เต็ม
# วิธีแก้ไข: ใช้ Generator และ Batch Processing
import asyncio
async def fetch_in_batches():
"""ดึงข้อมูลเป็น Batch เพื่อประหยัด Memory"""
batch_size = 100000 # records ต่อ batch
batch = []
async for book in tardis_client.replay(
exchange="binance-futures",
symbols=["BTCUSDT"],
from_timestamp=from_ts,
to_timestamp=to_ts
):
batch.append(book)
if len(batch) >= batch_size:
# ประมวลผล Batch ก่อนที่จะดึงต่อ
processed = process_batch(batch)
save_to_parquet(processed, f"batch_{len(processed)}.parquet")
batch = [] # Clear memory
# ประมวลผล Batch สุดท้าย
if batch:
processed = process_batch(batch)
save_to_parquet(processed, "final_batch.parquet")
หรือใช้ Streaming ไปยัง Parquet โดยตรง
import pyarrow as pa
import pyarrow.parquet as pq
def stream_to_parquet():
"""Stream ข้อมูลไปยัง Parquet โดยตรง"""
writer = None
batch_count = 0
async for book in data_stream:
record = transform_book(book)
batch = pa.RecordBatch.from_pylist([record])
if writer is None:
schema = batch.schema
writer = pq.ParquetWriter('output.parquet', schema)
writer.write_batch(batch)
batch_count += 1
if batch_count % 100000 == 0:
print(f"Processed {batch_count} records")
writer.close()3. Timestamp Mismatch ระหว่าง Binance และ Local Time
สาเหตุ: Binance ใช้ UTC timestamp แต่เครื่อง Local อาจตั้ง Timezone ผิด
# วิธีแก้ไข: ตรวจสอบและแปลง Timezone ให้ถูกต้อง
from datetime import datetime, timezone
import pytz
def fix_timestamp_handling():
"""
แก้ไขปัญหา Timestamp Mismatch
Binance API ใช้ Milliseconds UTC
"""
# วิธีที่ 1: ใช้ UTC โดยตรง
utc_time = datetime.now(timezone.utc)
timestamp_ms = int(utc_time.timestamp() * 1000)
# วิธีที่ 2: แปลง Local Time เป็น UTC
local_tz = pytz.timezone('Asia/Bangkok') # หรือ timezone ที่ใช้
local_time = datetime.now(local_tz)
utc_time_correct = local_time.astimezone(pytz.UTC)
timestamp_correct = int(utc_time_correct.timestamp() * 1000)
# วิธีที่ 3: ใช้ pd.to_datetime กับ unit='ms' และ UTC
df['timestamp'] = pd.to_datetime(df['timestamp_ms'], unit='ms', utc=True)
df['timestamp'] = df['timestamp'].dt.tz_convert('Asia/Bangkok') # หรือ UTC
return timestamp_correct
ตรวจสอบว่า Timestamp ตรงกัน
def validate_timestamp():
"""ตรวจสอบว่า Timestamp ที่ได้มาจาก Binance ตรงกับที่คาดหวัง"""
# Binance จะ return timestamp เป็น milliseconds
# ตรวจสอบว่าอยู่ในช่วงที่กำหนด
expected_start = datetime(2024, 1, 1, tzinfo=timezone.utc)
expected_end = datetime(2024, 1, 2, tzinfo=timezone.utc)
actual_start = pd.Timestamp(expected_start).value // 10**6
actual_end = pd.Timestamp(expected_end).value // 10**6
print(f"Expected range: {actual_start} - {actual_end}")4. Rate Limit Error เมื่อดึงข้อมูลบ่อยเกินไป
สาเหตุ: Tardis มี Rate Limit สำหรับ Historical API ถ้าเรียกบ่อยเกินจะถูก Block ชั่วคราว
# วิธีแก้ไข: ใช้ Rate Limiter และ Caching
import asyncio
from asyncio_throttle import Throttler
import hashlib
class RateLimitedTardisClient:
"""Wrapper สำหรับ