如何在Python 3中计算移动平均值？
如何在Python 3中计算移动平均值？
提问于 2013-02-15 05:05:19
假设我有一个列表：
y = ['1', '2', '3', '4','5','6','7','8','9','10']我想创建一个计算移动n天平均值的函数。因此，如果n是5，我希望我的代码计算前1-5，添加它并找到平均值，它将是3.0，然后继续到2-6，计算平均值，它将是4.0，然后是3-7，4-8，5-9，6-10。
def moving_average(x:'list of prices', n):
for num in range(len(x)+1):
print(x[num-n:num])这似乎打印出了我想要的内容：
[]
[]
[]
[]
[]
['1', '2', '3', '4', '5']
['2', '3', '4', '5', '6']
['3', '4', '5', '6', '7']
['4', '5', '6', '7', '8']
['5', '6', '7', '8', '9']
['6', '7', '8', '9', '10']但是，我不知道如何计算这些列表中的数字。有什么想法吗？
回答 5
Stack Overflow用户
回答已采纳
发布于 2013-02-15 05:07:58
在使用itertools examples的Python文档的旧版本中有一个很棒的滑动窗口生成器
from itertools import islice
def window(seq, n=2):
"Returns a sliding window (of width n) over data from the iterable"
" s -> (s0,s1,...s[n-1]), (s1,s2,...,sn), ... "
it = iter(seq)
result = tuple(islice(it, n))
if len(result) == n:
yield result
for elem in it:
result = result[1:] + (elem,)
yield result使用你的移动平均值是微不足道的：
from __future__ import division # For Python 2
def moving_averages(values, size):
for selection in window(values, size):
yield sum(selection) / size对您的输入运行此命令(将字符串映射为整数)会得到以下结果：
>>> y= ['1', '2', '3', '4','5','6','7','8','9','10']
>>> for avg in moving_averages(map(int, y), 5):
... print(avg)
...
3.0
4.0
5.0
6.0
7.0
8.0要返回“不完整”集合的第一次n - 1迭代，只需稍微扩展一下moving_averages函数：
def moving_averages(values, size):
for _ in range(size - 1):
yield None
for selection in window(values, size):
yield sum(selection) / sizeStack Overflow用户
发布于 2013-02-19 02:15:15
虽然我喜欢Martijn's answer在这一点上，像乔治一样，我想知道这是否会更快，使用运行求和，而不是一遍又一遍地对几乎相同的数字应用sum()。
此外，在加速阶段将None值设为默认值的想法也很有趣。事实上，移动平均线可能有很多不同的场景。让我们将平均值的计算分成三个阶段：
窗口大小上升:开始迭代，其中当前迭代计数<窗口大小窗口大小稳定的进展:我们正好有窗口大小的元素数可用于计算正常的窗口大小下降:在输入数据的末尾，我们可以返回另一个平均“average := sum(x[iteration_counter-window_size:iteration_counter])/window_size
- Ramp”数字。
window_size - 1
下面是一个函数，它接受
作为data
- Arbitrary窗口大小的输入，
- 任意迭代(生成器很好) >= 1
- 参数，在那些阶段的Ramp Up/Down
- Callback函数的阶段期间，打开/关闭值的产生，以控制值的产生方式。这可用于不断提供默认值(例如
None)或提供
的部分平均值
代码如下：
from collections import deque
def moving_averages(data, size, rampUp=True, rampDown=True):
"""Slide a window of <size> elements over <data> to calc an average
First and last <size-1> iterations when window is not yet completely
filled with data, or the window empties due to exhausted <data>, the
average is computed with just the available data (but still divided
by <size>).
Set rampUp/rampDown to False in order to not provide any values during
those start and end <size-1> iterations.
Set rampUp/rampDown to functions to provide arbitrary partial average
numbers during those phases. The callback will get the currently
available input data in a deque. Do not modify that data.
"""
d = deque()
running_sum = 0.0
data = iter(data)
# rampUp
for count in range(1, size):
try:
val = next(data)
except StopIteration:
break
running_sum += val
d.append(val)
#print("up: running sum:" + str(running_sum) + " count: " + str(count) + " deque: " + str(d))
if rampUp:
if callable(rampUp):
yield rampUp(d)
else:
yield running_sum / size
# steady
exhausted_early = True
for val in data:
exhausted_early = False
running_sum += val
#print("st: running sum:" + str(running_sum) + " deque: " + str(d))
yield running_sum / size
d.append(val)
running_sum -= d.popleft()
# rampDown
if rampDown:
if exhausted_early:
running_sum -= d.popleft()
for (count) in range(min(len(d), size-1), 0, -1):
#print("dn: running sum:" + str(running_sum) + " deque: " + str(d))
if callable(rampDown):
yield rampDown(d)
else:
yield running_sum / size
running_sum -= d.popleft()它似乎比Martijn的版本快一点--尽管它要优雅得多。下面是测试代码：
print("")
print("Timeit")
print("-" * 80)
from itertools import islice
def window(seq, n=2):
"Returns a sliding window (of width n) over data from the iterable"
" s -> (s0,s1,...s[n-1]), (s1,s2,...,sn), ... "
it = iter(seq)
result = tuple(islice(it, n))
if len(result) == n:
yield result
for elem in it:
result = result[1:] + (elem,)
yield result
# Martijn's version:
def moving_averages_SO(values, size):
for selection in window(values, size):
yield sum(selection) / size
import timeit
problems = [int(i) for i in (10, 100, 1000, 10000, 1e5, 1e6, 1e7)]
for problem_size in problems:
print("{:12s}".format(str(problem_size)), end="")
so = timeit.repeat("list(moving_averages_SO(range("+str(problem_size)+"), 5))", number=1*max(problems)//problem_size,
setup="from __main__ import moving_averages_SO")
print("{:12.3f} ".format(min(so)), end="")
my = timeit.repeat("list(moving_averages(range("+str(problem_size)+"), 5, False, False))", number=1*max(problems)//problem_size,
setup="from __main__ import moving_averages")
print("{:12.3f} ".format(min(my)), end="")
print("")和输出：
Timeit
--------------------------------------------------------------------------------
10 7.242 7.656
100 5.816 5.500
1000 5.787 5.244
10000 5.782 5.180
100000 5.746 5.137
1000000 5.745 5.198
10000000 5.764 5.186 原来的问题现在可以通过这个函数调用来解决：
print(list(moving_averages(range(1,11), 5,
rampUp=lambda _: None,
rampDown=False)))输出：
[None, None, None, None, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0]Stack Overflow用户
发布于 2013-02-15 06:04:04
一种避免重新计算中间和的方法。
list=range(0,12)
def runs(v):
global runningsum
runningsum+=v
return(runningsum)
runningsum=0
runsumlist=[ runs(v) for v in list ]
result = [ (runsumlist[k] - runsumlist[k-5])/5 for k in range(0,len(list)+1)]打印结果
[2,3,4,5,6,7,8,9]使其运行(int(V)) ..然后..。repr( runsumlistk - runsumlistk-5)/5 )如果你不想携带数字字符串..
不带全局的Alt：
list = [float[x] for x in range(0,12)]
nave = 5
movingave = sum(list[:nave]/nave)
for i in range(len(list)-nave):movingave.append(movingave[-1]+(list[i+nave]-list[i])/nave)
print movingave 即使输入值是整数，也要确保进行浮点数学运算
[2.0,3.0,4.0,5.0,6.0,7.0,8.0,9,0]页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/14884017
复制相关文章
相似问题
腾讯云开发者
