3. 数据获取与清洗：用Python搞定高频行情数据

做量化交易的朋友都知道，策略好不好，数据说了算。尤其是高频交易，数据质量直接决定了你的回测是否可信。我见过太多人花大把时间调参数，结果数据本身就有问题——那真是白忙活一场。

这一章，咱们就聊聊怎么用Python获取和清洗高频行情数据。说白了，就是让你拿到干净、可用的tick级数据。

3.1 高频数据长什么样？

先搞清楚我们要处理的对象。高频行情数据，通常包含这些字段：

字段名	含义	示例值
timestamp	时间戳（精确到毫秒）	2024-01-15 09:30:00.123
symbol	合约代码	rb2405
last_price	最新成交价	3985.0
volume	成交量（本笔）	10
bid_price	买一价	3984.0
ask_price	卖一价	3986.0
bid_volume	买一量	200
ask_volume	卖一量	150

嗯，这里要注意：不同交易所的数据格式略有差异。比如中金所会多一个「当日累计成交量」，而上期所则没有。我个人习惯是统一做一次字段映射，后面处理起来省心很多。

3.2 数据获取的两种主流方式

高频数据的来源，无非两种：

• API实时订阅：通过CTP、XTP等接口直接接收行情流
• 历史数据文件：从数据商（如万得、聚宽、米筐）下载CSV或Parquet文件

咱们课程主要讲第二种——因为做策略研究，历史数据才是主力。实时数据更多是交易执行环节的事。

3.3 用Python读取高频CSV

假设你手头有一个CSV文件，里面是螺纹钢期货的tick数据。怎么读？

import pandas as pd

# 读取原始数据
df = pd.read_csv(
    'rb2405_tick_202401.csv',
    parse_dates=['timestamp'],
    dtype={
        'symbol': 'str',
        'last_price': 'float64',
        'volume': 'int32',
        'bid_price': 'float64',
        'ask_price': 'float64',
        'bid_volume': 'int32',
        'ask_volume': 'int32'
    }
)

# 看一眼数据长啥样
print(df.head())
print(df.info())

这里我特别强调一下dtype参数。如果不指定，pandas会默认用float64存所有数值列。但你想，成交量用int32就够了，能省一半内存。高频数据动辄几百万行，内存优化不是小事。

3.4 数据清洗：高频数据的「坑」

数据读进来了，但别急着用。高频数据里常见的坑，我一个个说。

3.4.1 时间戳对齐

不同交易所的时间精度不一样。上期所精确到秒，中金所精确到毫秒。如果你把两者混在一起分析，时间轴就对不齐。

# 统一到毫秒精度
df['timestamp'] = df['timestamp'].dt.floor('ms')

# 检查是否有重复时间戳
duplicated = df.duplicated(subset=['timestamp', 'symbol'])
print(f'重复记录数: {duplicated.sum()}')

3.4.2 异常价格过滤

行情数据偶尔会出现「打错价」——比如突然蹦出一个0或者一个天文数字。这种数据不处理，你的流动性指标会直接崩掉。

# 过滤明显异常的价格
def filter_price_outliers(df, col='last_price', n_std=5):
    mean = df[col].mean()
    std = df[col].std()
    lower = mean - n_std * std
    upper = mean + n_std * std
    return df[(df[col] >= lower) & (df[col] <= upper)]

df_clean = filter_price_outliers(df)
print(f'过滤前: {len(df)}, 过滤后: {len(df_clean)}')

3.4.3 盘口数据合理性检查

买一价必须小于卖一价，这是基本常识。但数据里偶尔会出现买一价高于卖一价的情况——通常是数据采集时的时序错乱导致的。

# 检查盘口合理性
invalid = df_clean[df_clean['bid_price'] >= df_clean['ask_price']]
print(f'盘口异常记录数: {len(invalid)}')

# 剔除异常
df_clean = df_clean[df_clean['bid_price'] < df_clean['ask_price']]

3.5 数据清洗的完整流程

上面几步是基础。在实际项目中，我一般会封装成一个清洗函数，一步到位：

def clean_tick_data(df):
    """高频tick数据清洗流水线"""
    # 1. 时间戳处理
    df['timestamp'] = pd.to_datetime(df['timestamp']).dt.floor('ms')
    
    # 2. 去重
    df = df.drop_duplicates(subset=['timestamp', 'symbol'])
    
    # 3. 价格异常过滤
    df = filter_price_outliers(df, 'last_price', 5)
    
    # 4. 盘口合理性
    df = df[df['bid_price'] < df['ask_price']]
    
    # 5. 成交量非负
    df = df[df['volume'] >= 0]
    
    # 6. 按时间排序
    df = df.sort_values('timestamp').reset_index(drop=True)
    
    return df

# 执行清洗
df_clean = clean_tick_data(df)

3.6 本章知识体系

下面这张图，帮你理清数据获取与清洗的完整脉络：

3.7 数据质量检查清单

清洗完了，怎么知道数据到底干不干净？我一般会跑一遍检查清单：

1. 时间连续性：相邻tick的时间差是否合理？有没有大段缺失？
2. 价格范围：是否在当日涨跌停板内？
3. 买卖价差：价差是否为正？有没有异常大的价差？
4. 成交量：是否有负值？是否有单笔成交量异常大？
5. 数据完整性：交易时段内是否有超过1秒没有数据？

好了，数据拿到手，也洗干净了。下一件事就是怎么从这些tick数据里提取流动性指标——比如买卖价差、订单簿深度、成交活跃度等等。这些内容，咱们下一章再聊。

---

公众号:蓝海数据掘金营，微信deep3321