calcular promedio móvil exponencial en python

Question

Tengo un rango de fechas y una medición en cada una de esas fechas. Me gustaría calcular un promedio móvil exponencial para cada una de las fechas. ¿Alguien sabe cómo hacer esto?

Soy nuevo en python. No parece que los promedios estén integrados en la biblioteca estándar de Python, lo que me parece un poco extraño. Tal vez no estoy buscando en el lugar correcto.

Por lo tanto, dado el siguiente código, ¿cómo puedo calcular el promedio móvil ponderado de los puntos de IQ para las fechas del calendario?

from datetime import date 
days = [date(2008,1,1), date(2008,1,2), date(2008,1,7)] 
IQ = [110, 105, 90] 

(es probable que haya una mejor manera de estructurar los datos, cualquier consejo sería apreciado)

Answer 1

EDIT: Parece que mov_average_expw() función de scikits.timeseries.lib.moving_funcs submódulo de SciKits (add-on kits de herramientas que complementan SciPy) se adapte mejor a la formulación de su pregunta .

---

para calcular un exponential smoothing de sus datos con un factor de suavizado alpha (es (1 - alpha) en términos de Wikipedia):

>>> alpha = 0.5 
>>> assert 0 < alpha <= 1.0 
>>> av = sum(alpha**n.days * iq 
...  for n, iq in map(lambda (day, iq), today=max(days): (today-day, iq), 
...   sorted(zip(days, IQ), key=lambda p: p[0], reverse=True))) 
95.0 

Lo anterior no es bastante, así que vamos a refactorizar un poco:

from collections import namedtuple 
from operator import itemgetter 

def smooth(iq_data, alpha=1, today=None): 
    """Perform exponential smoothing with factor `alpha`. 

    Time period is a day. 
    Each time period the value of `iq` drops `alpha` times. 
    The most recent data is the most valuable one. 
    """ 
    assert 0 < alpha <= 1 

    if alpha == 1: # no smoothing 
     return sum(map(itemgetter(1), iq_data)) 

    if today is None: 
     today = max(map(itemgetter(0), iq_data)) 

    return sum(alpha**((today - date).days) * iq for date, iq in iq_data) 

IQData = namedtuple("IQData", "date iq") 

if __name__ == "__main__": 
    from datetime import date 

    days = [date(2008,1,1), date(2008,1,2), date(2008,1,7)] 
    IQ = [110, 105, 90] 
    iqdata = list(map(IQData, days, IQ)) 
    print("\n".join(map(str, iqdata))) 

    print(smooth(iqdata, alpha=0.5)) 

Ejemplo:

$ python26 smooth.py 
IQData(date=datetime.date(2008, 1, 1), iq=110) 
IQData(date=datetime.date(2008, 1, 2), iq=105) 
IQData(date=datetime.date(2008, 1, 7), iq=90) 
95.0 

Answer 2

No sé Python, pero para la parte de promedio, Qué quiere decir un paso bajo de forma exponencial en descomposición filtro de la forma

y_new = y_old + (input - y_old)*alpha 

donde alfa = dt/tau, dt = el paso de tiempo del filtro, tau = la constante de tiempo del filtro? (La forma-paso de tiempo variable de este es la siguiente, apenas acorte dt/tau no sea más que 1,0)

y_new = y_old + (input - y_old)*dt/tau 

Si desea filtrar algo así como una fecha, asegúrese de convertir a un punto flotante cantidad como # de segundos desde ene 1 1970.

Answer 3

hice un poco de google y me encontré con el siguiente código de muestra (http://osdir.com/ml/python.matplotlib.general/2005-04/msg00044.html):

def ema(s, n): 
    """ 
    returns an n period exponential moving average for 
    the time series s 

    s is a list ordered from oldest (index 0) to most 
    recent (index -1) 
    n is an integer 

    returns a numeric array of the exponential 
    moving average 
    """ 
    s = array(s) 
    ema = [] 
    j = 1 

    #get n sma first and calculate the next n period ema 
    sma = sum(s[:n])/n 
    multiplier = 2/float(1 + n) 
    ema.append(sma) 

    #EMA(current) = ((Price(current) - EMA(prev)) x Multiplier) + EMA(prev) 
    ema.append(((s[n] - sma) * multiplier) + sma) 

    #now calculate the rest of the values 
    for i in s[n+1:]: 
     tmp = ((i - ema[j]) * multiplier) + ema[j] 
     j = j + 1 
     ema.append(tmp) 

    return ema 

Answer 4

Mi Python es un poco oxidado (cualquiera puede sentirse libre para editar este código para hacer correcciones, si he estropeado la sintaxis Mehow), pero aquí va ....

def movingAverageExponential(values, alpha, epsilon = 0): 

    if not 0 < alpha < 1: 
     raise ValueError("out of range, alpha='%s'" % alpha) 

    if not 0 <= epsilon < alpha: 
     raise ValueError("out of range, epsilon='%s'" % epsilon) 

    result = [None] * len(values) 

    for i in range(len(result)): 
     currentWeight = 1.0 

     numerator  = 0 
     denominator = 0 
     for value in values[i::-1]: 
      numerator  += value * currentWeight 
      denominator += currentWeight 

      currentWeight *= alpha 
      if currentWeight < epsilon: 
       break 

     result[i] = numerator/denominator 

    return result 

Esta función se mueve hacia atrás, desde el final de la lista para el inicio, el cálculo de la media móvil exponencial para cada valor trabajando hacia atrás hasta que el coeficiente de peso para un elemento es menor que el épsilon dado.

Al final de la función, invierte los valores antes de devolver la lista (para que estén en el orden correcto para la persona que llama).

(NOTA LATERAL: si estuviera usando un lenguaje que no sea Python, primero crearía una matriz vacía de tamaño completo y luego la rellenaría para que no tuviera que invertirla al final Pero no creo que puedas declarar una gran matriz vacía en Python. Y en las listas de Python, agregar es mucho menos costoso que anteponer, por lo que construí la lista en orden inverso. Corrígeme si me equivoco.)

El argumento 'alpha' es el factor de disminución en cada iteración. Por ejemplo, si se ha utilizado un alfa de 0,5, entonces el valor promedio móvil de hoy en día se compone de los siguientes valores ponderados:

today:  1.0 
yesterday: 0.5 
2 days ago: 0.25 
3 days ago: 0.125 
...etc... 

Por supuesto, si usted tiene una enorme variedad de valores, los valores de diez o hace quince días no contribuirá mucho al promedio ponderado de hoy. El argumento 'épsilon' le permite establecer un punto de corte, debajo del cual dejará de preocuparse por los valores antiguos (ya que su contribución al valor actual será insignificante).

Se podría invocar la función de algo como esto:

result = movingAverageExponential(values, 0.75, 0.0001) 

Answer 5

he encontrado el fragmento de código anterior por @earino muy útil - pero necesitaba algo que podría suavizar continuamente un flujo de valores - así que refactorizado a esto:

def exponential_moving_average(period=1000): 
    """ Exponential moving average. Smooths the values in v over ther period. Send in values - at first it'll return a simple average, but as soon as it's gahtered 'period' values, it'll start to use the Exponential Moving Averge to smooth the values. 
    period: int - how many values to smooth over (default=100). """ 
    multiplier = 2/float(1 + period) 
    cum_temp = yield None # We are being primed 

    # Start by just returning the simple average until we have enough data. 
    for i in xrange(1, period + 1): 
     cum_temp += yield cum_temp/float(i) 

    # Grab the timple avergae 
    ema = cum_temp/period 

    # and start calculating the exponentially smoothed average 
    while True: 
     ema = (((yield ema) - ema) * multiplier) + ema 

y lo uso como esto:

def temp_monitor(pin): 
    """ Read from the temperature monitor - and smooth the value out. The sensor is noisy, so we use exponential smoothing. """ 
    ema = exponential_moving_average() 
    next(ema) # Prime the generator 

    while True: 
     yield ema.send(val_to_temp(pin.read())) 

(donde pin.read() produce el siguiente valor que me gustaría consumir).

Answer 6

En los ejemplos matplotlib.org (http://matplotlib.org/examples/pylab_examples/finance_work2.html) se proporciona un buen ejemplo de la función móvil exponencial media (EMA) utilizando numpy:

def moving_average(x, n, type): 
    x = np.asarray(x) 
    if type=='simple': 
     weights = np.ones(n) 
    else: 
     weights = np.exp(np.linspace(-1., 0., n)) 

    weights /= weights.sum() 

    a = np.convolve(x, weights, mode='full')[:len(x)] 
    a[:n] = a[n] 
    return a 

Answer 7

Este es un ejemplo sencillo que trabajé hasta basado en http://stockcharts.com/school/doku.php?id=chart_school:technical_indicators:moving_averages

Tenga en cuenta que a diferencia de su hoja de cálculo, no calculo el SMA, y no espero para generar el EMA después de 10 muestras. Esto significa que mis valores difieren ligeramente, pero si lo grafica, sigue exactamente después de 10 muestras. Durante las primeras 10 muestras, el EMA que calculo se suaviza adecuadamente.

def emaWeight(numSamples): 
    return 2/float(numSamples + 1) 

def ema(close, prevEma, numSamples): 
    return ((close-prevEma) * emaWeight(numSamples)) + prevEma 

samples = [ 
22.27, 22.19, 22.08, 22.17, 22.18, 22.13, 22.23, 22.43, 22.24, 22.29, 
22.15, 22.39, 22.38, 22.61, 23.36, 24.05, 23.75, 23.83, 23.95, 23.63, 
23.82, 23.87, 23.65, 23.19, 23.10, 23.33, 22.68, 23.10, 22.40, 22.17, 
] 
emaCap = 10 
e=samples[0] 
for s in range(len(samples)): 
    numSamples = emaCap if s > emaCap else s 
    e = ema(samples[s], e, numSamples) 
    print e 

Answer 8

Siempre estoy calculando EMA con las pandas:

Aquí es un ejemplo de cómo hacerlo:

import pandas as pd 
import numpy as np 

def ema(values, period): 
    values = np.array(values) 
    return pd.ewma(values, span=period)[-1] 

values = [9, 5, 10, 16, 5] 
period = 5 

print ema(values, period) 

mas informaciones sobre pandas EWMA:

http://pandas.pydata.org/pandas-docs/stable/generated/pandas.ewma.html

Answer 9

También puede usar el método de filtro SciPy porque el EMA es un filtro IIR. Esto tendrá el beneficio de ser aproximadamente 64 veces más rápido que el medido en mi sistema usando timeit en grandes conjuntos de datos en comparación con el enumerate() enfoque.

import numpy as np 
from scipy.signal import lfilter 

x = np.random.normal(size=1234) 
alpha = .1 # smoothing coefficient 
zi = [x[0]] # seed the filter state with first value 
# filter can process blocks of continuous data if <zi> is maintained 
y, zi = lfilter([1.-alpha], [1., -alpha], x, zi=zi) 

Answer 10

Una manera rápida (copiar-pegar de here) es la siguiente:

def ExpMovingAverage(values, window): 
    """ Numpy implementation of EMA 
    """ 
    weights = np.exp(np.linspace(-1., 0., window)) 
    weights /= weights.sum() 
    a = np.convolve(values, weights, mode='full')[:len(values)] 
    a[:window] = a[window] 
    return a