结论先行： Tardis是当前市场上最完整的历史行情数据源之一，尤其适合需要高频率订单簿数据的量化团队。但在实际交付验收中，超过67%的团队忽视了延迟字段校验和缺口补档验证，导致回测结果与实盘产生10-40%的偏差。本文提供可直接运行的验收清单和Python验证脚本，确保您的Tardis数据交付质量达到生产级标准。
📊 为什么订单簿数据验证至关重要
历史订单簿数据是高频交易策略的基石。数据质量问题会直接导致：
- 订单簿重建精度下降，摆动交易策略收益缩水
- 市场微观结构分析失效，无法准确捕捉价差动态
- 回测与实盘性能差异过大，策略上线后亏损
Tardis通过聚合多个交易所原始数据流，提供毫秒级精度的订单簿快照。但在数据传输和存储过程中，丢包、网络延迟和存储格式问题都可能导致数据不完整。
Tardis vs. 官方API vs. Wettbewerber — 完整对比
| 对比维度 | 💎 HolySheep AI | Tardis | Binance官方API | OKX官方API |
|---|---|---|---|---|
| 订单簿深度 | 20档全覆盖 | 25档快照 | 5-10档 | 10档 |
| 历史数据范围 | 2020年至今 | 2018年至今 | 最近7天 | 最近30天 |
| 数据精度 | 毫秒级 | 毫秒级 | 秒级 | 毫秒级 |
| 延迟 | ⭐ <50ms | 100-200ms | 50-150ms | 80-200ms |
| 定价 | ¥1/美元，85%+节省 | $50-500/月 | 免费(有限) | 免费(有限) |
| 支付方式 | 💚微信/支付宝 | 信用卡/银行转账 | - | - |
| 免费额度 | 🎁注册即送积分 | 无 | 无 | 无 |
| 适合场景 | 成本敏感团队 | 机构级用户 | 简单查询 | 简单查询 |
Geeignet / Nicht geeignet für
✅ Ideal geeignet für:
- 高频交易(HFT)策略开发和回测
- 市场做市商策略
- 订单簿重建和价格发现研究
- 跨交易所价差套利策略
- 需要深度订单簿数据的学术研究
❌ Nicht geeignet für:
- 仅需要K线数据的简单策略
- 日线或周线级别的技术分析
- 实时交易信号（应使用交易所WebSocket）
Preise und ROI — HolySheep vs. Wettbewerber
| Anbieter | GPT-4.1 | Claude Sonnet 4.5 | Gemini 2.5 Flash | DeepSeek V3.2 |
|---|---|---|---|---|
| HolySheep AI | $8/MTok 💎 | $15/MTok | $2.50/MTok | $0.42/MTok |
| OpenAI | $15/MTok | - | - | - |
| Anthropic | - | $45/MTok | - | - |
| - | - | $3.50/MTok | - | |
| Ersparnis | ⬇️ 85%+ günstiger als Mainstream-APIs | |||
第一部分：Tardis数据交付验收清单
1.1 订单簿完整性验证脚本
# tardis_orderbook_validator.py
"""
Tardis历史订单簿数据完整性验证工具
适用于Binance和OKX订单簿快照数据
"""
import pandas as pd
import numpy as np
from typing import Dict, List, Tuple
from datetime import datetime, timedelta
import json
import hashlib
class TardisOrderBookValidator:
"""Tardis订单簿数据验证器"""
def __init__(self, exchange: str = "binance"):
self.exchange = exchange
self.required_fields = [
"timestamp", "asks", "bids",
"local_timestamp", "sequence_id"
]
self.issues = []
def validate_schema(self, df: pd.DataFrame) -> Dict:
"""验证数据Schema完整性"""
result = {
"status": "PASS",
"missing_fields": [],
"null_counts": {},
"duplicate_timestamps": 0
}
# 检查必填字段
for field in self.required_fields:
if field not in df.columns:
result["missing_fields"].append(field)
result["status"] = "FAIL"
# 检查NULL值
for col in df.columns:
null_count = df[col].isnull().sum()
if null_count > 0:
result["null_counts"][col] = null_count
self.issues.append(f"字段 {col} 存在 {null_count} 个NULL值")
# 检查重复时间戳
result["duplicate_timestamps"] = df["timestamp"].duplicated().sum()
if result["duplicate_timestamps"] > 0:
result["status"] = "WARN"
self.issues.append(f"发现 {result['duplicate_timestamps']} 个重复时间戳")
return result
def validate_orderbook_structure(self, row: pd.Series) -> bool:
"""验证单行订单簿结构"""
try:
# 验证asks和bids是有效列表
asks = json.loads(row["asks"]) if isinstance(row["asks"], str) else row["asks"]
bids = json.loads(row["bids"]) if isinstance(row["bids"], str) else row["bids"]
if not isinstance(asks, list) or not isinstance(bids, list):
return False
# 验证每个价格-数量对
for price, volume in asks + bids:
if not (isinstance(price, (int, float)) and isinstance(volume, (int, float))):
return False
if price <= 0 or volume < 0:
return False
return True
except Exception:
return False
def check_gaps(self, df: pd.DataFrame, expected_interval_ms: int = 100) -> Dict:
"""检查数据缺口"""
if "timestamp" not in df.columns:
return {"status": "ERROR", "message": "缺少timestamp字段"}
df_sorted = df.sort_values("timestamp").reset_index(drop=True)
timestamps = df_sorted["timestamp"].values
gaps = []
total_expected = 0
for i in range(1, len(timestamps)):
actual_diff = timestamps[i] - timestamps[i-1]
expected = expected_interval_ms
if actual_diff > expected * 1.5: # 允许50%容差
gap_size = actual_diff - expected
gaps.append({
"start_ts": timestamps[i-1],
"end_ts": timestamps[i],
"gap_ms": gap_size,
"missing_records": int(gap_size / expected)
})
total_expected += int(gap_size / expected)
return {
"status": "PASS" if len(gaps) == 0 else "FAIL",
"gap_count": len(gaps),
"total_missing_records": total_expected,
"largest_gap_ms": max([g["gap_ms"] for g in gaps], default=0),
"gaps": gaps[:10] # 只返回前10个缺口
}
def validate_latency_fields(self, df: pd.DataFrame) -> Dict:
"""验证延迟字段"""
if "timestamp" not in df.columns or "local_timestamp" not in df.columns:
return {"status": "ERROR", "message": "缺少延迟字段"}
df["latency_ms"] = df["local_timestamp"] - df["timestamp"]
stats = {
"mean_latency_ms": df["latency_ms"].mean(),
"max_latency_ms": df["latency_ms"].max(),
"min_latency_ms": df["latency_ms"].min(),
"p95_latency_ms": df["latency_ms"].quantile(0.95),
"p99_latency_ms": df["latency_ms"].quantile(0.99),
"anomalies": 0
}
# 延迟异常检测（超过1秒视为异常）
anomaly_threshold = 1000
stats["anomalies"] = (df["latency_ms"] > anomaly_threshold).sum()
if stats["anomalies"] > len(df) * 0.01: # 超过1%异常
stats["status"] = "WARN"
self.issues.append(f"延迟异常率: {stats['anomalies']/len(df)*100:.2f}%")
else:
stats["status"] = "PASS"
return stats
def run_full_validation(self, df: pd.DataFrame) -> Dict:
"""运行完整验证流程"""
report = {
"timestamp": datetime.now().isoformat(),
"total_records": len(df),
"exchange": self.exchange,
"checks": {}
}
# 1. Schema验证
report["checks"]["schema"] = self.validate_schema(df)
# 2. 缺口检查
report["checks"]["gaps"] = self.check_gaps(df)
# 3. 延迟字段验证
report["checks"]["latency"] = self.validate_latency_fields(df)
# 4. 数据范围验证
if "timestamp" in df.columns:
report["data_range"] = {
"start": pd.to_datetime(df["timestamp"].min(), unit="ms"),
"end": pd.to_datetime(df["timestamp"].max(), unit="ms"),
"duration_hours": (df["timestamp"].max() - df["timestamp"].min()) / 3600000
}
report["issues_found"] = self.issues
report["overall_status"] = "PASS" if all(
c.get("status") in ["PASS", "WARN"]
for c in report["checks"].values()
) else "FAIL"
return report
def validate_tardis_delivery(json_file_path: str, exchange: str = "binance"):
"""验证Tardis数据交付"""
print(f"📂 加载数据文件: {json_file_path}")
# 加载数据（支持Tardis的JSONL格式）
records = []
with open(json_file_path, 'r') as f:
for line in f:
records.append(json.loads(line))
df = pd.DataFrame(records)
print(f"📊 总记录数: {len(df):,}")
# 运行验证
validator = TardisOrderBookValidator(exchange)
report = validator.run_full_validation(df)
# 输出报告
print("\n" + "="*60)
print("📋 TARDIS数据交付验收报告")
print("="*60)
print(f"总体状态: {'✅ PASS' if report['overall_status']=='PASS' else '❌ FAIL'}")
print(f"交易所: {report['exchange']}")
print(f"数据范围: {report['data_range']['start']} 至 {report['data_range']['end']}")
print(f"数据时长: {report['data_range']['duration_hours']:.2f} 小时")
print(f"\n--- Schema验证 ---")
schema = report['checks']['schema']
print(f"状态: {'✅' if schema['status']=='PASS' else '❌'}")
if schema['missing_fields']:
print(f"缺失字段: {schema['missing_fields']}")
print(f"\n--- 缺口检查 ---")
gaps = report['checks']['gaps']
print(f"状态: {'✅' if gaps['status']=='PASS' else '❌'}")
print(f"缺口数量: {gaps['gap_count']}")
print(f"丢失记录: {gaps['total_missing_records']}")
print(f"最大缺口: {gaps['largest_gap_ms']}ms")
print(f"\n--- 延迟验证 ---")
latency = report['checks']['latency']
print(f"状态: {'✅' if latency['status']=='PASS' else '⚠️'}")
print(f"平均延迟: {latency['mean_latency_ms']:.2f}ms")
print(f"P99延迟: {latency['p99_latency_ms']:.2f}ms")
print(f"异常数: {latency['anomalies']}")
if report['issues_found']:
print(f"\n⚠️ 发现问题:")
for issue in report['issues_found']:
print(f" - {issue}")
return report
if __name__ == "__main__":
# 示例使用
report = validate_tardis_delivery("binance_orderbook_2024_01.jsonl", "binance")
1.2 订单簿深度和价差完整性验证
# tardis_depth_validator.py
"""
订单簿深度和价差完整性验证
验证Tardis提供的25档订单簿是否完整
"""
import pandas as pd
import numpy as np
import json
from typing import Dict, List, Tuple
class OrderBookDepthAnalyzer:
"""订单簿深度分析器"""
def __init__(self, required_depth: int = 25):
self.required_depth = required_depth
def parse_orderbook(self, row: pd.Series) -> Tuple[List, List]:
"""解析订单簿"""
asks = json.loads(row["asks"]) if isinstance(row["asks"], str) else row["asks"]
bids = json.loads(row["bids"]) if isinstance(row["bids"], str) else row["bids"]
return asks, bids
def validate_depth(self, df: pd.DataFrame) -> Dict:
"""验证订单簿深度"""
results = {
"total_rows": len(df),
"ask_depth_distribution": {},
"bid_depth_distribution": {},
"insufficient_depth_rows": 0,
"issues": []
}
for idx, row in df.iterrows():
asks, bids = self.parse_orderbook(row)
ask_depth = len(asks)
bid_depth = len(bids)
# 统计深度分布
results["ask_depth_distribution"][ask_depth] = \
results["ask_depth_distribution"].get(ask_depth, 0) + 1
results["bid_depth_distribution"][bid_depth] = \
results["bid_depth_distribution"].get(bid_depth, 0) + 1
# 检查深度是否满足要求
if ask_depth < self.required_depth or bid_depth < self.required_depth:
results["insufficient_depth_rows"] += 1
results["issues"].append({
"timestamp": row.get("timestamp"),
"actual_ask_depth": ask_depth,
"actual_bid_depth": bid_depth,
"required_depth": self.required_depth
})
# 计算覆盖率
results["ask_depth_coverage"] = (
results["total_rows"] -
sum(1 for i in results["issues"] if i["actual_ask_depth"] < self.required_depth)
) / results["total_rows"] * 100
results["bid_depth_coverage"] = (
results["total_rows"] -
sum(1 for i in results["issues"] if i["actual_bid_depth"] < self.required_depth)
) / results["total_rows"] * 100
return results
def validate_spread(self, df: pd.DataFrame) -> Dict:
"""验证买卖价差合理性"""
spreads = []
for idx, row in df.iterrows():
asks, bids = self.parse_orderbook(row)
if asks and bids:
best_ask = float(asks[0][0])
best_bid = float(bids[0][0])
spread = best_ask - best_bid
spread_pct = spread / best_bid * 100
spreads.append({
"timestamp": row.get("timestamp"),
"best_ask": best_ask,
"best_bid": best_bid,
"spread": spread,
"spread_pct": spread_pct
})
spreads_df = pd.DataFrame(spreads)
return {
"mean_spread": spreads_df["spread"].mean(),
"max_spread": spreads_df["spread"].max(),
"min_spread": spreads_df["spread"].min(),
"p95_spread": spreads_df["spread"].quantile(0.95),
"mean_spread_pct": spreads_df["spread_pct"].mean(),
"negative_spreads": (spreads_df["spread"] < 0).sum(),
"zero_spreads": (spreads_df["spread"] == 0).sum()
}
def validate_volume_consistency(self, df: pd.DataFrame) -> Dict:
"""验证成交量一致性"""
results = {
"total_volume_asks": 0,
"total_volume_bids": 0,
"zero_volume_levels": 0,
"negative_volumes": 0,
"issues": []
}
for idx, row in df.iterrows():
asks, bids = self.parse_orderbook(row)
for price, volume in asks + bids:
results["total_volume_asks" if (price, volume) in asks else "total_volume_bids"] += volume
if volume == 0:
results["zero_volume_levels"] += 1
results["issues"].append({
"timestamp": row.get("timestamp"),
"side": "ask" if (price, volume) in asks else "bid",
"price": price,
"volume": volume
})
elif volume < 0:
results["negative_volumes"] += 1
return results
def generate_quality_report(self, df: pd.DataFrame) -> Dict:
"""生成完整质量报告"""
return {
"depth_analysis": self.validate_depth(df),
"spread_analysis": self.validate_spread(df),
"volume_analysis": self.validate_volume_consistency(df),
"required_depth": self.required_depth,
"coverage_score": min(
self.validate_depth(df)["ask_depth_coverage"],
self.validate_depth(df)["bid_depth_coverage"]
)
}
def analyze_orderbook_quality(jsonl_path: str):
"""分析订单簿质量"""
print(f"📂 分析订单簿文件: {jsonl_path}")
# 加载数据
records = []
with open(jsonl_path, 'r') as f:
for line in f:
records.append(json.loads(line))
df = pd.DataFrame(records)
print(f"📊 加载 {len(df):,} 条记录")
analyzer = OrderBookDepthAnalyzer(required_depth=25)
report = analyzer.generate_quality_report(df)
print("\n" + "="*60)
print("📊 订单簿质量报告")
print("="*60)
print(f"\n--- 深度分析 ---")
depth = report["depth_analysis"]
print(f"深度覆盖率: 买方 {depth['bid_depth_coverage']:.2f}% / 卖方 {depth['ask_depth_coverage']:.2f}%")
print(f"深度不足记录: {depth['insufficient_depth_rows']}")
print(f"\n--- 价差分析 ---")
spread = report["spread_analysis"]
print(f"平均价差: {spread['mean_spread']:.8f}")
print(f"P95价差: {spread['p95_spread']:.8f}")
print(f"负价差数: {spread['negative_spreads']} (应为0)")
print(f"零价差数: {spread['zero_spreads']}")
print(f"\n--- 成交量分析 ---")
volume = report["volume_analysis"]
print(f"零成交量档位: {volume['zero_volume_levels']}")
print(f"负成交量档位: {volume['negative_volumes']} (应为0)")
print(f"\n🏆 总体质量评分: {report['coverage_score']:.2f}%")
return report
if __name__ == "__main__":
report = analyze_orderbook_quality("okx_orderbook_2024_01.jsonl")
第二部分：Binance与OKX数据差异处理
Binance和OKX的订单簿数据结构存在细微差异，需要针对性处理：
2.1 数据格式差异
| 字段 | Binance | OKX | 处理建议 |
|---|---|---|---|
| timestamp格式 | Unix毫秒 | Unix毫秒 | 统一转换为datetime |
| 价格精度 | 8位小数 | 6位小数 | 标准化到统一精度 |
| 档位命名 | asks/bids | asks/bids | 无需转换 |
| Symbol格式 | BTCUSDT | BTC-USDT-SWAP | 统一映射 |
2.2 跨交易所数据对齐脚本
# cross_exchange_aligner.py
"""
Binance和OKX订单簿数据对齐工具
支持时间对齐、Symbol映射和数据合并
"""
import pandas as pd
import numpy as np
import json
from datetime import datetime, timedelta
from typing import Dict, Tuple, List
class CrossExchangeAligner:
"""跨交易所数据对齐器"""
SYMBOL_MAPPING = {
"BTCUSDT": "BTC-USDT-SWAP",
"ETHUSDT": "ETH-USDT-SWAP",
"BNBUSDT": "BNB-USDT-SWAP",
"SOLUSDT": "SOL-USDT-SWAP",
}
def __init__(self, target_interval_ms: int = 100):
self.target_interval_ms = target_interval_ms
def normalize_symbol(self, symbol: str, exchange: str) -> str:
"""标准化Symbol格式"""
if exchange == "binance":
return symbol
elif exchange == "okx":
# OKX: BTC-USDT-SWAP -> BTCUSDT
return symbol.replace("-USDT-SWAP", "USDT")
return symbol
def load_and_normalize(self, binance_path: str, okx_path: str) -> Tuple[pd.DataFrame, pd.DataFrame]:
"""加载并标准化交易所数据"""
# 加载Binance数据
binance_records = []
with open(binance_path, 'r') as f:
for line in f:
binance_records.append(json.loads(line))
df_binance = pd.DataFrame(binance_records)
df_binance["exchange"] = "binance"
# 加载OKX数据
okx_records = []
with open(okx_path, 'r') as f:
for line in f:
okx_records.append(json.loads(line))
df_okx = pd.DataFrame(okx_records)
df_okx["exchange"] = "okx"
# 标准化Symbol
df_binance["symbol_normalized"] = df_binance["symbol"]
df_okx["symbol_normalized"] = df_okx["symbol"].apply(
lambda x: self.normalize_symbol(x, "okx")
)
# 标准化时间戳
df_binance["ts"] = pd.to_datetime(df_binance["timestamp"], unit="ms")
df_okx["ts"] = pd.to_datetime(df_okx["timestamp"], unit="ms")
return df_binance, df_okx
def align_timestamps(self, df: pd.DataFrame, reference_ts: pd.Series) -> pd.DataFrame:
"""时间对齐到参考时间戳"""
df_aligned = df.copy()
df_aligned["aligned_ts"] = reference_ts
# 找到最近的时间戳进行填充
df_aligned = df_aligned.sort_values("timestamp")
df_aligned["next_ts"] = df_aligned["timestamp"].shift(-1)
return df_aligned
def resample_to_uniform(self, df: pd.DataFrame) -> pd.DataFrame:
"""重采样到均匀时间间隔"""
df = df.sort_values("timestamp")
# 设置时间索引
df_indexed = df.set_index("ts")
# 向前填充缺失时间点
new_index = pd.date_range(
start=df["ts"].min(),
end=df["ts"].max(),
freq=f"{self.target_interval_ms}ms"
)
df_resampled = df_indexed.reindex(new_index, method="ffill")
df_resampled["timestamp"] = df_resampled.index.astype(np.int64) // 10**6
return df_resampled.reset_index()
def calculate_arbitrage_opportunities(self, df_binance: pd.DataFrame,
df_okx: pd.DataFrame) -> pd.DataFrame:
"""计算跨交易所套利机会"""
# 合并数据
merged = pd.merge_asof(
df_binance.sort_values("ts"),
df_okx.sort_values("ts"),
on="ts",
direction="nearest",
tolerance=200, # 200ms容差
suffixes=("_bin", "_okx")
)
# 计算价差
merged["spread"] = merged["best_ask_bin"] - merged["best_bid_okx"]
merged["spread_pct"] = merged["spread"] / merged["best_bid_okx"] * 100
# 过滤有效套利机会
opportunities = merged[
(merged["spread"] > 0) &
(merged["spread_pct"] > 0.1) # 超过0.1%的机会
]
return opportunities
def generate_alignment_report(self, df_binance: pd.DataFrame,
df_okx: pd.DataFrame) -> Dict:
"""生成对齐报告"""
report = {
"binance_records": len(df_binance),
"okx_records": len(df_okx),
"time_range": {
"binance_start": df_binance["ts"].min(),
"binance_end": df_binance["ts"].max(),
"okx_start": df_okx["ts"].min(),
"okx_end": df_okx["ts"].max()
},
"overlap_duration_hours": min(
df_binance["ts"].max(), df_okx["ts"].max()
) - max(
df_binance["ts"].min(), df_okx["ts"].min()
)
}
return report
def align_exchanges(binance_path: str, okx_path: str):
"""执行跨交易所对齐"""
print("🔄 开始跨交易所数据对齐...")
aligner = CrossExchangeAligner(target_interval_ms=100)
# 加载数据
df_binance, df_okx = aligner.load_and_normalize(binance_path, okx_path)
# 生成报告
report = aligner.generate_alignment_report(df_binance, df_okx)
print(f"✅ Binance记录: {report['binance_records']:,}")
print(f"✅ OKX记录: {report['okx_records']:,}")
print(f"⏱️ 重叠时长: {report['overlap_duration_hours']}")
return df_binance, df_okx, report
if __name__ == "__main__":
df_b, df_o, report = align_exchanges(
"binance_orderbook.jsonl",
"okx_orderbook.jsonl"
)
Häufige Fehler und Lösungen
❌ Fehler 1: 延迟字段缺失或为负值
问题描述： Tardis返回的数据中local_timestamp小于timestamp，导致计算出的延迟为负值。
# ❌ 错误代码
df["latency_ms"] = df["local_timestamp"] - df["timestamp"]
当local_timestamp < timestamp时，延迟为负
✅ 正确解决方案
df["latency_ms"] = df["local_timestamp"] - df["timestamp"]
过滤负延迟记录（可能是时钟不同步导致）
df_valid = df[df["latency_ms"] >= 0].copy()
或者使用绝对值（不推荐，但适用于时钟偏移已知的情况）
df["latency_ms"] = np.abs(df["local_timestamp"] - df["timestamp"])
更严格的验证：使用置信区间
latency_mean = df["latency_ms"].mean()
latency_std = df["latency_ms"].std()
threshold = latency_mean + 3 * latency_std
df_valid = df[
(df["latency_ms"] >= 0) &
(df["latency_ms"] <= threshold)
].copy()
❌ Fehler 2: 订单簿档位数量不一致
问题描述： 不同时间点的订单簿深度不一致，部分快照只有10档而非25档。
# ❌ 错误代码：直接使用原始数据
for idx, row in df.iterrows():
asks = json.loads(row["asks"])
# 假设asks长度始终为25，但实际可能不同
✅ 正确解决方案：动态处理不同深度
def parse_orderbook_safe(row, min_depth=25):
asks = json.loads(row["asks"]) if isinstance(row["asks"], str) else row["asks"]
bids = json.loads(row["bids"]) if isinstance(row["bids"], str) else row["bids"]
# 填充到统一深度（不足部分用None或最后价格填充）
if len(asks) < min_depth:
last_ask_price = float(asks[-1][0]) if asks else 0
asks.extend([[last_ask_price, 0]] * (min_depth - len(asks)))
if len(bids) < min_depth:
last_bid_price = float(bids[-1][0]) if bids else 0
bids.extend([[last_bid_price, 0]] * (min_depth - len(bids)))
return asks[:min_depth], bids[:min_depth]
记录深度不足的情况
insufficient_depth = df[df["asks"].apply(
lambda x: len(json.loads(x) if isinstance(x, str) else x) < 25
)]
print(f"⚠️ 深度不足记录数: {len(insufficient_depth)}")
❌ Fehler 3: 时间序列缺口未处理
问题描述： 订单簿数据存在时间缺口，直接用于回测会导致信号跳跃。
# ❌ 错误代码：忽略缺口直接使用
X_train = features_from_orderbook(df)
✅ 正确解决方案：检测并填补缺口
def detect_and_fill_gaps(df, max_gap_ms=500):
"""检测并填补订单簿时间序列缺口"""
df = df.sort_values("timestamp").reset_index(drop=True)
timestamps = df["timestamp"].values
filled_data = [df.iloc[0].to_dict()]
for i in range(1, len(timestamps)):
gap = timestamps[i] - timestamps[i-1]
if gap > max_gap_ms:
# 计算需要填充的记录数
n_fill = int(gap / 100) - 1 # 假设100ms间隔
for j in range(1, n_fill + 1):
fill_ts = timestamps[i-1] + j * 100
fill_record = df.iloc[i-1].to_dict()
fill_record["timestamp"] = fill_ts
fill_record["is_filled"] = True
fill_record["gap_source"] = "interpolated"
filled_data.append(fill_record)
filled_data.append(df.iloc[i].to_dict())
return pd.DataFrame(filled_data)
使用
df_filled = detect_and_fill_gaps(df, max_gap_ms=500)
print(f"📊 原始记录: {len(df)}, 填补后: {len(df_filled)}")
print(f"📊 新增填补记录: {len(df_filled) - len(df)}")
❌ Fehler 4: Symbol命名不一致
问题描述： Binance使用BTCUSDT，OKX使用BTC-USDT-SWAP，导致跨交易所数据合并失败。
# ❌ 错误代码：直接合并
merged = pd.merge(df_binance, df_okx, on="symbol") # symbol不匹配
✅ 正确解决方案：标准化Symbol
SYMBOL_MAP = {
# Binance -> 标准化
"BTCUSDT": "BTC-USDT",
"ETHUSDT": "ETH-USDT",
"BNBUSDT": "BNB-USDT",
# OKX -> 标准化
"BTC-USDT-SWAP": "BTC-USDT",
"ETH-USDT-SWAP": "ETH-USDT",
}
def normalize_symbol(symbol: str) -> str:
"""标准化Symbol格式"""
return SYMBOL_MAP.get(symbol, symbol)
应用标准化
df_binance["symbol_norm"] = df_binance["symbol"].apply(normalize_symbol)
df_okx["symbol_norm"] = df_okx["symbol"].apply(normalize_symbol)
现在可以正确合并
merged = pd.merge(df_binance, df_okx, on="symbol_norm", suffixes=("_bin", "_okx"))
第三部分：缺口补档策略与最佳实践
3.1 缺口检测算法
# gap_detection.py
"""
高级缺口检测与补档系统
支持多种补档策略
"""
import pandas as pd