我觉得使用瓶颈可以很好地解决这个问题

参见下面的基本示例:

import numpy as np
import bottleneck as bn

a = np.random.randint(4, 1000, size=100)
mm = bn.move_mean(a, window=5, min_count=1)

“mm”是“a”的移动平均值。 “窗口”是考虑移动均值的最大条目数。 "min_count"是考虑移动平均值的最小条目数(例如，对于前几个元素或如果数组有nan值)。

好在瓶颈有助于处理nan值，而且非常高效。

我知道这是一个老问题，但这里有一个解决方案，它不使用任何额外的数据结构或库。它在输入列表的元素数量上是线性的，我想不出任何其他方法来使它更有效(实际上，如果有人知道更好的分配结果的方法，请告诉我)。

注意:使用numpy数组而不是列表会快得多，但我想消除所有依赖关系。通过多线程执行也可以提高性能

该函数假设输入列表是一维的，所以要小心。

### Running mean/Moving average
def running_mean(l, N):
    sum = 0
    result = list( 0 for x in l)

    for i in range( 0, N ):
        sum = sum + l[i]
        result[i] = sum / (i+1)

    for i in range( N, len(l) ):
        sum = sum - l[i-N] + l[i]
        result[i] = sum / N

    return result

例子

假设我们有一个列表data =[1,2,3,4,5,6]，我们想在它上面计算周期为3的滚动平均值，并且你还想要一个与输入列表相同大小的输出列表(这是最常见的情况)。

第一个元素的索引为0，因此滚动平均值应该在索引为-2、-1和0的元素上计算。显然，我们没有data[-2]和data[-1](除非您想使用特殊的边界条件)，因此我们假设这些元素为0。这相当于对列表进行零填充，除了我们实际上不填充它，只是跟踪需要填充的索引(从0到N-1)。

所以，对于前N个元素，我们只是在累加器中不断地把元素加起来。

result[0] = (0 + 0 + 1) / 3  = 0.333    ==   (sum + 1) / 3
result[1] = (0 + 1 + 2) / 3  = 1        ==   (sum + 2) / 3
result[2] = (1 + 2 + 3) / 3  = 2        ==   (sum + 3) / 3

从元素N+1开始，简单的累加是行不通的。我们期望的结果是[3]=(2 + 3 + 4)/3 = 3，但这与(sum + 4)/3 = 3.333不同。

计算正确值的方法是用sum+4减去数据[0]= 1，从而得到sum+4 - 1 = 9。

这是因为目前sum =数据[0]+数据[1]+数据[2]，但对于每个i >= N也是如此，因为在减法之前，sum是数据[i-N] +…+ data[i-2] + data[i-1]。

如果你选择自己生成，而不是使用现有的库，请注意浮点错误并尽量减少其影响:

class SumAccumulator:
    def __init__(self):
        self.values = [0]
        self.count = 0

    def add( self, val ):
        self.values.append( val )
        self.count = self.count + 1
        i = self.count
        while i & 0x01:
            i = i >> 1
            v0 = self.values.pop()
            v1 = self.values.pop()
            self.values.append( v0 + v1 )

    def get_total(self):
        return sum( reversed(self.values) )

    def get_size( self ):
        return self.count

如果所有的值都是大致相同的数量级，那么这将通过始终添加大致相似的数量级值来帮助保持精度。

另一个解决方案是使用标准库和deque:

from collections import deque
import itertools

def moving_average(iterable, n=3):
    # http://en.wikipedia.org/wiki/Moving_average
    it = iter(iterable) 
    # create an iterable object from input argument
    d = deque(itertools.islice(it, n-1))  
    # create deque object by slicing iterable
    d.appendleft(0)
    s = sum(d)
    for elem in it:
        s += elem - d.popleft()
        d.append(elem)
        yield s / n

# example on how to use it
for i in  moving_average([40, 30, 50, 46, 39, 44]):
    print(i)

# 40.0
# 42.0
# 45.0
# 43.0

上述所有的解决方案都很差，因为它们缺乏

由于本机python而不是numpy向量化实现， 数值稳定性，由于numpy使用不当。cumsum或 由于O(len(x) * w)实现为卷积的速度。

鉴于

import numpy
m = 10000
x = numpy.random.rand(m)
w = 1000

注意x_[:w].sum()等于x[:w-1].sum()。因此，对于第一个平均值，numpy.cumsum(…)加上x[w] / w(通过x_[w+1] / w)，并减去0(从x_[0] / w)。结果是x[0:w].mean()

通过cumsum，您将通过添加x[w+1] / w并减去x[0] / w来更新第二个平均值，从而得到x[1:w+1].mean()。

这将一直进行，直到到达x[-w:].mean()。

x_ = numpy.insert(x, 0, 0)
sliding_average = x_[:w].sum() / w + numpy.cumsum(x_[w:] - x_[:-w]) / w

这个解是向量化的，O(m)，可读且数值稳定。

使用@Aikude的变量，我编写了一行程序。

import numpy as np

mylist = [1, 2, 3, 4, 5, 6, 7]
N = 3

mean = [np.mean(mylist[x:x+N]) for x in range(len(mylist)-N+1)]
print(mean)

>>> [2.0, 3.0, 4.0, 5.0, 6.0]