如何在指派问题（匈牙利算法）中禁用超时时差的事件匹配？

背景与动机 #

我正在使用scipy库中匈牙利算法的Python实现来匹配两组时间事件。目前整体流程可行，但我设置了时差限制（例如0.1秒），不希望出现任何时差超过该限制的匹配结果。

无时差限制的示例 #

from dataclasses import dataclass
from scipy.optimize import linear_sum_assignment

@dataclass(frozen=True)
class Event:
    t: float

expected = [Event(0), Event(1), Event(2)]
detected = [          Event(1), Event(2), Event(3)] # shifted to show
                                                    # time matching

def tdiff(e1: Event, e2: Event) -> float:
    return abs(e1.t - e2.t)

cost_matrix = [
    [
        tdiff(e1, e2)
        for e2 in detected
    ]
    for e1 in expected
]
print('cost matrix:')
for row in cost_matrix:
    print('[', ', '.join(map(str, row)), ']')

row_idx, col_idx = linear_sum_assignment(cost_matrix)
for i, j in zip(row_idx, col_idx):
    exp_ev = expected[i]
    det_ev = detected[j]
    td = cost_matrix[i][j]
    print('matches %s and %s, tdiff: %.3f' % (exp_ev, det_ev, td))

输出结果如下：

cost matrix:
[ 1, 2, 3 ]
[ 0, 1, 2 ]
[ 1, 0, 1 ]
matches Event(t=0) and Event(t=1), tdiff: 1.000
matches Event(t=1) and Event(t=2), tdiff: 1.000
matches Event(t=2) and Event(t=3), tdiff: 1.000

显然这不符合我的需求，因为所有匹配的时差都超过了0.1秒的限制。

已尝试的方案 #

我的解决方案是人为提高“非期望”匹配在成本矩阵中的成本，之后再过滤得到仅包含“期望”匹配的结果：

from dataclasses import dataclass
from scipy.optimize import linear_sum_assignment

@dataclass(frozen=True)
class Event:
    t: float

TDIFF_MAX = 0.1
expected = [Event(0), Event(1), Event(2)]
detected = [          Event(1), Event(2), Event(3)] # shifted to show
                                                    # time matching

def tdiff(e1: Event, e2: Event) -> float:
    return abs(e1.t - e2.t)

orig_cost_matrix = [
    [
        tdiff(e1, e2)
        for e2 in detected
    ]
    for e1 in expected
]
orig_max = max(map(max, orig_cost_matrix))

print('original cost matrix:')
for row in orig_cost_matrix:
    print('[', ', '.join(map(str, row)), ']')

cost_matrix = [
    [
        x if x <= TDIFF_MAX else orig_max + 1
        for x in row
    ]
    for row in orig_cost_matrix
]
print('modified cost matrix:')
for row in cost_matrix:
    print('[', ', '.join(map(str, row)), ']')

row_idx, col_idx = linear_sum_assignment(cost_matrix)
for i, j in zip(row_idx, col_idx):
    exp_ev = expected[i]
    det_ev = detected[j]
    td = cost_matrix[i][j]
    if td <= TDIFF_MAX:
        print('matches %s and %s, tdiff: %.3f' % (exp_ev, det_ev, td))

输出结果：

original cost matrix:
[ 1, 2, 3 ]
[ 0, 1, 2 ]
[ 1, 0, 1 ]
modified cost matrix:
[ 4, 4, 4 ]
[ 0, 4, 4 ]
[ 4, 0, 4 ]
matches Event(t=1) and Event(t=1), tdiff: 0.000
matches Event(t=2) and Event(t=2), tdiff: 0.000

该方案可行，但我存在两点顾虑：

• 代码不够直观，可读性较差
• 在实际场景中，expected和detected数据集可能非常庞大，且由于时差限制很小，大部分匹配都是“非期望”的，这会导致方案效率低下，浪费大量CPU资源

问题 #

是否存在更优的方法来“禁用”某些可能的匹配（不一定局限于匈牙利算法），尤其适用于每个Event仅存在1、2或3个可能匹配的大数据场景？

我主要关注算法逻辑，欢迎提供其他语言的实现或伪代码。

更优解决方案 #

针对你的场景（每个Event仅1-3个可行匹配），推荐两种高效且直观的思路：

1. 排序+贪心匹配（适合时间有序场景）

如果expected和detected序列是按时间戳严格递增排序的，这是最优方案，时间复杂度O(n+m)：

• 用双指针遍历两个序列，只保留时差符合要求的匹配
• 逻辑简单，代码可读性极强，完全不需要构建成本矩阵

伪代码：

TDIFF_MAX = 0.1
i = 0, j = 0
matches = []
while i < len(expected) and j < len(detected):
    diff = abs(expected[i].t - detected[j].t)
    if diff <= TDIFF_MAX:
        matches.append( (expected[i], detected[j]) )
        i += 1
        j += 1
    elif expected[i].t < detected[j].t - TDIFF_MAX:
        # 当前预期事件已过，没有可用检测事件匹配
        i += 1
    else:
        # 当前检测事件已过，没有可用预期事件匹配
        j += 1

Python实现示例：

from dataclasses import dataclass

@dataclass(frozen=True)
class Event:
    t: float

TDIFF_MAX = 0.1
expected = [Event(0), Event(1), Event(2)]
detected = [Event(1), Event(2), Event(3)]

i = j = 0
matches = []
while i < len(expected) and j < len(detected):
    diff = abs(expected[i].t - detected[j].t)
    if diff <= TDIFF_MAX:
        matches.append((expected[i], detected[j]))
        i += 1
        j += 1
    elif expected[i].t < detected[j].t - TDIFF_MAX:
        i += 1
    else:
        j += 1

for exp_ev, det_ev in matches:
    print(f'matches {exp_ev} and {det_ev}, tdiff: %.3f' % abs(exp_ev.t - det_ev.t))

2. 稀疏二分图+全局最优匹配（适合无序场景）

如果事件序列无序，或者需要全局最优匹配而非局部贪心结果，可以构建仅包含有效匹配的稀疏二分图，再用支持稀疏输入的匹配算法：

• 只保留时差≤0.1秒的节点对，避免构建全量稠密矩阵
• 用networkx的最大权重匹配（转成本为负权重）或专门的稀疏匈牙利算法实现，效率远高于全量矩阵方案

Python实现示例（networkx）：

from dataclasses import dataclass
import networkx as nx

@dataclass(frozen=True)
class Event:
    t: float

TDIFF_MAX = 0.1
expected = [Event(0), Event(1), Event(2)]
detected = [Event(1), Event(2), Event(3)]

# 构建二分图，区分两组节点
G = nx.Graph()
for idx, e in enumerate(expected):
    G.add_node(f"exp_{idx}", event=e, type="expected")
for idx, d in enumerate(detected):
    G.add_node(f"det_{idx}", event=d, type="detected")

# 仅添加符合时差要求的边，权重为负时差（networkx求最大权重对应最小成本）
for exp_idx, e in enumerate(expected):
    for det_idx, d in enumerate(detected):
        diff = abs(e.t - d.t)
        if diff <= TDIFF_MAX:
            G.add_edge(f"exp_{exp_idx}", f"det_{det_idx}", weight=-diff)

# 求解最大权重匹配（保证全局最优）
matching = nx.max_weight_matching(G, maxcardinality=True)

# 解析匹配结果
for node1, node2 in matching:
    if G.nodes[node1]['type'] == 'expected':
        exp_ev, det_ev = G.nodes[node1]['event'], G.nodes[node2]['event']
    else:
        exp_ev, det_ev = G.nodes[node2]['event'], G.nodes[node1]['event']
    print(f'matches {exp_ev} and {det_ev}, tdiff: %.3f' % abs(exp_ev.t - det_ev.t))

方案对比

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