我们从Python开源项目中，提取了以下10个代码示例，用于说明如何使用talib.ADX。
def add_ADX(self, timeperiod=14, type='line', color='secondary', **kwargs): """Average Directional Movement Index.""" if not (self.has_high and self.has_low and self.has_close): raise Exception() utils.kwargs_check(kwargs, VALID_TA_KWARGS) if 'kind' in kwargs: type = kwargs['kind'] name = 'ADX({})'.format(str(timeperiod)) self.sec[name] = dict(type=type, color=color) self.ind[name] = talib.ADX(self.df[self.hi].values, self.df[self.lo].values, self.df[self.cl].values, timeperiod)
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 _test_adx(self): code = '300033' df_five_hisdat = mysql.getFiveHisdat(code, start_day='2017-6-14') #df_five_hisdat = pd.read_csv('../test.csv') df_five_hisdat = df_five_hisdat.sort_index() #agl.print_df(df_five_hisdat) highs, lows, closes = df_five_hisdat['h'], df_five_hisdat['l'], df_five_hisdat['c'] adx = ADX(highs, lows, closes) #ui.DrawTs(pl, ts=closes,high=adx) print(adx[-1]) adx = TDX_ADX(highs, lows, closes) print(adx[-1], len(adx)) print(adx[-5:]) #???? ???? #ui.DrawTs(pl, ts=closes[-100:],high=adx[-100:])
def calculate_indicator(stock_df): periods = [3, 5, 10, 20, 30, 60] # MA for period in periods: stock_df['MA' + str(period)] = talib.MA(stock_df['close'].values, timeperiod=period) # EMA periods = [3, 5, 10, 20, 30, 60] for period in periods: stock_df['EMA' + str(period)] = talib.EMA(stock_df['close'].values, timeperiod=period) # AMTMA periods = [5, 10, 20] for period in periods: stock_df['AMTMA' + str(period)] = talib.MA(stock_df['amount'].values, timeperiod=period) # ATR periods = [5, 10, 20] for period in periods: stock_df['ATR' + str(period)] = talib.ATR(stock_df['high'].values, stock_df['low'].values, stock_df['close'].values, timeperiod=period) # ADX period = 14 stock_df['ADX' + str(period)] = talib.ADX(stock_df['high'].values, stock_df['low'].values, stock_df['close'].values, timeperiod=period) # MACD stock_df['MACD_DIFF'], stock_df['MACD_DEA'], stock_df['MACD_HIST'] = talib.MACD( stock_df['close'].values, fastperiod=12, slowperiod=26, signalperiod=9) # CCI period = 14 stock_df['CCI' + str(period)] = talib.CCI(stock_df['high'].values, stock_df['low'].values, stock_df['close'].values, timeperiod=period) # MFI period = 14 stock_df['MFI' + str(period)] = talib.MFI(stock_df['high'].values, stock_df['low'].values, stock_df['close'].values, stock_df['volume'].values, timeperiod=period) # ROCP periods = [5, 10, 20] for period in periods: stock_df['ROCP' + str(period)] = talib.ROCP(stock_df['close'].values, timeperiod=period)
def calculate_adx(self, period_name, close): adx = talib.ADX(self.highs, self.lows, close, timeperiod=14) self.current_indicators[period_name]['adx'] = adx[-1]
def adx(self, n, array=False): """ADX??""" result = talib.ADX(self.high, self.low, self.close, n) if array: return result return result[-1] # ----------------------------------------------------------------------
def TDX_ADX(highs, lows, closes): """??????ADX, ?????test_adx return: np.ndarray MTR:=EXPMEMA(MAX(MAX(HIGH-LOW,ABS(HIGH-REF(CLOSE,1))),ABS(REF(CLOSE,1)-LOW)),N); HD :=HIGH-REF(HIGH,1); LD :=REF(LOW,1)-LOW; DMP:=EXPMEMA(IF(HD>0&&HD>LD,HD,0),N); DMM:=EXPMEMA(IF(LD>0&&LD>HD,LD,0),N); PDI: DMP*100/MTR; MDI: DMM*100/MTR; ADX: EXPMEMA(ABS(MDI-PDI)/(MDI+PDI)*100,MM); ADXR:EXPMEMA(ADX,MM); """ assert(len(closes)>30) highs = np.array(highs) lows = np.array(lows) closes = np.array(closes) mtr = np.zeros(len(closes)) for i, v in np.ndenumerate(closes): i = i[0] if i>0: y = closes[i-1] mtr[i] = max(max(highs[i]-lows[i], abs(highs[i]-y)), abs(y-lows[i])) n = 14 mm = 6 mtr = talib.EMA(mtr, n) hd = np.zeros(len(highs)) ld = np.zeros(len(highs)) for i, v in np.ndenumerate(highs): i = i[0] if i>0: hd[i] = highs[i] - highs[i-1] ld[i] = lows[i-1] - lows[i] if not (hd[i] > 0 and hd[i]>ld[i]): hd[i] = 0 if not (ld[i]>0 and ld[i]>hd[i]): ld[i] = 0 dmp = talib.EMA(hd, n) dmm = talib.EMA(ld, n) pdi = dmp * 100 / mtr mdi = dmm * 100 / mtr adx = np.zeros(len(mdi)) for i, v in np.ndenumerate(mdi): i = i[0] adx[i] = abs(mdi[i]-pdi[i]) / (mdi[i]+pdi[i])*100 adx = talib.EMA(adx, mm) return adx
def ADX(highs, lows, closes): """??????, ADX????????????????????? DMI??ADX?DX""" highs = np.array(highs) lows = np.array(lows) closes = np.array(closes) return talib.ADX(highs, lows, closes, timeperiod=14)
def oscillator2(data): float_close = Data.toFloatArray(df['Close']) float_high = Data.toFloatArray(df['High']) float_low = Data.toFloatArray(df['Low']) float_open = Data.toFloatArray(df['Open']) adx_values = tl.ADX(np.array(float_high),np.array(float_low),np.array(float_close), timeperiod = 14) dmi = tl.DX(np.array(float_high),np.array(float_low),np.array(float_close), timeperiod = 14) mdmi = tl.MINUS_DI(np.array(float_high),np.array(float_low),np.array(float_close), timeperiod = 14) rsi = tl.RSI(np.array(float_close),timeperiod = 4 ) signals = [] flag = 0 for i in xrange(40 , len(adx_values) - 2): if flag ==0: if adx_values[i]>20 and dmi[i]>mdmi[i] and df.loc[i+1, 'Open']> (df.loc[i, 'Close']+1.8) and rsi[i]<50: signal = ['HSI', df.loc[i+1, 'Date'], 'Long', df.loc[i+1, 'Close']] flag =1 signals.append(signal) if adx_values[i]>20 and dmi[i]<mdmi[i] and df.loc[i+1, 'Open']< (df.loc[i, 'Close']-1.8) and rsi[i]<50: signal = ['HSI', df.loc[i+1, 'Date'], 'Short', df.loc[i+1, 'Close']] flag =2 signals.append(signal) elif flag ==1: if df.loc[i, 'Close']>= signal[3]*1.01 or df.loc[i, 'Close']<= signal[3]*0.90 or (df.loc[i, 'Date']-signal[1])>timedelta(days=5): signal = ['HSI', df.loc[i, 'Date'], 'Short', df.loc[i+1, 'Open']] flag = 0 signals.append(signal) elif flag ==2: if df.loc[i, 'Close']<= signal[3]*0.99 or df.loc[i, 'Close']>= signal[3]*1.10 or (df.loc[i, 'Date']-signal[1])>timedelta(days=5): signal = ['HSI', df.loc[i+1, 'Date'], 'Long', df.loc[i+1, 'Close']] flag = 0 signals.append(signal) sig = pd.DataFrame(signals, columns=['Code', 'Time', 'Action', 'Price']) print sig['Time'][10]-sig['Time'][0] profits = [] print sig for k in range(0,len(signals)/2): if sig['Action'][k*2] == "Long": profit = sig['Price'][k*2+1] - sig['Price'][k*2] else: profit = sig['Price'][k*2]- sig['Price'][k*2+1] profits.append(profit) print np.sum(profits) print(profits) ###### PLOT ####### longSignals = sig[sig['Action'] == 'Long'] shortSignals = sig[sig['Action'] == 'Short'] plt.plot(df['Date'], df['Close'], longSignals['Time'], longSignals['Price'], 'r^', shortSignals['Time'], shortSignals['Price'], 'gv', markersize=10) red_patch = mpatches.Patch(color='red', label='Long') green_patch = mpatches.Patch(color='green', label='Short') plt.legend(handles=[red_patch, green_patch]) plt.grid() plt.show() ###### PLOT #######
