我们从Python开源项目中，提取了以下16个代码示例，用于说明如何使用talib.CCI。
def add_CCI(self, timeperiod=14, type='line', color='secondary', **kwargs): """Channel Commodity 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 = 'CCI({})'.format(str(timeperiod)) self.sec[name] = dict(type=type, color=color) self.ind[name] = talib.CCI(self.df[self.hi].values, self.df[self.lo].values, self.df[self.cl].values, timeperiod)
def CCI(security_list, timeperiod=14): # ???????????? if isinstance(security_list, str): security_list = [security_list] # ?? CCI cci = {} for stock in security_list: security_data = attribute_history( stock, timeperiod * 2, '1d', ['close', 'high', 'low'], df=False) close_CCI = security_data['close'] high_CCI = security_data['high'] low_CCI = security_data['low'] cci[stock] = talib.CCI(high_CCI, low_CCI, close_CCI, timeperiod) return cci # ATR
def pre_data(stick_code,ktype='D'): # ktype in ('D','W','M') global df db = m_db2() try: if ktype == 'D': df = db.get_data("select * from t_stick_data_d where code = '"+stick_code+"' and date > '2015-09-01';")#and date>'2015-05-01' elif ktype == 'W': df = db.get_data("select * from t_stick_data_w where code = '"+stick_code+"' ;")#and date>'2015-05-01' elif ktype == 'M': df = db.get_data("select * from t_stick_data_m where code = '" + stick_code + "' ;") # and date>'2015-05-01' except Exception as e: print('ERR:',e) return df['cci'] = ta.CCI(df['high'].values.astype('double'),df['low'].values.astype('double'),df['close'].values.astype('double')) df['diff'],df['dea'],df['macd'] = ta.MACD(df['close'].values.astype('double'),fastperiod=12, slowperiod=26, signalperiod=9) df['obv'] = ta.OBV(df['close'].values.astype('double'),df['vol'].values.astype('double')) df['volma5']=ta.MA(df['vol'].values.astype('double'),5); df['volma20'] = ta.MA(df['vol'].values.astype('double'), 20); df['MA20'] = ta.MA(df['close'].values.astype('double'), 20) df['MA60'] = ta.MA(df['close'].values.astype('double'), 60) df['cwbili']=0 df['pricebili']=0 return df # draw
def pre_data(stick_code,ktype='D',today=''): # ktype in ('D','W','M') #today='2010-01-01' if '' == today: today = datetime.date.today().strftime('%Y-%m-%d') # begindate = datetime.date.today() - datetime.timedelta(days=13) global df db = m_db2() try: if ktype == 'D': df = db.get_data("select * from t_stick_data_d where code = '"+stick_code+"' and date > '2015-09-01' and date <='"+today+"' order by date asc;")#and date>'2015-05-01' elif ktype == 'W': df = db.get_data("select * from t_stick_data_w where code = '"+stick_code+"' ;")#and date>'2015-05-01' elif ktype == 'M': df = db.get_data("select * from t_stick_data_m where code = '" + stick_code + "' ;") # and date>'2015-05-01' except Exception as e: #print('ERR:',e) return df['cci'] = ta.CCI(df['high'].values.astype('double'),df['low'].values.astype('double'),df['close'].values.astype('double')) df['diff'],df['dea'],df['macd'] = ta.MACD(df['close'].values.astype('double'),fastperiod=12, slowperiod=26, signalperiod=9) df['obv'] = ta.OBV(df['close'].values.astype('double'),df['vol'].values.astype('double')) df['volma5']=ta.MA(df['vol'].values.astype('double'),5); df['volma13'] = ta.MA(df['vol'].values.astype('double'), 13); df['volma20'] = ta.MA(df['vol'].values.astype('double'), 20); df['volma34'] = ta.MA(df['vol'].values.astype('double'), 34); df['MA20'] = ta.MA(df['close'].values.astype('double'), 20) df['MA60'] = ta.MA(df['close'].values.astype('double'), 60) df['MA5'] = ta.MA(df['close'].values.astype('double'), 5) df['MA13'] = ta.MA(df['close'].values.astype('double'), 13) df['MA34'] = ta.MA(df['close'].values.astype('double'), 34) df['MA89'] = ta.MA(df['close'].values.astype('double'), 89) df['MA144'] = ta.MA(df['close'].values.astype('double'), 144) df['cwbili']=0 df['pricebili']=0 return df # draw
def pre_data(stick_code,ktype='D'): # ktype in ('D','W','M') global df db = m_db2() try: if ktype == 'D': df = db.get_data("select * from t_stick_data_d where code = '"+stick_code+"' and date > '2015-09-01';")#and date>'2015-05-01' elif ktype == 'W': df = db.get_data("select * from t_stick_data_w where code = '"+stick_code+"' ;")#and date>'2015-05-01' elif ktype == 'M': df = db.get_data("select * from t_stick_data_m where code = '" + stick_code + "' ;") # and date>'2015-05-01' except Exception as e: print('ERR:',e) return df['cci'] = ta.CCI(df['high'].values.astype('double'),df['low'].values.astype('double'),df['close'].values.astype('double')) df['diff'],df['dea'],df['macd'] = ta.MACD(df['close'].values.astype('double'),fastperiod=12, slowperiod=26, signalperiod=9) df['obv'] = ta.OBV(df['close'].values.astype('double'),df['vol'].values.astype('double')) df['volma5']=ta.MA(df['vol'].values.astype('double'),5); df['volma20'] = ta.MA(df['vol'].values.astype('double'), 20); df['MA20'] = ta.MA(df['close'].values.astype('double'), 5) #print(df) #i= ta.CDLCONCEALBABYSWALL(df['open'].values.astype('double'),df['high'].values.astype('double'), # df['low'].values.astype('double'),df['close'].values.astype('double'),) #print(i) return df # draw
def CCI(df, n): PP = (df['high'] + df['low'] + df['close']) / 3 CCI = pd.Series((PP - pd.rolling_mean(PP, n)) / pd.rolling_std(PP, n) / 0.015, name = 'CCI' + str(n)) return CCI
def cci(df, n): real = talib.CCI(df['high'][(-n-1):], df['low'][(-n-1):], df['close'][(-n-1):], timeperiod=n) df['CCI' + str(n)][-1] = real[-1] #Coppock Curve
def CCI_CN(high, low, close, timeperiod=14) : len1 = len(high) len2 = len(low) len3 = len(close) if len1 != len2 or len1 != len3: print ("CCI_CN input invalid for len:%s %s %s " %(str(len1),str(len2),str(len3))) return np.array(np.nan) cci = tl.CCI(high, low, close, timeperiod=timeperiod) return cci #MACD
def CCI_DAY(self, context, security, data={}, ref=0): CCI = self.CCI_DATA_DAY(context, security, data, ref+1) if np.isnan(CCI[-1]): return 0 return CCI[-1-ref]
def CCI_DATA(self, context, security, freq = 'D', data={}, dataCount=1): #sma target round2 precision = 14 high, low, close = self.GET_PERIOD_DATA(context, security, freq, data, dataCount+precision) if np.isnan(close[-1]): return np.array([np.nan]) CCI = self.CCI_CN(high, low, close) if len(CCI) > precision: CCI = CCI[-dataCount:] else: #print "security:%s no len data precison %s" %(str(security), len(CCI)) pass decimal.getcontext().rounding=decimal.ROUND_HALF_UP CCI = np.array([float(decimal.Decimal(s).quantize(decimal.Decimal('0.00'))) for s in CCI]) return CCI
def cci(self, n, array=False): """CCI??""" result = talib.CCI(self.high, self.low, self.close, n) if array: return result return result[-1] # ----------------------------------------------------------------------
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 pre_data(self,stick_code, ktype='D', beginday='',endday=''): # ktype in ('D','W','M') # today='2010-01-01' if '' == beginday: begindaytmp = datetime.date.today() - datetime.timedelta(days=13) beginday = begindaytmp.strftime('%Y-%m-%d') if '' == endday: endday = datetime.date.today().strftime('%Y-%m-%d') df ='' print(beginday,endday) try: if ktype == 'D': df = self.get_data( "select * from t_stick_data_d \ where code = '" + stick_code + "' and date > '"+beginday+"' \ and date <='" + endday + "' order by date asc;") # and date>'2015-05-01' elif ktype == 'W': df = self.get_data( "select * from t_stick_data_w \ where code = '" + stick_code + "' and date > '"+beginday+"' \ and date <='" + endday + "' order by date asc;") # and date>'2015-05-01' elif ktype == 'M': df = self.get_data( "select * from t_stick_data_m \ where code = '" + stick_code + "' and date > '"+beginday+"' \ and date <='" + endday + "' order by date asc;") # and date>'2015-05-01' except Exception as e: # print('ERR:',e) return df['cci'] = ta.CCI(df['high'].values.astype('double'), df['low'].values.astype('double'), df['close'].values.astype('double')) df['diff'], df['dea'], df['macd'] = ta.MACD(df['close'].values.astype('double'), fastperiod=12, slowperiod=26, signalperiod=9) df['obv'] = ta.OBV(df['close'].values.astype('double'), df['vol'].values.astype('double')) df['volma5'] = ta.MA(df['vol'].values.astype('double'), 5); df['volma20'] = ta.MA(df['vol'].values.astype('double'), 20); df['MA20'] = ta.MA(df['close'].values.astype('double'), 20) df['MA60'] = ta.MA(df['close'].values.astype('double'), 60) df['cwbili'] = 0 df['pricebili'] = 0 return df # xx=m_db2(); # df=xx.pre_data('000157',ktype='W') # print(df) # xx.insert_data('t_stick_data_m_test',df.head(20).as_matrix()) # xx.commit()
def calculate_features(df): """ Method which calculates and generates features """ close = df['close'].values high = df['high'].values low = df['low'].values volume = df['volume'].values last_row = df.tail(1).copy() # ************** Calc EMAs ema_periods = [2, 4, 8, 12, 16, 20] for ema_period in ema_periods: ema = talib.EMA(close[-ema_period:], timeperiod=ema_period)[-1] last_row['ema' + str(ema_period)] = ema # ************** Calc RSIs rsi_periods = [5] for rsi_period in rsi_periods: rsi = talib.RSI(close[-rsi_period:], timeperiod=rsi_period-1)[-1] last_row['rsi' + str(rsi_period)] = rsi last_row['rsi_above_50' + str(rsi_period)] = int(rsi > 50.0) # ************** Calc CCIs cci_periods = [5] for cci_period in cci_periods: cci = talib.CCI(high[-cci_period:], low[-cci_period:], close[-cci_period:], timeperiod=cci_period)[-1] last_row['cci' + str(cci_period)] = cci # ************** Calc MACD 1 macd_periods = [34] for macd_period in macd_periods: macd, macd_signal, _ = talib.MACD(close[-macd_period:], fastperiod=12, slowperiod=26, signalperiod=9) macd = macd[-1] signal_line = macd_signal[-1] last_row['macd_above_signal' + str(macd_period)] = int(macd > signal_line) last_row['macd_above_zero' + str(macd_period)] = int(macd > 0.0) # ************** Calc OBVs obv_periods = [2, 4, 8, 12, 16, 20] for obv_period in obv_periods: obv = talib.OBV(close[-obv_period:], volume[-obv_period:])[-1] last_row['obv' + str(obv_period)] = obv return last_row
