2. 数据准备与清洗：高频数据获取、跳脱噪声处理、非同步交易校正

做异质波动率因子，第一步不是写模型，而是搞定数据。

我见过太多人，模型写得花里胡哨，结果数据一塌糊涂，最后归因都找不到方向。说白了，因子研究里，数据清洗占掉 70% 的精力，一点都不夸张。

这一章，我们就来聊聊高频数据的那些坑。我会把我在实盘项目中踩过的雷，一个一个说清楚。

2.1 高频数据获取：从哪里来，要注意什么

高频数据，通常指 tick 级数据，或者 1 分钟、5 分钟的快照数据。国内常见的来源有交易所直连、第三方数据商（如 Wind、聚宽、Tushare），还有自己搭建的行情网关。

我个人习惯，做研究阶段先用第三方数据商，省心。但上生产环境，必须走交易所直连，延迟和完整性不是一个量级。

举个例子，获取某只股票 1 分钟 K 线数据的伪代码：

# 伪代码示例：获取1分钟高频数据
import pandas as pd
from data_source import get_tick_data

# 获取原始tick数据
tick_df = get_tick_data('000001.SZ', '2024-01-01', '2024-01-31')

# 重采样为1分钟K线
minute_df = tick_df.resample('1T', on='timestamp').agg({
    'price': 'ohlc',   # 开盘、最高、最低、收盘
    'volume': 'sum'
})

# 检查缺失时间段
missing = minute_df[minute_df['volume'] == 0]
if len(missing) > 0:
    print(f"发现 {len(missing)} 个缺失分钟线，需处理")

2.2 跳脱噪声处理：那些不该出现的价格尖峰

高频数据里，跳脱噪声（Jump Noise）是最常见的脏数据。比如一笔成交价突然从 10 元跳到 100 元，下一秒又回到 10 元。这通常是因为：

• 交易所撮合系统异常（比如大额错单）
• 数据商传输过程中丢包或重复
• 股票停牌后复牌首笔成交异常

怎么处理？ 我一般用两种方法，结合起来用效果最好。

方法一：基于价格变化率的阈值过滤

设定一个合理的阈值，比如单笔价格变化超过 5% 或 10%，就标记为可疑。但要注意，不同股票的波动率不同，阈值不能一刀切。

def filter_jump_noise(df, threshold=0.05):
    """
    基于价格变化率过滤跳脱噪声
    threshold: 单笔价格变化率阈值，默认5%
    """
    df['price_change'] = df['price'].pct_change()
    # 标记异常点
    df['is_jump'] = df['price_change'].abs() > threshold
    # 用前一个有效价格替换
    df.loc[df['is_jump'], 'price'] = df['price'].shift(1)
    return df

方法二：基于中位数滤波

这个方法更稳健。用当前价格前后各 K 个点的中位数，来替代异常值。我习惯用 K=5，效果不错。

from scipy.signal import medfilt

def median_filter_prices(df, kernel_size=5):
    """
    中位数滤波去噪
    kernel_size: 窗口大小，必须为奇数
    """
    df['price_filtered'] = medfilt(df['price'], kernel_size)
    return df

2.3 非同步交易校正：不同股票的时间对齐问题

这是高频数据里最头疼的问题，没有之一。

为什么？因为 A 股市场里，不同股票的交易活跃度天差地别。贵州茅台可能每秒都有成交，但一只冷门小盘股可能几分钟才有一笔。当你计算两只股票的协方差或相关性时，时间戳对不上，结果就会严重偏误。

非同步交易校正，说白了就是让不同股票的数据站在同一个时间点上说话。

常用的校正方法

1. 线性插值法： 最简单，但会引入未来信息（前向插值）或滞后信息（后向插值）。
2. 前向填充法： 用最近一笔成交价填充空白时间点。这是业界最常用的方法。
3. Refresh Time 方法： 只保留所有股票都有成交的时间点。数据量会大幅减少，但最干净。
4. Hayashi-Yoshida 协方差估计： 直接处理非同步数据，不需要插值。适合做波动率建模。

我个人习惯，做因子研究时用 Refresh Time 方法，因为干净、可解释。做实时交易时用前向填充，因为延迟低。

def refresh_time_alignment(df_dict):
    """
    Refresh Time 方法对齐多只股票的高频数据
    df_dict: 字典，key为股票代码，value为DataFrame（含timestamp和price列）
    """
    # 找出所有股票都有数据的时间点
    common_times = None
    for code, df in df_dict.items():
        times = set(df['timestamp'])
        if common_times is None:
            common_times = times
        else:
            common_times = common_times.intersection(times)
    
    common_times = sorted(list(common_times))
    
    # 提取对齐后的数据
    aligned_data = {}
    for code, df in df_dict.items():
        aligned_data[code] = df[df['timestamp'].isin(common_times)].copy()
    
    return aligned_data, common_times

2.4 本章知识体系总览

下面这张图，把数据准备与清洗的核心流程串起来了。你可以把它当作一个检查清单，每次做数据清洗时对照着走一遍。

2.5 实战建议：我的数据清洗流水线

最后，分享一个我目前在用的数据清洗流水线。你直接拿去用，稍微改改参数就能跑。

1. Step 1： 获取原始 tick 数据，检查时间戳是否连续。如果有缺失，记录日志并决定是否补全。
2. Step 2： 跑一遍跳脱噪声检测。我习惯用动态阈值（5 倍滚动标准差），标记出所有可疑点。
3. Step 3： 人工复核标记点。对于明显异常（比如价格翻倍又回来），直接删除或替换。对于模棱两可的，保留但打上标签。
4. Step 4： 重采样到目标频率（比如 5 分钟）。这一步顺便做了非同步校正——用前向填充。
5. Step 5： 最终检查。计算每只股票的日收益率、波动率，跟公开数据比对。如果偏差超过 1%，回去查原因。

---