ジョージのように、これに関するMartijnの答えが好きですがsum()、ほとんど同じ数字に何度も何度も適用する代わりに、実行中の合計を使用することでこれが速くならないのではないかと思っていました。
Noneまた、ランプアップ フェーズ中に値をデフォルトとして設定するというアイデアも興味深いものです。実際、移動平均について考えられるさまざまなシナリオがたくさんあるかもしれません。平均の計算を 3 つのフェーズに分けてみましょう。
- ランプアップ: 現在の反復回数 < ウィンドウ サイズで反復を開始する
- 着実な進歩: 法線を計算するために使用できる正確なウィンドウ サイズの数の要素があります。
average := sum(x[iteration_counter-window_size:iteration_counter])/window_size
- Ramp Down: 入力データの最後に、別の
window_size - 1「平均」数値を返すことができます。
ここに受け入れる関数があります
- データの入力としての任意のイテラブル (ジェネレーターは問題ありません)
- 任意のウィンドウ サイズ >= 1
- Ramp Up/Down のフェーズ中に値の生成をオン/オフするパラメータ
- 値の生成方法を制御するこれらのフェーズのコールバック関数。これを使用して、デフォルト (例: ) を常に提供し
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]