ในโลกของ DeFi และ DeFi Options การวิเคราะห์ Implied Volatility (IV) Surface ถือเป็นหัวใจสำคัญสำหรับนักเทรดและนักพัฒนาที่ต้องการสร้าง стратегия การซื้อขายอย่างมีประสิทธิภาพ บทความนี้จะพาคุณสร้างระบบ Archive ข้อมูล IV Surface จาก Deribit ผ่าน Tardis API พร้อมทั้งวิธีการ Backtest ด้วย Python ฉบับสมบูรณ์

ทำความรู้จักกับ IV Surface และความสำคัญ

IV Surface หรือ Volatility Surface คือการแสดงภาพ Implied Volatility ของ Options ตาม Strike Price และ Time to Expiration ซึ่งช่วยให้เราเข้าใจพฤติกรรมราคาของ Options ในสถานการณ์ต่างๆ การเก็บข้อมูล IV Surface ย้อนหลังช่วยให้เราสามารถ:

การตั้งค่า Environment และ Dependencies

เริ่มต้นด้วยการติดตั้ง Python packages ที่จำเป็นสำหรับโปรเจกต์นี้

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


สำหรับการเชื่อมต่อกับ Tardis API


สมัคร API Key ที่ https://tardis.dev/



ติดตั้ง PostgreSQL client สำหรับเก็บข้อมูล

pip install psycopg2-binary sqlalchemy[postgresql]

ดึงข้อมูล Deribit Options จาก Tardis API

Tardis API ให้บริการข้อมูล Tick-by-Tick จาก Deribit ซึ่งรวมถึง Order Book และ Trade Data ของ Options เราสามารถใช้ Python Client ดึงข้อมูลและคำนวณ IV จาก Bid/Ask Prices ได้โดยตรง

import asyncio
from tardis_client import TardisClient, Channel
import pandas as pd
from datetime import datetime, timedelta
import numpy as np


Tardis API Configuration

TARDIS_API_KEY = "your_tardis_api_key"
EXCHANGE = "deribit"
INSTRUMENT_TYPE = "option"

async def fetch_options_data(start_date, end_date, instrument_name=None):
    """
    ดึงข้อมูล Options จาก Tardis API
    """
    client = TardisClient(api_key=TARDIS_API_KEY)
    
    # กำหนดช่วงเวลาที่ต้องการ
    from_timestamp = int(start_date.timestamp() * 1000)
    to_timestamp = int(end_date.timestamp() * 1000)
    
    # สร้าง Channel Filter สำหรับ Options
    channels = []
    
    if instrument_name:
        # ดึงข้อมูลเฉพาะ Instrument
        channels.append(Channel.from_description(
            f"{EXCHANGE}:{instrument_name}:book"
        ))
        channels.append(Channel.from_description(
            f"{EXCHANGE}:{instrument_name}:trade"
        ))
    else:
        # ดึงข้อมูล Options ทั้งหมด
        channels.append(Channel.from_description(
            f"{EXCHANGE}:*:*:book"
        ))
        channels.append(Channel.from_description(
            f"{EXCHANGE}:*:*:trade"
        ))
    
    trades_data = []
    book_data = []
    
    # วนลูปดึงข้อมูลตามช่วงเวลา
    async for local_timestamp, message in client.stream(
        channels=channels,
        from_timestamp=from_timestamp,
        to_timestamp=to_timestamp,
        from_id=None
    ):
        if message.type == "book":
            book_data.append({
                'timestamp': local_timestamp,
                'instrument': message.instrument_name,
                'bids': message.bids,
                'asks': message.asks
            })
        elif message.type == "trade":
            trades_data.append({
                'timestamp': local_timestamp,
                'instrument': message.instrument_name,
                'price': message.price,
                'size': message.size,
                'side': message.side
            })
    
    return pd.DataFrame(trades_data), pd.DataFrame(book_data)


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

start = datetime(2026, 4, 1, 0, 0, 0)
end = datetime(2026, 4, 30, 23, 59, 59)

trades_df, books_df = asyncio.run(
    fetch_options_data(start, end, "BTC-29MAY26-95000-C")
)

print(f"ดึงข้อมูลสำเร็จ: {len(trades_df)} trades, {len(books_df)} order books")
print(trades_df.head())

คำนวณ Implied Volatility จาก Order Book Data

การคำนวณ IV ต้องอาศัย Black-Scholes Model แบบหลัก ซึ่งในที่นี้เราจะใช้ Newton-Raphson Method เพื่อหาค่า Volatility ที่ทำให้ Theoretical Price เท่ากับ Market Price

from scipy.stats import norm
from scipy.optimize import newton
from scipy.interpolate import griddata
import warnings
warnings.filterwarnings('ignore')

def black_scholes_call(S, K, T, r, sigma):
    """
    คำนวณ Call Option Price ด้วย Black-Scholes
    """
    if T <= 0 or sigma <= 0:
        return max(S - K, 0)
    
    d1 = (np.log(S / K) + (r + 0.5 * sigma ** 2) * T) / (sigma * np.sqrt(T))
    d2 = d1 - sigma * np.sqrt(T)
    
    call_price = S * norm.cdf(d1) - K * np.exp(-r * T) * norm.cdf(d2)
    return call_price

def calculate_implied_volatility(market_price, S, K, T, r, option_type='call'):
    """
    คำนวณ IV โดยใช้ Newton-Raphson Method
    """
    if T <= 0:
        return np.nan
    
    # ตรวจสอบ Parity
    intrinsic = max(S - K, 0) if option_type == 'call' else max(K - S, 0)
    if market_price <= intrinsic:
        return np.nan
    
    def objective(sigma):
        if option_type == 'call':
            return black_scholes_call(S, K, T, r, sigma) - market_price
        else:
            return black_scholes_put(S, K, T, r, sigma) - market_price
    
    try:
        # เริ่มต้นด้วย Volatility ที่ 50%
        iv = newton(objective, 0.5, maxiter=100)
        return iv if 0.01 < iv < 5.0 else np.nan
    except:
        return np.nan

def black_scholes_put(S, K, T, r, sigma):
    """
    คำนวณ Put Option Price ด้วย Black-Scholes
    """
    if T <= 0 or sigma <= 0:
        return max(K - S, 0)
    
    d1 = (np.log(S / K) + (r + 0.5 * sigma ** 2) * T) / (sigma * np.sqrt(T))
    d2 = d1 - sigma * np.sqrt(T)
    
    put_price = K * np.exp(-r * T) * norm.cdf(-d2) - S * norm.cdf(-d1)
    return put_price

def build_iv_surface(books_df, trades_df, S, T, r=0.01):
    """
    สร้าง IV Surface จากข้อมูล Order Book
    """
    # รวมข้อมูล bid/ask จาก order book
    iv_data = []
    
    for _, row in books_df.iterrows():
        instrument = row['instrument']
        
        # Parse Strike และ Expiry จาก instrument name
        # Format: BTC-29MAY26-95000-C หรือ BTC-29MAY26-95000-P
        try:
            parts = instrument.split('-')
            expiry_str = parts[1]  # 29MAY26
            strike = float(parts[2])
            option_type = 'call' if parts[3] == 'C' else 'put'
        except:
            continue
        
        bids = row['bids']
        asks = row['asks']
        
        if len(bids) > 0 and len(asks) > 0:
            mid_price = (float(bids[0][0]) + float(asks[0][0])) / 2
            spread = float(asks[0][0]) - float(bids[0][0])
            
            # คำนวณ IV จาก Mid Price
            iv = calculate_implied_volatility(mid_price, S, strike, T, r, option_type)
            
            iv_data.append({
                'timestamp': row['timestamp'],
                'instrument': instrument,
                'strike': strike,
                'option_type': option_type,
                'mid_price': mid_price,
                'spread': spread,
                'implied_volatility': iv,
                'moneyness': S / strike
            })
    
    return pd.DataFrame(iv_data)


ตัวอย่างการสร้าง IV Surface


สมมติ BTC Price อยู่ที่ $95,000 และ Expiry 30 วัน

S_btc = 95000  # Spot Price
T_days = 30
T_years = T_days / 365
r_rate = 0.01  # Risk-free rate

iv_surface_df = build_iv_surface(books_df, trades_df, S_btc, T_years, r_rate)
print(f"IV Surface Data Points: {len(iv_surface_df)}")
print(iv_surface_df.head(10))

สร้าง Visualization ของ Volatility Surface

การแสดงผล Volatility Surface แบบ 3D ช่วยให้เห็นภาพรวมของตลาด Options ได้ชัดเจนขึ้น เราจะใช้ Plotly สร้าง Interactive Chart ที่สามารถหมุนและซูมได้

import plotly.graph_objects as go
from plotly.subplots import make_subplots

def plot_iv_surface_3d(iv_surface_df, snapshot_time=None):
    """
    สร้าง 3D Volatility Surface Plot
    """
    # กรองข้อมูลตาม timestamp
    if snapshot_time:
        df = iv_surface_df[iv_surface_df['timestamp'] <= snapshot_time]
    else:
        df = iv_surface_df.dropna(subset=['implied_volatility'])
    
    # สร้าง Grid สำหรับ Interpolation
    strikes = df['strike'].unique()
    moneyness = df['moneyness'].unique()
    
    # สร้าง Pivot Table
    pivot = df.pivot_table(
        values='implied_volatility',
        index='moneyness',
        columns='strike',
        aggfunc='mean'
    )
    
    # Interpolate สำหรับค่าที่หายไป
    pivot_interpolated = pivot.interpolate(method='linear', axis=1)
    pivot_interpolated = pivot_interpolated.interpolate(method='linear', axis=0)
    
    # สร้าง 3D Surface Plot
    fig = go.Figure()
    
    fig.add_trace(go.Surface(
        x=pivot_interpolated.columns,
        y=pivot_interpolated.index,
        z=pivot_interpolated.values,
        colorscale='Viridis',
        colorbar=dict(title='IV'),
        name='IV Surface'
    ))
    
    fig.update_layout(
        title=f'Volatility Surface (BTC Options) - {snapshot_time}',
        scene=dict(
            xaxis_title='Strike Price',
            yaxis_title='Moneyness (S/K)',
            zaxis_title='Implied Volatility'
        ),
        width=900,
        height=700
    )
    
    return fig

def plot_volatility_smile(iv_df, expiry_group):
    """
    สร้าง Volatility Smile Plot
    """
    df = iv_df[iv_df['expiry_days'] == expiry_group].dropna()
    
    fig = make_subplots(rows=1, cols=2, 
                       subplot_titles=('Call Options', 'Put Options'))
    
    calls = df[df['option_type'] == 'call']
    puts = df[df['option_type'] == 'put']
    
    fig.add_trace(go.Scatter(
        x=calls['strike'],
        y=calls['implied_volatility'],
        mode='lines+markers',
        name='Call IV',
        line=dict(color='blue')
    ), row=1, col=1)
    
    fig.add_trace(go.Scatter(
        x=puts['strike'],
        y=puts['implied_volatility'],
        mode='lines+markers',
        name='Put IV',
        line=dict(color='red')
    ), row=1, col=2)
    
    fig.update_layout(title=f'Volatility Smile - {expiry_group} Days to Expiry')
    
    return fig


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

fig_3d = plot_iv_surface_3d(iv_surface_df)
fig_3d.show()


หรือบันทึกเป็น HTML

fig_3d.write_html('iv_surface_3d.html')

Archiving System สำหรับ Historical IV Data

การสร้างระบบ Archive ที่มีประสิทธิภาพช่วยให้เราสามารถ Query ข้อมูลย้อนหลังได้อย่างรวดเร็ว ระบบนี้ใช้ PostgreSQL เป็น Data Warehouse และ Time-series Indexing สำหรับการค้นหาที่รวดเร็ว

from sqlalchemy import create_engine, Column, Float, String, DateTime, Integer, Index
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime
import json

Base = declarative_base()

class IVSnapshot(Base):
    """
    Table สำหรับเก็บ IV Surface Snapshots
    """
    __tablename__ = 'iv_snapshots'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    timestamp = Column(DateTime, nullable=False, index=True)
    instrument_name = Column(String(100), nullable=False)
    strike = Column(Float, nullable=False)
    expiry = Column(DateTime, nullable=False)
    option_type = Column(String(10), nullable=False)
    spot_price = Column(Float, nullable=False)
    bid_price = Column(Float)
    ask_price = Column(Float)
    mid_price = Column(Float)
    implied_volatility = Column(Float, index=True)
    moneyness = Column(Float)
    
    # Composite Index สำหรับ Time-series Query
    __table_args__ = (
        Index('idx_timestamp_instrument', 'timestamp', 'instrument_name'),
        Index('idx_timestamp_strike', 'timestamp', 'strike'),
    )

class ArchiveManager:
    """
    Manager สำหรับจัดการ IV Surface Archive
    """
    
    def __init__(self, db_connection_string):
        self.engine = create_engine(db_connection_string)
        Base.metadata.create_all(self.engine)
        self.Session = sessionmaker(bind=self.engine)
    
    def save_snapshot(self, snapshot_data):
        """
        บันทึก IV Surface Snapshot
        """
        session = self.Session()
        try:
            records = []
            for item in snapshot_data:
                record = IVSnapshot(
                    timestamp=item['timestamp'],
                    instrument_name=item['instrument'],
                    strike=item['strike'],
                    expiry=item['expiry'],
                    option_type=item['option_type'],
                    spot_price=item['spot_price'],
                    bid_price=item.get('bid_price'),
                    ask_price=item.get('ask_price'),
                    mid_price=item.get('mid_price'),
                    implied_volatility=item.get('implied_volatility'),
                    moneyness=item.get('moneyness')
                )
                records.append(record)
            
            session.bulk_save_objects(records)
            session.commit()
            return len(records)
        except Exception as e:
            session.rollback()
            raise e
        finally:
            session.close()
    
    def query_iv_surface(self, start_date, end_date, strike_range=None):
        """
        Query IV Surface ตามช่วงเวลาและ Strike Range
        """
        session = self.Session()
        try:
            query = session.query(IVSnapshot).filter(
                IVSnapshot.timestamp.between(start_date, end_date)
            )
            
            if strike_range:
                query = query.filter(
                    IVSnapshot.strike.between(strike_range[0], strike_range[1])
                )
            
            return query.all()
        finally:
            session.close()
    
    def get_volatility_smile(self, timestamp, expiry_date):
        """
        ดึงข้อมูล Volatility Smile ณ เวลาที่กำหนด
        """
        session = self.Session()
        try:
            return session.query(IVSnapshot).filter(
                IVSnapshot.timestamp == timestamp,
                IVSnapshot.expiry == expiry_date
            ).order_by(IVSnapshot.strike).all()
        finally:
            session.close()


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

DB_CONNECTION = "postgresql://user:password@localhost:5432/iv_archive"

archive_manager = ArchiveManager(DB_CONNECTION)


บันทึก Snapshot

snapshot = [
    {
        'timestamp': datetime.now(),
        'instrument': 'BTC-29MAY26-95000-C',
        'strike': 95000,
        'expiry': datetime(2026, 5, 29),
        'option_type': 'call',
        'spot_price': 95000,
        'mid_price': 0.05,
        'implied_volatility': 0.65,
        'moneyness': 1.0
    }
]

saved_count = archive_manager.save_snapshot(snapshot)
print(f"บันทึกสำเร็จ: {saved_count} records")

Backtesting Framework สำหรับ Options Strategy

การทำ Backtest บน IV Surface ที่เก็บไว้ช่วยให้เราทดสอบกลยุทธ์ต่างๆ ได้อย่างมีประสิทธิภาพ เช่น Delta Hedging, Straddle/Strangle, หรือ Calendar Spread

class OptionsBacktester:
    """
    Backtesting Framework สำหรับ Options Strategies
    """
    
    def __init__(self, archive_manager, initial_capital=100000):
        self.archive = archive_manager
        self.capital = initial_capital
        self.positions = []
        self.trades = []
        self.portfolio_value = [initial_capital]
    
    def execute_straddle(self, expiry_date, strike, entry_time, exit_time):
        """
        กลยุทธ์ Straddle: ซื้อ Call และ Put ที่ Strike เดียวกัน
        """
        # ดึงข้อมูล IV ณ เวลาเข้า
        entry_iv = self.archive.query_iv_surface(
            entry_time, entry_time, strike_range=(strike-1000, strike+1000)
        )
        
        # คำนวณต้นทุน Straddle
        call_iv = next((x.implied_volatility for x in entry_iv 
                       if x.option_type == 'call' and x.strike == strike), None)
        put_iv = next((x.implied_volatility for x in entry_iv 
                      if x.option_type == 'put' and x.strike == strike), None)
        
        if call_iv and put_iv:
            entry_cost = call_iv + put_iv
            
            # ดึงข้อมูล IV ณ เวลาออก
            exit_iv = self.archive.query_iv_surface(
                exit_time, exit_time, strike_range=(strike-1000, strike+1000)
            )
            
            exit_call_iv = next((x.implied_volatility for x in exit_iv 
                                if x.option_type == 'call'), None)
            exit_put_iv = next((x.implied_volatility for x in exit_iv 
                               if x.option_type == 'put'), None)
            
            if exit_call_iv and exit_put_iv:
                exit_value = exit_call_iv + exit_put_iv
                pnl = exit_value - entry_cost
                pnl_percent = (pnl / entry_cost) * 100
                
                return {
                    'strategy': 'Straddle',
                    'strike': strike,
                    'entry_iv': entry_cost,
                    'exit_iv': exit_value,
                    'pnl': pnl,
                    'pnl_percent': pnl_percent
                }
        
        return None
    
    def run_volatility_mean_reversion(self, start_date, end_date, 
                                      lookback_days=30, entry_threshold=2.0):
        """
        กลยุทธ์ Volatility Mean Reversion
        ซื้อเมื่อ IV ต่ำกว่า Mean - Threshold * Std
        ขายเมื่อ IV สูงกว่า Mean + Threshold * Std
        """
        results = []
        current_date = start_date
        
        while current_date < end_date:
            # คำนวณ Historical Mean IV
            historical = self.archive.query_iv_surface(
                current_date - timedelta(days=lookback_days),
                current_date
            )
            
            if len(historical) > 100:
                iv_values = [h.implied_volatility for h in historical 
                            if h.implied_volatility]
                mean_iv = np.mean(iv_values)
                std_iv = np.std(iv_values)
                
                # ดึง IV ปัจจุบัน
                current_iv = self.archive.query_iv_surface(
                    current_date, current_date
                )
                
                if current_iv:
                    current_mean_iv = np.mean([c.implied_volatility 
                                              for c in current_iv 
                                              if c.implied_volatility])
                    
                    # ตรวจสอบสัญญาณ
                    if current_mean_iv < mean_iv - entry_threshold * std_iv:
                        # ส่งสัญญาณ Long Vega
                        results.append({
                            'date': current_date,
                            'signal': 'LONG_VEGA',
                            'current_iv': current_mean_iv,
                            'mean_iv': mean_iv,
                            'z_score': (current_mean_iv - mean_iv) / std_iv
                        })
                    elif current_mean_iv > mean_iv + entry_threshold * std_iv:
                        # ส่งสัญญาณ SHORT_VEGA
                        results.append({
                            'date': current_date,
                            'signal': 'SHORT_VEGA',
                            'current_iv': current_mean_iv,
                            'mean_iv': mean_iv,
                            'z_score': (current_mean_iv - mean_iv) / std_iv
                        })
            
            current_date += timedelta(days=1)
        
        return pd.DataFrame(results)


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

backtester = OptionsBacktester(archive_manager, initial_capital=100000)

start_date = datetime(2026, 3, 1)
end_date = datetime(2026, 4, 30)


รัน Mean Reversion Strategy

signals = backtester.run_volatility_mean_reversion(
    start_date, end_date, 
    lookback_days=30, 
    entry_threshold=1.5
)

print(f"สัญญาณที่พบ: {len(signals)}")
print(signals.head())

ใช้ HolySheep AI วิเคราะห์ IV Surface Patterns

นอกจากการคำนวณด้วย Traditional Methods แล้ว คุณยังสามารถใช้ AI Models จาก HolySheep AI เพื่อวิเคราะห์ Patterns ใน IV Surface ได้อย่างมีประสิทธิภาพ ตัวอย่างเช่น การตรวจจับ Volatility Regime Changes หรือการทำนาย IV ล่วงหน้า

import requests
import json


HolySheep AI API Configuration

BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"

def analyze_iv_patterns_with_ai(iv_surface_df, model="gpt-4.1"):
    """
    ใช้ AI วิเคราะห์ Patterns ใน IV Surface
    """
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    
    # เตรียมข้อมูล IV Surface สำหรับ Analysis
    summary_stats = iv_surface_df.groupby(['option_type', 'moneyness']).agg({
        'implied_volatility': ['mean', 'std', 'min', 'max']
    }).round(4)
    
    # สร้าง Prompt สำหรับ AI
    prompt = f"""
    วิเคราะห์ IV Surface สำหรับ Options Trading:

    ข้อมูลสรุป (Implied Volatility by Moneyness):
    {summary_stats.to_string()}

    กรุณาวิเคราะห์:
    1. Volatility Skew - ระบุว่า Skew เป็นแบบ Forward หรือ Backward
    2. Smile Effect - อธิบายรูปร่างของ Smile Curve
    3. Regime Detection - ระบุว่าตลาดอยู่ในช่วง Low/High Volatility
    4. Trading Signals - เสนอกลยุทธ์ที่เหมาะสมกับสถานการณ์ปัจจุบัน
    5. Risk Factors - ระบุความเสี่ยงที่ควรระวัง
    """
    
    payload = {
        "model": model,
        "messages": [
            {
                "role": "system",
                "content": "คุณเป็นผู้เชี่ยวชาญด้าน Options Trading และ Volatility Analysis"
            },
            {
                "role": "user", 
                "content": prompt
            }
        ],
        "temperature": 0.3,
        "max_tokens": 2000
    }
    
    response = requests.post(
        f"{BASE_URL}/chat/completions",
        headers=headers,
        json=payload,
        timeout=30
    )
    
    if response.status_code == 200:
        result = response.json()
        return result['choices'][0]['message']['content']
    else:
        raise Exception(f"API Error: {response.status_code} - {response.text}")

def detect_volatility_regime_change(iv_history_df, window=20):
    """
    ตรวจจับ Volatility Regime Change ด้วย AI
    """
    # คำนวณ Rolling Statistics
    iv_history_df['rolling_mean'] = iv_history_df['implied_volatility'].rolling(window).mean()
    iv_history_df['rolling

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