第3章数据格式标准化

3. 数据格式标准化：JSON/CSV/ProtoBuf格式解析、统一时间戳处理（UTC转本地）、价格精度归一化

做量化交易，最头疼的事是什么？

我个人觉得，不是策略亏钱，而是数据打架。

你从交易所A拿到的行情是JSON，从数据商B买的历史数据是CSV，内部系统又用ProtoBuf。时间戳有的用UTC，有的用本地时间。价格精度更乱，有的四位小数，有的八位。不统一？那策略跑起来就是灾难。

这一节，我们就来解决这个「数据格式标准化」的问题。

3.1 三种主流格式的解析实战

先说JSON。这是最通用的格式，但解析时有个坑——浮点数精度丢失。

我在项目中遇到过，某交易所的成交价是12345.67890123，Python的json.loads直接读成12345.67890123，看着没问题。但当你做累加或比较时，浮点误差就出来了。你想想看，几百万笔交易累下来，误差能到几分钱。

我的建议：用Decimal代替float解析价格字段。

import json
from decimal import Decimal

def parse_trade_json(raw: str) -> dict:
    data = json.loads(raw, parse_float=Decimal)
    # 现在 data['price'] 是 Decimal 类型，精度无损
    return data

再说CSV。这东西看着简单，但列名不一致是家常便饭。有的叫timestamp，有的叫time，有的叫date_time。我习惯在解析时做一个列名映射表，统一成内部标准。

import pandas as pd

COLUMN_MAP = {
    'timestamp': 'ts',
    'time': 'ts',
    'date_time': 'ts',
    'price': 'px',
    'last_price': 'px',
    'volume': 'vol',
    'amount': 'vol'
}

def parse_csv(filepath: str) -> pd.DataFrame:
    df = pd.read_csv(filepath)
    df.rename(columns=COLUMN_MAP, inplace=True)
    # 只保留我们关心的列
    return df[['ts', 'px', 'vol']]

最后是ProtoBuf。这东西性能好，但调试麻烦。我建议在开发阶段，先转成JSON再调试，上线后再用二进制格式。

import market_pb2  # 假设这是生成的proto类

def parse_proto(raw_bytes: bytes) -> dict:
    tick = market_pb2.Tick()
    tick.ParseFromString(raw_bytes)
    return {
        'ts': tick.timestamp,
        'px': tick.price,
        'vol': tick.volume
    }

3.2 统一时间戳处理：UTC转本地

时间戳的坑，我踩过不止一次。

交易所给的时间，99%是UTC。但你的策略跑在本地服务器，可能是东八区。直接拿UTC算指标？那开盘时间、收盘时间全对不上。

我曾经因为忘记转时区，导致一个日内策略在下午3点触发了收盘信号——但实际收盘是下午3点30分。那一次回测结果看起来很好，实盘却亏了。嗯，教训深刻。

注意：不要用datetime.replace来转时区！它会直接替换小时数，不考虑夏令时。

正确的做法是用pytz或zoneinfo（Python 3.9+）。

from datetime import datetime, timezone
import pytz

def utc_to_local(utc_ts: datetime, local_tz: str = 'Asia/Shanghai') -> datetime:
    """
    将UTC时间戳转为本地时间
    """
    # 确保是UTC aware
    if utc_ts.tzinfo is None:
        utc_ts = utc_ts.replace(tzinfo=timezone.utc)
    
    local = pytz.timezone(local_tz)
    return utc_ts.astimezone(local)

# 使用示例
utc_time = datetime(2024, 1, 1, 10, 0, 0)  # 假设这是UTC
local_time = utc_to_local(utc_time)
print(local_time)  # 输出: 2024-01-01 18:00:00+08:00

如果你处理的是Unix时间戳（整数或浮点数），那就更简单了：

import pandas as pd

def unix_to_local(ts_seconds: float, tz: str = 'Asia/Shanghai') -> pd.Timestamp:
    return pd.to_datetime(ts_seconds, unit='s', utc=True).tz_convert(tz)

3.3 价格精度归一化

不同交易所的价格精度不一样。比特币有的交易所给到8位小数，有的给到2位。你直接拿过来算，0.00000001和0.00能一样吗？

我建议统一归一化到最小价格变动单位（tick size）。比如某合约的tick size是0.01，那所有价格都四舍五入到0.01。

from decimal import Decimal, ROUND_HALF_UP

def normalize_price(price: Decimal, tick_size: Decimal) -> Decimal:
    """
    将价格归一化到指定的tick size
    """
    quantize_str = str(tick_size)
    return price.quantize(Decimal(quantize_str), rounding=ROUND_HALF_UP)

# 示例
tick = Decimal('0.01')
print(normalize_price(Decimal('123.456'), tick))  # 输出: 123.46
print(normalize_price(Decimal('123.454'), tick))  # 输出: 123.45

小技巧：如果数据量很大，可以用numpy的向量化操作来加速归一化。但注意，numpy不支持Decimal，所以精度要求高的场景还是用Decimal。

3.4 知识体系总览

下面这张图，是我对数据格式标准化的整体理解。你可以把它当作一个检查清单：

说白了，数据标准化就是三个步骤：解析 → 转时区 → 对齐精度。每一步都有坑，但只要你按照上面的方法做，基本能避开90%的问题。

我个人习惯在项目初期就写好标准化的工具函数，放在一个data_utils.py里。这样后续所有数据接入，都走同一套代码。省心，也省得背锅。

  核心原则：数据标准化不是「做完一次就完事」，而是「每次接入新数据源都要做」。把它当作一个持续的过程，而不是一次性任务。

好了，这一节的内容就到这里。记住，数据标准化做得好，策略开发就成功了一半。剩下的，就是让策略去赚钱了。

📚 课程目录

3. 数据格式标准化：JSON/CSV/ProtoBuf格式解析、统一时间戳处理（UTC转本地）、价格精度归一化

3.1 三种主流格式的解析实战

3.2 统一时间戳处理：UTC转本地

3.3 价格精度归一化

3.4 知识体系总览