导言：为什么Tick数据是高频交易的生命线
在加密货币高频交易（HFT）领域，历史Tick数据是策略研发和回测的基石。与传统的OHLCV分钟数据不同，Tick数据记录每一笔成交的精确价格和成交量，时间戳精度达到毫秒级（latence <1ms）。正是这种粒度让交易者能够识别订单簿动态、检测冰山订单、捕捉流动性转移模式。
本文将带你从零开始构建完整的数据获取管道，涵盖Binance、Bybit、OKX等主流交易所API的实际调用，以及数据清洗、存储和性能优化的实战技巧。作为HolySheep AI的技术团队，我们在为量化机构搭建数据基础设施时，积累了大量的踩坑经验，这些都将毫无保留地分享给你。
一、Tick数据基础概念与数据结构
1.1 什么是Tick数据？
Tick数据是金融市场上每一次交易发生时的最小信息单元。在加密货币领域，一个标准的Tick包含：
- timestamp：交易发生时间（Unix毫秒时间戳或ISO 8601格式）
- symbol：交易对标识符，如 BTCUSDT、ETHUSDT
- price：成交价格（精度通常为8位小数）
- quantity：成交数量
- side：买方主动成交（buy）或卖方主动成交（sell）
- trade_id：交易所分配的唯一成交编号
{
"event_time": 1704067200000,
"symbol": "BTCUSDT",
"price": 46250.50,
"quantity": 0.00234,
"side": "buy",
"trade_id": 1234567890123,
"is_maker": false
}1.2 Tick数据 vs K线数据：为什么粒度决定策略质量
让我们用实际数据对比说明差异。以2024年1月BTC/USDT市场为例：
# 1分钟K线数据示例
KLine_1m = {
"open": 46200.00,
"high": 46280.50,
"low": 46150.25,
"close": 46250.00,
"volume": 125.4321,
"timestamp": 1704067200000 # 1分钟内所有信息被压缩为1个数据点
}
同1分钟内的Tick数据（实际可能包含数百到数千条）
Tick_sample = [
{"time": 1704067200001, "price": 46200.50, "qty": 0.00100, "side": "buy"},
{"time": 1704067200023, "price": 46201.00, "qty": 0.05000, "side": "buy"},
{"time": 1704067200045, "price": 46200.75, "qty": 0.00200, "side": "sell"},
# ... 实际1分钟内可能有500+条Tick
{"time": 1704067200599, "price": 46250.00, "qty": 0.01000, "side": "buy"}
]关键差异：K线丢失了价格微观波动路径、订单流方向、大单分割模式等对HFT策略至关重要的信息。
二、主流交易所API数据获取实战
2.1 Binance获取历史成交记录
Binance是目前交易量最大的加密货币交易所，其历史成交API（Historical Trades）支持获取任意时间范围的Tick数据。
import requests
import time
import pandas as pd
from datetime import datetime, timedelta
class BinanceTickDataFetcher:
"""Binance历史Tick数据获取器"""
BASE_URL = "https://api.binance.com/api/v3"
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_historical_trades(self, symbol: str, limit: int = 1000,
from_id: int = None) -> list:
"""
获取历史成交记录
参数:
symbol: 交易对，如 'BTCUSDT'
limit: 单次最大获取数量 (1-1000)
from_id: 从指定trade_id开始获取（用于分页）
返回:
包含Tick数据的列表
"""
endpoint = f"{self.BASE_URL}/historicalTrades"
params = {
"symbol": symbol.upper(),
"limit": min(limit, 1000) # API限制单次最多1000条
}
if from_id:
params["fromId"] = from_id
response = self.session.get(endpoint, params=params)
response.raise_for_status()
trades = response.json()
print(f"获取 {symbol} Tick数据 {len(trades)} 条")
return trades
def get_trades_by_timerange(self, symbol: str, start_time: int,
end_time: int) -> pd.DataFrame:
"""
按时间范围批量获取Tick数据
参数:
symbol: 交易对
start_time: 开始时间（毫秒Unix时间戳）
end_time: 结束时间（毫秒Unix时间戳）
返回:
Pandas DataFrame格式的Tick数据
"""
all_trades = []
current_from_id = None
# 批量获取直到覆盖整个时间范围
while True:
trades = self.get_historical_trades(
symbol=symbol,
limit=1000,
from_id=current_from_id
)
if not trades:
break
# 过滤时间范围
filtered = [t for t in trades
if start_time <= t["timestamp"] <= end_time]
all_trades.extend(filtered)
# 获取下一批的起始ID
min_timestamp = min(t["timestamp"] for t in trades)
if min_timestamp < start_time:
break
current_from_id = trades[0]["id"] - 1
# API限速：每分钟1200请求，约50ms间隔
time.sleep(0.05)
df = pd.DataFrame(all_trades)
if not df.empty:
df["datetime"] = pd.to_datetime(df["timestamp"], unit="ms")
df = df.sort_values("timestamp").reset_index(drop=True)
return df
使用示例：获取BTCUSDT最近1小时的Tick数据
if __name__ == "__main__":
fetcher = BinanceTickDataFetcher()
end_time = int(datetime.now().timestamp() * 1000)
start_time = int((datetime.now() - timedelta(hours=1)).timestamp() * 1000)
df = fetcher.get_trades_by_timerange(
symbol="BTCUSDT",
start_time=start_time,
end_time=end_time
)
print(f"\n数据概览：")
print(f" 总记录数: {len(df)}")
print(f" 时间范围: {df['datetime'].min()} 至 {df['datetime'].max()}")
print(f" 价格范围: {df['price'].min():.2f} - {df['price'].max():.2f}")
print(f" 成交额总计: ${df['price'].multiply(df['qty']).sum():,.2f}")2.2 Bybit统一合约交易数据接口
Bybit以其低延迟著称，是高频交易者的热门选择。其最近成交查询API支持更灵活的参数配置。
import hashlib
import hmac
import requests
import time
from typing import Optional, List, Dict
import pandas as pd
class BybitTickDataFetcher:
"""Bybit合约Tick数据获取器"""
BASE_URL = "https://api.bybit.com"
def __init__(self, api_key: str = None, api_secret: str = None):
"""
初始化Bybit数据获取器
注意：公共API不需要签名，数据获取通常使用公共接口
"""
self.api_key = api_key
self.api_secret = api_secret
self.session = requests.Session()
self.recv_window = 5000 # 请求有效期（毫秒）
def _generate_signature(self, param_str: str) -> str:
"""生成HMAC SHA256签名"""
return hmac.new(
self.api_secret.encode("utf-8"),
param_str.encode("utf-8"),
hashlib.sha256
).hexdigest()
def get_public_recent_trades(self, category: str = "linear",
symbol: str = "BTCUSDT",
limit: int = 1000) -> List[Dict]:
"""
公共接口：获取最近成交
参数:
category: 合约类型 (linear=U永续, inverse=反向, option=期权)
symbol: 交易对
limit: 返回数量 (1-1000)
返回:
成交记录列表
"""
endpoint = "/v5/market/recent-trade"
params = {
"category": category,
"symbol": symbol,
"limit": limit
}
url = f"{self.BASE_URL}{endpoint}"
response = self.session.get(url, params=params)
data = response.json()
if data["retCode"] != 0:
raise ValueError(f"API错误: {data['retMsg']}")
return data["result"]["list"]
def get_trades_with_cursor(self, category: str = "linear",
symbol: str = "BTCUSDT",
limit: int = 1000,
cursor: Optional[str] = None) -> tuple:
"""
使用Cursor分页获取历史成交
返回:
(trades_list, next_cursor)
"""
endpoint = "/v5/market/recent-trade"
params = {
"category": category,
"symbol": symbol,
"limit": limit
}
if cursor:
params["cursor"] = cursor
url = f"{self.BASE_URL}{endpoint}"
response = self.session.get(url, params=params)
data = response.json()
if data["retCode"] != 0:
raise ValueError(f"API错误: {data['retMsg']}")
result = data["result"]
trades = result["list"]
next_cursor = result.get("nextPageCursor")
return trades, next_cursor
def batch_fetch_trades(self, symbol: str, target_count: int = 50000,
delay_ms: int = 100) -> pd.DataFrame:
"""
批量获取指定数量的历史成交
参数:
symbol: 交易对
target_count: 目标获取数量
delay_ms: 请求间隔（毫秒），避免触发限速
返回:
包含所有Tick的DataFrame
"""
all_trades = []
cursor = None
while len(all_trades) < target_count:
trades, cursor = self.get_trades_with_cursor(
symbol=symbol,
cursor=cursor,
limit=min(1000, target_count - len(all_trades))
)
all_trades.extend(trades)
print(f"已获取 {len(all_trades)}/{target_count} 条成交记录")
if not cursor:
break
time.sleep(delay_ms / 1000) # 限速保护
# 转换为DataFrame并标准化
df = pd.DataFrame(all_trades)
if not df.empty:
# 数据类型转换
df["trade_time"] = pd.to_datetime(df["tradeTime"].astype(int), unit="ms")
df["price"] = df["price"].astype(float)
df["size"] = df["size"].astype(float)
df = df.sort_values("trade_time").reset_index(drop=True)
return df
使用示例：获取Bybit BTCUSDT永续合约最近50000条成交
if __name__ == "__main__":
fetcher = BybitTickDataFetcher()
df = fetcher.batch_fetch_trades(
symbol="BTCUSDT",
target_count=50000,
delay_ms=100
)
print(f"\n=== Bybit BTCUSDT Tick数据统计 ===")
print(f"获取时间: {df['trade_time'].min()} ~ {df['trade_time'].max()}")
print(f"价格范围: ${df['price'].min():.2f} ~ ${df['price'].max():.2f}")
print(f"平均成交间隔: {df['trade_time'].diff().mean()}")
# 计算订单流不平衡
df["is_buy"] = df["side"] == "Buy"
ofi = df["is_buy"].rolling(100).sum() - (~df["is_buy"]).rolling(100).sum()
print(f"订单流不平衡(OFI)均值: {ofi.mean():.2f}")2.3 OKX交易所Tick数据获取
import requests
import pandas as pd
from datetime import datetime, timedelta
from typing import Optional
class OKXTickDataFetcher:
"""OKX交易所Tick数据获取器"""
BASE_URL = "https://www.okx.com"
def __init__(self):
self.session = requests.Session()
self.session.headers.update({
"Content-Type": "application/json"
})
def get_history_trades(self, inst_id: str = "BTC-USDT-SWAP",
limit: int = 100) -> list:
"""
获取OKX历史成交
参数:
inst_id: 合约ID (如 BTC-USDT-SWAP 永续合约)
limit: 获取数量 (1-100, 默认100)
返回:
成交记录列表
"""
endpoint = "/api/v5/market/trades"
params = {
"instId": inst_id,
"limit": str(min(limit, 100))
}
response = self.session.get(f"{self.BASE_URL}{endpoint}", params=params)
data = response.json()
if data["code"] != "0":
raise ValueError(f"OKX API错误: {data['msg']}")
return data["data"]
def fetch_historical_by_time(self, inst_id: str = "BTC-USDT-SWAP",
start: Optional[str] = None,
end: Optional[str] = None,
limit: int = 100) -> pd.DataFrame:
"""
按时间范围获取历史成交
时间格式: ISO 8601, 如 "2024-01-01T00:00:00.000Z"
"""
endpoint = "/api/v5/market/history-trades"
params = {
"instId": inst_id,
"limit": str(limit)
}
if start:
params["after"] = str(int(datetime.fromisoformat(
start.replace("Z", "+00:00")
).timestamp() * 1000))
if end:
params["before"] = str(int(datetime.fromisoformat(
end.replace("Z", "+00:00")
).timestamp() * 1000))
response = self.session.get(f"{self.BASE_URL}{endpoint}", params=params)
data = response.json()
if data["code"] != "0":
raise ValueError(f"OKX API错误: {data['msg']}")
df = pd.DataFrame(data["data"])
if not df.empty:
df["ts"] = pd.to_datetime(df["ts"].astype(int), unit="ms")
df["px"] = df["px"].astype(float)
df["sz"] = df["sz"].astype(float)
df = df.sort_values("ts").reset_index(drop=True)
return df
测试OKX数据获取
if __name__ == "__main__":
fetcher = OKXTickDataFetcher()
# 最近成交
trades = fetcher.get_history_trades(inst_id="BTC-USDT-SWAP", limit=500)
print(f"OKX最近500条成交获取成功")
# 按时间范围获取
df = fetcher.fetch_historical_by_time(
inst_id="BTC-USDT-SWAP",
start="2024-01-15T00:00:00.000Z",
end="2024-01-15T01:00:00.000Z",
limit=1000
)
print(f"时间范围数据: {len(df)} 条")
print(df.head())三、Tick数据存储与性能优化
3.1 数据存储选型对比
对于高频Tick数据，存储方案的选择直接影响回测速度和数据处理效率。以下是主流方案对比：
| 存储方案 | 写入速度 | 查询速度 | 存储成本 | 适用场景 | 压缩率 |
|---|---|---|---|---|---|
| Parquet + S3 | 中等 | 良好 | 低 | 云端架构、冷数据 | 3-5x |
| TimescaleDB | 高 | 优秀 | 中等 | 时序分析、实时监控 | 2-3x |
| ClickHouse | 极高 | 极优秀 | 中-高 | 超大规模回测、OLAP | 2-4x |
| Arrow IPC | 极高 | 优秀 | 低 | 本地回测、内存映射 | 2-3x |
| SQLite + WAL | 高 | 良好 | 极低 | 小规模研究、个人项目 | 1x |
3.2 高效Tick数据存储实现
import pyarrow as pa
import pyarrow.parquet as pq
import pyarrow.ipc as ipc
from pathlib import Path
from datetime import datetime, timedelta
import pandas as pd
import numpy as np
import struct
import os
class TickDataStorage:
"""高效Tick数据存储管理器"""
def __init__(self, base_path: str = "./tick_data"):
self.base_path = Path(base_path)
self.base_path.mkdir(parents=True, exist_ok=True)
# Arrow模式定义
self.schema = pa.schema([
("timestamp", pa.uint64), # Unix毫秒
("symbol", pa.string()),
("price", pa.float64()),
("quantity", pa.float64()),
("side", pa.uint8()), # 0=buy, 1=sell
("trade_id", pa.uint64()),
("flags", pa.uint8()) # 扩展标志位
])
def save_parquet(self, df: pd.DataFrame, symbol: str,
date: datetime) -> str:
"""
保存为Parquet格式（按交易日分区）
优势：列式存储、压缩率高、支持列裁剪
"""
date_str = date.strftime("%Y-%m-%d")
file_path = self.base_path / symbol / f"{symbol}_{date_str}.parquet"
file_path.parent.mkdir(parents=True, exist_ok=True)
# 确保数据类型正确
df_out = df.copy()
df_out["timestamp"] = df["timestamp"].astype(np.uint64)
df_out["side"] = df["side"].map({"buy": 0, "sell": 1, "Buy": 0, "Sell": 1})
df_out["side"] = df_out["side"].astype(np.uint8)
table = pa.Table.from_pandas(df_out, schema=self.schema)
pq.write_table(table, file_path, compression="snappy")
file_size = file_path.stat().st_size / 1024 / 1024
print(f"保存完成: {file_path} ({file_size:.2f} MB)")
return str(file_path)
def save_arrow_ipc(self, df: pd.DataFrame, symbol: str,
date: datetime) -> str:
"""
保存为Arrow IPC格式（适合内存映射快速读取）
优势：零拷贝读取、极致性能
"""
date_str = date.strftime("%Y-%m-%d")
file_path = self.base_path / symbol / f"{symbol}_{date_str}.arrow"
file_path.parent.mkdir(parents=True, exist_ok=True)
table = pa.Table.from_pandas(df, schema=self.schema)
with pa.OSFile(str(file_path), 'wb') as sink:
with ipc.new_file(sink, table.schema) as writer:
writer.write_table(table)
return str(file_path)
def load_parquet(self, symbol: str, date: datetime) -> pd.DataFrame:
"""加载指定日期的Tick数据"""
date_str = date.strftime("%Y-%m-%d")
file_path = self.base_path / symbol / f"{symbol}_{date_str}.parquet"
if not file_path.exists():
raise FileNotFoundError(f"数据文件不存在: {file_path}")
# 只加载需要的列（列裁剪优化）
table = pq.read_table(file_path)
return table.to_pandas()
def load_timerange(self, symbol: str, start: datetime,
end: datetime) -> pd.DataFrame:
"""
加载时间范围内的所有Tick数据
自动合并多个日期文件
"""
files = []
current = start.date()
while current <= end.date():
date_str = current.strftime("%Y-%m-%d")
file_path = self.base_path / symbol / f"{symbol}_{date_str}.parquet"
if file_path.exists():
files.append(file_path)
current += timedelta(days=1)
if not files:
raise FileNotFoundError(f"未找到 {symbol} 在 {start} 至 {end} 的数据")
# 读取并合并
tables = [pq.read_table(f) for f in files]
combined = pa.concat_tables(tables)
df = combined.to_pandas()
df = df[(df["timestamp"] >= int(start.timestamp() * 1000)) &
(df["timestamp"] <= int(end.timestamp() * 1000))]
return df.sort_values("timestamp").reset_index(drop=True)
def get_storage_stats(self, symbol: str) -> dict:
"""获取存储统计信息"""
symbol_path = self.base_path / symbol
if not symbol_path.exists():
return {"files": 0, "total_size_mb": 0, "dates": []}
files = list(symbol_path.glob("*.parquet"))
total_size = sum(f.stat().st_size for f in files)
dates = sorted(set(
f.stem.split("_")[1] for f in files
))
return {
"files": len(files),
"total_size_mb": total_size / 1024 / 1024,
"dates": dates
}
使用示例
if __name__ == "__main__":
storage = TickDataStorage("./tick_data")
# 模拟获取的Tick数据
sample_data = pd.DataFrame({
"timestamp": np.arange(1704067200000, 1704070800000, 100),
"symbol": ["BTCUSDT"] * len(range(0, 3600000, 100)),
"price": 46250 + np.random.randn(len(range(0, 3600000, 100))).cumsum() * 0.5,
"quantity": np.random.exponential(0.01, len(range(0, 3600000, 100))),
"side": np.random.choice(["buy", "sell"], len(range(0, 3600000, 100))),
"trade_id": np.arange(1704067200000, 1704070800000, 100)
})
# 保存数据
today = datetime.now()
storage.save_parquet(sample_data, "BTCUSDT", today)
storage.save_arrow_ipc(sample_data, "BTCUSDT", today)
# 查看统计
stats = storage.get_storage_stats("BTCUSDT")
print(f"\n存储统计: {stats}")四、数据质量控制与清洗
import pandas as pd
import numpy as np
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime, timedelta
@dataclass
class DataQualityReport:
"""数据质量报告"""
total_records: int
valid_records: int
duplicate_count: int
outlier_count: int
missing_fields: int
time_gaps: List[tuple]
price_jumps: List[tuple]
def print_summary(self):
print("=" * 50)
print("数据质量报告")
print("=" * 50)
print(f"总记录数: {self.total_records}")
print(f"有效记录: {self.valid_records}")
print(f"重复记录: {self.duplicate_count}")
print(f"异常值: {self.outlier_count}")
print(f"缺失字段: {self.missing_fields}")
print(f"时间间隙: {len(self.time_gaps)} 处")
print(f"价格跳变: {len(self.price_jumps)} 处")
class TickDataCleaner:
"""Tick数据清洗工具"""
def __init__(self, max_price_jump_pct: float = 0.01,
min_tick_interval_ms: int = 10):
"""
参数:
max_price_jump_pct: 最大允许价格跳变比例（1%）
min_tick_interval_ms: 最小成交间隔（10ms）
"""
self.max_price_jump = max_price_jump_pct
self.min_interval = min_tick_interval_ms
def validate_and_clean(self, df: pd.DataFrame) -> tuple[pd.DataFrame, DataQualityReport]:
"""
完整的数据验证和清洗流程
返回:
(清洗后的DataFrame, 质量报告)
"""
original_count = len(df)
# 1. 去除重复记录
df = df.drop_duplicates(subset=["timestamp", "trade_id"], keep="first")
dup_count = original_count - len(df)
# 2. 去除缺失值
required_fields = ["timestamp", "price", "quantity"]
df = df.dropna(subset=required_fields)
missing_count = len(df) - len(df.dropna(subset=required_fields))
# 3. 数据类型标准化
df["timestamp"] = df["timestamp"].astype(np.int64)
df["price"] = df["price"].astype(np.float64)
df["quantity"] = df["quantity"].astype(np.float64)
# 4. 去除价格异常值
df = self._remove_price_outliers(df)
# 5. 检测时间间隙
time_gaps = self._detect_time_gaps(df)
# 6. 检测价格跳变
price_jumps = self._detect_price_jumps(df)
# 生成报告
report = DataQualityReport(
total_records=original_count,
valid_records=len(df),
duplicate_count=dup_count,
outlier_count=0, # 在_remove_price_outliers中统计
missing_fields=missing_count,
time_gaps=time_gaps,
price_jumps=price_jumps
)
return df, report
def _remove_price_outliers(self, df: pd.DataFrame,
window: int = 100,
std_threshold: float = 5.0) -> pd.DataFrame:
"""
使用滚动Z-score移除价格异常值
"""
df = df.copy().sort_values("timestamp").reset_index(drop=True)
# 计算滚动均值和标准差
rolling_mean = df["price"].rolling(window, center=True, min_periods=1).mean()
rolling_std = df["price"].rolling(window, center=True, min_periods=1).std()
z_scores = np.abs((df["price"] - rolling_mean) / rolling_std)
outlier_mask = z_scores > std_threshold
outlier_count = outlier_mask.sum()
if outlier_count > 0:
print(f"发现 {outlier_count} 个价格异常值")
df = df[~outlier_mask].reset_index(drop=True)
return df
def _detect_time_gaps(self, df: pd.DataFrame,
max_gap_ms: int = 60000) -> List[tuple]:
"""
检测异常大的时间间隙
返回:
[(start_time, end_time, gap_ms), ...]
"""
df = df.sort_values("timestamp").reset_index(drop=True)
time_diffs = df["timestamp"].diff()
gaps = []
gap_indices = time_diffs > max_gap_ms
for idx in df[gap_indices].index[1:]:
start = df.loc[idx-1, "timestamp"]
end = df.loc[idx, "timestamp"]
gap_ms = end - start
gaps.append((start, end, gap_ms))
return gaps
def _detect_price_jumps(self, df: pd.DataFrame) -> List[tuple]:
"""
检测价格大幅跳变
返回:
[(timestamp, price_before, price_after, jump_pct), ...]
"""
df = df.sort_values("timestamp").reset_index(drop=True)
price_changes = df["price"].pct_change()
jumps = []
jump_mask = np.abs(price_changes) > self.max_price_jump
for idx in df[jump_mask].index[1:]:
timestamp = df.loc[idx, "timestamp"]
price_before = df.loc[idx-1, "price"]
price_after = df.loc[idx, "price"]
jump_pct = (price_after - price_before) / price_before * 100
jumps.append((timestamp, price_before, price_after, jump_pct))
return jumps
def resample_to_volume_bars(self, df: pd.DataFrame,
volume_per_bar: float) -> pd.DataFrame:
"""
重采样为成交量柱（Volume Bars）
相比时间柱，成交量柱在市场微观结构分析中更有意义
"""
df = df.copy().sort_values("timestamp").reset_index(drop=True)
df["cumvol"] = df["quantity"].cumsum()
df["bar_id"] = (df["cumvol"] / volume_per_bar).astype(int)
bars = df.groupby("bar_id").agg({
"timestamp": ["first", "last"],
"price": ["first", "last", "mean"],
"quantity": "sum",
"symbol": "first"
})
bars.columns = ["start_time", "end_time", "open", "close",
"vwap", "volume", "symbol"]
return bars
使用示例
if __name__ == "__main__":
cleaner = TickDataCleaner(max_price_jump_pct=0.005)
# 模拟带噪声的数据
np.random.seed(42)
n = 10000
df = pd.DataFrame({
"timestamp": np.arange(n) * 100 + 1704067200000,
"symbol": "BTCUSDT",
"price": 46250 + np.random.randn(n).cumsum() * 0.5,
"quantity": np.random.exponential(0.01, n),
"trade_id": np.arange(n) + 1704067200000,
"side": np.random.choice(["buy", "sell"], n)
})
# 添加一些异常
df.loc[100:105, "price"] = 50000 # 价格异常
df.loc[500, "timestamp"] = df.loc[500, "timestamp"] + 60000 # 时间间隙
# 清洗数据
df_clean, report = cleaner.validate_and_clean(df)
report.print_summary()
print(f"\n清洗后数据量: {len(df_clean)} 条")五、高频策略数据管道完整实现
import asyncio
import aiohttp
from concurrent.futures import ThreadPoolExecutor
import pandas as pd
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import time
from dataclasses import dataclass
from enum import Enum
class Exchange(Enum):
BINANCE = "binance"
BYBIT = "bybit"
OKX = "okx"
@dataclass
class FetchConfig:
exchange: Exchange
symbol: str
start_time: int # Unix ms
end_time: int # Unix ms
rate_limit_rpm: int = 600 # 每分钟请求数
class HighFrequencyDataPipeline:
"""
高频策略数据管道
功能：
1. 多交易所并行数据获取
2. 自动限速保护
3. 断点续传
4. 数据质量实时验证
"""
def __init__(self, max_workers: int = 3):
self.max_workers = max_workers
self.executor = ThreadPoolExecutor(max_workers=max_workers)
self.rate_limiter = RateLimiter()
async def fetch_binance_range(self, symbol: str,
start: datetime,
end: datetime) -> pd.DataFrame:
"""获取Binance指定时间范围数据"""
from_id = None
all_trades = []
current_time = int(start.timestamp() * 1000)
end_time = int(end.timestamp() * 1000)
while current_time < end_time:
await self.rate_limiter.acquire()
url = "https://api.binance.com/api/v3/historicalTrades"
params = {
"symbol": symbol.upper(),
"limit": 1000
}
if from_id:
params["fromId"] = from_id
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as resp:
if resp.status != 200:
print(f"请求失败: {resp.status}")
break
trades = await resp.json()
if not trades:
break