第3章 Tick数据质量评估

3、Tick数据质量评估：数据完整性检查、缺失值识别与统计、异常值检测（价格跳空、成交量突变）、数据延迟与时间戳对齐问题

好，咱们进入第三章。说实话，Tick数据质量评估这块，是我在实战中踩坑最多的地方。很多新手拿到数据就开始算因子、跑策略，结果模型一上线就崩。为什么？因为数据本身就有问题。你想想看，如果原材料是坏的，再好的厨子也做不出好菜。

我个人习惯，拿到任何Tick数据，第一件事不是分析，而是做质量评估。这就像医生看病，先得做个体检。今天我就把这份「体检清单」掰开揉碎了讲给你听。

核心观点：数据质量评估不是可选项，而是必选项。跳过这一步，后面的所有工作都是在沙滩上建城堡。

3.1 数据完整性检查

完整性检查，说白了就是看看数据有没有「缺胳膊少腿」。我遇到过最离谱的一次，某交易所的Tick数据居然少了整整一个交易日的记录，要不是做了完整性检查，我那个策略就全废了。

完整性检查主要看三件事：

时间范围完整性：数据覆盖的时间段是否符合预期？有没有整段缺失？
交易日完整性：每个交易日的数据是否都存在？节假日是否被正确处理？
字段完整性：每条记录是否都包含必要的字段？比如时间戳、价格、成交量等。

我一般用这样的代码来快速检查：

import pandas as pd
import numpy as np

def check_data_completeness(df, date_col='datetime', freq='1min'):
    """
    检查Tick数据完整性
    """
    # 检查时间范围
    print(f"数据时间范围: {df[date_col].min()} 到 {df[date_col].max()}")
    
    # 检查交易日完整性
    df['date'] = df[date_col].dt.date
    trading_days = df['date'].nunique()
    print(f"交易日数量: {trading_days}")
    
    # 检查每个交易日的记录数
    daily_counts = df.groupby('date').size()
    print(f"每日记录数统计:\n{daily_counts.describe()}")
    
    # 检查是否有整段缺失
    time_diff = df[date_col].diff().dropna()
    expected_diff = pd.Timedelta(freq)
    anomalies = time_diff[time_diff > expected_diff * 2]
    
    if len(anomalies) > 0:
        print(f"发现 {len(anomalies)} 处时间间隔异常")
        return anomalies
    else:
        print("时间间隔正常")
        return None

小技巧：对于高频Tick数据，我建议用1分钟或5分钟作为基准频率来检查。太细了容易误报，太粗了又可能漏掉问题。

3.2 缺失值识别与统计

缺失值处理，是数据清洗的必修课。Tick数据里的缺失值，往往不是简单的NaN，而是「隐式缺失」——比如某几秒的数据完全没出现。

我总结了几种常见的缺失模式：

显式缺失：字段值为NaN或None
隐式缺失：时间戳不连续，中间有跳过的时段
部分缺失：某些字段有值，某些字段缺失

举个例子，我曾经处理过一份期货Tick数据，发现每天下午14:30-14:35之间总有5分钟的数据缺失。一开始以为是系统bug，后来才发现是交易所的例行维护时间。嗯，这种「假缺失」需要特别注意。

def identify_missing_values(df, time_col='timestamp', freq='1S'):
    """
    识别缺失值，包括隐式缺失
    """
    # 显式缺失统计
    missing_counts = df.isnull().sum()
    print("显式缺失统计:")
    print(missing_counts[missing_counts > 0])
    
    # 隐式缺失检测
    df = df.sort_values(time_col)
    full_range = pd.date_range(
        start=df[time_col].min(),
        end=df[time_col].max(),
        freq=freq
    )
    
    actual_times = set(df[time_col])
    missing_times = [t for t in full_range if t not in actual_times]
    
    print(f"\n隐式缺失数量: {len(missing_times)}")
    
    # 找出连续缺失超过5秒的时段
    if missing_times:
        missing_series = pd.Series(missing_times)
        gaps = missing_series.diff() > pd.Timedelta(seconds=1)
        gap_starts = missing_series[gaps]
        gap_ends = missing_series.shift(-1)[gaps]
        
        for start, end in zip(gap_starts, gap_ends):
            if (end - start) > pd.Timedelta(seconds=5):
                print(f"长时间缺失: {start} 到 {end}")
    
    return missing_times

警告：千万不要直接删除所有缺失值！我曾经这么干过，结果把一个重要行情波动给删掉了，策略回测结果完全失真。缺失值处理要结合业务逻辑来判断。

3.3 异常值检测

异常值检测，是Tick数据清洗中最刺激的部分。为什么？因为Tick数据里的异常，往往意味着交易机会或者数据错误。

3.3.1 价格跳空检测

价格跳空，就是相邻两个Tick之间的价格变化超过了正常范围。我见过最夸张的一次，某股票在1秒内从10元跳到100元，后来发现是数据录入错误。

检测方法其实不复杂：

def detect_price_gaps(df, price_col='price', threshold=0.05):
    """
    检测价格跳空
    threshold: 跳空阈值，比如0.05表示5%
    """
    df = df.sort_values('timestamp')
    price_changes = df[price_col].pct_change()
    
    gaps = df[abs(price_changes) > threshold].copy()
    gaps['change_pct'] = price_changes[abs(price_changes) > threshold]
    
    print(f"发现 {len(gaps)} 处价格跳空")
    if len(gaps) > 0:
        print("跳空详情:")
        print(gaps[['timestamp', price_col, 'change_pct']].head(10))
    
    return gaps

避坑指南：我曾经把阈值设得太小，结果把正常的高频波动也当成了异常。后来我改用「滚动窗口+标准差」的方法，效果好了很多。具体来说，就是用过去N个Tick的价格变化标准差作为动态阈值。

3.3.2 成交量突变检测

成交量突变，往往比价格跳空更隐蔽。为什么？因为成交量没有「正常范围」这个概念。一只股票平时每分钟成交100手，突然变成10000手，这可能是大单进场，也可能是数据错误。

我常用的方法是：

def detect_volume_spikes(df, volume_col='volume', window=20, zscore_threshold=3):
    """
    检测成交量突变
    使用Z-score方法
    """
    df = df.sort_values('timestamp')
    
    # 计算滚动均值和标准差
    rolling_mean = df[volume_col].rolling(window=window).mean()
    rolling_std = df[volume_col].rolling(window=window).std()
    
    # 计算Z-score
    z_scores = (df[volume_col] - rolling_mean) / rolling_std
    
    # 标记异常
    spikes = df[abs(z_scores) > zscore_threshold].copy()
    spikes['z_score'] = z_scores[abs(z_scores) > zscore_threshold]
    
    print(f"发现 {len(spikes)} 处成交量异常")
    
    return spikes

说实话，成交量突变检测最难的地方在于区分「真异常」和「假异常」。真异常是数据错误，假异常是真实的交易行为。我一般会结合价格变化一起看——如果成交量暴增但价格没怎么动，那很可能是数据问题。

3.4 数据延迟与时间戳对齐

数据延迟，是Tick数据清洗中最头疼的问题。你想想看，不同交易所、不同数据源的时间戳精度不一样，有的精确到毫秒，有的精确到微秒，甚至还有精确到纳秒的。怎么对齐？

我遇到过最坑的一次，某数据源的时间戳居然用的是本地时间，而不是交易所时间。结果夏令时切换那天，所有数据都偏移了1小时。嗯，从那以后我养成了一个习惯：拿到数据先检查时间戳的时区。

时间戳对齐的核心步骤：

统一时区：全部转换为UTC或交易所本地时间
统一精度：全部截断到相同的精度（比如毫秒）
处理延迟：识别并标记延迟超过阈值的数据

def align_timestamps(df, time_col='timestamp', target_freq='1S', timezone='UTC'):
    """
    时间戳对齐处理
    """
    # 统一时区
    if df[time_col].dt.tz is None:
        df[time_col] = df[time_col].dt.tz_localize(timezone)
    else:
        df[time_col] = df[time_col].dt.tz_convert(timezone)
    
    # 统一精度（截断到秒）
    df['aligned_time'] = df[time_col].dt.floor(target_freq)
    
    # 检测延迟
    df['delay'] = (df[time_col] - df['aligned_time']).dt.total_seconds()
    delayed = df[df['delay'] > 0.1]  # 延迟超过100毫秒
    
    print(f"延迟数据数量: {len(delayed)}")
    if len(delayed) > 0:
        print(f"最大延迟: {delayed['delay'].max():.3f}秒")
    
    return df

注意：时间戳对齐不是简单的「四舍五入」。比如你按1秒对齐，那么0.999秒和1.001秒的数据会被分到不同的桶里。我建议用「最近邻」或「线性插值」的方法来处理边界情况。

3.5 知识体系总览

说了这么多，我画了一张图帮你理清思路。这张图涵盖了Tick数据质量评估的完整流程：

这张图把今天讲的内容串起来了。你从中心开始，沿着四个方向走，就能完成一次完整的Tick数据质量评估。我个人建议，每次拿到新数据，都按这个流程走一遍，花不了多少时间，但能避免很多坑。

好了，这一章的内容就到这里。数据质量评估是Tick数据处理的基石，打好这个基础，后面的工作才能顺利进行。记住我常说的那句话：数据质量决定策略上限，清洗质量决定策略下限。

📚 课程目录

3、Tick数据质量评估：数据完整性检查、缺失值识别与统计、异常值检测（价格跳空、成交量突变）、数据延迟与时间戳对齐问题

3.1 数据完整性检查

3.2 缺失值识别与统计

3.3 异常值检测

3.3.1 价格跳空检测

3.3.2 成交量突变检测

3.4 数据延迟与时间戳对齐

3.5 知识体系总览