移动平均过滤器怎么样?它也是一个单行程序，它的优点是，如果你需要矩形以外的东西，你可以很容易地操作窗口类型。一个n长的简单移动平均数组a:

lfilter(np.ones(N)/N, [1], a)[N:]

应用三角形窗口后:

lfilter(np.ones(N)*scipy.signal.triang(N)/N, [1], a)[N:]

注:我通常会在最后丢弃前N个样本作为假的，因此[N:]，但这是没有必要的，只是个人选择的问题。

出于教学目的，让我再添加两个Numpy解决方案(比cumsum解决方案慢):

import numpy as np
from numpy.lib.stride_tricks import as_strided

def ra_strides(arr, window):
    ''' Running average using as_strided'''
    n = arr.shape[0] - window + 1
    arr_strided = as_strided(arr, shape=[n, window], strides=2*arr.strides)
    return arr_strided.mean(axis=1)

def ra_add(arr, window):
    ''' Running average using add.reduceat'''
    n = arr.shape[0] - window + 1
    indices = np.array([0, window]*n) + np.repeat(np.arange(n), 2)
    arr = np.append(arr, 0)
    return np.add.reduceat(arr, indices )[::2]/window

使用的函数:as_strided, add.reduceat

这个问题现在甚至比NeXuS上个月写的时候更古老，但我喜欢他的代码处理边缘情况的方式。然而，因为它是一个“简单移动平均”，它的结果滞后于它们应用的数据。我认为，通过对基于卷积()的方法应用类似的方法，可以以比NumPy的模式valid、same和full更令人满意的方式处理边缘情况。

我的贡献使用了一个中央运行平均值，以使其结果与他们的数据相一致。当可供使用的全尺寸窗口的点太少时，将从数组边缘的连续较小窗口计算运行平均值。[实际上，从连续较大的窗口，但这是一个实现细节。]

import numpy as np

def running_mean(l, N):
    # Also works for the(strictly invalid) cases when N is even.
    if (N//2)*2 == N:
        N = N - 1
    front = np.zeros(N//2)
    back = np.zeros(N//2)

    for i in range(1, (N//2)*2, 2):
        front[i//2] = np.convolve(l[:i], np.ones((i,))/i, mode = 'valid')
    for i in range(1, (N//2)*2, 2):
        back[i//2] = np.convolve(l[-i:], np.ones((i,))/i, mode = 'valid')
    return np.concatenate([front, np.convolve(l, np.ones((N,))/N, mode = 'valid'), back[::-1]])

它相对较慢，因为它使用了卷积()，并且可能会被真正的Pythonista修饰很多，但是，我相信这个想法是成立的。

对于一个简短、快速的解决方案，在一个循环中完成所有事情，没有依赖关系，下面的代码工作得很好。

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

for i, x in enumerate(mylist, 1):
    cumsum.append(cumsum[i-1] + x)
    if i>=N:
        moving_ave = (cumsum[i] - cumsum[i-N])/N
        #can do stuff with moving_ave here
        moving_aves.append(moving_ave)

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

注意:使用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

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