3. 行情数据采集与存储：实时数据流接入、分布式存储方案与数据清洗对齐

高频行情数据的采集与存储是整个A/B测试管线的基石。数据流的实时性、存储的可扩展性以及清洗对齐的精度，直接决定了后续策略回测与效果评估的可靠性。本节将深入剖析从交易所原始行情到可分析数据集的完整技术链路。

3.1 实时数据流接入架构

高频行情数据源通常包括交易所的Level-2快照、逐笔成交、订单簿变更等。接入层需要解决高吞吐、低延迟、断线重连与数据完整性校验四大挑战。

3.1.1 接入层组件与职责

3.1.2 实时数据流接入流程

1. 连接建立： 行情网关通过TCP/UDP多播连接交易所数据源，采用心跳机制维持会话。
2. 协议解析： 将二进制报文（如STEP、Binary、FIX/FAST）解码为结构化行情事件。
3. 序列号校验： 检查消息序列号连续性，发现缺失时触发重传请求或从备份通道补数。
4. 消息序列化： 将行情事件序列化为Avro/Protobuf格式，写入Kafka Topic。
5. 数据持久化： Kafka Broker将消息落盘，并保留指定时间窗口（通常7-30天）。

3.2 分布式存储方案：Kafka 与 InfluxDB

高频行情数据具有写多读少、时序性强、数据量大的特点。传统关系型数据库难以胜任，业界普遍采用分层存储策略：Kafka 负责实时消息缓冲与短期存储，InfluxDB 负责时序数据的长期存储与高效查询。

3.2.1 Kafka：实时消息总线

Kafka 作为分布式消息队列，在高频行情场景中承担数据管道核心角色。其关键配置直接影响A/B测试的数据完整性。

3.2.2 InfluxDB：时序数据存储

InfluxDB 专为时序数据优化，支持高写入吞吐、自动降采样与高效范围查询。在A/B测试场景中，通常将清洗对齐后的行情数据写入InfluxDB，用于策略回测与效果分析。

3.2.3 分层存储策略对比

3.3 数据清洗与对齐

原始行情数据存在缺失、重复、乱序、时间戳不一致等问题。在A/B测试中，这些噪声会直接污染策略效果评估结果。数据清洗与对齐是保证测试结论可信的关键步骤。

3.3.1 常见数据质量问题

3.3.2 清洗流程

1. 去重： 基于交易所消息序列号（SeqNo）或（时间戳+价格+成交量）组合键，保留首条记录。
2. 排序： 按交易所原始时间戳（Exchange Timestamp）排序，而非网关接收时间。
3. 缺失填充： 对于短时间（< 1ms）缺失，使用前一条数据填充；长时间缺失则标记为“数据空洞”。
4. 异常过滤： 使用3σ原则或固定阈值过滤价格/成交量异常值，并记录日志供人工审核。
5. 时间戳对齐： 将不同数据源（快照、逐笔、订单簿）的时间戳统一到同一时钟基准（如NTP同步后的纳秒时间戳）。

3.3.3 数据对齐策略

在A/B测试中，需要将不同频率、不同来源的行情数据对齐到统一的时间网格上，以便进行策略信号计算与效果对比。

• 时间桶对齐： 将连续行情按固定时间窗口（如1ms、10ms）切分，每个桶内取首笔或末笔数据。
• 事件驱动对齐： 以逐笔成交或订单簿变更为触发事件，将快照数据对齐到最近的事件时间戳。
• 插值对齐： 对于缺失的时间点，使用线性插值或前值填充法生成对齐数据。

# 示例：基于Pandas的时间桶对齐
import pandas as pd

def align_to_time_bucket(df, bucket_ms=10):
    """
    将原始行情数据对齐到固定时间桶
    df: DataFrame 包含 'timestamp' 列（纳秒）
    bucket_ms: 时间桶宽度（毫秒）
    """
    df['bucket'] = (df['timestamp'] // (bucket_ms * 1_000_000)) * (bucket_ms * 1_000_000)
    aligned = df.groupby('bucket').agg({
        'price': 'last',      # 取桶内最后一笔价格
        'volume': 'sum',      # 桶内成交量累加
        'bid': 'last',
        'ask': 'last'
    }).reset_index()
    return aligned

3.4 数据质量监控与校验

建立自动化数据质量监控体系，是保障A/B测试数据可靠性的最后一道防线。建议在清洗对齐后，对每个数据批次执行以下校验：