如何统计指定区间在min/max数据集内的出现次数?
高效统计区间在min/max数组中的出现次数优化方案
我需要统计一组自定义区间(由lb和ub生成的(x,y)对,其中y>x)在min/max数据集的出现情况,并计算对应的project值。初始实现用三重循环,效率偏低,希望找到更优的向量化实现方式。
测试数据生成代码
import numpy as np; import pandas as pd np.random.seed(42) df = pd.DataFrame(np.random.rand(1000, 2), columns=["Min", "Max"]) \ .mul(6).query("Min <= Max").reset_index(drop=True) lb = np.arange(1, 5, 0.1) ub = np.arange(1, 5, 0.1) a = df[['Min','Max']].to_numpy() factor = 1
初始低效实现(三重循环)
arry = [] for row in a: for x in lb: for y in ub: lbExists = 0 rangeExists = 0 if y <= x: continue if x >= row[0] and x <= row[1]: lbExists = 1 if x >= row[0] and y <= row[1]: rangeExists = 1 project = 0 if lbExists == 1 and rangeExists == 0: project -= factor if lbExists == 1 and rangeExists == 1: project = (y - x) * factor arry.append([round(x, 2), round(y, 2), row[0], row[1], lbExists, rangeExists , round(project,2)])
截断输出示例
[[1.0, 1.1, 1.34, 5.8, 0, 0, 0], [1.0, 1.2, 1.34, 5.8, 0, 0, 0], ...
向量化优化实现
利用numpy的广播机制,将三重循环转化为批量矩阵运算,大幅降低Python循环的开销,提升运行效率。
代码实现
# 1. 生成所有有效(x,y)组合,过滤y<=x的情况 x_grid, y_grid = np.meshgrid(lb, ub) mask = y_grid > x_grid x_valid = x_grid[mask].reshape(-1, 1) # 形状: (有效组合数, 1) y_valid = y_grid[mask].reshape(-1, 1) # 形状: (有效组合数, 1) # 2. 批量计算lbExists和rangeExists row_min = a[:, 0].reshape(1, -1) # 形状: (1, 数据集行数) row_max = a[:, 1].reshape(1, -1) # 形状: (1, 数据集行数) # 广播运算:每个(x,y)和所有数据集区间做批量比较 lbExists = ((x_valid >= row_min) & (x_valid <= row_max)).astype(int) # 形状: (有效组合数, 数据集行数) rangeExists = ((x_valid >= row_min) & (y_valid <= row_max)).astype(int) # 形状: (有效组合数, 数据集行数) # 3. 批量计算project值 project = np.zeros_like(lbExists, dtype=np.float64) # 条件1:lbExists=1且rangeExists=0 → project = -factor mask1 = (lbExists == 1) & (rangeExists == 0) project[mask1] = -factor # 条件2:lbExists=1且rangeExists=1 → project = (y-x)*factor mask2 = (lbExists == 1) & (rangeExists == 1) project[mask2] = (y_valid[mask2] - x_valid[mask2]) * factor # 4. 整理成和原循环一致的输出结构 x_repeat = np.repeat(x_valid, a.shape[0], axis=0).flatten() y_repeat = np.repeat(y_valid, a.shape[0], axis=0).flatten() row_min_repeat = np.tile(row_min, x_valid.shape[0]).flatten() row_max_repeat = np.tile(row_max, x_valid.shape[0]).flatten() lbExists_flat = lbExists.flatten() rangeExists_flat = rangeExists.flatten() project_flat = project.flatten() # 四舍五入对齐原输出格式 result = np.column_stack([ np.round(x_repeat, 2), np.round(y_repeat, 2), row_min_repeat, row_max_repeat, lbExists_flat, rangeExists_flat, np.round(project_flat, 2) ])
统计每个区间的出现次数
如果需要统计每个(x,y)区间在数据集中的符合条件次数,可直接对结果按行求和:
# 统计每个(x,y)的lbExists=1的次数 lb_counts = lbExists.sum(axis=1) # 统计每个(x,y)的rangeExists=1的次数 range_counts = rangeExists.sum(axis=1) # 整理成DataFrame方便查看 stats_df = pd.DataFrame({ 'x': np.round(x_valid.flatten(), 2), 'y': np.round(y_valid.flatten(), 2), 'lb_exists_count': lb_counts, 'range_exists_count': range_counts })
效率说明
原三重循环时间复杂度为O(RLU)(R为数据集行数,L/U为lb/ub长度),总运算量约1.6e6次。向量化实现利用numpy的C级底层运算,避免了Python循环的额外开销,实际运行速度可提升10~100倍(取决于硬件配置)。
内容的提问来源于stack exchange,提问作者Jared King
相关产品推荐
相关产品推荐

