第四章：数据准备与清洗——行情数据的“炼金术”

说实话，很多做量化交易的朋友，一上来就急着写策略、跑回测。但我个人经验告诉我，数据准备这一步，至少决定了你系统成败的50%。你想想看，如果喂给模型的数据都是脏的、乱的、不对齐的，那出来的结果能靠谱吗？

这一章，我们就来聊聊怎么把Level-1和Level-2的原始行情数据，变成可以直接用于VWAP/TWAP执行的“干净数据”。

4.1 Level-1 vs Level-2：你到底需要什么？

先搞清楚这两个数据源的区别。Level-1，也就是我们常说的“快照数据”，它只提供当前最优的买卖盘口。而Level-2，也就是“深度数据”，它能看到逐笔委托和逐笔成交。

我在搭建第一个执行系统时，就犯过傻——只用了Level-1数据做TWAP切片。结果呢？大单进场时，盘口瞬间被吃掉，我的切片价格完全失真。后来我学乖了，对于VWAP这种需要精确计算成交量的策略，Level-2是必须的。

4.2 获取行情数据：从API到本地存储

获取数据的方式，我一般分两种：实时流和离线快照。实时流用WebSocket，离线快照用REST API。这里我展示一个简单的Level-2数据获取示例：

import websocket
import json
import pandas as pd
from datetime import datetime

# 这是我常用的一个WebSocket连接模板
def on_message(ws, message):
    data = json.loads(message)
    # 只保留我们需要的字段
    clean_data = {
        'timestamp': datetime.now(),
        'bid_price': data['bids'][0][0],
        'bid_volume': data['bids'][0][1],
        'ask_price': data['asks'][0][0],
        'ask_volume': data['asks'][0][1],
        'total_volume': data['volume']
    }
    # 直接写入DataFrame，方便后续处理
    df = pd.DataFrame([clean_data])
    df.to_csv('level2_data.csv', mode='a', header=False)

ws = websocket.WebSocketApp(
    "wss://api.yourbroker.com/level2",
    on_message=on_message
)
ws.run_forever()

4.3 处理缺失值与异常值：别让脏数据毁了你的策略

行情数据里，缺失值和异常值简直是家常便饭。比如网络抖动导致某条数据没收到，或者交易所撮合引擎抽风，给你返回一个离谱的价格。

我处理缺失值的逻辑很简单：

• 时间戳缺失： 用前向填充（ffill），因为行情是连续的，前一秒的数据比凭空猜测靠谱。
• 价格缺失： 如果连续缺失超过3个时间片，我直接丢弃这个区间。为什么？因为VWAP计算需要连续的价格序列，断点太多会扭曲结果。
• 成交量缺失： 用0填充，并标记为“可疑”。在后续计算中，我会降低这个时间片的权重。

异常值处理，我有个“三西格玛法则”的变种：

def detect_outliers(series, threshold=3):
    # 用滚动窗口计算均值和标准差
    rolling_mean = series.rolling(window=20).mean()
    rolling_std = series.rolling(window=20).std()
    
    # 标记异常
    outliers = (abs(series - rolling_mean) > threshold * rolling_std)
    
    # 我曾经遇到过一个极端情况：某只股票因为乌龙指，价格瞬间跳了20%
    # 这种异常如果直接剔除，会导致VWAP计算严重偏差
    # 所以我加了一个逻辑：如果异常值超过前值的50%，直接丢弃该时间片
    extreme_outliers = (abs(series.pct_change()) > 0.5)
    
    return outliers | extreme_outliers

4.4 数据对齐与重采样：让不同频率的数据“说同一种语言”

这是最容易被忽视的一步。Level-1数据可能是3秒一次，Level-2数据是逐笔的，而你的策略可能需要1秒的切片。怎么对齐？

我常用的方法是“时间戳对齐+线性插值”。举个例子：

import pandas as pd

# 假设level1_data是3秒间隔，level2_data是逐笔
level1 = pd.read_csv('level1.csv', index_col='timestamp', parse_dates=True)
level2 = pd.read_csv('level2.csv', index_col='timestamp', parse_dates=True)

# 重采样到1秒
level1_resampled = level1.resample('1S').ffill()  # 前向填充
level2_resampled = level2.resample('1S').ohlc()   # OHLC聚合

# 对齐两个DataFrame
aligned_data = pd.concat([level1_resampled, level2_resampled], axis=1)
aligned_data = aligned_data.dropna()  # 删除任何一方缺失的时间片

这里有个坑：重采样时，一定要考虑交易时间。比如A股有午休，如果你用24小时重采样，会把中午的空白时间也算进去。我一般会先过滤掉非交易时段。

4.5 知识体系总览

下面这张图，是我对本章内容的一个总结。你可以把它当作数据准备的“检查清单”：

嗯，这张图基本概括了从原始数据到可用数据的完整链路。你按照这个流程走，至少能避免80%的数据问题。