第3章时间戳标准化

第三节：时间戳标准化——把那些乱七八糟的时间格式收拾得服服帖帖

做量化交易的朋友，十有八九都被时间戳坑过。

我刚开始搞多市场数据清洗那会儿，就吃过这个亏。明明两个交易所的数据看起来都对得上，一算价差，全乱套了。查了半天，原来是时间戳差了8个小时——一个用的UTC，一个用的北京时间。

嗯，今天咱们就来聊聊这个事儿。

3.1 时间戳的三大“门派”

说白了，不同交易所的时间格式，无非就这三种：

UTC时间：比如 2025-01-15T08:30:00Z，后面带个Z，表示世界标准时间。很多国际交易所都用这个。
本地时间：比如 2025-01-15 16:30:00，不带时区信息。你得自己猜它是哪个时区的。国内交易所喜欢这么干。
毫秒级时间戳：比如 1736901000000，一串数字。看着简单，但不同系统可能用秒、毫秒、甚至微秒，一不小心就搞错。

你想想看，如果A交易所给的是UTC，B交易所给的是北京时间，C交易所给了一串数字。你要把它们对齐到同一根K线上，不统一处理怎么行？

3.2 我的统一方案：全部转成毫秒级时间戳

我个人习惯，不管数据源是什么格式，最终都转成毫秒级时间戳。为什么？

计算快：整数加减，比字符串快得多
精度可控：毫秒级足够应对大部分行情
存储省：一个Long类型就搞定

下面是我常用的处理逻辑，用Python写的，你们感受一下：

import pandas as pd
from datetime import datetime, timezone, timedelta

def normalize_timestamp(ts, source_format='utc'):
    """
    统一转成毫秒级时间戳
    source_format: 'utc', 'local', 'milli', 'micro'
    """
    if source_format == 'utc':
        # 处理 '2025-01-15T08:30:00Z' 这种格式
        dt = datetime.fromisoformat(ts.replace('Z', '+00:00'))
        return int(dt.timestamp() * 1000)
    
    elif source_format == 'local':
        # 假设是北京时间（UTC+8）
        dt = datetime.strptime(ts, '%Y-%m-%d %H:%M:%S')
        dt = dt.replace(tzinfo=timezone(timedelta(hours=8)))
        return int(dt.timestamp() * 1000)
    
    elif source_format == 'milli':
        # 已经是毫秒级，直接返回
        return int(ts)
    
    elif source_format == 'micro':
        # 微秒转毫秒
        return int(ts) // 1000
    
    else:
        raise ValueError(f'未知的时间格式: {source_format}')

小提示：我建议你把这个函数封装成一个工具类，放在项目的utils目录下。因为后面几乎每个清洗步骤都会用到它。

3.3 避坑指南：我曾经踩过的三个坑

坑一：时区信息丢失

我曾经从某交易所拿到一批历史数据，时间戳看起来是UTC，但其实是本地时间。我直接按UTC处理，结果所有K线都偏移了8小时。后来我养成了一个习惯：拿到数据先画个时间分布图，看看交易时间是不是符合常识。

坑二：毫秒和微秒傻傻分不清

有一次，一个数据源给的时间戳是13位的，我以为是毫秒。结果一算，日期对不上。仔细一看，人家给的是微秒（16位）。嗯，从那以后，我写了个自动检测函数：

def detect_timestamp_unit(ts):
    """自动判断时间戳单位"""
    ts_str = str(int(ts))
    length = len(ts_str)
    
    if length == 10:
        return 'second'
    elif length == 13:
        return 'milli'
    elif length == 16:
        return 'micro'
    else:
        return 'unknown'

坑三：夏令时

这个坑主要出现在欧美交易所。有些交易所用夏令时，有些不用。我建议统一转成UTC，彻底避开这个麻烦。

3.4 实战：多市场时间对齐流程

下面这张图，是我在实际项目中用的时间对齐流程。你看一眼就明白了：

流程其实不复杂，核心就三步：

检测格式：先判断数据源是什么格式
统一转换：调用标准化函数转成毫秒戳
校验对齐：检查转换后的时间是否合理

  核心原则：不管数据源多乱，最终只保留一种时间格式。我选毫秒级时间戳，因为它兼顾了精度和计算效率。

3.5 校验环节：别让脏数据溜过去

转换完了就完事儿了？不是的。我每次都会做一轮校验：

def validate_timestamps(series):
    """校验时间戳是否合理"""
    # 1. 检查是否有空值
    if series.isnull().any():
        print(f'发现 {series.isnull().sum()} 个空值')
    
    # 2. 检查是否在合理范围内
    now = int(datetime.now().timestamp() * 1000)
    if series.max() > now + 86400000:  # 允许未来1天
        print('警告：存在未来时间戳')
    
    # 3. 检查是否单调递增
    if not series.is_monotonic_increasing:
        print('警告：时间戳不是单调递增的')
    
    # 4. 检查重复值
    if series.duplicated().any():
        print(f'发现 {series.duplicated().sum()} 个重复时间戳')

这四步走完，基本能过滤掉90%的时间戳问题。剩下的10%，嗯，那就得靠经验了。

特别注意：千万不要相信数据源说的时间格式。我遇到过好几次，文档写的是UTC，实际给的是本地时间。一定要自己验证一遍。

好了，时间戳标准化这块儿，说白了就是：统一格式、统一时区、统一精度。把这三点做到位，后面的数据对齐工作就轻松多了。

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

📚 课程目录

第三节：时间戳标准化——把那些乱七八糟的时间格式收拾得服服帖帖

3.1 时间戳的三大“门派”

3.2 我的统一方案：全部转成毫秒级时间戳

3.3 避坑指南：我曾经踩过的三个坑

3.4 实战：多市场时间对齐流程

3.5 校验环节：别让脏数据溜过去