Calcular média móvel exponencial em python

Question

Eu tenho um intervalo de datas e uma medição em cada uma dessas datas. Eu gostaria de calcular uma média móvel exponencial para cada uma das datas. Alguém sabe como fazer isso?

Eu sou novo para python. Não parece que as médias são construídos para a biblioteca python padrão, o que me parece um pouco estranho. Talvez eu não estou procurando no lugar certo.

Assim, dado o seguinte código, como eu poderia calcular o móvel ponderada média de pontos de QI para datas de calendário?

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

(provavelmente há uma maneira melhor de estruturar os dados, qualquer conselho seria apreciado)

Answer 1

EDIT: Parece que mov_average_expw() função de scikits.timeseries.lib.moving_funcs submodule de SciKits (add-on toolkits que o complemento SciPy ) melhor se adequa a formulação da sua pergunta.

---

Para calcular uma exponencial de seus dados com um alpha fator de suavização (é (1 - alpha) em termos da Wikipédia):

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

O acima não é muito, então vamos refatorar-lo um pouco:

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

Exemplo:

$ 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

Eu fiz um pouco de googling e encontrei o código de exemplo a seguir ( 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 3

Estou sempre cálculo EMAs com Pandas:

Aqui está um exemplo de como fazê-lo:

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)

Mais informações sobre Pandas EWMA:

http://pandas.pydata.org/pandas- docs / estável / gerado / pandas.ewma.html

Answer 4

O meu python é um pouco enferrujado pouco (ninguém pode se sentir livre para editar este código para fazer as correções, se eu errei a sintaxe de alguma forma), mas aqui vai ....

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 função se move para trás, a partir do fim da lista para o início, calculando a média móvel exponencial para cada valor de trabalho para trás até que o coeficiente de peso para um elemento é menor do que a dada epsilon.

No final da função, ele inverte os valores antes de retornar a lista (de modo que eles estão na ordem correta para o chamador).

(SIDE NOTA: se eu estava usando um idioma diferente do python, eu criar uma matriz vazia de tamanho completo em primeiro lugar e, em seguida, preenchê-lo para trás-fim, de modo que eu não teria que reverter isso no final. mas eu não acho que você pode declarar um array vazio grande em python. E em listas de python, anexando é muito menos dispendioso do que prepending, razão pela qual eu construí a lista em ordem inversa. por favor, me corrija se eu estiver errado).

O argumento 'alfa' é o fator de decaimento em cada iteração. Por exemplo, se você usou um alfa de 0,5, então o valor de média móvel de hoje seria composto dos seguintes valores ponderados:

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

Claro que, se você tem uma enorme variedade de valores, os valores de dez ou quinze dias atrás não vai contribuir muito para média ponderada de hoje. O argumento 'epsilon' permite que você defina um ponto de corte, abaixo do qual você deixará de se preocupar com valores antigos (uma vez que a sua contribuição para o valor de hoje será insignificante).

Você iria invocar a função de algo como isto:

result = movingAverageExponential(values, 0.75, 0.0001)

Answer 5

Nos exemplos matplotlib.org ( http://matplotlib.org/examples/pylab_examples/finance_work2. html ) é fornecido um exemplo de boa função móvel exponencial médio (EMA) usando 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 6

Eu não sei Python, mas para a parte média, que quer dizer um decaimento exponencial filtro low-pass do formulário

y_new = y_old + (input - y_old)*alpha

onde alfa = dt / tau, dt = o instante temporal de filtro, tau = a constante de tempo do filtro? (A forma variável-iteração deste é como se segue, apenas clip dt / tau para não ser mais do que 1.0)

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

Se você quiser algo filtro como uma data, certifique-se de converter a uma quantidade de ponto flutuante como # de segundos desde 1 de janeiro de 1970.

Answer 7

Você também pode usar o método do filtro SciPy porque o EMA é um filtro IIR. Isto terá a vantagem de ser cerca de 64 vezes mais rápido como medido no meu sistema usando timeit em grandes conjuntos de dados quando comparado com o enumerate () abordagem.

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 8

Eu encontrei o trecho de código acima por @earino bastante útil - mas eu precisava de algo que poderia continuamente suavizar um fluxo de valores - então eu reformulado-lo para isto:

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

e eu usá-lo como este:

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

(onde pin.read () produz o valor seguinte eu gostaria de consumir).

Answer 9

Aqui está um exemplo simples que eu trabalhei com base nas http: // stockcharts.com/school/doku.php?id=chart_school:technical_indicators:moving_averages

Note que, ao contrário em sua planilha, eu não calcular a SMA, e eu não esperar para gerar o EMA depois de 10 amostras. Isso significa que os meus valores diferem ligeiramente, mas se você traçar isso, segue-se exatamente após 10 amostras. Durante as primeiras 10 amostras, o EMA calculo é apropriadamente alisado.

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 10

Uma maneira rápida (copiar-colar de aqui ) é o seguinte:

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

Answer 11

Eu estou usando uma lista e uma taxa de decaimento como entradas. Espero que este pequeno função com apenas duas linhas pode ajudá-lo aqui, considerando recursão profunda não é estável em python.

def expma(aseries, ratio):
    return sum([ratio*aseries[-x-1]*((1-ratio)**x) for x in range(len(aseries))])