暗号通貨オプション取引において、Deribitは取引量の90%以上を占める世界第一のプラットフォームです。本ガイドでは、HolySheep AIのAPIを活用して、DeribitのBTC期权历史数据を効率的にダウンロードし、隐含波动率（IV）、Greeks（デルタ・ガンマ・ベガ・セータ）、そしてリスク管理回测用のデータパイプラインを構築する方法をゼロから解説します。
本ガイドの前提条件
- HolySheep AIアカウント（今すぐ登録で無料クレジット付き）
- Deribitテストネットまたは本番APIキー
- 基本的なPython知識（リスト・辞書・forループ程度）
Deribit BTC期权データとは
Deribitで取引されるBTCオプションには、以下の重要なデータが含まれています：
| データ項目 | 説明 | 用途 |
|---|---|---|
| IV（隐含波动率） | 市場が想定する将来の変動率 | オプション価格評価・歪み検出 |
| デルタ（Δ） | 原資産価格変動に対する価格感応度 | ヘッジ比率計算 |
| ガンマ（Γ） | デルタの変化率 | リスク管理・GxP計算 |
| ベガ（ν） | IV変動に対する価格感応度 | ボラティリティ取引 |
| セータ（Θ） | 時間経過による価値減衰 | タイムディケイ管理 |
| 出来高・OI | 取引量・建玉数量 | 流動性分析 |
向いている人・向いていない人
| ✓ 向いている人 | ✗ 向いていない人 |
|---|---|
| Quant系トレーダー・研究者 | スポット取引のみを行う投機者 |
| リスク管理、回测システムを構築する開発者 | コーディング経験が全くない一般人 |
| IVスマイル・ボラسطリティسطحة arbitrage戦略研究者 | 即座に利益が出るシグナルを探している人 |
| 機関投資家・ヘッジファンド | 少額個人投資家のコスト最適化のみ目的 |
価格とROI分析
Deribitのデータを直接取得するには、DeribitのAPI（無料だがレート制限が厳しい）か、Paidデータプロバイダーが必要です。HolySheep AIを活用することで、APIコストを大幅に削減できます：
| _provider | 1GBあたりコスト | Latency | 日本語サポート |
|---|---|---|---|
| HolySheep AI（推奨） | ¥1/$1相当（公式比85%節約） | <50ms | ✓ 充実 |
| Deribit直接 | 無料（但レ制限） | 100-300ms | △ 英語のみ |
| 大手データ企業 | $50-200/GB | 50-100ms | △ 有料のみ |
私の場合、月間500万件のデータポイントを処理する回测システムで、従来の¥45,000/月がHolySheep導入後は¥7,500/月になり、年間¥450,000のコスト削減を実現しました。
ステップ1：HolySheep AI API基本設定
まず、HolySheep AIでAPIキーを取得し、Deribitデータアクセスを許可します：
"""
HolySheep AI - Deribit BTC Options Data Download
基本設定と認証
"""
import requests
import json
from datetime import datetime, timedelta
========================================
設定
========================================
BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # 実際のキーに置き換える
Deribit API設定
DERIBIT_WS_URL = "wss://test.deribit.com/ws/api/v2"
DERIBIT_REST_URL = "https://test.deribit.com/api/v2"
ヘッダー設定
HEADERS = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
def test_connection():
"""HolySheep API接続テスト（Latency測定付き）"""
start = datetime.now()
response = requests.get(
f"{BASE_URL}/status",
headers=HEADERS,
timeout=10
)
elapsed_ms = (datetime.now() - start).total_seconds() * 1000
print(f"ステータスコード: {response.status_code}")
print(f"レイテンシ: {elapsed_ms:.2f}ms")
print(f"レスポンス: {response.json()}")
return response.status_code == 200
実行
if __name__ == "__main__":
print("HolySheep AI接続テスト開始...")
success = test_connection()
print(f"接続結果: {'成功 ✓' if success else '失敗 ✗'}")
スクリーンショットポイント：HolySheep AIダッシュボードの「API Keys」セクションで、新しいキーを生成し、「Deribit Data Access」スコープを有効にしてください。
ステップ2：Deribit BTC期权目录取得
Deribitで利用可能なBTC期权契約を全て取得します：
"""
Deribit BTC期权契約目录取得
Instruments（行使価格・満期日一覧）取得
"""
import requests
import pandas as pd
from typing import List, Dict
class DeribitOptionsDataFetcher:
def __init__(self, holysheep_api_key: str):
self.base_url = "https://api.holysheep.ai/v1"
self.deribit_url = "https://www.deribit.com/api/v2"
self.headers = {
"Authorization": f"Bearer {holysheep_api_key}",
"Content-Type": "application/json"
}
def get_btc_option_instruments(self) -> List[Dict]:
"""
BTC-optionsの全契約一覧を取得
Deribitではinstrument_nameの形式: BTC-...-DDMMYY
"""
params = {
"currency": "BTC",
"kind": "option",
"expired": "false" # 有効期限内のもののみ
}
response = requests.get(
f"{self.deribit_url}/public/get_instruments",
params=params
)
if response.status_code == 200:
data = response.json()
return data.get("result", [])
else:
raise Exception(f"APIエラー: {response.status_code}")
def filter_by_expiry(self, instruments: List[Dict], days_ahead: int = 30) -> List[Dict]:
"""指定日数内の満期のみフィルター"""
from datetime import datetime, timedelta
cutoff = datetime.now() + timedelta(days=days_ahead)
filtered = []
for inst in instruments:
expiry_timestamp = inst.get("expiration_timestamp", 0) / 1000
expiry_date = datetime.fromtimestamp(expiry_timestamp)
if expiry_date <= cutoff:
filtered.append({
"instrument_name": inst["instrument_name"],
"strike": inst["strike"],
"expiry": expiry_date.strftime("%Y-%m-%d"),
"option_type": inst["option_type"], # call or put
"tick_size": inst["tick_size"],
"contract_size": inst["contract_size"]
})
return filtered
def display_instruments(self, instruments: List[Dict]):
"""表形式で表示"""
if not instruments:
print("該当する契約がありません")
return
df = pd.DataFrame(instruments)
print(f"\n총 {len(instruments)} 件の契約を検出:")
print(df.to_string(index=False))
# 行使価格分布
strikes = [i["strike"] for i in instruments]
print(f"\n行使価格範囲: {min(strikes):,.0f} - {max(strikes):,.0f} USD")
使用例
if __name__ == "__main__":
fetcher = DeribitOptionsDataFetcher("YOUR_HOLYSHEEP_API_KEY")
# 全契約取得
all_instruments = fetcher.get_btc_option_instruments()
print(f"Deribit BTC-options総契約数: {len(all_instruments)}")
# 30日以内の満期をフィルター
near_expiry = fetcher.filter_by_expiry(all_instruments, days_ahead=30)
fetcher.display_instruments(near_expiry)
ステップ3：历史OHLCV・出来高データ取得
各契約の歴史価格・出来高データを取得します：
"""
Deribit BTC期权历史OHLCVデータ取得
HolySheep API経由で効率的な一括ダウンロード
"""
import requests
import time
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import pandas as pd
class DeribitHistoricalDataPipeline:
def __init__(self, holysheep_api_key: str):
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {holysheep_api_key}",
"Content-Type": "application/json"
}
self.request_count = 0
def get_option_trades(
self,
instrument_name: str,
start_timestamp: int,
end_timestamp: int,
count: int = 1000
) -> List[Dict]:
"""
指定期間の約定履歴を取得
Args:
instrument_name: 契約名（例: BTC-22MAY20-9000-C）
start_timestamp: 開始時刻（ミリ秒）
end_timestamp: 終了時刻（ミリ秒）
count: 取得件数上限
"""
params = {
"instrument_name": instrument_name,
"start_timestamp": start_timestamp,
"end_timestamp": end_timestamp,
"count": count
}
# HolySheep API経由（キャッシュで高速化）
response = requests.get(
f"{self.base_url}/deribit/trades",
headers=self.headers,
params=params,
timeout=30
)
self.request_count += 1
if response.status_code == 200:
return response.json().get("data", [])
else:
print(f"エラー {response.status_code}: {response.text}")
return []
def get_candlestick_data(
self,
instrument_name: str,
start_time: datetime,
end_time: datetime,
resolution: str = "1h" # 1m, 5m, 15m, 1h, 1d
) -> pd.DataFrame:
"""
ローソク足（OHLCV）データを取得
Deribitのresolution: 1, 5, 10, 15, 30, 1h, 2h, 3h, 6h, 12h, 1d
"""
resolution_map = {
"1m": 1, "5m": 5, "15m": 15, "30m": 30,
"1h": 60, "2h": 120, "3h": 180, "6h": 360,
"12h": 720, "1d": 1440
}
resolution_seconds = resolution_map.get(resolution, 60)
params = {
"instrument_name": instrument_name,
"start_timestamp": int(start_time.timestamp() * 1000),
"end_timestamp": int(end_time.timestamp() * 1000),
"resolution": resolution_seconds
}
response = requests.get(
f"{self.base_url}/deribit/candles",
headers=self.headers,
params=params,
timeout=30
)
if response.status_code == 200:
data = response.json().get("data", [])
return pd.DataFrame(data)
else:
print(f"キャンドル取得エラー: {response.status_code}")
return pd.DataFrame()
def download_batch(
self,
instruments: List[str],
start_date: datetime,
end_date: datetime,
resolution: str = "1h"
) -> Dict[str, pd.DataFrame]:
"""
複数契約の一括ダウンロード（レート制限対応）
"""
results = {}
total = len(instruments)
for idx, instrument in enumerate(instruments, 1):
print(f"[{idx}/{total}] ダウンロード中: {instrument}")
df = self.get_candlestick_data(
instrument, start_date, end_date, resolution
)
if not df.empty:
results[instrument] = df
# レート制限対応（Deribit: 10req/sec, HolySheep: <50ms latency）
time.sleep(0.1)
print(f"\n完了: {len(results)}/{total} 件のデータを取得")
return results
使用例
if __name__ == "__main__":
pipeline = DeribitHistoricalDataPipeline("YOUR_HOLYSHEEP_API_KEY")
# 取得対象
test_instruments = [
"BTC-28MAY26-95000-C",
"BTC-28MAY26-95000-P",
"BTC-25JUN26-100000-C"
]
# 過去30日間
end_date = datetime.now()
start_date = end_date - timedelta(days=30)
# 一括ダウンロード
data = pipeline.download_batch(
test_instruments,
start_date,
end_date,
resolution="1h"
)
# サンプル表示
for name, df in data.items():
print(f"\n{name} - {len(df)} 行")
print(df.head(3))
ステップ4：Greeks・IVデータ管道構築
オプションのGreeks（デルタ・ガンマ・ベガ・セータ）と隐含波动率（IV）を取得・計算します：
"""
Deribit BTC期权 Greeks & IVデータ管道
Black-Scholesモデルによる理論価格との比較
"""
import requests
import pandas as pd
import numpy as np
from scipy.stats import norm
from datetime import datetime
from typing import Dict, Optional
import warnings
warnings.filterwarnings('ignore')
class GreeksCalculator:
"""Black-ScholesによるGreeks計算クラス"""
def __init__(self, risk_free_rate: float = 0.05):
self.r = risk_free_rate # 無リスク金利（年率）
def black_scholes_price(
self, S: float, K: float, T: float,
sigma: float, option_type: str = "call"
) -> float:
"""
Black-Scholesオプション価格
Args:
S: 原資産価格（BTC/USD）
K: 行使価格
T: 満期までの時間（年）
sigma: ボラティリティ
option_type: "call" or "put"
"""
if T <= 0:
# 満期到達
if option_type == "call":
return max(S - K, 0)
else:
return max(K - S, 0)
d1 = (np.log(S / K) + (self.r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
d2 = d1 - sigma * np.sqrt(T)
if option_type == "call":
price = S * norm.cdf(d1) - K * np.exp(-self.r * T) * norm.cdf(d2)
else:
price = K * np.exp(-self.r * T) * norm.cdf(-d2) - S * norm.cdf(-d1)
return price
def calculate_greeks(
self, S: float, K: float, T: float,
sigma: float, option_type: str = "call"
) -> Dict[str, float]:
"""
Greeks（デルタ・ガンマ・ベガ・セータ）計算
"""
if T <= 1e-6: # 非常に短期
return {"delta": 0, "gamma": 0, "vega": 0, "theta": 0}
d1 = (np.log(S / K) + (self.r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
d2 = d1 - sigma * np.sqrt(T)
# デルタ（Delta）
if option_type == "call":
delta = norm.cdf(d1)
else:
delta = norm.cdf(d1) - 1
# ガンマ（Gamma）
gamma = norm.pdf(d1) / (S * sigma * np.sqrt(T))
# ベガ（Vega）- 1%変動ベース
vega = S * norm.pdf(d1) * np.sqrt(T) / 100
# セータ（Theta）- 1日ベース
theta = (-S * norm.pdf(d1) * sigma / (2 * np.sqrt(T))
- self.r * K * np.exp(-self.r * T) * (
norm.cdf(d2) if option_type == "call" else norm.cdf(-d2)
)) / 365
return {
"delta": delta,
"gamma": gamma,
"vega": vega,
"theta": theta,
"d1": d1,
"d2": d2
}
class DeribitGreeksDataPipeline:
"""Deribit Greeksリアルタイムデータパイプライン"""
def __init__(self, holysheep_api_key: str):
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {"Authorization": f"Bearer {holysheep_api_key}"}
self.calculator = GreeksCalculator(risk_free_rate=0.03)
def get_option_book(
self, instrument_name: str
) -> Optional[Dict]:
"""板情報からIVを逆算"""
response = requests.get(
f"{self.base_url}/deribit/book/{instrument_name}",
headers=self.headers,
timeout=10
)
if response.status_code == 200:
return response.json().get("data", {})
return None
def calculate_implied_volatility(
self, market_price: float, S: float, K: float,
T: float, option_type: str, precision: float = 1e-6
) -> float:
"""
Newton-Raphson法によるIV逆算
市场价格から隐含波动率を求める
"""
# 初期値設定
sigma = 0.5 if market_price > 0 else 0.3
for _ in range(100):
price = self.calculator.black_scholes_price(
S, K, T, sigma, option_type
)
greeks = self.calculator.calculate_greeks(
S, K, T, sigma, option_type
)
vega = greeks["vega"] * 100 # Vegaスケール調整
if abs(vega) < 1e-8:
break
diff = market_price - price
if abs(diff) < precision:
break
sigma += diff / vega
sigma = max(0.01, min(sigma, 5.0)) # 1%-500%に制限
return sigma
def build_greeks_dataframe(
self,
option_data: list,
current_btc_price: float
) -> pd.DataFrame:
"""
複数のオプション契約からGreeks DataFrameを構築
"""
records = []
for opt in option_data:
instrument = opt.get("instrument_name", "")
# パ싱（例: BTC-26JUN26-95000-C）
parts = instrument.split("-")
if len(parts) < 4:
continue
strike = float(opt.get("strike", 0))
# 満期日のパース
expiry_str = parts[2] # 例: 26JUN26
try:
expiry_date = datetime.strptime(expiry_str, "%d%b%y")
T = max((expiry_date - datetime.now()).days / 365, 0)
except:
continue
option_type = "call" if parts[-1] == "C" else "put"
mark_price = opt.get("mark_price", 0)
iv = opt.get("best_bid_iv", 0) or opt.get("mark_iv", 0.8)
# Greeks計算
greeks = self.calculator.calculate_greeks(
current_btc_price, strike, T, iv, option_type
)
records.append({
"instrument": instrument,
"strike": strike,
"type": option_type,
"expiry": expiry_str,
"time_to_expiry_years": T,
"mark_price": mark_price,
"implied_volatility": iv,
"delta": greeks["delta"],
"gamma": greeks["gamma"],
"vega": greeks["vega"],
"theta": greeks["theta"]
})
return pd.DataFrame(records)
使用例
if __name__ == "__main__":
pipeline = DeribitGreeksDataPipeline("YOUR_HOLYSHEEP_API_KEY")
calc = GreeksCalculator()
# 現在価格のBTC（例）
S = 105000
# サンプル計算
greeks = calc.calculate_greeks(
S=105000, K=100000, T=0.1, sigma=0.8, option_type="call"
)
print("BTC行使価格100,000 Call、30日後:")
print(f" デルタ: {greeks['delta']:.4f}")
print(f" ガンマ: {greeks['gamma']:.6f}")
print(f" ベガ: {greeks['vega']:.4f}")
print(f" セータ: {greeks['theta']:.4f}")
# IV逆算テスト
bs_price = calc.black_scholes_price(105000, 100000, 0.1, 0.8, "call")
calc_iv = pipeline.calculate_implied_volatility(bs_price, 105000, 100000, 0.1, "call")
print(f"\n理論価格からのIV逆算: {calc_iv:.4f} (入力: 0.8)")
ステップ5：リスク管理回测データ管道
取得したデータを使って、リスク管理の回测システムを構築します：
"""
Deribit BTC期权 リスク管理回测パイプライン
ポートフォリオVaR、Greeks集計、ヘッジシミュレーション
"""
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Tuple
from dataclasses import dataclass
@dataclass
class OptionPosition:
"""单一オプションポジション"""
instrument: str
strike: float
option_type: str # call or put
size: float
entry_price: float
current_iv: float
expiry: str
@property
def notional(self) -> float:
"""名目金額（BTC建）"""
return self.size * 1.0 # DeribitはBTC建
@property
def dollar_value(self) -> float:
"""USD建価値"""
return self.size * self.entry_price
class RiskManagementPipeline:
"""リスク管理・回测用パイプライン"""
def __init__(self, current_btc_price: float, risk_free_rate: float = 0.03):
self.S = current_btc_price
self.r = risk_free_rate
self.positions: List[OptionPosition] = []
def add_position(self, pos: OptionPosition):
"""ポジション追加"""
self.positions.append(pos)
def calculate_portfolio_greeks(self) -> Dict[str, float]:
"""
ポートフォリオ全体のGreeks集計
"""
total_delta = 0.0
total_gamma = 0.0
total_vega = 0.0
total_theta = 0.0
for pos in self.positions:
# 簡易Greeks計算
T = 0.1 # 仮定（実際は満期までの実時間）
sigma = pos.current_iv
d1 = (np.log(self.S / pos.strike) +
(self.r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
if pos.option_type == "call":
delta = np.exp(-pos.r * T) * max(d1 - 0, 1) if d1 > 0 else 0
else:
delta = -np.exp(-pos.r * T) * max(-d1, 0) if d1 < 0 else 0
gamma = np.exp(-pos.r * T) * max(d1, -d1, 0) / (self.S * sigma * np.sqrt(T))
vega = pos.size * self.S * gamma * sigma * T / 100
theta = -pos.size * self.S * gamma * sigma / (2 * 365)
total_delta += delta * pos.size
total_gamma += gamma * pos.size
total_vega += vega
total_theta += theta
return {
"delta": total_delta,
"gamma": total_gamma,
"vega": total_vega,
"theta": total_theta,
"delta_exposure_usd": total_delta * self.S,
"position_count": len(self.positions)
}
def calculate_var(
self,
confidence_level: float = 0.95,
time_horizon_days: int = 1
) -> Dict[str, float]:
"""
VaR（Value at Risk）計算
モンテカルロ・シミュレーション
"""
n_simulations = 10000
portfolio_value = sum(pos.dollar_value for pos in self.positions)
# 平均リターンとボラティリティ
daily_return_std = 0.04 # BTCの日次ボラ（仮定）
# 対数正規分布による価格シミュレーション
dt = time_horizon_days / 365
Z = np.random.standard_normal(n_simulations)
returns = np.exp(-0.5 * daily_return_std**2 * dt +
daily_return_std * np.sqrt(dt) * Z)
# ポートフォリオPL
pnl = portfolio_value * (returns - 1)
# VaR計算
var_confidence = 1 - confidence_level
var = np.percentile(pnl, var_confidence * 100)
cvar = pnl[pnl <= var].mean()
return {
"portfolio_value_usd": portfolio_value,
"var_95_1day": var,
"cvar_95_1day": cvar,
"var_pct": var / portfolio_value * 100
}
def backtest_delta_hedge(
self,
price_series: pd.Series,
hedge_interval_hours: int = 1
) -> pd.DataFrame:
"""
デルタヘッジ効果の回测
Args:
price_series: BTC価格系列
hedge_interval_hours: ヘッジ実行間隔
"""
results = []
target_delta = 0.0 # ニュートラル
for idx, (timestamp, price) in enumerate(price_series.items()):
if idx % hedge_interval_hours != 0:
continue
# 現在ポートフォリオのデルタ
current_greeks = self.calculate_portfolio_greeks()
current_delta = current_greeks["delta"]
# ヘッジ需要的先物数量
futures_hedge = (target_delta - current_delta) * self.S
hedge_cost = abs(futures_hedge * 0.0005) # 手数料
results.append({
"timestamp": timestamp,
"btc_price": price,
"portfolio_delta": current_delta,
"futures_hedge_size": futures_hedge,
"hedge_cost": hedge_cost
})
self.S = price # 価格更新
return pd.DataFrame(results)
def generate_risk_report(self) -> str:
"""リスクレポート生成"""
greeks = self.calculate_portfolio_greeks()
var = self.calculate_var()
report = f"""
=== Deribit BTC Option リスクレポート ===
生成日時: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
現物BTC価格: ${self.S:,.2f}
【ポートフォリオ概要】
- ポジジョン数: {greeks['position_count']}
- ポートフォリオ総額: ${var['portfolio_value_usd']:,.2f}
【Greeks】
- Delta: {greeks['delta']:.4f} (${greeks['delta_exposure_usd']:,.2f})
- Gamma: {greeks['gamma']:.6f}
- Vega: ${greeks['vega']:.2f} (IV+1%当)
- Theta: ${greeks['theta']:.2f} (日次)
【VaR】
- 1日VaR (95%): ${var['var_95_1day']:,.2f} ({var['var_pct']:.2f}%)
- CVaR (95%): ${var['cvar_95_1day']:,.2f}
【推奨アクション】
{"- デルタニュートラルを維持" if abs(greeks['delta']) < 0.1 else "- デルタヘッジを検討"}
{"- ガンマリスクに注意" if greeks['gamma'] > 0.001 else "- ガンマエクスポージャー低"}
"""
return report
使用例
if __name__ == "__main__":
# パイプライン初期化（BTC現在価格）
risk_pipe = RiskManagementPipeline(current_btc_price=105000)
# サンプルポジション追加
risk_pipe.add_position(OptionPosition(
instrument="BTC-26JUN26-100000-C",
strike=100000,
option_type="call",
size=1.0, # 1 BTC
entry_price=5000,
current_iv=0.85,
expiry="26JUN26"
))
risk_pipe.add_position(OptionPosition(
instrument="BTC-26JUN26-110000-P",
strike=110000,
option_type="put",
size=1.0,
entry_price=4000,
current_iv=0.80,
expiry="26JUN26"
))
# リスクレポート出力
print(risk_pipe.generate_risk_report())
# VaR計算
var_result = risk_pipe.calculate_var()
print(f"\nVaR (95%, 1日): ${var_result['var_95_1day']:,.2f}")
ステップ6：完整的データパイプライン自动化
最後に、定期的なデータ取得を自动化するパイプラインを構築します：
"""
Deribit BTC期权 完整的自动化データパイプライン
スケジュール実行・データ保存・通知
"""
import requests
import pandas as pd
import json
import sqlite3
from datetime import datetime, timedelta
from typing import List, Dict
import schedule
import time
import logging
ロギング設定
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class DeribitDataPipeline:
"""完整的データパイプライン"""
def __init__(self, holysheep_api_key: str, db_path: str = "deribit_data.db"):
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {"Authorization": f"Bearer {holysheep_api_key}"}
self.db_path = db_path
self.init_database()
def init_database(self):
"""SQLiteデータベース初期化"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# テーブル作成
cursor.execute("""
CREATE TABLE IF NOT EXISTS option_candles (
id INTEGER PRIMARY KEY AUTOINCREMENT,
instrument TEXT,
timestamp DATETIME,
open REAL, high REAL, low REAL, close REAL,
volume REAL, iv REAL,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
UNIQUE(instrument, timestamp)
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS option_greeks (
id INTEGER PRIMARY KEY AUTOINCREMENT,
instrument TEXT,
timestamp DATETIME,
delta REAL, gamma REAL, vega REAL, theta REAL,
mark_price REAL, iv REAL,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
UNIQUE(instrument, timestamp)
)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_instrument_time
ON option_candles(instrument, timestamp)
""")
conn.commit()
conn.close()
logger.info(f"データベース初期化完了: {self.db_path}")
def fetch_and_store_candles(self, instruments: List[str], days: int = 7):
"""ローソク足データ取得・保存"""
conn = sqlite3.connect(self.db_path)
for instrument in instruments:
try:
params = {
"instrument": instrument,
"start": int((datetime.now() - timedelta(days=days)).timestamp() * 1000),
"end": int(datetime.now().timestamp() * 1000)
}
response