Existe-t-il une fonction SciPy ou une fonction NumPy ou un module pour Python qui calcule la moyenne de fonctionnement d'un tableau 1D dans une fenêtre spécifique?

Pour une solution courte et rapide qui fait le tout en une seule boucle, sans dépendances, le code ci-dessous fonctionne très bien.

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)

UPD: des solutions plus efficaces ont été proposées par Alleo et jasaarim .

Vous pouvez utiliser np.convolvepour cela:

np.convolve(x, np.ones((N,))/N, mode='valid')

Explication

La moyenne courante est un cas de l'opération mathématique de convolution . Pour la moyenne courante, vous faites glisser une fenêtre le long de l'entrée et calculez la moyenne du contenu de la fenêtre. Pour les signaux 1D discrets, la convolution est la même chose, sauf qu'au lieu de la moyenne, vous calculez une combinaison linéaire arbitraire, c'est-à-dire multipliez chaque élément par un coefficient correspondant et additionnez les résultats. Ces coefficients, un pour chaque position dans la fenêtre, sont parfois appelés le noyau de convolution . Maintenant, la moyenne arithmétique de N valeurs est (x_1 + x_2 + ... + x_N) / N, donc le noyau correspondant est (1/N, 1/N, ..., 1/N), et c'est exactement ce que nous obtenons en utilisant np.ones((N,))/N.

Bords

L' modeargument de np.convolvespécifie comment gérer les arêtes. J'ai choisi le validmode ici parce que je pense que c'est ainsi que la plupart des gens s'attendent à ce que la course à pied fonctionne, mais vous pouvez avoir d'autres priorités. Voici un graphique qui illustre la différence entre les modes:

import numpy as np
import matplotlib.pyplot as plt
modes = ['full', 'same', 'valid']
for m in modes:
    plt.plot(np.convolve(np.ones((200,)), np.ones((50,))/50, mode=m));
plt.axis([-10, 251, -.1, 1.1]);
plt.legend(modes, loc='lower center');
plt.show()

Solution efficace

La convolution est bien meilleure qu'une approche simple, mais (je suppose) qu'elle utilise FFT et donc assez lente. Cependant, spécialement pour le calcul du fonctionnement, l'approche suivante fonctionne bien

def running_mean(x, N):
    cumsum = numpy.cumsum(numpy.insert(x, 0, 0)) 
    return (cumsum[N:] - cumsum[:-N]) / float(N)

Le code à vérifier

In[3]: x = numpy.random.random(100000)
In[4]: N = 1000
In[5]: %timeit result1 = numpy.convolve(x, numpy.ones((N,))/N, mode='valid')
10 loops, best of 3: 41.4 ms per loop
In[6]: %timeit result2 = running_mean(x, N)
1000 loops, best of 3: 1.04 ms per loop

Notez que numpy.allclose(result1, result2)est True, deux méthodes sont équivalentes. Plus N est grand, plus la différence de temps est grande.

avertissement: bien que cumsum soit plus rapide, il y aura une augmentation des erreurs en virgule flottante qui peuvent rendre vos résultats invalides / incorrects / inacceptables

les commentaires ont souligné ce problème d'erreur en virgule flottante ici, mais je le rends plus évident ici dans la réponse. .

# demonstrate loss of precision with only 100,000 points
np.random.seed(42)
x = np.random.randn(100000)+1e6
y1 = running_mean_convolve(x, 10)
y2 = running_mean_cumsum(x, 10)
assert np.allclose(y1, y2, rtol=1e-12, atol=0)

• plus vous accumulez de points, plus l'erreur en virgule flottante est grande (donc 1e5 points est perceptible, 1e6 points est plus significatif, plus de 1e6 et vous voudrez peut-être réinitialiser les accumulateurs)
• vous pouvez tricher en utilisant np.longdoublemais votre erreur en virgule flottante deviendra toujours significative pour un nombre relativement important de points (environ> 1e5 mais dépend de vos données)
• vous pouvez tracer l'erreur et la voir augmenter relativement rapidement
• la solution de convolution est plus lente mais n'a pas cette perte de précision en virgule flottante
• la solution uniform_filter1d est plus rapide que cette solution cumulative ET n'a pas cette perte de précision en virgule flottante

Mise à jour: l'exemple ci-dessous montre l'ancienne pandas.rolling_meanfonction qui a été supprimée dans les versions récentes de pandas. Un équivalent moderne de l'appel de fonction ci-dessous serait

In [8]: pd.Series(x).rolling(window=N).mean().iloc[N-1:].values
Out[8]: 
array([ 0.49815397,  0.49844183,  0.49840518, ...,  0.49488191,
        0.49456679,  0.49427121])

pandas est plus approprié pour cela que NumPy ou SciPy. Sa fonction rolling_mean fait le travail de manière pratique. Il renvoie également un tableau NumPy lorsque l'entrée est un tableau.

Il est difficile de battre les rolling_meanperformances avec une implémentation Python pure personnalisée. Voici un exemple de performance par rapport à deux des solutions proposées:

In [1]: import numpy as np

In [2]: import pandas as pd

In [3]: def running_mean(x, N):
   ...:     cumsum = np.cumsum(np.insert(x, 0, 0)) 
   ...:     return (cumsum[N:] - cumsum[:-N]) / N
   ...:

In [4]: x = np.random.random(100000)

In [5]: N = 1000

In [6]: %timeit np.convolve(x, np.ones((N,))/N, mode='valid')
10 loops, best of 3: 172 ms per loop

In [7]: %timeit running_mean(x, N)
100 loops, best of 3: 6.72 ms per loop

In [8]: %timeit pd.rolling_mean(x, N)[N-1:]
100 loops, best of 3: 4.74 ms per loop

In [9]: np.allclose(pd.rolling_mean(x, N)[N-1:], running_mean(x, N))
Out[9]: True

Il existe également de belles options pour gérer les valeurs de bord.

Vous pouvez calculer une moyenne courante avec:

import numpy as np

def runningMean(x, N):
    y = np.zeros((len(x),))
    for ctr in range(len(x)):
         y[ctr] = np.sum(x[ctr:(ctr+N)])
    return y/N

Mais c'est lent.

Heureusement, numpy inclut une fonction de convolution que nous pouvons utiliser pour accélérer les choses. La moyenne courante équivaut à une convolution xavec un vecteur Nlong, avec tous les membres égaux à 1/N. L'implémentation numpy de convolve inclut le transitoire de départ, vous devez donc supprimer les N-1 premiers points:

def runningMeanFast(x, N):
    return np.convolve(x, np.ones((N,))/N)[(N-1):]

Sur ma machine, la version rapide est 20 à 30 fois plus rapide, en fonction de la longueur du vecteur d'entrée et de la taille de la fenêtre de calcul de la moyenne.

Notez que convolve inclut un 'same'mode qui semble devoir résoudre le problème transitoire de départ, mais il le sépare entre le début et la fin.

ou module pour python qui calcule

dans mes tests sur Tradewave.net TA-lib gagne toujours:

import talib as ta
import numpy as np
import pandas as pd
import scipy
from scipy import signal
import time as t

PAIR = info.primary_pair
PERIOD = 30

def initialize():
    storage.reset()
    storage.elapsed = storage.get('elapsed', [0,0,0,0,0,0])

def cumsum_sma(array, period):
    ret = np.cumsum(array, dtype=float)
    ret[period:] = ret[period:] - ret[:-period]
    return ret[period - 1:] / period

def pandas_sma(array, period):
    return pd.rolling_mean(array, period)

def api_sma(array, period):
    # this method is native to Tradewave and does NOT return an array
    return (data[PAIR].ma(PERIOD))

def talib_sma(array, period):
    return ta.MA(array, period)

def convolve_sma(array, period):
    return np.convolve(array, np.ones((period,))/period, mode='valid')

def fftconvolve_sma(array, period):    
    return scipy.signal.fftconvolve(
        array, np.ones((period,))/period, mode='valid')    

def tick():

    close = data[PAIR].warmup_period('close')

    t1 = t.time()
    sma_api = api_sma(close, PERIOD)
    t2 = t.time()
    sma_cumsum = cumsum_sma(close, PERIOD)
    t3 = t.time()
    sma_pandas = pandas_sma(close, PERIOD)
    t4 = t.time()
    sma_talib = talib_sma(close, PERIOD)
    t5 = t.time()
    sma_convolve = convolve_sma(close, PERIOD)
    t6 = t.time()
    sma_fftconvolve = fftconvolve_sma(close, PERIOD)
    t7 = t.time()

    storage.elapsed[-1] = storage.elapsed[-1] + t2-t1
    storage.elapsed[-2] = storage.elapsed[-2] + t3-t2
    storage.elapsed[-3] = storage.elapsed[-3] + t4-t3
    storage.elapsed[-4] = storage.elapsed[-4] + t5-t4
    storage.elapsed[-5] = storage.elapsed[-5] + t6-t5    
    storage.elapsed[-6] = storage.elapsed[-6] + t7-t6        

    plot('sma_api', sma_api)  
    plot('sma_cumsum', sma_cumsum[-5])
    plot('sma_pandas', sma_pandas[-10])
    plot('sma_talib', sma_talib[-15])
    plot('sma_convolve', sma_convolve[-20])    
    plot('sma_fftconvolve', sma_fftconvolve[-25])

def stop():

    log('ticks....: %s' % info.max_ticks)

    log('api......: %.5f' % storage.elapsed[-1])
    log('cumsum...: %.5f' % storage.elapsed[-2])
    log('pandas...: %.5f' % storage.elapsed[-3])
    log('talib....: %.5f' % storage.elapsed[-4])
    log('convolve.: %.5f' % storage.elapsed[-5])    
    log('fft......: %.5f' % storage.elapsed[-6])

résultats:

[2015-01-31 23:00:00] ticks....: 744
[2015-01-31 23:00:00] api......: 0.16445
[2015-01-31 23:00:00] cumsum...: 0.03189
[2015-01-31 23:00:00] pandas...: 0.03677
[2015-01-31 23:00:00] talib....: 0.00700  # <<< Winner!
[2015-01-31 23:00:00] convolve.: 0.04871
[2015-01-31 23:00:00] fft......: 0.22306

Pour une solution prête à l'emploi, voir https://scipy-cookbook.readthedocs.io/items/SignalSmooth.html . Il fournit une moyenne mobile avec le flattype de fenêtre. Notez que c'est un peu plus sophistiqué que la simple méthode de convolution à faire soi-même, car elle tente de gérer les problèmes au début et à la fin des données en les reflétant (ce qui peut ou non fonctionner dans votre cas. ..).

Pour commencer, vous pouvez essayer:

a = np.random.random(100)
plt.plot(a)
b = smooth(a, window='flat')
plt.plot(b)

Vous pouvez utiliser scipy.ndimage.filters.uniform_filter1d :

import numpy as np
from scipy.ndimage.filters import uniform_filter1d
N = 1000
x = np.random.random(100000)
y = uniform_filter1d(x, size=N)

uniform_filter1d:

• donne la sortie avec la même forme numpy (c'est-à-dire le nombre de points)
• permet plusieurs façons de gérer la bordure où 'reflect'est la valeur par défaut, mais dans mon cas, je voulais plutôt'nearest'

C'est aussi assez rapide (près de 50 fois plus rapide np.convolveet 2 à 5 fois plus rapide que l'approche cumsum donnée ci-dessus ):

%timeit y1 = np.convolve(x, np.ones((N,))/N, mode='same')
100 loops, best of 3: 9.28 ms per loop

%timeit y2 = uniform_filter1d(x, size=N)
10000 loops, best of 3: 191 µs per loop

voici 3 fonctions qui vous permettent de comparer l'erreur / la vitesse de différentes implémentations:

from __future__ import division
import numpy as np
import scipy.ndimage.filters as ndif
def running_mean_convolve(x, N):
    return np.convolve(x, np.ones(N) / float(N), 'valid')
def running_mean_cumsum(x, N):
    cumsum = np.cumsum(np.insert(x, 0, 0))
    return (cumsum[N:] - cumsum[:-N]) / float(N)
def running_mean_uniform_filter1d(x, N):
    return ndif.uniform_filter1d(x, N, mode='constant', origin=-(N//2))[:-(N-1)]

Je sais que c'est une vieille question, mais voici une solution qui n'utilise aucune structure de données ou bibliothèque supplémentaire. Il est linéaire dans le nombre d'éléments de la liste d'entrée et je ne peux penser à aucun autre moyen de le rendre plus efficace (en fait, si quelqu'un connaît une meilleure façon d'allouer le résultat, faites-le moi savoir).

REMARQUE: ce serait beaucoup plus rapide en utilisant un tableau numpy au lieu d'une liste, mais je voulais éliminer toutes les dépendances. Il serait également possible d'améliorer les performances par une exécution multi-thread

La fonction suppose que la liste d'entrée est unidimensionnelle, soyez donc prudent.

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

Exemple

Supposons que nous ayons une liste data = [ 1, 2, 3, 4, 5, 6 ]sur laquelle nous voulons calculer une moyenne glissante avec une période de 3, et que vous souhaitiez également une liste de sortie de la même taille que celle d'entrée (c'est le plus souvent le cas).

Le premier élément a l'index 0, donc la moyenne mobile doit être calculée sur les éléments d'index -2, -1 et 0. Évidemment, nous n'avons pas de données [-2] et de données [-1] (sauf si vous souhaitez utiliser conditions aux limites), donc nous supposons que ces éléments sont 0. Cela équivaut à un remplissage à zéro de la liste, sauf que nous ne la remplissons pas réellement, gardons simplement une trace des indices qui nécessitent un remplissage (de 0 à N-1).

Donc, pour les N premiers éléments, nous continuons simplement à additionner les éléments dans un accumulateur.

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

A partir des éléments N + 1 vers l'avant, l'accumulation simple ne fonctionne pas. nous attendons result[3] = (2 + 3 + 4)/3 = 3mais c'est différent de (sum + 4)/3 = 3.333.

La façon de calculer la valeur correcte est de soustraire data[0] = 1de sum+4, donnant ainsi sum + 4 - 1 = 9.

Cela se produit parce qu'actuellement sum = data[0] + data[1] + data[2], mais c'est aussi vrai pour tout i >= Nparce que, avant la soustraction, sumc'est data[i-N] + ... + data[i-2] + data[i-1].

Je pense que cela peut être résolu avec élégance en utilisant un goulot d'étranglement

Voir l'exemple de base ci-dessous:

import numpy as np
import bottleneck as bn

a = np.random.randint(4, 1000, size=100)
mm = bn.move_mean(a, window=5, min_count=1)

• "mm" est la moyenne mobile de "a".

• "window" est le nombre maximum d'entrées à considérer pour la moyenne mobile.

• "min_count" est le nombre minimal d'entrées à considérer pour la moyenne mobile (par exemple pour les premiers éléments ou si le tableau a des valeurs nan).

La bonne partie est que Bottleneck aide à gérer les valeurs nanométriques et il est également très efficace.

Je n'ai pas encore vérifié à quelle vitesse c'est, mais vous pouvez essayer:

from collections import deque

cache = deque() # keep track of seen values
n = 10          # window size
A = xrange(100) # some dummy iterable
cum_sum = 0     # initialize cumulative sum

for t, val in enumerate(A, 1):
    cache.append(val)
    cum_sum += val
    if t < n:
        avg = cum_sum / float(t)
    else:                           # if window is saturated,
        cum_sum -= cache.popleft()  # subtract oldest value
        avg = cum_sum / float(n)

Cette réponse contient des solutions utilisant la bibliothèque standard Python pour trois scénarios différents.

Moyenne mobile avec itertools.accumulate

Il s'agit d'une solution Python 3.2+ efficace en termes de mémoire calculant la moyenne mobile sur un itérable de valeurs en tirant parti itertools.accumulate.

>>> from itertools import accumulate
>>> values = range(100)

Notez qu'il valuespeut être tout itérable, y compris les générateurs ou tout autre objet qui produit des valeurs à la volée.

Tout d'abord, construisez paresseusement la somme cumulée des valeurs.

>>> cumu_sum = accumulate(value_stream)

Ensuite, enumeratela somme cumulée (commençant à 1) et construisez un générateur qui donne la fraction des valeurs accumulées et l'indice d'énumération actuel.

>>> rolling_avg = (accu/i for i, accu in enumerate(cumu_sum, 1))

Vous pouvez émettre means = list(rolling_avg)si vous avez besoin de toutes les valeurs en mémoire à la fois ou appeler de manière nextincrémentielle.
(Bien sûr, vous pouvez également effectuer une itération rolling_avgavec une forboucle, qui appellera nextimplicitement.)

>>> next(rolling_avg) # 0/1
>>> 0.0
>>> next(rolling_avg) # (0 + 1)/2
>>> 0.5
>>> next(rolling_avg) # (0 + 1 + 2)/3
>>> 1.0

Cette solution peut être écrite comme une fonction comme suit.

from itertools import accumulate

def rolling_avg(iterable):
    cumu_sum = accumulate(iterable)
    yield from (accu/i for i, accu in enumerate(cumu_sum, 1))
    

Une coroutine à laquelle vous pouvez envoyer des valeurs à tout moment

Cette coroutine consomme les valeurs que vous lui envoyez et conserve une moyenne courante des valeurs vues jusqu'à présent.

Il est utile lorsque vous n'avez pas d'itérables de valeurs, mais que vous acquérez les valeurs à faire la moyenne une par une à différents moments de la vie de votre programme.

def rolling_avg_coro():
    i = 0
    total = 0.0
    avg = None

    while True:
        next_value = yield avg
        i += 1
        total += next_value
        avg = total/i
        

La coroutine fonctionne comme ceci:

>>> averager = rolling_avg_coro() # instantiate coroutine
>>> next(averager) # get coroutine going (this is called priming)
>>>
>>> averager.send(5) # 5/1
>>> 5.0
>>> averager.send(3) # (5 + 3)/2
>>> 4.0
>>> print('doing something else...')
doing something else...
>>> averager.send(13) # (5 + 3 + 13)/3
>>> 7.0

Calcul de la moyenne sur une fenêtre glissante de taille N

Cette fonction de générateur prend une taille de fenêtre itérable et de fenêtre N et donne la moyenne sur les valeurs actuelles à l'intérieur de la fenêtre. Il utilise a deque, qui est une structure de données similaire à une liste, mais optimisée pour des modifications rapides ( pop, append) aux deux points de terminaison .

from collections import deque
from itertools import islice

def sliding_avg(iterable, N):        
    it = iter(iterable)
    window = deque(islice(it, N))        
    num_vals = len(window)

    if num_vals < N:
        msg = 'window size {} exceeds total number of values {}'
        raise ValueError(msg.format(N, num_vals))

    N = float(N) # force floating point division if using Python 2
    s = sum(window)
    
    while True:
        yield s/N
        try:
            nxt = next(it)
        except StopIteration:
            break
        s = s - window.popleft() + nxt
        window.append(nxt)
        

Voici la fonction en action:

>>> values = range(100)
>>> N = 5
>>> window_avg = sliding_avg(values, N)
>>> 
>>> next(window_avg) # (0 + 1 + 2 + 3 + 4)/5
>>> 2.0
>>> next(window_avg) # (1 + 2 + 3 + 4 + 5)/5
>>> 3.0
>>> next(window_avg) # (2 + 3 + 4 + 5 + 6)/5
>>> 4.0

Un peu tard à la fête, mais j'ai créé ma propre petite fonction qui ne s'enroule PAS autour des extrémités ou des pads avec des zéros qui sont ensuite utilisés pour trouver la moyenne également. Un autre traitement est qu'il rééchantillonne également le signal en des points linéairement espacés. Personnalisez le code à volonté pour obtenir d'autres fonctionnalités.

La méthode est une simple multiplication matricielle avec un noyau gaussien normalisé.

def running_mean(y_in, x_in, N_out=101, sigma=1):
    '''
    Returns running mean as a Bell-curve weighted average at evenly spaced
    points. Does NOT wrap signal around, or pad with zeros.

    Arguments:
    y_in -- y values, the values to be smoothed and re-sampled
    x_in -- x values for array

    Keyword arguments:
    N_out -- NoOf elements in resampled array.
    sigma -- 'Width' of Bell-curve in units of param x .
    '''
    N_in = size(y_in)

    # Gaussian kernel
    x_out = np.linspace(np.min(x_in), np.max(x_in), N_out)
    x_in_mesh, x_out_mesh = np.meshgrid(x_in, x_out)
    gauss_kernel = np.exp(-np.square(x_in_mesh - x_out_mesh) / (2 * sigma**2))
    # Normalize kernel, such that the sum is one along axis 1
    normalization = np.tile(np.reshape(sum(gauss_kernel, axis=1), (N_out, 1)), (1, N_in))
    gauss_kernel_normalized = gauss_kernel / normalization
    # Perform running average as a linear operation
    y_out = gauss_kernel_normalized @ y_in

    return y_out, x_out

Une utilisation simple sur un signal sinusoïdal avec un bruit distribué normal ajouté:

Au lieu de numpy ou scipy, je recommanderais aux pandas de le faire plus rapidement:

df['data'].rolling(3).mean()

Cela prend la moyenne mobile (MA) de 3 périodes de la colonne «données». Vous pouvez également calculer les versions décalées, par exemple celle qui exclut la cellule actuelle (décalée une vers l'arrière) peut être calculée facilement comme:

df['data'].shift(periods=1).rolling(3).mean()

Il y a un commentaire de mab enfoui dans l'une des réponses ci-dessus qui a cette méthode. bottlenecka move_meanqui est une moyenne mobile simple:

import numpy as np
import bottleneck as bn

a = np.arange(10) + np.random.random(10)

mva = bn.move_mean(a, window=2, min_count=1)

min_countest un paramètre pratique qui prendra essentiellement la moyenne mobile jusqu'à ce point dans votre tableau. Si vous ne définissez pas min_count, il sera égal window, et tout jusqu'à windowpoints le sera nan.

Une autre approche pour trouver la moyenne mobile sans utiliser numpy, panda

import itertools
sample = [2, 6, 10, 8, 11, 10]
list(itertools.starmap(lambda a,b: b/a, 
               enumerate(itertools.accumulate(sample), 1)))

imprimera [2.0, 4.0, 6.0, 6.5, 7.4, 7.833333333333333]

Cette question est maintenant encore plus ancienne que lorsque NeXuS en a parlé le mois dernier, MAIS j'aime la façon dont son code traite les cas extrêmes. Cependant, comme il s'agit d'une «moyenne mobile simple», ses résultats sont en retard par rapport aux données auxquelles ils s'appliquent. Je pensais que le traitement des cas de pointe de manière plus satisfaisante que les modes de numpy valid, sameet fullpourrait être obtenue en appliquant une approche similaire à une convolution()méthode basée.

Ma contribution utilise une moyenne mobile centrale pour aligner ses résultats avec leurs données. Lorsqu'il y a trop peu de points disponibles pour que la fenêtre pleine taille puisse être utilisée, les moyennes courantes sont calculées à partir de fenêtres successivement plus petites aux bords du tableau. [En fait, à partir de fenêtres successivement plus grandes, mais c'est un détail d'implémentation.]

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

C'est relativement lent car il utilise convolve()et pourrait probablement être beaucoup amélioré par un vrai Pythonista, cependant, je pense que l'idée tient.

Il existe de nombreuses réponses ci-dessus sur le calcul d'une moyenne courante. Ma réponse ajoute deux fonctionnalités supplémentaires:

1. ignore les valeurs nan
2. calcule la moyenne des N valeurs voisines NON comprenant la valeur d'intérêt elle-même

Cette deuxième caractéristique est particulièrement utile pour déterminer les valeurs qui diffèrent d'un certain montant de la tendance générale.

J'utilise numpy.cumsum car c'est la méthode la plus rapide ( voir la réponse d'Alleo ci-dessus ).

N=10 # number of points to test on each side of point of interest, best if even
padded_x = np.insert(np.insert( np.insert(x, len(x), np.empty(int(N/2))*np.nan), 0, np.empty(int(N/2))*np.nan ),0,0)
n_nan = np.cumsum(np.isnan(padded_x))
cumsum = np.nancumsum(padded_x) 
window_sum = cumsum[N+1:] - cumsum[:-(N+1)] - x # subtract value of interest from sum of all values within window
window_n_nan = n_nan[N+1:] - n_nan[:-(N+1)] - np.isnan(x)
window_n_values = (N - window_n_nan)
movavg = (window_sum) / (window_n_values)

Ce code fonctionne uniquement pour les N pairs. Il peut être ajusté pour les nombres impairs en changeant le np.insert de padded_x et n_nan.

Exemple de sortie (raw en noir, movavg en bleu):

Ce code peut être facilement adapté pour supprimer toutes les valeurs de moyenne mobile calculées à partir de moins de cutoff = 3 valeurs non nan.

window_n_values = (N - window_n_nan).astype(float) # dtype must be float to set some values to nan
cutoff = 3
window_n_values[window_n_values<cutoff] = np.nan
movavg = (window_sum) / (window_n_values)

Utiliser uniquement la bibliothèque standard Python (mémoire efficace)

Donnez simplement une autre version de l'utilisation de la bibliothèque standard dequeuniquement. Je suis assez surpris que la plupart des réponses utilisent pandasou numpy.

def moving_average(iterable, n=3):
    d = deque(maxlen=n)
    for i in iterable:
        d.append(i)
        if len(d) == n:
            yield sum(d)/n

r = moving_average([40, 30, 50, 46, 39, 44])
assert list(r) == [40.0, 42.0, 45.0, 43.0]

En fait, j'ai trouvé une autre implémentation dans les documents python

def moving_average(iterable, n=3):
    # moving_average([40, 30, 50, 46, 39, 44]) --> 40.0 42.0 45.0 43.0
    # http://en.wikipedia.org/wiki/Moving_average
    it = iter(iterable)
    d = deque(itertools.islice(it, n-1))
    d.appendleft(0)
    s = sum(d)
    for elem in it:
        s += elem - d.popleft()
        d.append(elem)
        yield s / n

Cependant, la mise en œuvre me semble un peu plus complexe qu'elle ne devrait l'être. Mais il doit être dans la documentation python standard pour une raison, quelqu'un pourrait-il commenter l'implémentation de la mienne et la documentation standard?

Avec les variables de @ Aikude, j'ai écrit une ligne.

import numpy as np

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

mean = [np.mean(mylist[x:x+N]) for x in range(len(mylist)-N+1)]
print(mean)

>>> [2.0, 3.0, 4.0, 5.0, 6.0]

Bien qu'il existe des solutions à cette question ici, veuillez jeter un coup d'œil à ma solution. C'est très simple et fonctionne bien.

import numpy as np
dataset = np.asarray([1, 2, 3, 4, 5, 6, 7])
ma = list()
window = 3
for t in range(0, len(dataset)):
    if t+window <= len(dataset):
        indices = range(t, t+window)
        ma.append(np.average(np.take(dataset, indices)))
else:
    ma = np.asarray(ma)

En lisant les autres réponses, je ne pense pas que ce soit ce que la question demandait, mais je suis arrivé ici avec le besoin de garder une moyenne mobile d'une liste de valeurs qui augmentait en taille.

Donc, si vous souhaitez conserver une liste de valeurs que vous acquérez quelque part (un site, un appareil de mesure, etc.) et la moyenne des dernières nvaleurs mises à jour, vous pouvez utiliser le code ci-dessous, qui minimise l'effort d'ajout de nouvelles éléments:

class Running_Average(object):
    def __init__(self, buffer_size=10):
        """
        Create a new Running_Average object.

        This object allows the efficient calculation of the average of the last
        `buffer_size` numbers added to it.

        Examples
        --------
        >>> a = Running_Average(2)
        >>> a.add(1)
        >>> a.get()
        1.0
        >>> a.add(1)  # there are two 1 in buffer
        >>> a.get()
        1.0
        >>> a.add(2)  # there's a 1 and a 2 in the buffer
        >>> a.get()
        1.5
        >>> a.add(2)
        >>> a.get()  # now there's only two 2 in the buffer
        2.0
        """
        self._buffer_size = int(buffer_size)  # make sure it's an int
        self.reset()

    def add(self, new):
        """
        Add a new number to the buffer, or replaces the oldest one there.
        """
        new = float(new)  # make sure it's a float
        n = len(self._buffer)
        if n < self.buffer_size:  # still have to had numbers to the buffer.
            self._buffer.append(new)
            if self._average != self._average:  # ~ if isNaN().
                self._average = new  # no previous numbers, so it's new.
            else:
                self._average *= n  # so it's only the sum of numbers.
                self._average += new  # add new number.
                self._average /= (n+1)  # divide by new number of numbers.
        else:  # buffer full, replace oldest value.
            old = self._buffer[self._index]  # the previous oldest number.
            self._buffer[self._index] = new  # replace with new one.
            self._index += 1  # update the index and make sure it's...
            self._index %= self.buffer_size  # ... smaller than buffer_size.
            self._average -= old/self.buffer_size  # remove old one...
            self._average += new/self.buffer_size  # ...and add new one...
            # ... weighted by the number of elements.

    def __call__(self):
        """
        Return the moving average value, for the lazy ones who don't want
        to write .get .
        """
        return self._average

    def get(self):
        """
        Return the moving average value.
        """
        return self()

    def reset(self):
        """
        Reset the moving average.

        If for some reason you don't want to just create a new one.
        """
        self._buffer = []  # could use np.empty(self.buffer_size)...
        self._index = 0  # and use this to keep track of how many numbers.
        self._average = float('nan')  # could use np.NaN .

    def get_buffer_size(self):
        """
        Return current buffer_size.
        """
        return self._buffer_size

    def set_buffer_size(self, buffer_size):
        """
        >>> a = Running_Average(10)
        >>> for i in range(15):
        ...     a.add(i)
        ...
        >>> a()
        9.5
        >>> a._buffer  # should not access this!!
        [10.0, 11.0, 12.0, 13.0, 14.0, 5.0, 6.0, 7.0, 8.0, 9.0]

        Decreasing buffer size:
        >>> a.buffer_size = 6
        >>> a._buffer  # should not access this!!
        [9.0, 10.0, 11.0, 12.0, 13.0, 14.0]
        >>> a.buffer_size = 2
        >>> a._buffer
        [13.0, 14.0]

        Increasing buffer size:
        >>> a.buffer_size = 5
        Warning: no older data available!
        >>> a._buffer
        [13.0, 14.0]

        Keeping buffer size:
        >>> a = Running_Average(10)
        >>> for i in range(15):
        ...     a.add(i)
        ...
        >>> a()
        9.5
        >>> a._buffer  # should not access this!!
        [10.0, 11.0, 12.0, 13.0, 14.0, 5.0, 6.0, 7.0, 8.0, 9.0]
        >>> a.buffer_size = 10  # reorders buffer!
        >>> a._buffer
        [5.0, 6.0, 7.0, 8.0, 9.0, 10.0, 11.0, 12.0, 13.0, 14.0]
        """
        buffer_size = int(buffer_size)
        # order the buffer so index is zero again:
        new_buffer = self._buffer[self._index:]
        new_buffer.extend(self._buffer[:self._index])
        self._index = 0
        if self._buffer_size < buffer_size:
            print('Warning: no older data available!')  # should use Warnings!
        else:
            diff = self._buffer_size - buffer_size
            print(diff)
            new_buffer = new_buffer[diff:]
        self._buffer_size = buffer_size
        self._buffer = new_buffer

    buffer_size = property(get_buffer_size, set_buffer_size)

Et vous pouvez le tester avec, par exemple:

def graph_test(N=200):
    import matplotlib.pyplot as plt
    values = list(range(N))
    values_average_calculator = Running_Average(N/2)
    values_averages = []
    for value in values:
        values_average_calculator.add(value)
        values_averages.append(values_average_calculator())
    fig, ax = plt.subplots(1, 1)
    ax.plot(values, label='values')
    ax.plot(values_averages, label='averages')
    ax.grid()
    ax.set_xlim(0, N)
    ax.set_ylim(0, N)
    fig.show()

Qui donne:

Une autre solution utilisant simplement une bibliothèque standard et deque:

from collections import deque
import itertools

def moving_average(iterable, n=3):
    # http://en.wikipedia.org/wiki/Moving_average
    it = iter(iterable) 
    # create an iterable object from input argument
    d = deque(itertools.islice(it, n-1))  
    # create deque object by slicing iterable
    d.appendleft(0)
    s = sum(d)
    for elem in it:
        s += elem - d.popleft()
        d.append(elem)
        yield s / n

# example on how to use it
for i in  moving_average([40, 30, 50, 46, 39, 44]):
    print(i)

# 40.0
# 42.0
# 45.0
# 43.0

À des fins éducatives, permettez-moi d'ajouter deux autres solutions Numpy (qui sont plus lentes que la solution 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

Fonctions utilisées: as_strided , add.reduceat

Toutes les solutions susmentionnées sont médiocres car elles manquent

• vitesse due à un python natif au lieu d'une implémentation vectorisée numpy,
• stabilité numérique due à une mauvaise utilisation de numpy.cumsum, ou
• vitesse due aux O(len(x) * w)implémentations sous forme de convolutions.

Donné

import numpy
m = 10000
x = numpy.random.rand(m)
w = 1000

Notez que x_[:w].sum()égale x[:w-1].sum(). Donc, pour la première moyenne, l' numpy.cumsum(...)addition x[w] / w(via x_[w+1] / w) et la soustraction 0(de x_[0] / w). Cela se traduit parx[0:w].mean()

Via cumsum, vous mettrez à jour la deuxième moyenne en ajoutant x[w+1] / wet en soustrayant en plus x[0] / w, ce qui donne x[1:w+1].mean().

Cela continue jusqu'à ce qu'il x[-w:].mean()soit atteint.

x_ = numpy.insert(x, 0, 0)
sliding_average = x_[:w].sum() / w + numpy.cumsum(x_[w:] - x_[:-w]) / w

Cette solution est vectorisée O(m), lisible et numériquement stable.

Que diriez-vous d' un filtre de moyenne mobile ? C'est aussi un one-liner et a l'avantage, que vous pouvez facilement manipuler le type de fenêtre si vous avez besoin d'autre chose que le rectangle, ie. une moyenne mobile simple de N-long d'un tableau a:

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

Et avec la fenêtre triangulaire appliquée:

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

Remarque: Je jette généralement les N premiers échantillons comme faux donc [N:]à la fin, mais ce n'est pas nécessaire et la question d'un choix personnel uniquement.

Si vous choisissez de lancer le vôtre, plutôt que d'utiliser une bibliothèque existante, soyez conscient de l'erreur de virgule flottante et essayez de minimiser ses effets:

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

Si toutes vos valeurs sont à peu près du même ordre de grandeur, cela aidera à préserver la précision en ajoutant toujours des valeurs de magnitudes à peu près similaires.