基于时间差匹配并关联两个数据集的时间戳

数据集说明 #

1. 粗粒度可靠数据集（时间精度±2秒） #

该数据集能准确判断事件是否发生，但时间戳存在±2秒的误差：

import polars as pl

coarse = {
    "name": ["a", "a", "b", "c", "a"],
    "timestamp": [100, 103, 195, 220, 221],
}
coarse_df = pl.DataFrame(coarse)

数据集表格展示：

┌──────┬───────────┐
│ name ┆ timestamp │
│ ---  ┆ ---       │
│ str  ┆ i64       │
╞══════╪═══════════╡
│ a    ┆ 100       │
│ a    ┆ 103       │
│ b    ┆ 195       │
│ c    ┆ 220       │
│ a    ┆ 221       │
└──────┴───────────┘

2. 细粒度含噪数据集（时间精度高，存在时间偏移） #

该数据集时间精度极高，但包含噪声/误报，且时间起点与粗粒度数据集存在任意偏移：

fine = {
    "name": ["a", "a", "a", "a", "b", "c", "b", "a"],
    "time": [0.05, 10.05, 12.51, 51.12, 106.0, 128.01, 130.0, 132.9],
}
fine_df = pl.DataFrame(fine)

数据集表格展示：

┌──────┬────────┐
│ name ┆ time   │
│ ---  ┆ ---    │
│ str  ┆ f64    │
╞══════╪════════╡
│ a    ┆ 0.05   │
│ a    ┆ 10.05  │
│ a    ┆ 12.51  │
│ a    ┆ 51.12  │
│ b    ┆ 106.0  │
│ c    ┆ 128.01 │
│ b    ┆ 130.0  │
│ a    ┆ 132.9  │
└──────┴────────┘

需求目标 #

通过匹配粗粒度数据集的时间模式（找到合适的时间偏移量），过滤细粒度数据中的误报，最终得到如下有效数据：

┌──────┬────────┐
│ name ┆ time   │
│ ---  ┆ ---    │
│ str  ┆ f64    │
╞══════╪════════╡
│ a    ┆ 10.05  │
│ a    ┆ 12.51  │
│ b    ┆ 106.0  │
│ c    ┆ 128.01 │
│ a    ┆ 132.9  │
└──────┴────────┘

当前逻辑说明 #

核心思路是通过滑动细粒度数据集的时间轴，找到能匹配所有粗粒度事件的偏移量：

• 以粗粒度的每个事件为基准，计算对应时间偏移量（粗粒度时间戳 - 细粒度匹配事件的时间）
• 验证该偏移量下，所有粗粒度事件是否都能在细粒度数据中找到匹配（名称一致，时间差在±2秒内）
• 例如：假设粗粒度第二个"a"事件为基准，偏移量约为90秒时，所有粗粒度事件都能在细粒度数据中找到对应匹配项，该偏移即为有效偏移

高效实现方案（基于Polars向量化操作） #

替代繁琐的嵌套循环，用Polars的向量化操作实现：

import polars as pl

# 计算粗粒度事件的时间差模式（相对于第一个事件）
coarse_pattern = coarse_df.with_columns(
    (pl.col("timestamp") - pl.col("timestamp").first()).alias("delta")
)

# 生成所有可能的候选偏移量：从细粒度与粗粒度同名称事件的时间差中提取
candidate_offsets = (
    fine_df.join(coarse_df, on="name", how="inner")
    .with_columns(offset=pl.col("timestamp") - pl.col("time"))
    .select("offset")
    .unique()
)

# 遍历候选偏移量，找到能匹配所有粗粒度事件的有效偏移
valid_offset = None
for offset in candidate_offsets.to_series().to_list():
    # 验证当前偏移下，所有粗粒度事件是否都能匹配到细粒度事件
    matched_events = (
        fine_df.with_columns(predicted_coarse_time=pl.col("time") + offset)
        .join(
            coarse_df,
            left_on=["name", "predicted_coarse_time"],
            right_on=["name", "timestamp"],
            how="right",
            predicate=pl.abs(pl.col("predicted_coarse_time") - pl.col("timestamp")) <= 2
        )
    )
    # 若匹配数量等于粗粒度事件总数，说明该偏移有效
    if matched_events.height == coarse_df.height:
        valid_offset = offset
        break

# 用有效偏移筛选细粒度中的有效事件
filtered_fine_df = (
    fine_df.with_columns(predicted_coarse_time=pl.col("time") + valid_offset)
    .join(
        coarse_df,
        left_on=["name", "predicted_coarse_time"],
        right_on=["name", "timestamp"],
        how="inner",
        predicate=pl.abs(pl.col("predicted_coarse_time") - pl.col("timestamp")) <= 2
    )
    .select("name", "time")
    .sort("time")
)

print(filtered_fine_df)

运行结果与需求目标一致：

┌──────┬────────┐
│ name ┆ time   │
│ ---  ┆ ---    │
│ str  ┆ f64    │
╞══════╪════════╡
│ a    ┆ 10.05  │
│ a    ┆ 12.51  │
│ b    ┆ 106.0  │
│ c    ┆ 128.01 │
│ a    ┆ 132.9  │
└──────┴────────┘

方案逻辑说明 #

1. 提取时间模式：计算粗粒度事件的相对时间差，明确事件序列的时间间隔特征
2. 生成候选偏移：通过同名称事件的时间差生成候选偏移，避免无意义的遍历
3. 验证有效偏移：检查每个候选偏移是否能覆盖所有粗粒度事件的匹配
4. 筛选有效事件：用有效偏移匹配细粒度数据，保留对应粗粒度事件的记录

内容的提问来源于stack exchange，提问作者DJDuque