3、数据解析与清洗：JSON数据解析、数据校验、异常值处理、时间戳对齐

好，咱们进入第三章。说实话，这一章才是真正考验功力的地方。

你想想看，交易所给你的原始数据，就像刚从泥里挖出来的土豆——带着土、带着根、还有虫眼。你不能直接拿去做饭。我见过太多新手，拿到数据就急着画K线，结果图出来全是锯齿，还以为是市场波动大，其实是数据没洗干净。

这一章，咱们就专门干「洗土豆」的活。我会把我在生产环境中踩过的坑，一个一个说给你听。

3.1 JSON数据解析：别被嵌套搞晕

交易所返回的订单簿数据，99%都是JSON格式。但不同交易所的JSON结构，那真是五花八门。

举个例子，币安的深度数据长这样：

{
  "lastUpdateId": 1027024,
  "bids": [
    ["0.0024", "10.0"],
    ["0.0023", "5.0"]
  ],
  "asks": [
    ["0.0025", "12.0"],
    ["0.0026", "8.0"]
  ]
}

看着简单吧？但坑就在那个嵌套数组里。bids和asks都是二维数组，每个元素是[价格, 数量]。我刚开始做的时候，直接用data['bids'][0][0]去取价格，结果某天交易所改了字段顺序，程序直接崩了。

⚠️ 注意： 永远不要硬编码数组索引。用命名元组或数据类来解析。

我个人习惯的做法是，先定义一个数据类：

from dataclasses import dataclass
from typing import List, Tuple

@dataclass
class OrderBookLevel:
    price: float
    quantity: float
    is_bid: bool

@dataclass
class OrderBookSnapshot:
    update_id: int
    bids: List[OrderBookLevel]
    asks: List[OrderBookLevel]

def parse_orderbook(raw: dict) -> OrderBookSnapshot:
    """解析原始JSON为结构化数据"""
    bids = [
        OrderBookLevel(price=float(p), quantity=float(q), is_bid=True)
        for p, q in raw['bids']
    ]
    asks = [
        OrderBookLevel(price=float(p), quantity=float(q), is_bid=False)
        for p, q in raw['asks']
    ]
    return OrderBookSnapshot(
        update_id=raw['lastUpdateId'],
        bids=bids,
        asks=asks
    )

这样写的好处是，如果交易所改了字段名，你只需要改一个地方。我在项目中遇到过，某交易所突然把lastUpdateId改成了updateId，幸好我有这层封装，改一行代码就搞定了。

3.2 数据校验：别信交易所的鬼话

你以为交易所给的数据就一定是对的？太天真了。

我曾经在凌晨三点被报警电话吵醒，说系统显示BTC价格跌到0了。我一看，原来是交易所返回了一个空数组。嗯，你没看错，空数组。

核心原则： 信任，但要验证。对每一笔数据都要做完整性检查。

我建议至少做这三层校验：

  1. 结构校验：检查必要的字段是否存在，类型是否正确
  2. 范围校验：价格不能为负，数量不能为0，时间戳不能是未来
  3. 逻辑校验：买单价格必须低于卖单价格（否则就是套利机会，但通常不会出现在正常数据里）

代码实现大概是这样的：

def validate_orderbook(data: OrderBookSnapshot) -> bool:
    """校验订单簿数据的合法性"""
    # 1. 结构校验
    if not data.bids or not data.asks:
        return False
    
    # 2. 范围校验
    for level in data.bids + data.asks:
        if level.price <= 0 or level.quantity <= 0:
            return False
    
    # 3. 逻辑校验
    best_bid = max(level.price for level in data.bids)
    best_ask = min(level.price for level in data.asks)
    if best_bid >= best_ask:
        return False
    
    return True
💡 小技巧： 校验失败时不要直接丢弃数据，先记录日志。有时候连续校验失败，说明交易所接口出了问题，你需要及时切换备用数据源。

3.3 异常值处理：那些离谱的数据点

数据校验通过，不代表数据就干净了。还有一类问题叫「异常值」。

举个例子，某交易所的深度数据里，突然出现一个买单价格是100万，而当前市价才5万。这明显是交易所的撮合引擎抽风了，或者某个交易员手滑输错了。

你怎么处理？

我的做法是引入「价格合理性窗口」：

def filter_outliers(levels: List[OrderBookLevel], 
                    reference_price: float, 
                    window: float = 0.2) -> List[OrderBookLevel]:
    """
    过滤异常价格
    window: 允许偏离参考价格的比例，0.2表示±20%
    """
    lower = reference_price * (1 - window)
    upper = reference_price * (1 + window)
    return [l for l in levels if lower <= l.price <= upper]

参考价格怎么取？我一般用最近一笔成交价，或者用订单簿的中间价。注意，这个窗口不能设得太小，否则会把正常的深度数据也过滤掉。我个人习惯设20%，在极端行情下可以放宽到50%。

⚠️ 注意： 异常值过滤一定要有日志记录。我曾经因为过滤太狠，导致系统以为市场深度为0，直接触发了风控熔断。后来我加了一条规则：如果过滤后数据量不足原来的10%，就报警人工介入。

3.4 时间戳对齐：让数据说同一种语言

这是最容易被忽视，但也是最容易出问题的地方。

不同交易所的时间戳格式不一样。有的用毫秒，有的用微秒，有的甚至用字符串。更坑的是，有些交易所的服务器时间和你本地时间有偏差。

你想想看，如果你把不同交易所的数据放在一起分析，时间戳没对齐，那得出的结论全是错的。

我的标准化方案是这样的：

from datetime import datetime, timezone
import pandas as pd

def normalize_timestamp(ts, source: str) -> pd.Timestamp:
    """
    将各种格式的时间戳统一为UTC毫秒
    source: 'binance', 'okx', 'bybit' 等
    """
    if isinstance(ts, (int, float)):
        # 判断是秒还是毫秒
        if ts > 1e12:  # 微秒
            ts = ts / 1000
        elif ts > 1e9:  # 纳秒
            ts = ts / 1e6
        # 统一转为毫秒
        return pd.Timestamp(ts, unit='ms', tz='UTC')
    elif isinstance(ts, str):
        # 处理ISO格式字符串
        return pd.Timestamp(ts).tz_convert('UTC')
    else:
        raise ValueError(f"不支持的时间戳格式: {type(ts)}")

这里有个细节：我统一用毫秒。为什么？因为大多数交易所的深度更新频率在100ms左右，毫秒精度够用了。用微秒反而会增加存储和计算开销。

关键点： 时间戳对齐后，一定要做「时间戳连续性检查」。如果相邻两条数据的时间差超过预期（比如超过1秒），说明中间可能有数据丢失。

我曾经在回测时发现策略收益异常高，查了半天，原来是数据有10秒的缺口，系统把缺口前后的数据当成连续的了，算出来的收益率自然离谱。从那以后，我每次加载数据都会跑一遍连续性检查：

def check_timestamp_continuity(timestamps: List[pd.Timestamp], 
                                max_gap: pd.Timedelta = pd.Timedelta(seconds=1)):
    """检查时间戳是否连续"""
    gaps = timestamps[1:] - timestamps[:-1]
    large_gaps = gaps[gaps > max_gap]
    if len(large_gaps) > 0:
        print(f"发现 {len(large_gaps)} 处数据缺口")
        for idx, gap in enumerate(large_gaps[:5]):  # 只打印前5个
            print(f"  缺口 {idx+1}: {gap}")

3.5 本章知识体系

说了这么多，咱们用一张图来总结一下整个数据清洗的流程：

数据解析与清洗流程 原始JSON数据 JSON解析 → 数据类 三层数据校验 异常值过滤 时间戳标准化 清洗后的干净数据 校验失败 → 记录日志 过滤过多 → 报警

这张图把咱们今天讲的内容串起来了。从原始JSON开始，经过解析、校验、异常值处理、时间戳对齐，最后得到干净可用的数据。每一步都有对应的代码实现和避坑指南。

嗯，数据清洗这部分就讲到这里。记住一句话：脏数据进，脏数据出。你花在清洗上的每一分钟，都会在后面的分析和交易中加倍回报给你。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →