1. 跳水回放数据源接入与时间对齐

数据源构成与对齐目标

本案例接入的原始行情由两类数据流组成：逐笔成交（trade）与盘口快照（snapshot）。两者在原始文件中往往使用不同的时间基准——逐笔成交常带毫秒级成交时间，盘口快照则可能按固定间隔（如每 3 秒）推送，且时间戳字段可能为交易所时间、本地接收时间或字符串格式。回放前必须把它们统一到同一时间轴，否则会出现“成交已发生但盘口还停留在上一档”的错位。

对齐的核心不是简单排序，而是建立统一时间基准 + 单调递增事件序列。我们约定：以交易所时间为准，统一转换为毫秒整数时间戳（epoch ms），并保证同一时间戳下快照先于成交生效（因为快照代表该时刻的盘口状态，成交是在该状态上发生的）。

时间戳统一的具体做法

原始数据中时间字段常见三种形态："09:31:02.350"（当日时分秒毫秒字符串）、"2024-06-12 09:31:02.350"（带日期）、以及整数毫秒。统一策略是：先解析为 Python datetime，再转为 epoch 毫秒整数。若原始只有时间没有日期，需补一个基准交易日，避免跨日回放时时间倒退。

from datetime import datetime, timezone

TRADE_DATE = "2024-06-12"  # 示例交易日，实际由数据文件元信息提供

def to_epoch_ms(ts, base_date=TRADE_DATE):
    """将多种时间格式统一为 epoch 毫秒整数。"""
    if isinstance(ts, (int, float)):
        return int(ts)
    s = str(ts).strip()
    # 带日期
    for fmt in ("%Y-%m-%d %H:%M:%S.%f", "%Y-%m-%d %H:%M:%S"):
        try:
            dt = datetime.strptime(s, fmt)
            return int(dt.replace(tzinfo=timezone.utc).timestamp() * 1000)
        except ValueError:
            pass
    # 仅时间，补基准日期
    for fmt in ("%H:%M:%S.%f", "%H:%M:%S"):
        try:
            dt = datetime.strptime(f"{base_date} {s}", f"%Y-%m-%d {fmt}")
            return int(dt.replace(tzinfo=timezone.utc).timestamp() * 1000)
        except ValueError:
            pass
    raise ValueError(f"无法解析时间戳: {ts!r}")

注意：这里统一用 UTC 时区做转换，只要两类数据用同一规则，相对顺序就正确；若原始数据本身是本地时间且带时区偏移，应在解析时显式指定，避免混用。

对齐与一致性验证

对齐后生成统一事件列表，每条事件带 ts_ms、type（snapshot/trade）与原始载荷。排序键为 (ts_ms, type_priority)，其中快照优先级 0、成交优先级 1，保证同一毫秒内快照先生效。随后做两项一致性检查：

• 单调性检查：排序后 ts_ms 不得回退，若回退说明解析或时区有误。
• 覆盖性检查：每笔成交的时间应落在某个快照区间内（前快照 ≤ 成交时间 < 后快照），否则标记为“无盘口覆盖”的孤立成交，供后续回放时提示。

def build_aligned_stream(trades, snaps):
    events = []
    for s in snaps:
        events.append({"ts_ms": to_epoch_ms(s["snap_time"]), "type": "snapshot", "data": s})
    for t in trades:
        events.append({"ts_ms": to_epoch_ms(t["trade_time"]), "type": "trade", "data": t})
    # 同刻快照优先
    events.sort(key=lambda e: (e["ts_ms"], 0 if e["type"] == "snapshot" else 1))
    # 单调性检查
    for a, b in zip(events, events[1:]):
        assert b["ts_ms"] >= a["ts_ms"], "时间轴出现回退"
    return events

def check_coverage(events):
    """标记无盘口覆盖的成交。"""
    last_snap_ts = None
    orphan = []
    for e in events:
        if e["type"] == "snapshot":
            last_snap_ts = e["ts_ms"]
        elif last_snap_ts is None:
            orphan.append(e)
    return orphan

可复用的对齐结果结构

对齐产物建议保存为一份事件流文件（如 JSONL），每行一条事件，字段固定为 ts_ms、type、payload。这样后续回放模块只需按行读取、按 ts_ms 推进，不必再关心原始时间格式差异。示例首行：

{"ts_ms": 1718184662000, "type": "snapshot", "payload": {"bid1": 10.25, "ask1": 10.26, "...": "..."}}
{"ts_ms": 1718184662350, "type": "trade", "payload": {"price": 10.26, "volume": 300, "direction": "B"}}

至此，逐笔成交与盘口快照已落在同一毫秒时间轴上，且同刻顺序明确，可直接作为后续合并回放流的输入。