数据获取与清洗：API接口调用、缺失值处理、异常值检测、时间对齐

做高频交易的人都知道一句话：垃圾进，垃圾出。数据质量不过关，后面再牛的策略也是白搭。今天我就把数据获取与清洗这块的实战经验掰开揉碎了讲给你听。

一、API接口调用：别让数据源卡住你的脖子

我个人习惯用WebSocket做实时行情，RESTful API做历史数据补全。为什么？因为高频场景下，毫秒级的延迟差异可能就是盈亏的分水岭。

先看一个典型的WebSocket订阅代码：

import asyncio
import websockets
import json

async def subscribe_depth(symbols, callback):
    uri = "wss://api.exchange.com/ws"
    async with websockets.connect(uri) as ws:
        # 订阅深度数据
        sub_msg = {
            "op": "subscribe",
            "args": [f"depth:{s}" for s in symbols]
        }
        await ws.send(json.dumps(sub_msg))
        
        # 持续接收数据
        while True:
            try:
                msg = await asyncio.wait_for(ws.recv(), timeout=5)
                data = json.loads(msg)
                await callback(data)
            except asyncio.TimeoutError:
                # 心跳检测
                await ws.ping()
                continue

这里有个坑——心跳超时。我在项目中遇到过，某交易所的WebSocket连接如果30秒没收到数据就会自动断开。你想想看，凌晨三点策略跑得好好的，突然断连了，等早上发现时已经亏了一裤兜。所以一定要加心跳重连机制。

二、缺失值处理：高频数据里的"黑洞"

高频数据里缺失值太常见了。交易所撮合引擎偶尔抽风、网络抖动、甚至行情快照和增量数据没对齐，都会导致数据空洞。

我一般把缺失值分成两类：

类型	特征	处理方法
随机缺失	单个tick丢失，前后数据正常	线性插值或前向填充
连续缺失	连续多个tick丢失	标记为无效区间，或使用波动率模型估算

看一个实际的处理代码：

import pandas as pd
import numpy as np

def handle_missing_ticks(df, method='linear'):
    """
    处理高频tick数据中的缺失值
    
    Parameters:
    df: DataFrame, 包含'timestamp'和'price'列
    method: str, 'linear'或'ffill'
    """
    # 先检查缺失比例
    missing_ratio = df['price'].isnull().sum() / len(df)
    
    if missing_ratio > 0.05:
        print(f"警告：缺失比例达到{missing_ratio:.2%}，建议检查数据源")
        # 标记这些区间
        df['data_quality'] = 'good'
        df.loc[df['price'].isnull(), 'data_quality'] = 'suspect'
    
    if method == 'linear':
        df['price'] = df['price'].interpolate(method='linear')
    elif method == 'ffill':
        df['price'] = df['price'].fillna(method='ffill')
    
    return df

三、异常值检测：揪出那些"离谱"的数据

高频数据里的异常值，说白了就是那些明显不符合市场规律的报价。比如卖一价突然比买一价还低，或者价格瞬间跳了10%又弹回来。

我常用的检测方法有三种：

• Z-score法：适合正态分布的数据，但高频收益率的分布其实有厚尾特征
• IQR法：对极端值更鲁棒，我一般用这个做第一道过滤
• 基于波动率的动态阈值：用过去N个tick的波动率做自适应阈值，这是我最推荐的方法

来看动态阈值的实现：

def detect_anomalies(price_series, window=100, n_std=5):
    """
    基于滚动波动率的异常检测
    
    原理：用过去window个tick的波动率估算当前合理价格范围
    """
    # 计算对数收益率
    log_returns = np.log(price_series / price_series.shift(1))
    
    # 滚动标准差
    rolling_std = log_returns.rolling(window=window).std()
    rolling_mean = log_returns.rolling(window=window).mean()
    
    # 动态阈值
    upper_bound = rolling_mean + n_std * rolling_std
    lower_bound = rolling_mean - n_std * rolling_std
    
    # 标记异常
    anomalies = (log_returns > upper_bound) | (log_returns < lower_bound)
    
    return anomalies

嗯，这里要注意。动态阈值法在波动率突变的时候会失效。比如重大新闻发布后，市场波动率瞬间放大，这时候用过去100个tick算出来的阈值会偏小，导致大量正常数据被误判为异常。

四、时间对齐：让不同数据源"说同一种语言"

做高频交易最头疼的事之一，就是不同交易所、不同数据源的时间戳格式不统一。有的用Unix毫秒，有的用纳秒，还有的用字符串格式的ISO时间。

我总结了一套标准流程：

1. 统一时区：全部转成UTC，别用本地时间
2. 统一精度：高频数据至少保留到毫秒，最好到微秒
3. 对齐方式：用"最近邻"对齐，别用"前向填充"

为什么不用前向填充？举个例子，A交易所的tick在10:00:00.500到达，B交易所的下一个tick在10:00:00.501到达。如果你用前向填充，会把A的报价"复制"到B的时间点上，但实际上这两个报价差了0.5毫秒，在高频策略里这就是天壤之别。

def align_timestamps(df_a, df_b, tolerance='1ms'):
    """
    将两个数据源按时间戳对齐
    
    使用pandas的merge_asof实现最近邻对齐
    """
    # 确保时间戳是datetime类型
    df_a['timestamp'] = pd.to_datetime(df_a['timestamp'], unit='ms')
    df_b['timestamp'] = pd.to_datetime(df_b['timestamp'], unit='ms')
    
    # 排序
    df_a = df_a.sort_values('timestamp')
    df_b = df_b.sort_values('timestamp')
    
    # 最近邻对齐
    aligned = pd.merge_asof(
        df_a, df_b,
        on='timestamp',
        tolerance=pd.Timedelta(tolerance),
        direction='nearest'
    )
    
    return aligned

五、实战中的完整数据管道

说了这么多，我把整个流程串起来。下面这张图展示了我个人在项目中常用的数据清洗管道：

你看，整个流程其实不复杂。但每个环节都有它的门道。我做了这么多年量化，最大的体会就是：数据清洗不是一次性工作，而是需要持续迭代的过程。你今天觉得完美的清洗逻辑，明天可能就被新的数据特征打脸。

---