含重复/缺失ID的段数组高效轨迹重构方案问询

问题输入 #

算法输出为段列表，每个段由一对ID表示，单段归属某条轨迹，实际场景存在ID重复、ID缺失问题。以下代码可生成结构一致的示例数组：

import numpy as np
np.random.seed(42)
#ids pairs are less the further you go on the track
ids_ammount= np.arange(26,8,-2)
# the array is a 2 column array  with conected segments
ids_array = np.zeros((np.sum(ids_ammount),2))
idx_0 = 0
idx_1 = 0
for i, ids in enumerate(ids_ammount):
  idx_1+=ids
  #first column is in order from smaller to largest id
  ids_array[:,0][idx_0:idx_1] = np.sort(np.random.randint(i * 10,10*(i+1),size=ids))
  ids_array[:,1][idx_0:idx_1] = np.random.randint((i + 1) * 10,10*(i+2),size=ids)
  idx_0+= ids

目标输出 #

需要输出9类数组，分别对应含1个连接到9个连接的ID序列（k个连接对应序列包含k+1个ID），示例格式如下：

• 1个连接：single_conection = [[0,18],[67,73]...]
• 2个连接：double_conection = [[1,14,26], [53,68,79],....]
• ...
• 9个连接：nine_conections = [[3,15,28,36,41,59,66,79,82,99]....]

核心约束 #

• ID存在重复
• 部分ID不存在
• ID最大重复次数不固定

解决思路 #

这个问题本质是有向图的路径枚举问题：每个ID对是一条起始ID -> 终止ID的有向边，轨迹就是图中从入度为0的起点出发的连通路径，不需要提前统计ID重复次数，也不需要补全缺失ID。
具体流程：

1. 一次遍历所有段构建邻接表：字典的键为起始ID，值为该ID可直接连接的所有下游ID列表，同时统计每个ID的入度。
2. 提取所有轨迹起点：所有入度为0、且作为边起点存在的ID，就是每条轨迹的起始点。
3. 迭代式深度优先遍历枚举所有路径：从每个起点出发沿邻接表向下遍历，遇到分支（ID重复对应多条出边）自动拆分为多条独立路径，遇到无下游节点（ID缺失）自动终止路径，路径长度达到10（对应9个连接）时截断存储，同时增加环检测逻辑避免异常ID导致死循环。
4. 遍历过程中直接按路径的连接数（路径长度-1）归类到对应结果数组中。

实现代码 #

import numpy as np
from collections import defaultdict

def reconstruct_tracks(ids_array):
    # 构建邻接表、统计入度
    adj = defaultdict(list)
    in_degree = defaultdict(int)
    all_nodes = set()
    for u, v in ids_array:
        u, v = int(u), int(v)
        adj[u].append(v)
        in_degree[v] += 1
        all_nodes.add(u)
        all_nodes.add(v)
    
    # 提取所有起点（入度为0的节点）
    starts = [node for node in all_nodes if in_degree[node] == 0]
    # 初始化9类结果，索引i对应i+1个连接的序列
    results = [[] for _ in range(9)]

    # 迭代DFS遍历，避免递归深度溢出
    for start in starts:
        stack = [(start, [start])]
        while stack:
            current_node, path = stack.pop()
            path_len = len(path)
            # 路径达到9个连接（10个节点），直接存入对应分类
            if path_len == 10:
                results[8].append(path.copy())
                continue
            # 当前节点无下游连接，按当前长度归类
            if current_node not in adj or not adj[current_node]:
                conn_num = path_len - 1
                if 1 <= conn_num <= 9:
                    results[conn_num - 1].append(path.copy())
                continue
            # 遍历所有下游分支，处理ID重复带来的多路径
            for next_node in adj[current_node]:
                # 检测环，避免死循环
                if next_node in path:
                    conn_num = path_len - 1
                    if 1 <= conn_num <= 9:
                        results[conn_num - 1].append(path.copy())
                    continue
                new_path = path.copy()
                new_path.append(next_node)
                stack.append((next_node, new_path))
    
    return results

# 示例数据测试
np.random.seed(42)
ids_ammount= np.arange(26,8,-2)
ids_array = np.zeros((np.sum(ids_ammount),2))
idx_0 = 0
idx_1 = 0
for i, ids in enumerate(ids_ammount):
  idx_1+=ids
  ids_array[:,0][idx_0:idx_1] = np.sort(np.random.randint(i * 10,10*(i+1),size=ids))
  ids_array[:,1][idx_0:idx_1] = np.random.randint((i + 1) * 10,10*(i+2),size=ids)
  idx_0+= ids

res = reconstruct_tracks(ids_array)
single_connection, double_connection, three_connection, four_connection, five_connection, six_connection, seven_connection, eight_connection, nine_connection = res

方案优势 #

• 时间复杂度为O(N+P)，N为总段数，P为总路径数，十万级段数可在毫秒级完成计算，远高于多层嵌套判断的实现
• 不需要提前统计ID最大重复次数，重复ID带来的分支会在遍历过程中自动拆分处理
• ID缺失时路径自动终止，不需要额外的缺失值补全或判断逻辑
• 自带环检测，不会因异常ID导致死循环
• 遍历过程中直接完成结果分类，不需要二次遍历筛选

内容的提问来源于stack exchange，提问作者alejandro maza villalpando