📚 课程目录第2章 Tick数据清洗

2. Tick数据清洗：缺失值处理、异常值检测（跳空、零价格）、时间戳对齐与标准化

高频行情数据（Tick数据）在采集、传输与存储过程中，极易受到网络延迟、交易所撮合机制、硬件故障等因素影响，产生缺失值、异常跳空、零价格以及时间戳错位等问题。若未经清洗直接用于策略回测或实盘计算，将导致信号失真、回测过拟合甚至实盘亏损。本节将系统讲解Tick数据清洗的三大核心环节：缺失值处理、异常值检测与时间戳对齐与标准化。

2.1 缺失值处理

缺失值在高频数据中表现为某一时刻的行情字段（如最新价、成交量、买卖盘口）为空或NaN。常见原因包括：交易所数据推送间歇、网络丢包、数据库写入失败等。处理策略需根据缺失模式与业务场景选择。

缺失模式	典型场景	推荐处理方法
单点随机缺失	某一Tick的买一价缺失	前向填充（ffill）或线性插值
连续片段缺失	网络中断导致10秒无数据	标记为无效区间，不填充；或使用同期历史均值
开盘/收盘边界缺失	集合竞价阶段数据不完整	使用交易所官方开盘价填充，或删除该区间
盘口深度缺失	某一档位买卖量缺失	使用相邻档位加权估计，或置零并标记

  实践要点： 对于高频策略，连续缺失超过3个Tick时，建议直接丢弃该区间而非强行填充，以避免引入虚假价格路径。

import pandas as pd
import numpy as np

# 示例：Tick数据前向填充与线性插值
tick_data = pd.DataFrame({
    'time': pd.to_datetime(['09:30:00.000', '09:30:00.100', '09:30:00.200',
                            '09:30:00.300', '09:30:00.400']),
    'price': [100.0, np.nan, 100.05, np.nan, 100.10]
})

# 前向填充
tick_data['price_ffill'] = tick_data['price'].ffill()

# 线性插值
tick_data['price_interp'] = tick_data['price'].interpolate(method='linear')

print(tick_data)

2.2 异常值检测：跳空与零价格

异常值在高频数据中主要表现为两类：价格跳空（价格在极短时间内发生不合理跃变）和零价格（价格或成交量为0）。这些异常值会严重扭曲统计特征与策略信号。

2.2.1 跳空检测

跳空检测的核心逻辑是：当前Tick价格与前一Tick价格的变动幅度超过预设阈值（如固定价差倍数或百分比）。对于不同品种，阈值需动态调整。

固定阈值法： 设定最大绝对变动（如股票0.5元，期货2个最小变动价位）。
统计阈值法： 使用滚动窗口的均值±k倍标准差（k通常取3~5）。
基于波动率的方法： 使用前一分钟已实现波动率作为基准，超过5倍视为跳空。

注意： 跳空检测需区分“真实跳空”（如重大新闻导致的价格突变）与“数据错误”。建议结合成交量与盘口数据辅助判断：若跳空伴随成交量异常放大，可能是真实事件；若成交量极低或为零，则大概率是数据错误。

def detect_price_jump(series, threshold_mult=5):
    """
    基于滚动均值和标准差检测价格跳空
    """
    rolling_mean = series.rolling(window=20, min_periods=5).mean()
    rolling_std = series.rolling(window=20, min_periods=5).std()
    upper_bound = rolling_mean + threshold_mult * rolling_std
    lower_bound = rolling_mean - threshold_mult * rolling_std
    jumps = (series > upper_bound) | (series < lower_bound)
    return jumps

# 示例使用
price_series = tick_data['price'].fillna(method='ffill')
jump_flags = detect_price_jump(price_series)
print("跳空标记：\n", jump_flags)

2.2.2 零价格与零成交量处理

零价格或零成交量通常由交易所初始化、数据解析错误或撮合异常产生。处理原则：

零价格： 直接标记为缺失值，按缺失值策略处理；若连续出现，则删除对应区间。
零成交量： 若价格不变，可视为无成交Tick，保留时间戳但标记；若价格变动但成交量为零，则极大概率是错误数据，需删除。

高效技巧： 使用向量化操作一次性过滤零价格与跳空，避免逐行循环。

# 零价格过滤
tick_data.loc[tick_data['price'] == 0, 'price'] = np.nan
tick_data['price'] = tick_data['price'].ffill()

# 零成交量且价格变动 → 删除
mask = (tick_data['volume'] == 0) & (tick_data['price'].diff().abs() > 0)
tick_data = tick_data[~mask].copy()

2.3 时间戳对齐与标准化

高频数据的时间戳来自不同交易所、不同数据源，其精度、时区、格式差异巨大。时间戳对齐与标准化是后续多源数据融合与策略计算的基础。

2.3.1 时间戳精度统一

常见精度包括：秒级、毫秒级、微秒级、纳秒级。统一规则：

将所有时间戳转换为纳秒级整数（如Unix纳秒时间戳）作为内部存储格式。
展示或计算时再按需转换为可读格式。
对于精度不足的数据（如只有秒级），采用均匀分布假设或交易所官方时间戳进行插值对齐。

2.3.2 时区标准化

跨市场数据（如A股、港股、美股）需统一到同一时区。推荐做法：

内部统一使用UTC时间存储。
在分析或回测时，根据品种转换为交易所本地时间。
注意夏令时切换对美股数据的影响。

2.3.3 对齐到统一时间网格

不同Tick数据到达时间不均匀，需对齐到固定时间网格（如每100ms一个Tick）。常用方法：

对齐方法	描述	适用场景
最近邻对齐	将每个Tick对齐到最近的时间网格点	低频策略、盘口快照
线性插值对齐	在两个Tick之间线性插值得到网格点价格	高频策略、连续价格序列
前向填充对齐	使用网格点之前的最新Tick数据填充	回测模拟、避免未来信息
成交量加权对齐	以成交量为权重计算网格点价格	流动性敏感策略

  关键原则： 时间戳对齐必须避免“未来信息泄露”。在回测中，只能使用网格点之前已发生的数据，不能使用未来Tick信息进行插值。

import pandas as pd

# 示例：将Tick数据对齐到100ms时间网格
tick_data['time_ns'] = tick_data['time'].astype(np.int64)  # 转为纳秒
grid_start = tick_data['time_ns'].min()
grid_end = tick_data['time_ns'].max()
grid_step = 100_000_000  # 100ms = 100,000,000纳秒

time_grid = pd.date_range(start=grid_start, end=grid_end, freq='100ms')
aligned_df = pd.DataFrame(index=time_grid)

# 前向填充对齐（避免未来信息）
aligned_df['price'] = tick_data.set_index('time')['price'].reindex(
    time_grid, method='ffill'
)

print(aligned_df.head(10))

2.4 清洗流程整合与质量验证

将缺失值处理、异常值检测、时间戳对齐整合为一个完整的清洗流水线，并加入质量验证环节，确保输出数据可靠。

推荐实践： 将清洗逻辑封装为可复用的Pipeline类，支持链式调用与参数配置，便于多品种、多周期复用。

class TickCleaner:
    def __init__(self, max_jump_std=5, fill_method='ffill', time_grid='100ms'):
        self.max_jump_std = max_jump_std
        self.fill_method = fill_method
        self.time_grid = time_grid
        self.cleaning_log = []
    
    def clean(self, df):
        # 1. 零价格处理
        df.loc[df['price'] == 0, 'price'] = np.nan
        self.cleaning_log.append(f"零价格标记：{(df['price'].isna()).sum()} 条")
        
        # 2. 缺失值填充
        if self.fill_method == 'ffill':
            df['price'] = df['price'].ffill()
        elif self.fill_method == 'interpolate':
            df['price'] = df['price'].interpolate(method='linear')
        
        # 3. 跳空检测
        jumps = detect_price_jump(df['price'], self.max_jump_std)
        df.loc[jumps, 'price'] = np.nan
        df['price'] = df['price'].ffill()
        self.cleaning_log.append(f"跳空检测并修复：{jumps.sum()} 条")
        
        # 4. 时间戳对齐
        df = df.set_index('time').reindex(
            pd.date_range(df['time'].min(), df['time'].max(), freq=self.time_grid),
            method='ffill'
        ).reset_index()
        
        return df
    
    def report(self):
        return "\n".join(self.cleaning_log)

# 使用示例
cleaner = TickCleaner(max_jump_std=4, fill_method='interpolate', time_grid='50ms')
cleaned_data = cleaner.clean(tick_data)
print(cleaner.report())

质量验证清单： 清洗完成后，务必检查以下指标：
① 数据量变化（清洗前后行数对比）
② 缺失值比例（应降至0或接近0）
③ 价格序列的极值是否在合理范围内
④ 时间戳是否严格单调递增且无重复
⑤ 对齐后的时间网格是否完整覆盖交易时段

通过以上系统化的清洗流程，Tick数据将从原始、杂乱的状态转变为干净、对齐、可用的高质量数据集，为后续的特征工程与策略建模奠定坚实基础。