2. 数据获取与预处理：从源头到可用数据

做量化交易，尤其是高频策略，最头疼的往往不是策略本身，而是数据。我常说一句话：「策略决定上限，数据决定下限」。你想想看，如果拿到的逐笔数据都是脏的、乱的、时间戳对不齐的，再牛的模型也白搭。

这一章，我们就来聊聊怎么把原始数据变成能用的「干净数据」。我会结合我这些年踩过的坑，给你拆解每一步。

2.1 数据来源：交易所与数据商

逐笔数据，也就是 Level-2 行情里的核心内容。它记录了每一笔成交的细节：时间、价格、成交量、买卖方向等等。

获取渠道主要有两个：

• 交易所直连：比如上交所的 L2 行情、深交所的 L2 行情。延迟最低，但门槛高，需要券商资质或机构席位。
• 第三方数据商：比如万得、聚宽、通联数据、RiceQuant 等。方便，但数据质量参差不齐。

我个人习惯用第三方数据商，因为省事。但要注意一点：不同数据商对同一笔成交的解析可能不同。我曾经遇到过，同一个时间点的同一笔交易，A 数据商标记为「主动买」，B 数据商标记为「主动卖」。嗯，这里要留个心眼。

2.2 数据格式解析：CSV vs Parquet

拿到数据后，第一件事就是看格式。最常见的两种：

格式	优点	缺点	我的建议
CSV	通用、可读性强、Excel 能打开	文件大、读写慢、不支持压缩	小数据量（<1GB）可以用
Parquet	列式存储、压缩率高、读写快	需要特定库（如 pyarrow）	大数据量（>1GB）首选

举个例子，我手头有一份某天的逐笔数据，CSV 格式 2.3GB，换成 Parquet 后只有 400MB。读取速度从 30 秒降到了 3 秒。你说香不香？

代码示例：

import pandas as pd

# 读取 CSV
df_csv = pd.read_csv('tick_data.csv')

# 读取 Parquet
df_parquet = pd.read_parquet('tick_data.parquet')

# 保存为 Parquet
df_csv.to_parquet('tick_data.parquet', compression='snappy')

2.3 数据清洗：去重、缺失值、异常值

数据拿到手，别急着分析。先做三件事：去重、补缺失、过滤异常。

2.3.1 去重

数据商有时候会重复推送同一笔成交。怎么判断？看时间戳 + 成交编号 + 价格 + 数量，四个字段完全一致就是重复。

# 去重
df = df.drop_duplicates(subset=['time', 'trade_id', 'price', 'volume'])

我曾经遇到过一个极端案例：某数据商在某个时间段内，把同一笔成交重复推送了 5 次。如果不做去重，你的成交量统计会直接翻 5 倍，策略信号全乱套。

2.3.2 缺失值处理

逐笔数据里，缺失值通常出现在价格或成交量字段。原因可能是数据商采集时丢包了。

处理方法：

• 直接删除：如果缺失比例很小（<0.1%），直接删掉这行。
• 向前填充：用上一笔成交的价格填充。适合价格连续的场景。
• 插值：用前后两笔的平均值。适合成交量。

# 删除缺失值
df = df.dropna(subset=['price', 'volume'])

# 向前填充
df['price'] = df['price'].ffill()

2.3.3 异常值过滤

什么叫异常值？比如价格突然跳到 0 元，或者成交量突然变成 99999999 手。这些明显是数据错误。

我常用的方法：

• 价格范围过滤：比如股票价格不可能低于 0.01 元，也不可能高于 1000 元（除非特殊情况）。
• 成交量过滤：单笔成交量超过该股票日均成交量的 10%，基本就是异常。
• 涨跌幅过滤：单笔成交价格相比上一笔涨跌幅超过 10%，大概率是错误数据。

# 价格过滤
df = df[(df['price'] >= 0.01) & (df['price'] <= 1000)]

# 成交量过滤（假设日均成交量是 100000 手）
df = df[df['volume'] <= 100000 * 0.1]

2.4 时间对齐与标准化

这是最容易被忽视的一步，也是坑最多的一步。

不同数据商的时间戳格式可能不一样：有的用毫秒，有的用微秒，有的直接用字符串。而且，交易所的时钟和数据商的时钟可能存在偏差。

我的做法：

1. 统一时间格式：全部转为 datetime64[ns] 类型。
2. 对齐到毫秒：逐笔数据的时间精度通常是毫秒级，微秒级的数据可以四舍五入到毫秒。
3. 检查时间连续性：如果两笔成交之间间隔超过 1 秒，说明可能有数据缺失。

# 时间标准化
df['time'] = pd.to_datetime(df['time'], unit='ms')

# 对齐到毫秒
df['time'] = df['time'].dt.floor('ms')

# 检查时间间隔
df['time_diff'] = df['time'].diff().dt.total_seconds()
print(df['time_diff'].describe())

2.5 知识体系总览

下面这张图，概括了本章的核心逻辑。你可以把它当作一个检查清单：

这张图把整个流程串起来了。从数据获取到时间对齐，每一步都不能跳过。我见过不少新手，拿到数据直接跑策略，结果回测曲线漂亮得不行，实盘一塌糊涂。为什么？数据没洗干净。

好了，这一章的内容就到这里。数据预处理是基本功，也是最能体现一个量化工程师水平的地方。下一章，我们会深入逐笔数据的核心字段，看看每一列到底代表什么。

---