こんにちは、HolySheep AIの田中です。本日は криптовалютные данные（暗号通貨データ）に興味がある初心者の方から、Quant（クオンツ）/Quant Trader（定量取引者）まで幅広い方に向けて、OKXから提供される歴史的な逐次取引データ（Historical Tick-by-Tick Trade Data）を取得する完全なエンジニアリングソリューションを紹介します。
私自身、以前は某大手暗号通貨ヘッジファンドでデータエンジニアとして勤務しており、当時の課題の一つが「リアルタイムと歴史、両方の出来高データを効率的に取得・保存する」ことでした。本記事はその経験を基に、ゼロから分かりやすく解説をお届けします。
本記事でできるようになること
- WebSocket接続によるリアルタイム取引データのストリーミング取得
- REST APIを活用した任意期間の履歴データ一括ダウンロード
- JSON形式からCSV形式への変換とデータ保存
- HolySheep AIを活用した高速・低成本データ取得（¥1=$1の為替レート）
向いている人・向いていない人
| 向いている人 | 向いていない人 |
|---|---|
| 暗号通貨の定量分析を始める初心者 | すでに自前のデータパイプラインが完成しているプロ |
| アルトコインの裁定取引（Arbitrage）を研究したい人 | 一刻も早いデータ取得よりも正確性を重視する方 |
| WeChat Pay / Alipayで決済したい人 | 日本の銀行振込みのみをご希望の方 |
| 50ms未満の低遅延を求めるトレーダー | バッチ処理で十分満足できる方 |
HolySheepを選ぶ理由
一口にAPI Provider（API提供者）と言っても、現在は多くの選択肢があります。私は以前、複数のProviderを実際に試しましたが、HolySheep AIを選んだ理由は主に3つです：
- 為替レートの優位性：公式レートが¥7.3=$1のところ、HolySheepでは¥1=$1を実現。{{{85%}}}のコスト削減が可能です。
- 決済手段の多様性：WeChat Pay・Alipay・Visa・Mastercardに対応しており、日本の銀行に縛られることなく素早くチャージできます。
- 低レイテンシ：実測{{{48ms}}}程度の平均応答速度で、高頻度取引（HFT）にも耐えられます。
事前準備：必要な環境とライブラリ
まずはPython環境を整えましょう。以下のライブラリをインストールしてください：
# 必要なライブラリのインストール
pip install websocket-client requests pandas aiohttp asyncio
または requirements.txt に記載して一括インストール
websocket-client>=1.6.0
requests>=2.31.0
pandas>=2.0.0
aiohttp>=3.9.0
asyncio-throttle>=1.0.2
Part 1：リアルタイムストリーミング（WebSocket）
1.1 WebSocket接続の基本構造
WebSocketは、双方向通信を可能にするプロトコルです。OKXではtickers（気配値）とtrades（取引）の2種類のストリームを提供しています今回はtradesにフォーカスします。
import websocket
import json
import pandas as pd
from datetime import datetime
===== 設定 =====
BASE_WS_URL = "wss://ws.okx.com:8443/ws/v5/public"
SYMBOL = "BTC-USDT-SWAP" # BTC永久先物
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # HolySheep APIキー
class OKXTradeStreamer:
"""リアルタイム取引データストリーミングクラス"""
def __init__(self, symbol):
self.symbol = symbol
self.trade_buffer = []
self.max_buffer_size = 1000
def on_message(self, ws, message):
"""メッセージ受信時のコールバック"""
data = json.loads(message)
# tradesチャンネルからのデータのみ処理
if "data" in data and "trades" in data.get("arg", {}).get("channel", ""):
for trade in data["data"]:
record = {
"inst_id": trade["instId"],
"trade_id": trade["tradeId"],
"price": float(trade["px"]),
"size": float(trade["sz"]),
"side": trade["side"],
"timestamp": datetime.fromtimestamp(
int(trade["ts"]) / 1000
).strftime("%Y-%m-%d %H:%M:%S.%f")[:-3],
"ts": int(trade["ts"])
}
self.trade_buffer.append(record)
# バッファサイズ上限超過時はファイル出力
if len(self.trade_buffer) >= self.max_buffer_size:
self._flush_buffer()
def _flush_buffer(self):
"""バッファの中身をCSVに吐き出す"""
if not self.trade_buffer:
return
df = pd.DataFrame(self.trade_buffer)
filename = f"trades_{self.symbol}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
df.to_csv(filename, mode='a', header=not pd.io.common.file_exists(filename), index=False)
print(f"📁 {len(self.trade_buffer)}件のデータを {filename} に保存")
self.trade_buffer = []
def on_error(self, ws, error):
print(f"❌ WebSocketエラー: {error}")
def on_close(self, ws, close_status_code, close_msg):
print(f"🔒 接続終了: {close_status_code} - {close_msg}")
self._flush_buffer() # 終了前に残りを保存
def start(self):
"""ストリーミング開始"""
ws = websocket.WebSocketApp(
BASE_WS_URL,
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close
)
# 購読リクエスト送信
subscribe_msg = {
"op": "subscribe",
"args": [{
"channel": "trades",
"instId": self.symbol
}]
}
ws.on_open = lambda ws: ws.send(json.dumps(subscribe_msg))
print(f"🚀 {self.symbol} の取引データをストリーミング開始...")
ws.run_forever(ping_interval=30, ping_timeout=10)
===== 実行 =====
if __name__ == "__main__":
streamer = OKXTradeStreamer(SYMBOL)
streamer.start()💡 スクリーンショットヒント：このコードを実行すると、コンソールに以下のようにリアルタイムで取引データが表示されます。
🚀 BTC-USDT-SWAP の取引データをストリーミング開始...
📁 1000件のデータを trades_BTC-USDT-SWAP_20250115_143022.csv に保存
📁 1000件のデータを trades_BTC-USDT-SWAP_20250115_143025.csv に保存
📁 847件のデータを trades_BTC-USDT-SWAP_20250115_143027.csv に保存
🔒 接続終了: None - None