ในโลกของ DeFi และ Cryptocurrency derivatives การทำ backtesting ที่แม่นยำเป็นกุญแจสำคัญสู่ความสำเร็จของกลยุทธ์การซื้อขาย Options บน Deribit ซึ่งเป็นตลาด Options ที่ใหญ่ที่สุดในโลกสำหรับ BTC และ ETH วันนี้ผมจะพาทุกท่านไปสำรวจวิธีการดึงข้อมูล Orderbook อย่างครบถ้วนด้วย Tardis API พร้อมทั้งกระบวนการจัดเก็บในรูปแบบ Parquet ที่ช่วยให้การประมวลผลข้อมูลย้อนหลังทำได้อย่างรวดเร็วและมีประสิทธิภาพสูงสุด
ทำไมต้องเป็น Deribit Options Orderbook?
Deribit ครอบครองส่วนแบ่งการซื้อขาย Options บน Bitcoin กว่า 80% ของตลาดทั่วโลก ทำให้ข้อมูล Orderbook จากแพลตฟอร์มนี้มีความสำคัญอย่างยิ่งสำหรับนักเทรดและนักวิจัย ข้อมูล Orderbook ที่มีคุณภาพจะช่วยให้เราสามารถ:
- วิเคราะห์ Implied Volatility Surface อย่างแม่นยำ
- ทดสอบกลยุทธ์ Delta Hedging และ Gamma Scalping
- ประเมินความลึกของตลาด (Market Depth) ตาม Strike Price ต่างๆ
- คำนวณต้นทุนในการเข้าออกสถานะอย่างแท้จริง
Tardis API: ผู้ช่วยดึงข้อมูลตลาดระดับ Professional
Tardis API เป็นบริการที่รวบรวมข้อมูลตลาดคริปโตจากหลายแพลตฟอร์มในรูปแบบที่พร้อมสำหรับการใช้งาน รองรับ Deribit ทั้ง Futures และ Options พร้อมข้อมูล Orderbook, Trades และ Funding Rates ในความละเอียดระดับ Millisecond
การติดตั้งและ Setup สภาพแวดล้อม
เริ่มต้นด้วยการติดตั้ง dependencies ที่จำเป็นสำหรับโปรเจกต์นี้
# ติดตั้ง packages ที่จำเป็น
pip install tardis-client pandas pyarrow fastparquet requests aiohttp
สำหรับ visualization
pip install plotly kaleido
สำหรับ technical indicators
pip install ta-lib pandas-ta
ตรวจสอบ version ที่ติดตั้ง
python -c "import tardis_client; print(f'Tardis Client: {tardis_client.__version__}')"
python -c "import pandas; print(f'Pandas: {pandas.__version__}')"
python -c "import pyarrow; print(f'PyArrow: {pyarrow.__version__}')"โครงสร้างข้อมูล Deribit Options Orderbook
ก่อนจะเริ่มดึงข้อมูล เราต้องเข้าใจโครงสร้างข้อมูล Orderbook จาก Deribit ก่อน โดยข้อมูลประกอบด้วย Bids (คำสั่งซื้อ) และ Asks (คำสั่งขาย) พร้อมราคาและปริมาณ ซึ่งจะถูกจัดเก็บในรูปแบบ JSON ที่มีโครงสร้างซ้อนกัน
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from tardis_client import TardisClient, Channel
from datetime import datetime, timedelta
import asyncio
import json
class DeribitOrderbookCollector:
"""
คลาสสำหรับดึงข้อมูล Orderbook จาก Deribit Options ผ่าน Tardis API
และจัดเก็บในรูปแบบ Parquet สำหรับ Backtesting
"""
def __init__(self, api_key: str, exchange: str = "deribit"):
self.api_key = api_key
self.exchange = exchange
self.client = None
self.orderbook_data = []
async def fetch_options_orderbook(
self,
start_date: datetime,
end_date: datetime,
instrument_name: str = "BTC-*-*-*" # รองรับ wildcards
):
"""
ดึงข้อมูล Options Orderbook ตามช่วงเวลาที่กำหนด
Parameters:
-----------
start_date : datetime
วันที่เริ่มต้นการดึงข้อมูล
end_date : datetime
วันที่สิ้นสุดการดึงข้อมูล
instrument_name : str
ชื่อ instrument สำหรับ filter (รองรับ wildcards)
"""
self.client = TardisClient(api_key=self.api_key)
# กำหนด channels ที่ต้องการดึง
channels = [
Channel(name=instrument_name, exchange=self.exchange)
]
print(f"เริ่มดึงข้อมูล: {start_date} ถึง {end_date}")
print(f"Instrument: {instrument_name}")
# ดึงข้อมูลแบบ realtime/replay
self.orderbook_data = []
async for message in self.client.replay(
channels=channels,
from_date=start_date,
to_date=end_date
):
if message.type == "orderbook_snapshot" or message.type == "orderbook_update":
record = self._parse_orderbook_message(message)
if record:
self.orderbook_data.append(record)
# แสดงความคืบหน้าทุก 10,000 records
if len(self.orderbook_data) % 10000 == 0:
print(f"ดึงข้อมูลได้แล้ว: {len(self.orderbook_data):,} records")
def _parse_orderbook_message(self, message) -> dict:
"""แปลง message จาก Tardis เป็น dictionary ที่มีโครงสร้างเรียบง่าย"""
timestamp = pd.to_datetime(message.timestamp, unit='ms')
# แปลง bids และ asks เป็น DataFrame
bids_df = pd.DataFrame(message.bids, columns=['price', 'amount'])
asks_df = pd.DataFrame(message.asks, columns=['price', 'amount'])
# คำนวณ metrics พื้นฐาน
best_bid = float(bids_df['price'].iloc[0]) if len(bids_df) > 0 else None
best_ask = float(asks_df['price'].iloc[0]) if len(asks_df) > 0 else None
spread = best_ask - best_bid if best_bid and best_ask else None
spread_pct = (spread / best_bid * 100) if spread and best_bid else None
# คำนวณ volume รวม
total_bid_volume = bids_df['amount'].sum()
total_ask_volume = asks_df['amount'].sum()
return {
'timestamp': timestamp,
'instrument_name': message.name,
'best_bid': best_bid,
'best_ask': best_ask,
'spread': spread,
'spread_pct': spread_pct,
'total_bid_volume': total_bid_volume,
'total_ask_volume': total_ask_volume,
'bid_ask_ratio': total_bid_volume / total_ask_volume if total_ask_volume > 0 else None,
'orderbook_bids': message.bids[:10], # เก็บ 10 ระดับแรก
'orderbook_asks': message.asks[:10],
'local_timestamp': datetime.now()
}
def to_parquet(self, filepath: str, partition_by: str = 'date'):
"""
จัดเก็บข้อมูลเป็น Parquet file พร้อม partitioning
Parameters:
-----------
filepath : str
path สำหรับบันทึกไฟล์
partition_by : str
คอลัมน์สำหรับ partitioning ('date', 'instrument', 'hour')
"""
df = pd.DataFrame(self.orderbook_data)
if df.empty:
print("ไม่มีข้อมูลที่จะบันทึก")
return
# เพิ่มคอลัมน์สำหรับ partitioning
if partition_by == 'date':
df['date'] = df['timestamp'].dt.date
elif partition_by == 'hour':
df['hour'] = df['timestamp'].dt.floor('H')
elif partition_by == 'instrument':
df['expiry'] = df['instrument_name'].str.extract(r'-(\d{2}[A-Z]{3}\d{2})-')
# แปลง lists เป็น JSON string สำหรับ Parquet
df['orderbook_bids'] = df['orderbook_bids'].apply(json.dumps)
df['orderbook_asks'] = df['orderbook_asks'].apply(json.dumps)
# เลือก columns ที่จะเก็บ
columns_to_save = [
'timestamp', 'instrument_name', 'best_bid', 'best_ask',
'spread', 'spread_pct', 'total_bid_volume', 'total_ask_volume',
'bid_ask_ratio', 'orderbook_bids', 'orderbook_asks'
]
if partition_by in df.columns:
columns_to_save.append(partition_by)
df = df[columns_to_save]
# บันทึกเป็น Parquet
table = pa.Table.from_pandas(df)
pq.write_table(
table,
filepath,
compression='snappy',
use_dictionary=True
)
print(f"บันทึกข้อมูลสำเร็จ: {len(df):,} records")
print(f"ขนาดไฟล์: {df.memory_usage(deep=True).sum() / 1024**2:.2f} MB")
print(f"Path: {filepath}")
ตัวอย่างการใช้งาน
async def main():
collector = DeribitOrderbookCollector(
api_key="YOUR_TARDIS_API_KEY" # ใส่ API key ของคุณ
)
# ดึงข้อมูลย้อนหลัง 7 วัน
end_date = datetime.now()
start_date = end_date - timedelta(days=7)
# ดึงข้อมูล BTC Options
await collector.fetch_options_orderbook(
start_date=start_date,
end_date=end_date,
instrument_name="BTC-*-*-*" # ดึงทุก BTC Options
)
# บันทึกเป็น Parquet
collector.to_parquet(
filepath="./data/deribit_options_orderbook.parquet",
partition_by='date'
)
return collector.orderbook_data
if __name__ == "__main__":
data = asyncio.run(main())การประมวลผลข้อมูลสำหรับ Backtesting
เมื่อได้ข้อมูล Orderbook แล้ว ขั้นตอนต่อไปคือการประมวลผลเพื่อให้พร้อมสำหรับการทำ Backtest ต่างๆ
import pandas as pd
import numpy as np
from typing import Dict, List, Tuple
class OptionsBacktestProcessor:
"""
คลาสสำหรับประมวลผลข้อมูล Orderbook เพื่อใช้ใน Backtesting
"""
def __init__(self, parquet_path: str):
self.parquet_path = parquet_path
self.df = None
def load_data(self, date_filter: List[str] = None) -> pd.DataFrame:
"""
โหลดข้อมูลจาก Parquet file
Parameters:
-----------
date_filter : List[str]
รายชื่อวันที่ที่ต้องการโหลด (format: 'YYYY-MM-DD')
"""
# อ่านข้อมูลจาก Parquet
if date_filter:
self.df = pd.read_parquet(
self.parquet_path,
filters=[('date', 'in', date_filter)]
)
else:
self.df = pd.read_parquet(self.parquet_path)
# แปลง timestamp
self.df['timestamp'] = pd.to_datetime(self.df['timestamp'])
self.df = self.df.sort_values('timestamp')
print(f"โหลดข้อมูลสำเร็จ: {len(self.df):,} records")
print(f"ช่วงเวลา: {self.df['timestamp'].min()} ถึง {self.df['timestamp'].max()}")
return self.df
def calculate_slippage(
self,
order_size: float,
side: str = 'buy',
depth_levels: int = 5
) -> pd.DataFrame:
"""
คำนวณ slippage จาก Orderbook depth
Parameters:
-----------
order_size : float
ขนาด order ที่ต้องการคำนวณ slippage (ในหน่วย BTC)
side : str
'buy' หรือ 'sell'
depth_levels : int
จำนวน levels ของ orderbook ที่ใช้คำนวณ
Returns:
--------
DataFrame พร้อมคอลัมน์ slippage และ effective_price
"""
def calc_slippage_row(row, order_size, side):
# Parse JSON strings กลับเป็น list
asks = json.loads(row['orderbook_asks']) if isinstance(row['orderbook_asks'], str) else row['orderbook_asks']
bids = json.loads(row['orderbook_bids']) if isinstance(row['orderbook_bids'], str) else row['orderbook_bids']
if side == 'buy':
levels = asks[:depth_levels]
else:
levels = bids[:depth_levels]
if not levels:
return pd.Series({
'slippage_bps': np.nan,
'effective_price': np.nan,
'filled_levels': 0,
'vwap': np.nan
})
remaining_size = order_size
total_cost = 0
filled_levels = 0
for price, amount in levels:
if remaining_size <= 0:
break
fill_amount = min(remaining_size, float(amount))
total_cost += fill_amount * float(price)
remaining_size -= fill_amount
filled_levels += 1
if remaining_size > 0:
# Order ไม่สามารถ filled ครบได้
return pd.Series({
'slippage_bps': np.nan,
'effective_price': np.nan,
'filled_levels': filled_levels,
'vwap': np.nan
})
vwap = total_cost / order_size
best_price = float(levels[0][0])
slippage_bps = (vwap - best_price) / best_price * 10000
if side == 'sell':
slippage_bps = -slippage_bps # slippage เป็นลบเมื่อขาย
return pd.Series({
'slippage_bps': slippage_bps,
'effective_price': vwap,
'filled_levels': filled_levels,
'vwap': vwap
})
# คำนวณ slippage ทีละ row (ใช้ apply)
slippage_df = self.df.apply(
lambda row: calc_slippage_row(row, order_size, side),
axis=1
)
self.df = pd.concat([self.df, slippage_df], axis=1)
return self.df
def calculate_market_impact(
self,
order_sizes: List[float] = [0.1, 0.25, 0.5, 1.0]
) -> pd.DataFrame:
"""
คำนวณ market impact สำหรับหลายขนาด order
ผลลัพธ์จะเป็น DataFrame ที่มี market impact สำหรับแต่ละขนาด
"""
results = []
for size in order_sizes:
print(f"กำลังคำนวณ market impact สำหรับ order size: {size} BTC")
df_slippage = self.calculate_slippage(
order_size=size,
side='buy'
)
results.append({
'order_size': size,
'avg_slippage_bps': df_slippage['slippage_bps'].mean(),
'median_slippage_bps': df_slippage['slippage_bps'].median(),
'p95_slippage_bps': df_slippage['slippage_bps'].quantile(0.95),
'unfillable_rate': df_slippage['slippage_bps'].isna().mean() * 100
})
return pd.DataFrame(results)
def get_volatility_surface(self) -> pd.DataFrame:
"""
สร้าง Implied Volatility Surface จาก Orderbook data
โดยใช้ spread เป็น proxy
"""
# Group by instrument และคำนวณ statistics
surface = self.df.groupby('instrument_name').agg({
'spread_pct': ['mean', 'std', 'median'],
'bid_ask_ratio': ['mean', 'median'],
'total_bid_volume': ['mean', 'sum'],
'total_ask_volume': ['mean', 'sum'],
'timestamp': ['min', 'max', 'count']
}).reset_index()
# Flatten column names
surface.columns = [
'instrument_name',
'avg_spread_pct', 'std_spread_pct', 'median_spread_pct',
'avg_bid_ask_ratio', 'median_bid_ask_ratio',
'avg_bid_volume', 'total_bid_volume',
'avg_ask_volume', 'total_ask_volume',
'first_timestamp', 'last_timestamp', 'num_records'
]
return surface
def export_for_backtest(self, output_path: str):
"""
Export ข้อมูลที่ประมวลผลแล้วสำหรับใช้ใน backtesting framework
"""
# เลือก columns ที่จำเป็น
backtest_df = self.df[[
'timestamp',
'instrument_name',
'best_bid',
'best_ask',
'spread',
'spread_pct',
'total_bid_volume',
'total_ask_volume'
]].copy()
# Export เป็น Parquet อีกครั้ง (กรณีมีการคำนวณเพิ่ม)
backtest_df.to_parquet(
output_path,
compression='snappy',
index=False
)
# Export เป็น CSV สำหรับดูข้อมูลเบื้องต้น
csv_path = output_path.replace('.parquet', '.csv')
backtest_df.to_csv(csv_path, index=False)
print(f"Export สำเร็จ:")
print(f" - Parquet: {output_path}")
print(f" - CSV: {csv_path}")
ตัวอย่างการใช้งาน Processor
processor = OptionsBacktestProcessor(
parquet_path="./data/deribit_options_orderbook.parquet"
)
โหลดข้อมูล
df = processor.load_data()
คำนวณ slippage สำหรับ order size ต่างๆ
market_impact = processor.calculate_market_impact(
order_sizes=[0.1, 0.25, 0.5, 1.0, 2.0]
)
print("\nMarket Impact Analysis:")
print(market_impact)
สร้าง Volatility Surface
vol_surface = processor.get_volatility_surface()
print("\nVolatility Surface Sample:")
print(vol_surface.head(10))
Export สำหรับ Backtesting
processor.export_for_backtest("./data/backtest_ready.parquet")การใช้งานร่วมกับ HolySheep AI สำหรับ Analysis
หลังจากได้ข้อมูลที่ประมวลผลแล้ว คุณสามารถนำไปวิเคราะห์เพิ่มเติมด้วย AI Models จาก HolySheep AI เพื่อทำ Sentiment Analysis หรือ Pattern Recognition จากข้อมูล Orderbook
import requests
import json
class HolySheepAIClient:
"""
Client สำหรับเชื่อมต่อกับ HolySheep AI API
ใช้สำหรับวิเคราะห์ข้อมูล Orderbook ด้วย AI
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
def analyze_orderbook_pattern(
self,
orderbook_data: dict,
analysis_type: str = "pattern_recognition"
) -> dict:
"""
วิเคราะห์ Orderbook pattern ด้วย AI
Parameters:
-----------
orderbook_data : dict
ข้อมูล Orderbook ที่รวบรวมได้
analysis_type : str
ประเภทการวิเคราะห์ (pattern_recognition, sentiment, prediction)
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
# สร้าง prompt สำหรับ AI
prompt = self._build_analysis_prompt(orderbook_data, analysis_type)
payload = {
"model": "gpt-4.1", # ใช้ GPT-4.1 สำหรับ complex analysis
"messages": [
{
"role": "system",
"content": "คุณเป็นผู้เชี่ยวชาญด้าน Cryptocurrency Options Trading และ Market Microstructure"
},
{
"role": "user",
"content": prompt
}
],
"temperature": 0.3,
"max_tokens": 2000
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
if response.status_code == 200:
result = response.json()
return {
"success": True,
"analysis": result['choices'][0]['message']['content'],
"usage": result.get('usage', {})
}
else:
return {
"success": False,
"error": response.text
}
def _build_analysis_prompt(self, data: dict, analysis_type: str) -> str:
"""สร้าง prompt สำหรับการวิเคราะห์แต่ละประเภท"""
best_bid = data.get('best_bid', 'N/A')
best_ask = data.get('best_ask', 'N/A')
spread_pct = data.get('spread_pct', 'N/A')
bid_volume = data.get('total_bid_volume', 'N/A')
ask_volume = data.get('total_ask_volume', 'N/A')
base_prompt = f"""
วิเคราะห์ข้อมูล Options Orderbook ต่อไปนี้:
Best Bid: ${best_bid}
Best Ask: ${best_ask}
Spread: {spread_pct}%
Total Bid Volume: {bid_volume}
Total Ask Volume: {ask_volume}
"""
if analysis_type == "pattern_recognition":
base_prompt += """
กรุณาระบุ:
1. รูปแบบ Orderbook ที่พบ (เช่น bid wall, ask wall, vacuum, balanced)
2. ระดับความผันผวนของตลาด (low/medium/high)
3. สัญญาณที่อาจเกิดขึ้น (manipulation, arbitrage, smart money)
"""
elif analysis_type == "sentiment":
base_prompt += """
กรุณาระบุ:
1. ความรู้สึกของตลาดโดยรวม (bullish/bearish/neutral)
2. ดุลยพินิจระหว่างผู้ซื้อและผู้ขาย
3. แนวโน้มที่อาจเกิดขึ้นในระยะสั้น
"""
return base_prompt
ตัวอย่างการใช้งาน
holy_client = HolySheepAIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
วิเคราะห์ pattern จาก snapshot ล่าสุด
latest_orderbook = {
'best_bid': 45000,
'best_ask': 45050,
'spread_pct': 0.11,
'total_bid_volume': 1500,
'total_ask_volume': 800
}
result = holy_client.analyze_orderbook_pattern(
orderbook_data=latest_orderbook,
analysis_type="pattern_recognition"
)
if result['success']:
print("ผลการวิเคราะห์จาก AI:")
print(result['analysis'])
print(f"\nค่าใช้จ่าย: ${result['usage']['total_tokens'] / 1000 * 8:.4f}") # GPT-4.1 pricing
else:
print(f"เกิดข้อผิดพลาด: {result['error']}")เหมาะกับใคร / ไม่เหมาะกับใคร
| เหมาะกับใคร | ไม่เหมาะกับใคร |
|---|---|
| นักเทรด Options มืออาชีพที่ต้องการ Backtesting ที่แม่นยำ | ผู้เริ่มต้นที่ยังไม่มีความรู้พื้นฐานเกี่ยวกับ Options หรือ Python |
| ทีม Quant ที่ต้องการข้อมูลคุณภาพสูงสำหรับสร้างกลยุทธ์ | นักเทรดรา�
แหล่งข้อมูลที่เกี่ยวข้องบทความที่เกี่ยวข้อง🔥 ลอง HolySheep AIเกตเวย์ AI API โดยตรง รองรับ Claude, GPT-5, Gemini, DeepSeek — หนึ่งคีย์ ไม่ต้อง VPN |