我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用pandas.ewma()。
def calculate_atr(self, window=14): """ Average True Range Shows volatility of market ATR(t) = ((n-1) * ATR(t-1) + Tr(t)) / n where Tr(t) = Max(Abs(High - Low), Abs(High - Close(t - 1)), Abs(Low - Close(t - 1)); """ self.set_max_win(window) i = 0 tr_l = [0] for i in range(self.df.index[-1]): tr = max(self.df.get_value(i + 1, 'High'), self.df.get_value(i, 'Adj_Close')) - min(self.df.get_value(i + 1, 'Low'), self.df.get_value(i, 'Adj_Close')) tr_l.append(tr) name = 'atr_%s' % (window) self.df[name] = pd.ewma(pd.Series(tr_l), span=window, min_periods=window) return name
def calculate_trix(self, window=15): """ Triple Exponential Moving Average Smooth the insignificant movements TR(t) / TR(t-1) where TR(t) = EMA(EMA(EMA(Price(t)))) over n days period """ self.set_max_win(window) #ignore produced warnings for now #TODO ema = pd.ewma(self.df['Adj_Close'], span=window, min_periods=window-1) ema = pd.ewma(ema, span=window, min_periods=window-1) ema = pd.ewma(ema, span=window, min_periods=window-1) roc_l = [0] for i in range(1, len(self.df.index) - 1): roc_l.append((ema[i + 1] - ema[i]) / ema[i]) name = "trix_%s" % (window) self.df[name] = pd.Series(roc_l) return name
def RSI(df, n): i = 0 UpI = [0] DoI = [0] while i + 1 <= df.index[-1]: UpMove = df.get_value(i + 1, 'High') - df.get_value(i, 'High') DoMove = df.get_value(i, 'Low') - df.get_value(i + 1, 'Low') if UpMove > DoMove and UpMove > 0: UpD = UpMove else: UpD = 0 UpI.append(UpD) if DoMove > UpMove and DoMove > 0: DoD = DoMove else: DoD = 0 DoI.append(DoD) i = i + 1 UpI = pd.Series(UpI) DoI = pd.Series(DoI) PosDI = pd.Series(pd.ewma(UpI, span = n, min_periods = n - 1)) NegDI = pd.Series(pd.ewma(DoI, span = n, min_periods = n - 1)) RSI = pd.Series(PosDI / (PosDI + NegDI), name = 'RSI_' + str(n)) df = df.join(RSI) return df #True Strength Index
def main(): parser = argparse.ArgumentParser() parser.add_argument('files', metavar='filename', type=str, nargs='*') args = parser.parse_args() args = vars(args) files = args['files'] assert len(files) == 2 targets_df = pd.read_csv(files[0], header=0, index_col=False) predict_df = pd.read_csv(files[1], header=0, index_col=False) column = targets_df.columns[1] targets = targets_df.as_matrix(columns=[column]) #predict_df[column] = pd.ewma(predict_df[column], com=1, adjust=False) predictions = predict_df.as_matrix(columns=[column]) rmse, mse = calc_rmse(predictions, targets) print("RMSE: %f, MSE: %f" % (rmse, mse))
def EMA(self, param): return pd.ewma(param[0], span=param[1], adjust = True)
def MEMA(self, param): return pd.ewma(param[0], span=param[1] * 2 - 1, adjust = False)
def SMA(self, param): M = param[2] if param[2] == 0: M = 1 return pd.ewma(param[0], span = 2 * param[1] / M - 1)
def ema(f, c, p = 20): r"""Calculate the mean on a rolling basis. Parameters ---------- f : pandas.DataFrame Dataframe containing the column ``c``. c : str Name of the column in the dataframe ``f``. p : int The period over which to calculate the rolling mean. Returns ------- new_column : pandas.Series (float) The array containing the new feature. References ---------- *An exponential moving average (EMA) is a type of moving average that is similar to a simple moving average, except that more weight is given to the latest data* [IP_EMA]_. .. [IP_EMA] http://www.investopedia.com/terms/e/ema.asp """ new_column = pd.ewma(f[c], span=p) return new_column # # Function maratio #
def rolling_weighted_mean(series, window=200, min_periods=None): min_periods = window if min_periods is None else min_periods try: return series.ewm(span=window, min_periods=min_periods).mean() except BaseException: return pd.ewma(series, span=window, min_periods=min_periods) # ---------------------------------------------
def ATR(df, n = 20): tr = TR(df) ts_atr = pd.ewma(tr, span=n, min_periods = n-1, adjust = False) ts_atr.name = 'ATR'+str(n) return ts_atr
def TRIX(df, n): EX1 = pd.ewma(df['close'], span = n, min_periods = n - 1, adjust = False) EX2 = pd.ewma(EX1, span = n, min_periods = n - 1, adjust = False) EX3 = pd.ewma(EX2, span = n, min_periods = n - 1, adjust = False) return pd.Series(EX3/EX3.shift(1) - 1, name = 'Trix' + str(n)) #Average Directional Movement Index
def ADX(df, n): return pd.Series(talib.ADX(df['high'].values, df['low'].values, df['close'].values, timeperiod = n), index = df.index, name = 'ADX_%s' % str(n)) # UpMove = df['high'] - df['high'].shift(1) # DoMove = df['low'].shift(1) - df['low'] # UpD = pd.Series(UpMove) # DoD = pd.Series(DoMove) # UpD[(UpMove<=DoMove)|(UpMove <= 0)] = 0 # DoD[(DoMove<=UpMove)|(DoMove <= 0)] = 0 # ATRs = ATR(df,span = n, min_periods = n) # PosDI = pd.Series(pd.ewma(UpD, span = n, min_periods = n - 1) / ATRs) # NegDI = pd.Series(pd.ewma(DoD, span = n, min_periods = n - 1) / ATRs) # ADX = pd.Series(pd.ewma(abs(PosDI - NegDI) / (PosDI + NegDI), span = n_ADX, min_periods = n_ADX - 1), name = 'ADX' + str(n) + '_' + str(n_ADX)) # return ADX
def MACD(df, n_fast, n_slow, n_signal): EMAfast = pd.Series(pd.ewma(df['close'], span = n_fast, min_periods = n_slow - 1)) EMAslow = pd.Series(pd.ewma(df['close'], span = n_slow, min_periods = n_slow - 1)) MACD = pd.Series(EMAfast - EMAslow, name = 'MACD' + str(n_fast) + '_' + str(n_slow) + '_' + str(n_signal)) MACDsig = pd.Series(pd.ewma(MACD, span = n_signal, min_periods = n_signal - 1), name = 'MACDsig' + str(n_fast) + '_' + str(n_slow) + '_' + str(n_signal)) MACDhist = pd.Series(MACD - MACDsig, name = 'MACDhist' + str(n_fast) + '_' + str(n_slow) + '_' + str(n_signal)) return pd.concat([MACD, MACDsig, MACDhist], join='outer', axis=1)
def MassI(df): Range = df['high'] - df['low'] EX1 = pd.ewma(Range, span = 9, min_periods = 8) EX2 = pd.ewma(EX1, span = 9, min_periods = 8) Mass = EX1 / EX2 MassI = pd.Series(pd.rolling_sum(Mass, 25), name = 'MassIndex') return MassI #Vortex Indicator
def TSI(df, r, s): M = pd.Series(df['close'].diff(1)) aM = abs(M) EMA1 = pd.Series(pd.ewma(M, span = r, min_periods = r - 1)) aEMA1 = pd.Series(pd.ewma(aM, span = r, min_periods = r - 1)) EMA2 = pd.Series(pd.ewma(EMA1, span = s, min_periods = s - 1)) aEMA2 = pd.Series(pd.ewma(aEMA1, span = s, min_periods = s - 1)) TSI = pd.Series(EMA2 / aEMA2, name = 'TSI' + str(r) + '_' + str(s)) return TSI #Accumulation/Distribution
def Chaikin(df): ad = (2 * df['close'] - df['high'] - df['low']) / (df['high'] - df['low']) * df['volume'] Chaikin = pd.Series(pd.ewma(ad, span = 3, min_periods = 2) - pd.ewma(ad, span = 10, min_periods = 9), name = 'Chaikin') return Chaikin #Money Flow Index and Ratio
def COPP(df, n): M = df['close'].diff(int(n * 11 / 10) - 1) N = df['close'].shift(int(n * 11 / 10) - 1) ROC1 = M / N M = df['close'].diff(int(n * 14 / 10) - 1) N = df['close'].shift(int(n * 14 / 10) - 1) ROC2 = M / N Copp = pd.Series(pd.ewma(ROC1 + ROC2, span = n, min_periods = n), name = 'Copp' + str(n)) return Copp #Keltner Channel
def FISHER(df, n, smooth_p = 0.7, smooth_i = 0.7): roll_high = pd.rolling_max(df.high, n) roll_low = pd.rolling_min(df.low, n) price_loc = (df.close - roll_low)/(roll_high - roll_low) * 2.0 - 1 sm_price = pd.Series(pd.ewma(price_loc, com = 1.0/smooth_p - 1, adjust = False), name = 'FISHER_P') fisher_ind = 0.5 * np.log((1 + sm_price)/(1 - sm_price)) sm_fisher = pd.Series(pd.ewma(fisher_ind, com = 1.0/smooth_i - 1, adjust = False), name = 'FISHER_I') return pd.concat([sm_price, sm_fisher], join='outer', axis=1)
def TEMA(ts, n): n = int(n) ts_ema1 = pd.Series( pd.ewma(ts, span = n, adjust = False), name = 'EMA' + str(n) ) ts_ema2 = pd.Series( pd.ewma(ts_ema1, span = n, adjust = False), name = 'EMA2' + str(n) ) ts_ema3 = pd.Series( pd.ewma(ts_ema2, span = n, adjust = False), name = 'EMA3' + str(n) ) ts_tema = pd.Series( 3 * ts_ema1 - 3 * ts_ema2 + ts_ema3, name = 'TEMA' + str(n) ) return ts_tema
def EMA(df, n): EMA = pd.Series(pd.ewma(df['Close'], span = n, min_periods = n - 1), name = 'EMA_' + str(n)) df = df.join(EMA) return df #Momentum
def ATR(df, n): i = 0 TR_l = [0] while i < df.index[-1]: TR = max(df.get_value(i + 1, 'High'), df.get_value(i, 'Close')) - min(df.get_value(i + 1, 'Low'), df.get_value(i, 'Close')) TR_l.append(TR) i = i + 1 TR_s = pd.Series(TR_l) ATR = pd.Series(pd.ewma(TR_s, span = n, min_periods = n), name = 'ATR_' + str(n)) df = df.join(ATR) return df #Bollinger Bands
def STO(df, n): SOk = pd.Series((df['Close'] - df['Low']) / (df['High'] - df['Low']), name = 'SO%k') SOd = pd.Series(pd.ewma(SOk, span = n, min_periods = n - 1), name = 'SO%d_' + str(n)) df = df.join(SOd) return df #Trix
def TRIX(df, n): EX1 = pd.ewma(df['Close'], span = n, min_periods = n - 1) EX2 = pd.ewma(EX1, span = n, min_periods = n - 1) EX3 = pd.ewma(EX2, span = n, min_periods = n - 1) i = 0 ROC_l = [0] while i + 1 <= df.index[-1]: ROC = (EX3[i + 1] - EX3[i]) / EX3[i] ROC_l.append(ROC) i = i + 1 Trix = pd.Series(ROC_l, name = 'Trix_' + str(n)) df = df.join(Trix) return df #Average Directional Movement Index
def ADX(df, n, n_ADX): i = 0 UpI = [] DoI = [] while i + 1 <= df.index[-1]: UpMove = df.get_value(i + 1, 'High') - df.get_value(i, 'High') DoMove = df.get_value(i, 'Low') - df.get_value(i + 1, 'Low') if UpMove > DoMove and UpMove > 0: UpD = UpMove else: UpD = 0 UpI.append(UpD) if DoMove > UpMove and DoMove > 0: DoD = DoMove else: DoD = 0 DoI.append(DoD) i = i + 1 i = 0 TR_l = [0] while i < df.index[-1]: TR = max(df.get_value(i + 1, 'High'), df.get_value(i, 'Close')) - min(df.get_value(i + 1, 'Low'), df.get_value(i, 'Close')) TR_l.append(TR) i = i + 1 TR_s = pd.Series(TR_l) ATR = pd.Series(pd.ewma(TR_s, span = n, min_periods = n)) UpI = pd.Series(UpI) DoI = pd.Series(DoI) PosDI = pd.Series(pd.ewma(UpI, span = n, min_periods = n - 1) / ATR) NegDI = pd.Series(pd.ewma(DoI, span = n, min_periods = n - 1) / ATR) ADX = pd.Series(pd.ewma(abs(PosDI - NegDI) / (PosDI + NegDI), span = n_ADX, min_periods = n_ADX - 1), name = 'ADX_' + str(n) + '_' + str(n_ADX)) df = df.join(ADX) return df #MACD, MACD Signal and MACD difference
def MassI(df): Range = df['High'] - df['Low'] EX1 = pd.ewma(Range, span = 9, min_periods = 8) EX2 = pd.ewma(EX1, span = 9, min_periods = 8) Mass = EX1 / EX2 MassI = pd.Series(pd.rolling_sum(Mass, 25), name = 'Mass Index') df = df.join(MassI) return df #Vortex Indicator: http://www.vortexindicator.com/VFX_VORTEX.PDF
def TSI(df, r, s): M = pd.Series(df['Close'].diff(1)) aM = abs(M) EMA1 = pd.Series(pd.ewma(M, span = r, min_periods = r - 1)) aEMA1 = pd.Series(pd.ewma(aM, span = r, min_periods = r - 1)) EMA2 = pd.Series(pd.ewma(EMA1, span = s, min_periods = s - 1)) aEMA2 = pd.Series(pd.ewma(aEMA1, span = s, min_periods = s - 1)) TSI = pd.Series(EMA2 / aEMA2, name = 'TSI_' + str(r) + '_' + str(s)) df = df.join(TSI) return df #Accumulation/Distribution
def Chaikin(df): ad = (2 * df['Close'] - df['High'] - df['Low']) / (df['High'] - df['Low']) * df['Volume'] Chaikin = pd.Series(pd.ewma(ad, span = 3, min_periods = 2) - pd.ewma(ad, span = 10, min_periods = 9), name = 'Chaikin') df = df.join(Chaikin) return df #Money Flow Index and Ratio
def COPP(df, n): M = df['Close'].diff(int(n * 11 / 10) - 1) N = df['Close'].shift(int(n * 11 / 10) - 1) ROC1 = M / N M = df['Close'].diff(int(n * 14 / 10) - 1) N = df['Close'].shift(int(n * 14 / 10) - 1) ROC2 = M / N Copp = pd.Series(pd.ewma(ROC1 + ROC2, span = n, min_periods = n), name = 'Copp_' + str(n)) df = df.join(Copp) return df #Keltner Channel
def draw_trend(timeSeries, size): rol_mean = pd.rolling_mean(timeSeries,size) rol_weighted_mean = pd.ewma(timeSeries, span=size) # f = plt.figure(facecolor='white') # plt.plot(timeSeries,'b',rol_mean,'g',rol_weighted_mean,'r') # plt.legend(loc='best') # plt.title('Rolling Mean') # plt.show() return rol_weighted_mean
def calculate_ema(self, window, name=None, col='Adj_Close'): self.set_max_win(window) if name is None: name = "ema_%s" % (window) self.df[name] = pd.ewma(self.df[col], span=window, min_periods=window) return name
def calculate_macd(self, signal=9, fast=12, slow=26, col='Adj_Close'): """ MACD """ self.set_max_win(slow) signal_name = "signal_%s" % (signal) # ignore warnings for now # TODO fast_ema = pd.ewma(self.df[col], span=fast, min_periods=fast) slow_ema = pd.ewma(self.df[col], span=slow, min_periods=slow) name = "macd_%s_%s" % (fast, slow) self.df[name] = fast_ema - slow_ema self.calculate_ema(signal, col=name, name=signal_name) return name
def calculate_adx(self, window=14, window_adx=14): """ Average Directional Index Discover if trend is developing Sum((+DI-(-DI))/(+DI+(-DI))/n """ self.set_max_win(window) i = 0 UpI = [] DoI = [] while i + 1 <= self.df.index[-1]: UpMove = self.df.get_value(i + 1, 'High') - self.df.get_value(i, 'High') DoMove = self.df.get_value(i, 'Low') - self.df.get_value(i + 1, 'Low') if UpMove > DoMove and UpMove > 0: UpD = UpMove else: UpD = 0 UpI.append(UpD) if DoMove > UpMove and DoMove > 0: DoD = DoMove else: DoD = 0 DoI.append(DoD) i = i + 1 atr_name = "atr_%s" % (window) self.calculate_atr(window) PosDI = pd.ewma(pd.Series(UpI), span=window, min_periods=window-1) / self.df[atr_name] NegDI = pd.ewma(pd.Series(DoI), span=window, min_periods=window-1) / self.df[atr_name] name = "adx_%s_%s" % (window, window_adx) self.df[name] = pd.Series(pd.ewma(abs(PosDI - NegDI) / (PosDI + NegDI), span=window_adx, min_periods=window_adx-1)) return name
def ema(df, periods=12): # compute exponential moving average #return pd.ewma(df, span = periods) return df.ewm(span=periods, adjust=True, min_periods=0, ignore_na=False).mean()
