如何在Python中匹配含异常值的两组大规模(x,y)点集？

针对行结构点集匹配问题（含异常值、非简单平移对应关系），可以通过先按行聚类拆分，再逐行匹配的思路解决，具体步骤和代码如下：

核心思路 #

行结构点集的特点是同一行的点y坐标高度集中，不同行的y坐标差异明显。利用这一特性，先通过聚类将两组点分别拆分为独立的行，过滤掉异常值；再对每一对对应行内的点按x轴排序后做匹配，既缩小了匹配范围，又避免了全局变换（如RST）不适用的问题。

具体实现步骤 #

1. 行聚类与异常值过滤 #

使用DBSCAN密度聚类对两组点分别按y轴聚类，自动识别行结构，同时标记并过滤掉不在任何行内的异常点（噪声）。聚类后按行的平均y值排序，保证两组的行顺序对应。

2. 行内点匹配 #

对于每一对对应行：

• 若行内点数一致，直接按x轴排序后按位置一一匹配；
• 若点数不一致，用动态规划算法寻找最优匹配（最小化x坐标差异的平方和），确保匹配的合理性。

代码示例 #

import numpy as np
from sklearn.cluster import DBSCAN
import matplotlib.pyplot as plt

# 替换为你的点集数据加载逻辑，比如从文件读取
# points1 = np.loadtxt("your_points1_path.txt")
# points2 = np.loadtxt("your_points2_path.txt")

def cluster_rows(points, eps=0.5, min_samples=5):
    # 基于y坐标做DBSCAN聚类，识别行并过滤噪声
    db = DBSCAN(eps=eps, min_samples=min_samples).fit(points[:, [1]])
    valid_mask = db.labels_ != -1
    clustered_points = points[valid_mask]
    clustered_labels = db.labels_[valid_mask]
    
    # 按行的平均y值排序，保证行顺序一致
    row_groups = []
    for label in np.unique(clustered_labels):
        row_points = clustered_points[clustered_labels == label]
        row_groups.append((np.mean(row_points[:, 1]), row_points))
    row_groups.sort(key=lambda x: x[0])
    return [group[1] for group in row_groups]

# 对两组点进行行聚类
rows1 = cluster_rows(points1)
rows2 = cluster_rows(points2)

# 逐行匹配点对
matches = []
for row1, row2 in zip(rows1, rows2):
    # 行内按x坐标排序
    sorted_row1 = row1[np.argsort(row1[:, 0])]
    sorted_row2 = row2[np.argsort(row2[:, 0])]
    
    if len(sorted_row1) == len(sorted_row2):
        # 点数一致时直接按位置匹配
        matches.extend(zip(sorted_row1, sorted_row2))
    else:
        # 点数不一致时用动态规划找最优匹配
        n, m = len(sorted_row1), len(sorted_row2)
        # 构建成本矩阵，存储匹配到当前位置的最小成本
        cost_matrix = np.zeros((n+1, m+1))
        cost_matrix[1:, 0] = np.inf
        cost_matrix[0, 1:] = np.inf
        
        for i in range(1, n+1):
            for j in range(1, m+1):
                x_diff = sorted_row1[i-1, 0] - sorted_row2[j-1, 0]
                cost = x_diff ** 2
                cost_matrix[i][j] = min(
                    cost_matrix[i-1][j],    # 跳过row1的当前点
                    cost_matrix[i][j-1],    # 跳过row2的当前点
                    cost_matrix[i-1][j-1] + cost  # 匹配两个点
                )
        
        # 回溯找到匹配路径
        i, j = n, m
        while i > 0 and j > 0:
            if cost_matrix[i][j] == cost_matrix[i-1][j-1] + (sorted_row1[i-1,0]-sorted_row2[j-1,0])**2:
                matches.append((sorted_row1[i-1], sorted_row2[j-1]))
                i -= 1
                j -= 1
            elif cost_matrix[i][j] == cost_matrix[i-1][j]:
                i -= 1
            else:
                j -= 1

# 可视化匹配结果（可选）
plt.scatter(points1[:,0], points1[:,1], c='blue', label='Set 1')
plt.scatter(points2[:,0], points2[:,1], c='red', label='Set 2')
for p1, p2 in matches:
    plt.plot([p1[0], p2[0]], [p1[1], p2[1]], c='gray', linestyle='--', alpha=0.6)
plt.legend()
plt.title("Point Matching Result")
plt.show()

关键参数与优化建议 #

• 聚类参数调整：eps需根据行间距设置（比如行y差为1时，eps取0.3~0.5）；min_samples设为每行的最小点数，避免误判少量异常点为行。
• 行数不一致处理：如果两组行数不同，可先计算每行的特征（平均y、点数量、x范围），通过余弦相似度或欧氏距离找到最匹配的行对，再进行行内匹配。
• 大数量点优化：若每行点数过千，动态规划的二维成本矩阵会占用大量内存，可改为一维数组滚动更新，降低内存消耗。
• 匹配阈值过滤：可在动态规划匹配后，对匹配点对的距离设置阈值，过滤掉差异过大的匹配结果。

内容的提问来源于stack exchange，提问作者exsurge-domine