Tardis Data สำหรับ Bybit: คู่มือ Quantitative Backtesting ด้วย Trades, Quotes และ Funding Rates

การทำ Quantitative Trading ที่ประสบความสำเร็จต้องอาศัยข้อมูลที่แม่นยำและครอบคลุม โดยเฉพาะตลาด Crypto ที่มีความผันผวนสูงอย่าง Bybit การใช้ Tardis Data ร่วมกับ AI อย่าง HolySheep AI จะช่วยให้คุณวิเคราะห์ข้อมูลย้อนหลัง (Backtest) ได้อย่างมีประสิทธิภาพ ในบทความนี้เราจะสอนการดึงข้อมูล Trades, Quotes และ Funding Rates จาก Bybit ผ่าน Tardis API พร้อมโค้ด Python ที่พร้อมใช้งานจริง

Tardis Data คืออะไร และทำไมต้องใช้กับ Bybit

Tardis เป็นบริการรวบรวมข้อมูลตลาด Crypto คุณภาพสูงระดับ Enterprise ที่รวบรวม Raw Market Data จาก Exchange ชั้นนำรวมถึง Bybit ข้อมูลที่ได้ประกอบด้วย:

• Trades: ข้อมูลการซื้อขายที่เกิดขึ้นจริง ราคา ปริมาณ และ timestamp
• Quotes (Orderbook): ข้อมูลราคา Bid/Ask ที่รอการซื้อขาย
• Funding Rates: อัตราดอกเบี้ยราย 8 ชั่วโมงของสัญญา Perpetual
• Candles/OHLCV: ข้อมูลแท่งเทียนที่รวม Open, High, Low, Close, Volume

การติดตั้งและเริ่มต้นใช้งาน Tardis API

# ติดตั้ง dependencies ที่จำเป็น
pip install tardis-client pandas numpy requests

หรือใช้ poetry
poetry add tardis-client pandas numpy requests

# นำเข้า libraries และตั้งค่า API
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import requests

ตั้งค่า Tardis API Key (สมัครได้ที่ https://tardis.dev)
TARDIS_API_KEY = "your_tardis_api_key"

ฟังก์ชันดึงข้อมูล Trades จาก Bybit
def get_bybit_trades(symbol="BTCUSDT", start_date=None, end_date=None, limit=100000):
    """
    ดึงข้อมูล Trades จาก Bybit ผ่าน Tardis API
    symbol: สัญลักษณ์คู่เทรด เช่น BTCUSDT, ETHUSDT
    start_date: วันที่เริ่มต้น (datetime object)
    end_date: วันที่สิ้นสุด (datetime object)
    limit: จำนวน records สูงสุดที่ต้องการ
    """
    if start_date is None:
        start_date = datetime.utcnow() - timedelta(days=1)
    if end_date is None:
        end_date = datetime.utcnow()
    
    # API Endpoint สำหรับ Bybit historical data
    url = f"https://api.tardis.dev/v1/derivatives/bybit/trades"
    
    params = {
        "symbol": symbol,
        "from": start_date.isoformat(),
        "to": end_date.isoformat(),
        "limit": limit,
        "api_key": TARDIS_API_KEY
    }
    
    response = requests.get(url, params=params)
    response.raise_for_status()
    
    data = response.json()
    
    # แปลงเป็น DataFrame
    df = pd.DataFrame(data)
    
    # ปรับ format timestamp
    if "timestamp" in df.columns:
        df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
    
    return df

ฟังก์ชันดึงข้อมูล Quotes/Orderbook
def get_bybit_quotes(symbol="BTCUSDT", start_date=None, end_date=None):
    """
    ดึงข้อมูล Quotes (Orderbook) จาก Bybit ผ่าน Tardis API
    ข้อมูลนี้สำคัญสำหรับการวิเคราะห์ Spread และ Liquidity
    """
    if start_date is None:
        start_date = datetime.utcnow() - timedelta(hours=1)
    if end_date is None:
        end_date = datetime.utcnow()
    
    url = f"https://api.tardis.dev/v1/derivatives/bybit/quotes"
    
    params = {
        "symbol": symbol,
        "from": start_date.isoformat(),
        "to": end_date.isoformat(),
        "api_key": TARDIS_API_KEY
    }
    
    response = requests.get(url, params=params)
    response.raise_for_status()
    
    data = response.json()
    df = pd.DataFrame(data)
    
    if "timestamp" in df.columns:
        df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
    
    return df

ฟังก์ชันดึงข้อมูล Funding Rates
def get_bybit_funding_rates(symbol="BTCUSDT", days=30):
    """
    ดึงข้อมูล Funding Rates จาก Bybit
    ข้อมูลนี้สำคัญสำหรับการวิเคราะห์ Sentiment และความต้องการถือ Position
    """
    end_date = datetime.utcnow()
    start_date = end_date - timedelta(days=days)
    
    url = f"https://api.tardis.dev/v1/derivatives/bybit/funding-rates"
    
    params = {
        "symbol": symbol,
        "from": start_date.isoformat(),
        "to": end_date.isoformat(),
        "api_key": TARDIS_API_KEY
    }
    
    response = requests.get(url, params=params)
    response.raise_for_status()
    
    data = response.json()
    df = pd.DataFrame(data)
    
    if "timestamp" in df.columns:
        df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
    
    return df

ตัวอย่างการใช้งาน
if __name__ == "__main__":
    # ดึงข้อมูล BTCUSDT ย้อนหลัง 7 วัน
    trades_df = get_bybit_trades(
        symbol="BTCUSDT",
        start_date=datetime.utcnow() - timedelta(days=7),
        end_date=datetime.utcnow(),
        limit=500000
    )
    
    print(f"ดึงข้อมูลสำเร็จ: {len(trades_df)} trades")
    print(trades_df.head())

สร้างระบบ Quantitative Backtest พื้นฐาน

หลังจากได้ข้อมูล Trades, Quotes และ Funding Rates แล้ว ต่อไปจะเป็นการสร้างระบบ Backtest เบื้องต้นที่ใช้งานได้จริง

import pandas as pd
import numpy as np
from typing import List, Dict, Tuple
from dataclasses import dataclass
from datetime import datetime

@dataclass
class Trade:
    """โครงสร้างข้อมูลสำหรับการซื้อขายในระบบ Backtest"""
    timestamp: datetime
    side: str  # 'buy' or 'sell'
    price: float
    volume: float
    fee: float = 0.0004  # Bybit spot fee พื้นฐาน

@dataclass
class Position:
    """โครงสร้างข้อมูล Position ปัจจุบัน"""
    entry_price: float
    quantity: float
    side: str  # 'long' or 'short'
    entry_time: datetime

class BybitBacktester:
    """
    ระบบ Backtest สำหรับ Bybit Perpetual Futures
    รองรับการวิเคราะห์ด้วย Trades, Quotes และ Funding Rates
    """
    
    def __init__(self, initial_capital: float = 10000.0, fee_rate: float = 0.0004):
        self.initial_capital = initial_capital
        self.capital = initial_capital
        self.fee_rate = fee_rate
        self.position: Position = None
        self.trades_log: List[Dict] = []
        self.equity_curve: List[float] = []
        
        # ข้อมูลสำหรับวิเคราะห์
        self.trades_df = None
        self.quotes_df = None
        self.funding_df = None
        
    def load_data(self, 
                   trades_df: pd.DataFrame,
                   quotes_df: pd.DataFrame = None,
                   funding_df: pd.DataFrame = None):
        """โหลดข้อมูลเข้าระบบ Backtest"""
        self.trades_df = trades_df.copy()
        self.quotes_df = quotes_df.copy() if quotes_df is not None else None
        self.funding_df = funding_df.copy() if funding_df is not None else None
        
        # เพิ่มคอลัมน์ timestamp ถ้ายังไม่มี
        if "timestamp" in self.trades_df.columns:
            self.trades_df.set_index("timestamp", inplace=True)
            
        print(f"โหลดข้อมูลเสร็จสมบูรณ์:")
        print(f"  - Trades: {len(self.trades_df)} records")
        print(f"  - Time range: {self.trades_df.index.min()} ถึง {self.trades_df.index.max()}")
        
    def calculate_spread(self) -> pd.Series:
        """คำนวณ Spread จากข้อมูล Quotes"""
        if self.quotes_df is None or "bidPrice" not in self.quotes_df.columns:
            return None
            
        spread = self.quotes_df["askPrice"] - self.quotes_df["bidPrice"]
        spread_pct = (spread / self.quotes_df["askPrice"]) * 100
        
        return spread_pct
        
    def calculate_volatility(self, window: int = 20) -> pd.Series:
        """คำนวณ Volatility จากข้อมูล Trades"""
        if self.trades_df is None or "price" not in self.trades_df.columns:
            return None
            
        returns = self.trades_df["price"].pct_change()
        volatility = returns.rolling(window=window).std() * np.sqrt(365 * 24)
        
        return volatility
        
    def open_position(self, side: str, price: float, quantity: float, timestamp: datetime):
        """เปิด Position ใหม่"""
        # คำนวณค่าธรรมเนียม
        fee = price * quantity * self.fee_rate
        self.capital -= fee
        
        self.position = Position(
            entry_price=price,
            quantity=quantity,
            side=side,
            entry_time=timestamp
        )
        
        self.trades_log.append({
            "timestamp": timestamp,
            "action": "OPEN",
            "side": side,
            "price": price,
            "quantity": quantity,
            "fee": fee,
            "capital": self.capital
        })
        
    def close_position(self, price: float, timestamp: datetime):
        """ปิด Position ปัจจุบัน"""
        if self.position is None:
            return
            
        # คำนวณ PnL
        if self.position.side == "long":
            pnl = (price - self.position.entry_price) * self.position.quantity
        else:
            pnl = (self.position.entry_price - price) * self.position.quantity
            
        # ค่าธรรมเนียมการปิด
        fee = price * self.position.quantity * self.fee_rate
        
        self.capital += pnl - fee
        
        self.trades_log.append({
            "timestamp": timestamp,
            "action": "CLOSE",
            "side": self.position.side,
            "entry_price": self.position.entry_price,
            "exit_price": price,
            "quantity": self.position.quantity,
            "pnl": pnl,
            "fee": fee,
            "capital": self.capital
        })
        
        self.position = None
        
    def apply_funding(self, funding_rate: float, timestamp: datetime):
        """หัก/รับ Funding Rate ราย 8 ชั่วโมง"""
        if self.position is None:
            return
            
        # Long position จ่าย funding ถ้า funding_rate > 0
        # Short position รับ funding ถ้า funding_rate > 0
        if self.position.side == "long":
            funding_cost = self.position.entry_price * self.position.quantity * funding_rate
            self.capital -= funding_cost
        else:
            funding_cost = self.position.entry_price * self.position.quantity * funding_rate
            self.capital += funding_cost
            
        self.trades_log.append({
            "timestamp": timestamp,
            "action": "FUNDING",
            "funding_rate": funding_rate,
            "capital": self.capital
        })
        
    def get_performance_summary(self) -> Dict:
        """สรุปผลการ Backtest"""
        if not self.trades_log:
            return {}
            
        df = pd.DataFrame(self.trades_log)
        
        # กรองเฉพาะ Close trades
        closed_trades = df[df["action"] == "CLOSE"]
        
        if len(closed_trades) == 0:
            return {"message": "ไม่มีการปิด Position ในช่วง Backtest"}
            
        total_pnl = closed_trades["pnl"].sum()
        win_rate = (closed_trades["pnl"] > 0).sum() / len(closed_trades)
        avg_win = closed_trades[closed_trades["pnl"] > 0]["pnl"].mean()
        avg_loss = closed_trades[closed_trades["pnl"] < 0]["pnl"].mean()
        
        # คำนวณ Max Drawdown
        df["equity"] = df["capital"].fillna(method="ffill")
        rolling_max = df["equity"].expanding().max()
        drawdown = (df["equity"] - rolling_max) / rolling_max
        max_drawdown = drawdown.min()
        
        # Sharpe Ratio
        returns = closed_trades["pnl"] / self.initial_capital
        sharpe = returns.mean() / returns.std() * np.sqrt(252) if returns.std() > 0 else 0
        
        return {
            "initial_capital": self.initial_capital,
            "final_capital": self.capital,
            "total_pnl": total_pnl,
            "total_return": (self.capital - self.initial_capital) / self.initial_capital * 100,
            "total_trades": len(closed_trades),
            "win_rate": win_rate * 100,
            "avg_win": avg_win,
            "avg_loss": avg_loss,
            "profit_factor": abs(avg_win / avg_loss) if avg_loss != 0 else 0,
            "max_drawdown": max_drawdown * 100,
            "sharpe_ratio": sharpe,
            "max_drawdown_pct": max_drawdown * 100
        }

ตัวอย่างการใช้งาน Backtester
def run_sample_backtest():
    """ตัวอย่างการรัน Backtest ด้วยข้อมูลจริง"""
    
    # สร้าง Backtester
    backtester = BybitBacktester(initial_capital=10000.0, fee_rate=0.0004)
    
    # ดึงข้อมูล (ตัวอย่าง - ใช้ฟังก์ชันจากด้านบน)
    # trades_df = get_bybit_trades(symbol="BTCUSDT", days=7)
    # quotes_df = get_bybit_quotes(symbol="BTCUSDT", hours=1)
    # funding_df = get_bybit_funding_rates(symbol="BTCUSDT", days=7)
    
    # โหลดข้อมูล
    # backtester.load_data(trades_df, quotes_df, funding_df)
    
    # รัน Backtest (Logic ตัวอย่าง)
    # for idx, row in trades_df.iterrows():
    #     # เพิ่ม Logic การซื้อขายของคุณที่นี่
    #     pass
    
    # แสดงผล
    # summary = backtester.get_performance_summary()
    # print(summary)
    
    pass

print("ระบบ Backtest เบื้องต้นพร้อมใช้งาน!")

การใช้ AI วิเคราะห์ข้อมูล Backtest ด้วย HolySheep

หลังจากได้ผลลัพธ์ Backtest แล้ว การวิเคราะห์เชิงลึกต้องอาศัย AI ที่มีความสามารถในการประมวลผลข้อมูลจำนวนมากและให้คำแนะนำที่เป็นประโยชน์ HolySheep AI เป็นแพลตฟอร์ม AI API ที่รวดเร็วและประหยัด รองรับโมเดลหลากหลาย เหมาะสำหรับการวิเคราะห์ข้อมูล Quant

import requests
import json

การใช้ HolySheep AI สำหรับวิเคราะห์ผล Backtest
def analyze_backtest_results_with_ai(backtest_summary: dict, trades_log: list):
    """
    ใช้ HolySheep AI วิเคราะห์ผลลัพธ์ Backtest และให้คำแนะนำ
    """
    
    # สร้าง Prompt สำหรับ AI
    prompt = f"""
ข้อมูลผล Backtest
    
สรุปผลการทดสอบ
- เงินทุนเริ่มต้น: ${backtest_summary.get('initial_capital', 0):,.2f}
- เงินทุนสุดท้าย: ${backtest_summary.get('final_capital', 0):,.2f}
- กำไร/ขาดทุนรวม: ${backtest_summary.get('total_pnl', 0):,.2f} ({backtest_summary.get('total_return', 0):.2f}%)
- จำนวนการซื้อขาย: {backtest_summary.get('total_trades', 0)} ครั้ง
- Win Rate: {backtest_summary.get('win_rate', 0):.2f}%
- กำไรเฉลี่ยต่อการชนะ: ${backtest_summary.get('avg_win', 0):,.2f}
- ขาดทุนเฉลี่ยต่อการแพ้: ${backtest_summary.get('avg_loss', 0):,.2f}
- Profit Factor: {backtest_summary.get('profit_factor', 0):.2f}
- Max Drawdown: {backtest_summary.get('max_drawdown', 0):.2f}%
- Sharpe Ratio: {backtest_summary.get('sharpe_ratio', 0):.2f}

คำถาม
1. กลยุทธ์นี้มีประสิทธิภาพอย่างไร?
2. มีจุดอ่อนอะไรที่ควรปรับปรุง?
3. แนะนำการปรับปรุงอย่างไร?
4. ควร Risk Management อย่างไร?

รูปแบบคำตอบที่ต้องการ
ให้ตอบเป็นภาษาไทย พร้อมสรุปประเด็นสำคัญ 3-5 ข้อ และแนะนำเชิงปฏิบัติที่สามารถทำได้ทันที
"""
    
    # เรียกใช้ HolySheep API
    url = "https://api.holysheep.ai/v1/chat/completions"
    
    headers = {
        "Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": "gpt-4.1",  # โมเดลที่เหมาะสมสำหรับการวิเคราะห์ข้อมูล
        "messages": [
            {"role": "system", "content": "คุณเป็นผู้เชี่ยวชาญด้าน Quantitative Trading และ Crypto Markets"},
            {"role": "user", "content": prompt}
        ],
        "temperature": 0.7,
        "max_tokens": 2000
    }
    
    try:
        response = requests.post(url, headers=headers, json=payload, timeout=30)
        response.raise_for_status()
        
        result = response.json()
        ai_analysis = result["choices"][0]["message"]["content"]
        
        return {
            "success": True,
            "analysis": ai_analysis,
            "model_used": "gpt-4.1",
            "cost_estimate": "$0.008"  # ประมาณการค่าใช้จ่าย
        }
        
    except requests.exceptions.RequestException as e:
        return {
            "success": False,
            "error": str(e)
        }

ตัวอย่างการใช้งาน
if __name__ == "__main__":
    # ข้อมูลตัวอย่าง
    sample_summary = {
        "initial_capital": 10000.0,
        "final_capital": 12450.0,
        "total_pnl": 2450.0,
        "total_return": 24.5,
        "total_trades": 45,
        "win_rate": 62.2,
        "avg_win": 150.0,
        "avg_loss": -85.0,
        "profit_factor": 1.76,
        "max_drawdown": -8.5,
        "sharpe_ratio": 1.85
    }
    
    result = analyze_backtest_results_with_ai(sample_summary, [])
    
    if result["success"]:
        print("=" * 60)
        print("📊 ผลการวิเคราะห์จาก AI")
        print("=" * 60)
        print(result["analysis"])
        print("=" * 60)
        print(f"💰 ค่าใช้จ่ายโดยประมาณ: {result.get('cost_estimate', 'N/A')}")
        print(f"🤖 โมเดลที่ใช้: {result.get('model_used', 'N/A')}")

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

ราคาและ ROI