root/lang/python/tremolo/indications/moving_average.py @ 31994

#!/usr/bin/env python
#-*- coding:utf-8 -*-

import sys, os
# TREMOLO_PATH = os.path.split( os.path.dirname(__file__) )[0]
# if TREMOLO_PATH not in sys.path: sys.path.append(TREMOLO_PATH)
# from assets.foundation import Asset
from foundation import Indication

class MovingAverage (Indication):
        """
        すべての移動平均指標の基本となるクラスです.
        """
        @classmethod
        def isRealIndication(cls): return False
        @classmethod
        def getType(cls): return float
        
        def __init__(self, asset, parent=None): Indication.__init__(self, asset, parent)
        
        def getDescription(self): return "Indication::MovingAverage"
        
        def evaluate(self, t): raise NotImplementedError
        
        def getPreviousData(self, t, n=5):
                """
                t時点からn回前までの複数のデータを取得し、リストで返します.
                """
                self.asset.lock(t)
                tp = self.asset.getType()
                dat = []
                if tp == tuple:
                        dat = [self.asset.get(t-i)[0] for i in xrange(n)]
                elif tp == int or tp == float:
                        dat = [self.asset.get(t-i) for i in xrange(n)]
                self.asset.unlock()
                return dat

class SMA(MovingAverage):
        """
        単純移動平均を表すクラスです.
        """
        @classmethod
        def isRealIndication(cls): return True
        
        def __init__(self, asset, parent=None, length=5):
                from scipy.ndimage.measurements import mean
                MovingAverage.__init__(self, asset, parent)
                if type(length) == int and length > 0: self.length = length
                else: raise TypeError("the variable 'length' is not valid.")
        
        def getDescription(self): return "Indication::SimpleMovingAverage"
        
        def evaluate(self, t):
                dat = self.getPreviousData(t, self.length)
                return mean(dat)

class WMA(MovingAverage):
        """
        加重移動平均を表すクラスです.
        """
        @classmethod
        def isRealIndication(cls): return True
        
        def __init__(self, asset, parent=None, length=5):
                MovingAverage.__init__(self, asset, parent)
                if type(length) == int and length > 0:
                        self.__length = length
                        self.__denominator = length*(length+1)/2.0
                        self.__cache_list  = [(length-i) for i in xrange(length)]
                else: raise TypeError("the variable 'length' is not valid.")
        
        def getDescription(self): return "Indication::WeightedMovingAverage"
        
        def setLength(self, length):
                if type(length) == int and length > 0:
                        self.__length = length
                        self.__denominator = length*(length+1)/2.0
                        self.__cache_list  = [(length-i) for i in xrange(length)]
                else: raise TypeError("the variable 'length' is not valid.")
        
        def evaluate(self, t):
                dat       = self.getPreviousData(t, self.__length)
                numerator = sum([x*dat[i] for (i, x) in enumerate(self.__cache_list)])
                return numerator / self.__denominator

class EMA(MovingAverage):
        """
        指数加重移動平均を表すクラスです.
        """
        @classmethod
        def isRealIndication(cls): return True
        
        def getDescription(self): return "Indication::ExponentiallyWeightedMovingAverage"
        
        def __init__(self, asset, parent=None, length=5, alpha=-1, approximation=False):
                """
                オブジェクトのコンストラクタです。
                
                length        : 過去に参照する長さの値
                alpha         : 減衰率.0から1の間で指定.-1の場合は適切な値に設定される
                approximation : 近似を行って計算を行うかどうか.デフォルトはFalse
                """
                MovingAverage.__init__(self, asset, parent)
                
                if alpha == -1 or (type(alpha)==float and 0.0 < alpha < 1.0):
                        self.__alpha = alpha
                else: raise TypeError("the variable 'alpha' is not valid.")
                if type(length) == int and length > 0:
                        self.__length = length
                else: raise TypeError("the variable 'length' is not valid.")
                
                self.__initCache()
                self.approximation = approximation
        
        def __initCache(self):
                k = self.__alpha == -1 and 2.0/(self.__length+1) or self.__alpha
                oneMinusK = 1.0 - k
                self.__cache_list = [(oneMinusK**i) for i in xrange(self.__length)]
        
        def setAlpha(self, alpha):
                if alpha == -1 or (type(alpha)==float and 0.0 < alpha < 1.0):
                        self.__alpha = alpha
                else: raise TypeError("the variable 'alpha' is not valid.")
                self.__initCache()
        
        def setLength(self, length):
                if type(length) == int and length > 0:
                        self.__length = length
                else: raise TypeError("the variable 'length' is not valid.")
                self.__initCache()
        
        def evaluate(self, t):
                n = self.__length
                dat = self.getPreviousData(t, n)
                numerator   = sum([x*dat[i] for (i, x) in enumerate(self.__cache_list)])
                if self.approximation:
                        denominator = 1.0/k
                        return numerator / denominator
                else:
                        denominator = sum(self.__cache_list)
                        return numerator / denominator