在Python中计算重叠滑动窗口中的值

Question

给定一个已排序的值数组 a 和一个范围数组 bins，如何最有效地计算在每个范围 rng 中有多少个 a 值？

目前我正在执行以下操作：

def sliding_count(a, end, window, start=0, step=1):
    bins = [(x, x + window) for x in range(start, (end + 1) - window, step)]
    counts = np.zeros(len(bins))
    for i, rng in enumerate(bins):
        count = len(a[np.where(np.logical_and(a>=rng[0], a<=rng[1]))])
        counts[i] = count
    return counts

a = np.array([1, 5, 8, 11, 14, 19])
end = 20
window = 10
sliding_count(a, end, window)

这将返回预期的数组

array([3., 4., 3., 3., 4., 4., 3., 3., 3., 3., 3.])

但是我感觉一定有更有效的方法来做这件事吧？

- Michael Hall

---

在我看来，这是相当有效的。你是否正在寻找更高效的方法？ - JE_Muc

你知道吗，range返回一个对象，你可以在其上执行x in range(...)并获得真/假的结果。 - Nullman

@Nullman 是的，我试过了，但这意味着我必须为 a 中的每个元素迭代一次 bins。 - Michael Hall

窗口的大小是否总是步长的倍数？ - Mad Physicist

Answer 1

import numpy as np

def alt(a, end, window, start=0, step=1):
    bin_starts = np.arange(start, end+1-window, step)
    bin_ends = bin_starts + window
    last_index = np.searchsorted(a, bin_ends, side='right')
    first_index = np.searchsorted(a, bin_starts, side='left')
    return  last_index - first_index

def sliding_count(a, end, window, start=0, step=1):
    bins = [(x, x + window) for x in range(start, (end + 1) - window, step)]
    counts = np.zeros(len(bins))
    for i, rng in enumerate(bins):
        count = len(a[np.where(np.logical_and(a>=rng[0], a<=rng[1]))])
        counts[i] = count
    return counts

a = np.array([1, 5, 8, 11, 14, 19])
end = 20
window = 10

print(sliding_count(a, end, window))
# [3. 4. 3. 3. 4. 4. 3. 3. 3. 3. 3.]

print(alt(a, end, window))
# [3 4 3 3 4 4 3 3 3 3 3]

---

alt的工作原理：

生成区间(bin)的起始值和结束值:

In [73]: bin_starts = np.arange(start, end+1-window, step); bin_starts
Out[73]: array([ 0,  1,  2,  3,  4,  5,  6,  7,  8,  9, 10])

In [74]: bin_ends = bin_starts + window; bin_ends
Out[74]: array([10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20])

由于a已经按顺序排列，因此可以使用np.searchsorted来查找每个值在bin_starts和bin_ends中的第一个和最后一个索引位置：

In [75]: last_index = np.searchsorted(a, bin_ends, side='right'); last_index
Out[75]: array([3, 4, 4, 4, 5, 5, 5, 5, 5, 6, 6])

In [76]: first_index = np.searchsorted(a, bin_starts, side='left'); first_index
Out[76]: array([0, 0, 1, 1, 1, 1, 2, 2, 2, 3, 3])

count 就是索引之间的差值：

In [77]: last_index - first_index
Out[77]: array([3, 4, 3, 3, 4, 4, 3, 3, 3, 3, 3])

---

这里是一个perfplot，比较了alt和sliding_count在a长度的函数中的性能：

import perfplot

def make_array(N):
    a = np.random.randint(10, size=N)
    a = a.cumsum()
    return a

def using_sliding(a):
    return sliding_count(a, end, window)

def using_alt(a):
    return alt(a, end, window)

perfplot.show(
    setup=make_array,
    kernels=[using_sliding, using_alt],
    n_range=[2**k for k in range(22)],
    logx=True,
    logy=True,
    xlabel='len(a)')

Perfplot还会检查using_sliding返回的值是否等于using_alt返回的值。

Matt Timmermans提出的想法，即“从该bin的计数中减去position_in_a”，启发了这个解决方案。

- unutbu

Answer 2

一个bin中的元素个数是指小于等于b.end的元素个数减去小于b.start的元素个数。

因此，您可以创建一个按起始位置排序的“starts”数组和一个按结束位置排序的“ends”数组。然后按步骤遍历所有3个数组。当您超过a中的每个x时，将通过x < b.start前进并从该bin的计数中“减去”position_in_a。然后在通过x <= b.end后进并将position_in_a“添加”到该bin的计数中。

总复杂度为O(N log N)，主要由排序开始和结束数组决定。遍历3个数组并调整计数的时间复杂度为O(N)。

在您的代码中，您已经生成了已排序的箱子数组，因此如果您可以这样做，那么您可以跳过排序步骤，总复杂度为O（a.length+bin_count）。我甚至不会费心去生成该数组，因为您可以轻松地从索引计算出起始值和结束值。

- Matt Timmermans

Answer 3

类似这样的东西（？）：

def sliding_count(a, nx0, nx1, window):
    bin0 = np.arange(nx0,nx1,1)
    bin1 = bin0 + window 
    count = np.zeros((nx1-nx0), dtype=int)

    for j in range(nx1-nx0):
        count[j] = np.sum(a<=bin1[j]) - np.sum(a<bin0[j])
    return count

#---- main ---------------  
nx0, nx1, window = 0, 11, 10
a = np.array([1, 5, 8, 11, 14, 19])
sliding_count(a, nx0, nx1, window)

array([3, 4, 3, 3, 4, 4, 3, 3, 3, 3, 3])

我没有检查bin0 = np.arange(nx0,nx1,1)中的nx0>0和step>1的代码。因此，对于这种情况，for循环的长度必须进行修改。

- pyano