calcola la media mobile esponenziale in python

Question

Ho un intervallo di date e una misurazione su ciascuna di tali date. Vorrei calcolare una media mobile esponenziale per ciascuna delle date. Qualcuno sa come farlo?

Sono nuovo di Python. Non sembra che le medie siano integrate nella libreria standard di Python, il che mi sembra un po 'strano. Forse non sto cercando nel posto giusto.

Quindi, dato il seguente codice, come posso calcolare la media ponderata mobile dei punti QI per le date del calendario?

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

(c'è probabilmente un modo migliore per strutturare i dati, qualsiasi consiglio sarebbe apprezzato)

Answer 1

EDIT: Sembra che mov_average_expw() funzione da scikits.timeseries.lib.moving_funcs sottomodulo da SciKits (kit di strumenti aggiuntivi che completano SciPy ) si adatta meglio alla formulazione della tua domanda.

---

Per calcolare un smoothing esponenziale dei tuoi dati con un fattore di smoothing alpha (è (1 - alpha) nei termini di 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

Quanto sopra non è carino, quindi riformattiamolo un po ':

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

Esempio:

$ 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

Ho fatto un po 'di ricerche su Google e ho trovato il seguente codice di esempio ( 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

Calcolo sempre gli EMA con i panda:

Ecco un esempio su come farlo:

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)

Ulteriori informazioni su Pandas EWMA:

http://pandas.pydata.org/pandas- docs / stabile / generata / pandas.ewma.html

Answer 4

Il mio pitone è un po 'arrugginito (chiunque può sentirsi libero di modificare questo codice per apportare correzioni, se in qualche modo ho incasinato la sintassi), ma qui 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

Questa funzione si sposta all'indietro, dalla fine dell'elenco all'inizio, calcolando la media mobile esponenziale per ciascun valore lavorando all'indietro fino a quando il coefficiente di peso per un elemento è inferiore al epsilon dato.

Alla fine della funzione, inverte i valori prima di restituire l'elenco (in modo che siano nell'ordine corretto per il chiamante).

(NOTA LATERALE: se usassi un linguaggio diverso da Python, creerei prima un array vuoto a grandezza naturale e poi lo riempirei all'indietro, in modo da non doverlo invertire alla fine. Ma non credo che tu possa dichiarare un grande array vuoto in Python. E negli elenchi Python, aggiungere è molto meno costoso di anteporre, motivo per cui ho creato l'elenco in ordine inverso. Correggimi se sbaglio.)

L'argomento 'alpha' è il fattore di decadimento su ogni iterazione. Ad esempio, se hai utilizzato un valore alfa di 0,5, il valore della media mobile di oggi sarebbe composto dai seguenti valori ponderati:

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

Naturalmente, se hai una vasta gamma di valori, i valori di dieci o quindici giorni fa non contribuiranno molto alla media ponderata di oggi. L'argomento 'epsilon' ti consente di impostare un punto di interruzione, al di sotto del quale cesserai di preoccuparti dei vecchi valori (poiché il loro contributo al valore di oggi sarà insignificante).

Dovresti invocare la funzione in questo modo:

result = movingAverageExponential(values, 0.75, 0.0001)

Answer 5

In esempi matplotlib.org ( http://matplotlib.org/examples/pylab_examples/finance_work2. html ) viene fornito un buon esempio della funzione di media mobile esponenziale (EMA) utilizzando 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

Non conosco Python, ma per la parte media si intende un filtro passa-basso in decadenza esponenziale del modulo

y_new = y_old + (input - y_old)*alpha

dove alpha = dt / tau, dt = il timestep del filtro, tau = la costante di tempo del filtro? (la forma a tempo variabile di questo è la seguente, basta tagliare dt / tau per non essere più di 1.0)

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

Se vuoi filtrare qualcosa come una data, assicurati di convertirlo in una quantità in virgola mobile come # di secondi dal 1 ° gennaio 1970.

Answer 7

Puoi anche usare il metodo di filtro SciPy perché l'EMA è un filtro IIR. Ciò avrà il vantaggio di essere circa 64 volte più veloce rispetto al mio sistema usando timeit su grandi set di dati rispetto all'approccio enumerate () .

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

Ho trovato abbastanza utile lo snippet di codice sopra di @earino - ma avevo bisogno di qualcosa che potesse continuamente smussare un flusso di valori - quindi l'ho rifattorizzato in questo modo:

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 lo uso così:

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

(dove pin.read () produce il prossimo valore che vorrei consumare).

Answer 9

Ecco un semplice esempio che ho elaborato basato su http: // stockcharts.com/school/doku.php?id=chart_school:technical_indicators:moving_averages

Nota che, diversamente dal loro foglio di calcolo, non calcolo lo SMA e non aspetto di generare l'EMA dopo 10 campioni. Ciò significa che i miei valori differiscono leggermente, ma se lo traccia, segue esattamente dopo 10 campioni. Durante i primi 10 campioni, l'EMA I calcolato viene adeguatamente levigato.

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

Un modo rapido (copiato e incollato da qui ) è il seguente:

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

Sto usando un elenco e un tasso di decadimento come input. Spero che questa piccola funzione con solo due righe possa aiutarti qui, considerando che la ricorsione profonda non è stabile in Python.

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