在高频交易和量化策略开发中，订单簿（Order Book）数据的完整性直接决定了策略的有效性。作为一名在 HolySheep AI 负责行情数据架构的工程师，本文将从实战角度详细介绍如何系统化验收 Tardis 交付的 Binance 和 OKX 历史行情数据，确保订单簿完整性、延迟字段准确性和缺口补档机制完善。
为什么订单簿数据验收如此重要
我曾经历过一个惨痛的教训：某量化团队使用未经严格验收的订单簿数据进行回测，实盘上线后才发现历史数据中存在大量缺口，导致均值回归策略的买卖信号严重偏移。最终该团队在一天内亏损超过 15%，这充分说明了数据质量对交易系统的致命影响。
Tardis 作为专业的加密货币历史行情数据提供商，覆盖了 Binance、OKX、Bybit 等主流交易所的 tick 级数据。但数据交付并不意味着数据可用——我们需要一套完整的验收流程来确保数据符合生产环境标准。
验收清单概览
- 订单簿完整性验证：检查 bid/ask 价格连续性和深度覆盖
- 延迟字段准确性：验证 timestamp、local_timestamp 和 exchange_timestamp 的逻辑一致性
- 缺口检测与补档：识别并修复数据中的时间戳断层
- 数据类型和精度：确保价格、数量字段的数值精度符合要求
- 边界条件处理：验证停盘、熔断等特殊场景的数据处理
订单簿完整性验证
1. 价格连续性检查
订单簿的价格连续性是基础中的基础。正常市场情况下，相邻档位之间的价差（spread）应该呈现合理的分布。如果发现价格跳空超过正常范围，可能是数据采集或传输过程中出现了问题。
import pandas as pd
import numpy as np
from typing import Dict, List, Tuple
class OrderBookValidator:
"""Tardis 订单簿数据验证器"""
def __init__(self, exchange: str = "binance"):
self.exchange = exchange
self.max_spread_ratio = 0.05 # 最大价差比例为 5%
def check_price_continuity(self, df: pd.DataFrame, symbol: str) -> Dict:
"""
检查订单簿价格连续性
返回：{
'is_valid': bool,
'issues': List[Dict],
'spread_stats': Dict
}
"""
# 提取 bid 和 ask 价格
bids = df[[col for col in df.columns if 'bid' in col.lower() and 'price' in col.lower()]]
asks = df[[col for col in df.columns if 'ask' in col.lower() and 'price' in col.lower()]]
# 计算最佳买卖价差
best_bid = df['bid_price_0'] if 'bid_price_0' in df.columns else bids.min(axis=1)
best_ask = df['ask_price_0'] if 'ask_price_0' in df.columns else asks.min(axis=1)
spread = best_ask - best_bid
spread_ratio = spread / ((best_bid + best_ask) / 2)
# 检测异常价差
abnormal_spreads = df[spread_ratio > self.max_spread_ratio]
return {
'is_valid': len(abnormal_spreads) == 0,
'total_records': len(df),
'abnormal_count': len(abnormal_spreads),
'abnormal_ratio': len(abnormal_spreads) / len(df),
'spread_stats': {
'mean': spread_ratio.mean(),
'median': spread_ratio.median(),
'p95': spread_ratio.quantile(0.95),
'max': spread_ratio.max()
},
'issues': self._categorize_spread_issues(abnormal_spreads)
}
def check_depth_completeness(self, df: pd.DataFrame, expected_levels: int = 20) -> Dict:
"""
检查订单簿深度完整性
确保每档都有 bid 和 ask 数据
"""
issues = []
# 检查 bid 档位
for i in range(expected_levels):
bid_price_col = f'bid_price_{i}'
bid_qty_col = f'bid_qty_{i}'
if bid_price_col in df.columns:
null_prices = df[bid_price_col].isnull().sum()
null_qty = df[bid_qty_col].isnull().sum() if bid_qty_col in df.columns else 0
if null_prices > 0 or null_qty > 0:
issues.append({
'side': 'bid',
'level': i,
'null_prices': int(null_prices),
'null_qty': int(null_qty)
})
# 类似的检查 ask 档位...
return {
'is_valid': len(issues) == 0,
'expected_levels': expected_levels,
'issues': issues,
'completeness_ratio': 1 - (len(issues) / (expected_levels * 2))
}
def _categorize_spread_issues(self, abnormal_df: pd.DataFrame) -> List[Dict]:
"""对价差异常进行分类"""
categories = []
if len(abnormal_df) > 0:
categories.append({
'type': 'large_spread',
'description': '存在异常大的买卖价差',
'count': len(abnormal_df)
})
return categories
使用示例
validator = OrderBookValidator(exchange="binance")
df = pd.read_parquet("binance_btcusdt_orderbook_2024.parquet")
result = validator.check_price_continuity(df, "BTCUSDT")
print("订单簿验证器初始化完成")2. 数量和金额统计验证
除了价格，连续性还需要验证订单数量的合理性。负数数量、极端大数或零值都可能表示数据问题。
from dataclasses import dataclass
from decimal import Decimal
import statistics
@dataclass
class VolumeThresholds:
"""交易量阈值配置"""
max_single_order: float = 1000.0 # 单笔订单最大数量
min_single_order: float = 0.0001 # 最小交易量
max_total_bid_ratio: float = 0.8 # bid 总量占总量的最大比例
z_score_threshold: float = 5.0 # Z-score 异常检测阈值
class VolumeAnalyzer:
"""交易量异常检测器"""
def __init__(self, thresholds: VolumeThresholds = None):
self.thresholds = thresholds or VolumeThresholds()
def detect_anomalies(self, df: pd.DataFrame) -> Dict:
"""
检测订单簿数量异常
"""
anomalies = {
'negative_quantities': [],
'zero_quantities': [],
'extreme_values': [],
'imbalance_alerts': []
}
# 检查所有 bid 和 ask 数量列
qty_cols = [col for col in df.columns if 'qty' in col.lower()]
for col in qty_cols:
# 负数检测
negatives = df[df[col] < 0]
if len(negatives) > 0:
anomalies['negative_quantities'].append({
'column': col,
'count': len(negatives),
'sample_values': negatives[col].head().tolist()
})
# 零值检测
zeros = df[df[col] == 0]
if len(zeros) > len(df) * 0.1: # 超过 10% 为零值
anomalies['zero_quantities'].append({
'column': col,
'count': len(zeros),
'ratio': len(zeros) / len(df)
})
# 极端值检测（使用 Z-score）
mean_val = df[col].mean()
std_val = df[col].std()
if std_val > 0:
z_scores = ((df[col] - mean_val) / std_val).abs()
extremes = df[z_scores > self.thresholds.z_score_threshold]
if len(extremes) > 0:
anomalies['extreme_values'].append({
'column': col,
'count': len(extremes),
'threshold': self.thresholds.z_score_threshold
})
# 计算订单簿不平衡度
if 'bid_qty_0' in df.columns and 'ask_qty_0' in df.columns:
total_bid = df[[col for col in df.columns if 'bid_qty' in col]].sum(axis=1)
total_ask = df[[col for col in df.columns if 'ask_qty' in col]].sum(axis=1)
imbalance = (total_bid - total_ask) / (total_bid + total_ask)
# 检测极端不平衡
extreme_imb = df[imbalance.abs() > 0.5]
if len(extreme_imb) > 0:
anomalies['imbalance_alerts'].append({
'type': 'extreme_imbalance',
'count': len(extreme_imb),
'threshold': 0.5
})
return {
'has_anomalies': any(len(v) > 0 for v in anomalies.values()),
'anomalies': anomalies,
'summary': {
'total_anomaly_types': sum(1 for v in anomalies.values() if len(v) > 0)
}
}
集成验证报告生成
def generate_validation_report(df: pd.DataFrame, symbol: str) -> str:
"""生成完整的验证报告"""
validator = OrderBookValidator()
analyzer = VolumeAnalyzer()
continuity_result = validator.check_price_continuity(df, symbol)
depth_result = validator.check_depth_completeness(df)
volume_result = analyzer.detect_anomalies(df)
report = f"""
=== {symbol} 订单簿验证报告 ===
1. 价格连续性检查
状态: {'通过' if continuity_result['is_valid'] else '失败'}
总记录数: {continuity_result['total_records']}
异常比例: {continuity_result['abnormal_ratio']:.2%}
价差统计:
- 平均: {continuity_result['spread_stats']['mean']:.4%}
- 中位数: {continuity_result['spread_stats']['median']:.4%}
- P95: {continuity_result['spread_stats']['p95']:.4%}
2. 深度完整性检查
状态: {'通过' if depth_result['is_valid'] else '失败'}
完整度: {depth_result['completeness_ratio']:.2%}
问题数: {len(depth_result['issues'])}
3. 数量异常检测
状态: {'发现异常' if volume_result['has_anomalies'] else '正常'}
异常类型数: {volume_result['summary']['total_anomaly_types']}
"""
return report
print("交易量分析器初始化完成")延迟字段验证
Tardis 数据中包含三个关键时间戳字段，理解它们的含义对于验证数据质量至关重要：
- timestamp：Tardis 服务器接收数据的时间戳（UTC）
- local_timestamp：交易所原始时间戳，转换为 UTC
- exchange_timestamp：部分交易所提供的原始时间戳（可选）
from datetime import datetime, timezone
from typing import Optional, Tuple
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class LatencyValidator:
"""
验证 Tardis 数据中的延迟字段
关键指标：
- 网络延迟：从交易所到 Tardis 服务器的传输时间
- 乱序检测：识别 timestamp 不递增的记录
- 时区一致性：确保所有时间戳使用统一时区
"""
# Binance 和 OKX 的典型延迟范围（毫秒）
EXPECTED_LATENCY_RANGE = {
'binance': (5, 100), # 5-100ms
'okx': (10, 150) # 10-150ms
}
def __init__(self, exchange: str):
self.exchange = exchange.lower()
self.latency_range = self.EXPECTED_LATENCY_RANGE.get(
self.exchange,
(5, 200) # 默认范围
)
def calculate_latency(self, df: pd.DataFrame) -> pd.DataFrame:
"""
计算每条记录的实际延迟
延迟 = timestamp - local_timestamp
"""
if 'timestamp' not in df.columns or 'local_timestamp' not in df.columns:
raise ValueError("缺少必要的时间戳字段")
# 转换为 datetime（如果是 timestamp 格式）
ts = pd.to_datetime(df['timestamp'])
local_ts = pd.to_datetime(df['local_timestamp'])
# 计算延迟（毫秒）
latency_ms = (ts - local_ts).dt.total_seconds() * 1000
return latency_ms
def detect_out_of_order(self, df: pd.DataFrame) -> Dict:
"""
检测乱序数据
正常情况下 timestamp 应该严格递增
"""
ts = pd.to_datetime(df['timestamp'])
# 检测 timestamp 不递增的情况
ooo_indices = ts.diff() < pd.Timedelta(0)
ooo_count = ooo_indices.sum()
# 提取乱序记录
out_of_order_df = df[ooo_indices].copy() if ooo_count > 0 else pd.DataFrame()
return {
'total_records': len(df),
'out_of_order_count': int(ooo_count),
'out_of_order_ratio': float(ooo_count / len(df)),
'is_acceptable': ooo_count / len(df) < 0.001, # 允许小于 0.1%
'sample_records': out_of_order_df.head(5).to_dict('records') if ooo_count > 0 else []
}
def validate_latency_distribution(self, df: pd.DataFrame) -> Dict:
"""
验证延迟分布是否符合预期
异常高的延迟可能表示：
1. 网络问题
2. 数据采集服务中断
3. 交易所限流
"""
latency = self.calculate_latency(df)
# 统计信息
stats = {
'mean_ms': float(latency.mean()),
'median_ms': float(latency.median()),
'std_ms': float(latency.std()),
'p50': float(latency.quantile(0.50)),
'p95': float(latency.quantile(0.95)),
'p99': float(latency.quantile(0.99)),
'max_ms': float(latency.max()),
'min_ms': float(latency.min())
}
# 检测异常延迟
min_latency, max_latency = self.latency_range
# 延迟过低（可能是时间同步问题）
low_latency = latency[latency < min_latency]
# 延迟过高
high_latency = latency[latency > max_latency]
# 负延迟（严重问题）
negative_latency = latency[latency < 0]
issues = []
if len(negative_latency) > 0:
issues.append({
'type': 'negative_latency',
'count': len(negative_latency),
'description': '负延迟表示 timestamp < local_timestamp，这是严重的数据问题'
})
if len(low_latency) > 0:
issues.append({
'type': 'unusually_low_latency',
'count': len(low_latency),
'min_expected_ms': min_latency,
'description': '延迟低于预期范围，可能是时间同步问题'
})
if len(high_latency) > 0:
issues.append({
'type': 'high_latency',
'count': len(high_latency),
'max_expected_ms': max_latency,
'description': '延迟高于预期范围，需要检查网络或服务状态'
})
return {
'is_valid': len(issues) == 0,
'stats': stats,
'issues': issues,
'low_latency_count': len(low_latency),
'high_latency_count': len(high_latency),
'negative_latency_count': len(negative_latency)
}
def check_timestamp_consistency(self, df: pd.DataFrame) -> Dict:
"""
检查同一时间窗口内的数据一致性
"""
ts = pd.to_datetime(df['timestamp'])
# 按秒分组，统计每秒消息数
msg_per_second = ts.dt.floor('S').value_counts().sort_index()
# 检测消息数异常（过多或过少）
mean_msgs = msg_per_second.mean()
std_msgs = msg_per_second.std()
anomalies = []
for ts_val, count in msg_per_second.items():
z_score = (count - mean_msgs) / std_msgs if std_msgs > 0 else 0
if abs(z_score) > 3: # 超过 3 个标准差
anomalies.append({
'timestamp': str(ts_val),
'count': int(count),
'z_score': float(z_score)
})
return {
'is_consistent': len(anomalies) / len(msg_per_second) < 0.05,
'anomaly_ratio': len(anomalies) / len(msg_per_second) if len(msg_per_second) > 0 else 0,
'mean_msgs_per_second': float(mean_msgs),
'anomalies': anomalies[:10] # 只返回前 10 个
}
使用示例
def validate_tardis_data(file_path: str, exchange: str) -> Dict:
"""
完整的 Tardis 数据验证流程
"""
logger.info(f"开始验证 {exchange} 数据: {file_path}")
df = pd.read_parquet(file_path)
validator = LatencyValidator(exchange)
results = {
'exchange': exchange,
'record_count': len(df),
'time_range': {
'start': str(df['timestamp'].min()),
'end': str(df['timestamp'].max())
}
}
# 1. 乱序检测
results['out_of_order'] = validator.detect_out_of_order(df)
# 2. 延迟分布验证
results['latency'] = validator.validate_latency_distribution(df)
# 3. 时间戳一致性检查
results['consistency'] = validator.check_timestamp_consistency(df)
# 生成报告
is_valid = (
results['out_of_order']['is_acceptable'] and
results['latency']['is_valid'] and
results['consistency']['is_consistent']
)
results['final_verdict'] = 'PASS' if is_valid else 'FAIL'
logger.info(f"验证完成: {results['final_verdict']}")
return results
print("延迟字段验证器初始化完成")缺口检测与补档机制
历史数据中的时间戳缺口是量化策略开发中的隐形杀手。即便是微小的数据缺口，也可能导致指标计算错误、信号延迟或遗漏。因此，建立完善的缺口检测和补档机制是数据验收的关键环节。
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
from collections import defaultdict
import hashlib
@dataclass
class GapConfig:
"""缺口检测配置"""
symbol: str
exchange: str
expected_interval_ms: int # 预期时间间隔（毫秒）
max_gap_threshold_ms: int # 最大允许缺口（超过此值判定为缺口）
check_interval: str = '1min' # 检查间隔
def __post_init__(self):
# 根据交易对类型设置合适的参数
if 'usdt' in self.symbol.lower() or 'usd' in self.symbol.lower():
self.expected_interval_ms = 100 # USDT 永续合约 100ms
else:
self.expected_interval_ms = 1000 # 其他 1s
@dataclass
class GapRecord:
"""缺口记录"""
start_time: pd.Timestamp
end_time: pd.Timestamp
gap_duration_ms: float
gap_size: int # 缺失的消息数
severity: str # minor, major, critical
cause: Optional[str] = None
class GapDetector:
"""
Tardis 数据缺口检测与补档管理
常见缺口原因：
1. 网络中断
2. 交易所维护
3. Tardis 服务重启
4. 限流导致的数据丢失
"""
SEVERITY_THRESHOLDS = {
'minor': 1, # 1-10 个预期间隔
'major': 10, # 10-100 个预期间隔
'critical': 100 # 超过 100 个预期间隔
}
def __init__(self, config: GapConfig):
self.config = config
self.gaps: List[GapRecord] = []
def detect_gaps(self, df: pd.DataFrame) -> Dict:
"""
检测数据中的时间戳缺口
策略：
1. 计算相邻记录的时间差
2. 与预期间隔比较
3. 超过阈值标记为缺口
"""
if 'timestamp' not in df.columns:
raise ValueError("数据中缺少 timestamp 字段")
ts = pd.to_datetime(df['timestamp']).sort_values()
time_diffs = ts.diff()
# 转换为毫秒
diffs_ms = time_diffs.dt.total_seconds() * 1000
# 找出超过阈值的间隔
threshold = self.config.max_gap_threshold_ms
gap_indices = diffs_ms > threshold
# 收集缺口信息
gaps = []
gap_starts = ts[gap_indices].index
for idx in gap_starts:
gap_start = ts[idx]
gap_end = ts[idx + 1] if idx + 1 < len(ts) else None
if gap_end is None:
continue
gap_duration = (gap_end - gap_start).total_seconds() * 1000
expected_intervals = gap_duration / self.config.expected_interval_ms
# 确定严重程度
if expected_intervals > self.SEVERITY_THRESHOLDS['critical']:
severity = 'critical'
elif expected_intervals > self.SEVERITY_THRESHOLDS['major']:
severity = 'major'
else:
severity = 'minor'
gaps.append(GapRecord(
start_time=gap_start,
end_time=gap_end,
gap_duration_ms=gap_duration,
gap_size=int(expected_intervals),
severity=severity,
cause=self._infer_gap_cause(gap_duration)
))
self.gaps = gaps
# 统计摘要
by_severity = defaultdict(int)
for gap in gaps:
by_severity[gap.severity] += 1
total_gap_time = sum(g.gap_duration_ms for g in gaps)
return {
'total_gaps': len(gaps),
'by_severity': dict(by_severity),
'total_gap_duration_ms': total_gap_time,
'total_gap_duration_hours': total_gap_time / 3600000,
'gap_ratio': total_gap_time / ((ts.max() - ts.min()).total_seconds() * 1000),
'gaps': [
{
'start': str(g.start_time),
'end': str(g.end_time),
'duration_ms': g.gap_duration_ms,
'severity': g.severity,
'cause': g.cause
}
for g in gaps
],
'is_acceptable': self._evaluate_acceptance(gaps)
}
def _infer_gap_cause(self, duration_ms: float) -> str:
"""推断缺口原因"""
# 常见的固定时间窗口
known_windows = {
3600000: "交易所 1 小时维护",
300000: "交易所 5 分钟维护",
60000: "交易所 1 分钟快照",
5000: "网络抖动",
30000: "Tardis 服务重启"
}
for expected_ms, cause in known_windows.items():
if abs(duration_ms - expected_ms) < 5000: # 5 秒容差
return cause
return "未知原因"
def _evaluate_acceptance(self, gaps: List[GapRecord]) -> Dict:
"""评估缺口是否在可接受范围内"""
# 关键策略：不允许 critical 级别缺口
critical_gaps = [g for g in gaps if g.severity == 'critical']
# major 缺口占比不应超过 1%
major_gaps = [g for g in gaps if g.severity == 'major']
total_expected = sum(g.gap_size for g in gaps)
critical_ratio = sum(g.gap_size for g in critical_gaps) / total_expected if total_expected > 0 else 0
return {
'is_acceptable': len(critical_gaps) == 0 and critical_ratio < 0.01,
'critical_count': len(critical_gaps),
'critical_ratio': critical_ratio,
'recommendation': 'REJECT' if len(critical_gaps) > 0 else 'ACCEPT_WITH_NOTES'
}
def generate_fill_request(self) -> Dict:
"""
生成缺口补档请求
用于向 Tardis 提交补档申请
"""
if not self.gaps:
return {'has_gaps': False, 'requests': []}
requests = []
for gap in self.gaps:
if gap.severity in ['major', 'critical']:
requests.append({
'symbol': self.config.symbol,
'exchange': self.config.exchange,
'start_time': gap.start_time.isoformat(),
'end_time': gap.end_time.isoformat(),
'priority': 'high' if gap.severity == 'critical' else 'medium'
})
return {
'has_gaps': True,
'total_requests': len(requests),
'requests': requests
}
完整的验收工作流
class TardisDeliveryValidator:
"""
Tardis 数据交付完整验收工作流
包含三大核心验证：
1. 订单簿完整性
2. 延迟字段准确性
3. 缺口检测与补档
"""
def __init__(self, symbol: str, exchange: str):
self.symbol = symbol
self.exchange = exchange
self.orderbook_validator = OrderBookValidator(exchange)
self.latency_validator = LatencyValidator(exchange)
gap_config = GapConfig(
symbol=symbol,
exchange=exchange,
expected_interval_ms=100,
max_gap_threshold_ms=500 # 超过 500ms 视为缺口
)
self.gap_detector = GapDetector(gap_config)
def run_full_validation(self, df: pd.DataFrame) -> Dict:
"""执行完整验收流程"""
results = {
'metadata': {
'symbol': self.symbol,
'exchange': self.exchange,
'record_count': len(df),
'validation_time': datetime.now(timezone.utc).isoformat()
}
}
# 1. 订单簿验证
results['orderbook'] = {
'continuity': self.orderbook_validator.check_price_continuity(df, self.symbol),
'depth': self.orderbook_validator.check_depth_completeness(df)
}
# 2. 延迟验证
results['latency'] = {
'distribution': self.latency_validator.validate_latency_distribution(df),
'out_of_order': self.latency_validator.detect_out_of_order(df)
}
# 3. 缺口检测
results['gaps'] = self.gap_detector.detect_gaps(df)
# 综合判定
results['verdict'] = self._calculate_final_verdict(results)
return results
def _calculate_final_verdict(self, results: Dict) -> Dict:
"""计算最终验收结论"""
checks = {
'orderbook_continuity': results['orderbook']['continuity']['is_valid'],
'orderbook_depth': results['orderbook']['depth']['is_valid'],
'latency_acceptable': results['latency']['distribution']['is_valid'],
'out_of_order_acceptable': results['latency']['out_of_order']['is_acceptable'],
'gaps_acceptable': results['gaps']['is_acceptable']
}
passed = sum(1 for v in checks.values() if v)
total = len(checks)
return {
'passed_checks': passed,
'total_checks': total,
'pass_rate': passed / total,
'status': 'PASS' if passed == total else ('CONDITIONAL_PASS' if passed >= 4 else 'FAIL'),
'checks': checks
}
API 补档请求示例（使用 HolySheep AI）
def request_gap_fill_via_api(gaps: List[Dict], api_key: str) -> Dict:
"""
通过 HolySheep AI API 请求 Tardis 补档
API 端点: https://api.holysheep.ai/v1
"""
import requests
# 注意：这里仅为示例，实际使用时需要替换为真实的 Tardis 补档 API
# HolySheep AI 提供对多种数据源的集成访问
endpoint = "https://api.holysheep.ai/v1/tardis/fill-gaps"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"gaps": gaps,
"priority": "high"
}
# 实际调用时取消注释：
# response = requests.post(endpoint, json=payload, headers=headers)
# return response.json()
return {"status": "demo", "message": "实际使用时调用真实 API"}
print("缺口检测与补档系统初始化完成")数据类型和精度验证
数值精度问题可能导致累计误差，特别是在高频交易场景下。以下是需要重点检查的数据类型：
- 价格字段：应使用足够的有效数字位数
- 数量字段：应保留原始精度，避免四舍五入
- 时间戳：应为毫秒或微秒级精度
- 字段类型：确保数值字段为数字类型而非字符串
import pyarrow as pa
from decimal import Decimal
class DataTypeValidator:
"""数据类型和精度验证器"""
EXPECTED_TYPES = {
'bid_price': ['float64', 'float32', 'double'],
'ask_price': ['float64', 'float32', 'double'],
'bid_qty': ['float64', 'float32', 'double'],
'ask_qty': ['float64', 'float32', 'double'],
'timestamp': ['int64', 'timestamp[ms]', 'timestamp[us]']
}
def validate_schema(self, df: pd.DataFrame) -> Dict:
"""验证 DataFrame 的数据类型是否符合预期"""
issues = []
for col in df.columns:
col_lower = col.lower()
# 确定预期的数据类型
expected_type = None
for key, types in self.EXPECTED_TYPES.items():
if key in col_lower:
expected_type = types
break
if expected_type:
actual_type = str(df[col].dtype)
if actual_type not in expected_type:
issues.append({
'column': col,
'expected': expected_type,
'actual': actual_type,
'severity': 'error'
})
# 检查是否为对象类型（通常表示数据问题）
if df[col].dtype == 'object':
issues.append({
'column': col,
'expected': 'numeric type',
'actual': 'object',
'severity': 'warning'
})
return {
'is_valid': len(issues) == 0,
'issues': issues
}
def check_precision_loss(self, df: pd.DataFrame, source_schema: pa.Schema) -> Dict:
"""
检查是否存在精度损失
比较源数据模式和目标数据模式
"""
precision_issues = []
for field in source_schema:
col_name = field.name
if col_name not in df.columns:
continue
source_type = field.type
# 对于浮点类型，检查是否会丢失精度
if pa.types.is_float64(source_type):
# 检查是否有超过 float64 表示范围的整数
if df[col_name].dtype == 'float32':
precision_issues.append({
'column': col_name,
'source_type': 'float64',
'target_type': 'float32',
'risk': 'potential_precision_loss'
})
# 对于 Decimal 类型
if pa.types.is_decimal(source_type):
decimal_info = {
'precision': source_type.precision,
'scale': source_type.scale
}
# 检查是否有数值超过精度范围
max_representable = 10 ** (decimal_info['precision'] - decimal_info['scale'])
if df[col_name].abs().max() > max_representable:
precision_issues.append({
'column': col_name,
'decimal_info': decimal_info,
'actual_max': float(df[col_name].abs().max()),
'risk': 'overflow'
})
return {
'has_issues': len(precision_issues) > 0,
'