ในฐานะวิศวกรข้อมูลที่ทำงานด้าน DeFi research มาหลายปี ผมเคยเจอปัญหาเรื่องการเข้าถึงข้อมูล derivatives คุณภาพสูงสำหรับการวิเคราะห์ options chain และ funding rates อยู่บ่อยครั้ง วันนี้ผมจะมาแชร์ประสบการณ์การใช้งาน Tardis CSV dataset ร่วมกับ HolySheep AI ในการวิเคราะห์ข้อมูลสินทรัพย์อนุพันธ์แบบครบวงจร

Tardis CSV Dataset คืออะไร และทำไมต้องสนใจ

Tardis Finance เป็นแพลตฟอร์มที่รวบรวม historical market data ของ exchange หลายราย ครอบคลุม perpetual futures, options, spot และ funding rate data ในรูปแบบ CSV ที่พร้อมใช้งานทันที จุดเด่นคือความละเอียดถึงระดับ millisecond-level timestamps และการ normalize ข้อมูลให้อยู่ใน format เดียวกันทั้งหมด

สำหรับงานวิเคราะห์ derivatives ข้อมูลที่มีประโยชน์มาก ได้แก่:

การดาวน์โหลดและโครงสร้างข้อมูล

ผมจะแสดงวิธีการดึงข้อมูล options chain และ funding rate จาก Tardis โดยใช้ Python แบบ production-ready

import pandas as pd
from pathlib import Path
from datetime import datetime, timedelta
import asyncio
import aiohttp

class TardisDataLoader:
    """
    Production-grade data loader สำหรับ Tardis CSV datasets
    รองรับ options chain และ funding rate data
    """
    
    BASE_URL = "https://data.tardis.dev/v1/files"
    
    EXCHANGES = {
        'deribit': {
            'options': 'deribit/options',
            'funding': 'deribit/funding_rate'
        },
        'binance': {
            'perpetuals': 'binance/um/futures',
            'funding': 'binance/um/funding_rate'
        },
        'okx': {
            'options': 'okx/options',
            'funding': 'okx/funding_rate'
        }
    }
    
    def __init__(self, api_token: str):
        self.api_token = api_token
        self.session = None
        self._cache = {}
    
    async def __aenter__(self):
        timeout = aiohttp.ClientTimeout(total=300)
        self.session = aiohttp.ClientSession(timeout=timeout)
        return self
    
    async def __aexit__(self, *args):
        if self.session:
            await self.session.close()
    
    async def download_options_chain(
        self, 
        exchange: str, 
        symbol: str, 
        date: str
    ) -> pd.DataFrame:
        """
        ดึงข้อมูล options chain สำหรับ symbol ที่กำหนด
        
        Args:
            exchange: 'deribit', 'binance', 'okx'
            symbol: 'BTC', 'ETH', etc.
            date: 'YYYY-MM-DD'
        
        Returns:
            DataFrame พร้อม options chain data
        """
        url = f"{self.BASE_URL}/{self.EXCHANGES[exchange]['options']}/{symbol}-{date}.csv.gz"
        
        headers = {'Authorization': f'Bearer {self.api_token}'}
        
        async with self.session.get(url, headers=headers) as resp:
            if resp.status == 404:
                raise FileNotFoundError(f"No data for {symbol} on {date}")
            resp.raise_for_status()
            
            # อ่าน compressed CSV trực tiếp
            import io
            import gzip
            
            content = await resp.read()
            df = pd.read_csv(
                io.BytesIO(gzip.decompress(content)),
                compression='gzip'
            )
            
            return self._normalize_options_data(df, exchange)
    
    def _normalize_options_data(
        self, 
        df: pd.DataFrame, 
        exchange: str
    ) -> pd.DataFrame:
        """Normalize options data ให้อยู่ใน format มาตรฐาน"""
        
        column_mapping = {
            'deribit': {
                'instrument_name': 'symbol',
                'creation_timestamp': 'timestamp',
                'expiration_timestamp': 'expiry_timestamp',
                'open_interest': 'open_interest',
                'index_price': 'underlying_price',
                'mark_price': 'mark_price'
            },
            'binance': {
                'symbol': 'symbol',
                'timestamp': 'timestamp',
                'openInterest': 'open_interest',
                'markPrice': 'mark_price'
            }
        }
        
        mapping = column_mapping.get(exchange, {})
        df = df.rename(columns=mapping)
        
        # คำนวณ implied volatility จาก mark price
        if 'mark_price' in df.columns and 'underlying_price' in df.columns:
            df['iv_estimate'] = self._estimate_iv(df)
        
        return df
    
    def _estimate_iv(self, df: pd.DataFrame) -> pd.Series:
        """Black-Scholes implied volatility approximation"""
        import numpy as np
        
        # Simplified IV estimation สำหรับ ATM options
        T = (df.get('expiry_timestamp', 0) - df.get('timestamp', 0)) / (365 * 24 * 3600)
        T = T.replace(0, 0.01)  # กัน division by zero
        
        S = df['underlying_price']
        K = df.get('strike', S)  # ถ้าไม่มี strike ใช้ underlying เป็น ATM
        V = df['mark_price']
        
        # Simplified vega approximation
        vega_approx = S * np.sqrt(T) * 0.4
        intrinsic = np.maximum(S - K, 0)
        extrinsic = np.maximum(V - intrinsic, 0.01)
        
        return (extrinsic / vega_approx).clip(0.01, 5.0)
    
    async def download_funding_rate(
        self,
        exchange: str,
        symbols: list,
        start_date: str,
        end_date: str
    ) -> pd.DataFrame:
        """ดึงข้อมูล funding rate สำหรับหลาย symbols"""
        
        tasks = []
        current = datetime.strptime(start_date, '%Y-%m-%d')
        end = datetime.strptime(end_date, '%Y-%m-%d')
        
        while current <= end:
            date_str = current.strftime('%Y-%m-%d')
            for symbol in symbols:
                task = self._download_funding_day(exchange, symbol, date_str)
                tasks.append(task)
            current += timedelta(days=1)
        
        # ดาวน์โหลดแบบ concurrent (rate limit: 10 req/s)
        results = await self._concurrent_download(tasks, max_concurrent=5)
        
        return pd.concat(results, ignore_index=True)
    
    async def _download_funding_day(
        self, 
        exchange: str, 
        symbol: str, 
        date: str
    ) -> pd.DataFrame:
        """ดาวน์โหลด funding rate ของวันเดียว"""
        
        cache_key = f"{exchange}_{symbol}_{date}"
        if cache_key in self._cache:
            return self._cache[cache_key]
        
        url = f"{self.BASE_URL}/{self.EXCHANGES[exchange]['funding']}/{symbol}-{date}.csv.gz"
        headers = {'Authorization': f'Bearer {self.api_token}'}
        
        try:
            async with self.session.get(url, headers=headers) as resp:
                if resp.status == 404:
                    return pd.DataFrame()
                
                content = await resp.read()
                df = pd.read_csv(io.BytesIO(gzip.decompress(content)))
                
                self._cache[cache_key] = df
                return df
                
        except Exception as e:
            print(f"Error downloading {cache_key}: {e}")
            return pd.DataFrame()
    
    async def _concurrent_download(
        self, 
        tasks: list, 
        max_concurrent: int = 5
    ) -> list:
        """Execute tasks พร้อมกันด้วย semaphore สำหรับ rate limiting"""
        
        semaphore = asyncio.Semaphore(max_concurrent)
        
        async def bounded_task(task):
            async with semaphore:
                return await task
        
        return await asyncio.gather(*[bounded_task(t) for t in tasks])


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

async def main():
    async with TardisDataLoader(api_token='your_tardis_token') as loader:
        # ดึง options chain
        btc_options = await loader.download_options_chain(
            exchange='deribit',
            symbol='BTC',
            date='2024-01-15'
        )
        
        # ดึง funding rates
        funding_data = await loader.download_funding_rate(
            exchange='binance',
            symbols=['BTC', 'ETH'],
            start_date='2024-01-01',
            end_date='2024-01-15'
        )
        
        print(f"Options records: {len(btc_options)}")
        print(f"Funding records: {len(funding_data)}")

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

การวิเคราะห์ Options Chain เชิงลึก

หลังจากได้ข้อมูล options chain มาแล้ว ขั้นตอนต่อไปคือการวิเคราะห์เชิงลึก ผมจะสร้าง class สำหรับวิเคราะห์ Greeks, gamma exposure และการกระจายตัวของ open interest

import numpy as np
from scipy.stats import norm
from dataclasses import dataclass
from typing import Dict, List, Tuple
import pandas as pd

@dataclass
class GreeksResult:
    """ผลลัพธ์การคำนวณ Greeks สำหรับ options contract"""
    delta: float
    gamma: float
    theta: float
    vega: float
    rho: float
    implied_volatility: float

class OptionsChainAnalyzer:
    """
    Production-grade options chain analyzer
    คำนวณ Greeks, gamma exposure, ระดับ support/resistance
    """
    
    RISK_FREE_RATE = 0.05  # USD interest rate
    TRADING_DAYS = 252
    
    def __init__(self, underlying_price: float, risk_free_rate: float = None):
        self.S = underlying_price
        self.r = risk_free_rate or self.RISK_FREE_RATE
    
    def calculate_greeks(
        self,
        K: float,  # Strike price
        T: float,  # Time to expiry (in years)
        sigma: float,  # Implied volatility
        option_type: str = 'call'  # 'call' or 'put'
    ) -> GreeksResult:
        """
        คำนวณ option Greeks โดยใช้ Black-Scholes formula
        
        Performance: ~0.1ms per calculation
        """
        
        if T <= 0:
            T = 1e-6  # กัน log(0) error
        
        d1 = (np.log(self.S / K) + (self.r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
        d2 = d1 - sigma * np.sqrt(T)
        
        if option_type == 'call':
            delta = norm.cdf(d1)
            rho = K * T * norm.cdf(d2) / 100
        else:
            delta = norm.cdf(d1) - 1
            rho = -K * T * norm.cdf(-d2) / 100
        
        # Greeks calculations
        gamma = norm.pdf(d1) / (self.S * sigma * np.sqrt(T))
        vega = self.S * norm.pdf(d1) * np.sqrt(T) / 100  # per 1% vol change
        theta_call = (-self.S * norm.pdf(d1) * sigma / (2 * np.sqrt(T)) 
                     - self.r * K * np.exp(-self.r * T) * norm.cdf(d2)) / 365
        theta_put = (-self.S * norm.pdf(d1) * sigma / (2 * np.sqrt(T)) 
                    + self.r * K * np.exp(-self.r * T) * norm.cdf(-d2)) / 365
        theta = theta_call if option_type == 'call' else theta_put
        
        return GreeksResult(
            delta=delta,
            gamma=gamma,
            theta=theta,
            vega=vega,
            rho=rho,
            implied_volatility=sigma
        )
    
    def calculate_gamma_exposure(
        self,
        df: pd.DataFrame,
        option_type_col: str = 'option_type'
    ) -> pd.DataFrame:
        """
        คำนวณ Gamma Exposure (GEX) ของทั้ง chain
        
        GEX = Open Interest * Contract Size * Gamma * Spot Price
        
        สูตรนี้ใช้หา gamma squeeze potential และระดับ price sensitivity
        """
        
        contract_size = df.get('contract_size', 1)
        oi = df['open_interest']
        
        df = df.copy()
        df['T_years'] = (df['expiry_timestamp'] - df['timestamp']) / (365 * 24 * 3600)
        df['T_years'] = df['T_years'].clip(lower=1e-6)
        
        # คำนวณ d1 สำหรับทุก strike
        K = df['strike']
        sigma = df.get('iv_estimate', df.get('implied_volatility', 0.5))
        T = df['T_years']
        
        d1 = (np.log(self.S / K) + (self.r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
        gamma = norm.pdf(d1) / (self.S * sigma * np.sqrt(T))
        
        # Gamma exposure
        # ค่า positive = market maker long gamma
        # ค่า negative = market maker short gamma
        df['gamma_raw'] = gamma * oi * contract_size
        df['gamma_exposure'] = df['gamma_raw'] * self.S
        
        # กรองตาม strike price เพื่อหา key levels
        df['distance_from_spot'] = abs(K - self.S) / self.S
        
        return df
    
    def find_gamma_squeeze_levels(
        self,
        df: pd.DataFrame,
        top_n: int = 10
    ) -> List[Dict]:
        """
        หา strike levels ที่มี gamma exposure สูงที่สุด
        ซึ่งเป็นแนวรับ/แนวต้านที่มีนัยสำคัญ
        """
        
        df = self.calculate_gamma_exposure(df)
        df = df[df['distance_from_spot'] < 0.3]  # เฉพาะ strikes ใกล้ spot
        
        # รวม call และ put gamma
        call_gamma = df[df['option_type'] == 'call'].groupby('strike')['gamma_exposure'].sum()
        put_gamma = df[df['option_type'] == 'put'].groupby('strike')['gamma_exposure'].sum()
        
        all_gamma = pd.DataFrame({
            'call_gamma': call_gamma,
            'put_gamma': put_gamma
        }).fillna(0)
        
        all_gamma['total_gamma'] = all_gamma['call_gamma'] - all_gamma['put_gamma']
        all_gamma = all_gamma.sort_values('total_gamma', ascending=False)
        
        # Top levels
        top_levels = all_gamma.head(top_n).to_dict('index')
        
        return [
            {
                'strike': strike,
                'total_gamma': row['total_gamma'],
                'type': 'resistance' if row['total_gamma'] > 0 else 'support',
                'strength': abs(row['total_gamma']) / abs(all_gamma['total_gamma']).mean()
            }
            for strike, row in top_levels.items()
        ]
    
    def analyze_max_pain(
        self,
        df: pd.DataFrame
    ) -> Tuple[float, float]:
        """
        คำนวณ Max Pain strike price
        
        Max Pain = Strike ที่ทำให้ options หมดอายุ worthless มากที่สุด
        """
        
        strikes = df['strike'].unique()
        pain_scores = []
        
        for K in strikes:
            # สมมติว่า price ที่ expiry = K
            # คำนวณ intrinsic value loss สำหรับ calls
            call_loss = df[df['option_type'] == 'call'].apply(
                lambda row: max(0, K - row['underlying_price']) * row['open_interest'],
                axis=1
            ).sum()
            
            # คำนวณ intrinsic value loss สำหรับ puts
            put_loss = df[df['option_type'] == 'put'].apply(
                lambda row: max(0, row['underlying_price'] - K) * row['open_interest'],
                axis=1
            ).sum()
            
            pain_scores.append((K, call_loss + put_loss))
        
        pain_df = pd.DataFrame(pain_scores, columns=['strike', 'pain'])
        min_pain_idx = pain_df['pain'].idxmin()
        
        return pain_df.loc[min_pain_idx]


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

def analyze_options_chain(df: pd.DataFrame, spot_price: float = 45000):
    """วิเคราะห์ options chain แบบครบวงจร"""
    
    analyzer = OptionsChainAnalyzer(underlying_price=spot_price)
    
    # 1. หา Gamma Squeeze Levels
    gamma_levels = analyzer.find_gamma_squeeze_levels(df, top_n=5)
    
    print("=== Gamma Squeeze Levels ===")
    for level in gamma_levels:
        print(f"Strike ${level['strike']:,.0f}: "
              f"{level['type']} (strength: {level['strength']:.2f}x)")
    
    # 2. คำนวณ Max Pain
    max_pain_strike, min_pain = analyzer.analyze_max_pain(df)
    print(f"\n=== Max Pain ===")
    print(f"Max Pain Strike: ${max_pain_strike:,.0f}")
    print(f"Distance from Spot: {((max_pain_strike - spot_price) / spot_price * 100):.2f}%")
    
    # 3. คำนวณ aggregate Greeks
    total_delta = 0
    total_gamma = 0
    total_vega = 0
    
    for _, row in df.iterrows():
        greeks = analyzer.calculate_greeks(
            K=row['strike'],
            T=row.get('T_years', 0.1),
            sigma=row.get('iv_estimate', 0.5),
            option_type=row.get('option_type', 'call')
        )
        
        weight = row['open_interest'] * row.get('contract_size', 1)
        total_delta += greeks.delta * weight
        total_gamma += greeks.gamma * weight
        total_vega += greeks.vega * weight
    
    print(f"\n=== Aggregate Greeks ===")
    print(f"Net Delta: {total_delta:,.0f}")
    print(f"Net Gamma: {total_gamma:.4f}")
    print(f"Net Vega (per 1% vol): ${total_vega:,.0f}")
    
    return {
        'gamma_levels': gamma_levels,
        'max_pain': max_pain_strike,
        'aggregate_greeks': {
            'delta': total_delta,
            'gamma': total_gamma,
            'vega': total_vega
        }
    }

การวิเคราะห์ Funding Rate สำหรับ Perpetual Futures

Funding rate เป็นตัวชี้วัดสำคัญสำหรับการวิเคราะห์ sentiment และ market positioning ของ perpetual futures ผมจะสร้าง analyzer สำหรับ funding rate patterns

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

@dataclass
class FundingRateStats:
    """สถิติ funding rate ของ asset"""
    symbol: str
    mean_funding: float
    std_funding: float
    current_funding: float
    funding_accrued_7d: float
    annualized_rate: float
    sentiment: str  # 'bullish', 'bearish', 'neutral'
    extreme_event_count: int

class FundingRateAnalyzer:
    """
    วิเคราะห์ funding rate patterns สำหรับ perpetual futures
    
    Funding Rate มีความสำคัญในการบอก:
    1. ทิศทาง sentiment ของตลาด
    2. ความเสี่ยงของ long/short squeeze
    3. Arbitrage opportunities
    """
    
    EXTREME_THRESHOLD = 3  # Standard deviations for extreme events
    
    def __init__(self):
        self.cache = {}
    
    def analyze_funding_rate(
        self,
        df: pd.DataFrame,
        symbol: str,
        funding_interval_hours: float = 8.0
    ) -> FundingRateStats:
        """
        วิเคราะห์ funding rate ของ symbol เดียว
        
        Args:
            df: DataFrame ที่มี columns ['timestamp', 'funding_rate']
            symbol: ชื่อ asset
            funding_interval_hours: ช่วงเวลาของ funding (default: 8 ชม. สำหรับ Binance)
        """
        
        df = df.copy()
        df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
        df = df.sort_values('timestamp')
        
        funding = df['funding_rate']
        
        # สถิติพื้นฐาน
        mean_funding = funding.mean()
        std_funding = funding.std()
        current_funding = funding.iloc[-1]
        
        # คำนวณ funding ที่สะสมใน 7 วัน
        df_7d = df[df['timestamp'] >= df['timestamp'].max() - timedelta(days=7)]
        funding_accrued_7d = df_7d['funding_rate'].sum()
        
        # Annualized rate
        periods_per_day = 24 / funding_interval_hours
        annualized_rate = mean_funding * periods_per_day * 365 * 100
        
        # Sentiment analysis
        sentiment = self._determine_sentiment(
            current_funding, 
            mean_funding, 
            std_funding
        )
        
        # นับ extreme events
        extreme_threshold = mean_funding + self.EXTREME_THRESHOLD * std_funding
        extreme_events = (funding.abs() > extreme_threshold).sum()
        
        return FundingRateStats(
            symbol=symbol,
            mean_funding=mean_funding,
            std_funding=std_funding,
            current_funding=current_funding,
            funding_accrued_7d=funding_accrued_7d,
            annualized_rate=annualized_rate,
            sentiment=sentiment,
            extreme_event_count=extreme_events
        )
    
    def _determine_sentiment(
        self,
        current: float,
        mean: float,
        std: float
    ) -> str:
        """กำหนด sentiment จาก funding rate ปัจจุบัน"""
        
        z_score = (current - mean) / std if std > 0 else 0
        
        if z_score > 1.5:
            return 'bullish'  # Long ต้องจ่าย funding
        elif z_score < -1.5:
            return 'bearish'  # Short ต้องจ่าย funding
        else:
            return 'neutral'
    
    def detect_funding_rate_crossovers(
        self,
        df: pd.DataFrame,
        window_short: int = 8,
        window_long: int = 24
    ) -> pd.DataFrame:
        """
        ตรวจจับ funding rate crossovers
        
        Moving average crossover ของ funding rate
        สามารถใช้เป็น signal สำหรับ sentiment shifts
        """
        
        df = df.copy()
        df = df.sort_values('timestamp')
        
        # คำนวณ moving averages
        df['ma_short'] = df['funding_rate'].rolling(window_short).mean()
        df['ma_long'] = df['funding_rate'].rolling(window_long).mean()
        
        # หา crossover signals
        df['signal'] = 0
        df.loc[df['ma_short'] > df['ma_long'], 'signal'] = 1  # Bullish crossover
        df.loc[df['ma_short'] < df['ma_long'], 'signal'] = -1  # Bearish crossover
        
        # ระบุ crossover points
        df['prev_signal'] = df['signal'].shift(1)
        df['crossover'] = (df['signal'] != df['prev_signal']) & (df['signal'] != 0)
        
        crossovers = df[df['crossover']].copy()
        
        return crossovers[['timestamp', 'funding_rate', 'ma_short', 'ma_long', 'signal']]
    
    def calculate_funding_rate_volatility(
        self,
        df: pd.DataFrame
    ) -> Dict[str, float]:
        """
        คำนวณ volatility ของ funding rate
        
        High volatility = ตลาดไม่แน่นอน, มี sentiment swings
        Low volatility = ตลาด stable, consensus ชัดเจน
        """
        
        funding = df['funding_rate']
        
        # Rolling volatility (24h)
        rolling_vol = funding.rolling(3).std() * 100  # 3 periods = 24h for 8h funding
        
        return {
            'current_volatility': rolling_vol.iloc[-1],
            'mean_volatility': rolling_vol.mean(),
            'max_volatility': rolling_vol.max(),
            'volatility_percentile': (rolling_vol < rolling_vol.iloc[-1]).mean() * 100
        }
    
    def predict_next_funding(
        self,
        df: pd.DataFrame,
        method: str = 'arima'
    ) -> float:
        """
        ทำนาย funding rate ครั้งถัดไป
        
        ใช้ simple moving average หรือ ARIMA ตาม preference
        """
        
        recent_funding = df['funding_rate'].tail(24)  # 8 periods * 24h

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