Deribit 作为全球最大的加密货币期权交易所，其期权链数据是量化交易、波动率曲面建模、Gamma Scalping 策略开发的核心原料。本文将从实战角度详解三种获取 Deribit 期权链历史数据的方案，并重点介绍如何通过 HolySheep AI 的 Tardis.dev 数据中转服务实现低成本、高效率的数据管道搭建，配合 AI 辅助完成波动率分析与预测工作流。
方案对比：HolySheep vs 官方 API vs 其他数据中转
在开始技术细节前，先给出一个直观的横向对比，帮助你快速判断哪种方案最适合你的场景：
| 对比维度 | Deribit 官方 API | Tardis.dev 官方 | HolySheep AI 中转 |
|---|---|---|---|
| 汇率 | ¥7.3 = $1（美元结算） | ¥7.3 = $1 | ¥1 = $1（无损汇率） |
| 国内延迟 | 200-400ms（跨境） | 180-350ms | <50ms（国内直连） |
| 充值方式 | 国际信用卡/PayPal | 国际信用卡 | 微信/支付宝（人民币直充） |
| Deribit 数据 | 原始格式，需二次处理 | 标准化 JSON，含 Order Book | 同 Tardis 标准化格式 |
| 免费额度 | 无 | 注册送 $5 测试金 | 注册即送免费额度 |
| 数据深度 | 实时 + 部分历史 | Tick 级 + Order Book | 同 Tardis 全量数据 |
| 计费方式 | 按请求数 | 按数据量（$/GB 或 $/M messages） | 同 Tardis 透明计费 |
为什么选择 HolySheep 接入 Tardis.dev 数据
作为深度参与加密货币量化开发的工程师，我选择 HolySheep AI 的 Tardis.dev 数据中转主要有三个原因：
- 成本节省超过 85%：官方渠道以美元结算，按 ¥7.3=$1 计算，通过 HolySheep 的 ¥1=$1 无损汇率，实际支出直接打一折
- 国内直连延迟 <50ms：相比直接访问海外 API 的 200-400ms 延迟，在高频期权策略中这是决定性的性能优势
- 微信/支付宝充值：省去国际支付的手续费和换汇麻烦，资金到账即时
环境准备与依赖安装
开始之前，请确保你的 Python 环境满足以下条件，并安装必要的依赖包：
# 创建虚拟环境（推荐）
python -m venv tardis-env
source tardis-env/bin/activate # Linux/Mac
tardis-env\Scripts\activate # Windows
安装核心依赖
pip install tardis-client pandas numpy aiohttp asyncio
pip install python-dotenv # 用于管理 API Key
pip install plotly kaleido # 用于可视化波动率曲面
pip install scipy # 用于数值计算Tardis.dev Deribit 数据接口详解
Tardis.dev 提供的 Deribit 数据覆盖了期权链的完整生命周期数据，包括：
- book_L2_25.{instrument_name}.100ms：Level 2 Order Book，每 100ms 快照
- trades.{instrument_name}：逐笔成交数据
- deribit_series_index：期权序列指数（波动率曲面计算用）
- estimated_liquidation：预估强平价格
HolySheep API 中转配置与代码实现
通过 HolySheep 接入 Tardis.dev，你需要使用 HolySheep 的 base URL 替换官方端点，同时使用 HolySheep 的 API Key 进行认证。以下是完整的 Python 客户端实现：
#!/usr/bin/env python3
"""
Deribit 期权链历史数据下载客户端
通过 HolySheep AI Tardis.dev 中转接入
"""
import os
import asyncio
import aiohttp
import pandas as pd
from datetime import datetime, timedelta
from tardis_client import TardisClient, MessageType
from dotenv import load_dotenv
load_dotenv()
============ HolySheep API 配置 ============
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1/tardis"
Tardis.dev 通过 HolySheep 的端点映射
TARDIS_EXCHANGE = "deribit"
TARDIS_DATA_TYPE = "historical"
class HolySheepTardisClient:
"""通过 HolySheep 中转的 Tardis.dev 客户端"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
async def fetch_trades(
self,
exchange: str,
symbol: str,
from_time: datetime,
to_time: datetime
) -> pd.DataFrame:
"""
获取指定时间范围的期权成交数据
Args:
exchange: 交易所名称 (deribit)
symbol: 合约符号 (如 BTC-28MAR25-95000-C)
from_time: 开始时间
to_time: 结束时间
"""
url = f"{self.base_url}/replay"
# 构造请求参数
params = {
"exchange": exchange,
"symbols": [symbol],
"from": int(from_time.timestamp() * 1000),
"to": int(to_time.timestamp() * 1000),
"channels": [f"trades.{symbol}"]
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
trades_data = []
async with aiohttp.ClientSession() as session:
async with session.post(
url,
json=params,
headers=headers
) as response:
if response.status != 200:
error_text = await response.text()
raise ConnectionError(
f"Tardis API 请求失败: {response.status} - {error_text}"
)
# 实时处理返回的流式数据
async for line in response.content:
if line:
data = line.decode('utf-8').strip()
if data:
trades_data.append(data)
# 解析并转换为 DataFrame
return self._parse_trades(trades_data)
def _parse_trades(self, raw_data: list) -> pd.DataFrame:
"""解析原始成交数据为 DataFrame"""
records = []
for item in raw_data:
try:
import json
msg = json.loads(item)
if msg.get('type') == 'trade':
records.append({
'timestamp': pd.to_datetime(msg['timestamp'], unit='ms'),
'symbol': msg.get('symbol', msg.get('instrument_name')),
'side': msg.get('side'),
'price': float(msg.get('price', 0)),
'amount': float(msg.get('amount', msg.get('size', 0))),
'trade_id': msg.get('id', msg.get('trade_id'))
})
except (json.JSONDecodeError, KeyError) as e:
# 跳过无效数据
continue
return pd.DataFrame(records)
async def download_option_chain_trades():
"""下载期权链完整成交数据示例"""
# 初始化客户端
client = HolySheepTardisClient(api_key=HOLYSHEEP_API_KEY)
# 设置时间范围：过去 24 小时
end_time = datetime.utcnow()
start_time = end_time - timedelta(hours=24)
# BTC 期权活跃序列
btc_options = [
"BTC-28MAR25-95000-C", # Call
"BTC-28MAR25-95000-P", # Put
"BTC-28MAR25-100000-C",
"BTC-28MAR25-100000-P",
"BTC-28MAR25-105000-C",
]
all_trades = []
for symbol in btc_options:
print(f"正在下载 {symbol} 成交数据...")
try:
df = await client.fetch_trades(
exchange="deribit",
symbol=symbol,
from_time=start_time,
to_time=end_time
)
if not df.empty:
all_trades.append(df)
print(f" ✓ 获取 {len(df)} 条成交记录")
except Exception as e:
print(f" ✗ {symbol} 下载失败: {e}")
# 合并所有数据
if all_trades:
combined_df = pd.concat(all_trades, ignore_index=True)
combined_df.to_csv('deribit_options_trades.csv', index=False)
print(f"\n数据已保存至 deribit_options_trades.csv，共 {len(combined_df)} 条记录")
return combined_df
return None
if __name__ == "__main__":
# 执行下载
asyncio.run(download_option_chain_trades())Order Book 数据下载与订单簿重构
对于波动率曲面构建和流动性分析，Order Book 数据比成交数据更为关键。以下代码展示如何下载 L2 订单簿数据并重构买卖盘口：
#!/usr/bin/env python3
"""
Deribit 期权链 Order Book 数据下载与处理
用于波动率曲面构建和流动性分析
"""
import asyncio
import aiohttp
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, List, Optional
import json
@dataclass
class OrderBookLevel:
"""订单簿档位"""
price: float
amount: float
orders: int = 1
@property
def notional(self) -> float:
return self.price * self.amount
@dataclass
class OrderBookSnapshot:
"""订单簿快照"""
timestamp: datetime
symbol: str
bids: List[OrderBookLevel] = field(default_factory=list)
asks: List[OrderBookLevel] = field(default_factory=list)
@property
def best_bid(self) -> Optional[float]:
return self.bids[0].price if self.bids else None
@property
def best_ask(self) -> Optional[float]:
return self.asks[0].price if self.asks else None
@property
def spread(self) -> Optional[float]:
if self.best_bid and self.best_ask:
return self.best_ask - self.best_bid
return None
@property
def mid_price(self) -> Optional[float]:
if self.best_bid and self.best_ask:
return (self.best_bid + self.best_ask) / 2
return None
def implied_volatility_approx(self, time_to_expiry: float) -> float:
"""粗略估算隐含波动率（简化 Black-Scholes 逆向）"""
if not self.mid_price or time_to_expiry <= 0:
return None
# 简化估算，实际应用需要完整的 BS 公式
# 这里假设标的价格 = mid_price
S = self.mid_price
K = S # ATM 期权
r = 0.05 # 无风险利率
T = time_to_expiry
# 简化 IV 估算（OTM Call 的粗略公式）
if S > K: # Call
intrinsic = max(S - K, 0)
time_value = S - intrinsic
iv_approx = time_value / (S * np.sqrt(T)) if T > 0 else 0
return iv_approx * np.sqrt(T) * 100 if iv_approx else None
return None
class DeribitOrderBookDownloader:
"""Deribit 期权链 Order Book 数据下载器"""
BASE_URL = "https://api.holysheep.ai/v1/tardis"
def __init__(self, api_key: str):
self.api_key = api_key
self.snapshots: Dict[str, List[OrderBookSnapshot]] = defaultdict(list)
async def download_orderbook_series(
self,
exchange: str,
symbols: List[str],
from_time: datetime,
to_time: datetime,
channels: Optional[List[str]] = None
) -> Dict[str, pd.DataFrame]:
"""
下载多个期权合约的订单簿序列数据
Returns:
Dict[symbol, DataFrame]，每个合约的订单簿快照 DataFrame
"""
if channels is None:
channels = [f"book_L2_25.{sym}.100ms" for sym in symbols]
url = f"{self.BASE_URL}/replay"
params = {
"exchange": exchange,
"symbols": symbols,
"from": int(from_time.timestamp() * 1000),
"to": int(to_time.timestamp() * 1000),
"channels": channels,
"compression": "gzip"
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Accept-Encoding": "gzip, deflate"
}
# 解析 Order Book 更新
book_state = {} # 当前合约的完整订单簿状态
async with aiohttp.ClientSession() as session:
async with session.post(
url, json=params, headers=headers
) as response:
if response.status != 200:
raise ConnectionError(
f"Order Book 下载失败: HTTP {response.status}"
)
async for line in response.content:
if not line:
continue
try:
msg = json.loads(line.decode('utf-8'))
self._process_orderbook_message(msg, book_state)
except json.JSONDecodeError:
continue
# 转换为 DataFrame 格式
return self._state_to_dataframes(book_state)
def _process_orderbook_message(
self,
msg: dict,
book_state: dict
):
"""处理订单簿更新消息"""
msg_type = msg.get('type', '')
if msg_type not in ('book_snapshot', 'book_update'):
return
symbol = msg.get('instrument_name', msg.get('symbol'))
timestamp = pd.to_datetime(msg.get('timestamp', msg.get('local_timestamp')), unit='ms')
# 初始化或获取该合约的订单簿
if symbol not in book_state:
book_state[symbol] = {
'timestamp': [],
'bids': defaultdict(dict), # price -> {amount, orders}
'asks': defaultdict(dict),
'snapshots': []
}
state = book_state[symbol]
# 处理 bids
for bid in msg.get('bids', msg.get('bid', [])):
price, amount = float(bid[0]), float(bid[1])
if amount == 0:
book_state[symbol]['bids'].pop(price, None)
else:
book_state[symbol]['bids'][price] = {'amount': amount}
# 处理 asks
for ask in msg.get('asks', msg.get('ask', [])):
price, amount = float(ask[0]), float(ask[1])
if amount == 0:
book_state[symbol]['asks'].pop(price, None)
else:
book_state[symbol]['asks'][price] = {'amount': amount}
# 每秒采样一次快照（避免数据量过大）
if not state['snapshots'] or \
(timestamp - state['snapshots'][-1]['timestamp']).total_seconds() >= 1:
bids_sorted = sorted(book_state[symbol]['bids'].items(),
key=lambda x: -x[0])[:10]
asks_sorted = sorted(book_state[symbol]['asks'].items(),
key=lambda x: x[0])[:10]
snapshot = OrderBookSnapshot(
timestamp=timestamp,
symbol=symbol,
bids=[OrderBookLevel(price=p, amount=d['amount'])
for p, d in bids_sorted],
asks=[OrderBookLevel(price=p, amount=d['amount'])
for p, d in asks_sorted]
)
state['snapshots'].append(snapshot)
def _state_to_dataframes(
self,
book_state: dict
) -> Dict[str, pd.DataFrame]:
"""将订单簿状态转换为 DataFrame"""
result = {}
for symbol, state in book_state.items():
records = []
for snap in state['snapshots']:
record = {
'timestamp': snap.timestamp,
'best_bid': snap.best_bid,
'best_ask': snap.best_ask,
'mid_price': snap.mid_price,
'spread': snap.spread,
'bid_depth_10': sum(b.amount for b in snap.bids),
'ask_depth_10': sum(a.amount for a in snap.asks),
'imbalance': (
sum(b.amount for b in snap.bids) -
sum(a.amount for a in snap.asks)
) / (
sum(b.amount for b in snap.bids) +
sum(a.amount for a in snap.asks) + 1e-10
)
}
records.append(record)
result[symbol] = pd.DataFrame(records)
return result
async def main():
"""示例：下载 BTC 期权链 Order Book 数据"""
api_key = "YOUR_HOLYSHEEP_API_KEY" # 替换为你的 HolySheep API Key
downloader = DeribitOrderBookDownloader(api_key)
# 下载最近 1 小时的 BTC ATM 期权
end_time = datetime.utcnow()
start_time = end_time - timedelta(hours=1)
symbols = [
"BTC-28MAR25-95000-C",
"BTC-28MAR25-100000-C",
"BTC-28MAR25-105000-C",
"BTC-28MAR25-95000-P",
"BTC-28MAR25-100000-P",
"BTC-28MAR25-105000-P",
]
print("开始下载 Order Book 数据...")
dfs = await downloader.download_orderbook_series(
exchange="deribit",
symbols=symbols,
from_time=start_time,
to_time=end_time
)
for symbol, df in dfs.items():
if not df.empty:
filename = f"orderbook_{symbol.replace('-', '_').replace('.', '_')}.csv"
df.to_csv(filename, index=False)
print(f"✓ {symbol}: {len(df)} 条快照 → {filename}")
print(f" 平均价差: {df['spread'].mean():.2f}")
print(f" 平均订单簿失衡: {df['imbalance'].mean():.4f}")
if __name__ == "__main__":
asyncio.run(main())AI 波动率分析工作流集成
获取历史数据后，我们可以结合大语言模型进行波动率曲面分析和策略生成。以下是一个完整的工作流示例：
#!/usr/bin/env python3
"""
AI 辅助的加密货币期权波动率分析工作流
结合 HolySheep AI API 进行波动率曲面分析与策略生成
"""
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from scipy.stats import norm
from scipy.optimize import brentq
import requests
import json
from typing import Dict, List, Tuple, Optional
HolySheep AI API 配置
HOLYSHEEP_API_BASE = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # 替换为你的 Key
class BlackScholes:
"""Black-Scholes 期权定价与 Greeks 计算"""
@staticmethod
def d1(S: float, K: float, T: float, r: float, sigma: float) -> float:
if T <= 0 or sigma <= 0:
return np.nan
return (np.log(S / K) + (r + sigma**2 / 2) * T) / (sigma * np.sqrt(T))
@staticmethod
def d2(S: float, K: float, T: float, r: float, sigma: float) -> float:
return BlackScholes.d1(S, K, T, r, sigma) - sigma * np.sqrt(T)
@staticmethod
def call_price(S: float, K: float, T: float, r: float, sigma: float) -> float:
if T <= 0:
return max(S - K, 0)
d1 = BlackScholes.d1(S, K, T, r, sigma)
d2 = BlackScholes.d2(S, K, T, r, sigma)
return S * norm.cdf(d1) - K * np.exp(-r * T) * norm.cdf(d2)
@staticmethod
def put_price(S: float, K: float, T: float, r: float, sigma: float) -> float:
if T <= 0:
return max(K - S, 0)
d1 = BlackScholes.d1(S, K, T, r, sigma)
d2 = BlackScholes.d2(S, K, T, r, sigma)
return K * np.exp(-r * T) * norm.cdf(-d2) - S * norm.cdf(-d1)
@staticmethod
def implied_volatility(
market_price: float,
S: float, K: float, T: float,
r: float,
option_type: str = 'call',
high: float = 5.0,
low: float = 0.001
) -> float:
"""使用 Brent 方法求解隐含波动率"""
def objective(sigma):
if option_type == 'call':
price = BlackScholes.call_price(S, K, T, r, sigma)
else:
price = BlackScholes.put_price(S, K, T, r, sigma)
return price - market_price
try:
return brentq(objective, low, high, xtol=1e-6)
except ValueError:
return np.nan
@staticmethod
def greeks(S: float, K: float, T: float, r: float, sigma: float) -> Dict[str, float]:
"""计算期权 Greeks"""
d1 = BlackScholes.d1(S, K, T, r, sigma)
d2 = BlackScholes.d2(S, K, T, r, sigma)
delta_call = norm.cdf(d1)
delta_put = -norm.cdf(-d1)
gamma = norm.pdf(d1) / (S * sigma * np.sqrt(T))
vega = S * norm.pdf(d1) * np.sqrt(T) / 100 # 每 1% 波动率
theta_call = (
-S * norm.pdf(d1) * sigma / (2 * np.sqrt(T))
- r * K * np.exp(-r * T) * norm.cdf(d2)
) / 365
theta_put = (
-S * norm.pdf(d1) * sigma / (2 * np.sqrt(T))
+ r * K * np.exp(-r * T) * norm.cdf(-d2)
) / 365
return {
'delta_call': delta_call,
'delta_put': delta_put,
'gamma': gamma,
'vega': vega,
'theta_call': theta_call,
'theta_put': theta_put
}
class VolatilitySurfaceAnalyzer:
"""波动率曲面分析器"""
def __init__(self, api_key: str):
self.api_key = api_key
self.bs = BlackScholes()
def calculate_iv_surface(
self,
option_data: pd.DataFrame,
spot_price: float,
risk_free_rate: float = 0.05
) -> pd.DataFrame:
"""
从期权市场数据计算隐含波动率曲面
Args:
option_data: 包含 symbol, expiry, strike, market_price 等列
spot_price: 当前标的价格
risk_free_rate: 无风险利率
"""
results = []
for _, row in option_data.iterrows():
strike = row['strike']
expiry = row['expiry']
market_price = row['market_price']
option_type = 'call' if strike > spot_price else 'put'
# 计算到期时间（年化）
if isinstance(expiry, str):
expiry_date = pd.to_datetime(expiry)
else:
expiry_date = expiry
T = (expiry_date - datetime.now()).days / 365.25
if T <= 0:
continue
# 计算隐含波动率
iv = self.bs.implied_volatility(
market_price, spot_price, strike, T, risk_free_rate, option_type
)
if not np.isnan(iv):
greeks = self.bs.greeks(spot_price, strike, T, risk_free_rate, iv)
results.append({
'symbol': row.get('symbol', ''),
'strike': strike,
'expiry': expiry,
'T': T,
'moneyness': strike / spot_price,
'option_type': option_type,
'market_price': market_price,
'iv': iv * 100, # 转为百分比
'iv_bps': iv * 10000,
**greeks
})
return pd.DataFrame(results)
def get_vol_skew(self, surface_df: pd.DataFrame) -> Dict[str, float]:
"""计算波动率偏斜（Skew）"""
if surface_df.empty:
return {}
atm_options = surface_df[
(surface_df['moneyness'] > 0.95) &
(surface_df['moneyness'] < 1.05)
]
otm_calls = surface_df[
(surface_df['moneyness'] > 1.05) &
(surface_df['option_type'] == 'call')
]
otm_puts = surface_df[
(surface_df['moneyness'] < 0.95) &
(surface_df['option_type'] == 'put')
]
atm_iv = atm_options['iv'].mean() if not atm_options.empty else np.nan
skew_25d = np.nan # 25 delta skew
rr_25d = np.nan # Risk Reversal 25 delta
if not otm_calls.empty and not otm_puts.empty:
rr_25d = otm_calls['iv'].mean() - otm_puts['iv'].mean()
return {
'atm_iv': atm_iv,
'skew_25d': skew_25d,
'rr_25d': rr_25d
}
class AIAnalysisClient:
"""通过 HolySheep AI 进行期权分析的自然语言查询"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
def analyze_volatility(
self,
surface_data: pd.DataFrame,
question: str
) -> str:
"""
使用 AI 分析波动率曲面数据
Args:
surface_data: 隐含波动率曲面 DataFrame
question: 自然语言问题
"""
# 准备数据摘要
data_summary = surface_data.describe().to_string()
prompt = f"""你是一位专业的加密货币期权量化分析师。请根据以下隐含波动率曲面数据回答问题。
波动率曲面数据摘要：
{data_summary}
当前数据包含：strike(行权价), T(到期时间), moneyness(货币性), iv(隐含波动率%), delta, gamma, vega, theta
用户问题：{question}
请给出专业的分析回答，包括：
1. 数据解读
2. 潜在的交易机会或风险提示
3. 具体的数值依据
回答格式：中文，条理清晰，包含具体数字。"""
# 调用 HolySheep AI API
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "claude-sonnet-4-20250514", # 选择合适的模型
"messages": [
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 2000
}
response = requests.post(
f"{self.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"AI API 调用失败: {response.status_code} - {response.text}")
def generate_trading_signal(
self,
surface_data: pd.DataFrame,
spot_price: float,
historical_vol: float
) -> Dict:
"""使用 AI 生成交易信号"""
current_skew = self._calculate_current_skew(surface_data)
prompt = f"""作为期权交易专家，请根据以下数据生成交易信号：
当前标的价格：${spot_price}
历史波动率：{historical_vol*100:.2f}%
隐含波动率曲面状态：
- ATM IV: {current_skew.get('atm_iv', 0):.2f}%
- 25 Delta Risk Reversal: {current_skew.get('rr_25d', 0):.2f}%
请分析以下方面并给出 JSON 格式的交易建议：
{{
"signal": "BULLISH/BEARISH/NEUTRAL",
"confidence": 0.0-1.0,
"strategy": "具体策略名称",
"entry_conditions": ["条件1", "条件2"],
"risk_management": {{
"stop_loss": "止损条件",
"take_profit": "止盈条件",
"max_position_size": 0.1
}},
"reasoning": "分析理由"
}}
只输出 JSON，不要有其他内容。"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "claude-sonnet-4-20250514",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.2,
"max_tokens": 1000
}
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
if response.status_code == 200:
result = response.json()
content = result['choices'][0]['message']['content']
return json.loads(content)
else:
raise Exception(f"AI API 调用失败: {response.status_code}")
def _calculate_current_skew(self, surface_df: pd.DataFrame) -> Dict[str, float]:
"""计算当前波动率偏斜状态"""
analyzer = VolatilitySurfaceAnalyzer(self.api_key)
return analyzer.get_vol_skew(surface_df)
async def demo_workflow():
"""演示完整的工作流"""
# 1. 模拟期权数据（实际应用中从 Tardis 下载）
sample_data = pd.DataFrame([
{'symbol': 'BTC-28MAR25-95000-C', 'strike': 95000, 'expiry': '2025-03-28', 'market_price': 8500},
{'symbol': 'BTC-28MAR25-100000-C', 'strike': 100000, 'expiry': '2025-03-28', 'market_price': 5200},
{'symbol': 'BTC-28MAR25-105000-C', 'strike': 105000, 'expiry': '2025-03-28', 'market_price': 2800},
{'symbol': 'BTC-28MAR25-95000-P', 'strike': 95000, 'expiry': '2025