暗号通貨取引所の歴史的틱データ需求急増中。本稿では、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モデルにアクセスでき、月間のコストを大幅に削减できました。

数据源对比表

数据源 数据类型 深度 延迟 成本 适合场景

관련 리소스

관련 문서

🔥 HolySheep AI를 사용해 보세요

직접 AI API 게이트웨이. Claude, GPT-5, Gemini, DeepSeek 지원. VPN 불필요.

👉 무료 가입 →