自作の取引戦略をバックテストしたい。でも「tickデータってなに？」「APIってどう使うの？」という状態から始まった筆者と同じレベルのあなたへ。この記事では、OKX exchangeの歷史 tick データをREST APIでダウンロードし、バックテスト用水準に清洗する方法をゼロ부터解説します。
tickデータとは？バックテスト为什么要用？
tickデータは、ago約1回の取引情報を記録した最小単位のデータです。每一tickに以下の情報が含まれます：
- timestamp：取引時刻（ミリ秒精度）
- price：約定価格
- volume：取引数量
- side：買い（buy）または売り（sell）
FXや股票的每日收盘データではなく、tickレベルで分析することで、短時間で地利確約や約定スリッページなどの現実的なシミュレーションが可能になります。筆者も最初は日足データでバックテストしていましたが、實際の取引では約定しないケースが多すぎて信頼性が低かったのです。
所需环境整备
以下の環境を事前に準備してください：
- Python 3.8以上
- pip（Pythonパッケージマネージャー）
- Internet соединение
必要なライブラリをインストールします：
pip install requests pandas numpy python-dateutilOKX REST API 基本構造
OKXは金融系cryptocurrency exchangeとして、历史データを取得するための公开REST APIを提供しています。このAPIは認証不要のエンドポイント比较多のため、気軽に试用开始できます。
ベースURLは以下の通りです：
OKX_PUBLIC_BASE_URL = "https://www.okx.com"
OKX_TRADE_DATA_ENDPOINT = "/api/v5/market/trades"【実践コード1】单一交易对の历史 tickデータ下载
import requests
import pandas as pd
from datetime import datetime, timezone
import time
class OKXHistoricalDataDownloader:
"""OKX исторических тиковых данных REST скачиваниеと清洗"""
def __init__(self):
self.base_url = "https://www.okx.com"
self.session = requests.Session()
self.session.headers.update({
'Content-Type': 'application/json',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
})
def get_trades(self, inst_id: str, limit: int = 100) -> pd.DataFrame:
"""
指定取引対象の历史tickデータを取得
Args:
inst_id: 取引対象ID（例：BTC-USDT）
limit: 取得件数（最大100）
Returns:
tickデータDataFrame
"""
endpoint = "/api/v5/market/trades"
params = {
'instId': inst_id,
'limit': limit
}
url = f"{self.base_url}{endpoint}"
print(f"[INFO] データ取得中: {inst_id}")
response = self.session.get(url, params=params)
response.raise_for_status()
data = response.json()
if data.get('code') != '0':
raise ValueError(f"APIエラー: {data.get('msg')}")
trades = data.get('data', [])
df = pd.DataFrame(trades)
# カラム名を统一
df = df.rename(columns={
'instId': 'symbol',
'tradeId': 'trade_id',
'px': 'price',
'sz': 'volume',
'side': 'side',
'ts': 'timestamp',
'tickCtx': 'tick_ctx'
})
# timestampをdatetimeに変換
df['datetime'] = pd.to_datetime(
df['timestamp'].astype(np.int64),
unit='ms',
utc=True
)
df['datetime'] = df['datetime'].dt.tz_convert('Asia/Shanghai')
# numeric型的转换
df['price'] = df['price'].astype(float)
df['volume'] = df['volume'].astype(float)
print(f"[SUCCESS] {len(df)}件のtickデータを取得")
return df
def get_historical_trades(self, inst_id: str, start_date: str, end_date: str) -> pd.DataFrame:
"""
指定期間内の全历史tickデータを递归取得
Args:
inst_id: 取引対象ID
start_date: 開始日（YYYY-MM-DD）
end_date: 終了日（YYYY-MM-DD）
"""
start_ts = pd.Timestamp(start_date, tz='Asia/Shanghai').timestamp() * 1000
end_ts = pd.Timestamp(end_date, tz='Asia/Shanghai').timestamp() * 1000
all_trades = []
after_ts = start_ts
request_count = 0
while after_ts < end_ts:
params = {
'instId': inst_id,
'after': str(int(after_ts)), # afterパラメータで页送
'limit': 100
}
url = f"{self.base_url}/api/v5/market/trades"
response = self.session.get(url, params=params)
if response.status_code != 200:
print(f"[WARN] リクエスト失败: {response.status_code}")
time.sleep(2)
continue
data = response.json()
trades = data.get('data', [])
if not trades:
break
all_trades.extend(trades)
after_ts = int(trades[-1]['ts']) + 1
request_count += 1
# Rate Limit対応（每分120リクエスト）
time.sleep(0.5)
if request_count % 100 == 0:
print(f"[INFO] {len(all_trades)}件取得完了...")
df = pd.DataFrame(all_trades)
if df.empty:
return df
# データ清洗処理
df = df.rename(columns={
'instId': 'symbol',
'tradeId': 'trade_id',
'px': 'price',
'sz': 'volume',
'side': 'side',
'ts': 'timestamp'
})
df['datetime'] = pd.to_datetime(df['timestamp'].astype(np.int64), unit='ms', utc=True)
df['datetime'] = df['datetime'].dt.tz_convert('Asia/Shanghai')
df['price'] = df['price'].astype(float)
df['volume'] = df['volume'].astype(float)
print(f"[COMPLETE] 合计 {len(df)}件のtickデータを取得")
return df
使用例
downloader = OKXHistoricalDataDownloader()
BTC-USDTの单日数据获取
df_btc = downloader.get_trades(inst_id='BTC-USDT', limit=100)
print(df_btc.head(10))【実践コード2】バックテスト用水準へのデータ清洗
import pandas as pd
import numpy as np
from typing import Tuple, Optional
import warnings
warnings.filterwarnings('ignore')
class TickDataCleaner:
"""tick数据的清洗・リサンプリング处理"""
def __init__(self, data_path: Optional[str] = None, df: Optional[pd.DataFrame] = None):
"""
Args:
data_path: CSVファイルパス
df: 内存DataFrame
"""
if data_path:
self.df = pd.read_csv(data_path, parse_dates=['datetime'])
elif df is not None:
self.df = df.copy()
else:
raise ValueError("data_pathまたはdfのいずれかを指定してください")
def basic_cleaning(self) -> 'TickDataCleaner':
"""基本的清洗处理"""
df = self.df.copy()
# 1. 重複除去
before_len = len(df)
df = df.drop_duplicates(subset=['timestamp', 'trade_id'])
after_len = len(df)
if before_len != after_len:
print(f"[CLEAN] 重複データ除去: {before_len} → {after_len} ({before_len - after_len}件)")
# 2. 異常値去除
df = self._remove_outliers(df)
# 3. 欠損値処理
df = df.dropna(subset=['price', 'volume'])
# 4. 时间序列排序
df = df.sort_values('datetime').reset_index(drop=True)
# 5. インデックス设定
df = df.set_index('datetime')
self.df = df
return self
def _remove_outliers(self, df: pd.DataFrame, price_std_threshold: float = 5.0) -> pd.DataFrame:
"""価格异常値の除去"""
df = df.copy()
# 价格变动率检测
df['price_pct_change'] = df['price'].pct_change()
# 5标准差を超える变动を异常値として标记
mean_pct = df['price_pct_change'].mean()
std_pct = df['price_pct_change'].std()
outlier_mask = np.abs(df['price_pct_change'] - mean_pct) > (std_pct * price_std_threshold)
outlier_count = outlier_mask.sum()
if outlier_count > 0:
print(f"[CLEAN] 異常値除去: {outlier_count}件 ({outlier_count/len(df)*100:.2f}%)")
df.loc[outlier_mask, 'price'] = np.nan
df['price'] = df['price'].interpolate(method='linear')
df = df.drop(columns=['price_pct_change'])
return df
def resample_to_ohlcv(self, timeframe: str = '1T') -> pd.DataFrame:
"""
tickデータをOHLCV形式にリサンプリング
Args:
timeframe: リサンプル周期（'1T'=1分, '5T'=5分, '1H'=1時間）
Returns:
OHLCV格式DataFrame
"""
df = self.df.copy()
ohlcv = pd.DataFrame()
ohlcv['open'] = df['price'].resample(timeframe).first()
ohlcv['high'] = df['price'].resample(timeframe).max()
ohlcv['low'] = df['price'].resample(timeframe).min()
ohlcv['close'] = df['price'].resample(timeframe).last()
ohlcv['volume'] = df['volume'].resample(timeframe).sum()
ohlcv['trade_count'] = df['volume'].resample(timeframe).count()
# 欠損値処理（取引 없는时间段）
ohlcv = ohlcv.dropna()
print(f"[RESAMPLE] {timeframe}周期のOHLCVデータにリサンプリング: {len(ohlcv)}件")
return ohlcv
def calculate_spread_and_slippage(self, fee_rate: float = 0.0004) -> pd.DataFrame:
"""
スプレッドとスリッページを計算
Args:
fee_rate: 取引手数料率（OKX先物: 0.04% = 0.0004）
"""
df = self.df.copy()
# buyとsellの分离
buys = df[df['side'] == 'buy'].copy()
sells = df[df['side'] == 'sell'].copy()
# sell価格（maker注文）与える
sells = sells.rename(columns={'price': 'maker_price'})
# 聚合到同一时间戳
buys_agg = buys.groupby(buys.index).agg({
'price': 'first',
'volume': 'sum',
'trade_id': 'count'
}).rename(columns={'trade_id': 'buy_count'})
sells_agg = sells.groupby(sells.index).agg({
'maker_price': 'first',
'volume': 'sum',
'trade_id': 'count'
}).rename(columns={'trade_id': 'sell_count'})
result = buys_agg.join(sells_agg, how='outer')
# スプレッド计算
result['spread'] = result['maker_price'] - result['price']
result['spread_pct'] = result['spread'] / result['price']
# 取引コスト计算
result['transaction_cost'] = result['price'] * fee_rate * 2
print(f"[ANALYSIS] 平均スプレッド: {result['spread_pct'].mean()*100:.4f}%")
print(f"[ANALYSIS] 平均取引コスト: {result['transaction_cost'].mean():.4f}")
return result
def export_for_backtesting(self, output_path: str, format: str = 'csv'):
"""
バックテスト用データを出力
Args:
output_path: 输出ファイルパス
format: 出力形式（'csv', 'parquet', 'feather'）
"""
df = self.df.copy()
if format == 'csv':
df.to_csv(output_path)
elif format == 'parquet':
df.to_parquet(output_path)
elif format == 'feather':
df.to_feather(output_path)
print(f"[EXPORT] {output_path} に出力完了: {len(df)}件")
def get_statistics(self) -> dict:
"""データ統計情報を取得"""
stats = {
'total_records': len(self.df),
'date_range': f"{self.df.index.min()} ~ {self.df.index.max()}",
'unique_trades': self.df['trade_id'].nunique() if 'trade_id' in self.df.columns else None,
'price_range': f"{self.df['price'].min():.2f} ~ {self.df['price'].max():.2f}",
'avg_volume': self.df['volume'].mean(),
'total_volume': self.df['volume'].sum(),
}
return stats
使用例
cleaner = TickDataCleaner(df=df_btc)
cleaner.basic_cleaning()
stats = cleaner.get_statistics()
print("\n=== データ統計 ===")
for key, value in stats.items():
print(f"{key}: {value}")
1分足OHLCVに変換
ohlcv_1m = cleaner.resample_to_ohlcv(timeframe='1T')
print(ohlcv_1m.head())
バックテスト用に出力
cleaner.export_for_backtesting('btcusdt_1m_ohlcv.csv', format='csv')HolySheep AIとの組み合わせ：AI驅動分析への展開
清洗済みtickデータ或いはOHLCVデータがあれば、HolySheep AIのAPIを組み合わせることで、更なる分析扩张が可能です。例えば：
- パターン認識：Gemini 2.5 Flash（$2.50/MTok）で価格パターンの自动分類
- 感情分析：DeepSeek V3.2（$0.42/MTok）で市場感情指标的抽出
- 予測モデル：Claude Sonnet 4.5（$15/MTok）でトレンド予測
HolySheep AIはhttps://api.holysheep.ai/v1をベースURLとし、$1=¥1の汇率（公式¥7.3=$1の85%節約）でAPI利用可能です。WeChat Pay/Alipay対応で<50msの低レイテンシ、登録特典の無料クレジットもあります。
# HolySheep AIで市場分析プロンプトを送信する例
import requests
def analyze_market_with_holysheep(ohlcv_summary: str) -> str:
"""
HolySheep AI APIを使用して市場分析を実行
Args:
ohlcv_summary: 分析対象のOHLCVサマリー
Returns:
AI生成の分析结果
"""
api_key = "YOUR_HOLYSHEEP_API_KEY" # HolySheep AIのAPIキー
base_url = "https://api.holysheep.ai/v1"
prompt = f"""
以下のOHLCVデータに基づく短期トレンド分析を行ってください：
{ohlcv_summary}
分析項目：
1. 当前位置判断（上昇/下落/保ち合い）
2. ボラティリティ評価（高い/普通/低い）
3. 取引機会の示唆（買い/売り/様子見）
"""
payload = {
"model": "gemini-2.5-flash", # $2.50/MTok - コスト эффектив的な分析
"messages": [
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 500
}
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
response = requests.post(
f"{base_url}/chat/completions",
headers=headers,
json=payload
)
if response.status_code == 200:
result = response.json()
return result['choices'][0]['message']['content']
else:
raise Exception(f"API Error: {response.status_code} - {response.text}")
使用例
summary = ohlcv_1m.tail(20).to_string()
analysis = analyze_market_with_holysheep(summary)
print("AI分析結果:", analysis)よくあるエラーと対処法
エラー1：APIリクエスト回数上限超过
错误信息：requests.exceptions.HTTPError: 429 Client Error: Too Many Requests
原因：OKX APIは1分間に120リクエストのRate Limitがあります。递归取得時にリクエスト间隔が短すぎると発生します。
解決コード：
import time
from functools import wraps
def rate_limit_handler(max_retries=5, base_delay=1.0):
"""Rate Limit対応デコレータ"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
retries = 0
delay = base_delay
while retries < max_retries:
try:
return func(*args, **kwargs)
except requests.exceptions.HTTPError as e:
if e.response.status_code == 429:
print(f"[WARN] Rate Limit. {delay}秒後に再試行...")
time.sleep(delay)
delay *= 2 # 指数バックオフ
retries += 1
else:
raise
raise Exception(f"Max retries ({max_retries}) exceeded")
return wrapper
return decorator
使用例
@rate_limit_handler(max_retries=5, base_delay=1.0)
def get_trades_with_retry(inst_id: str, after: str = None):
params = {'instId': inst_id, 'limit': 100}
if after:
params['after'] = after
response = requests.get(
"https://www.okx.com/api/v5/market/trades",
params=params
)
response.raise_for_status()
return response.json()エラー2：タイムスタンプ形式不正
错误信息：ValueError: Invalid unit-supplied for timestamp conversion
原因：OKX APIのタイムスタンプはミリ秒単位の文字列で返されますが、秒単位として处理试图みると発生します。
解決コード：
def safe_timestamp_conversion(timestamp_series):
"""
安全的timestamp转换（ミリ秒→datetime）
Args:
timestamp_series: ミリ秒タイムスタンプのシリーズ
Returns:
DatetimeIndex
"""
# 文字列の場合、整数に変換
if timestamp_series.dtype == 'object':
ts_numeric = pd.to_numeric(timestamp_series, errors='coerce')
else:
ts_numeric = timestamp_series
# ミリ秒单位确认（13桁ならミリ秒、10桁なら秒）
sample_val = ts_numeric.dropna().iloc[0] if len(ts_numeric.dropna()) > 0 else 0
if sample_val > 1e12: # 13桁以上 = ミリ秒
unit = 'ms'
else: # 10桁 = 秒
unit = 's'
print(f"[INFO] タイムスタンプ形式: {unit}単位")
return pd.to_datetime(ts_numeric, unit=unit, utc=True).dt.tz_convert('Asia/Shanghai')
使用例
df['datetime'] = safe_timestamp_conversion(df['timestamp'])エラー3：データ间隙导致的分析误差
错误信息：バックテスト结果が実際の取引结果と大きく異なる（Gap问题）
原因：取引していない時間帯があると、ohlcvリサンプル時に欠損值ができたり、价格が急激に変化してスリッページが大きくなります。
解決コード：
def detect_and_handle_gaps(df: pd.DataFrame, max_gap_minutes: int = 60) -> pd.DataFrame:
"""
データ间隙を検出し、フラグ付け
Args:
df: datetimeインデックスを持つDataFrame
max_gap_minutes: 间隙とみなす最小分钟数
Returns:
间隙フラグ付きのDataFrame
"""
df = df.copy()
# 前回からの时间差を计算
time_diff = df.index.to_series().diff()
time_diff_minutes = time_diff.dt.total_seconds() / 60
# 间隙を检测
gap_mask = time_diff_minutes > max_gap_minutes
df['has_gap'] = gap_mask
df['gap_duration_minutes'] = time_diff_minutes.where(gap_mask, 0)
gap_count = gap_mask.sum()
if gap_count > 0:
print(f"[WARNING] {gap_count}件のデータ间隙を検出")
print(f"[WARNING] 最大间隙: {time_diff_minutes.max():.0f}分")
# 间隙情報を表示
gaps = df[df['has_gap']].index
for gap_start in gaps:
gap_end_idx = df.index.get_loc(gap_start) + 1
if gap_end_idx < len(df):
gap_end = df.index[gap_end_idx]
print(f" - {gap_start} ~ {gap_end}")
return df
使用例
df_with_gaps = detect_and_handle_gaps(ohlcv_1m, max_gap_minutes=60)
间隙が存在する足をフラグ付け（バックテストで处理分开）
df_clean = df_with_gaps[df_with_gaps['gap_duration_minutes'] == 0]
print(f"[INFO] 间隙除外後のデータ: {len(df_clean)}件")まとめ：バックテスト用水準へのデータ整備フロー
- データ下载：OKX REST APIからtickデータを批量取得
- 基本清洗：重複除去・異常値处理・欠損値填补
- 品質确认：间隙検出・統計情報確認
- リ샘플링：tick→OHLCV（1分足・5分足・1時間足等）
- AI分析：HolySheep AIで дополнительный分析
このフローを確立すれば任何の取引戦略のバックテストが現実的なデータで検証可能です。tickレベルの细致な分析は、短期トレーダーにとって特に重要な基础技术となります。
HolySheep AIの<50ms低レイテンシ環境なら、リアルタイム分析にも耐えられます。注册特典の無料クレジットで気軽に试すことができます。