บทความนี้สอนวิธีดึงข้อมูลตลาดคริปโตแบบ High-Frequency จาก Tardis.dev เข้าสู่ Python Pandas เพื่อใช้สร้างระบบ Backtesting สำหรับ Quantitative Trading โดยเน้นการประมวลผล Order Book, Trade Data และ Candlestick อย่างมีประสิทธิภาพ เหมาะสำหรับนักเทรดและนักพัฒนาที่ต้องการทดสอบกลยุทธ์ด้วยข้อมูลจริงจาก Exchange ยอดนิยม

Tardis.dev คืออะไร ทำไมต้องใช้สำหรับ Backtesting

Tardis.dev เป็นบริการ Historical Market Data API ที่รวบรวมข้อมูลตลาดคริปโตครบถ้วนที่สุดรายการหนึ่ง ครอบคลุม Exchange กว่า 50 แห่ง รวมถึง Binance, Bybit, OKX, BitMEX และ Deribit มีจุดเด่นด้านคุณภาพข้อมูลระดับ Tick-by-Tick พร้อม WebSocket Streaming และ REST API สำหรับดึงข้อมูลย้อนหลังหลายปี

ประเภทข้อมูลที่รองรับ

ติดตั้งและเตรียม Environment

# สร้าง Virtual Environment แยกสำหรับโปรเจกต์
python -m venv quant_backtest
source quant_backtest/bin/activate  # Windows: quant_backtest\Scripts\activate


ติดตั้ง Dependencies

pip install pandas numpy requests asyncio aiohttp
pip install pandas>=2.0.0 numpy>=1.24.0
pip install tardis-client  # Official Python Client ของ Tardis.dev


ตรวจสอบเวอร์ชัน

python -c "import pandas; import numpy; print(f'Pandas: {pandas.__version__}')"

วิธีที่ 1: ดึงข้อมูล Trades ผ่าน REST API

วิธีนี้เหมาะสำหรับดึงข้อมูล Trade History ย้อนหลังในช่วงเวลาที่กำหนด เช่น ดึงข้อมูล 1 เดือนของ BTCUSDT Perpetual

import pandas as pd
import requests
from datetime import datetime, timedelta
import time

class TardisDataFetcher:
    """คลาสสำหรับดึงข้อมูลตลาดจาก Tardis.dev API"""
    
    BASE_URL = "https://api.tardis.dev/v1"
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({"Authorization": api_key})
    
    def get_trades(
        self,
        exchange: str,
        symbol: str,
        start_date: datetime,
        end_date: datetime,
        limit: int = 100000
    ) -> pd.DataFrame:
        """
        ดึงข้อมูล Trade History จาก Exchange ที่กำหนด
        
        Args:
            exchange: ชื่อ Exchange เช่น 'binance', 'bybit'
            symbol: สัญลักษณ์คู่เทรด เช่น 'BTCUSDT'
            start_date: วันที่เริ่มต้น
            end_date: วันที่สิ้นสุด
            limit: จำนวน Records สูงสุดต่อ Request
        
        Returns:
            DataFrame ที่มีคอลัมน์: timestamp, price, amount, side
        """
        url = f"{self.BASE_URL}/ trades"
        params = {
            "exchange": exchange,
            "symbol": symbol,
            "from": int(start_date.timestamp()),
            "to": int(end_date.timestamp()),
            "limit": limit
        }
        
        all_trades = []
        cursor = None
        
        print(f"กำลังดึงข้อมูล {exchange}:{symbol} จาก {start_date} ถึง {end_date}")
        
        while True:
            if cursor:
                params["cursor"] = cursor
            
            response = self.session.get(url, params=params)
            response.raise_for_status()
            
            data = response.json()
            
            if not data.get("trades"):
                break
            
            all_trades.extend(data["trades"])
            cursor = data.get("nextCursor")
            
            if not cursor:
                break
            
            # Tardis.dev มี Rate Limit ดังนั้นต้องหน่วงเวลา
            time.sleep(0.5)
            
            print(f"ดึงได้แล้ว {len(all_trades)} records...")
        
        # แปลงเป็น DataFrame
        df = pd.DataFrame(all_trades)
        
        if not df.empty:
            df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
            df["price"] = df["price"].astype(float)
            df["amount"] = df["amount"].astype(float)
            
            # เรียงลำดับตามเวลา
            df = df.sort_values("timestamp").reset_index(drop=True)
        
        print(f"สรุป: ดึงข้อมูลได้ทั้งหมด {len(df)} trades")
        
        return df


ตัวอย่างการใช้งาน

API_KEY = "YOUR_TARDIS_API_KEY"  # สมัครที่ https://tardis.dev

fetcher = TardisDataFetcher(API_KEY)


ดึงข้อมูล BTCUSDT Perpetual 7 วันย้อนหลัง

end_date = datetime.now()
start_date = end_date - timedelta(days=7)

btc_trades = fetcher.get_trades(
    exchange="binance",
    symbol="BTCUSDT",
    start_date=start_date,
    end_date=end_date
)

print(btc_trades.head(10))
print(f"\nข้อมูลครอบคลุม: {btc_trades['timestamp'].min()} ถึง {btc_trades['timestamp'].max()}")

วิธีที่ 2: ดึงข้อมูล Order Book ผ่าน WebSocket Streaming

สำหรับข้อมูล Order Book ที่ละเอียดกว่า ต้องใช้ WebSocket Streaming ซึ่งให้ข้อมูลแบบ Real-time แต่สามารถ Replay ได้เช่นกัน

import asyncio
import aiohttp
import pandas as pd
from dataclasses import dataclass, field
from typing import List, Dict
from datetime import datetime
import json

@dataclass
class OrderBookLevel:
    """โครงสร้างข้อมูลระดับราคาของ Order Book"""
    price: float
    amount: float
    side: str  # 'bid' หรือ 'ask'

class TardisWebSocketCollector:
    """คลาสสำหรับรวบรวมข้อมูล Order Book ผ่าน WebSocket"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.trades: List[Dict] = []
        self.orderbook_snapshots: List[Dict] = []
        self.liquidations: List[Dict] = []
    
    async def fetch_replayed_data(
        self,
        exchange: str,
        symbols: List[str],
        start_date: datetime,
        end_date: datetime
    ):
        """
        ดึงข้อมูลย้อนหลังผ่าน Replay API
        
        วิธีนี้เหมาะสำหรับดึงข้อมูลปริมาณมากๆ โดยไม่ต้องรอ Real-time
        """
        replay_url = f"https://api.tardis.dev/v1/feeds/{exchange}:{','.join(symbols)}"
        
        params = {
            "from": int(start_date.timestamp()),
            "to": int(end_date.timestamp()),
            "channels": "trade,book"
        }
        
        headers = {"Authorization": self.api_key}
        
        print(f"เริ่ม Replay ข้อมูล {exchange}:{symbols}")
        print(f"ช่วงเวลา: {start_date} ถึง {end_date}")
        
        async with aiohttp.ClientSession() as session:
            # ใช้ SSE (Server-Sent Events) สำหรับดึงข้อมูล
            async with session.get(
                replay_url,
                params=params,
                headers=headers
            ) as response:
                
                async for line in response.content:
                    line = line.decode("utf-8").strip()
                    
                    if not line or not line.startswith("data:"):
                        continue
                    
                    try:
                        data = json.loads(line[5:])
                        await self._process_message(data)
                    except json.JSONDecodeError:
                        continue
                    
                    # แสดงความคืบหน้า
                    if len(self.trades) % 10000 == 0 and len(self.trades) > 0:
                        print(f"ประมวลผลได้ {len(self.trades)} trades, "
                              f"{len(self.orderbook_snapshots)} snapshots")
        
        print(f"เสร็จสิ้น! รวม {len(self.trades)} trades, "
              f"{len(self.orderbook_snapshots)} orderbook snapshots")
    
    async def _process_message(self, data: Dict):
        """ประมวลผลข้อความจาก WebSocket"""
        msg_type = data.get("type") or data.get("channel")
        
        if msg_type == "trade":
            self.trades.append({
                "timestamp": pd.to_datetime(data["timestamp"], unit="ms"),
                "symbol": data.get("symbol"),
                "price": float(data["price"]),
                "amount": float(data["amount"]),
                "side": data.get("side", "buy" if data.get("is_buyer_maker") else "sell")
            })
        
        elif msg_type == "book" or msg_type == "snapshot":
            self.orderbook_snapshots.append({
                "timestamp": pd.to_datetime(data["timestamp"], unit="ms"),
                "symbol": data.get("symbol"),
                "bids": [[float(p), float(a)] for p, a in data.get("bids", [])],
                "asks": [[float(p), float(a)] for p, a in data.get("asks", [])]
            })
        
        elif msg_type == " liquidation":
            self.liquidations.append({
                "timestamp": pd.to_datetime(data["timestamp"], unit="ms"),
                "symbol": data.get("symbol"),
                "side": data.get("side"),
                "price": float(data["price"]),
                "amount": float(data["amount"])
            })
    
    def to_dataframes(self) -> Dict[str, pd.DataFrame]:
        """แปลงข้อมูลที่รวบรวมเป็น DataFrame"""
        return {
            "trades": pd.DataFrame(self.trades) if self.trades else pd.DataFrame(),
            "orderbook": pd.DataFrame(self.orderbook_snapshots) if self.orderbook_snapshots else pd.DataFrame(),
            "liquidations": pd.DataFrame(self.liquidations) if self.liquidations else pd.DataFrame()
        }

async def main():
    """ตัวอย่างการใช้งาน WebSocket Replay"""
    api_key = "YOUR_TARDIS_API_KEY"
    
    collector = TardisWebSocketCollector(api_key)
    
    # ดึงข้อมูล 1 ชั่วโมงของ BTCUSDT Perpetual
    end_time = datetime.now()
    start_time = end_time - timedelta(hours=1)
    
    await collector.fetch_replayed_data(
        exchange="binance",
        symbols=["BTCUSDT"],
        start_date=start_time,
        end_date=end_time
    )
    
    # แปลงเป็น DataFrame
    dfs = collector.to_dataframes()
    
    if not dfs["trades"].empty:
        trades_df = dfs["trades"]
        print("\n=== Trade Data Sample ===")
        print(trades_df.head())
        print(f"\nTime Range: {trades_df['timestamp'].min()} - {trades_df['timestamp'].max()}")
        print(f"Total Volume: {trades_df['amount'].sum():,.2f} BTC")
        print(f"Avg Trade Size: {trades_df['amount'].mean():,.6f} BTC")


รัน Asyncio

if __name__ == "__main__":
    asyncio.run(main())

วิธีที่ 3: ดึงข้อมูล OHLCV/Candlestick พร้อมสร้าง Features

สำหรับการสร้างกลยุทธ์ Trading ที่ใช้ Technical Indicators ข้อมูล Candlestick เป็นสิ่งจำเป็น ตัวอย่างนี้แสดงการดึงข้อมูลและคำนวณ Features สำหรับ Machine Learning

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class TradingFeatureEngineer:
    """คลาสสำหรับสร้าง Features สำหรับ Quant Trading"""
    
    def __init__(self, trades_df: pd.DataFrame):
        """
        Args:
            trades_df: DataFrame ที่มีคอลัมน์ timestamp, price, amount, side
        """
        self.trades = trades_df.copy()
        self.candles = None
    
    def resample_to_candles(self, timeframe: str = "1T") -> pd.DataFrame:
        """
        แปลง Trade Data เป็น Candlestick
        
        Args:
            timeframe: ระยะเวลาของแท่งเทียน เช่น '1T' = 1 นาที, '5T' = 5 นาที, '1H' = 1 ชั่วโมง
        """
        self.trades.set_index("timestamp", inplace=True)
        
        # Resample OHLCV
        self.candles = self.trades["price"].resample(timeframe).ohlc()
        self.candles["volume"] = self.trades["amount"].resample(timeframe).sum()
        self.candles["trade_count"] = self.trades["price"].resample(timeframe).count()
        
        # คำนวณ VWAP (Volume Weighted Average Price)
        self.candles["vwap"] = (
            (self.trades["price"] * self.trades["amount"])
            .resample(timeframe).sum() 
            / self.candles["volume"]
        )
        
        # คำนวณ Buy/Sell Volume
        buy_vol = self.trades[self.trades["side"] == "buy"]["amount"].resample(timeframe).sum()
        sell_vol = self.trades[self.trades["side"] == "sell"]["amount"].resample(timeframe).sum()
        
        self.candles["buy_volume"] = buy_vol.fillna(0)
        self.candles["sell_volume"] = sell_vol.fillna(0)
        self.candles["buy_ratio"] = self.candles["buy_volume"] / self.candles["volume"]
        
        # คำนวณ Volume Imbalance
        self.candles["volume_imbalance"] = (
            (self.candles["buy_volume"] - self.candles["sell_volume"]) 
            / self.candles["volume"]
        )
        
        # Reset Index
        self.candles = self.candles.reset_index()
        self.candles.rename(columns={"timestamp": "datetime"}, inplace=True)
        
        # ลบ NaN rows
        self.candles.dropna(inplace=True)
        
        return self.candles
    
    def add_technical_indicators(self) -> pd.DataFrame:
        """เพิ่ม Technical Indicators พื้นฐาน"""
        if self.candles is None:
            raise ValueError("ต้องเรียก resample_to_candles() ก่อน")
        
        df = self.candles.copy()
        
        # Simple Moving Averages
        for period in [5, 10, 20, 50, 200]:
            df[f"SMA_{period}"] = df["close"].rolling(window=period).mean()
        
        # Exponential Moving Averages
        for period in [12, 26]:
            df[f"EMA_{period}"] = df["close"].ewm(span=period, adjust=False).mean()
        
        # MACD
        df["MACD"] = df["EMA_12"] - df["EMA_26"]
        df["MACD_signal"] = df["MACD"].ewm(span=9, adjust=False).mean()
        df["MACD_histogram"] = df["MACD"] - df["MACD_signal"]
        
        # RSI (Relative Strength Index)
        delta = df["close"].diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
        rs = gain / loss
        df["RSI_14"] = 100 - (100 / (1 + rs))
        
        # Bollinger Bands
        df["BB_middle"] = df["close"].rolling(window=20).mean()
        bb_std = df["close"].rolling(window=20).std()
        df["BB_upper"] = df["BB_middle"] + (bb_std * 2)
        df["BB_lower"] = df["BB_middle"] - (bb_std * 2)
        df["BB_width"] = (df["BB_upper"] - df["BB_lower"]) / df["BB_middle"]
        df["BB_position"] = (df["close"] - df["BB_lower"]) / (df["BB_upper"] - df["BB_lower"])
        
        # ATR (Average True Range)
        high_low = df["high"] - df["low"]
        high_close = np.abs(df["high"] - df["close"].shift())
        low_close = np.abs(df["low"] - df["close"].shift())
        tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
        df["ATR_14"] = tr.rolling(window=14).mean()
        
        # Volume Profile Features
        df["volume_SMA_20"] = df["volume"].rolling(window=20).mean()
        df["volume_ratio"] = df["volume"] / df["volume_SMA_20"]
        
        # Momentum
        df["momentum_5"] = df["close"] / df["close"].shift(5) - 1
        df["momentum_10"] = df["close"] / df["close"].shift(10) - 1
        
        # Returns
        df["returns"] = df["close"].pct_change()
        df["log_returns"] = np.log(df["close"] / df["close"].shift(1))
        df["volatility_20"] = df["returns"].rolling(window=20).std()
        
        return df


ตัวอย่างการใช้งาน

if __name__ == "__main__":
    # สมมติว่ามี trades_df จาก TardisDataFetcher
    # trades_df = fetcher.get_trades("binance", "BTCUSDT", start_date, end_date)
    
    # สร้าง Features
    engineer = TradingFeatureEngineer(trades_df)
    
    # แปลงเป็น Candlestick 15 นาที
    candles = engineer.resample_to_candles("15T")
    
    # เพิ่ม Technical Indicators
    features_df = engineer.add_technical_indicators()
    
    print("=== Sample Features ===")
    print(features_df.tail(10))
    print(f"\nจำนวน Features: {len(features_df.columns)}")
    print(f"Features ที่สร้าง: {list(features_df.columns)}")

สร้างระบบ Backtesting พื้นฐาน

หลังจากมีข้อมูลและ Features แล้ว มาสร้างระบบ Backtesting อย่างง่ายสำหรับทดสอบกลยุทธ์

import pandas as pd
import numpy as np
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class Trade:
    """โครงสร้างข้อมูลการเทรด"""
    entry_time: pd.Timestamp
    exit_time: pd.Timestamp
    side: str  # 'long' หรือ 'short'
    entry_price: float
    exit_price: float
    size: float
    pnl: float
    pnl_pct: float

class SimpleBacktester:
    """ระบบ Backtesting แบบง่ายสำหรับ Mean Reversion Strategy"""
    
    def __init__(self, initial_capital: float = 10000, commission: float = 0.0004):
        """
        Args:
            initial_capital: ทุนเริ่มต้น (USD)
            commission: ค่าคอมมิชชันต่อครั้ง (0.04% = 4 bps)
        """
        self.initial_capital = initial_capital
        self.commission = commission
        self.trades: List[Trade] = []
        self.equity_curve = []
    
    def run_mean_reversion(
        self,
        df: pd.DataFrame,
        entry_threshold: float = -1.5,
        exit_threshold: float = 0.5,
        lookback: int = 20,
        stop_loss: float = 0.02
    ):
        """
        รัน Mean Reversion Strategy
        
        Strategy Logic:
        - เข้า Long เมื่อ RSI < entry_threshold (Oversold)
        - ออกเมื่อ RSI > exit_threshold หรือ ถึง Stop Loss
        """
        position = None
        entry_price = 0
        entry_time = None
        size = 0
        
        equity = self.initial_capital
        
        for i in range(lookback, len(df)):
            row = df.iloc[i]
            prev_rows = df.iloc[i-lookback:i]
            
            # คำนวณ Position Size
            if position is None:
                size = equity * 0.95  # ใช้ 95% ของ Equity
            else:
                size = equity  # Maintain Position Size
            
            # Entry Signal: RSI Oversold
            if position is None and row["RSI_14"] < entry_threshold:
                position = "long"
                entry_price = row["close"]
                entry_time = row["datetime"]
                
                # คำนวณค่าคอมมิชชัน
                equity -= size * self.commission
            
            # Exit Conditions
            elif position == "long":
                pnl_pct = (row["close"] - entry_price) / entry_price
                
                # Stop Loss
                if pnl_pct <= -stop_loss:
                    exit_price = entry_price * (1 - stop_loss)
                    self._close_trade(entry_time, row["datetime"], "long", 
                                     entry_price, exit_price, size)
                    equity = size * (1 + pnl_pct) - (size * self.commission)
                    position = None
                
                # Take Profit: RSI Overbought หรือ Mean Reversion
                elif row["RSI_14"] > exit_threshold or pnl_pct > -pnl_pct * 2:
                    exit_price = row["close"]
                    self._close_trade(entry_time, row["datetime"], "long",
                                     entry_price, exit_price, size)
                    equity = size * (1 + pnl_pct) - (size * self.commission)
                    position = None
            
            self.equity_curve.append({
                "datetime": row["datetime"],
                "equity": equity,
                "position": position
            })
        
        return self._generate_report()
    
    def _close_trade(self, entry_time, exit_time, side, entry, exit, size):
        """บันทึกการเทรด"""
        pnl_pct = (exit - entry) / entry if side == "long" else (entry - exit) / entry
        pnl = size * pnl_pct
        
        trade = Trade(
            entry_time=entry_time,
            exit_time=exit_time,
            side=side,
            entry_price=entry,
            exit_price=exit,
            size=size,
            pnl=pnl,
            pnl_pct=pnl_pct
        )
        self.trades.append(trade)
    
    def _generate_report(self) -> dict:
        """สร้างรายงานผล Backtest"""
        if not self.trades:
            return {"error": "No trades executed"}
        
        equity_df = pd.DataFrame(self.equity_curve)
        trades_df = pd.DataFrame([vars(t) for t in self.trades])
        
        winning_trades = trades_df[trades_df["pnl"] > 0]
        losing_trades = trades_df[trades_df["pnl"] <= 0]
        
        final_equity = equity_df["equity"].iloc[-1]
        total_return = (final_equity - self.initial_capital) / self.initial_capital
        
        # Calculate Sharpe Ratio
        returns = equity_df["equity"].pct_change().dropna()
        sharpe = returns.mean() / returns.std() * np.sqrt(252 * 390)  # แท่งเทียน 1 นาที
        
        # Max Drawdown
        cumulative = equity_df["equity"]
        running_max = cumulative.expanding().max()
        drawdown = (cumulative - running_max) / running_max
        max_drawdown = drawdown.min()
        
        return {
            "initial_capital": self.initial_capital,
            "final_equity": final_equity,
            "total_return": total_return,
            "total_trades": len(self.trades),
            "winning_trades": len(winning_trades),
            "losing_trades": len(losing_trades),
            "win_rate": len(winning_trades) / len(self.trades) * 100,
            "avg_win": winning_trades["pnl"].mean() if len(winning_trades) > 0 else 0,
            "avg_loss": losing_trades["pnl"].mean() if len(losing_trades) > 0 else 0,
            "profit_factor": (
                abs(winning_trades["pnl"].sum() / losing_trades["pnl"].sum()) 
                if len(losing_trades) > 0 and losing_trades["pnl"].sum() != 0 
                else float("inf")
            ),
            "sharpe_ratio": sharpe,
            "max_drawdown": max_drawdown,
            "equity_curve": equity_df
        }


ตัวอย่างการใช้งาน

if __name__ == "__main__":
    # 假设มี features_df จากตัวอย่างก่อนหน้า
    backtester = SimpleBacktester(initial_capital=10000)
    
    results = backtester.run_mean_reversion(
        features_df,
        entry_threshold=30,
        exit_threshold=50,
        lookback=20,
        stop_loss=0.02
    )
    
    print("=== Backtest Results ===")
    print(f"Initial Capital: ${results['initial_capital']:,.2f}")
    print(f"Final Equity: ${results['final_equity']:,.2f}")
    print(f"Total Return: {results['total_return']*100:.2f}%")
    print(f"Total Trades: {results['total_trades']}")
    print(f"Win Rate: {results['win_rate']:.2f}%")
    print(f"Profit Factor: {results['profit_factor']:.2f}")
    print(f"Sharpe Ratio: {results['sharpe_ratio']:.3f}")
    print(f"Max Drawdown: {results['max_drawdown']*100:.2f}%")

เหมาะกับใคร / ไม่เหมาะกับใคร

เหมาะกับ