在国内做高频加密货币量化交易，最头疼的不是策略开发，而是数据源可靠性。我见过太多团队花三个月写好策略，回测跑得漂亮，实盘一上线就亏成狗——问题往往出在数据质量上：逐笔成交跳帧、Order Book快照失真、强平记录缺失……今天用实际案例讲清楚如何用Tardis API做数据质量评估与异常检测，文末附我踩过的坑和解决方案。
先说个真实案例：2024年Q4，某高频策略团队用某数据商的历史Tick数据回测，年化收益38%。实盘三个月，收益-12%。我帮他们做数据审计，发现40%的成交量数据存在虚假放大——交易所原始数据的"隐单"被错误标记为真实成交。这一查，光debug就花了两周。所以数据质量检测不是可选项，是量化系统的生命线。
Tardis API数据质量评估体系
数据质量评估通常从四个维度入手：完整性（Completeness）、连续性（Continuity）、一致性（Consistency）、时效性（Timeliness）。我用HolySheep接入Tardis.dev的加密货币高频数据中转服务，支持Binance/Bybit/OKX/Deribit等主流交易所的逐笔成交、Order Book、强平、资金费率全量历史数据，国内直连延迟<50ms，比直接调用Tardis官方快3-5倍。
基础数据接入
#!/usr/bin/env python3
"""
Tardis API 数据质量评估基础接入
通过 HolySheep 中转服务获取加密货币高频历史数据
"""
import asyncio
import aiohttp
import json
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import statistics
class TardisDataQualityChecker:
def __init__(self, api_key: str, exchange: str = "binance"):
# 通过 HolySheep 中转 Tardis API
self.base_url = "https://api.holysheep.ai/tardis/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
self.exchange = exchange
self.symbols_cache = {}
async def fetch_trades(self, symbol: str, start_time: int, end_time: int) -> List[Dict]:
"""
获取指定时间段的逐笔成交数据
start_time/end_time: Unix timestamp (milliseconds)
"""
url = f"{self.base_url}/trades"
params = {
"exchange": self.exchange,
"symbol": symbol,
"from": start_time,
"to": end_time,
"limit": 100000
}
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=self.headers, params=params) as resp:
if resp.status == 200:
data = await resp.json()
return data.get("data", [])
else:
error_text = await resp.text()
raise Exception(f"API Error {resp.status}: {error_text}")
async def fetch_orderbook_snapshot(self, symbol: str, start_time: int, end_time: int) -> List[Dict]:
"""获取Order Book快照数据"""
url = f"{self.base_url}/book-snapshots"
params = {
"exchange": self.exchange,
"symbol": symbol,
"from": start_time,
"to": end_time,
"limit": 50000
}
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=self.headers, params=params) as resp:
if resp.status == 200:
data = await resp.json()
return data.get("data", [])
else:
raise Exception(f"Failed to fetch orderbook: {resp.status}")
async def fetch_liquidations(self, symbol: str, start_time: int, end_time: int) -> List[Dict]:
"""获取强平历史数据"""
url = f"{self.base_url}/liquidations"
params = {
"exchange": self.exchange,
"symbol": symbol,
"from": start_time,
"to": end_time,
"limit": 50000
}
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=self.headers, params=params) as resp:
if resp.status == 200:
data = await resp.json()
return data.get("data", [])
else:
raise Exception(f"Failed to fetch liquidations: {resp.status}")
async def main():
checker = TardisDataQualityChecker(
api_key="YOUR_HOLYSHEEP_API_KEY", # 替换为你的 HolySheep API Key
exchange="binance"
)
# 获取最近1小时的BTCUSDT逐笔成交数据
end_time = int(datetime.now().timestamp() * 1000)
start_time = end_time - 3600 * 1000
trades = await checker.fetch_trades("BTCUSDT", start_time, end_time)
print(f"获取到 {len(trades)} 条成交记录")
print(f"时间范围: {datetime.fromtimestamp(start_time/1000)} ~ {datetime.fromtimestamp(end_time/1000)}")
if __name__ == "__main__":
asyncio.run(main())这段代码演示了通过注册HolySheep获取Tardis数据的标准流程。国内直连延迟<50ms，相比直接调用Tardis官方API的200-400ms延迟，响应速度提升明显。
四大维度数据质量检测实现
#!/usr/bin/env python3
"""
Tardis API 数据质量评估核心实现
完整性、连续性、一致性、时效性四大维度检测
"""
import asyncio
import aiohttp
from collections import defaultdict
from datetime import datetime, timedelta
from typing import Dict, List, Tuple, Optional
import statistics
class DataQualityAnalyzer:
"""数据质量分析器"""
def __init__(self, api_key: str):
self.base_url = "https://api.holysheep.ai/tardis/v1"
self.headers = {"Authorization": f"Bearer {api_key}"}
# ============ 完整性检测 ============
def check_completeness(self, trades: List[Dict]) -> Dict:
"""
检测数据完整性：缺失字段、异常空值、数据量统计
"""
results = {
"total_records": len(trades),
"missing_fields": {},
"null_ratios": {},
"completeness_score": 0.0
}
if not trades:
return results
# 必检字段
required_fields = ["id", "price", "amount", "side", "timestamp"]
field_stats = defaultdict(lambda: {"total": 0, "null": 0, "empty": 0})
for trade in trades:
for field in required_fields:
field_stats[field]["total"] += 1
value = trade.get(field)
if value is None:
field_stats[field]["null"] += 1
elif value == "" or value == 0:
if field not in ["amount"]: # amount可能为0但仍是有效数据
field_stats[field]["empty"] += 1
# 计算缺失率
for field, stats in field_stats.items():
null_ratio = (stats["null"] + stats["empty"]) / max(stats["total"], 1)
results["null_ratios"][field] = round(null_ratio * 100, 2)
if null_ratio > 0:
results["missing_fields"][field] = {
"null_count": stats["null"],
"empty_count": stats["empty"],
"missing_rate": f"{null_ratio*100:.2f}%"
}
# 完整性得分 (0-100)
avg_missing = statistics.mean(results["null_ratios"].values())
results["completeness_score"] = round(max(0, 100 - avg_missing), 2)
return results
# ============ 连续性检测（跳帧检测）============
def check_continuity(self, trades: List[Dict], expected_interval_ms: int = 100) -> Dict:
"""
检测数据连续性：识别跳帧、重复记录、时间逆序
预期间隔：逐笔成交默认100ms（10条/秒）
"""
results = {
"total_gaps": 0,
"gaps": [],
"duplicate_ids": [],
"out_of_order": [],
"continuity_score": 0.0
}
if len(trades) < 2:
return results
# 按时间排序
sorted_trades = sorted(trades, key=lambda x: x.get("timestamp", 0))
# 检测ID重复
seen_ids = set()
for trade in sorted_trades:
trade_id = trade.get("id")
if trade_id in seen_ids:
results["duplicate_ids"].append(trade_id)
seen_ids.add(trade_id)
# 检测时间间隔异常
gap_threshold = expected_interval_ms * 5 # 超过5倍预期间隔判定为跳帧
prev_timestamp = sorted_trades[0].get("timestamp", 0)
for i in range(1, len(sorted_trades)):
curr_timestamp = sorted_trades[i].get("timestamp", 0)
interval = curr_timestamp - prev_timestamp
if interval > gap_threshold:
results["total_gaps"] += 1
results["gaps"].append({
"index": i,
"gap_ms": interval,
"expected_ms": expected_interval_ms,
"gap_ratio": round(interval / expected_interval_ms, 1),
"timestamp_before": prev_timestamp,
"timestamp_after": curr_timestamp
})
if interval < 0:
results["out_of_order"].append({
"index": i,
"timestamp": curr_timestamp,
"prev_timestamp": prev_timestamp
})
prev_timestamp = curr_timestamp
# 连续性得分
gap_ratio = results["total_gaps"] / max(len(sorted_trades) - 1, 1)
dup_ratio = len(results["duplicate_ids"]) / max(len(sorted_trades), 1)
results["continuity_score"] = round(max(0, 100 - (gap_ratio * 50 + dup_ratio * 50)), 2)
return results
# ============ 一致性检测 ============
def check_consistency(self, trades: List[Dict], orderbook: List[Dict] = None) -> Dict:
"""
检测数据一致性：价格合理性、成交量正负、订单簿均衡
"""
results = {
"price_anomalies": [],
"amount_anomalies": [],
"side_anomalies": [],
"consistency_score": 0.0,
"warnings": []
}
if not trades:
return results
prices = [float(t["price"]) for t in trades if t.get("price")]
amounts = [float(t["amount"]) for t in trades if t.get("amount")]
# 价格一致性检查
if prices:
mean_price = statistics.mean(prices)
stdev_price = statistics.stdev(prices) if len(prices) > 1 else 0
# 3σ原则检测价格尖峰
for i, trade in enumerate(trades):
price = float(trade.get("price", 0))
deviation = abs(price - mean_price)
if stdev_price > 0 and deviation > 3 * stdev_price:
results["price_anomalies"].append({
"index": i,
"price": price,
"mean": round(mean_price, 2),
"stdev": round(stdev_price, 2),
"deviation_sigma": round(deviation / stdev_price, 2),
"timestamp": trade.get("timestamp")
})
# 成交量正负检查
for i, trade in enumerate(trades):
amount = float(trade.get("amount", 0))
if amount <= 0:
results["amount_anomalies"].append({
"index": i,
"amount": amount,
"timestamp": trade.get("timestamp")
})
# Side与成交方向一致性（买单成交价应≥当前盘口卖一，卖单相反）
for i, trade in enumerate(trades):
side = trade.get("side", "").lower()
price = float(trade.get("price", 0))
# 简化检测：仅记录需要人工确认的异常
if side not in ["buy", "sell"]:
results["side_anomalies"].append({
"index": i,
"side": side,
"timestamp": trade.get("timestamp")
})
# 一致性得分
anomaly_count = len(results["price_anomalies"]) + len(results["amount_anomalies"])
anomaly_ratio = anomaly_count / max(len(trades), 1)
results["consistency_score"] = round(max(0, 100 - anomaly_ratio * 1000), 2)
return results
# ============ 时效性检测 ============
def check_timeliness(self, data: List[Dict], source_type: str = "trades") -> Dict:
"""
检测数据时效性：延迟情况、数据新鲜度
"""
results = {
"record_count": len(data),
"time_range": {},
"latency_stats": {},
"timeliness_score": 0.0
}
if not data:
return results
timestamps = [d.get("timestamp", 0) for d in data if d.get("timestamp")]
if timestamps:
min_ts = min(timestamps)
max_ts = max(timestamps)
results["time_range"] = {
"start": datetime.fromtimestamp(min_ts / 1000).isoformat(),
"end": datetime.fromtimestamp(max_ts / 1000).isoformat(),
"duration_ms": max_ts - min_ts
}
# 计算平均延迟（假设数据采集时间=当前时间）
now_ms = int(datetime.now().timestamp() * 1000)
delays = [now_ms - ts for ts in timestamps]
avg_delay = statistics.mean(delays)
median_delay = statistics.median(delays)
results["latency_stats"] = {
"avg_delay_ms": round(avg_delay, 2),
"median_delay_ms": round(median_delay, 2),
"max_delay_ms": max(delays),
"min_delay_ms": min(delays)
}
# 时效性得分：平均延迟<1s为满分
if avg_delay < 1000:
results["timeliness_score"] = 100
elif avg_delay < 5000:
results["timeliness_score"] = 80
else:
results["timeliness_score"] = max(0, 60 - (avg_delay - 5000) / 1000)
return results
async def run_full_quality_check():
"""执行完整数据质量检测"""
analyzer = DataQualityAnalyzer(api_key="YOUR_HOLYSHEEP_API_KEY")
# 获取测试数据（最近5分钟BTCUSDT逐笔成交）
end_time = int(datetime.now().timestamp() * 1000)
start_time = end_time - 300 * 1000
# 获取成交数据
async with aiohttp.ClientSession() as session:
url = f"{analyzer.base_url}/trades"
params = {"exchange": "binance", "symbol": "BTCUSDT", "from": start_time, "to": end_time, "limit": 10000}
async with session.get(url, headers=analyzer.headers, params=params) as resp:
trades = (await resp.json()).get("data", [])
# 执行四大维度检测
print("=" * 60)
print("数据质量检测报告")
print("=" * 60)
completeness = analyzer.check_completeness(trades)
print(f"\n【完整性】得分: {completeness['completeness_score']}/100")
print(f" - 总记录数: {completeness['total_records']}")
print(f" - 字段缺失: {completeness['missing_fields']}")
continuity = analyzer.check_continuity(trades, expected_interval_ms=100)
print(f"\n【连续性】得分: {continuity['continuity_score']}/100")
print(f" - 跳帧数量: {continuity['total_gaps']}")
print(f" - 重复ID: {len(continuity['duplicate_ids'])}")
consistency = analyzer.check_consistency(trades)
print(f"\n【一致性】得分: {consistency['consistency_score']}/100")
print(f" - 价格异常: {len(consistency['price_anomalies'])}")
print(f" - 成交量异常: {len(consistency['amount_anomalies'])}")
timeliness = analyzer.check_timeliness(trades)
print(f"\n【时效性】得分: {timeliness['timeliness_score']}/100")
print(f" - 平均延迟: {timeliness['latency_stats'].get('avg_delay_ms', 'N/A')}ms")
overall_score = statistics.mean([
completeness['completeness_score'],
continuity['continuity_score'],
consistency['consistency_score'],
timeliness['timeliness_score']
])
print(f"\n{'=' * 60}")
print(f"综合评分: {round(overall_score, 2)}/100")
print("=" * 60)
if __name__ == "__main__":
asyncio.run(run_full_quality_check())实际运行这段代码，你能看到四个维度的详细评分。我自己的经验是：综合评分低于85分的数据集，直接弃用或联系数据源修复；85-95分之间需要记录异常日志，人工复核；95分以上才能用于实盘策略。
常见异常数据检测模式与处理策略
1. 价格尖峰（Price Spike）
这是最常见的异常类型，通常由交易所故障、刷量机器人或数据采集器bug导致。识别标准：价格偏离均值超过3个标准差，或单笔成交价格与前后5笔均值差异超过1%。
#!/usr/bin/env python3
"""
异常数据检测与处理模块
Price Spike、Volume Wash、Order Book Imbalance 检测
"""
import statistics
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass
@dataclass
class AnomalyRecord:
"""异常记录数据结构"""
anomaly_type: str
index: int
severity: str # LOW, MEDIUM, HIGH, CRITICAL
details: Dict
original_value: float
suggested_value: Optional[float] = None
class AnomalyDetector:
"""异常数据检测器"""
def __init__(self, sigma_threshold: float = 3.0):
"""
sigma_threshold: 标准差倍数阈值，超过则判定为异常
"""
self.sigma_threshold = sigma_threshold
def detect_price_spike(self, trades: List[Dict], window_size: int = 20) -> List[AnomalyRecord]:
"""
检测价格尖峰异常
使用滑动窗口均值±标准差判断
"""
anomalies = []
for i in range(window_size, len(trades)):
window = trades[i-window_size:i]
prices = [float(t["price"]) for t in window if t.get("price")]
if len(prices) < 10:
continue
mean_price = statistics.mean(prices)
stdev_price = statistics.stdev(prices)
current_price = float(trades[i].get("price", 0))
deviation = abs(current_price - mean_price)
if stdev_price > 0 and deviation > self.sigma_threshold * stdev_price:
sigma_ratio = deviation / stdev_price
# 计算修正值（使用前后窗口均值）
prev_window = trades[max(0, i-window_size):i]
next_window = trades[i+1:min(len(trades), i+window_size)]
all_prices = [float(t["price"]) for t in prev_window + next_window if t.get("price")]
corrected_price = statistics.mean(all_prices) if all_prices else mean_price
severity = "HIGH" if sigma_ratio > 5 else "MEDIUM" if sigma_ratio > 4 else "LOW"
anomalies.append(AnomalyRecord(
anomaly_type="PRICE_SPIKE",
index=i,
severity=severity,
details={
"timestamp": trades[i].get("timestamp"),
"mean": round(mean_price, 2),
"stdev": round(stdev_price, 4),
"sigma_ratio": round(sigma_ratio, 2)
},
original_value=current_price,
suggested_value=round(corrected_price, 2)
))
return anomalies
def detect_volume_wash(self, trades: List[Dict], min_tick_size: float = 0.1) -> List[AnomalyRecord]:
"""
检测成交量清洗（虚假成交量）
识别模式：小间隔+固定数量+同向连续成交
"""
anomalies = []
for i in range(len(trades) - 5):
batch = trades[i:i+6]
# 检查时间间隔是否异常小（<10ms）
timestamps = [b.get("timestamp", 0) for b in batch]
intervals = [timestamps[j+1] - timestamps[j] for j in range(len(timestamps)-1)]
avg_interval = statistics.mean(intervals)
if avg_interval > 10: # 平均间隔超过10ms，跳过
continue
# 检查成交量是否相同
amounts = [float(b.get("amount", 0)) for b in batch]
if len(set(amounts)) > 2: # 超过2种不同数量，跳过
continue
# 检查方向是否一致
sides = [b.get("side", "").lower() for b in batch]
if len(set(sides)) > 1: # 方向不一致，跳过
continue
# 判定为可疑成交量清洗
anomalies.append(AnomalyRecord(
anomaly_type="VOLUME_WASH",
index=i,
severity="HIGH",
details={
"timestamp_start": timestamps[0],
"timestamp_end": timestamps[-1],
"interval_ms": round(avg_interval, 2),
"unique_amounts": len(set(amounts)),
"side": sides[0],
"affected_records": len(batch)
},
original_value=sum(amounts)
))
return anomalies
def detect_orderbook_imbalance(self, snapshots: List[Dict]) -> List[AnomalyRecord]:
"""
检测订单簿失衡
当bid/ask量比值超过5:1或1:5时标记为异常
"""
anomalies = []
for i, snapshot in enumerate(snapshots):
bids = snapshot.get("bids", [])
asks = snapshot.get("asks", [])
if not bids or not asks:
continue
bid_volume = sum(float(b.get("size", 0)) for b in bids[:10])
ask_volume = sum(float(a.get("size", 0)) for a in asks[:10])
if bid_volume == 0 or ask_volume == 0:
continue
imbalance_ratio = max(bid_volume, ask_volume) / min(bid_volume, ask_volume)
if imbalance_ratio > 5:
side = "BID_HEAVY" if bid_volume > ask_volume else "ASK_HEAVY"
severity = "CRITICAL" if imbalance_ratio > 10 else "HIGH"
anomalies.append(AnomalyRecord(
anomaly_type="ORDERBOOK_IMBALANCE",
index=i,
severity=severity,
details={
"timestamp": snapshot.get("timestamp"),
"bid_volume": round(bid_volume, 4),
"ask_volume": round(ask_volume, 4),
"imbalance_ratio": round(imbalance_ratio, 2),
"side": side
},
original_value=imbalance_ratio,
suggested_value=None # 订单簿失衡无法简单修正
))
return anomalies
============ 异常数据处理策略 ============
class AnomalyProcessor:
"""异常数据处理器"""
@staticmethod
def interpolate_price(anomalies: List[AnomalyRecord], trades: List[Dict]) -> List[Dict]:
"""
策略1：线性插值修正价格
用前后有效数据的均值替代异常值
"""
processed = [dict(t) for t in trades] # 深拷贝
# 按索引降序处理，避免索引偏移
for anomaly in sorted(anomalies, key=lambda x: x.index, reverse=True):
if anomaly.anomaly_type == "PRICE_SPIKE" and anomaly.suggested_value:
processed[anomaly.index]["price"] = anomaly.suggested_value
processed[anomaly.index]["_corrected"] = True
processed[anomaly.index]["_original_price"] = anomaly.original_value
return processed
@staticmethod
def remove_wash_trades(anomalies: List[AnomalyRecord], trades: List[Dict]) -> List[Dict]:
"""
策略2：移除虚假成交量记录
直接删除检测出的成交量清洗批次
"""
# 收集所有需要删除的索引
indices_to_remove = set()
for anomaly in anomalies:
if anomaly.anomaly_type == "VOLUME_WASH":
details = anomaly.details
start_idx = anomaly.index
count = details.get("affected_records", 0)
indices_to_remove.update(range(start_idx, start_idx + count))
# 过滤掉异常记录
processed = [t for i, t in enumerate(trades) if i not in indices_to_remove]
return processed
@staticmethod
def flag_for_review(anomalies: List[AnomalyRecord], trades: List[Dict]) -> List[Dict]:
"""
策略3：标记待人工复核
不修改原数据，仅添加标记字段
"""
processed = [dict(t) for t in trades]
anomaly_indices = {a.index for a in anomalies}
for i, trade in enumerate(processed):
if i in anomaly_indices:
trade["_flagged"] = True
trade["_flag_reason"] = next(
(a.anomaly_type for a in anomalies if a.index == i), "UNKNOWN"
)
return processed
使用示例
async def demo_anomaly_processing():
detector = AnomalyDetector(sigma_threshold=3.0)
processor = AnomalyProcessor()
# 假设 trades 是从 HolySheep API 获取的原始数据
# 1. 检测异常
price_spikes = detector.detect_price_spike(trades, window_size=20)
volume_washes = detector.detect_volume_wash(trades)
all_anomalies = price_spikes + volume_washes
print(f"检测到 {len(all_anomalies)} 条异常记录")
for a in all_anomalies[:5]: # 打印前5条
print(f" [{a.severity}] {a.anomaly_type} at index {a.index}: {a.details}")
# 2. 选择处理策略
# 策略A：修正价格 + 移除虚假成交量
corrected_trades = processor.interpolate_price(price_spikes, trades)
clean_trades = processor.remove_wash_trades(volume_washes, corrected_trades)
# 策略B：仅标记，不修改
flagged_trades = processor.flag_for_review(all_anomalies, trades)
print(f"处理后记录数: {len(clean_trades)} (原始: {len(trades)})")
return clean_trades, flagged_trades2. 强平数据异常处理
强平数据有个独特的问题：交易所的强平记录往往滞后3-5秒，且存在大量"预告型"数据（交易所提前宣告某大户即将强平但实际未成交）。我通常会交叉验证资金费率、标记价格、Order Book深度三个维度。
#!/usr/bin/env python3
"""
强平数据交叉验证与异常检测
"""
import asyncio
import aiohttp
from datetime import datetime
from typing import List, Dict, Tuple, Optional
class LiquidationValidator:
"""强平数据验证器"""
def __init__(self, api_key: str):
self.base_url = "https://api.holysheep.ai/tardis/v1"
self.headers = {"Authorization": f"Bearer {api_key}"}
async def fetch_funding_rate(self, exchange: str, symbol: str,
start_time: int, end_time: int) -> List[Dict]:
"""获取资金费率历史"""
async with aiohttp.ClientSession() as session:
url = f"{self.base_url}/funding-rate"
params = {
"exchange": exchange,
"symbol": symbol,
"from": start_time,
"to": end_time,
"limit": 1000
}
async with session.get(url, headers=self.headers, params=params) as resp:
return (await resp.json()).get("data", [])
async def cross_validate_liquidations(
self,
liquidations: List[Dict],
funding_rates: List[Dict],
orderbooks: List[Dict]
) -> Dict:
"""
交叉验证强平数据
验证逻辑：
1. 强平时段（通常在资金费率结算前15分钟）应该有高资金费率
2. 强平价格应与标记价格在合理范围内（±0.5%）
3. Order Book 应显示明显的失衡
"""
validation_results = {
"total_liquidations": len(liquidations),
"validated": [],
"suspicious": [],
"invalid": [],
"statistics": {}
}
# 构建资金费率查找表（按时间戳）
fr_lookup = {}
for fr in funding_rates:
ts = fr.get("timestamp", 0)
fr_lookup[ts] = fr
# 构建 Order Book 查找表
ob_lookup = {}
for ob in orderbooks:
ts = ob.get("timestamp", 0)
ob_lookup[ts] = ob
for liq in liquidations:
ts = liq.get("timestamp", 0)
liq_price = float(liq.get("price", 0))
liq_side = liq.get("side", "") # long or short
issues = []
# 检查1：附近是否有高资金费率（>0.01%）
nearby_fr = None
for fr_ts in sorted(fr_lookup.keys()):
if abs(fr_ts - ts) < 60000: # 1分钟内
nearby_fr = fr_lookup[fr_ts]
break
if nearby_fr:
fr_rate = float(nearby_fr.get("rate", 0))
if abs(fr_rate) < 0.0001: # 资金费率低于0.01%
issues.append(f"Low funding rate: {fr_rate*100:.4f}%")
# 检查2：标记价格合理性（假设我们从Order Book获取）
nearby_ob = None
for ob_ts in sorted(ob_lookup.keys()):
if abs(ob_ts - ts) < 5000: # 5秒内
nearby_ob = ob_lookup[ob_ts]
break
if nearby_ob and nearby_ob.get("mark_price"):
mark_price = float(nearby_ob["mark_price"])
price_diff = abs(liq_price - mark_price) / mark_price
if price_diff > 0.005: # 偏离超过0.5%
issues.append(f"Mark price deviation: {price_diff*100:.2f}%")
# 分类处理
if len(issues) == 0:
validation_results["validated"].append(liq)
elif len(issues) == 1:
liq_copy = dict(liq)
liq_copy["_warnings"] = issues
validation_results["suspicious"].append(liq_copy)
else:
liq_copy = dict(liq)
liq_copy["_issues"] = issues
validation_results["invalid"].append(liq_copy)
# 统计
total = len(liquidations)
validation_results["statistics"] = {
"validated_rate": f"{len(validation_results['validated'])/max(total,1)*100:.1f}%",
"suspicious_rate": f"{len(validation_results['suspicious'])/max(total,1)*100:.1f}%",
"invalid_rate": f"{len(validation_results['invalid'])/max(total,1)*100:.1f}%"
}
return validation_results
async def demo_liquidation_validation():
validator = LiquidationValidator(api_key="YOUR_HOLYSHEEP_API_KEY")
# 获取最近1小时的强平数据
end_time = int(datetime.now().timestamp() * 1000)
start_time = end_time - 3600 * 1000
# 并行获取多源数据
async with aiohttp.ClientSession() as session:
# 获取强平数据
liq_url = f"{validator.base_url}/liquidations"
liq_params = {"exchange": "binance", "symbol": "BTCUSDT", "from": start_time, "to": end_time}
liq_resp = await session.get(liq_url, headers=validator.headers, params=liq_params)
liquidations = (await liq_resp.json()).get("data", [])
# 获取资金费率
funding_rates = await validator.fetch_funding_rate("binance", "BTCUSDT", start_time, end_time)
# 获取Order Book快照
ob_url = f"{validator.base_url}/book-snapshots"
ob_params = {"exchange": "binance", "symbol":