Python速度计算优化：大数据集下最大速度计算性能提升咨询

我写了一个Python脚本，用两个列表分别存储测量距离和耗时，用来计算特定距离区间或时间窗口内的最大速度（比如2秒、10秒时间窗口，100米、1公里距离区间）。

当前实现是遍历每个测量点，找到当前点往后满足指定距离/时间的终点，再计算速度。数据量小时没问题，但数据量大时计算要花好几分钟，同类闭源软件却只需要几秒，求性能优化方案。

当前代码 #

distance=[1,2,6,8,13,15,18,24,27,34,37,38,44,47,48,49]
time=[1,2,3,4,5,7,8,9,10,11,12,13,15,16,17,18]


def calculate(arr,unit,type):
    pointer=-1
    max=0
    for xy in arr:
        pointer+=1
        try:        
            if type=='distance':
                 nxt=next(x[0] for x in enumerate(distance) if x[1] > (distance[pointer]+unit))
            else:
                 nxt=next(x[0] for x in enumerate(time) if x[1] > (time[pointer]+unit))

            vald=distance[nxt]-distance[pointer]
            valc=time[nxt]-time[pointer]
            value=round((vald/valc)*1.94384449,2)

            if value > max and value < 45:
                 max=value
        except:
            pass

        return(max)

print(calculate(distance,2,"time"))
print(calculate(distance,10,"time"))
print(calculate(distance,10,"distance")) 

性能优化方案 #

核心优化思路：双指针替代暴力遍历 #

原代码每次找终点都要从头枚举，时间复杂度是O(n²)，数据量大时必然卡顿。利用distance和time都是递增序列的特性，用双指针可以把时间复杂度降到O(n)，每个元素最多被访问一次。

优化后的代码 #

distance = [1,2,6,8,13,15,18,24,27,34,37,38,44,47,48,49]
time = [1,2,3,4,5,7,8,9,10,11,12,13,15,16,17,18]
SPEED_CONVERSION = 1.94384449  # 提前定义速度转换常量

def calculate_max_speed(unit, window_type):
    max_speed = 0.0
    n = len(distance)
    right_ptr = 0  # 双指针的右指针，跟随左指针单向移动
    
    for left_ptr in range(n):
        # 根据窗口类型移动右指针，直到满足区间条件或到达数组末尾
        if window_type == 'distance':
            target = distance[left_ptr] + unit
            while right_ptr < n and distance[right_ptr] <= target:
                right_ptr += 1
        else:  # 时间窗口类型
            target = time[left_ptr] + unit
            while right_ptr < n and time[right_ptr] <= target:
                right_ptr += 1
        
        # 跳过右指针越界的情况
        if right_ptr >= n:
            continue
        
        # 计算速度，避免除以0
        delta_dist = distance[right_ptr] - distance[left_ptr]
        delta_time = time[right_ptr] - time[left_ptr]
        if delta_time <= 0:
            continue
        
        speed = round((delta_dist / delta_time) * SPEED_CONVERSION, 2)
        if speed > max_speed and speed < 45:
            max_speed = speed
    
    return max_speed

print(calculate_max_speed(2, "time"))
print(calculate_max_speed(10, "time"))
print(calculate_max_speed(10, "distance"))

其他优化细节 #

• 移除冗余参数：原函数的arr参数未实际使用，直接删除，让逻辑更清晰
• 替换异常处理：用边界判断替代try-except，减少异常捕获的性能开销
• 提前定义常量：把速度转换系数提取为全局常量，避免循环内重复计算
• 增加除以0防护：避免极端情况下出现除零错误

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