Python中的移动平均

Question

有没有一种更“Python风”的方法来创建一个包含某个函数的滑动平均值的列表呢？

我在读了一篇有趣的文章，讲的是火星人、黑箱和柯西分布后，觉得自己计算一下柯西分布的滑动平均值也挺有意思的：

import math 
import random

def cauchy(location, scale):
    p = 0.0
    while p == 0.0:
        p = random.random()
    return location + scale*math.tan(math.pi*(p - 0.5))

# is this next block of code a good way to populate running_avg?
sum = 0
count = 0
max = 10
running_avg = []
while count < max:
    num = cauchy(3,1)
    sum += num
    count += 1
    running_avg.append(sum/count)

print running_avg     # or do something else with it, besides printing

我觉得这个方法是可行的，但我很好奇，是否有比用循环和计数器更优雅的方法来构建这个running_avg列表，比如使用列表推导式。

虽然有一些相关的问题，但它们涉及的都是更复杂的情况（比如小窗口大小、指数加权），或者并不是专门针对Python的：

• 如何在Python中计算指数移动平均
• 如何高效地计算滑动标准差？
• 计算列表的移动平均

Answer 1

我这里有两个可能的解决方案给你。它们都是通用的运行平均数函数，可以处理任何数字列表。（其实可以改造成处理任何可迭代的对象）

基于生成器的：

nums = [cauchy(3,1) for x in xrange(10)]

def running_avg(numbers):
    for count in xrange(1, len(nums)+1):
        yield sum(numbers[:count])/count

print list(running_avg(nums))

基于列表推导的（其实和前面的代码是一样的）：

nums = [cauchy(3,1) for x in xrange(10)]

print [sum(nums[:count])/count for count in xrange(1, len(nums)+1)]

兼容生成器的生成器方法：

编辑：我刚刚测试了一下，看看能不能轻松让我的解决方案兼容生成器，以及它的性能如何。这是我得到的结果。

def running_avg(numbers):
    sum = 0
    for count, number in enumerate(numbers):
        sum += number
        yield sum/(count+1)

下面是性能统计，值得一看。

性能特点：

编辑：我还决定测试一下Orip有趣的多生成器用法，看看对性能的影响。

使用timeit和以下设置（1,000,000次迭代，重复3次）：

print "Generator based:", ', '.join(str(x) for x in Timer('list(running_avg(nums))', 'from __main__ import nums, running_avg').repeat())
print "LC based:", ', '.join(str(x) for x in Timer('[sum(nums[:count])/count for count in xrange(1, len(nums)+1)]', 'from __main__ import nums').repeat())
print "Orip's:", ', '.join(str(x) for x in Timer('list(itertools.islice(running_avgs, 10))', 'from __main__ import itertools, running_avgs').repeat())

print "Generator-compatabile Generator based:", ', '.join(str(x) for x in Timer('list(running_avg(nums))', 'from __main__ import nums, running_avg').repeat())

我得到了以下结果：

Generator based: 17.653908968, 17.8027219772, 18.0342400074
LC based: 14.3925321102, 14.4613749981, 14.4277560711
Orip's: 30.8035550117, 30.3142540455, 30.5146529675

Generator-compatabile Generator based: 3.55352187157, 3.54164409637, 3.59098005295

代码请查看评论：

Orip's genEx based: 4.31488609314, 4.29926609993, 4.30518198013 

结果以秒为单位，显示出LC新的兼容生成器的方法始终更快，不过你的结果可能会有所不同。我认为我原来的生成器和新的生成器之间的巨大差异在于，新的方法并不是实时计算总和。

Answer 2

你可以使用协程。协程和生成器有点像，但它允许你传入值。协程是在Python 2.5版本中引入的，所以在这个版本之前的Python是无法使用的。

def running_average():
    sum = 0.0
    count = 0
    value = yield(float('nan'))
    while True:
        sum += value
        count += 1
        value = yield(sum/count)

ravg = running_average()
next(ravg)   # advance the corutine to the first yield

for i in xrange(10):
    avg = ravg.send(cauchy(3,1))
    print 'Running average: %.6f' % (avg,)

作为列表推导式：

ravg = running_average()
next(ravg)
ravg_list = [ravg.send(cauchy(3,1)) for i in xrange(10)]

编辑：

• 使用 next() 函数，而不是 it.next() 方法。这样做是为了让它也能在Python 3中使用。next() 函数也已经被移植到Python 2.6及以上版本。
  在Python 2.5中，你可以用 it.next() 替换调用，或者自己定义一个 next 函数。
  （感谢Adam Parkin）

Answer 3

你可以写一个生成器：

def running_average():
  sum = 0
  count = 0
  while True:
    sum += cauchy(3,1)
    count += 1
    yield sum/count

或者，假如你有一个用来生成Cauchy数的生成器，还有一个用来计算累加和的工具函数，你就可以写出一个简洁的生成器表达式：

# Cauchy numbers generator
def cauchy_numbers():
  while True:
    yield cauchy(3,1)

# running sum utility function
def running_sum(iterable):
  sum = 0
  for x in iterable:
    sum += x
    yield sum

# Running averages generator expression (** the neat part **)
running_avgs = (sum/(i+1) for (i,sum) in enumerate(running_sum(cauchy_numbers())))

# goes on forever
for avg in running_avgs:
  print avg

# alternatively, take just the first 10
import itertools
for avg in itertools.islice(running_avgs, 10):
  print avg