经纬度近邻聚类：寻找100米范围内关联配送物品组

嘿，这个问题本质上是空间连通分量的求解问题——你要找的那些物品组，其实就是把距离小于100米的物品视为「连通」，然后所有能通过这种连通关系间接关联起来的物品就构成一个组（哪怕组里两个物品直接距离超100米，只要中间有其他物品搭桥就行）。下面是我整理的分步技术方案，从建模到实现都给你捋清楚：

把每个物品看成图里的一个节点，如果两个物品的实际距离小于100米，就在它们之间连一条无向边。你的需求就变成了：找出这个图里的所有连通分量——每个连通分量就是一个符合要求的物品组。

这里分三个关键环节，每个环节都有优化空间，避免暴力计算拖慢速度：

2.1 准确计算球面距离 #

因为是经纬度坐标，不能直接用平面欧氏距离（误差会很大），得用球面距离公式：

• Haversine公式：计算速度快，误差在几米内，完全满足100米阈值的需求；
• Vincenty公式：精度更高，但计算稍慢，适合对距离精度要求极高的场景。

2.2 用空间索引避免暴力遍历 #

如果物品数量N很大（比如上万甚至更多），两两计算距离的O(n²)复杂度会直接卡爆。这时候必须用空间索引来缩小计算范围：

• 网格划分：把地图切成边长100米的网格，每个物品放进对应的网格里，然后每个物品只需要和自身网格+相邻8个网格的物品计算距离，能大幅减少计算量；
• KD树/R树：用现成的空间索引库（比如Python的scipy.spatial.KDTree、rtree），快速找出每个物品周围100米内的所有邻近物品，不用遍历全量数据。

2.3 高效求解连通分量 #

当你找到所有距离小于100米的物品对后，用**并查集（Union-Find/DSU）**数据结构来计算连通分量是最高效的——它的时间复杂度接近O(n)，比BFS/DFS更适合大规模数据。

给你写个可落地的流程，包含关键代码片段：

3.1 准备数据 #

先把物品和对应的经纬度整理成结构化数据：

items = [
    {"id": "Item1", "lat": 39.9042, "lon": 116.4074},
    {"id": "Item2", "lat": 39.9045, "lon": 116.4076},
    # 更多物品...
]

3.2 实现Haversine距离计算 #

import math

def haversine_distance(lat1, lon1, lat2, lon2):
    # 转弧度
    lat1_rad = math.radians(lat1)
    lon1_rad = math.radians(lon1)
    lat2_rad = math.radians(lat2)
    lon2_rad = math.radians(lon2)
    
    dlat = lat2_rad - lat1_rad
    dlon = lon2_rad - lon1_rad
    
    a = math.sin(dlat/2)**2 + math.cos(lat1_rad) * math.cos(lat2_rad) * math.sin(dlon/2)**2
    c = 2 * math.atan2(math.sqrt(a), math.sqrt(1-a))
    # 地球半径取6371公里，转成米
    return 6371000 * c

3.3 用KDTree加速邻近查找（可选：转平面投影提升精度） #

如果你的物品集中在小范围区域，建议先把经纬度转成UTM平面坐标，这样用欧氏距离判断更准确：

from scipy.spatial import KDTree
from pyproj import Transformer

# 选对应区域的UTM投影（比如北京附近用EPSG:32650）
transformer = Transformer.from_crs("EPSG:4326", "EPSG:32650")
# 转平面坐标
projected_coords = [transformer.transform(item["lat"], item["lon"]) for item in items]

# 构建KDTree，查询每个点100米内的邻居
tree = KDTree(projected_coords)
neighbors_list = tree.query_ball_point(projected_coords, r=100)

3.4 用并查集分组 #

class UnionFind:
    def __init__(self, size):
        self.parent = list(range(size))
        self.rank = [0]*size
    
    def find(self, x):
        # 路径压缩
        if self.parent[x] != x:
            self.parent[x] = self.find(self.parent[x])
        return self.parent[x]
    
    def union(self, x, y):
        # 按秩合并
        x_root = self.find(x)
        y_root = self.find(y)
        if x_root == y_root:
            return
        if self.rank[x_root] < self.rank[y_root]:
            self.parent[x_root] = y_root
        else:
            self.parent[y_root] = x_root
            if self.rank[x_root] == self.rank[y_root]:
                self.rank[x_root] += 1

# 初始化并查集
uf = UnionFind(len(items))

# 合并所有连通的物品
for i, neighbors in enumerate(neighbors_list):
    for j in neighbors:
        if i < j:  # 避免重复合并同一对
            uf.union(i, j)

# 整理分组结果
groups = {}
for idx, item in enumerate(items):
    root = uf.find(idx)
    if root not in groups:
        groups[root] = []
    groups[root].append(item["id"])

# 过滤掉单个物品的组（如果需求不允许单个物品成组）
valid_groups = [group for group in groups.values() if len(group) > 1]

# 输出结果
for idx, group in enumerate(valid_groups, 1):
    print(f"Group {idx}: {group}")

• 投影选择：如果物品跨多个UTM分区，或者是全球范围，建议用支持球面距离的索引（比如rtree结合PostGIS），或者动态匹配UTM分区；
• 大规模数据处理：如果N达到10^5级别，Python的内存可能吃紧，建议用PostGIS数据库——它内置ST_DWithin函数找邻近点，还有ST_ClusterDBSCAN直接生成连通分组，效率更高；
• 边界确认：提前确认单个物品是否算有效组，根据需求调整最后过滤逻辑；
• 精度验证：可以拿已知距离的坐标测试距离计算函数，确保误差在可接受范围内。

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