3. 行情数据特征:来源、频率与质量挑战
在坏账率哨兵回放验证中,行情数据是驱动回放引擎的“燃料”。不同来源、不同频率的数据,其质量特征与潜在风险截然不同。本章系统梳理行情数据的三大核心维度,帮助你在回放验证前建立数据质量基线。
3.1 行情数据来源:交易所 vs OTC
行情数据按交易场所可分为两大类:交易所数据与场外交易(OTC)数据。两者在数据格式、更新机制、可信度上存在显著差异。
| 对比维度 | 交易所数据 | OTC 数据 |
|---|---|---|
| 数据格式 | 标准化(如 FIX/FAST 协议) | 非标准化(邮件、聊天记录、自定义 API) |
| 更新机制 | 实时撮合推送,毫秒级 | 双边报价,分钟级甚至延迟 |
| 可信度 | 高(受监管、可审计) | 中低(依赖对手方信用) |
| 回放适用性 | 直接用于 Tick 级回放 | 需清洗、对齐时间戳后使用 |
| 常见坏账风险 | 数据缺失、延迟、错单 | 报价虚假、重复、时间戳漂移 |
实战提示:在哨兵回放验证中,优先使用交易所行情数据作为“基准线”。若必须引入 OTC 数据,建议先做 时间戳对齐校验 和 报价合理性过滤,否则坏账率回放结果可能被数据噪声主导。
3.2 数据频率:Tick、分钟、日线
行情数据频率决定了回放的时间粒度与计算精度。不同频率适用于不同的坏账检测场景。
- Tick 级数据:逐笔成交或逐笔报价,精度最高,适合高频交易坏账检测(如闪电崩盘、流动性瞬时枯竭)。数据量巨大,单日可达数百万条。
- 分钟级数据:聚合后的 K 线(1分钟、5分钟、15分钟等),平衡精度与存储,适合日内趋势型坏账模式识别。
- 日线数据:每日收盘价、开盘价、最高最低价,适合中长期坏账率趋势分析,但无法捕捉日内突变。
频率选择原则:哨兵回放验证的坏账检测窗口应至少比数据频率高一个数量级。例如,若检测 1 秒内的流动性断裂,必须使用 Tick 级数据;若检测日终结算坏账,日线数据即可。
3.3 数据质量挑战与应对
行情数据在实际生产环境中面临多重质量挑战,直接回放会导致坏账率误判。以下是三大核心挑战及应对策略。
- 数据缺失:因网络中断、交易所故障、数据商延迟导致的时间段空白。应对:使用前向填充、插值或标记缺失区间,在回放日志中记录缺失率。
- 异常值:价格或成交量超出合理范围(如价格跳空 100 倍、成交量突增 1000 倍)。应对:设置阈值过滤,结合波动率带(如 Bollinger Bands)识别异常。
- 时间戳漂移:数据时间戳与标准时间不一致,导致回放顺序错乱。应对:使用 NTP 同步后的时间戳,对 OTC 数据做时间戳对齐与去重。
3.4 数据质量评估指标
在回放验证前,建议对行情数据做量化评估。以下指标可作为数据质量基线检查清单。
| 指标 | 定义 | 合格阈值 | 检测方法 |
|---|---|---|---|
| 缺失率 | 缺失时间戳占总时间戳比例 | < 0.1% | 时间序列连续性扫描 |
| 异常值率 | 超出 5 倍标准差的数据点比例 | < 0.01% | Z-score 或 IQR 检测 |
| 时间戳抖动 | 相邻时间戳间隔的标准差 | < 1ms(Tick 级) | 差分序列统计 |
| 重复率 | 完全相同的时间戳-价格-成交量记录比例 | 0% | 哈希去重 |
注意:若行情数据缺失率超过 1% 或异常值率超过 0.1%,建议先进行数据修复,否则回放验证的坏账率结果将不可信。在哨兵回放日志中,必须记录数据质量评估结果,作为坏账率置信度的辅助信息。
3.5 实战:行情数据质量检查代码片段
以下 Python 代码演示如何对 Tick 级行情数据做快速质量检查,输出缺失率与异常值率。
import pandas as pd
import numpy as np
def check_tick_quality(df, price_col='price', time_col='timestamp'):
"""
对 Tick 级行情数据进行质量检查
:param df: DataFrame,包含时间戳和价格列
:param price_col: 价格列名
:param time_col: 时间戳列名
:return: dict,包含缺失率、异常值率、重复率
"""
# 1. 缺失率:检查时间戳是否连续(假设 Tick 频率为 1ms)
expected_count = (df[time_col].max() - df[time_col].min()) / 1_000_000 # 微秒转秒
actual_count = len(df)
missing_rate = 1 - (actual_count / expected_count) if expected_count > 0 else 0
# 2. 异常值率:基于 Z-score
z_scores = np.abs((df[price_col] - df[price_col].mean()) / df[price_col].std())
outlier_rate = (z_scores > 5).mean()
# 3. 重复率:基于时间戳+价格+成交量联合去重
dup_rate = df.duplicated(subset=[time_col, price_col, 'volume']).mean()
return {
'missing_rate': missing_rate,
'outlier_rate': outlier_rate,
'duplicate_rate': dup_rate
}
# 使用示例
# quality = check_tick_quality(tick_df)
# print(f"缺失率: {quality['missing_rate']:.4%}")
# print(f"异常值率: {quality['outlier_rate']:.4%}")
# print(f"重复率: {quality['duplicate_rate']:.4%}")
扩展建议:在实际哨兵回放系统中,可将上述质量检查函数嵌入数据加载管道,在回放开始前自动输出数据质量报告。若质量指标未通过阈值,系统应发出警告并暂停回放,避免基于脏数据产生误判。