2. 逐笔委托与逐笔成交的时序对齐

对齐任务拆解：两条流、两个键、一个顺序约束

逐笔委托流（order stream）与逐笔成交流（trade stream）来自不同通道，时间戳精度、到达顺序、字段口径都不完全一致。对齐的目标不是简单拼接，而是回答三个问题：哪一笔成交由哪一笔委托触发、委托到达与成交执行谁先谁后、合并后的事件序列是否满足因果约束。本例用订单号作为主键、时间戳作为次序键，构建一张合并事件表，并用三条规则验证顺序合理性。

对齐键与顺序约束

对齐的核心是双键匹配：order_id 决定“哪条成交属于哪条委托”，ts 决定“谁先谁后”。成交流中的 order_id 指向触发该笔成交的委托单，因此以委托流为左表、成交流为右表做左连接，可以保留“已委托但尚未成交”的订单，这对后续计算挂单时长、成交概率等特征很关键。合并后必须满足三条约束：

• 因果约束：同一 order_id 下，ORDER 事件的 ts 必须小于等于其 TRADE 事件的 ts；
• 单调约束：按 ts 升序排列后，event_seq 必须严格递增，不允许出现时间倒挂；
• 延迟约束：lag = ts_trade − ts_order 必须非负，且不超过该市场合理的撮合延迟上限（示例中取 500 ms 作为教学阈值）。

用 pandas 实现对齐与校验

下面代码自包含：构造两份示例数据（明确为教学示例，非实测），完成字段规整、双键合并、事件序号生成与三条约束校验，并输出违规明细。

import pandas as pd

# 教学示例数据（非实测）
orders = pd.DataFrame({
    "order_id": ["O1", "O2", "O3", "O4"],
    "ts_order": pd.to_datetime([
        "2024-06-03 09:30:00.120", "2024-06-03 09:30:00.180",
        "2024-06-03 09:30:00.260", "2024-06-03 09:30:00.410"]),
    "side": ["B", "S", "B", "S"],
    "order_price": [10.02, 10.03, 10.03, 10.01],
    "order_qty": [300, 500, 200, 400],
})

trades = pd.DataFrame({
    "trade_id": ["T1", "T2", "T3", "T4"],
    "ts_trade": pd.to_datetime([
        "2024-06-03 09:30:00.150", "2024-06-03 09:30:00.210",
        "2024-06-03 09:30:00.290", "2024-06-03 09:30:00.430"]),
    "order_id": ["O1", "O2", "O3", "O4"],
    "trade_price": [10.02, 10.03, 10.03, 10.01],
    "trade_qty": [100, 300, 200, 150],
})

# 1) 双键左连接：order_id 匹配，保留未成交委托
merged = orders.merge(trades, on="order_id", how="left")

# 2) 展开为长表：每条委托与每条成交各成一行事件
order_events = merged[["order_id", "ts_order", "side", "order_price", "order_qty"]].copy()
order_events = order_events.rename(columns={"ts_order": "ts"})
order_events["event_type"] = "ORDER"
order_events["price"] = order_events["order_price"]
order_events["qty"] = order_events["order_qty"]

trade_events = merged.dropna(subset=["trade_id"])[
    ["order_id", "ts_trade", "trade_price", "trade_qty"]].copy()
trade_events = trade_events.rename(columns={"ts_trade": "ts"})
trade_events["event_type"] = "TRADE"
trade_events["price"] = trade_events["trade_price"]
trade_events["qty"] = trade_events["trade_qty"]

events = pd.concat([order_events, trade_events], ignore_index=True)
events = events.sort_values(["ts", "event_type"], ascending=[True, False]).reset_index(drop=True)
events["event_seq"] = events.index + 1

# 3) 计算 lag：同一 order_id 下 TRADE 相对 ORDER 的延迟
order_ts = events[events.event_type == "ORDER"].set_index("order_id")["ts"]
events["lag_ms"] = events.apply(
    lambda r: (r["ts"] - order_ts[r["order_id"]]).total_seconds() * 1000
    if r["event_type"] == "TRADE" else None, axis=1)

# 4) 三条约束校验
viol_causal = events[(events.event_type == "TRADE") & (events.lag_ms < 0)]
viol_mono = events[events.ts.diff().dt.total_seconds().fillna(0) < 0]
viol_lag = events[(events.event_type == "TRADE") & (events.lag_ms > 500)]

print(events[["event_seq", "event_type", "order_id", "ts", "price", "qty", "lag_ms"]])
print("因果违规:", len(viol_causal), "| 单调违规:", len(viol_mono), "| 延迟违规:", len(viol_lag))

结果判读与常见异常

在示例数据上运行后，event_seq 依次为 O1-ORDER、O1-TRADE、O2-ORDER、O2-TRADE、O3-ORDER、O3-TRADE、O4-ORDER、O4-TRADE，三条违规计数均为 0，说明对齐结果满足因果与单调约束。实际数据中更常见的是以下三类异常，需要在对齐阶段就识别出来：