3. 数据完整性校验：时间戳连续性检查、缺失值检测、重复数据去重、数据边界校验

行情数据这东西，说白了就是钱。一秒的缺失，可能让你错过几百万的行情。我做了这么多年量化，见过太多因为数据校验没做好，回测曲线漂亮得像假的一样，实盘一跑就崩的案例。

今天咱们就聊聊数据完整性校验。嗯，这活儿不性感，但绝对保命。

3.1 时间戳连续性检查

时间戳是行情数据的灵魂。没有连续的时间戳，你的策略就像在断桥上开车——随时可能翻。

为什么要检查连续性？

你想想看，如果某根K线的时间戳突然跳了10秒，或者从9:30直接跳到9:35，中间那5分钟去哪了？我个人习惯，拿到数据第一件事就是检查时间戳的间隔是否一致。

怎么检查？

我一般用两种方法：

1. 差值法：计算相邻时间戳的差值，看是否有异常值。
2. 期望值法：根据起始时间和频率，生成一个完整的时间序列，然后对比实际数据。

import pandas as pd
import numpy as np

def check_timestamp_continuity(df, freq='1min'):
    """
    检查时间戳连续性
    df: 包含'timestamp'列的DataFrame
    freq: 期望的频率，如'1min', '1s'
    """
    # 计算相邻时间戳差值
    df = df.sort_values('timestamp')
    df['time_diff'] = df['timestamp'].diff()
    
    # 期望的间隔（以秒为单位）
    expected_diff = pd.Timedelta(freq).total_seconds()
    
    # 找出异常点
    anomalies = df[df['time_diff'] != pd.Timedelta(seconds=expected_diff)]
    
    if len(anomalies) > 0:
        print(f"发现 {len(anomalies)} 个时间戳异常点")
        print(anomalies[['timestamp', 'time_diff']].head())
    else:
        print("时间戳连续性检查通过")
    
    return anomalies

3.2 缺失值检测

缺失值，说白了就是数据里的「坑」。你回测时可能觉得没什么，但实盘时策略遇到缺失值，轻则报错，重则直接爆仓。

常见的缺失值类型：

类型	表现	常见原因
NaN	数值字段为空	数据源传输中断、解析错误
零值	价格或成交量为0	交易所停牌、数据填充错误
异常值	价格突然跳变	数据合并错误、除权除息未处理

检测方法：

我个人习惯用两步走：

1. 统计法：直接统计每列的缺失值数量。
2. 逻辑法：检查价格、成交量等字段是否在合理范围内。

def detect_missing_values(df):
    """
    检测缺失值
    """
    # 统计NaN
    nan_counts = df.isnull().sum()
    print("NaN统计：")
    print(nan_counts[nan_counts > 0])
    
    # 检查零值（价格字段不应该为0）
    price_cols = ['open', 'high', 'low', 'close']
    for col in price_cols:
        if col in df.columns:
            zero_count = (df[col] == 0).sum()
            if zero_count > 0:
                print(f"警告：{col} 字段有 {zero_count} 个零值")
    
    # 检查负值
    for col in price_cols:
        if col in df.columns:
            neg_count = (df[col] < 0).sum()
            if neg_count > 0:
                print(f"错误：{col} 字段有 {neg_count} 个负值")

3.3 重复数据去重

重复数据，看起来无害，实际上能毁掉你的策略。你想想看，如果同一根K线出现了两次，你的策略可能会在同一时间点开两次仓，或者计算出的指标完全失真。

重复数据的来源：

• 数据源重复推送
• 数据合并时未去重
• 数据库写入时出现重复

去重策略：

我一般按优先级处理：

1. 完全重复：所有字段都一样，直接删除一条。
2. 时间戳重复但数据不同：保留最后一条（最新数据），或者取平均值。
3. 部分字段重复：根据业务逻辑判断保留哪条。

def deduplicate_data(df, keep='last'):
    """
    去重处理
    keep: 'first'保留第一条，'last'保留最后一条
    """
    # 基于时间戳去重
    df_dedup = df.drop_duplicates(subset=['timestamp'], keep=keep)
    
    # 检查去重效果
    dup_count = len(df) - len(df_dedup)
    if dup_count > 0:
        print(f"去除了 {dup_count} 条重复数据")
    
    return df_dedup

3.4 数据边界校验

数据边界校验，说白了就是检查数据是否在合理范围内。比如某只股票的价格突然从10块跳到1000块，这明显不合理。

校验维度：

维度	检查内容	示例
价格范围	是否在合理区间内	A股价格一般在0.01-1000之间
涨跌幅	单日涨跌幅是否超限	A股主板±10%，科创板±20%
成交量	是否为正数	成交量不能为负
OHLC关系	高>低，开>=低，收>=低	最高价必须大于等于最低价

def validate_boundaries(df, symbol=''):
    """
    数据边界校验
    """
    errors = []
    
    # 1. OHLC关系校验
    if not (df['high'] >= df['low']).all():
        errors.append("存在 high < low 的数据")
    if not (df['high'] >= df['close']).all():
        errors.append("存在 high < close 的数据")
    if not (df['high'] >= df['open']).all():
        errors.append("存在 high < open 的数据")
    
    # 2. 涨跌幅校验（A股主板±10%）
    df['pct_change'] = df['close'].pct_change()
    extreme_changes = df[abs(df['pct_change']) > 0.1]
    if len(extreme_changes) > 0:
        errors.append(f"发现 {len(extreme_changes)} 个涨跌幅超过10%的数据点")
    
    # 3. 成交量校验
    if (df['volume'] < 0).any():
        errors.append("存在负成交量")
    
    if errors:
        print(f"边界校验发现 {len(errors)} 个问题：")
        for err in errors:
            print(f"  - {err}")
    else:
        print("边界校验通过")
    
    return errors

3.5 整体校验流程

说了这么多，咱们来画个流程图，看看整个校验流程怎么走。

这个流程看起来简单，但实际做起来，每一步都可能遇到坑。我个人习惯，每个步骤都加上日志记录，方便事后追溯。