数据采集与清洗：获取订单历史数据、数据清洗与预处理、处理缺失值与异常值

做订单流预测，第一步不是建模，而是搞数据。

我见过太多人一上来就调参、跑模型，结果数据里全是坑。最后模型跑出来，回测曲线漂亮得像假的一样——嗯，其实就是假的。数据脏了，后面全白搭。

这一章，咱们就聊聊怎么把原始订单数据变成能用的干净数据。说白了，就是给数据洗个澡，把脏东西、坏数据都剔掉。

2.1 获取订单历史数据

数据源的选择，我踩过不少坑。早期我用过一些免费接口，数据经常断流，或者时间戳对不上。后来我总结了一套自己的数据获取策略。

常见数据来源：

• 交易所API：比如Binance、OKX、Coinbase的REST或WebSocket接口。实时性最好，但历史数据有限制。
• 第三方数据商：像Kaiko、CoinMetrics、QuantConnect。数据质量高，但要花钱。
• 开源数据集：Kaggle、Google BigQuery上有人分享。适合练手，但注意时效性。

我个人习惯用交易所API直接拉取。为什么？因为数据源最原始，没有经过第三方加工，你能控制清洗过程。

下面是我常用的数据拉取代码片段：

import requests
import pandas as pd
from datetime import datetime

def fetch_order_book(symbol='BTCUSDT', limit=1000):
    url = f'https://api.binance.com/api/v3/depth'
    params = {
        'symbol': symbol,
        'limit': limit
    }
    resp = requests.get(url, params=params)
    data = resp.json()
    
    # 解析为DataFrame
    bids = pd.DataFrame(data['bids'], columns=['price', 'volume'])
    asks = pd.DataFrame(data['asks'], columns=['price', 'volume'])
    
    # 转换数据类型
    bids[['price', 'volume']] = bids[['price', 'volume']].astype(float)
    asks[['price', 'volume']] = asks[['price', 'volume']].astype(float)
    
    return bids, asks

# 调用示例
bids_df, asks_df = fetch_order_book()
print(f'获取到 {len(bids_df)} 条买单，{len(asks_df)} 条卖单')

这里要注意：交易所API返回的数据，价格和成交量都是字符串。不转成float，后面计算会出问题。我曾经因为这个bug排查了一整天，最后发现是类型没转换。

2.2 数据清洗与预处理

数据拿到手，别急着用。先看看长什么样。

清洗步骤我一般分四步走：

1. 去重：同一时间戳的重复记录，只保留一条。
2. 排序：按时间戳升序排列，保证时序不乱。
3. 格式统一：时间戳转成datetime，价格和成交量统一为float。
4. 对齐：不同数据源的时间戳要对齐到同一粒度（比如1秒或1毫秒）。

你想想看，如果数据里混着几笔乱序的成交记录，模型会学到什么？它会以为未来的信息可以预测过去——这完全违背了因果律。

下面是一个完整的清洗函数：

def clean_order_data(df):
    # 1. 去重
    df = df.drop_duplicates(subset=['timestamp', 'order_id'])
    
    # 2. 排序
    df = df.sort_values('timestamp').reset_index(drop=True)
    
    # 3. 格式统一
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    df['price'] = df['price'].astype(float)
    df['volume'] = df['volume'].astype(float)
    
    # 4. 检查时间间隔是否均匀
    time_diffs = df['timestamp'].diff().dropna()
    if time_diffs.std() > pd.Timedelta(seconds=1):
        print('警告：时间间隔不均匀，可能存在缺失数据')
    
    return df

# 使用
clean_df = clean_order_data(raw_df)

2.3 处理缺失值与异常值

数据清洗完，接下来就是处理缺失值和异常值。这两个问题在订单流数据里特别常见。

2.3.1 缺失值处理

为什么会缺失数据？原因很多：网络波动、交易所维护、数据商更新延迟。我遇到过最离谱的一次，某数据商因为服务器硬盘满了，直接跳过了3个小时的订单数据。

处理策略：

• 删除：如果缺失比例小于1%，直接删掉那几行。简单粗暴，但有效。
• 插值：用前后时间点的均值填充。适合价格数据，但不适合成交量（成交量波动大）。
• 前向填充：用上一个有效值填充。适合订单簿快照数据，因为订单簿变化是连续的。

def handle_missing_values(df, method='ffill'):
    if method == 'ffill':
        df = df.fillna(method='ffill')
    elif method == 'linear':
        df = df.interpolate(method='linear')
    elif method == 'drop':
        df = df.dropna()
    
    # 检查是否还有缺失
    if df.isnull().sum().sum() > 0:
        print('仍有缺失值，建议检查数据源')
    
    return df

2.3.2 异常值处理

异常值在订单流数据里很常见。比如某笔成交价格突然偏离市场价10%，或者某笔成交量突然是平时的100倍。

怎么识别异常值？

• Z-score方法：计算每个值偏离均值的标准差倍数。一般超过3个标准差就算异常。
• IQR方法：四分位距法。低于Q1-1.5*IQR或高于Q3+1.5*IQR的算异常。
• 业务规则：比如价格不能为负，成交量不能为0（除非是撤单）。

我个人更倾向于用IQR方法。为什么？因为订单流数据往往不是正态分布，Z-score的假设不成立。IQR对偏态分布更鲁棒。

def detect_outliers_iqr(df, column='price'):
    Q1 = df[column].quantile(0.25)
    Q3 = df[column].quantile(0.75)
    IQR = Q3 - Q1
    
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = Q3 + 1.5 * IQR
    
    outliers = df[(df[column] < lower_bound) | (df[column] > upper_bound)]
    
    print(f'检测到 {len(outliers)} 条异常值')
    return outliers

# 标记异常值，而不是直接删除
df['is_outlier'] = False
outlier_idx = detect_outliers_iqr(df).index
df.loc[outlier_idx, 'is_outlier'] = True

2.4 数据清洗流程总览

说了这么多，咱们用一张图把整个流程串起来。这样你脑子里就有个清晰的框架了。

这张图把整个流程串起来了。你照着这个顺序做，基本不会出大问题。

2.5 实战中的几个坑

最后分享几个我实际踩过的坑，希望能帮你省点时间。

• 时间戳时区问题：交易所API返回的时间戳通常是UTC，但有些数据商可能用本地时间。不统一时区，回测结果会偏移。我习惯全部转成UTC+0。
• 成交量单位不一致：有的交易所用张数，有的用币数。比如BTC合约，一张可能是0.01 BTC。不换算直接对比，数据会差两个数量级。
• 订单簿快照的更新频率：有些交易所每100ms推送一次快照，有些是实时推送。如果混用不同频率的数据，模型会学到错误的时序关系。

好了，数据清洗这块就聊到这儿。记住一句话：数据干净了，模型才能好好干活。别急着跑模型，先把数据伺候好。

---