如何将calc_pulses函数向量化为Pandas/Polars实现？

首先明确原函数的核心逻辑（假设输入time是单调递增的时间序列，这是原逻辑成立的前提）：

• 从第一个时间点开始，标记属于第1个脉冲的点（与起始点的时间差≤pgt）
• 当遇到与当前脉冲起始点的时间差>pgt+dt的点时，启动新脉冲，更新起始点为当前时间，脉冲数+1
• 处于两个脉冲之间的区间（时间差在(pgt, pgt+dt]内）的点标记为NaN

以下是完全基于numpy的向量化实现，避免显式循环，适合批量数据调用：

import numpy as np

def calc_pulses_vectorized(time, pgt, dt):
    if len(time) == 0:
        return np.array([])
    
    # 1. 标记所有脉冲起始点（第一个点默认是起始点）
    # 计算每个点对应的触发阈值：当前脉冲起始点 + pgt + dt
    thresholds = np.full(len(time), time[0] + pgt + dt)
    # 更新阈值：当当前时间超过前一个阈值时，更新为当前时间对应的新阈值
    thresholds = np.maximum.accumulate(
        np.where(
            time > np.concatenate([[np.inf], thresholds[:-1]]),
            time + pgt + dt,
            thresholds
        )
    )
    # 确定哪些点是新脉冲的起始点
    pulse_trigger = np.concatenate([[True], time[1:] > thresholds[:-1]])
    
    # 2. 计算每个点对应的脉冲编号
    pulse_num = np.cumsum(pulse_trigger)
    
    # 3. 生成每个点对应的脉冲起始时间数组
    trigger_indices = np.where(pulse_trigger)[0]
    # 计算每个脉冲覆盖的点数，用于扩展起始时间数组
    pulse_lengths = np.diff(np.concatenate([trigger_indices, [len(time)]]))
    begin_array = np.repeat(time[trigger_indices], pulse_lengths)
    
    # 4. 判断每个点是否属于当前脉冲的有效区间，生成结果
    in_pulse = time <= begin_array + pgt
    res = np.where(in_pulse, pulse_num, np.nan)
    
    return res

实现说明 #

1. 阈值更新：用np.maximum.accumulate实现动态阈值的累积更新，替代循环中的begin变量更新
2. 脉冲起始点标记：通过比较当前时间与前一个阈值，确定新脉冲的触发位置
3. 脉冲编号生成：对起始点标记数组做累积和，直接得到每个点的脉冲编号
4. 有效区间判断：生成每个点对应的脉冲起始时间，对比判断是否在有效范围内，最终生成结果

性能对比 #

相比Cython循环实现，向量化版本在处理大规模数据或多组批量数据时，能充分利用numpy的底层优化（SIMD指令、内存连续访问），速度提升显著；同时避免了Cython的编译依赖，代码更易维护。

验证一致性 #

可以通过以下代码验证向量化版本与原函数结果一致：

# 生成测试数据
time = np.linspace(0, 10, 100)
pgt = 1.0
dt = 0.5

# 调用两个版本
res_original = calc_pulses(time, pgt, dt)
res_vectorized = calc_pulses_vectorized(time, pgt, dt)

# 验证结果一致（忽略NaN的比较）
np.testing.assert_array_equal(res_original[~np.isnan(res_original)], res_vectorized[~np.isnan(res_vectorized)])

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