NumPy 数组中滑动窗口中的最大值答案

【问题标题】：Max in a sliding window in NumPy arrayNumPy 数组中滑动窗口中的最大值
【发布时间】：2017-09-03 10:52:52
【问题描述】：

我想创建一个数组，该数组包含在给定 numpy 数组中移动的窗口的所有 max()es。如果这听起来令人困惑，我很抱歉。我举个例子。输入：

[ 6,4,8,7,1,4,3,5,7,2,4,6,2,1,3,5,6,3,4,7,1,9,4,3,2 ]

我的窗口宽度为 5 的输出应该是这样的：

[     8,8,8,7,7,7,7,7,7,6,6,6,6,6,6,7,7,9,9,9,9     ]

每个数字应为输入数组宽度为 5 的子数组的最大值：

[ 6,4,8,7,1,4,3,5,7,2,4,6,2,1,3,5,6,3,4,7,1,9,4,3,2 ]
  \       /                 \       /
   \     /                   \     /
    \   /                     \   /
     \ /                       \ /
[     8,8,8,7,7,7,7,7,7,6,6,6,6,6,6,7,7,9,9,9,9     ]

我没有在 numpy 中找到一个开箱即用的函数来执行此操作（但如果有一个函数，我不会感到惊讶；我并不总是按照 numpy 开发人员的想法来思考）。我考虑过为我的输入创建一个移动的 2D 版本：

[ [ 6,4,8,7,1,4,3,5,7,8,4,6,2,1,3,5,6,3,4,7,1 ]
  [ 4,8,7,1,4,3,5,7,8,4,6,2,1,3,5,6,3,4,7,1,9 ]
  [ 8,7,1,4,3,5,7,8,4,6,2,1,3,5,6,3,4,7,1,9,4 ]
  [ 7,1,4,3,5,7,8,4,6,2,1,3,5,6,3,4,7,1,9,4,3 ]
  [ 1,4,3,5,7,8,4,6,2,1,3,5,6,3,4,7,1,9,4,3,2 ] ]

然后我可以在此应用np.max(input, 0) 并得到我的结果。但这在我的情况下似乎效率不高，因为我的数组和窗口宽度都可能很大（>1000000 个条目和>100000 个窗口宽度）。数据或多或少会被窗口宽度放大。

我也考虑过以某种方式使用np.convolve()，但无法找到实现目标的方法。

任何想法如何有效地做到这一点？

【问题讨论】：

标签： python performance numpy scipy max

【解决方案1】：

方法#1：你可以使用1D max filter from Scipy -

from scipy.ndimage.filters import maximum_filter1d

def max_filter1d_valid(a, W):
    hW = (W-1)//2 # Half window size
    return maximum_filter1d(a,size=W)[hW:-hW]

方法#2：这是strides 的另一种方法：strided_app 创建一个2D 转换版本作为数组视图非常有效，这应该让我们可以使用任何自定义归约操作之后沿着第二个轴 -

def max_filter1d_valid_strided(a, W):
    return strided_app(a, W, S=1).max(axis=1)

运行时测试-

In [55]: a = np.random.randint(0,10,(10000))

# @Abdou's solution using pandas rolling
In [56]: %timeit pd.Series(a).rolling(5).max().dropna().tolist()
1000 loops, best of 3: 999 µs per loop

In [57]: %timeit max_filter1d_valid(a, W=5)
    ...: %timeit max_filter1d_valid_strided(a, W=5)
    ...: 
10000 loops, best of 3: 90.5 µs per loop
10000 loops, best of 3: 87.9 µs per loop

【讨论】：

这听起来很有希望，将性能与pandas 解决方案进行比较。不幸的是，对于我正在处理的数组，这引发了ValueError: array is too big.。亲自尝试：a = np.arange(1000000)、np.lib.stride_tricks.as_strided(a, shape=(1000, len(a)-1000+1), strides=(a.strides[0], a.strides[0]))。在实践中，我将需要大小为 10m 或更大的数组中大小为 100k 的窗口。你有什么解决办法吗？
@Alfe 只需使用他提出的scipy.ndimage.maximum_filter1d 方法即可。它几乎一样快，即使对于大型数组也应该非常有效。
@MSeifert 不幸的是，它比熊猫rolling_max() 慢，在我的测试中，我的实际尺寸比我的实际尺寸低约 2 倍。
这很有趣，因为在我的计算机上maximum_filter1d 对于 100k 的窗口大小和 10m 的数组大小来说要快 3-4 倍。您使用的是两个软件包的最新版本吗？
@Alfe a = np.arange(1000000), np.lib.stride_tricks.as_strided(a, shape=(1000, len(a)-1000+1), strides=(a.strides[0], a.strides[0])) 对我来说效果很好。你能报告你的 NumPy、Pandas 和 Scipy 版本吗？

【解决方案2】：

Pandas 对 Series 和 DataFrame 都有滚动方法，可以在这里使用：

import pandas as pd

lst = [6,4,8,7,1,4,3,5,7,8,4,6,2,1,3,5,6,3,4,7,1,9,4,3,2]
lst1 = pd.Series(lst).rolling(5).max().dropna().tolist()

# [8.0, 8.0, 8.0, 7.0, 7.0, 8.0, 8.0, 8.0, 8.0, 8.0, 6.0, 6.0, 6.0, 6.0, 6.0, 7.0, 7.0, 9.0, 9.0, 9.0, 9.0]

为了保持一致性，您可以将lst1 的每个元素强制转换为int：

[int(x) for x in lst1]

# [8, 8, 8, 7, 7, 8, 8, 8, 8, 8, 6, 6, 6, 6, 6, 7, 7, 9, 9, 9, 9]

【讨论】：

我发现您可以用更简单的方式重新表述您的解决方案：a = np.array(…)、pd.rolling_max(a, window=5)。到目前为止，这听起来像是我正在处理的尺寸的最佳选择。不过，如果@Divakar 的strides 解决方案适用于我的尺寸，它会更快，所以我还在等待接受这个答案。
新版本的pandas告诉我以后不再支持我的缩写，所以你的是最好的解决方案。

【解决方案3】：

我现在已经尝试了几种变体，并宣布 Pandas 版本是这场性能竞赛的获胜者。我尝试了几种变体，甚至使用二叉树（在纯 Python 中实现）来快速计算任意子范围的最大值。（可按需提供来源）。我自己想出的最好的算法是使用环形缓冲区的普通滚动窗口；只有在本次迭代中删除了当前最大值时，才需要完全重新计算最大值；否则它将保持或增加到下一个新值。与旧库相比，这种纯 Python 实现比其他库更快。

最后我发现有问题的库的版本是高度相关的。我主要仍在使用的相当旧的版本比现代版本慢得多。以下是 1M 个数字的数字，rollingMax'ed 大小为 100k 的窗口：

         old (slow HW)           new (better HW)
scipy:   0.9.0:  21.2987391949   0.13.3:  11.5804400444
pandas:  0.7.0:  13.5896410942   0.18.1:   0.0551438331604
numpy:   1.6.1:   1.17417216301  1.8.2:    0.537392139435

这里是使用 ringbuffer 的纯 numpy 版本的实现：

def rollingMax(a, window):
  def eachValue():
    w = a[:window].copy()
    m = w.max()
    yield m
    i = 0
    j = window
    while j < len(a):
      oldValue = w[i]
      newValue = w[i] = a[j]
      if newValue > m:
        m = newValue
      elif oldValue == m:
        m = w.max()
      yield m
      i = (i + 1) % window
      j += 1
  return np.array(list(eachValue()))

对于我的输入，这很有效，因为我正在处理各个方向都有很多峰值的音频数据。如果您将不断减小的信号放入其中（例如-np.arange(10000000)），那么您将遇到最坏的情况（也许您应该在这种情况下反转输入和输出）。

我只是将其包含在内，以防有人想在具有旧库的机器上执行此任务。

【讨论】：

【解决方案4】：

首先，我认为你的解释有误，因为你的解释开头的初始输入数组的第 10 个元素等于 8，而在下面，你应用窗口的地方，它是 2。

更正后，我认为执行您想要的代码如下：

import numpy as np
a=np.array([ 6,4,8,7,1,4,3,5,7,8,4,6,2,1,3,5,6,3,4,7,1,9,4,3,2 ])
window=5
for i in range(0,len(a)-window,1): 
    b[i] = np.amax(a[i:i+window])

我认为，这种方式比创建输入的移位 2D 版本更好，因为当您创建这样的版本时，您需要使用比使用原始输入数组更多的内存，因此如果输入可能会耗尽内存很大。

【讨论】：

天哪，你是对的！我在写我的问题的过程中改变了我的输入，以便展示更多的案例。我并没有因此而受到影响。我现在已经解决了。对于您的建议：我想避免在我的输入上出现任何 Python 编写的循环，因为这总是比使用 numpy、scipy、pandas 等包的任何功能慢。如果您认为您的解决方案可以竞争，请提供 timeits。否则：当然，这很简单，也是一个很好的解决方案。它只是不符合我的性能预期。

【解决方案5】：

如果您有二维数据，例如股票价格，并且想要获得滚动最大值或其他数据，这将有效。不使用迭代计算。

n = 5  # size of rolling window

data_expanded = np.expand_dims(data, 1)
data_shift = [np.roll(data_expanded, shift=-i, axis=2) for i in range(n)]
data_shift = np.concatenate(data_shift, axis=1)

data_max = np.max(data_shift, axis=1)  # max, mean, std...

【讨论】：

for i in range(n) 在我看来非常可疑，就像一个迭代。在我的情况下，n 将非常大，例如。 G。两秒的 96kHz 音频样本，因此 n > 150000。但无论如何感谢您的贡献，欢迎来到 StackOverflow :-)

【解决方案6】：

从Numpy 1.20 开始，sliding_window_view 提供了一种在元素窗口中滑动/滚动的方式。然后您可以找到最大值的窗口：

from numpy.lib.stride_tricks import sliding_window_view

# values = np.array([6,4,8,7,1,4,3,5,7,2,4,6,2,1,3,5,6,3,4,7,1,9,4,3,2])
np.max(sliding_window_view(values, window_shape = 5), axis = 1)
# array([8, 8, 8, 7, 7, 7, 7, 7, 7, 6, 6, 6, 6, 6, 6, 7, 7, 9, 9, 9, 9])

地点：

window_shape是滑动窗口的大小
np.max(array, axis = 1) 查找每个子数组的最大值

而滑动的中间结果是：

sliding_window_view(values, window_shape = 5)
# array([[6, 4, 8, 7, 1],
#        [4, 8, 7, 1, 4],
#        [8, 7, 1, 4, 3],
#        ...
#        [7, 1, 9, 4, 3],
#        [1, 9, 4, 3, 2]])

【讨论】：