如何高效实现百万级事件时间标签的多重复通道分类？

问题背景 #

我有100万个单调有序的无符号整数（uint）时间标签，每个标签对应一次检测事件（无标签则表示对应时间段无检测）。同时存在一张单调有序、通道窗口无重叠的时序表，定义了若干分类规则：若某个通道指定的时间段内出现至少一次检测事件，则该通道标记为“事件发生（True）”。时序表的通道规则会按固定周期（single_run_duration）循环生效。

当前采用嵌套循环实现：遍历每个时间标签，再逐个检查所有通道条目，处理速度极慢，需要更高效的算法优化。

原有实现伪代码：

tags = [8,500,1001,1265,1266,1502,1600,2990]
table = [0,1260,1480]
window = 10
single_run_duration = 1500
reps = ceil(max(tags)/single_run_duration)
storage = zero_array(len(table),reps)

for ti in tags:
   for ch in table:
      if ch <= (ti % single_run_duration) <= ch+window:
         run = floor(ti/single_run_duration)
         storage(ch,run) = True
// 最终storage结果：
storage = [[True,True],[True,False],[False,False]]

核心问题分析 #

原有嵌套循环的时间复杂度为O(N*M)（N为标签数，M为通道数），当N达到1e6时，即使M不大，计算量也会指数级增长，导致速度瓶颈。

但我们可以利用两个关键特性优化：

• 标签列表是单调有序的
• 时序表通道是单调有序且窗口无重叠的

优化方案1：双指针遍历法 #

利用标签和通道的有序性，通过双指针避免重复遍历通道，将时间复杂度降至接近O(N)。

实现步骤 #

1. 先将所有标签按周期分组：每个标签ti对应周期run = ti // single_run_duration，提取周期内的时间t_in_run = ti % single_run_duration
2. 对每个周期的标签列表（本身也是有序的），用双指针同步遍历标签和通道：因为标签递增，通道窗口有序，指针无需回退，一次遍历即可完成匹配

优化后伪代码 #

tags = [8,500,1001,1265,1266,1502,1600,2990]
table = [0,1260,1480]
window = 10
single_run_duration = 1500

# 按周期分组标签
tag_groups = defaultdict(list)
max_run = 0
for ti in tags:
    run = ti // single_run_duration
    tag_groups[run].append(ti % single_run_duration)
    if run > max_run:
        max_run = run
reps = max_run + 1
storage = zero_array(len(table), reps)

# 遍历每个周期的标签组
for run in tag_groups:
    t_list = tag_groups[run]
    ch_idx = 0
    num_channels = len(table)
    for t_in_run in t_list:
        # 利用有序性，指针只前进不后退
        while ch_idx < num_channels:
            ch_start = table[ch_idx]
            ch_end = ch_start + window
            if t_in_run > ch_end:
                # 当前标签超过通道窗口，移到下一个通道
                ch_idx += 1
            elif t_in_run >= ch_start:
                # 匹配到当前通道，标记为True
                storage[ch_idx][run] = True
                # 通道无重叠，直接处理下一个标签
                break
            else:
                # 因标签递增，此情况不会出现，直接跳出
                break

优化方案2：二分查找法 #

针对每个周期的标签列表（有序），用二分查找快速判断通道窗口内是否存在匹配标签，适合通道数较少的场景。

实现步骤 #

1. 同样先按周期分组标签
2. 对每个周期的标签列表，遍历每个通道，用二分查找定位窗口内的标签范围，若存在则标记通道为True

优化后伪代码 #

# 需引入二分查找工具（如Python的bisect模块）
tags = [8,500,1001,1265,1266,1502,1600,2990]
table = [0,1260,1480]
window = 10
single_run_duration = 1500

tag_groups = defaultdict(list)
max_run = 0
for ti in tags:
    run = ti // single_run_duration
    tag_groups[run].append(ti % single_run_duration)
    if run > max_run:
        max_run = run
reps = max_run + 1
storage = zero_array(len(table), reps)

for run in tag_groups:
    t_list = tag_groups[run]
    for ch_idx in range(len(table)):
        ch_start = table[ch_idx]
        ch_end = ch_start + window
        # 查找第一个>=ch_start的标签位置
        left = bisect.bisect_left(t_list, ch_start)
        # 查找第一个>ch_end的标签位置
        right = bisect.bisect_right(t_list, ch_end)
        # 若存在重叠标签，标记通道为True
        if left < right:
            storage[ch_idx][run] = True

复杂度对比 #

内容的提问来源于stack exchange，提问作者P. Egli