计算python中的指数移动平均线

Question

我有一系列日期和每个日期的测量.我想计算每个日期的指数移动平均线.有人知道怎么做这个吗？

我是python的新手.看起来平均值并没有内置到标准的python库中,这让我觉得有些奇怪.也许我不是在寻找合适的地方.

因此,给定以下代码,我如何计算日历日期IQ点的移动加权平均值？

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

(可能有更好的方法来构建数据,任何建议都将受到赞赏)

Answer 1

编辑:似乎mov_average_expw()来自SciKits的scikits.timeseries.lib.movi​​ng_funcs子模块的函数(补充SciPy的附加工具包)更适合您的问题的措辞.

使用平滑因子计算数据的指数平滑alpha((1 - alpha)以维基百科的术语表示):

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

以上不是很好,所以让我们重构一下:

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

例:

$ 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

我做了一些谷歌搜索,我发现了以下示例代码(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

我一直在用熊猫计算EMA:

这是一个如何做到的例子:

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)

关于熊猫EWMA的更多信息:

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

Answer 4

您还可以使用 SciPy 滤波器方法，因为 EMA 是 IIR 滤波器。与enumerate()方法相比，在我的系统上使用timeit对大型数据集进行测量时，这将具有大约 64 倍的好处。

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 5

我不了解Python，但对于平均部分，您的意思是形式为指数衰减的低通滤波器

y_new = y_old + (input - y_old)*alpha

其中alpha = dt / tau，dt =滤波器的时间步长，tau =滤波器的时间常数？（此方法的可变时步形式如下，仅将dt / tau剪辑为不大于1.0）

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

如果要过滤日期等内容，请确保将其转换为自1970年1月1日以来的秒数等浮点数。

Answer 6

我的python有点生疏(任何人都可以随意编辑这段代码进行更正,如果我以某种方式弄乱了语法),但是这里......

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

该函数从列表的末尾向后移动,通过向后工作计算每个值的指数移动平均值,直到元素的权重系数小于给定的epsilon.

在函数结束时,它会在返回列表之前反转这些值(这样它们对调用者的顺序正确).

(侧面注意:如果我使用的语言不是python,我先创建一个全尺寸的空数组,然后按顺序填充它,这样我就不必在最后反转它了.但我不是我认为你可以在python中声明一个大的空数组.而在python列表中,追加比预先添加要便宜得多,这就是为什么我以相反的顺序构建列表.如果我错了,请纠正我.)

'alpha'参数是每次迭代的衰减因子.例如,如果您使用0.5的alpha,那么今天的移动平均值将由以下加权值组成:

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

当然,如果你有大量的价值观,那么十天或十五天前的价值对今天的加权平均数的贡献不大.'epsilon'参数允许您设置一个截止点,低于该截止点​​,您将不再关注旧值(因为它们对今天的价值的贡献将是无关紧要的).

你调用这样的函数:

result = movingAverageExponential(values, 0.75, 0.0001)

Answer 7

在matplotlib.org示例(http://matplotlib.org/examples/pylab_examples/finance_work2.html)中提供了一个使用numpy的指数移动平均(EMA)函数的一个很好的例子:

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