3、时间序列对齐：不同交易所时间戳统一、时区转换、非交易时段过滤
做量化交易，最头疼的问题之一就是时间。
你想想看，A股、港股、美股、期货、币圈……每个交易所都有自己的时间规则。有的用UTC，有的用本地时间，有的还分夏令时冬令时。我刚开始做多市场策略的时候，就吃过这个亏——数据对不齐，信号全乱套了。
这一章，咱们就专门解决这个问题。
3.1 为什么时间对齐这么重要？
说白了，时间戳是数据的“身份证”。
如果身份证对不上，你后面做的所有特征工程、回测、实盘，都是白搭。我见过一个团队，因为没处理好时区问题，回测年化收益30%，实盘直接亏了10%。原因很简单：回测时用的历史数据是UTC时间，实盘信号触发用的是本地时间，差了8个小时。
嗯，这种坑，踩一次就够了。
核心原则： 所有数据必须统一到同一个时间基准上，通常是UTC时间。
3.2 时间戳统一：从混乱到有序
不同交易所返回的时间戳格式五花八门。我整理了一下，常见的有这几种：
| 交易所 | 原始时间格式 | 说明 |
|---|---|---|
| Binance | 毫秒级Unix时间戳 | 如 1700000000000 |
| Coinbase | ISO 8601字符串 | 如 2024-01-01T00:00:00Z |
| 上交所 | 本地时间字符串 | 如 2024-01-01 09:30:00 |
| CME期货 | 纳秒级Unix时间戳 | 如 1700000000000000000 |
我个人习惯的做法是：全部转成毫秒级Unix时间戳。为什么？因为整数运算快，而且精度够用。
import pandas as pd
import numpy as np
from datetime import datetime, timezone
def unify_timestamp(ts, source_format='unix_ms'):
"""
统一时间戳格式为毫秒级Unix时间戳
"""
if source_format == 'unix_ms':
# 已经是毫秒级，直接返回
return ts
elif source_format == 'unix_ns':
# 纳秒转毫秒
return ts // 1_000_000
elif source_format == 'iso8601':
# ISO字符串转毫秒
dt = datetime.fromisoformat(ts.replace('Z', '+00:00'))
return int(dt.timestamp() * 1000)
elif source_format == 'local_str':
# 本地时间字符串，需要指定时区
# 这里以上海为例
dt = datetime.strptime(ts, '%Y-%m-%d %H:%M:%S')
# 假设是上海时区
from pytz import timezone
shanghai_tz = timezone('Asia/Shanghai')
dt = shanghai_tz.localize(dt)
return int(dt.timestamp() * 1000)
else:
raise ValueError(f'未知的时间格式: {source_format}')
# 示例用法
binance_ts = 1700000000000
coinbase_ts = '2024-01-01T00:00:00Z'
shanghai_ts = '2024-01-01 09:30:00'
print(unify_timestamp(binance_ts, 'unix_ms')) # 1700000000000
print(unify_timestamp(coinbase_ts, 'iso8601')) # 1704067200000
print(unify_timestamp(shanghai_ts, 'local_str')) # 1704065400000
小技巧： 我建议你在数据库里专门建一个字段叫
ts_unified，所有数据入库前都先过一遍这个函数。这样后面查询、join、计算都省心。
3.3 时区转换：别让夏令时坑了你
时区转换本身不难，难的是夏令时。
我曾经在美股策略上栽过跟头。美国夏令时切换那天，我的数据源突然少了1个小时的数据。排查了半天才发现，原来是夏令时开始那天凌晨2点直接跳到3点，那1个小时根本不存在。
处理时区转换，我推荐用 pytz 库，它内置了完整的时区数据库，包括夏令时规则。
import pytz
from datetime import datetime
def convert_timezone(ts_ms, from_tz='UTC', to_tz='America/New_York'):
"""
将毫秒级时间戳从一个时区转换到另一个时区
"""
# 先转成datetime对象
dt_utc = datetime.fromtimestamp(ts_ms / 1000, tz=pytz.UTC)
# 获取源时区
from_zone = pytz.timezone(from_tz)
to_zone = pytz.timezone(to_tz)
# 转换
dt_from = dt_utc.astimezone(from_zone)
dt_to = dt_utc.astimezone(to_zone)
return dt_to
# 示例：将UTC时间转成纽约时间
utc_ts = 1700000000000 # 2023-11-14 22:13:20 UTC
ny_time = convert_timezone(utc_ts, 'UTC', 'America/New_York')
print(ny_time) # 2023-11-14 17:13:20-05:00
注意： 夏令时切换时，一天可能只有23小时或25小时。如果你的策略对时间精度要求高（比如高频交易），一定要做特殊处理。我一般会在切换日前后各加一个缓冲期，手动检查数据完整性。
3.4 非交易时段过滤：别让垃圾数据污染你的模型
每个交易所都有非交易时段。比如A股中午休市1.5小时，美股盘前盘后交易量很小，币圈虽然24小时交易但也有流动性极低的时段。
这些时段的数据，说白了就是噪音。如果你不过滤掉，特征工程算出来的指标全是错的。
我一般会维护一个“交易时段表”，针对每个交易所、每个品种单独定义。
| 交易所 | 交易时段（本地时间） | 说明 |
|---|---|---|
| 上交所 | 09:30-11:30, 13:00-15:00 | 中午休市，周末休市 |
| 港交所 | 09:30-12:00, 13:00-16:00 | 中午休市，周末休市 |
| 纽交所 | 09:30-16:00（东部时间） | 盘前盘后交易量小，建议过滤 |
| Binance | 24小时 | 但凌晨2-4点流动性极低 |
def filter_trading_hours(df, exchange='SH', ts_col='ts_unified'):
"""
过滤非交易时段的数据
df: DataFrame，必须包含统一时间戳列
exchange: 交易所代码
"""
# 先转成本地时间
if exchange == 'SH':
local_tz = 'Asia/Shanghai'
trading_sessions = [
('09:30', '11:30'),
('13:00', '15:00')
]
elif exchange == 'US':
local_tz = 'America/New_York'
trading_sessions = [
('09:30', '16:00')
]
else:
raise ValueError(f'未知交易所: {exchange}')
# 转换时间戳到本地时间
df['local_time'] = df[ts_col].apply(
lambda x: convert_timezone(x, 'UTC', local_tz)
)
# 提取时分
df['hour_min'] = df['local_time'].dt.strftime('%H:%M')
# 判断是否在交易时段内
def is_trading(hour_min):
for start, end in trading_sessions:
if start <= hour_min <= end:
return True
return False
df['is_trading'] = df['hour_min'].apply(is_trading)
# 过滤
result = df[df['is_trading']].copy()
# 清理临时列
result.drop(columns=['local_time', 'hour_min', 'is_trading'], inplace=True)
return result
# 示例
# filtered_df = filter_trading_hours(raw_df, exchange='SH')
避坑指南： 我曾经在过滤币圈数据时，直接把凌晨2-4点的数据全删了。结果后来发现，有些重大新闻恰恰是在那个时段发布的，导致策略错过了大行情。所以我的建议是：过滤要谨慎，最好保留原始数据，只在计算特征时跳过非活跃时段。
3.5 实战：完整的时间对齐流水线
好了，我们把上面三个步骤串起来，形成一个完整的流水线。
def time_alignment_pipeline(df, exchange, ts_col='timestamp', source_format='unix_ms'):
"""
完整的时间对齐流水线
1. 统一时间戳
2. 时区转换（如果需要）
3. 过滤非交易时段
"""
# 第一步：统一时间戳
df['ts_unified'] = df[ts_col].apply(
lambda x: unify_timestamp(x, source_format)
)
# 第二步：时区转换（这里统一到UTC）
# 如果原始数据已经是UTC，这步可以跳过
# 但为了保险，我一般都会显式转换一次
df['ts_utc'] = df['ts_unified'] # 假设已经是UTC
# 第三步：过滤非交易时段
df_filtered = filter_trading_hours(df, exchange, 'ts_utc')
return df_filtered
# 使用示例
# raw_data = pd.read_csv('shanghai_stock_2024.csv')
# clean_data = time_alignment_pipeline(raw_data, exchange='SH', source_format='local_str')
核心要点： 时间对齐不是一次性工作。每次接入新数据源、每次切换交易所、每次遇到夏令时切换，都要重新检查一遍。我建议你把时间对齐逻辑封装成一个独立的模块，方便复用和测试。
3.6 知识体系总览
下面这张图，帮你把本章的核心逻辑串起来：
嗯，这张图把整个流程讲清楚了。从原始数据进来，经过三步处理，最终得到干净、对齐的数据，可以直接喂给特征工程模块。
我个人习惯把时间对齐放在数据清洗的第一步。为什么？因为后面的所有操作——重采样、计算指标、合并数据——都依赖正确的时间戳。这一步做扎实了，后面就顺了。
最后提醒一句： 时间对齐没有银弹。每个交易所、每个品种都有自己的规则。我的做法是：先写一个通用的框架，然后针对每个数据源写具体的适配器。这样既保证了统一性，又保留了灵活性。
公众号:蓝海资料掘金营，微信deep3321