暗号通貨のアルゴリズムトレード開發において、高品質な历史データなしには十分なバックテストは実現できません。私は2024年からOKX先物交易市场での高频トレーディングシステム構築に携わり、Tardis.devを活用したデータパイプラインの構築了大量の実証を繰り返してきました。本記事では、Tardis.devの基本設定からOKX先物データの取得、实际的なコード例、そして笔者が遇到过かった落とし穴とその対処法を详细介绍していきます。
Tardis.devとは
Tardis.devは、加密货币交易所の历史データを提供するSaaSプラットフォームです。OKX、Binance、Bybit、Deribitなどの主要取引所に対応し、约50 Tb的超高速アクセス可能なマーケットデータセットを提供します。特に先物取引のレバレッジ取引データは、スポットデータよりも取得难度が高く、Tardis.devの提供する统合されたAPIは研究者やトレーダーにとって貴重なリソースとなっています。
私の場合、OKXのUSDT先物ミニッツ数据进行收集する際、朝の注文流れパターンを分析するために、历史データが必要でした。Tardis.devは1分足データの取得が非常に简单で、Python環境があれば10分以内に最初のデータを取得できます。
OKX先物データの种类と構造
OKXの先物データは複数の種類があり、目的に合ったデータを選ぶ必要があります。以下の表は私が実際に利用したことのある数据类型をまとめたものです。
| 数据类型 | 內容 | 更新頻度 | 用途 | 価格（月额） |
|---|---|---|---|---|
| Merged Traded | 約定履歴（ Tick） | リアルタイム | 高频バックテスト | $99〜 |
| Agg Trades | 集約約定 | リアルタイム | 中频戦略検証 | $49〜 |
| Klines（1m〜1M） | OHLCV足データ | リアルタイム | 一般的な分析 | $29〜 |
| Orderbook Snapshots | 板情報スナップショット | 1秒每 | 流動性分析 | $199〜 |
| Instruments | 取引產商品情報 | 静态 | システム初期化 | 免费 |
私が高频戦略の検証に使っているのはMerged Tradedデータです。约定単位の詳細なデータを取得することで、ミリ秒単位のエントリー判定が可能になります。ただし、1 BTC先物的一年分のMerged Tradedデータは約2GBに達することもありますので、ストレージの見積もりも忘れずに行いましょう。
APIアクセスと初期設定
Tardis.devへのアクセスはREST API経由が基本となります。以下のコードはPython环境下での初期設定と认证处理の例です。
# tardis_client.py
import requests
import pandas as pd
from datetime import datetime, timedelta
import time
class TardisClient:
"""Tardis.dev APIクライアント for OKX先物データ取得"""
BASE_URL = "https://api.tardis.dev/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
})
def get_available_exchanges(self) -> list:
"""利用可能な取引所リストを取得"""
response = self.session.get(f'{self.BASE_URL}/exchanges')
response.raise_for_status()
return response.json()
def get_okx_futures_symbols(self) -> list:
"""OKX先物の利用可能な取引ペアを取得"""
response = self.session.get(
f'{self.BASE_URL}/exchanges/okx-futures/symbols'
)
response.raise_for_status()
return response.json()
def fetch_klines(
self,
symbol: str,
start_date: datetime,
end_date: datetime,
limit: int = 1000
) -> pd.DataFrame:
"""
OKX先物のK線データを取得
Args:
symbol: 取引ペア（例：BTC-USDT-SWAP）
start_date: 開始日時
end_date: 終了日時
limit: 1リクエストあたりの上限（最大1000）
Returns:
pandas.DataFrame: OHLCVデータ
"""
params = {
'symbol': symbol,
'from': int(start_date.timestamp()),
'to': int(end_date.timestamp()),
'limit': limit,
'format': 'json'
}
all_data = []
current_start = start_date
while current_start < end_date:
params['from'] = int(current_start.timestamp())
params['to'] = int(min(
current_start + timedelta(hours=24),
end_date
).timestamp())
try:
response = self.session.get(
f'{self.BASE_URL}/exchanges/okx-futures/klines',
params=params
)
response.raise_for_status()
data = response.json()
if data:
all_data.extend(data)
print(f"取得完了: {len(data)}件 (現在: {current_start})")
else:
print(f"データなし: {current_start}")
# レート制限への対応（1秒間待機）
time.sleep(1)
# 次の取得期间に移動
current_start += timedelta(hours=24)
except requests.exceptions.RequestException as e:
print(f"エラー発生: {e}")
# 指数バックオフで再試行
time.sleep(60)
continue
if not all_data:
return pd.DataFrame()
# DataFrameに変換
df = pd.DataFrame(all_data, columns=[
'timestamp', 'open', 'high', 'low', 'close', 'volume'
])
df['datetime'] = pd.to_datetime(df['timestamp'], unit='s')
df.set_index('datetime', inplace=True)
df = df.astype({
'open': float, 'high': float,
'low': float, 'close': float, 'volume': float
})
return df
使用例
if __name__ == "__main__":
client = TardisClient(api_key="YOUR_TARDIS_API_KEY")
# 利用可能なペアを確認
symbols = client.get_okx_futures_symbols()
btc_symbols = [s for s in symbols if 'BTC' in s]
print(f"BTC先物ペア: {btc_symbols[:5]}")
# 直近1日分のデータを取得
end_date = datetime.now()
start_date = end_date - timedelta(days=1)
df = client.fetch_klines(
symbol='BTC-USDT-SWAP',
start_date=start_date,
end_date=end_date
)
print(f"取得データ: {len(df)}件")
print(df.tail())
高频バックテスト용 데이터パイプライン構築
高频ストラテジーでは、分足データだけでは不十分な場合があります。约定単位のデータを取得し、それを再構築することで、板寄せ計算による理論的な約定价格を再現できます。以下はMerged Tradesデータを活用したバックテスト用データセットの構築例です。
# backtest_data_pipeline.py
import asyncio
import aiohttp
import json
import zlib
import struct
from dataclasses import dataclass
from typing import List, Optional, Iterator
from datetime import datetime, timedelta
import pandas as pd
import numpy as np
@dataclass
class Trade:
"""約定データの構造体"""
timestamp: int # ミリ秒タイムスタンプ
price: float
volume: float
side: str # 'buy' or 'sell'
trade_id: int
class BacktestDataPipeline:
"""
高频バックテスト용 データパイプライン
OKX先物のMerged Tradesから Tickデータを生成
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.tardis.dev/v1"
self._rate_limit_delay = 0.1 # 100ms
def _get_headers(self) -> dict:
return {
'Authorization': f'Bearer {self.api_key}',
'Accept-Encoding': 'gzip, deflate'
}
async def fetch_trades_async(
self,
session: aiohttp.ClientSession,
symbol: str,
start_ts: int,
end_ts: int,
page: int = 1
) -> List[Trade]:
"""非同期でMerged Tradesデータを取得"""
url = f"{self.base_url}/exchanges/okx-futures/trades"
params = {
'symbol': symbol,
'from': start_ts,
'to': end_ts,
'page': page,
'limit': 50000 # 最大50,000件
}
async with session.get(url, params=params, headers=self._get_headers()) as resp:
if resp.status == 429:
await asyncio.sleep(5)
return await self.fetch_trades_async(
session, symbol, start_ts, end_ts, page
)
data = await resp.json()
if not data or 'data' not in data:
return []
trades = []
for item in data['data']:
trades.append(Trade(
timestamp=int(item['timestamp']),
price=float(item['price']),
volume=float(item['volume']),
side='buy' if item['side'] == 'buy' else 'sell',
trade_id=int(item['id'])
))
return trades
async def download_day_trades(
self,
symbol: str,
date: datetime
) -> List[Trade]:
"""
指定日の全約定データをダウンロード
Args:
symbol: 取引ペア（例：BTC-USDT-SWAP）
date: 対象日
Returns:
全約定データのリスト
"""
start_ts = int(date.replace(hour=0, minute=0, second=0).timestamp() * 1000)
end_ts = int(date.replace(hour=23, minute=59, second=59).timestamp() * 1000)
all_trades = []
connector = aiohttp.TCPConnector(limit=10)
async with aiohttp.ClientSession(connector=connector) as session:
# ページ数を先に確認
first_page = await self.fetch_trades_async(
session, symbol, start_ts, end_ts, page=1
)
if not first_page:
print(f"{date.date()}: データなし")
return []
all_trades.extend(first_page)
print(f"{date.date()} 1ページ目: {len(first_page)}件")
# 複数ページの处理（最大10ページ）
for page in range(2, 11):
await asyncio.sleep(self._rate_limit_delay)
trades = await self.fetch_trades_async(
session, symbol, start_ts, end_ts, page=page
)
if not trades:
break
all_trades.extend(trades)
print(f"{date.date()} {page}ページ目: {len(trades)}件")
return all_trades
def trades_to_tick_bars(
self,
trades: List[Trade],
threshold: float = 0.0001
) -> pd.DataFrame:
"""
約定データをTick足に変換
Args:
trades: 約定データのリスト
threshold: 価格変動閾値（%）— この値を超えると新しい足を生成
Returns:
Tick足DataFrame
"""
if not trades:
return pd.DataFrame()
bars = []
current_bar = {
'timestamp': trades[0].timestamp,
'open': trades[0].price,
'high': trades[0].price,
'low': trades[0].price,
'close': trades[0].price,
'volume': trades[0].volume,
'tick_count': 1,
'buy_volume': trades[0].volume if trades[0].side == 'buy' else 0,
'sell_volume': trades[0].volume if trades[0].side == 'sell' else 0
}
base_price = trades[0].price
for trade in trades[1:]:
price_change = abs(trade.price - base_price) / base_price
if price_change >= threshold:
# 新しい足を生成
bars.append(current_bar)
current_bar = {
'timestamp': trade.timestamp,
'open': trade.price,
'high': trade.price,
'low': trade.price,
'close': trade.price,
'volume': trade.volume,
'tick_count': 1,
'buy_volume': trade.volume if trade.side == 'buy' else 0,
'sell_volume': trade.volume if trade.side == 'sell' else 0
}
base_price = trade.price
else:
# 現在の足を更新
current_bar['high'] = max(current_bar['high'], trade.price)
current_bar['low'] = min(current_bar['low'], trade.price)
current_bar['close'] = trade.price
current_bar['volume'] += trade.volume
current_bar['tick_count'] += 1
if trade.side == 'buy':
current_bar['buy_volume'] += trade.volume
else:
current_bar['sell_volume'] += trade.volume
bars.append(current_bar)
df = pd.DataFrame(bars)
df['datetime'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('datetime', inplace=True)
# VWAP計算
df['vwap'] = (df['close'] * df['volume']).cumsum() / df['volume'].cumsum()
return df
async def generate_backtest_dataset(
self,
symbol: str,
start_date: datetime,
end_date: datetime,
output_dir: str = './data'
) -> pd.DataFrame:
"""
バックテスト用データセットを生成
全期間の約定データをダウンロードし Tick足に変換
"""
import os
os.makedirs(output_dir, exist_ok=True)
all_bars = []
current_date = start_date
while current_date <= end_date:
date_str = current_date.strftime('%Y%m%d')
csv_path = f"{output_dir}/{symbol}_{date_str}.csv"
if os.path.exists(csv_path):
print(f"{date_str}: キャッシュ済みスキップ")
df = pd.read_csv(csv_path, parse_dates=['datetime'], index_col='datetime')
all_bars.append(df)
current_date += timedelta(days=1)
continue
trades = await self.download_day_trades(symbol, current_date)
if trades:
bars = self.trades_to_tick_bars(trades)
bars.to_csv(csv_path)
print(f"{date_str}: {len(bars)}足のTick足を生成")
all_bars.append(bars)
else:
# 空ファイルを作成してスキップ
pd.DataFrame().to_csv(csv_path)
current_date += timedelta(days=1)
await asyncio.sleep(1) # レート制限対応
if all_bars:
return pd.concat(all_bars)
return pd.DataFrame()
使用例
async def main():
pipeline = BacktestDataPipeline(api_key="YOUR_TARDIS_API_KEY")
# 1週間分のデータを取得
end_date = datetime.now()
start_date = end_date - timedelta(days=7)
df = await pipeline.generate_backtest_dataset(
symbol='BTC-USDT-SWAP',
start_date=start_date,
end_date=end_date,
output_dir='./okx_btc_backtest'
)
print(f"総Tick足数: {len(df)}")
print(df.describe())
# データ分析：出来高分布
df['hour'] = df.index.hour
volume_by_hour = df.groupby('hour')['volume'].mean()
print("\n時間帯別平均出来高:")
print(volume_by_hour)
if __name__ == "__main__":
asyncio.run(main())
HolySheep AIとの統合：機械学習分析パイプライン
収集したバックテストデータは、そのままでは戦略の最適化には不十分です。出来高パターン認識、異常値検出、及市场構造の分類などには、機械学習モデルの活用が効果的です。HolySheep AIは такие задачиに最适合なAI APIを提供しており、私が普段使っている統合アーキテクチャをご紹介します。
| サービス | 主な用途 | レイテンシ | コスト効率 | 特徴 |
|---|---|---|---|---|
| HolySheep AI | パターン分析・異常検知 | <50ms | ¥1=$1（業界の85%节省） | WeChat Pay/Alipay対応、低遅延 |
| OpenAI | 高度な推論 | 200-500ms | $7.5/MTok | 信頼性高いがコスト高 |
| Anthropic | 長いコンテキスト処理 | 300-800ms | $15/MTok | コンテキスト長いが高コスト |
| コスト重視の推論 | 100-300ms | $2.5/MTok | バランス型 |
私は HolySheep AI を活用して、取得データを批量で分析し、出来高急増パターンや価格 操作の疑いがある足を自動検出するシステムを構築しています。特に DeepSeek V3.2 の出力コストが $0.42/MTok と非常に低く、大量データ処理に最適な点が嬉しいです。
# ai_analysis_pipeline.py
import os
import json
import asyncio
import aiohttp
from typing import List, Dict, Optional
from dataclasses import dataclass
from datetime import datetime
import pandas as pd
from concurrent.futures import ThreadPoolExecutor
@dataclass
class BacktestResult:
"""バックテスト分析结果"""
timestamp: datetime
pattern_type: str
confidence: float
description: str
risk_level: str # 'low', 'medium', 'high'
recommendation: str
class HolySheepClient:
"""
HolySheep AI APIクライアント
バックテスト分析用の軽量・高速推論
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
def analyze_volume_pattern(
self,
bar_data: dict,
model: str = "gpt-4.1"
) -> BacktestResult:
"""
単一バーの出来高パターンを分析
Args:
bar_data: OHLCVデータ辞書
model: 使用モデル（デフォルト: gpt-4.1）
Returns:
BacktestResult: 分析结果
"""
prompt = f"""
あなたは加密货币取引のテクニachalアナリストです。
以下の1分足を分析し、パターン分類を行ってください。
【データ】
- 時間: {bar_data.get('datetime')}
- 始値: {bar_data.get('open')}
- 高値: {bar_data.get('high')}
- 安値: {bar_data.get('low')}
- 終値: {bar_data.get('close')}
- 出来高: {bar_data.get('volume')}
- VWAP: {bar_data.get('vwap')}
- 買い出来高: {bar_data.get('buy_volume', 0)}
- 売り出来高: {bar_data.get('sell_volume', 0)}
【分析項目】
1. パターンタイプ（ прямоугольник、三角持ち合い、上昇トレンド、下降トレンド等）
2. 置信度（0-1）
3. リスクレベル（low/medium/high）
4. 推奨アクション
JSON形式で返答してください:
{{
"pattern_type": "string",
"confidence": float,
"description": "string",
"risk_level": "string",
"recommendation": "string"
}}
"""
payload = {
"model": model,
"messages": [
{"role": "system", "content": "あなたは 전문적인加密货币アナリストです。"},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 500
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
response = aiohttp.request(
"POST",
f"{self.BASE_URL}/chat/completions",
json=payload,
headers=headers
)
return self._parse_response(response)
def batch_analyze_volume_patterns(
self,
df: pd.DataFrame,
batch_size: int = 10,
model: str = "deepseek-v3.2"
) -> List[BacktestResult]:
"""
複数バーの出来高パターンを一括分析
Args:
df: Tick足/分足データ
batch_size: 批量处理サイズ
model: 使用モデル（コスト重視はdeepseek-v3.2）
Returns:
List[BacktestResult]: 分析结果リスト
"""
results = []
# 特徴量计算
df = df.copy()
df['price_change'] = df['close'].pct_change()
df['volume_ma'] = df['volume'].rolling(window=20).mean()
df['volume_ratio'] = df['volume'] / df['volume_ma']
df['buy_ratio'] = df.get('buy_volume', df['volume'] * 0.5) / df['volume']
bar_dict = df.to_dict('records')
# 批量リクエスト生成
prompts = []
for i in range(0, len(bar_dict), batch_size):
batch = bar_dict[i:i+batch_size]
prompt = f"""以下の{batch_size}件の加密货币足データを分析し、
各足の売買バランスと出来高异常を検出してください。
"""
for idx, bar in enumerate(batch):
prompt += f"""
【足 {idx+1}】
時間: {bar.get('datetime')}
終値: {bar.get('close')}
出来高: {bar.get('volume', 0):.2f}
出来高比率: {bar.get('volume_ratio', 1):.2f}
買い比率: {bar.get('buy_ratio', 0.5):.2f}
"""
prompt += """
各足について以下のJSON配列形式で返答:
[
{{"idx": 0, "pattern_type": "...", "confidence": 0.0-1.0, "risk_level": "...", "recommendation": "..."}},
...
]
"""
prompts.append((i, prompt))
# 並列処理でリクエスト送信
def analyze_batch(args):
idx, prompt = args
payload = {
"model": model,
"messages": [
{"role": "system", "content": "简洁扼要な分析をJSONで返答"},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 1000
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
try:
response = aiohttp.request(
"POST",
f"{self.BASE_URL}/chat/completions",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=30)
)
data = response.json()
if 'choices' in data:
content = data['choices'][0]['message']['content']
return idx, json.loads(content)
except Exception as e:
print(f"Batch {idx} Error: {e}")
return idx, []
with ThreadPoolExecutor(max_workers=5) as executor:
batch_results = list(executor.map(analyze_batch, prompts))
# 結果マージ
all_results = []
for idx, batch_result in sorted(batch_results, key=lambda x: x[0]):
if isinstance(batch_result, list):
all_results.extend(batch_result)
elif isinstance(batch_result, dict):
all_results.append(batch_result)
# BacktestResultに変換
for i, res in enumerate(all_results):
results.append(BacktestResult(
timestamp=pd.to_datetime(bar_dict[i].get('datetime')),
pattern_type=res.get('pattern_type', 'unknown'),
confidence=res.get('confidence', 0),
description=res.get('description', ''),
risk_level=res.get('risk_level', 'medium'),
recommendation=res.get('recommendation', '')
))
return results
使用例
if __name__ == "__main__":
holysheep = HolySheepClient(api_key="YOUR_HOLYSHEEP_API_KEY")
# テストデータ生成
test_data = []
for i in range(100):
test_data.append({
'datetime': datetime.now(),
'open': 65000 + i * 10,
'high': 65100 + i * 10,
'low': 64900 + i * 10,
'close': 65050 + i * 10,
'volume': 100 + abs(i % 20 - 10) * 20,
'vwap': 65050 + i * 10,
'buy_volume': 60 + (i % 10) * 4,
'sell_volume': 40 + (i % 10) * 4
})
df = pd.DataFrame(test_data)
# パターン分析実行
results = holysheep.batch_analyze_volume_patterns(
df,
batch_size=10,
model="deepseek-v3.2" # 低コストで高效
)
print(f"分析完了: {len(results)}件")
# 高リスク足を抽出
high_risk = [r for r in results if r.risk_level == 'high']
print(f"高リスク足: {len(high_risk)}件")
for result in high_risk[:5]:
print(f" {result.timestamp}: {result.pattern_type} ({result.confidence:.2f})")
向いている人・向いていない人
| 向いている人 | 向いていない人 |
|---|---|
| ✓ 量化投資を始める個人投資家・ conmem | ✗ 初期費用を払いたくない完全無料主義者 |
| ✓ 高频トレード戦略の検証を行うpropトレーダー | ✗ 単発の利用で十分な轻用量ユーザー |
| ✓ 加密货币取引データ研究を行う学术研究者 | ✗ 即時に全上場通貨のTickデータを必要とする人 |
| ✓ 機械学習モデル用の教師データ収集を検討しているMLエンジニア | ✗ 中国語サポートのみで十分な方 |
| ✓ 複数取引所の比較分析を行いたいクオンツチーム | ✗ 自前でKafka/リアルタイムストリーミングを構築できる大企業 |
価格とROI
Tardis.devの料金体系は、利用する数据类型と期間によって大きく異なります。以下に私が实战で見積もりを行った具体例をまとめます。
| 利用ケース | 数据类型 | 期間 | 推定コスト | 取代案との比較 |
|---|---|---|---|---|
| 个人トレーダー（バックテスト検証） | Klines 1m | 1年 | $29/月 × 12 = $348 | 自前構築：API費用$200+ 時間コスト$500 |
| クオンツ фонд（戦略开发） | Merged Trades | 2年 | $99/月 × 24 = $2,376 | 取引所直リクエスト：延迟大・可靠性低 |
| 学術研究（論文発表） | Klines + Agg Trades | 6ヶ月 | $49/月 × 6 = $294 | 手動収集：非現実的 |
| AI分析パイプライン（HolySheep統合） | Merged Trades | 1年 | $99/月 + 分析コスト$50 | HolySheep: $0.42/MTok × 1M = $420 |
HolySheep AIを併用する場合のデータ分析コストも考虑に入れる必要があります。例えば、100万件の Tick足を分析する場合、DeepSeek V3.2（$0.42/MTok）では非常に経済的です。一方、GPT-4.1（$8/MTok）ではコストが跳ね上がりますが、より高精度な分析が必要な场合には投资対効果が高いと言えます。
よくあるエラーと対処法
エラー1: 429 Too Many Requests
原因: APIリクエストのレート制限超过了
# ❌ 错误的実装
for i in range(1000):
data = fetch_trades(page=i) # 即座に1000リクエスト送信
✓ 正しい実装（指数バックオフ）
import time
def fetch_with_retry(url, max_retries=5):
for attempt in range(max_retries):
try:
response = requests.get(url)
if response.status_code == 429:
wait_time = 2 ** attempt + random.uniform(0, 1)
print(f"レート制限: {wait_time:.1f}秒待機...")
time.sleep(wait_time)
continue
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
if attempt == max_retries - 1:
raise
time.sleep(2 ** attempt)
return None
エラー2: データ欠損（Gap in data）
原因: メンテナンス期间的やAPI障害でデータがない
# データギャップ检测と补間
def detect_and_fill_gaps(df: pd.DataFrame, max_gap_seconds: int = 60) -> pd.DataFrame:
"""
欠損データを検出し、线形補間で埋める
Args:
df: タイムインデックスを持つDataFrame
max_gap_seconds: この値を超える空白を欠損として検出
Returns:
ギャップを埋めたDataFrame
"""
df = df.copy()
# 時間インデックスを確認
if not isinstance(df.index, pd.DatetimeIndex):
df.index = pd.to_datetime(df.index)
# ギャップ检测
time_diff = df.index.to_series().diff()
gaps = time_diff[time_diff > pd.Timedelta(seconds=max_gap_seconds)]
if not gaps.empty:
print(f"⚠️ {len(gaps)}件のギャップを検出:")
for idx, gap in gaps.items():
print(f" {idx}: +{gap}")
# 前後の値の 평균で補間（简单な方法）
# より高度な方法が必要な場合は.interpolate(method='time')
numeric_cols = df.select_dtypes(include=[np.number]).columns
df[numeric_cols] = df[numeric_cols].interpolate(method='time')
print("✅ ギャップを補間完了")
return df
エラー3: シンボル名不正確（Symbol not found）
原因: OKXのシンボル命名規則が異なる
# OKX先物シンボル マッピング
OKX_FUTURES_SYMBOLS = {
# 現物先物（SWAP）
'BTC-USDT-SWAP': 'BTC-USDT-SWAP',
'ETH-USDT-SWAP': 'ETH-USDT-SWAP',
# 季度先物（この日期により名称変更）
'BTC-USDT-240628': 'BTC-USDT-240628', # 2024年6月先物
'BTC-USDT-240329': 'BTC-USDT-240329', # 2024年3月先物
# 取り消し済み先物（历史データのみ）
'BTC-USDT-231229': 'BTC-USDT-231229', # 2023年12月先物（既に満期）
}
利用可能なシンボル確認
def list_available_okx_symbols(api_key: str) -> dict:
"""OKX先物の利用可能な全シンボルを取得"""
url = "https://api.tardis.dev/v1/exchanges/okx-futures/symbols"
headers = {"Authorization": f"Bearer {api_key}"}
response = requests.get(url, headers=headers)
response.raise_for_status()
symbols = response.json()
# カデゴリ別に整理
return {
'swap': [s