暗号資産取引の世界で、自动取引戦略の効果を検証するには、历史的なTick 数据が 필수입니다. このガイドでは、OKX取引所の永久先物（永续合约）からTardisを使用して 高品質なTickデータを取得し、Pythonでバックテスト框架を構築する方法を段階的に説明します.
Tardisとは
Tardisは криптовалют биржаからリアルタイムおよび历史的なマーケットデータを提供するSaaSプラットフォームです. OKX、Binance、Bybitなど 主要取引所のTickデータを 直接Pythonから利用でき、自作の取引戦略検証に最適です.
前提条件
- Python 3.8 이상
- Tardis.ai アカウント（무료 开始可能）
- OKX 先物거래权限
- 基本的なPython知識
1단계: Tardis Python SDK 설치
먼저 Tardis-machine 패키지를 설치합니다. 이 패키지는 OKX를 포함한 여러 거래소 데이터에 접근할 수 있게 해줍니다.
# Tardis SDK 설치
pip install tardis-machine
추가 의존성
pip install pandas numpy matplotlib
설치 확인
python -c "import tardis; print(tardis.__version__)"2단계: Tardis API 키 설정
Tardis는 API 키를 통해 데이터를 가져옵니다. 아래 코드로 자격 증명을 설정하세요.
import os
Tardis API 키 설정 (환경 변수 권장)
os.environ['TARDIS_API_KEY'] = 'your_tardis_api_key_here'
os.environ['TARDIS_API_PASSWORD'] = 'your_tardis_api_password_here'
확인
from tardis import Tardis
client = Tardis()
계정 정보 확인
print(client.account())
출력 예시: {'exchanges': ['okx', 'binance'], 'credits': 15000}
3단계: OKX永续合约Tickデータ取得
이제 OKX BTC/USDT永久先物 계약에서 특정 기간의 Tick 데이터를 가져와 보겠습니다.
from tardis_client import TardisClient
import asyncio
from datetime import datetime, timedelta
async def fetch_okx_tick_data():
"""OKX BTC/USDT永久先物 Tickデータ取得"""
client = TardisClient(api_key=os.environ['TARDIS_API_KEY'])
# 取得パラメータ設定
exchange = 'okx'
symbol = 'BTC-USDT-SWAP' # OKX永续合约形式
from_time = datetime(2025, 1, 1, 0, 0, 0) # 開始時刻
to_time = datetime(2025, 1, 2, 0, 0, 0) # 終了時刻
# TardisTimestamp形式に変換
from_timestamp = int(from_time.timestamp() * 1000)
to_timestamp = int(to_time.timestamp() * 1000)
# Tickデータストリーム
messages = client.replay(
exchange=exchange,
filters=[{
'type': 'trade',
'symbols': [symbol]
}],
from_timestamp=from_timestamp,
to_timestamp=to_timestamp
)
trades = []
async for message in messages:
if message['type'] == 'trade':
trades.append({
'timestamp': message['timestamp'],
'symbol': message['symbol'],
'side': message['side'],
'price': float(message['price']),
'amount': float(message['amount']),
'fee': message.get('fee', 0),
'contract_id': message.get('contractId', None)
})
return trades
データ取得実行
trades = asyncio.run(fetch_okx_tick_data())
print(f"取得完了: {len(trades)}件のTickデータを取得")
print(f"時間帯: {trades[0]['timestamp']} ~ {trades[-1]['timestamp']}")4단계: データ前処理と保存
取得したTickデータを 分析しやすい形式に変換し、Pandas DataFrameで保存します.
import pandas as pd
def process_tick_data(trades):
"""TickデータをDataFrameに変換・処理"""
df = pd.DataFrame(trades)
# タイムスタンプ変換
df['datetime'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('datetime', inplace=True)
# 価格系列リサンプリング（1分足）
df_1m = pd.DataFrame()
df_1m['open'] = df['price'].resample('1min').first()
df_1m['high'] = df['price'].resample('1min').max()
df_1m['low'] = df['price'].resample('1min').min()
df_1m['close'] = df['price'].resample('1min').last()
df_1m['volume'] = df['amount'].resample('1min').sum()
# 欠損値処理
df_1m.fillna(method='ffill', inplace=True)
return df, df_1m
データ処理
tick_df, ohlc_df = process_tick_data(trades)
print("=== Tickデータサンプル ===")
print(tick_df.head(10))
print(f"\n=== 1分足OHLCサンプル ===")
print(ohlc_df.head(10))
CSV保存
tick_df.to_csv('okx_btcusdt_ticks.csv')
ohlc_df.to_csv('okx_btcusdt_ohlc_1m.csv')
print("\n✅ データをCSVファイルとして保存しました")5단계: 単純移動平均線バックテスト
取得・加工したデータを使用して、簡単なSMA（単純移動平均線）クロステラリーのバックテストを実行します.
import numpy as np
class SimpleBacktester:
"""简单移動平均線（SMA）バックテストクラス"""
def __init__(self, df, fast_period=10, slow_period=30):
self.df = df.copy()
self.fast_period = fast_period
self.slow_period = slow_period
self.initial_balance = 10000 # 初始資金 USDT
self.balance = self.initial_balance
self.position = 0 # 持仓数量
self.trades = []
def calculate_indicators(self):
"""SMA計算"""
self.df['fast_sma'] = self.df['close'].rolling(self.fast_period).mean()
self.df['slow_sma'] = self.df['close'].rolling(self.slow_period).mean()
def run(self):
"""バックテスト実行"""
self.calculate_indicators()
for i in range(self.slow_period, len(self.df)):
current = self.df.iloc[i]
previous = self.df.iloc[i-1]
# ゴールデンクロス: 短期SMAが長期SMAを上抜ける
if previous['fast_sma'] < previous['slow_sma'] and current['fast_sma'] > current['slow_sma']:
if self.position == 0: # 持仓なし → 买入
self.position = self.balance / current['close']
self.balance = 0
self.trades.append({
'datetime': current.name,
'action': 'BUY',
'price': current['close'],
'position': self.position
})
# デッドクロス: 短期SMAが長期SMAを下抜ける
elif previous['fast_sma'] > previous['slow_sma'] and current['fast_sma'] < current['slow_sma']:
if self.position > 0: # 持仓あり → 売却
self.balance = self.position * current['close']
self.trades.append({
'datetime': current.name,
'action': 'SELL',
'price': current['close'],
'balance': self.balance
})
self.position = 0
# 最終持仓を売却
if self.position > 0:
final_price = self.df.iloc[-1]['close']
self.balance = self.position * final_price
self.trades.append({
'datetime': self.df.index[-1],
'action': 'FINAL_SELL',
'price': final_price,
'balance': self.balance
})
def get_results(self):
""" результаты возврата """
total_return = (self.balance - self.initial_balance) / self.initial_balance * 100
num_trades = len([t for t in self.trades if t['action'] in ['BUY', 'SELL']]) // 2
return {
'initial_balance': self.initial_balance,
'final_balance': self.balance,
'total_return_pct': total_return,
'num_trades': num_trades,
'trades': self.trades
}
バックテスト実行
backtester = SimpleBacktester(ohlc_df, fast_period=10, slow_period=30)
backtester.run()
results = backtester.get_results()
print("=== バックテスト結果 ===")
print(f"初始資金: ${results['initial_balance']:.2f}")
print(f"最終資金: ${results['final_balance']:.2f}")
print(f"総損益: {results['total_return_pct']:.2f}%")
print(f"取引回数: {results['num_trades']}回")
print(f"\n=== 取引履歴 ===")
for trade in results['trades']:
print(f"{trade['datetime']} | {trade['action']} | ${trade['price']:.2f}")6단계: 可视化
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
def plot_backtest_results(df, backtester):
"""バックテスト結果可視化"""
fig, axes = plt.subplots(2, 1, figsize=(14, 10), sharex=True)
# 価格・SMAチャート
ax1 = axes[0]
ax1.plot(df.index, df['close'], label='Close Price', alpha=0.7, linewidth=1)
ax1.plot(df.index, df['fast_sma'], label=f"SMA{backtester.fast_period}", linewidth=1.5)
ax1.plot(df.index, df['slow_sma'], label=f"SMA{backtester.slow_period}", linewidth=1.5)
# 取引ポイント标记
for trade in backtester.trades:
if trade['action'] == 'BUY':
ax1.scatter(trade['datetime'], trade['price'], color='green', marker='^', s=100, zorder=5)
elif trade['action'] == 'SELL':
ax1.scatter(trade['datetime'], trade['price'], color='red', marker='v', s=100, zorder=5)
ax1.set_ylabel('価格 (USDT)')
ax1.set_title('OKX BTC/USDT 永续合约 - SMAバックテスト')
ax1.legend()
ax1.grid(True, alpha=0.3)
# 資金曲線
ax2 = axes[1]
balance_history = [backtester.initial_balance]
dates = [df.index[backtester.slow_period]]
for trade in backtester.trades:
if trade['action'] == 'BUY':
balance_history.append(balance_history[-1])
dates.append(trade['datetime'])
elif trade['action'] in ['SELL', 'FINAL_SELL']:
balance_history.append(trade['balance'])
dates.append(trade['datetime'])
ax2.plot(dates, balance_history, color='blue', linewidth=2)
ax2.fill_between(dates, balance_history, alpha=0.3)
ax2.set_ylabel('资金 (USDT)')
ax2.set_xlabel('日時')
ax2.set_title('资金曲線')
ax2.grid(True, alpha=0.3)
ax2.xaxis.set_major_formatter(mdates.DateFormatter('%m-%d %H:%M'))
plt.xticks(rotation=45)
plt.tight_layout()
plt.savefig('backtest_result.png', dpi=150, bbox_inches='tight')
plt.show()
print("📊 チャートを 'backtest_result.png' として保存しました")
可視化実行
plot_backtest_results(ohlc_df, backtester)자주 발생하는 오류 해결
오류 1: Tardis API 키 인증 실패
# ❌ 错误示例
from tardis_client import TardisClient
client = TardisClient(api_key='invalid_key')
✅ 解決方法: 올바른 API 키 확인
import os
환경 변수에서 API 키 로드
TARDIS_API_KEY = os.environ.get('TARDIS_API_KEY')
TARDIS_API_PASSWORD = os.environ.get('TARDIS_API_PASSWORD')
if not TARDIS_API_KEY or not TARDIS_API_PASSWORD:
raise ValueError("TARDIS_API_KEY와 TARDIS_API_PASSWORD 환경 변수를 설정하세요")
client = TardisClient(api_key=TARDIS_API_KEY)
연결 테스트
try:
account = client.account()
print(f"✅ 인증 성공: 사용 가능 크레딧 {account['credits']}")
except Exception as e:
print(f"❌ 인증 실패: {e}")
# Tardis 대시보드에서 새 API 키 생성
# https://app.tardis.dev/profile/api-keys오류 2: OKX 계약 심볼 형식 오류
# ❌ 잘못된 심볼 형식
symbol = 'BTC-USDT' # 현물 형식
✅ 올바른 OKX 무기계약 심볼 형식
symbol = 'BTC-USDT-SWAP' # 무기계약
전체 지원 심볼 확인
async def list_okx_symbols():
client = TardisClient(api_key=TARDIS_API_KEY)
exchange_info = client.exchange('okx')
#永续合约 심볼만 필터링
swap_symbols = [
s for s in exchange_info.symbols
if 'SWAP' in s or 'perpetual' in str(exchange_info.symbols[s])
]
print("=== OKX 支持的永续合约 ===")
for sym in swap_symbols[:10]:
print(f" - {sym}")
return swap_symbols
symbols = asyncio.run(list_okx_symbols())오류 3: 데이터 복원 중 타임아웃 오류
# ❌ 대량 데이터 요청 시 타임아웃
messages = client.replay(
exchange='okx',
filters=[...],
from_timestamp=from_ts,
to_timestamp=to_ts # 너무 긴 기간
)
✅ 해결 방법: 기간 분할 및 재시도 로직
async def fetch_data_with_retry(client, exchange, filters, from_ts, to_ts, max_retries=3):
"""대량 데이터 복원 시 재시도 로직 포함"""
chunk_size = 24 * 60 * 60 * 1000 # 1일 단위
all_messages = []
current_start = from_ts
retry_count = 0
while current_start < to_ts:
current_end = min(current_start + chunk_size, to_ts)
for attempt in range(max_retries):
try:
messages = client.replay(
exchange=exchange,
filters=filters,
from_timestamp=current_start,
to_timestamp=current_end
)
async for msg in messages:
all_messages.append(msg)
print(f"✅ {pd.to_datetime(current_start, unit='ms')} ~ {pd.to_datetime(current_end, unit='ms')} 완료")
break
except Exception as e:
retry_count += 1
if attempt < max_retries - 1:
wait_time = 2 ** attempt # 지수 백오프
print(f"⚠️ 재시도 {attempt + 1}/{max_retries}, {wait_time}초 대기...")
await asyncio.sleep(wait_time)
else:
print(f"❌ Chunk 실패: {e}")
current_start = current_end
return all_messages
사용 예시
messages = await fetch_data_with_retry(
client=client,
exchange='okx',
filters=[{'type': 'trade', 'symbols': ['BTC-USDT-SWAP']}],
from_timestamp=from_ts,
to_timestamp=to_ts
)추가 팁과 모범 사례
데이터 품질 확인
def validate_tick_data(df):
"""Tickデータ品質 검증"""
issues = []
# 欠損値チェック
missing = df.isnull().sum()
if missing.any():
issues.append(f"欠損値: {missing[missing > 0].to_dict()}")
# 異常値チェック（価格）
price_zscore = (df['price'] - df['price'].mean()) / df['price'].std()
outliers = df[abs(price_zscore) > 5]
if not outliers.empty:
issues.append(f"異常値（5σ超）: {len(outliers)}件")
# 时间순チェック
if not df.index.is_monotonic_increasing:
issues.append("⚠️ 時刻が순서대로でない行が存在")
return {
'is_valid': len(issues) == 0,
'issues': issues,
'statistics': {
'total_ticks': len(df),
'time_range': f"{df.index[0]} ~ {df.index[-1]}",
'avg_spread': (df['high'] - df['low']).mean(),
'volume_total': df['volume'].sum()
}
}
validation = validate_tick_data(tick_df)
print("=== データ品質検証 ===")
print(f"有効性: {'✅ OK' if validation['is_valid'] else '❌ 要確認'}")
if validation['issues']:
for issue in validation['issues']:
print(f" - {issue}")
print(f"\n=== 統計情報 ===")
for k, v in validation['statistics'].items():
print(f" {k}: {v}")실전 최적화 제안
- 데이터 압축 저장: Parquet 형식으로 변환하면 CSV 대비 70% 저장 공간 절약
- 필요한 데이터만 취득: 모든 거래가 아닌 필터 조건을 사용하여 API 호출 최소화
- 지역 캐싱: 자주 사용하는 데이터셋을 로컬에 저장하여 반복 비용 절감
- 실시간 스트리밍: 백테스트 완료 후 같은 코드로 실시간 거래 가능
다음 단계
이 튜토리얼에서는 OKX永续合约의 Tick 데이터를 Tardis로 가져오고 Python으로 基本バックテスト框架を構築する方法を 학습했습니다. 다음 단계로는以下を学びことができます:
- より高度なインディケーター（RSI、MACD、 ボリンジャーバンド）実装
- リスク管理（ポジショニングSized、ストップロス、テイクプロフィット）追加
- Tardis 실시간ストリーミングへの移行
- 여러取引소・다수銘柄の同時バックテスト
криптовалют 거래 시스템을構築する際 HolySheep AI를 함께 활용하면、AI 기반 시장 분석 및 감정 분석을 백테스트 시스템과 통합하여 더욱 발전된 거래 전략을 开发할 수 있습니다.
👉 HolySheep AI 가입하고 무료 크레딧 받기