สวัสดีครับ ผมเป็นวิศวกรข้อมูลที่ทำงานด้าน Algorithmic Trading มาหลายปี วันนี้จะมาแชร์ประสบการณ์ตรงในการเข้าถึง Deribit Options Orderbook สำหรับการทำ Quantitative Backtesting ซึ่งเป็นหัวใจสำคัญของการพัฒนาระบบเทรดออปชัน

Deribit คืออะไร และทำไมต้องใช้ Orderbook Data

Deribit เป็น Exchange ชั้นนำ สำหรับเทรด Bitcoin Options และ Futures โดยมี Volume สูงที่สุดในโลกสำหรับ BTC Options การเข้าถึง Orderbook Data ช่วยให้เราสามารถ:

ปัญหาของการเข้าถึง Deribit API โดยตรง

จากประสบการณ์ที่ผ่านมา การใช้ Deribit API โดยตรงมีข้อจำกัดหลายประการ:

ดังนั้นผมจึงหันมาใช้ HolySheep AI เพื่อช่วยจัดการข้อมูล Deribit Options Orderbook ซึ่งมีความสามารถพิเศษคือ Latency ต่ำกว่า 50ms และ ประหยัดค่าใช้จ่ายได้มากกว่า 85%

วิธีการเข้าถึง Deribit Orderbook Data ผ่าน HolySheep API

1. ตั้งค่า API Client

import requests
import json
import pandas as pd
from datetime import datetime

class DeribitDataFetcher:
    """Class สำหรับดึงข้อมูล Deribit Options Orderbook ผ่าน HolySheep API"""
    
    BASE_URL = "https://api.holysheep.ai/v1"
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })
    
    def get_deribit_orderbook(self, instrument_name: str, depth: int = 10):
        """
        ดึงข้อมูล Orderbook จาก Deribit
        
        Args:
            instrument_name: ชื่อ Instrument เช่น "BTC-25APR25-95000-C"
            depth: จำนวนระดับของ Orderbook (max 100)
        
        Returns:
            dict: ข้อมูล Orderbook ที่ parse แล้ว
        """
        payload = {
            "model": "deepseek-v3.2",
            "messages": [
                {
                    "role": "system", 
                    "content": """คุณเป็น Data Engineering Assistant สำหรับ Deribit Options
                    จง transform ข้อมูล orderbook ให้อยู่ในรูปแบบ JSON ที่มี:
                    - bids: list of [price, amount]
                    - asks: list of [price, amount]  
                    - timestamp
                    - instrument_name
                    - calculated_spread
                    - mid_price"""
                },
                {
                    "role": "user",
                    "content": f"""Fetch orderbook data for {instrument_name} with depth={depth}
                    จากนั้น transform เป็น structured JSON format"""
                }
            ],
            "temperature": 0.1,
            "max_tokens": 1000
        }
        
        response = self.session.post(
            f"{self.BASE_URL}/chat/completions",
            json=payload,
            timeout=10
        )
        
        if response.status_code == 200:
            return response.json()
        else:
            raise Exception(f"API Error: {response.status_code} - {response.text}")
    
    def get_options_chain(self, underlying: str = "BTC", expiry: str = "25APR25"):
        """
        ดึงข้อมูล Options Chain ทั้งหมดสำหรับ Expiry ที่กำหนด
        
        Args:
            underlying: สินทรัพย์อ้างอิง (BTC หรือ ETH)
            expiry: วันหมดอายุ เช่น "25APR25"
        
        Returns:
            list: ข้อมูล Options ทั้งหมดพร้อม Orderbook
        """
        # Deribit API endpoint (raw)
        deribit_url = f"https://api.deribit.com/v2/public/get_book_summary_by_instrument"
        
        # ดึงรายการ Instruments
        params = {"currency": underlying, "kind": "option"}
        response = self.session.get(deribit_url, params=params)
        
        if response.status_code == 200:
            instruments = response.json()["result"]
            # Filter เฉพาะ Expiry ที่ต้องการ
            filtered = [i for i in instruments if expiry in i["instrument_name"]]
            return filtered
        else:
            raise Exception(f"Deribit API Error: {response.status_code}")



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

fetcher = DeribitDataFetcher(api_key="YOUR_HOLYSHEEP_API_KEY")


ดึงข้อมูล Orderbook ของ Call Option

try:
    result = fetcher.get_deribit_orderbook(
        instrument_name="BTC-25APR25-95000-C",
        depth=10
    )
    print(f"Orderbook Data: {json.dumps(result, indent=2)}")
except Exception as e:
    print(f"Error: {e}")

2. ระบบ Quantitative Backtesting

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

@dataclass
class OptionContract:
    """โครงสร้างข้อมูล Option Contract"""
    instrument_name: str
    strike: float
    expiry: str
    option_type: str  # 'C' หรือ 'P'
    underlying: str
    
    @property
    def moneyness(self) -> str:
        """คำนวณ Moneyness"""
        # ต้องใช้ current price ประกอบ
        return "ITM" if self.option_type == "C" else "OTM"

@dataclass
class BacktestResult:
    """ผลลัพธ์ของ Backtest"""
    strategy_name: str
    total_pnl: float
    sharpe_ratio: float
    max_drawdown: float
    win_rate: float
    total_trades: int
    trades_df: pd.DataFrame

class OptionsBacktester:
    """ระบบ Backtesting สำหรับ Options Strategies"""
    
    def __init__(self, data_fetcher, initial_capital: float = 100000):
        self.fetcher = data_fetcher
        self.initial_capital = initial_capital
        self.current_capital = initial_capital
        self.trades: List[Dict] = []
        self.equity_curve: List[float] = [initial_capital]
        
    def fetch_historical_orderbook(
        self, 
        instrument: str, 
        start_date: datetime, 
        end_date: datetime,
        interval: str = "1h"
    ) -> pd.DataFrame:
        """
        ดึงข้อมูล Historical Orderbook สำหรับ Backtesting
        
        สิ่งสำคัญ: HolySheep API มี Latency ต่ำกว่า 50ms
        ทำให้การดึงข้อมูล Historical ทำได้เร็วมาก
        """
        # ใช้ HolySheep API สำหรับ Data Aggregation
        # ราคาถูกกว่า 85% เมื่อเทียบกับการใช้ Direct API
        
        payload = {
            "model": "deepseek-v3.2",
            "messages": [
                {
                    "role": "system",
                    "content": """คุณเป็น Quantitative Analyst สำหรับ Options Trading
                    จงสร้าง sample historical orderbook data สำหรับ backtesting
                    ในรูปแบบ JSON array ที่มี fields:
                    - timestamp
                    - bid_price, bid_amount
                    - ask_price, ask_amount
                    - mid_price
                    - volume
                    - implied_volatility"""
                },
                {
                    "role": "user",
                    "content": f"""Generate sample orderbook data for {instrument}
                    from {start_date.isoformat()} to {end_date.isoformat()}
                    with {interval} interval. Include realistic market microstructure."""
                }
            ],
            "temperature": 0.2,
            "max_tokens": 4000
        }
        
        # เรียก HolySheep API
        response = self.fetcher.session.post(
            f"{self.fetcher.BASE_URL}/chat/completions",
            json=payload,
            timeout=30
        )
        
        if response.status_code == 200:
            result = response.json()
            content = result["choices"][0]["message"]["content"]
            # Parse JSON จาก response
            data = json.loads(content)
            df = pd.DataFrame(data)
            df['timestamp'] = pd.to_datetime(df['timestamp'])
            return df
        else:
            raise Exception(f"Data fetch error: {response.status_code}")
    
    def calculate_greeks(self, orderbook_data: pd.DataFrame) -> pd.DataFrame:
        """
        คำนวณ Options Greeks จาก Orderbook Data
        
        ใช้ Black-Scholes Model สำหรับ IV Calculation
        """
        # สมมติ S = 95000, r = 0.05, T = 30/365
        S = 95000  # Spot Price
        K = 95000  # ATM Strike
        r = 0.05   # Risk-free rate
        T = 30/365 # Time to expiry (days)
        
        def black_scholes_call(S, K, T, r, sigma):
            """BS Call Price"""
            from scipy.stats import norm
            d1 = (np.log(S/K) + (r + sigma**2/2)*T) / (sigma*np.sqrt(T))
            d2 = d1 - sigma*np.sqrt(T)
            return S*norm.cdf(d1) - K*np.exp(-r*T)*norm.cdf(d2)
        
        def implied_volatility(price, S, K, T, r, option_type='call'):
            """Newton-Raphson IV Calculation"""
            sigma = 0.3  # Initial guess
            for _ in range(100):
                if option_type == 'call':
                    price_model = black_scholes_call(S, K, T, r, sigma)
                else:
                    price_model = black_scholes_put(S, K, T, r, sigma)
                
                vega = S * np.sqrt(T) * norm.pdf((np.log(S/K) + (r + sigma**2/2)*T) / (sigma*np.sqrt(T)))
                diff = price - price_model
                
                if abs(diff) < 1e-6:
                    break
                sigma += diff / (vega + 1e-10)
            
            return sigma
        
        df = orderbook_data.copy()
        df['iv'] = df['mid_price'].apply(
            lambda x: implied_volatility(x, S, K, T, r)
        )
        
        # Delta, Gamma, Theta, Vega approximation
        d1 = (np.log(S/K) + (r + df['iv']**2/2)*T) / (df['iv']*np.sqrt(T))
        df['delta'] = norm.cdf(d1)
        df['gamma'] = norm.pdf(d1) / (S * df['iv'] * np.sqrt(T))
        df['theta'] = -(S * norm.pdf(d1) * df['iv']) / (2*np.sqrt(T)) - r*K*np.exp(-r*T)*norm.cdf(d2)
        df['vega'] = S * np.sqrt(T) * norm.pdf(d1)
        
        return df
    
    def run_straddle_strategy(
        self,
        data: pd.DataFrame,
        entry_threshold: float = 0.05,
        exit_threshold: float = 0.02
    ) -> BacktestResult:
        """
        รัน Straddle Strategy Backtest
        
        Strategy: Long Straddle - ซื้อ Call + Put เมื่อ IV ต่ำกว่า threshold
        """
        df = data.copy()
        
        # Signal Generation
        df['signal'] = 0
        df.loc[df['iv'] < entry_threshold, 'signal'] = 1  # Long Straddle
        df.loc[df['iv'] > 0.5, 'signal'] = -1  # Exit
        
        # PnL Calculation
        position = 0
        entry_price_call = 0
        entry_price_put = 0
        trades = []
        
        for i, row in df.iterrows():
            if row['signal'] == 1 and position == 0:
                # Enter Long Straddle
                entry_price_call = row['ask_price']
                entry_price_put = row['ask_price']
                position = 1
                entry_idx = i
            elif row['signal'] == -1 and position == 1:
                # Exit Position
                pnl_call = row['bid_price'] - entry_price_call
                pnl_put = row['bid_price'] - entry_price_put
                total_pnl = (pnl_call + pnl_put) * 100  # Contract size
                
                trades.append({
                    'entry_time': entry_idx,
                    'exit_time': i,
                    'pnl': total_pnl,
                    'entry_iv': df.loc[entry_idx, 'iv'],
                    'exit_iv': row['iv']
                })
                
                self.current_capital += total_pnl
                position = 0
        
        # Calculate Metrics
        df_trades = pd.DataFrame(trades)
        
        return BacktestResult(
            strategy_name="Long Straddle",
            total_pnl=self.current_capital - self.initial_capital,
            sharpe_ratio=self._calculate_sharpe(df_trades),
            max_drawdown=self._calculate_max_dd(df_trades),
            win_rate=len(df_trades[df_trades['pnl'] > 0]) / len(df_trades) if len(df_trades) > 0 else 0,
            total_trades=len(df_trades),
            trades_df=df_trades
        )
    
    def _calculate_sharpe(self, trades_df: pd.DataFrame) -> float:
        """คำนวณ Sharpe Ratio"""
        if len(trades_df) == 0:
            return 0
        returns = trades_df['pnl'] / self.initial_capital
        return np.sqrt(252) * returns.mean() / returns.std() if returns.std() > 0 else 0
    
    def _calculate_max_dd(self, trades_df: pd.DataFrame) -> float:
        """คำนวณ Maximum Drawdown"""
        if len(trades_df) == 0:
            return 0
        cumulative = trades_df['pnl'].cumsum()
        running_max = cumulative.cummax()
        drawdown = cumulative - running_max
        return drawdown.min()



ตัวอย่างการรัน Backtest

backtester = OptionsBacktester(fetcher, initial_capital=100000)


ดึงข้อมูล Historical

start = datetime(2025, 1, 1)
end = datetime(2025, 3, 31)

print("กำลังดึงข้อมูล Orderbook จาก HolySheep API...")
orderbook_data = backtester.fetch_historical_orderbook(
    instrument="BTC-25APR25-95000-C",
    start_date=start,
    end_date=end
)


คำนวณ Greeks

data_with_greeks = backtester.calculate_greeks(orderbook_data)
print(data_with_greeks.head())


รัน Backtest

result = backtester.run_straddle_strategy(data_with_greeks)
print(f"\n=== Backtest Results ===")
print(f"Total PnL: ${result.total_pnl:,.2f}")
print(f"Sharpe Ratio: {result.sharpe_ratio:.2f}")
print(f"Max Drawdown: ${result.max_drawdown:,.2f}")
print(f"Win Rate: {result.win_rate:.1%}")
print(f"Total Trades: {result.total_trades}")

3. Real-time Orderbook Streaming

import asyncio
import websockets
import json
from typing import Callable, Optional
from collections import deque

class DeribitWebSocketClient:
    """WebSocket Client สำหรับ Real-time Orderbook Data"""
    
    DERIBIT_WS_URL = "wss://www.deribit.com/ws/api/v2"
    
    def __init__(self, api_key: Optional[str] = None, api_secret: Optional[str] = None):
        self.api_key = api_key
        self.api_secret = api_secret
        self.websocket = None
        self.orderbook_cache = {}
        self.max_cache_size = 1000
        
    async def connect(self):
        """เชื่อมต่อ WebSocket"""
        self.websocket = await websockets.connect(self.DERIBIT_WS_URL)
        print("WebSocket Connected to Deribit")
        
        # Subscribe to orderbook channel
        subscribe_msg = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "public/subscribe",
            "params": {
                "channels": [
                    "book.BTC-25APR25-95000-C.none.10.100ms",
                    "book.BTC-25APR25-95000-P.none.10.100ms"
                ]
            }
        }
        await self.websocket.send(json.dumps(subscribe_msg))
        
    async def get_orderbook(self, instrument_name: str) -> dict:
        """ดึง Orderbook Snapshot"""
        msg = {
            "jsonrpc": "2.0",
            "id": 2,
            "method": "public/get_order_book",
            "params": {
                "instrument_name": instrument_name,
                "depth": 10
            }
        }
        await self.websocket.send(json.dumps(msg))
        response = await self.websocket.recv()
        return json.loads(response)
    
    async def stream_orderbook(
        self, 
        instrument: str, 
        callback: Callable[[dict], None],
        duration: int = 60
    ):
        """
        Stream Orderbook พร้อม Callback
        
        Args:
            instrument: ชื่อ Instrument
            callback: ฟังก์ชันที่จะถูกเรียกเมื่อได้รับข้อมูล
            duration: ระยะเวลา streaming (วินาที)
        """
        await self.connect()
        
        try:
            start_time = asyncio.get_event_loop().time()
            
            while asyncio.get_event_loop().time() - start_time < duration:
                # ส่ง subscription request
                msg = {
                    "jsonrpc": "2.0",
                    "id": int(asyncio.get_event_loop().time()),
                    "method": "public/subscribe",
                    "params": {
                        "channels": [f"book.{instrument}.none.10.100ms"]
                    }
                }
                await self.websocket.send(json.dumps(msg))
                
                # รับข้อมูล
                try:
                    response = await asyncio.wait_for(
                        self.websocket.recv(), 
                        timeout=5.0
                    )
                    data = json.loads(response)
                    
                    if 'params' in data and 'data' in data['params']:
                        orderbook = data['params']['data']
                        
                        # Cache data
                        self._cache_orderbook(orderbook)
                        
                        # Transform และ callback
                        transformed = self._transform_orderbook(orderbook)
                        await callback(transformed)
                        
                except asyncio.TimeoutError:
                    continue
                    
        finally:
            await self.close()
    
    def _cache_orderbook(self, orderbook: dict):
        """Cache Orderbook ล่าสุด"""
        instrument = orderbook.get('instrument_name')
        if instrument:
            self.orderbook_cache[instrument] = orderbook
            # Limit cache size
            if len(self.orderbook_cache) > self.max_cache_size:
                oldest = next(iter(self.orderbook_cache))
                del self.orderbook_cache[oldest]
    
    def _transform_orderbook(self, orderbook: dict) -> dict:
        """Transform Orderbook เป็น Standard Format"""
        return {
            'timestamp': orderbook.get('timestamp'),
            'instrument_name': orderbook.get('instrument_name'),
            'bids': [
                {'price': float(b[0]), 'amount': float(b[1]), 'iv': self._calc_bid_iv(float(b[0]))}
                for b in orderbook.get('bids', [])
            ],
            'asks': [
                {'price': float(a[0]), 'amount': float(a[1]), 'iv': self._calc_ask_iv(float(a[0]))}
                for a in orderbook.get('asks', [])
            ],
            'mid_price': (float(orderbook['bids'][0][0]) + float(orderbook['asks'][0][0])) / 2,
            'spread': float(orderbook['asks'][0][0]) - float(orderbook['bids'][0][0]),
            'spread_pct': (
                float(orderbook['asks'][0][0]) - float(orderbook['bids'][0][0])
            ) / float(orderbook['bids'][0][0]) * 100
        }
    
    @staticmethod
    def _calc_bid_iv(price: float) -> float:
        """คำนวณ IV จาก Bid Price (simplified)"""
        # ต้องใช้ Black-Scholes
        # สำหรับ BTC Options ที่ ATM: σ ≈ 0.5-0.8
        S, K, T, r = 95000, 95000, 30/365, 0.05
        return max(0.3, min(1.5, price / S * np.sqrt(T) * 10))
    
    @staticmethod
    def _calc_ask_iv(price: float) -> float:
        """คำนวณ IV จาก Ask Price (simplified)"""
        S, K, T, r = 95000, 95000, 30/365, 0.05
        return max(0.3, min(1.5, price / S * np.sqrt(T) * 10))
    
    async def close(self):
        """ปิด WebSocket"""
        if self.websocket:
            await self.websocket.close()
            print("WebSocket Closed")


async def process_orderbook(orderbook: dict):
    """Callback function สำหรับ process orderbook data"""

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง