第4章高频数据基础

4、高频数据基础：Tick级数据获取、数据清洗与对齐、时间戳处理与时钟同步

各位同学，欢迎来到第四章。

前面我们聊了碳市场的微观结构，讲了订单簿、流动性、价差这些概念。但说实话，那些都是「静态」的切片。真正做高频策略，你得跟数据「过日子」——每天面对的就是Tick级数据流。

这一章，我带你手把手搞定高频数据的三个核心环节：获取、清洗、对齐。尤其是时间戳和时钟同步，这里面的坑，我当年踩得那叫一个惨。

4.1 Tick级数据获取：从交易所到你的硬盘

先说说数据源。碳交易所的数据接口，跟股票期货不太一样。国内几个主要的碳交易所，比如上海环境能源交易所、湖北碳排放权交易中心，它们的数据推送方式各有各的脾气。

常见的获取方式有三种：

API直连：最理想的方式。交易所提供WebSocket或FIX协议接口，实时推送逐笔成交和订单簿快照。延迟最低，但门槛高，通常需要机构席位。
数据商购买：像万得、聚宽、通联数据这些，会提供整理好的Tick数据。省心，但贵，而且数据可能有几毫秒的延迟。
爬虫抓取：针对一些公开的行情页面。嗯，我建议你只用来做研究，实盘千万别用。延迟不稳定，还容易被封IP。

我个人习惯：实盘策略必须用API直连。回测研究可以用数据商的数据，但一定要确认数据的时间戳精度。我曾经遇到过数据商把毫秒级时间戳截断到秒，结果回测曲线漂亮得不行，实盘一跑就崩。

下面是一个简单的WebSocket订阅示例，用Python写的。注意，这只是伪代码，具体协议你得跟交易所确认。

import websocket
import json

def on_message(ws, message):
    tick = json.loads(message)
    # tick结构：{'symbol': 'CEA', 'price': 58.2, 'volume': 100, 'timestamp': 1700000000123}
    process_tick(tick)

def on_error(ws, error):
    print(f"连接异常: {error}")

def on_close(ws, close_status_code, close_msg):
    print("连接关闭，准备重连...")

def on_open(ws):
    # 订阅碳配额合约
    subscribe_msg = {"op": "subscribe", "args": ["CEA.202312"]}
    ws.send(json.dumps(subscribe_msg))

ws = websocket.WebSocketApp("wss://api.carbon-exchange.com/ws",
                            on_open=on_open,
                            on_message=on_message,
                            on_error=on_error,
                            on_close=on_close)
ws.run_forever()

避坑指南：WebSocket连接一定要加心跳和自动重连机制。我见过太多因为网络抖动导致数据断流，策略还在傻傻下单的情况。重连后还要做数据补全，否则时间序列会有空洞。

4.2 数据清洗：脏数据是策略的毒药

拿到原始Tick数据后，别急着用。先洗一洗。

碳市场的数据有多脏？我举个例子：某交易所曾经在收盘后推送了一条价格为0.01元的成交记录，明显是撮合引擎的bug。如果你不做清洗，策略会以为市场崩了，直接给你平仓。

清洗步骤，我一般按这个顺序来：

去重：同一笔成交可能被推送两次。用交易编号去重。
价格过滤：超出合理范围的价格直接剔除。比如CEA价格不可能低于10元，也不可能高于200元。
数量过滤：成交量不能为0或负数。
时间戳校验：时间戳不能是未来的，也不能比上一笔还早。
缺失值处理：对于短暂的缺失，可以用前一笔填充。超过1秒的缺失，建议标记为异常区间。

def clean_ticks(ticks):
    # 1. 去重
    ticks = ticks.drop_duplicates(subset=['trade_id'])
    
    # 2. 价格过滤
    ticks = ticks[(ticks['price'] >= 10) & (ticks['price'] <= 200)]
    
    # 3. 数量过滤
    ticks = ticks[ticks['volume'] > 0]
    
    # 4. 时间戳排序
    ticks = ticks.sort_values('timestamp')
    
    # 5. 检查时间戳是否倒挂
    time_diff = ticks['timestamp'].diff()
    invalid_idx = time_diff < 0
    if invalid_idx.any():
        print(f"发现{invalid_idx.sum()}条时间戳异常数据，已剔除")
        ticks = ticks[~invalid_idx]
    
    return ticks.reset_index(drop=True)

注意：清洗逻辑不能太激进。我曾经把「价格跳空」当成异常数据删掉了，结果发现那是交易所开盘集合竞价产生的正常跳空。嗯，那次回测结果直接失真了半个月。

4.3 时间戳处理：别让时钟骗了你

时间戳，是高频数据里最要命的东西。

你想想看，你的策略需要知道「A事件发生在B事件之前还是之后」。如果时间戳不准，因果关系就乱了。策略会以为「先有鸡还是先有蛋」这个问题有了新答案。

常见的时间戳问题：

精度不一致：有的交易所给毫秒，有的给微秒，有的甚至只给秒。
时区问题：国内交易所用北京时间，但你的服务器可能设了UTC。
时钟漂移：服务器运行久了，系统时钟会慢慢偏离真实时间。

我的处理原则很简单：统一转成UTC毫秒时间戳，用int64存储。别用字符串，别用datetime对象，计算效率差太多了。

import pandas as pd

def normalize_timestamp(ts, source_tz='Asia/Shanghai'):
    """
    将各种格式的时间戳统一为UTC毫秒
    """
    if isinstance(ts, (int, float)):
        # 假设是毫秒时间戳
        return int(ts)
    elif isinstance(ts, str):
        # 假设是字符串格式 '2024-01-15 09:30:00.123'
        dt = pd.to_datetime(ts)
        # 转换为UTC
        dt_utc = dt.tz_localize(source_tz).tz_convert('UTC')
        return int(dt_utc.timestamp() * 1000)
    else:
        raise ValueError(f"不支持的时间戳格式: {type(ts)}")

4.4 时钟同步：NTP是你的好朋友

好了，现在你的数据时间戳都统一了。但还有一个问题：你的服务器时钟准吗？

我做过一个测试：一台跑了三个月的服务器，系统时钟比真实时间慢了整整2.3秒。2.3秒啊兄弟们，在高频交易里，这足够让一个套利策略从盈利变成亏损。

解决方案：NTP（网络时间协议）

每台交易服务器都必须配置NTP服务，并且要配置多个NTP服务器做冗余。我建议至少配4个：两个国内的，两个国外的。

Linux下的配置示例：

# /etc/ntp.conf
server ntp.aliyun.com iburst
server ntp.tencent.com iburst
server pool.ntp.org iburst
server time.google.com iburst

# 启动并启用
systemctl enable ntpd
systemctl start ntpd

# 检查同步状态
ntpq -p

我的经验：光配NTP还不够。你还需要监控时钟偏移量。我写了一个监控脚本，每5分钟检查一次偏移，如果超过1毫秒就报警。别等到策略亏钱了才发现时钟有问题。

4.5 数据对齐：让不同数据源「对齐颗粒度」

最后一步，数据对齐。

你的策略可能需要同时用到成交数据、订单簿数据、还有外部数据（比如碳配额拍卖结果）。这些数据的时间戳精度不一样，推送频率也不一样。怎么把它们对齐到同一个时间轴上？

常用的对齐方法：

方法	适用场景	缺点
向前填充	低频数据对齐到高频	会引入未来信息（回测时要注意）
向后填充	高频数据对齐到低频	有延迟
线性插值	价格类数据	不适合离散事件（如成交）
时间桶聚合	生成等间隔的OHLC数据	丢失了桶内微观结构信息

我个人最常用的是时间桶聚合。把1秒切成100个10毫秒的桶，每个桶内取最后一笔成交作为该桶的代表。这样既保留了微观结构，又让数据规整了。

def bucket_align(ticks, bucket_ms=10):
    """
    将Tick数据按时间桶对齐
    """
    # 计算每个Tick所属的桶
    ticks['bucket'] = ticks['timestamp'] // bucket_ms
    
    # 每个桶取最后一笔
    aligned = ticks.groupby('bucket').last().reset_index()
    
    # 生成完整的时间桶序列
    full_buckets = range(aligned['bucket'].min(), aligned['bucket'].max() + 1)
    aligned = aligned.set_index('bucket').reindex(full_buckets, method='ffill')
    
    return aligned

一个小技巧：对齐的时候，记得把桶的边界时间戳也存下来。这样你回测的时候可以精确知道「这个价格是在桶的哪个位置产生的」。别问我为什么知道这个重要——有一次我回测一个抢单策略，就是因为桶内时间精度不够，导致模拟的成交顺序跟实际完全相反。

4.6 本章知识体系

说了这么多，我画了一张图帮你梳理一下整个流程。从数据源到最终可用的对齐数据，每一步都有坑，每一步都需要你亲自去验证。

嗯，这一章的内容就到这里。数据是策略的血液，血液不干净，策略迟早要出问题。下一章我们会聊到订单簿重建和微观结构特征提取，到时候这些清洗好的数据就能派上大用场了。

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

📚 课程目录

4、高频数据基础：Tick级数据获取、数据清洗与对齐、时间戳处理与时钟同步

4.1 Tick级数据获取：从交易所到你的硬盘

4.2 数据清洗：脏数据是策略的毒药

4.3 时间戳处理：别让时钟骗了你

4.4 时钟同步：NTP是你的好朋友

4.5 数据对齐：让不同数据源「对齐颗粒度」

4.6 本章知识体系