Come calcolare la media mobile in Python 3?

Question

Diciamo che ho una lista:

y = ['1', '2', '3', '4','5','6','7','8','9','10'] 

Voglio creare una funzione che calcola la media mobile n giorni. Quindi se n era 5, vorrei che il mio codice calcoli il primo 1-5, aggiungilo e trovi la media, che sarebbe 3.0, poi vai a 2-6, calcola la media, che sarebbe 4.0, quindi 3-7, 4-8, 5-9, 6-10.

Non voglio calcolare i primi n-1 giorni, quindi a partire dall'undicesimo giorno, contò i giorni precedenti.

def moving_average(x:'list of prices', n): 
    for num in range(len(x)+1): 
     print(x[num-n:num]) 

Questo sembra di stampare quello che voglio:

[] 
[] 
[] 
[] 
[] 

['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'] 

Tuttavia, non so come calcolare i numeri all'interno di questi elenchi. Qualche idea?

Answer 1

C'è un grande generatore di finestra scorrevole in una vecchia versione della documentazione Python con itertools examples:

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 

usando che le tue medie mobili è banale:

from __future__ import division # For Python 2 

def moving_averages(values, size): 
    for selection in window(values, size): 
     yield sum(selection)/size 

L'esecuzione di questo contro il vostro ingresso (mappando le stringhe ai numeri interi):

>>> 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 

Per re girano None i primi n - 1 iterazioni per i set di 'incomplete', basta espandere la funzione moving_averages un po ':

def moving_averages(values, size): 
    for _ in range(size - 1): 
     yield None 
    for selection in window(values, size): 
     yield sum(selection)/size 

Answer 2

Utilizzare le funzioni sum e map.

print(sum(map(int, x[num-n:num]))) 

La funzione map in Python 3 è sostanzialmente una versione pigra di questo:

[int(i) for i in x[num-n:num]] 

Sono sicuro che si può intuire ciò che la funzione sum fa.

Answer 3

Un approccio che evita di ricalcolare le somme intermedie ..

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)] 

risultato di stampa

[2,3,4,5,6,7,8,9] 

---

make that runs (int (v)) .. then .. rep r (runumlist [k] - runumlist [k-5])/5) se stai per portare in giro numeri uno stringhe ..

---

Alt senza il globale:

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 

essere sicuri di fare matematica a virgola mobile, anche se voi i valori di ingresso sono interi

[2.0,3.0,4.0,5.0,6.0,7.0,8.0,9,0] 

Answer 4

Mentre mi piace Martijn's answer su questo, come George, mi Stavo chiedendo se questo non sarebbe stato più veloce utilizzando una sommatoria in esecuzione invece di applicare lo sum() più e più volte su principalmente gli stessi numeri.

Anche l'idea di avere valori None come predefiniti durante la fase di rampa è interessante. In realtà ci possono essere molti scenari diversi che si potrebbero immaginare per le medie mobili. Dividiamo il calcolo delle medie in tre fasi:

1. Ramp Up: A partire iterazioni cui contare l'iterazione corrente < dimensione della finestra
2. progressi costanti: Abbiamo esattamente il numero dimensione della finestra di elementi a disposizione per calcolare un normale average := sum(x[iteration_counter-window_size:iteration_counter])/window_size
3. Rampa in basso: alla fine dei dati di input, è possibile restituire altri numeri "medi" di window_size - 1.

Ecco una funzione che accetta

• iterables arbitrari (generatori sono bene) come ingresso per i dati
• dimensioni delle finestre arbitrarie> = 1
• Parametri per accendere/spegnere produzione di valori durante le fasi per Rampa su/giù
• Funzioni di richiamata per quelle fasi per controllare come vengono prodotti i valori. Questo può essere usato per fornire costantemente un default (ad es None) o per fornire le medie parziali

Ecco il codice:

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() 

Sembra essere un po 'più veloce rispetto alla versione di Martijn - che è molto più elegante, però. Ecco il codice di prova:

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("") 

E l'output:

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 

La domanda iniziale può ora essere risolto con questa chiamata di funzione:

print(list(moving_averages(range(1,11), 5, 
          rampUp=lambda _: None, 
          rampDown=False))) 

L'output:

[None, None, None, None, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0] 

Answer 5

C'è un'altra soluzione che si estende g una ricetta itertoolspairwise().È possibile estendere questo a nwise(), che vi dà la finestra scorrevole (e funziona se l'iterabile è un generatore):

def nwise(iterable, n): 
    ts = it.tee(iterable, n) 
    for c, t in enumerate(ts): 
     next(it.islice(t, c, c), None) 
    return zip(*ts) 

def moving_averages_nw(iterable, n): 
    yield from (sum(x)/n for x in nwise(iterable, n)) 

>>> list(moving_averages_nw(range(1, 11), 5)) 
[3.0, 4.0, 5.0, 6.0, 7.0, 8.0] 

Mentre un costo relativamente elevato di installazione in breve iterable s questo costo si riduce a impatto più a lungo i dati impostato. Ciò utilizza sum() ma il codice è ragionevolmente elegante:

Timeit    MP   cfi   ***** 
-------------------------------------------------------------------------------- 
10     4.658  4.959  7.351 
100    5.144  4.070  4.234 
1000    5.312  4.020  3.977 
10000    5.317  4.031  3.966 
100000    5.508  4.115  4.087 
1000000   5.526  4.263  4.202 
10000000   5.632  4.326  4.242