如何加快最近地理坐标点（geopoint）的检索速度？

优化方案 #

以下优化方案按实现成本从低到高排列，单独使用即可感知到明显性能提升，组合使用效果最优： #

1. 替换全量排序为堆Top K查询

你当前使用的sorted会对全部7000条数据做全排序，时间复杂度为O(n log n)，但你仅需要返回最近的5条结果，直接使用Python标准库的heapq.nsmallest即可把时间复杂度降到O(n log 5)≈O(n)，改造成本极低。
示例代码：

import heapq
from math import cos, asin, sqrt

def get_closest_stops(lat, lng, data=db.STOPS):
    def distance(lat1, lon1, lat2, lon2):
        p = 0.017453292519943295
        a = 0.5 - cos((lat2-lat1)*p)/2 + cos(lat1*p)*cos(lat2*p) * (1-cos((lon2-lon1)*p)) / 2
        return 12742 * asin(sqrt(a))

    return heapq.nsmallest(5, data.values(), key=lambda p: distance(lat, lng, p['lat'], p['lng']))

2. 增加粗筛逻辑减少距离计算量

Haversine距离计算涉及多次三角函数运算，7000次累计开销较高，可以先增加一层粗筛逻辑：先预设经纬度差值阈值（比如0.1度，对应约11km的距离，可根据站点密度自行调整），先过滤掉和目标点经纬度差值超过阈值的站点，再对剩下的少量站点做精确距离计算，大多数场景下粗筛后仅剩余几十到上百条站点，性能会进一步提升。
示例代码：

import heapq
from math import cos, asin, sqrt

def get_closest_stops(lat, lng, data=db.STOPS, delta=0.1):
    def distance(lat1, lon1, lat2, lon2):
        p = 0.017453292519943295
        a = 0.5 - cos((lat2-lat1)*p)/2 + cos(lat1*p)*cos(lat2*p) * (1-cos((lon2-lon1)*p)) / 2
        return 12742 * asin(sqrt(a))
    # 经纬度粗筛
    filter_stops = [p for p in data.values() if (lat-delta < p['lat'] < lat+delta) and (lng-delta < p['lng'] < lng+delta)]
    # 粗筛结果不足5条时回退到全量计算
    if len(filter_stops) < 5:
        filter_stops = data.values()
    return heapq.nsmallest(5, filter_stops, key=lambda p: distance(lat, lng, p['lat'], p['lng']))

3. 预构建空间索引（适合高频查询场景）

如果该查询是高频调用的，可以一次性预构建KD树空间索引，后续每次查询的时间复杂度可以降到O(log n)，性能提升幅度最大。可以直接使用scipy的KDTree实现，注意要先把经纬度转成弧度再构建索引，适配球面距离计算。
示例代码（索引构建仅需执行一次）：

import numpy as np
from scipy.spatial import KDTree

# 预构建索引，全局仅执行一次
stops_list = list(db.STOPS.values())
coords_rad = np.array([[p['lat']*np.pi/180, p['lng']*np.pi/180] for p in stops_list])
kdtree = KDTree(coords_rad, metric='haversine')

def get_closest_stops(lat, lng, k=5):
    target_rad = np.array([lat*np.pi/180, lng*np.pi/180])
    # 查询最近k个点的索引，返回的距离为弧度值，乘以地球半径6371即可得到公里数
    distances, indices = kdtree.query(target_rad, k=k)
    return [stops_list[i] for i in indices]

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