使用python在移动间隔上查找最大值（和最小值）

Question

我有一个像

这样的数组

[5.5, 6.0, 6.0, 6.5, 6.0, 5.5, 5.5, 5.0, 4.5]. 

这个数组的所有数字相差0.5，两个连续数字的最大差异也是0.5（它们可以相同；如示例中所示）。并且有一个移动区间或框，例如，它涵盖了 3 个连续的数字，如下所示：

[(5.5, 6.0, 6.0), 6.5, 6.0, 5.5, 5.5, 5.0, 4.5]  # min: 5.5, max: 6.0

盒子一个接一个地向右移动：

[5.5, (6.0, 6.0, 6.5), 6.0, 5.5, 5.5, 5.0, 4.5]  # min: 6.0, max: 6.5

[5.5, 6.0, (6.0, 6.5, 6.0), 5.5, 5.5, 5.0, 4.5]  # min: 6.0, max: 6.5

问题是，我怎样才能找到每次盒子移动时盒子内数字的最小值和最大值？

当盒子和数组的大小像这个例子一样小时，我可以处理它，但我需要将它应用于数组大小 100000 和盒子大小 10000。使用我的方法（我使用 for-loop 计算每个最大值和最小值每次盒子通过），花费了太多时间（我还有 100 个数组要做并且需要重复运行）。有时间限制，所以我需要像计算一样在 0.5 秒内运行它。

Answer 1

看看熊猫的rolling windows：

>>> import pandas as pd
>>> L = [5.5, 6.0, 6.0, 6.5, 6.0, 5.5, 5.5, 5.0, 4.5]
>>> a = pd.DataFrame(L)
>>> pd.rolling_max(a, 3)
     0
0  NaN
1  NaN
2  6.0
3  6.5
4  6.5
5  6.5
6  6.0
7  5.5
8  5.5
>>> pd.rolling_min(a, 3)
     0
0  NaN
1  NaN
2  5.5
3  6.0
4  6.0
5  5.5
6  5.5
7  5.0
8  4.5

Answer 2

起初在我看来，这需要大列表中每个元素的最少 O(log(window_size)) 操作（请参阅我的其他答案）。但是@wim 向我指出了@adamax 在这篇文章中描述的真正卓越的算法：

Implement a queue in which push_rear(), pop_front() and get_min() are all constant time operations

这是一个实现。

在建议的 100000 个数字上以 1000 个窗口运行它需要 0.6 秒，而不是简单算法的 60 秒。

class MinMaxStack(object):

    def __init__(self):
        self.stack = []

    def push(self,val):
        if not self.stack:
            self.stack = [(val,val,val)]
        else:
            _,minimum,maximum = self.stack[-1]
            if val < minimum:
                self.stack.append((val,val,maximum))
            elif val > maximum:
                self.stack.append((val,minimum,val))
            else:
                self.stack.append((val,minimum,maximum))

    def pop(self):
        return self.stack.pop()

    def get_minimax(self):
        return self.stack[-1][1:]

    def __len__(self):
        return len(self.stack)

class RollingWindow(object):

    def __init__(self):
        self.push_stack = MinMaxStack()
        self.pop_stack = MinMaxStack()

    def push_only(self,o):
        self.push_stack.push(o)

    def push_and_pop(self,o):
        self.push_stack.push(o)
        if not self.pop_stack:
            for i in range(len(self.push_stack.stack)-1):
                self.pop_stack.push(self.push_stack.pop()[0])
            self.push_stack.pop()
        else:
            self.pop_stack.pop()

    def get_minimax(self):
        if not self.pop_stack:
            return self.push_stack.get_minimax()
        elif not self.push_stack:
            return self.pop_stack.get_minimax()
        mn1,mx1 = self.pop_stack.get_minimax()
        mn2,mx2 = self.push_stack.get_minimax()
        return min(mn1,mn2),max(mx1,mx2)



import time
import random
window = 10000
test_length = 100000
data = [random.randint(1,100) for i in range(test_length)]

s = time.time()

wr = RollingWindow()
answer1 = []
for i in range(test_length):
    if i < window:
        wr.push_only(data[i])
    else:
        wr.push_and_pop(data[i])
    answer1.append(wr.get_minimax())

print(s-time.time())

s = time.time()
answer2 = []
for i in range(test_length):
    if i+1 < window:
        current_window = i+1
    else:
        current_window = window
    answer2.append((min(data[i+1-current_window:i+1]),max(data[i+1-current_window:i+1])))

print(s-time.time())

if answer1 != answer2:
    print("Test Fail")

一些小的性能改进是可能的。这个版本不断增长和缩小用作堆栈的 python 列表。从不收缩它并使用结束指针会稍微快一些。但只有百分之几。如果您真的很想再获得几个百分比，您可以将两个堆栈合并到窗口类中并减少调用中的间接性。我构建了一个优化版本，将列表替换为collections.deque，并内联堆栈代码并将其缩短到 0.32 秒。

如果需要更高的速度，这将很容易在 C 或 Cython 中进行编码（特别是对于固定窗口大小），特别是如果您可以限制堆栈上值的类型。

Answer 3

l = [5.5, 6.0, 6.0, 6.5, 6.0, 5.5, 5.5, 5.0, 4.5]

windoSize = 3

for i in range(0,len(l)-windowSize+1):

    print max(l[i:i+windoSize])

输出：

6.0
6.5
6.5
6.5
6.0
5.5
5.5

Answer 4

这是一个滚动窗口，可以在熊猫中实现，如其他答案所示。

但是，如果您想自己实现它，下面的代码会有所帮助。这段代码可以进一步优化，可能更符合 Python 风格，但它应该可以很好地理解算法中发生的事情。

最初找到起始窗口的最小值和最大值。 初始化后，我们将子数组视为队列，只有 2 个值变得重要，新值被添加，旧值被删除。

如果旧值是最小值或最大值，我们重新计算最小值或最大值，否则我们检查新值是否为新最大值或最小值。

def updateMinMaxValues(minVal,maxVal,val):
    if val < minVal:
        minVal = val
    if val > maxVal:
        maxVal= val
    return minVal,maxVal

values = [5.5, 6.0, 6.0, 6.5, 6.0, 5.5, 5.5, 5.0, 4.5]
windowSize = 3
minVal,maxVal = min(values[:windowSize]),max(values[:windowSize])

print(minVal,maxVal)
for stepIndex in range(windowSize,len(values)):
    oldVal,newVal = values[stepIndex-windowSize],values[stepIndex]
    if oldVal == minVal:
        minVal = min(values[stepIndex-windowSize+1:stepIndex+1])
    if oldVal == maxVal:
        maxVal = max(values[stepIndex-(windowSize)+1:stepIndex+1])
    minVal,maxVal = updateMinMaxValues(minVal,maxVal,newVal)
    print(minVal,maxVal)

结果：

5.5 6.0
6.0 6.5
6.0 6.5
5.5 6.5
5.5 6.0
5.0 5.5
4.5 5.5

Answer 5

不确定是否有办法有效利用数字流的缓慢移动结构。

我认为最好的通用方法是使用优先队列。我在下面留下了如何做到这一点的描述。每个新数字进入窗口需要 O(log(window_size))。

然而，wim 对原帖的评论指出存在一个 O(1) 算法，在这篇帖子中描述：Implement a queue in which push_rear(), pop_front() and get_min() are all constant time operations

到目前为止，只需维护其中一个保持最小值和最大值的方法将是最好的解决方案。

但这里是我的尝试供参考：

维护一对优先级队列，一个用于最大，一个用于最小，每次都添加和删除一个条目。这增加了相当多的开销 每个新条目 [ O(log(window_size)) ]，但每个条目的行为都非常流畅，整体效率也不错。

Python heapq 模块是在 Python 中实现优先级队列的常用方法。但是，它不直接支持删除条目或修改它们的优先级。这可以通过在队列中从数字到位置添加字典索引来完成，而不会增加计算复杂度。要删除一个条目，您可以将其编号更新为极低（或分别为高）并重新堆化，使其移动到顶部并可以弹出。

这是一个例子，虽然我没有测试过，但看起来还不错：

http://code.activestate.com/recipes/522995-priority-dict-a-priority-queue-with-updatable-prio/

您需要消除字典中具有相同值的条目的歧义，或者为每个键保留多个值，以便在需要删除它们时找到所有实例。

Answer 6

在pandas的新版本中，你必须按照我在文档中所说的那样使用它：

>>> s = pd.Series([4, 3, 5, 2, 6])
>>> s.rolling(3).min()
0    NaN
1    NaN
2    3.0
3    2.0
4    2.0
dtype: float64