求纯Numpy优化方案：大规模频次数组的损失求和计算

你的问题核心是避开Python级别的循环/vectorize（本质还是循环），转而用Numpy的向量化操作一次性处理所有样本抽取和求和，这在100,000×1,000的大数组场景下会带来数量级的速度提升。

思路解析 #

原方法的瓶颈在于对每个元素单独调用np.random.choice，这会产生大量Python层的额外开销。我们可以换一种更高效的思路：

1. 先统计所有需要抽取的样本总数，一次性完成所有随机抽样（Numpy的批量随机操作远快于多次调用）
2. 用Numpy内置的分组求和工具，把对应频次的样本和计算出来
3. 将求和结果映射回原数组的对应位置

代码实现 #

import numpy as np

frequency = np.array( 
    [ [0, 0, 0],
      [0, 0, 0],
      [0, 0, 0],
      [0, 0, 0],
      [0, 0, 0],
      [0, 0, 0],
      [0, 0, 0],
      [0, 0, 0],
      [0, 0, 0],
      [0, 0, 1],
      [1, 2, 1],
      [1, 2, 1],
      [2, 4, 2],
      [2, 4, 2],
      [3, 5, 2],
    ] 
)
sev = np.array([1,1,2,2,1,2,3,4,5,1,1,2])

def calculate_insured_losses_numpy(frequency, severity_array):
    freq_flat = frequency.ravel()
    # 筛选非零频次并记录位置
    non_zero_mask = freq_flat > 0
    non_zero_freqs = freq_flat[non_zero_mask]
    
    if len(non_zero_freqs) == 0:
        return np.zeros_like(frequency)
    
    # 一次性抽取所有需要的样本
    total_samples = non_zero_freqs.sum()
    samples = np.random.choice(severity_array, size=total_samples, replace=True)
    
    # 计算分组分割点，用于后续求和
    split_points = np.cumsum(np.concatenate([[0], non_zero_freqs]))
    # 分组求和，得到每个非零频次对应的损失和
    non_zero_sums = np.add.reduceat(samples, split_points[:-1])
    
    # 构建结果数组并填充值
    result = np.zeros_like(freq_flat)
    result[non_zero_mask] = non_zero_sums
    return result.reshape(frequency.shape)

# 测试输出
print(calculate_insured_losses_numpy(frequency, sev))

速度优势 #

针对你的测试用例，这个方法的耗时大概在几十微秒级别（远低于原方法的291µs）；而对于100,000×1,000的大数组，原方法可能需要数分钟甚至更久，这个纯Numpy方法能在几秒内完成——因为所有核心操作都是Numpy底层的C实现，完全避开了Python循环的开销。

关键细节说明 #

• np.add.reduceat：Numpy原生的高效分组求和工具，比手动切片求和的效率高很多
• 批量抽样：一次性生成所有随机样本，避免了多次调用np.random.choice的额外开销
• 掩码处理：只处理非零频次，进一步减少不必要的计算量

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