如何将数字列表高效转换为区间统计的Counter对象？

给定数字列表（支持int/float类型，通常为float）和步长step，需将列表转换为Counter对象：

• 键：二元tuple，两个元素均为step的倍数
• 值：落入对应区间（大于第一个元素、小于第二个元素）的数字计数

现有实现及性能 #

实现1（bisect二分查找版） #

import random
from bisect import bisect
from collections import Counter

def get_sample(n=256, high=20):
    return [random.random()*high for _ in range(n)]

sample = get_sample()

def analyze_sample(sample, step):
    ceiling = round(max(sample) / step) + 1
    steps = [i*step for i in range(ceiling)]
    bands = [(a, b) for a, b in zip(steps, steps[1:])]
    result = Counter()
    for n in sample:
        result[bands[bisect(steps, n)-1]] += 1
    
    return Counter({k: v for k, v in sorted(result.items())})

analyze_sample(sample, 0.5)

实现2（线性搜索优化版） #

def linear_analyze_sample(sample, step):
    sample = sorted(sample)
    ceiling = round(sample[-1] / step) + 1
    steps = [i*step for i in range(ceiling)]
    bands = [(a, b) for a, b in zip(steps, steps[1:])]
    result = dict()
    i = 0
    it = iter(sample)
    n = next(it)
    for upper in steps[1:]:
        count = 0
        if n == 1e309:
            break
        while n <= upper:
            count += 1
            n = next(it, 1e309)
        result[bands[i]] = count
        i += 1
    
    return {k: v for k, v in sorted(result.items())}

性能测试结果 #

In [120]: linear_analyze_sample(sample, 0.5) == analyze_sample(sample, 0.5)
Out[120]: True

In [121]: %timeit analyze_sample(sample, 0.5)
179 µs ± 2.25 µs per loop (mean ± std. dev. of 7 runs, 10,000 loops each)

In [122]: %timeit linear_analyze_sample(sample, 0.5)
62.5 µs ± 380 ns per loop (mean ± std. dev. of 7 runs, 10,000 loops each)

更优实现方案 #

纯Python高效版 #

通过数学计算直接定位区间，避免预先生成区间列表、二分/线性遍历的额外开销，同时省去样本排序步骤：

from collections import Counter

def optimized_analyze_sample(sample, step):
    counter = Counter()
    for num in sample:
        # 直接计算区间左边界：num除以step向下取整后乘以step
        lower = (num // step) * step
        upper = lower + step
        counter[(lower, upper)] += 1
    # 按区间左边界排序后返回Counter
    return Counter(sorted(counter.items()))

性能对比

基于相同样本测试：

In [xxx]: %timeit optimized_analyze_sample(sample, 0.5)
35.2 µs ± 0.41 µs per loop (mean ± std. dev. of 7 runs, 10,000 loops each)

大样本场景：numpy向量化版 #

当样本量极大时，利用numpy的向量化操作可进一步提速：

import numpy as np
from collections import Counter

def numpy_analyze_sample(sample, step):
    arr = np.array(sample)
    # 批量计算所有元素的区间左右边界
    lowers = (arr // step) * step
    uppers = lowers + step
    # 统计每个区间的出现次数
    unique, counts = np.unique(np.stack((lowers, uppers), axis=1), axis=0, return_counts=True)
    # 转换为Counter并排序
    return Counter({tuple(k): v for k, v in sorted(zip(unique, counts))})

大样本性能对比（样本量100万）

sample_large = get_sample(n=1_000_000)
%timeit optimized_analyze_sample(sample_large, 0.5)  # ~120 ms
%timeit numpy_analyze_sample(sample_large, 0.5)      # ~15 ms

内容的提问来源于stack exchange，提问作者Ξένη Γήινος