计算给定窗口的滚动平均值和滚动最大值

Question

计算给定窗口 w1 的滚动平均值和滚动最大值。 但使用的是： 您应该编写一个接受迭代器并充当生成元组的生成器的类，如上所示。

例如，给定一个包含以下值的流。 W = 3。

[1, 2, 3, 4, 5, 6]

需要以下元组，其中“None”表示值不可用，在某些语言中为“NaN”：

(None, None) -- When we have input 1  --- As we dont have w num, ignore - None
(None, None) -- Now we have input 1,2 
(2, 3)       -- Now we have input 1,2,3
(3, 4)       -- Discard 1. Window [2,3,4]
(4, 5)       -- Discard 3. Window [3,4,5]
(5, 6)       -- Discard 4. Window [4,5,6]

在这个元组中，第一个 num 是 w (=3) 个数字的平均值。第二个 num 是 w(=3) 个数字中的最大值。

我还在学习 Python 中的迭代器和生成器/yield。

Que：所以生成器接受整个序列（或迭代器/列表）？如果什么 seq 很大吗？

我写了简单的解决方案，它可能不是有效的算法，但无论如何 我想先知道这对生成器是否正确：

 7 from collections import deque
 8 class Solution:
 9     def sliding_window_avg_max(self, nums, w=3):
 10         d = deque(maxlen = w)   
 11         total = 0
 12         for n in nums:
 13             if len(d) >= w:
 14                 total = total - d.pop() + n
 15             else:
 16                 total = total + n
 17             d.append(n)
 18             if len(d) >= w:
 19                 yield (total/float(w), max(d))
 20             else:
 21                 yield(None, None)
 22 
 23 s = Solution()
 24 a = [1, 2, 3, 4, 5, 6]
 25 for t in s.sliding_window_avg_max(a):
 26     print t 

编辑： 这不是作业问题。这是我想要的面试问题 也知道别人的想法。我真的很想知道生成器如何在这里提供帮助-它不需要列表（nums）在内存中吗？ 如果应该，我们将如何使用迭代器？

Answer 1

我会将您的方法分为：

1. 产生 W 数字元组的生成器。它可以保存 W+1 个值，因此它知道在添加新数字时要丢弃什么。
2. 计算当前元组的平均值和最大值的计算器。

这有两个原因。首先，在这种情况下，编码时间是有限的，如果您只是检查是否正确交付，调试生成器会更容易。其次，这是一种更灵活的方法——如果面试官要求添加最小偏差和标准偏差，那么计算模块中只需添加几行即可。

Answer 2

您应该能够将其转换为类，或将其作为方法添加到类中。

def sliding_window_avg_max(nums, w=3):
    if w < 1:
        raise ValueError('Window size must be > 0')
    for i in range(1, len(nums)+1):
        if i < w:
            yield None, None
        else:
            yield sum(nums[i-w:i])/float(w), max(nums[i-w:i])


>>> for t in sliding_window_avg_max([1, 2, 3, 4, 5, 6]):
...     print t
(None, None)
(None, None)
(2.0, 3)
(3.0, 4)
(4.0, 5)
(5.0, 6)

Answer 3

请注意，规范声明（强调我的）：

您应该编写一个接受迭代器并 充当 生成元组的生成器，如上所示。

我认为你误解了这个问题；我怀疑您被要求实施the iterator protocol。一种可能的解决方案如下所示：

from collections import deque


class Solution(object):
    """Apply a moving window to an iterator, calculate max and average.

    Arguments:
      iter_ (iterator): The iterator to process.
      w (int): The length of the moving window to apply.

    Attributes:
      window (deque): The moving window.

    Example:

        >>> list(Solution(range(1, 7), 3))
        [(None, None), (None, None), (2, 3), (3, 4), (4, 5), (5, 6)]

    """

    def __init__(self, iter_, w):
        self.iter = iter(iter_)
        self.w = w
        self.window = deque(maxlen=w)

    def __iter__(self):
        return self

    def next(self):
        """Calculate the next value in the series.

        Notes:
          If the window isn't full, return (None, None). 

        """
        self.window.append(next(self.iter))
        if len(self.window) == self.w:
            return self._avg(), self._max()
        return None, None

    def _max(self):
        """Calculate the maximum of the current window."""
        return max(self.window)

    def _avg(self):
        """Calculate the average of the current window."""
        return int(sum(self.window) / float(len(self.window)))

---

请注意，您也可以将__iter__ 实现为生成器，即删除next 并实现：

    def __iter__(self):
        for val in self.iter:
            self.window.append(val)
            if len(self.window) == self.w:
                yield self._avg(), self._max()
            else:
                yield None, None

无论哪种方式，您都确保来自 __iter__ 的返回是一个迭代器！

---

你的解决方案应该让你停下来；根据"Stop Writing Classes"，如果您的类实现了__init__ 和另一种方法（在这种情况下甚至没有__init__！），它不应该是一个类。