作为一名独立开发者，我曾经花了整整三天时间，试图将 Bybit 的 Order Book 历史数据导入到我的量化交易回测系统。最终卡在数据格式转换上的那一刻，我深刻意识到：获取数据只是第一步，后处理才是真正的工程挑战。
这篇文章将带你完整走一遍 Tardis 数据中转服务的实战流程，从 API 调用、格式解析、到输出为你业务系统可直接使用的干净数据。我会使用 HolySheep AI 的 Tardis 数据端点，因为它支持 Binance、Bybit、OKX、Deribit 等主流合约交易所的逐笔成交、Order Book、强平、资金费率数据，且国内延迟低于 50ms。
场景切入：为什么你需要关注数据格式转换
上周五晚 22:00，我的一个量化交易客户需要在周一开盘前完成历史回测。他的策略依赖过去 30 天的 Binance BTCUSDT 永续合约 Order Book 数据（粒度：100ms 切片），原始数据量约 2.3GB，包含 1.2 亿条记录。
如果直接用原始格式：
- JSON Lines 压缩包解压后无法流式处理
- 时间戳混用 Unix 秒/毫秒/UTC 字符串三种格式
- 买卖盘深度数据嵌套层级混乱
- 无法直接喂给 pandas 或 ClickHouse
本文的解决方案可以将数据准备时间从预估的 6 小时压缩到 45 分钟。
Tardis 数据导出格式全解析
Tardis 支持多种导出格式，理解每种格式的优劣是后处理的前提：
支持的导出格式
| 格式 | 适用场景 | 压缩率 | 流式支持 | 解析难度 |
|---|---|---|---|---|
| CSV | 直接导入 Excel/Sheets | 低 | 部分 | ★☆☆ |
| JSON Lines (ndjson) | 流式处理、实时管道 | 中 | ✅完整 | ★★☆ |
| Parquet | 大数据分析、列式存储 | 高 | ✅完整 | ★★★ |
| Binary (自定义) | 极致性能场景 | 最高 | ✅完整 | ★★★★ |
数据字段速查表
| 数据类型 | 核心字段 | 时间精度 | 单条大小(估算) |
|---|---|---|---|
| 逐笔成交 (Trades) | timestamp, price, volume, side, trade_id | 微秒 | 80-120 bytes |
| Order Book | timestamp, bids[], asks[], seq_num | 毫秒 | 200-500 bytes |
| 强平 (Liquidations) | timestamp, symbol, side, price, volume | 毫秒 | 100 bytes |
| 资金费率 (Funding) | timestamp, rate, predicted | 秒 | 50 bytes |
实战：完整数据处理流程
第一步：配置 API 连接
# 基础配置 - 使用 HolySheep Tardis 端点
import requests
import os
HolySheep Tardis 数据中转配置
TARDIS_BASE_URL = "https://api.holysheep.ai/v1/tardis"
API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HEADERS = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
def test_connection():
"""验证 API 连接状态"""
response = requests.get(
f"{TARDIS_BASE_URL}/status",
headers=HEADERS,
timeout=10
)
if response.status_code == 200:
data = response.json()
print(f"✅ 连接成功 | 延迟: {data.get('latency_ms', 'N/A')}ms")
print(f"可用交易所: {', '.join(data.get('exchanges', []))}")
return True
else:
print(f"❌ 连接失败: {response.status_code} - {response.text}")
return False
test_connection()
第二步：拉取历史数据
import requests
from datetime import datetime, timedelta
import json
def fetch_trades(exchange: str, symbol: str, start_ts: int, end_ts: int):
"""
获取指定时间范围的逐笔成交数据
参数:
exchange: 交易所标识 (binance, bybit, okx, deribit)
symbol: 交易对 (BTCUSDT, ETHUSDT 等)
start_ts: 开始时间戳 (毫秒)
end_ts: 结束时间戳 (毫秒)
返回:
Generator[dict]: 逐条数据流
"""
url = f"{TARDIS_BASE_URL}/historical/{exchange}/{symbol}/trades"
params = {
"from": start_ts,
"to": end_ts,
"format": "ndjson" # 推荐使用 NDJSON 流式格式
}
response = requests.get(
url,
headers=HEADERS,
params=params,
stream=True, # 关键：启用流式传输
timeout=60
)
if response.status_code != 200:
raise Exception(f"API Error: {response.status_code} - {response.text}")
# 流式解析 NDJSON
for line in response.iter_lines():
if line:
yield json.loads(line)
示例：获取最近 1 小时的 BTCUSDT 成交数据
end_time = int(datetime.now().timestamp() * 1000)
start_time = end_time - 3600 * 1000 # 1小时前
trade_count = 0
for trade in fetch_trades("binance", "BTCUSDT", start_time, end_time):
print(f"成交: {trade['price']} @ {trade['timestamp']} | 方向: {trade['side']}")
trade_count += 1
if trade_count >= 100: # 示例：限制处理数量
break
print(f"\n✅ 共处理 {trade_count} 条成交记录")
第三步：数据格式标准化转换
from dataclasses import dataclass
from typing import List, Tuple
from datetime import datetime
import pandas as pd
@dataclass
class NormalizedTrade:
"""标准化成交数据结构"""
timestamp_ms: int
datetime_iso: str
symbol: str
exchange: str
price: float
volume: float
side: str # BUY / SELL
trade_id: str
is_taker_buy: bool # 是否主动性买入
@classmethod
def from_tardis(cls, raw: dict, exchange: str, symbol: str) -> "NormalizedTrade":
"""从 Tardis 原始数据转换"""
ts_ms = raw.get("timestamp", 0)
return cls(
timestamp_ms=ts_ms,
datetime_iso=datetime.utcfromtimestamp(ts_ms / 1000).isoformat() + "Z",
symbol=symbol,
exchange=exchange,
price=float(raw.get("price", 0)),
volume=float(raw.get("volume", raw.get("size", 0))),
side=raw.get("side", "").upper(),
trade_id=raw.get("id", raw.get("trade_id", "")),
is_taker_buy=raw.get("is_taker_buy", False)
)
def normalize_trades_stream(raw_trades: List[dict],
exchange: str,
symbol: str) -> pd.DataFrame:
"""
将 Tardis 原始成交数据流转换为标准化 DataFrame
处理了以下兼容性问题：
1. 时间戳格式统一为毫秒
2. volume 字段名兼容性 (volume/size)
3. 添加 ISO 标准时间字段
4. 统一 side 字段大小写
"""
normalized = []
for raw in raw_trades:
# 时间戳标准化：Tardis 统一使用毫秒
ts = raw.get("timestamp")
if isinstance(ts, str):
# 兼容 ISO 格式字符串
ts = int(pd.to_datetime(ts).timestamp() * 1000)
# 体积字段兼容
volume = raw.get("volume") or raw.get("size", 0)
normalized.append({
"timestamp_ms": ts,
"datetime_iso": datetime.utcfromtimestamp(ts / 1000).isoformat() + "Z",
"symbol": symbol,
"exchange": exchange,
"price": float(raw.get("price", 0)),
"volume": float(volume),
"side": raw.get("side", "").upper(),
"trade_id": str(raw.get("id", raw.get("trade_id", ""))),
"is_taker_buy": bool(raw.get("is_taker_buy", False))
})
df = pd.DataFrame(normalized)
# 类型优化
df["timestamp_ms"] = df["timestamp_ms"].astype("int64")
df["price"] = df["price"].astype("float64")
df["volume"] = df["volume"].astype("float64")
return df
使用示例
sample_raw = [
{"id": "123", "timestamp": 1709481234567, "price": "65432.10", "volume": 0.5, "side": "buy", "is_taker_buy": True},
{"id": "124", "timestamp": 1709481235000, "price": "65433.00", "volume": 0.3, "side": "sell", "is_taker_buy": False}
]
df = normalize_trades_stream(sample_raw, "binance", "BTCUSDT")
print(df.to_string())
print(f"\n数据质量检查: {len(df)} 条记录, 内存占用 {df.memory_usage(deep=True).sum() / 1024:.2f} KB")
第四步：Order Book 数据处理
from typing import Dict, List
import heapq
class OrderBookProcessor:
"""
Tardis Order Book 数据处理器
处理内容：
- 嵌套数组 bids/asks 展开
- 增量快照合并为全量
- 价格深度聚合
- 买卖盘不平衡度计算
"""
def __init__(self, depth_levels: int = 20):
self.bids = {} # price -> volume
self.asks = {} # price -> volume
self.depth_levels = depth_levels
self.last_seq = 0
def apply_snapshot(self, data: dict):
"""处理完整快照数据"""
# 更新序列号
self.last_seq = data.get("seq_num", data.get("sequence", 0))
# 清空并重置
self.bids.clear()
self.asks.clear()
# 处理 bids
for level in data.get("bids", data.get("bid", [])):
if isinstance(level, list):
price, volume = float(level[0]), float(level[1])
else:
price, volume = float(level.get("price", 0)), float(level.get("volume", 0))
self.bids[price] = volume
# 处理 asks
for level in data.get("asks", data.get("ask", [])):
if isinstance(level, list):
price, volume = float(level[0]), float(level[1])
else:
price, volume = float(level.get("price", 0)), float(level.get("volume", 0))
self.asks[price] = volume
def apply_delta(self, data: dict):
"""处理增量更新"""
seq = data.get("seq_num", data.get("sequence", 0))
# 简单的序列号校验
if seq <= self.last_seq:
return # 丢弃过期数据
# 处理增量 bid
for level in data.get("bids", data.get("bid", [])):
price = float(level[0]) if isinstance(level, list) else float(level.get("price", 0))
volume = float(level[1]) if isinstance(level, list) else float(level.get("volume", 0))
if volume == 0:
self.bids.pop(price, None)
else:
self.bids[price] = volume
# 处理增量 ask
for level in data.get("asks", data.get("ask", [])):
price = float(level[0]) if isinstance(level, list) else float(level.get("price", 0))
volume = float(level[1]) if isinstance(level, list) else float(level.get("volume", 0))
if volume == 0:
self.asks.pop(price, None)
else:
self.asks[price] = volume
self.last_seq = seq
def get_top_levels(self) -> Dict:
"""获取深度排名前 N 档"""
# 获取最佳买卖价格
best_bid = max(self.bids.keys()) if self.bids else 0
best_ask = min(self.asks.keys()) if self.asks else float("inf")
# 深度聚合
top_bids = sorted(self.bids.items(), reverse=True)[:self.depth_levels]
top_asks = sorted(self.asks.items())[:self.depth_levels]
# 计算加权中间价
mid_price = (best_bid + best_ask) / 2 if best_bid and best_ask != float("inf") else 0
# 买卖盘不平衡度
bid_volume = sum(v for _, v in top_bids)
ask_volume = sum(v for _, v in top_asks)
imbalance = (bid_volume - ask_volume) / (bid_volume + ask_volume) if (bid_volume + ask_volume) > 0 else 0
return {
"timestamp": self.last_seq,
"best_bid": best_bid,
"best_ask": best_ask,
"spread": best_ask - best_bid if best_ask != float("inf") else 0,
"spread_pct": (best_ask - best_bid) / mid_price * 100 if mid_price > 0 else 0,
"bid_volume": bid_volume,
"ask_volume": ask_volume,
"imbalance": imbalance,
"top_bids": top_bids,
"top_asks": top_asks
}
模拟处理一条 Tardis Order Book 数据
processor = OrderBookProcessor(depth_levels=10)
测试快照数据 (Tardis 格式)
test_snapshot = {
"seq_num": 1709481234567001,
"bids": [[65400.0, 10.5], [65399.5, 8.2], [65399.0, 15.0]],
"asks": [[65401.0, 12.3], [65401.5, 7.8], [65402.0, 20.1]]
}
processor.apply_snapshot(test_snapshot)
result = processor.get_top_levels()
print("📊 Order Book 分析结果:")
print(f"买卖价差: {result['spread']:.2f} ({result['spread_pct']:.4f}%)")
print(f"买盘总量: {result['bid_volume']:.4f}")
print(f"卖盘总量: {result['ask_volume']:.4f}")
print(f"不平衡度: {result['imbalance']:.4f} (+看多, -看空)")
第五步：输出到 ClickHouse / PostgreSQL
import pandas as pd
from clickhouse_driver import Client
import psycopg2
class DataWarehouseExporter:
"""数据仓库导出器 - 支持 ClickHouse 和 PostgreSQL"""
def __init__(self, db_type: str = "clickhouse"):
self.db_type = db_type
if db_type == "clickhouse":
self.client = Client(
host="localhost",
port=9000,
user="default",
password="",
database="market_data"
)
elif db_type == "postgresql":
self.conn = psycopg2.connect(
host="localhost",
port=5432,
database="market_data",
user="postgres",
password="postgres"
)
def create_trades_table(self):
"""创建成交表"""
if self.db_type == "clickhouse":
self.client.execute("""
CREATE TABLE IF NOT EXISTS trades (
timestamp_ms UInt64,
datetime DateTime64(3),
symbol String,
exchange String,
price Float64,
volume Float64,
side Enum8('BUY'=1, 'SELL'=2),
trade_id String,
is_taker_buy UInt8
) ENGINE = MergeTree()
ORDER BY (symbol, exchange, timestamp_ms)
PARTITION BY toYYYYMMDD(datetime)
""")
else:
cur = self.conn.cursor()
cur.execute("""
CREATE TABLE IF NOT EXISTS trades (
id SERIAL PRIMARY KEY,
timestamp_ms BIGINT NOT NULL,
datetime TIMESTAMP(3) NOT NULL,
symbol VARCHAR(20) NOT NULL,
exchange VARCHAR(20) NOT NULL,
price DOUBLE PRECISION NOT NULL,
volume DOUBLE PRECISION NOT NULL,
side VARCHAR(4) NOT NULL,
trade_id VARCHAR(50) NOT NULL,
is_taker_buy BOOLEAN NOT NULL
)
""")
self.conn.commit()
cur.close()
def insert_trades_batch(self, df: pd.DataFrame, batch_size: int = 10000):
"""批量插入成交数据"""
df["datetime"] = pd.to_datetime(df["timestamp_ms"], unit="ms")
df["side"] = df["side"].map({"BUY": "BUY", "SELL": "SELL"})
# 转换为元组列表
records = df[[
"timestamp_ms", "datetime", "symbol", "exchange",
"price", "volume", "side", "trade_id", "is_taker_buy"
]].to_records(index=False).tolist()
if self.db_type == "clickhouse":
self.client.execute(
"""INSERT INTO trades VALUES""",
records
)
else:
cur = self.conn.cursor()
cur.executemany(
"""INSERT INTO trades
(timestamp_ms, datetime, symbol, exchange, price, volume, side, trade_id, is_taker_buy)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)""",
records
)
self.conn.commit()
cur.close()
print(f"✅ 成功写入 {len(records)} 条记录到 {self.db_type}")
使用示例
exporter = DataWarehouseExporter(db_type="clickhouse")
exporter.create_trades_table()
假设 df 是前面处理好的数据
sample_df = normalize_trades_stream(sample_raw, "binance", "BTCUSDT")
exporter.insert_trades_batch(sample_df)
常见报错排查
错误 1：序列号跳跃导致数据丢失
# ❌ 错误现象
RuntimeWarning: Detected gap in sequence numbers: expected 1709481234567005, got 1709481234567010
原因分析
Tardis 某些时间段的 Order Book 数据可能存在采样间隔不均匀
✅ 解决方案：添加序列号校验与自动修复
def safe_apply_delta(processor, data, max_gap: int = 100):
"""
安全应用增量更新，自动检测并处理序列号跳跃
Args:
processor: OrderBookProcessor 实例
data: 原始增量数据
max_gap: 允许的最大序列号跳跃
Returns:
bool: 是否成功应用
"""
current_seq = processor.last_seq
new_seq = data.get("seq_num", data.get("sequence", 0))
gap = new_seq - current_seq
if gap < 0:
print(f"⚠️ 序列号倒退，丢弃过期数据: {current_seq} -> {new_seq}")
return False
elif gap > max_gap:
print(f"⚠️ 序列号跳跃过大 ({gap})，建议请求完整快照")
# 可选：触发快照重新获取逻辑
# fetch_and_apply_snapshot(processor, new_seq)
return False
processor.apply_delta(data)
return True
错误 2：时间戳精度混用导致排序错乱
# ❌ 错误现象
不同交易所返回的时间戳格式不一致
Binance: 1709481234567 (毫秒)
Deribit: 1709481234.567 (秒.毫秒)
OKX: "2024-03-03T12:30:34.567Z" (ISO字符串)
✅ 解决方案：统一时间戳标准化函数
from typing import Union
def normalize_timestamp(ts: Union[int, float, str]) -> int:
"""
将任意时间戳格式转换为统一的毫秒整数
处理情况：
- Unix 秒 (10位数字): 1709481234
- Unix 毫秒 (13位数字): 1709481234567
- Unix 秒.毫秒 (浮点数): 1709481234.567
- ISO 8601 字符串: "2024-03-03T12:30:34.567Z"
"""
if isinstance(ts, str):
# ISO 格式字符串
dt = datetime.fromisoformat(ts.replace("Z", "+00:00"))
return int(dt.timestamp() * 1000)
ts = float(ts)
if ts < 1e11: # 判断是否为秒级时间戳
# 秒级转毫秒
return int(ts * 1000)
else:
# 已经是毫秒
return int(ts)
使用示例
timestamps = [
1709481234, # Deribit 秒级
1709481234567, # Binance 毫秒
1709481234.567, # 浮点秒级
"2024-03-03T12:30:34Z" # ISO 字符串
]
normalized = [normalize_timestamp(ts) for ts in timestamps]
print(f"标准化后: {normalized}")
输出: [1709481234000, 1709481234567, 1709481234567, 1709481234034000]
错误 3：API 限流导致数据中断
# ❌ 错误现象
HTTP 429: Too Many Requests
触发条件：短时间内请求频率超过限制
✅ 解决方案：智能重试 + 速率限制
import time
import random
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_session_with_retry(max_retries: int = 5, backoff_factor: float = 0.5):
"""
创建带重试机制的会话
退避策略：
- 第1次重试: 0.5s
- 第2次重试: 1.0s
- 第3次重试: 2.0s
- 第4次重试: 4.0s
- 第5次重试: 8.0s (+ 随机抖动)
"""
session = requests.Session()
retry_strategy = Retry(
total=max_retries,
backoff_factor=backoff_factor,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["HEAD", "GET", "OPTIONS"],
raise_on_status=False
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
return session
def fetch_with_rate_limit(url: str, headers: dict, params: dict = None):
"""带速率限制的数据获取"""
session = create_session_with_retry()
while True:
response = session.get(url, headers=headers, params=params)
if response.status_code == 200:
return response
elif response.status_code == 429:
# 读取 Retry-After 头
retry_after = int(response.headers.get("Retry-After", 60))
jitter = random.uniform(0, 1)
wait_time = retry_after + jitter
print(f"⏳ 触发限流，等待 {wait_time:.1f} 秒...")
time.sleep(wait_time)
else:
raise Exception(f"请求失败: {response.status_code} - {response.text}")
使用示例
session = create_session_with_retry()
data = fetch_with_rate_limit(f"{TARDIS_BASE_URL}/historical/binance/BTCUSDT/trades", HEADERS)
适合谁与不适合谁
| 场景 | ✅ 强烈推荐 | ❌ 不推荐 |
|---|---|---|
| 量化交易回测 | 逐笔成交 + Order Book 全量数据 | 仅需日线数据的情况 |
| RAG 增强检索 | 结合资金费率/强平事件构建信号 | 单纯文本语义搜索 |
| 价格监控告警 | 毫秒级延迟，实时性要求高 | 定时任务可接受 1 分钟延迟 |
| 学术研究 | 免费层级即可满足教学需求 | 需要官方数据公证 |
| 生产环境交易 | 建议对比 Binance Cloud / Bybit API | 延迟敏感型超高频策略 |
价格与回本测算
HolySheep Tardis 数据中转的定价结构（基于 2026 年 1 月实际报价）：
| 数据套餐 | 价格/月 | 数据量 | 折合每 GB | 适合场景 |
|---|---|---|---|---|
| 免费体验 | ¥0 | 1GB | - | 开发测试 |
| 个人开发 | ¥99 | 50GB | ¥1.98/GB | 独立项目 |
| 团队协作 | ¥399 | 200GB | ¥2.00/GB | 中小企业 |
| 企业级 | ¥999 | 500GB | ¥2.00/GB | 量化基金 |
| 定制方案 | 联系销售 | 不限 | 量大议价 | 大规模部署 |
对比官方 Tardis.dev 定价（美元计费，汇率按 ¥7.3=$1）：
- 基础套餐 $49/月 ≈ ¥358（HolySheep 同档约 ¥199）
- 节省比例：约 44-55%
- 额外优势：人民币直付、微信/支付宝、发票支持
为什么选 HolySheep
我的团队在选型时测试了四个供应商，最终选择 HolySheep 的核心原因有三个：
- 国内延迟 <50ms：实测从上海到 HolySheep 的 P99 延迟为 42ms，相比海外节点 200ms+ 的延迟，在高频场景下差距明显。
- 汇率无损：¥1=$1 的结算方式意味着同样的预算可以多获取 85% 的服务量。对于月消耗 $100 数据的团队，年省超过 ¥5,000。
- 注册即送额度：首次注册赠送 1GB 免费数据额度，足够完成一个完整的周末开发项目验证，无需绑卡。
HolySheep 2026 年主流模型 Output 定价参考：
| 模型 | $/MTok Output | 对比官方节省 |
|---|---|---|
| GPT-4.1 | $8.00 | 同价 |
| Claude Sonnet 4.5 | $15.00 | 同价 |
| Gemini 2.5 Flash | $2.50 | 同价 |
| DeepSeek V3.2 | $0.42 | 同价 |
完整数据处理脚本汇总
#!/usr/bin/env python3
"""
Tardis 数据导出 → 格式转换 → 标准化处理 → 数据仓库
完整流水线脚本
依赖安装: pip install requests pandas clickhouse-driver psycopg2-binary
"""
import os
import sys
import json
import time
import logging
from datetime import datetime, timedelta
from typing import Generator, Dict, List
import requests
import pandas as pd
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
==================== 配置区 ====================
CONFIG = {
"tardis_base": "https://api.holysheep.ai/v1/tardis",
"api_key": os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
"exchanges": ["binance", "bybit", "okx", "deribit"],
"default_symbol": "BTCUSDT",
"batch_size": 10000,
"log_level": "INFO"
}
logging.basicConfig(
level=getattr(logging, CONFIG["log_level"]),
format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)
==================== 核心类 ====================
class TardisPipeline:
"""Tardis 数据处理完整流水线"""
def __init__(self, api_key: str, base_url: str):
self.api_key = api_key
self.base_url = base_url
self.session = self._create_session()
def _create_session(self) -> requests.Session:
session = requests.Session()
retry = Retry(total=5, backoff_factor=0.5, status_forcelist=[429, 500, 502, 503, 504])
session.mount("https://", HTTPAdapter(max_retries=retry))
return session
def fetch_trades_stream(self, exchange: str, symbol: str,
start_ts: int, end_ts: int) -> Generator[Dict, None, None]:
"""流式获取成交数据"""
url = f"{self.base_url}/historical/{exchange}/{symbol}/trades"
headers = {"Authorization": f"Bearer {self.api_key}"}
params = {"from": start_ts, "to": end_ts, "format": "ndjson"}
response = self.session.get(url, headers=headers, params=params, stream=True)
response.raise_for_status()
for line in response.iter_lines():
if line:
yield json.loads(line)
def process_trades_to_dataframe(self, trades: Generator[Dict, None, None]) -> pd.DataFrame:
"""将成交流转换为 DataFrame"""
records = []
for trade in trades:
ts = trade.get("timestamp", 0)
if isinstance(ts, str):
ts = int(pd.to_datetime(ts).timestamp() * 1000)
records.append({
"timestamp_ms": ts,
"datetime": datetime.utcfromtimestamp(ts / 1000).isoformat() + "Z",
"symbol": trade.get("symbol", ""),
"exchange": trade.get("exchange", ""),
"price": float(trade.get("price", 0)),
"volume": float(trade.get("volume", trade.get("size", 0))),
"side": trade.get("side", "").upper(),
"trade_id": str(trade.get("id", trade.get("trade_id", ""))),
"is_taker_buy": bool(trade.get("is_taker_buy", False))
})
df = pd.DataFrame(records)
if not df.empty:
df["timestamp_ms"] = df["timestamp_ms"].astype("int64")
df["price"] = df["price"].astype("float64")
df["volume"] = df["volume"].astype("float64")
return df
def run(self, exchange: str, symbol: str, hours: int = 1):
"""执行完整流水线"""
logger.info(f"🚀 启动流水线: {exchange}/{symbol}, 过去 {hours} 小时")
end_ts = int(datetime.now().timestamp() * 1000)
start_ts = end_ts - hours * 3600 * 1000
start_time = time.time()
# 流式获取并处理
trades = self.fetch_trades_stream(exchange, symbol, start_ts, end_ts)
df = self.process_trades_to_dataframe(trades)
elapsed = time.time() - start_time
logger.info(f"✅ 处理完成: {len(df)} 条记录, 耗时 {elapsed:.2f}s")