原始数据解析：从混乱到有序的第一步

做量化交易的朋友都知道，交易所给的数据，从来不会干干净净地送到你手上。JSON、Protobuf、各种时间格式、不同精度……说白了，这就是一场数据清洗的硬仗。我个人习惯把这一步叫做「数据脱壳」——把交易所的原始数据，剥成我们内部能用的标准格式。

JSON 反序列化：最熟悉的陌生人

JSON 解析，听起来简单对吧？但我在项目中遇到过一个问题——交易所返回的订单簿深度数据，有时候一个字段是字符串，有时候是数字。比如价格字段，有时候是 "12345.67"，有时候是 12345.67。你想想看，如果直接拿字符串去计算，程序不会报错，但结果全错。

import json
import pandas as pd

def parse_orderbook_json(raw_json: str) -> dict:
    """
    解析订单簿 JSON 数据
    注意：价格和数量字段统一转为 float
    """
    data = json.loads(raw_json)
    
    # 我个人习惯先做类型校验
    bids = []
    for price_str, qty_str in data.get('bids', []):
        bids.append({
            'price': float(price_str),   # 强制转 float
            'quantity': float(qty_str)
        })
    
    asks = []
    for price_str, qty_str in data.get('asks', []):
        asks.append({
            'price': float(price_str),
            'quantity': float(qty_str)
        })
    
    return {
        'symbol': data['symbol'],
        'timestamp': data['timestamp'],
        'bids': bids,
        'asks': asks
    }

Protobuf 反序列化：性能与复杂度的权衡

为什么有些交易所要用 Protobuf？说白了，就是为了快。高频场景下，JSON 的解析开销太大了。Protobuf 的二进制格式，解析速度能快 3-5 倍。但代价是什么？你得先定义好 .proto 文件，而且调试起来比 JSON 麻烦得多。

# 假设交易所给了我们这样的 .proto 定义
# message OrderBook {
#     string symbol = 1;
#     int64 timestamp = 2;
#     repeated Level bids = 3;
#     repeated Level asks = 4;
# }
# message Level {
#     double price = 1;
#     double quantity = 2;
# }

import orderbook_pb2

def parse_orderbook_protobuf(binary_data: bytes) -> dict:
    """
    解析 Protobuf 格式的订单簿
    注意：时间戳是 int64 微秒
    """
    ob = orderbook_pb2.OrderBook()
    ob.ParseFromString(binary_data)
    
    # 我建议这里直接转成和 JSON 解析一样的 dict 结构
    # 这样下游代码不用关心数据来源
    return {
        'symbol': ob.symbol,
        'timestamp': ob.timestamp,  # 微秒时间戳
        'bids': [{'price': l.price, 'quantity': l.quantity} for l in ob.bids],
        'asks': [{'price': l.price, 'quantity': l.quantity} for l in ob.asks]
    }

时间戳标准化：UTC 是唯一真理

时间戳这个问题，我踩过的坑比代码行数还多。交易所给的时间戳，有的是秒级，有的是毫秒级，有的是微秒级。更坑的是，有些交易所给的是本地时间，有些给的是 UTC。你想想看，如果两个交易所的数据时间基准不一样，你的订单簿合并就是灾难。

我的原则很简单：内部统一用 UTC 毫秒时间戳。所有数据进来，先转成这个标准。

from datetime import datetime, timezone
import pandas as pd

def normalize_timestamp(ts: int, source_format: str) -> int:
    """
    标准化时间戳为 UTC 毫秒
    source_format: 's', 'ms', 'us', 'ns'
    """
    if source_format == 's':
        return ts * 1000
    elif source_format == 'ms':
        return ts
    elif source_format == 'us':
        return ts // 1000
    elif source_format == 'ns':
        return ts // 1_000_000
    else:
        raise ValueError(f"未知的时间戳格式: {source_format}")

# 如果需要显示为可读时间
def timestamp_to_utc_str(ts_ms: int) -> str:
    dt = datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc)
    return dt.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]

价格精度与数量精度：别让浮点数坑了你

做量化的人，最怕的就是浮点数精度问题。0.1 + 0.2 不等于 0.3，这种笑话在订单簿里会要命。交易所通常会给两个参数：tick_size（最小价格变动单位）和 lot_size（最小交易数量单位）。

举个例子：BTC-USDT 的 tick_size 可能是 0.01，意味着价格只能精确到小数点后两位。如果你算出来一个价格是 50000.001，那这个价格在交易所根本不存在。

def round_to_tick(price: float, tick_size: float) -> float:
    """
    将价格对齐到最小变动单位
    """
    return round(price / tick_size) * tick_size

def round_to_lot(quantity: float, lot_size: float) -> float:
    """
    将数量对齐到最小交易单位
    """
    return round(quantity / lot_size) * lot_size

# 实际使用示例
tick_size = 0.01
lot_size = 0.001

raw_price = 50000.005
raw_qty = 1.2345

clean_price = round_to_tick(raw_price, tick_size)  # 50000.01
clean_qty = round_to_lot(raw_qty, lot_size)        # 1.234

print(f"原始价格: {raw_price} -> 对齐后: {clean_price}")
print(f"原始数量: {raw_qty} -> 对齐后: {clean_qty}")

Symbol 映射表：统一你的交易世界

不同交易所对同一个交易对的命名可能完全不同。比如比特币对 USDT，币安叫 BTCUSDT，OKX 叫 BTC-USDT，Coinbase 叫 BTC-USD。如果你同时接入多个交易所，没有映射表的话，代码会变成一团乱麻。

我建议维护一个全局的 Symbol 映射表，把不同交易所的命名统一成内部标准。

# Symbol 映射表
SYMBOL_MAP = {
    'binance': {
        'BTCUSDT': 'BTC/USDT',
        'ETHUSDT': 'ETH/USDT',
        'SOLUSDT': 'SOL/USDT',
    },
    'okx': {
        'BTC-USDT': 'BTC/USDT',
        'ETH-USDT': 'ETH/USDT',
        'SOL-USDT': 'SOL/USDT',
    },
    'coinbase': {
        'BTC-USD': 'BTC/USD',
        'ETH-USD': 'ETH/USD',
    }
}

def normalize_symbol(raw_symbol: str, exchange: str) -> str:
    """
    将交易所的原始 symbol 转为内部标准格式
    """
    if exchange not in SYMBOL_MAP:
        raise ValueError(f"未知交易所: {exchange}")
    
    internal_symbol = SYMBOL_MAP[exchange].get(raw_symbol)
    if internal_symbol is None:
        raise ValueError(f"未知交易对: {raw_symbol} @ {exchange}")
    
    return internal_symbol

# 使用示例
print(normalize_symbol('BTCUSDT', 'binance'))   # BTC/USDT
print(normalize_symbol('BTC-USDT', 'okx'))      # BTC/USDT

实战：完整的解析流水线

好了，我们把上面所有的知识点串起来，写一个完整的解析函数。这个函数接收原始数据，输出标准化的订单簿结构。

def parse_orderbook_pipeline(raw_data: bytes, 
                              data_format: str,
                              exchange: str,
                              source_ts_format: str,
                              tick_size: float,
                              lot_size: float) -> dict:
    """
    完整的订单簿解析流水线
    """
    # 第一步：反序列化
    if data_format == 'json':
        parsed = parse_orderbook_json(raw_data.decode('utf-8'))
    elif data_format == 'protobuf':
        parsed = parse_orderbook_protobuf(raw_data)
    else:
        raise ValueError(f"不支持的数据格式: {data_format}")
    
    # 第二步：标准化 symbol
    internal_symbol = normalize_symbol(parsed['symbol'], exchange)
    
    # 第三步：标准化时间戳
    normalized_ts = normalize_timestamp(parsed['timestamp'], source_ts_format)
    
    # 第四步：精度处理
    clean_bids = []
    for bid in parsed['bids']:
        clean_bids.append({
            'price': round_to_tick(bid['price'], tick_size),
            'quantity': round_to_lot(bid['quantity'], lot_size)
        })
    
    clean_asks = []
    for ask in parsed['asks']:
        clean_asks.append({
            'price': round_to_tick(ask['price'], tick_size),
            'quantity': round_to_lot(ask['quantity'], lot_size)
        })
    
    return {
        'symbol': internal_symbol,
        'exchange': exchange,
        'timestamp': normalized_ts,
        'bids': clean_bids,
        'asks': clean_asks
    }

---

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