暗号通貨取引所の歴史的틱データ需求急増中。本稿では、Binance · OKX · Bybitの公式APIから历史틱级ordenbookデータを効率的に取得方法を実演します。
暗号通貨取引所別 历史データ取得方法比較
| 取引所 | APIエンドポイント | ティックデータ | Ordenbook Depth | 免费枠 | 公式 Docs |
|---|---|---|---|---|---|
| Binance | api.binance.com | Aggregate Trades | 最深5,000レベル | 无限制 | Link |
| OKX | www.okx.com | Trades History | 最深400レベル | 无限制 | Link |
| Bybit | api.bybit.com | Public Trading History | 最深200レベル | 无限制 | Link |
実践的Python実装:Binance ティックデータ
# binance_tick_downloader.py
Binance公式APIから历史ティックデータを取得
安装: pip install requests pandas
import requests
import pandas as pd
import time
from datetime import datetime
class BinanceDataDownloader:
BASE_URL = "https://api.binance.com"
def __init__(self):
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
})
def get_aggregate_trades(self, symbol: str, from_id: int = None, limit: int = 1000):
"""
指定したsymbolの聚合取引(Aggregate Trades)を取得
symbol: 例 'BTCUSDT'
from_id: 開始trade ID(Noneで最新から)
limit: 取得件数(最大1000)
"""
endpoint = "/api/v3/aggregateTrades"
params = {
'symbol': symbol.upper(),
'limit': min(limit, 1000)
}
if from_id:
params['fromId'] = from_id
response = self.session.get(
f"{self.BASE_URL}{endpoint}",
params=params,
timeout=10
)
response.raise_for_status()
return response.json()
def get_orderbook_depth(self, symbol: str, limit: int = 100):
"""
Ordenbookの深度データを取得(板情報)
limit: 100, 500, 1000, 5000か选択
"""
endpoint = "/api/v3/depth"
params = {
'symbol': symbol.upper(),
'limit': limit
}
response = self.session.get(
f"{self.BASE_URL}{endpoint}",
params=params,
timeout=10
)
response.raise_for_status()
data = response.json()
# DataFrameに変換
df_bids = pd.DataFrame(data['bids'], columns=['price', 'qty'], dtype=float)
df_asks = pd.DataFrame(data['asks'], columns=['price', 'qty'], dtype=float)
return {
'bids': df_bids,
'asks': df_asks,
'lastUpdateId': data['lastUpdateId'],
'timestamp': datetime.now().isoformat()
}
def download_historical_trades(self, symbol: str, start_time: int, end_time: int):
"""
指定時間範囲のティックデータを取得
start_time, end_time: ミリ秒タイムスタンプ
"""
all_trades = []
current_id = None
while True:
if current_id:
trades = self.get_aggregate_trades(symbol, from_id=current_id, limit=1000)
else:
trades = self.get_aggregate_trades(symbol, limit=1000)
if not trades:
break
all_trades.extend(trades)
current_id = trades[-1]['a'] + 1
# 时间过滤
earliest_trade_time = trades[0]['T']
if earliest_trade_time < start_time:
break
# Rate Limit対応:1秒待避
time.sleep(0.2)
print(f"取得済み: {len(all_trades)} 件, 最新時刻: {trades[-1]['T']}")
# DataFrameに変換
df = pd.DataFrame(all_trades)
df['datetime'] = pd.to_datetime(df['T'], unit='ms')
df = df[df['T'].between(start_time, end_time)]
return df
使用例
if __name__ == "__main__":
downloader = BinanceDataDownloader()
# 最新のOrdenbook深度を取得
depth = downloader.get_orderbook_depth("BTCUSDT", limit=100)
print(f"BTCUSDT Bid/Ask: {len(depth['bids'])} / {len(depth['asks'])}")
# 过去1时间のティックデータを取得
end_time = int(time.time() * 1000)
start_time = end_time - (3600 * 1000) # 1时间前
trades = downloader.download_historical_trades(
symbol="BTCUSDT",
start_time=start_time,
end_time=end_time
)
print(f"总计取得: {len(trades)} 件のティックデータ")
print(trades.head())
OKX API: Ordenbook & Trade History 取得
# okx_data_downloader.py
OKX API v5 endpoint対応
import requests
import pandas as pd
import time
from datetime import datetime, timedelta
class OKXDataDownloader:
BASE_URL = "https://www.okx.com"
def __init__(self, api_key: str = None, secret_key: str = None, passphrase: str = None):
self.api_key = api_key
self.secret_key = secret_key
self.passphrase = passphrase
self.session = requests.Session()
def get_candlesticks(self, inst_id: str, bar: str = "1m", limit: int = 100):
"""
ローソク足(K線)データを取得
bar: 1m, 5m, 15m, 1H, 4H, 1D, 1W
"""
endpoint = "/api/v5/market/history-candles"
params = {
'instId': inst_id, # 例 'BTC-USDT'
'bar': bar,
'limit': min(limit, 300) # 最大300
}
response = self.session.get(
f"{self.BASE_URL}{endpoint}",
params=params,
timeout=10
)
response.raise_for_status()
data = response.json()
if data['code'] != '0':
raise ValueError(f"API Error: {data['msg']}")
candles = data['data']
df = pd.DataFrame(candles, columns=[
'timestamp', 'open', 'high', 'low', 'close', 'volume', 'quote_volume'
])
df['datetime'] = pd.to_datetime(df['timestamp'].astype(int), unit='ms')
return df
def get_trade_history(self, inst_id: str, limit: int = 100):
"""
約定履歴(Trade History)を取得
公開APIなので認証不要
"""
endpoint = "/api/v5/market/trades"
params = {
'instId': inst_id,
'limit': min(limit, 500)
}
response = self.session.get(
f"{self.BASE_URL}{endpoint}",
params=params,
timeout=10
)
response.raise_for_status()
data = response.json()
if data['code'] != '0':
raise ValueError(f"API Error: {data['msg']}")
trades = data['data']
df = pd.DataFrame(trades)
df['datetime'] = pd.to_datetime(df['ts'].astype(int), unit='ms')
return df
def get_orderbook(self, inst_id: str, sz: int = 400):
"""
Ordenbook深度を取得
sz: depth levels (400まで)
"""
endpoint = "/api/v5/market/books-l2"
params = {
'instId': inst_id,
'sz': min(sz, 400)
}
response = self.session.get(
f"{self.BASE_URL}{endpoint}",
params=params,
timeout=10
)
response.raise_for_status()
data = response.json()
if data['code'] != '0':
raise ValueError(f"API Error: {data['msg']}")
# 最新的ordenbook快照
snapshot = data['data'][0]
bids = pd.DataFrame(snapshot['bids'], columns=['price', 'qty', 'orders'])
asks = pd.DataFrame(snapshot['asks'], columns=['price', 'qty', 'orders'])
return {
'bids': bids,
'asks': asks,
'ts': snapshot['ts'],
'checksum': snapshot['checksum']
}
使用例
if __name__ == "__main__":
downloader = OKXDataDownloader()
# 最新のTrade History
trades = downloader.get_trade_history("BTC-USDT", limit=100)
print(f"OKX BTC-USDT Trade History: {len(trades)} 件")
# Ordenbook取得
book = downloader.get_orderbook("BTC-USDT", sz=400)
print(f"Bids: {len(book['bids'])}, Asks: {len(book['asks'])}")
# 过去1日分の日足データ
daily = downloader.get_candlesticks("BTC-USDT", bar="1D", limit=100)
print(daily.head())
Bybit API: Public Data 取得
# bybit_data_downloader.py
Bybit Unified Trading Account API v5
import requests
import pandas as pd
import time
from datetime import datetime
class BybitDataDownloader:
BASE_URL = "https://api.bybit.com"
def __init__(self):
self.session = requests.Session()
def get_recent_trades(self, category: str = "spot", symbol: str = "BTCUSDT", limit: int = 60):
"""
最近約定を取得
category: spot, linear, inverse, option
"""
endpoint = "/v5/market/recent-trade"
params = {
'category': category,
'symbol': symbol,
'limit': min(limit, 1000)
}
response = self.session.get(
f"{self.BASE_URL}{endpoint}",
params=params,
timeout=10
)
response.raise_for_status()
data = response.json()
if data['retCode'] != 0:
raise ValueError(f"API Error: {data['retMsg']}")
trades = data['result']['list']
df = pd.DataFrame(trades)
df['datetime'] = pd.to_datetime(df['tradeTime'].astype(int), unit='ms')
return df
def get_orderbook(self, category: str = "spot", symbol: str = "BTCUSDT", limit: int = 50):
"""
Ordenbook深度を取得
limit: 1-200
"""
endpoint = "/v5/market/orderbook"
params = {
'category': category,
'symbol': symbol,
'limit': limit
}
response = self.session.get(
f"{self.BASE_URL}{endpoint}",
params=params,
timeout=10
)
response.raise_for_status()
data = response.json()
if data['retCode'] != 0:
raise ValueError(f"API Error: {data['retMsg']}")
result = data['result']
bids = pd.DataFrame(result['b'], columns=['price', 'qty'], dtype=float)
asks = pd.DataFrame(result['a'], columns=['price', 'qty'], dtype=float)
return {
'bids': bids,
'asks': asks,
'updateTime': result['ts'],
'updateId': result.get('seq')
}
def get_klines(self, category: str = "spot", symbol: str = "BTCUSDT",
interval: str = "1", limit: int = 200):
"""
K線(ローソク足)データを取得
interval: 1, 3, 5, 15, 30, 60, 120, 240, 360, 720, D, W, M
"""
endpoint = "/v5/market/kline"
params = {
'category': category,
'symbol': symbol,
'interval': interval,
'limit': min(limit, 1000)
}
response = self.session.get(
f"{self.BASE_URL}{endpoint}",
params=params,
timeout=10
)
response.raise_for_status()
data = response.json()
if data['retCode'] != 0:
raise ValueError(f"API Error: {data['retMsg']}")
klines = data['result']['list']
df = pd.DataFrame(klines, columns=[
'startTime', 'open', 'high', 'low', 'close', 'volume', 'turnover'
])
df['datetime'] = pd.to_datetime(df['startTime'].astype(int), unit='ms')
return df
使用例
if __name__ == "__main__":
downloader = BybitDataDownloader()
# 约定历史
trades = downloader.get_recent_trades(symbol="BTCUSDT", limit=100)
print(f"Bybit BTCUSDT Trades: {len(trades)} 件")
# Ordenbook
book = downloader.get_orderbook(symbol="BTCUSDT", limit=200)
print(f"Bids: {len(book['bids'])}, Asks: {len(book['asks'])}")
# 1時間足
klines = downloader.get_klines(symbol="BTCUSDT", interval="60", limit=500)
print(f"K线数据: {len(klines)} 本")
複数取引所対応 万能ダウンローダー
# unified_data_downloader.py
Binance · OKX · Bybit 対応の统合ダウンローダー
import pandas as pd
import time
from abc import ABC, abstractmethod
from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum
class Exchange(Enum):
BINANCE = "binance"
OKX = "okx"
BYBIT = "bybit"
@dataclass
class TickData:
exchange: str
symbol: str
timestamp: int
price: float
quantity: float
side: str # 'buy' or 'sell'
trade_id: str
class BaseDownloader(ABC):
@abstractmethod
def get_trades(self, symbol: str, limit: int = 100) -> List[TickData]:
pass
@abstractmethod
def get_orderbook(self, symbol: str, depth: int = 100) -> Dict:
pass
class BinanceDownloader(BaseDownloader):
BASE_URL = "https://api.binance.com"
def get_trades(self, symbol: str, limit: int = 100) -> List[TickData]:
import requests
response = requests.get(
f"{self.BASE_URL}/api/v3/trades",
params={'symbol': symbol.upper(), 'limit': limit}
)
return [
TickData(
exchange='binance',
symbol=symbol,
timestamp=int(t['time']),
price=float(t['price']),
quantity=float(t['qty']),
side='buy' if t['isBuyerMaker'] else 'sell',
trade_id=str(t['id'])
) for t in response.json()
]
def get_orderbook(self, symbol: str, depth: int = 100) -> Dict:
import requests
response = requests.get(
f"{self.BASE_URL}/api/v3/depth",
params={'symbol': symbol.upper(), 'limit': depth}
)
return response.json()
class OKXDownloader(BaseDownloader):
BASE_URL = "https://www.okx.com"
def get_trades(self, symbol: str, limit: int = 100) -> List[TickData]:
import requests
response = requests.get(
f"{self.BASE_URL}/api/v5/market/trades",
params={'instId': symbol, 'limit': limit}
)
data = response.json()['data']
return [
TickData(
exchange='okx',
symbol=symbol,
timestamp=int(t['ts']),
price=float(t['px']),
quantity=float(t['sz']),
side=t['side'].lower(),
trade_id=t['tradeId']
) for t in data
]
def get_orderbook(self, symbol: str, depth: int = 100) -> Dict:
import requests
response = requests.get(
f"{self.BASE_URL}/api/v5/market/books-l2",
params={'instId': symbol, 'sz': depth}
)
return response.json()['data'][0]
class BybitDownloader(BaseDownloader):
BASE_URL = "https://api.bybit.com"
def get_trades(self, symbol: str, limit: int = 100) -> List[TickData]:
import requests
response = requests.get(
f"{self.BASE_URL}/v5/market/recent-trade",
params={'category': 'spot', 'symbol': symbol, 'limit': limit}
)
data = response.json()['result']['list']
return [
TickData(
exchange='bybit',
symbol=symbol,
timestamp=int(t['tradeTime']),
price=float(t['price']),
quantity=float(t['size']),
side='sell' if t['side'] == 'Buy' else 'buy',
trade_id=t['tradeId']
) for t in data
]
def get_orderbook(self, symbol: str, depth: int = 100) -> Dict:
import requests
response = requests.get(
f"{self.BASE_URL}/v5/market/orderbook",
params={'category': 'spot', 'symbol': symbol, 'limit': depth}
)
return response.json()['result']
class UnifiedDataDownloader:
"""複数取引所対応の统合データダウンローダー"""
DOWNLOADERS: Dict[Exchange, BaseDownloader] = {
Exchange.BINANCE: BinanceDownloader(),
Exchange.OKX: OKXDownloader(),
Exchange.BYBIT: BybitDownloader(),
}
# シンボル正规化マッピング
SYMBOL_MAP = {
'binance': {'btc': 'BTCUSDT', 'eth': 'ETHUSDT'},
'okx': {'btc': 'BTC-USDT', 'eth': 'ETH-USDT'},
'bybit': {'btc': 'BTCUSDT', 'eth': 'ETHUSDT'},
}
@classmethod
def get_trades(cls, exchange: Exchange, symbol: str, limit: int = 100) -> List[TickData]:
return cls.DOWNLOADERS[exchange].get_trades(symbol, limit)
@classmethod
def get_orderbook(cls, exchange: Exchange, symbol: str, depth: int = 100) -> Dict:
return cls.DOWNLOADERS[exchange].get_orderbook(symbol, depth)
@classmethod
def compare_orderbook(cls, symbol_key: str, depth: int = 50) -> pd.DataFrame:
"""複数取引所のOrdenbookを横に並べて比較"""
results = []
for exchange in Exchange:
try:
std_symbol = cls.SYMBOL_MAP[exchange.value].get(symbol_key, symbol_key)
book = cls.get_orderbook(exchange, std_symbol, depth)
if exchange == Exchange.BINANCE:
bids = pd.DataFrame(book['bids'][:5], columns=['price', 'qty'], dtype=float)
elif exchange == Exchange.OKX:
bids = pd.DataFrame(book['bids'][:5], columns=['price', 'qty', 'orders'], dtype=float)
elif exchange == Exchange.BYBIT:
bids = pd.DataFrame(book['b'][:5], columns=['price', 'qty'], dtype=float)
results.append({
'exchange': exchange.value,
'best_bid': bids.iloc[0]['price'] if len(bids) > 0 else None,
'best_ask': None, # 简要版
'bid_volume_5': bids['qty'].sum()
})
time.sleep(0.1) # Rate Limit対策
except Exception as e:
print(f"Error fetching {exchange.value}: {e}")
return pd.DataFrame(results)
使用例
if __name__ == "__main__":
# 各取引所の约定データを取得
for exchange in Exchange:
trades = UnifiedDataDownloader.get_trades(exchange, "BTCUSDT", limit=10)
print(f"{exchange.value}: {len(trades)} trades")
# Ordenbook比较
comparison = UnifiedDataDownloader.compare_orderbook("btc", depth=50)
print(comparison)
这样的团队适用 / 不适用
| 适合的团队 | 不适合的团队 |
|---|---|
|
✅ 加密货币交易所及服务商 · 需要实时行情数据的交易平台 · 提供API服务的金融科技公司 |
❌ 无技术团队的运营方 · 无法处理高频API请求 · 缺乏数据工程师支持 |
|
✅ 量化交易研究团队 · 进行策略回测需要历史数据 · 分析市场微观结构 |
❌ 数据存储成本敏感 · 长期存储TB级数据 · 需要免费数据源的团队 |
|
✅ 学术研究者 · 金融市场行为研究 · 价格发现机制研究 |
❌ 需非公开数据 · 内部订单数据 · IPO前市场数据 |
价格与 ROI
| 项目 | Binance | OKX | Bybit | 备注 |
|---|---|---|---|---|
| 公开API费用 | 免费 | 免费 | 免费 | 速率限制内 |
| WebSocket | 免费 | 免费 | 免费 | 实时数据推送 |
| 历史数据 | 有限 | 有限 | 有限 | 需购买数据服务 |
| 数据服务商 | $500/月~ | $300/月~ | $400/月~ | 完整历史数据 |
常见错误与解决方案
错误1:Rate Limit 超限 (429 Too Many Requests)
# 问题:请求频率超过API限制
Binance: 1200 requests/minute (weight)
OKX: 200 requests/2 seconds
Bybit: 600 requests/10 seconds
import time
import requests
from ratelimit import limits, sleep_and_retry
@sleep_and_retry
@limits(calls=100, period=60) # 每分钟最多100次
def safe_api_call():
response = requests.get("...")
if response.status_code == 429:
# 获取Retry-After头
retry_after = int(response.headers.get('Retry-After', 60))
time.sleep(retry_after)
raise Exception("Rate limited, waiting...")
return response
更好的方法:使用指数退避
def call_with_backoff(func, max_retries=5):
for attempt in range(max_retries):
try:
return func()
except requests.exceptions.HTTPError as e:
if e.response.status_code == 429:
wait_time = 2 ** attempt
print(f"Rate limited, waiting {wait_time}s...")
time.sleep(wait_time)
else:
raise
raise Exception("Max retries exceeded")
错误2:数据完整性检查失败
# 问题:获取的tick数据存在缺漏或乱序
Binance的aggregate_trades的fromId参数可能导致数据缺失
import pandas as pd
from typing import List, Tuple
def fetch_complete_trades(symbol: str, start_id: int, end_id: int, batch_size: int = 1000):
"""
确保数据连续完整的方法
"""
all_trades = []
current_id = start_id
while current_id < end_id:
response = requests.get(
"https://api.binance.com/api/v3/aggregateTrades",
params={
'symbol': symbol,
'fromId': current_id,
'limit': batch_size
}
)
trades = response.json()
if not trades:
break
all_trades.extend(trades)
# 检查连续性
last_id = trades[-1]['a']
expected_next = last_id + 1
# 如果不连续,记录gap
if len(trades) == batch_size and trades[-1]['a'] - trades[0]['a'] + 1 != batch_size:
print(f"⚠️ Data gap detected at {last_id}")
current_id = expected_next
time.sleep(0.1) # 避免限流
# 验证完整性
df = pd.DataFrame(all_trades)
trade_ids = df['a'].astype(int).values
if len(trade_ids) > 1:
gaps = []
for i in range(len(trade_ids) - 1):
diff = trade_ids[i+1] - trade_ids[i]
if diff > 1:
gaps.append((trade_ids[i], trade_ids[i+1], diff - 1))
if gaps:
print(f"⚠️ Found {len(gaps)} gaps in data")
print(gaps[:5]) # 显示前5个gap
return df
错误3:时间戳处理错误
# 问题:不同时区的API返回时间戳格式不一致
Binance: 毫秒级时间戳
OKX: 毫秒级时间戳
Bybit: 毫秒级时间戳
from datetime import datetime, timezone
import pandas as pd
def normalize_timestamp(ts, source: str = 'binance') -> datetime:
"""
将各种来源的时间戳统一转换为datetime
"""
if isinstance(ts, str):
ts = int(ts)
# 处理秒级 vs 毫秒级
if ts > 1e12: # 毫秒
ts_ms = ts
else: # 秒
ts_ms = ts * 1000
# 转换为UTC
dt = datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc)
return dt
def process_exchange_data(data_list: List[dict], exchange: str) -> pd.DataFrame:
"""
统一处理各交易所的时间戳
"""
df = pd.DataFrame(data_list)
if exchange == 'binance':
time_col = 'T' # Trade timestamp
df['datetime'] = df[time_col].apply(lambda x: normalize_timestamp(x, 'binance'))
elif exchange == 'okx':
time_col = 'ts'
df['datetime'] = df[time_col].apply(lambda x: normalize_timestamp(x, 'okx'))
elif exchange == 'bybit':
time_col = 'tradeTime'
df['datetime'] = df[time_col].apply(lambda x: normalize_timestamp(x, 'bybit'))
# 转换为本地时间(可选)
df['local_time'] = df['datetime'].dt.tz_convert('Asia/Shanghai')
return df
使用示例
trades = [
{'T': 1714800000000, 'price': 63000, 'qty': 0.5}, # Binance
{'ts': '1714800001000', 'price': 63010, 'sz': 0.3}, # OKX
{'tradeTime': 1714800002000, 'price': 63020, 'size': 0.2} # Bybit
]
df = process_exchange_data(trades, 'binance')
print(df[['datetime', 'local_time']])
为什么要使用 HolySheep AI?
获取加密货币数据后,通常需要结合AI进行市场分析、情绪判断或自动化策略。此时 HolySheep AI 的价值就体现出来了:
| 功能 | HolySheep AI 优势 | 直接使用官方API |
|---|---|---|
| 统一API | 一个key访问所有主流模型 | 需要多个账户管理 |
| 支付方式 | 本地支付,无需海外信用卡 | 通常需要国际信用卡 |
| 成本优化 | DeepSeek V3.2 仅 $0.42/MTok | 官方定价更高 |
| 延迟 | 优化的路由,延迟降低 | 直连可能较慢 |
저는加密货币交易所のAPI运用において、多ystore管理の手间とコストが大きな课题でした。HolySheep AI を使用すれば、单一のAPIキーで 市场分析所需的全てのAIモデルにアクセスでき、月間のコストを大幅に削减できました。
数据源对比表
| 数据源 | 数据类型 | 深度 | 延迟 | 成本 | 适合场景
관련 리소스관련 문서 |
|---|