加速循环以用另一个数组中最接近的值填充数组

Gabriel 发表于 Dev

加布里埃尔

我有一段代码需要优化，因为我必须运行数千次。

它的作用是在给定数组的子列表中为随机浮点数找到最接近的浮点数，并将相应的浮点数（即具有相同索引）存储在该数组的另一个子表中。重复该过程，直到存储的浮点数的总和达到某个限制。

这是MWE为了使其更清楚：

import numpy as np

# Define array with two sub-lists.
a = [np.random.uniform(0., 100., 10000), np.random.random(10000)]

# Initialize empty final list.
b = []

# Run until the condition is met.
while (sum(b) < 10000):

    # Draw random [0,1) value.
    u = np.random.random()
    # Find closest value in sub-list a[1].
    idx = np.argmin(np.abs(u - a[1]))
    # Store value located in sub-list a[0].
    b.append(a[0][idx])

该代码相当简单，但是我还没有找到一种加速它的方法。我试图使我前段时间提出的类似问题中给出的出色（且非常快）的答案无济于事。

ali_m

好的，这是一个稍微偏左的建议。据我了解，您只是尝试从中的元素进行统一采样，a[0]直到列表总数超过某个限制为止。

尽管在内存方面会花费更多，但我认为您可能会发现，从头a[0]开始生成大量随机样本，然后求和，并首先发现超出限制的位置会更快。

例如：

import numpy as np

# array of reference float values, equivalent to a[0]
refs = np.random.uniform(0, 100, 10000)

def fast_samp_1(refs, lim=10000, blocksize=10000):

    # sample uniformally from refs
    samp = np.random.choice(refs, size=blocksize, replace=True)
    samp_sum = np.cumsum(samp)

    # find where the cumsum first exceeds your limit
    last = np.searchsorted(samp_sum, lim, side='right')
    return samp[:last + 1]

    # # if it's ok to be just under lim rather than just over then this might
    # # be quicker
    # return samp[samp_sum <= lim]

当然，如果blocksize元素样本的总和为<lim，那么将无法为您提供总和> = lim的样本。您可以检查是否存在这种情况，并在必要时循环添加到样本中。

def fast_samp_2(refs, lim=10000, blocksize=10000):

    samp = np.random.choice(refs, size=blocksize, replace=True)
    samp_sum = np.cumsum(samp)

    # is the sum of our current block of samples >= lim?
    while samp_sum[-1] < lim:

        # if not, we'll sample another block and try again until it is
        newsamp = np.random.choice(refs, size=blocksize, replace=True)
        samp = np.hstack((samp, newsamp))
        samp_sum = np.hstack((samp_sum, np.cumsum(newsamp) +  samp_sum[-1]))

    last = np.searchsorted(samp_sum, lim, side='right')
    return samp[:last + 1]

请注意，连接数组的速度相当慢，因此，最好使其blocksize足够大以合理地确保单个块的总和> =到您的限制而不会太大。

更新资料

我对您的原始功能做了一些调整，以使其语法与我的更加相似。

def orig_samp(refs, lim=10000):

    # Initialize empty final list.
    b = []

    a1 = np.random.random(10000)

    # Run until the condition is met.
    while (sum(b) < lim):

        # Draw random [0,1) value.
        u = np.random.random()
        # Find closest value in sub-list a[1].
        idx = np.argmin(np.abs(u - a1))
        # Store value located in sub-list a[0].
        b.append(refs[idx])

    return b

这是一些基准数据。

%timeit orig_samp(refs, lim=10000)
# 100 loops, best of 3: 11 ms per loop

%timeit fast_samp_2(refs, lim=10000, blocksize=1000)
# 10000 loops, best of 3: 62.9 µs per loop

这快了三个数量级。您可以通过减小块大小来提高性能-您基本上希望它比要取出的数组的长度舒适地更大。在这种情况下，您知道输出的平均长度约为200个元素，因为0到100之间的所有实数的平均值为50，并且10000/50 = 200。

更新2

获得加权样本而不是均匀样本很容易-您可以将p=参数传递给np.random.choice：

def weighted_fast_samp(refs, weights=None, lim=10000, blocksize=10000):

    samp = np.random.choice(refs, size=blocksize, replace=True, p=weights)
    samp_sum = np.cumsum(samp)

    # is the sum of our current block of samples >= lim?
    while samp_sum[-1] < lim:

        # if not, we'll sample another block and try again until it is
        newsamp = np.random.choice(refs, size=blocksize, replace=True, 
                                   p=weights)
        samp = np.hstack((samp, newsamp))
        samp_sum = np.hstack((samp_sum, np.cumsum(newsamp) +  samp_sum[-1]))

    last = np.searchsorted(samp_sum, lim, side='right')
    return samp[:last + 1]