我们从Python开源项目中，提取了以下28个代码示例，用于说明如何使用talib.SMA。
def main(): # read csv file and transform it to datafeed (df): df = jhta.CSV2DF('data.csv') # transform datafeed to heikin-ashi datafeed (df): # df = jhta.DF2HEIKIN_ASHI(df) # set numpy datafeed from df: df_numpy = { 'datetime': np.array(df['datetime']), 'Open': np.array(df['Open'], dtype='float'), 'High': np.array(df['High'], dtype='float'), 'Low': np.array(df['Low'], dtype='float'), 'Close': np.array(df['Close'], dtype='float'), 'Volume': np.array(df['Volume'], dtype='int') } # set pandas datafeed from numpy datafeed: # df_pandas = pd.Series(df_numpy) # set talib indicator: indicator = ta.SMA(df_numpy['Close'], 10) # set jhtalib indicator: indicator2 = jhta.SMA(df, 10) # loop through datafeed (df): i = 0 while i < len(df['Close']): # print row: print (df['datetime'][i]+' Open: '+str(df['Open'][i])+' High: '+str(df['High'][i])+' Low: '+str(df['Low'][i])+' Close: '+str(df['Close'][i])+' Volume: '+str(df['Volume'][i])) # print indicators and check for differences between talib and jhtalib: print (str(indicator[i])+' (talib)') print (str(indicator2[i])+' (jhTAlib)') i += 1
def boll(): #??tushare?????? df=ts.get_k_data('300580',start='2017-01-12',end='2017-05-26') #????? closed=df['close'].values upper,middle,lower=talib.BBANDS(closed,matype=talib.MA_Type.SMA) print upper,middle,lower plt.plot(upper) plt.plot(middle) plt.plot(lower) plt.grid() plt.show() diff1=upper-middle diff2=middle-lower print diff1 print diff2
def isBuy(context, analysis): # Bullish SMA Crossover if (getLast(analysis, 'sma_test') == 1): # Bullish MACD if (getLast(analysis, 'macd_test') == 1): return True # # Bullish Stochastics # if(getLast(analysis, 'stoch_over_sold') == 1): # return True # # Bullish RSI # if(getLast(analysis, 'rsi_over_sold') == 1): # return True return False
def isSell(context, analysis): # Bearish SMA Crossover if (getLast(analysis, 'sma_test') == 0): # Bearish MACD if (getLast(analysis, 'macd_test') == 0): return True # # Bearish Stochastics # if(getLast(analysis, 'stoch_over_bought') == 0): # return True # # Bearish RSI # if(getLast(analysis, 'rsi_over_bought') == 0): # return True return False
def OnBarUpdate(self, data=Data, bar=Bar): if len(self.C) < self.p_ma2: return # print('{0}-{1}'.format(self.D[-1], self.C[-1])) ma1 = talib.SMA(self.C, self.p_ma1) ma2 = talib.SMA(self.C, self.p_ma2) self.IndexDict['ma5'] = ma1 self.IndexDict['ma10'] = ma2 if self.PositionLong == 0: if ma1[-1] >= ma2[-1] and ma1[-2] < ma2[-2]: if self.PositionShort > 0: self.BuyToCover(self.O[-1], self.p_lots, '??') self.Buy(self.O[-1], self.p_lots, '??') elif self.PositionShort == 0: if ma1[-1] <= ma2[-1] and ma1[-2] > ma2[-2]: if self.PositionLong > 0: self.Sell(self.O[-1], self.p_lots, '??') self.SellShort(self.O[-1], self.p_lots, '??')
def ACO(self, df): """ Helper indicator :param df: :return: """ df_mid_points = (df['High'] + df['Low']) / 2 mid_points = Data.toFloatArray(df_mid_points) longav = tl.SMA(np.array(mid_points), timeperiod=40) shortav = tl.SMA(np.array(mid_points), timeperiod=15) A0 = longav - shortav Mavg = tl.SMA(A0, timeperiod=15) AcResult = tl.SMA(Mavg - A0, timeperiod=15) signals = np.diff(AcResult) return signals # if __name__ == "__main__": # np.set_printoptions(threshold=np.nan) # pd.set_option("display.max_rows", 280) # dt = Data() # df = dt.getCSVData() # #ACOscillator(df) # ACOscillator(df)
def handle_bar(context, bar_dict): prices = history_bars(context.s1, context.LONGPERIOD + 1, frequency, 'close') short_avg = talib.SMA(prices, context.SHORTPERIOD) long_avg = talib.SMA(prices, context.LONGPERIOD) # ????portfolio?????? cur_position = context.portfolio.positions[context.s1].quantity avg_price = context.portfolio.positions[context.s1].avg_price capital = cur_position * avg_price # ????portfolio???????????? shares = context.portfolio.cash / bar_dict[context.s1].close # ?????????? plot('capital', capital) # ???????????????????bar????????????? if short_avg[-1] - long_avg[-1] < 0 < long_avg[-2] - short_avg[-2] and cur_position > 0: # ???? order_target_value(context.s1, 0) # ???????????????????? if short_avg[-1] - long_avg[-1] > 0 > long_avg[-2] - short_avg[-2]: # ???? order_shares(context.s1, shares)
def recover_period(data): close = data.history(sid(8554),'close', 120, '1d') sma5 = ta.SMA(close, 5)[-1] # print sma5 sma20 = ta.SMA(close, 20)[-1] sma60 = ta.SMA(close, 60)[-1] # print sma5 # print sma20 # print sma60 short_ratio = sma5/sma20 if sma5/sma20 else 1 mid_ratio = sma20/sma60 if sma20/sma60 else 1 # long_ratio = sma5/sma60 if sma5/sma60 else 1 if (short_ratio > 1) and (mid_ratio > 1): return True else: return False
def handle_bar(context, bar_dict): # ????????????? # bar_dict[order_book_id] ?????????bar?? # context.portfolio ??????????????? # ??order_shares(id_or_ins, amount)?????? # TODO: ?????????? # ????????????????????? prices = history_bars(context.s1, context.LONGPERIOD+1, '1d', 'close') # ??talib????????????array????? short_avg = talib.SMA(prices, context.SHORTPERIOD) long_avg = talib.SMA(prices, context.LONGPERIOD) plot("short avg", short_avg[-1]) plot("long avg", long_avg[-1]) # ????portfolio?????? cur_position = context.portfolio.positions[context.s1].quantity # ????portfolio???????????? shares = context.portfolio.cash/bar_dict[context.s1].close # ??????????????????????bar?????????????????bar????????????? if short_avg[-1] - long_avg[-1] < 0 and short_avg[-2] - long_avg[-2] > 0 and cur_position > 0: # ???? order_target_value(context.s1, 0) # ???????????????????? if short_avg[-1] - long_avg[-1] > 0 and short_avg[-2] - long_avg[-2] < 0: # ???? order_shares(context.s1, shares)
def handle_bar(context, bar_dict): # ????????????? # bar_dict[order_book_id] ?????????bar?? # context.portfolio ??????????????? # ??order_shares(id_or_ins, amount)?????? # TODO: ?????????? # ????????????????????? prices = history_bars(context.s1, context.LONGPERIOD+1, '1d', 'close') # ??talib????????????array????? short_avg = talib.SMA(prices, context.SHORTPERIOD) long_avg = talib.SMA(prices, context.LONGPERIOD) plot("short avg", short_avg[-1]) plot("long avg", long_avg[-1]) # ????portfolio?????? cur_position = context.portfolio.positions[context.s1].quantity # ????portfolio???????????? shares = context.portfolio.cash / bar_dict[context.s1].close # ??????????????????????bar?????????????????bar????????????? if short_avg[-1] - long_avg[-1] < 0 and short_avg[-2] - long_avg[-2] > 0 and cur_position > 0: # ???? order_target_value(context.s1, 0) # ???????????????????? if short_avg[-1] - long_avg[-1] > 0 and short_avg[-2] - long_avg[-2] < 0: # ???? order_shares(context.s1, shares)
def baseAPI(): #??tushare?????? df=ts.get_k_data('300580',start='2017-01-12',end='2017-05-26') #????? closed=df['close'].values #??????????timeperiod??????? 5,10,20 ??????? ma5=talib.SMA(closed,timeperiod=5) ma10=talib.SMA(closed,timeperiod=10) ma20=talib.SMA(closed,timeperiod=20) #????????? print closed print ma5 print ma10 print ma20 #??plog???????????????? plt.plot(closed) plt.plot(ma5) plt.plot(ma10) plt.plot(ma20) #?????????????????? plt.grid() #??????????????? plt.show()
def ma_type_test(): #MA_Type: 0=SMA, 1=EMA, 2=WMA, 3=DEMA, 4=TEMA, 5=TRIMA, 6=KAMA, 7=MAMA, 8=T3 (Default=SMA) df=ts.get_k_data('300580',start='2017-01-12',end='2017-05-26') closed=df['close'].values sma=talib.MA(closed,timeperiod=10,matype=0) ema=talib.MA(closed,timeperiod=10,matype=1) wma=talib.MA(closed,timeperiod=10,matype=2) dema=talib.MA(closed,timeperiod=10,matype=3) tema=talib.MA(closed,timeperiod=10,matype=4) trima=talib.MA(closed,timeperiod=10,matype=5) kma=talib.MA(closed,timeperiod=10,matype=6) mama=talib.MA(closed,timeperiod=10,matype=7) t3=talib.MA(closed,timeperiod=10,matype=8) #ouput=talib.MA(closed,timeperiod=5,matype=0) print closed plt.ylim([0,40]) plt.plot(sma,'r--') plt.plot(ema,'g-*') plt.plot(wma) plt.plot(dema) plt.plot(tema) plt.plot(trima) plt.plot(kma) plt.plot(mama) plt.plot(t3) plt.grid() plt.text(7,30,'BST') plt.show()
def handle_bar(context, bar_dict): # ????????????? # bar_dict[order_book_id] ?????????bar?? # context.portfolio ??????????????? # ??order_shares(id_or_ins, amount)?????? # TODO: ?????????? stocks = set(context.to_buy) holdings = set(get_holdings(context)) to_buy = stocks - holdings holdings = set(get_holdings(context)) to_sell = holdings - stocks for stock in to_sell: if bar_dict[stock].is_trading: order_target_percent(stock , 0) to_buy = get_trading_stocks(to_buy, context, bar_dict) buy_rsi = [] for stock in to_buy: prices = history(context.SMAPERIOD+1,'1d','close')[stock].values avg = talib.SMA(prices,context.SMAPERIOD) if avg[-1] < prices[-1]: buy_rsi.append(stock) if len(buy_rsi) >0: #cash = context.portfolio.cash #average_value = 0.98 * cash / len(buy_rsi) for stock in buy_rsi: if bar_dict[stock].is_trading: #order_value(stock ,average_value) order_target_percent(stock,0.08)
def add_SMA(self, timeperiod=20, type='line', color='secondary', **kwargs): """Simple Moving Average.""" if not self.has_close: raise Exception() utils.kwargs_check(kwargs, VALID_TA_KWARGS) if 'kind' in kwargs: type = kwargs['kind'] name = 'SMA({})'.format(str(timeperiod)) self.pri[name] = dict(type=type, color=color) self.ind[name] = talib.SMA(self.df[self.cl].values, timeperiod)
def _moving_average(self, data, n): """ ??????????? """ data = transform2ndarray(data) return talib.SMA(data, n)
def calculate_avg_volume(self, period_name, volumes): avg_vol = talib.SMA(volumes, timeperiod=15) self.current_indicators[period_name]['avg_volume'] = avg_vol[-1]
def handle_bar(context, bar_dict): prices = history_bars(context.s1, context.LONGPERIOD + 1, '1d', 'close') # print(prices) short_avg = talib.SMA(prices, context.SHORTPERIOD) long_avg = talib.SMA(prices, context.LONGPERIOD) cur_position = context.portfolio.positions[context.s1].quantity shares = context.portfolio.cash / bar_dict[context.s1].close if short_avg[-1] < long_avg[-1] and short_avg[-2] > long_avg[-2] and cur_position > 0: # ?? order_target_value(context.s1, 0) if short_avg[-1] > long_avg[-1] and short_avg[-2] < long_avg[-2]: order_shares(context.s1, shares)
def handle_bar(context, bar_dict): prices = history_bars(context.s1, context.LONGPERIOD + 1, '1h', 'close') short_avg = talib.SMA(prices, context.SHORTPERIOD) long_avg = talib.SMA(prices, context.LONGPERIOD) cur_position = context.portfolio.positions[context.s1].quantity shares = context.portfolio.cash / bar_dict[context.s1].close if short_avg[-1] < long_avg[-1] and short_avg[-2] > long_avg[-2] and cur_position > 0: order_target_value(context.s1, 0) if short_avg[-1] > long_avg[-1] and short_avg[-2] < long_avg[-2]: order_shares(context.s1, shares)
def sma(self, n, array=False): """????""" result = talib.SMA(self.close, n) if array: return result return result[-1] # ----------------------------------------------------------------------
def sma(self, np_array, n, array=False): """????""" if n < 2: result = np_array else: result = talib.SMA(np_array, n) if array: return result return result[-1]
def ACOscillator(df): df_mid_points = (df['High'] + df['Low'])/2 mid_points = Data.toFloatArray(df_mid_points) longav = tl.SMA(np.array(mid_points), timeperiod=40) shortav = tl.SMA(np.array(mid_points),timeperiod =15) A0 = longav - shortav Mavg = tl.SMA(A0, timeperiod = 15) AcResult = tl.SMA(Mavg - A0, timeperiod = 15) signals = np.diff(AcResult) return signals
def history(self, databar, period, indicator): """Recieve mock data as databar dataframe format :databar: mock data bar, flexible, means will be changed by indicator, one dimension or several dimensions :period: time interval, flexbile too :indicator: include ['sma','macd','atr' ...] """ if indicator == 'sma': try: sma0 = talib.SMA(np.array(databar.low), timeperiod = period) sma1 = talib.SMA(np.array(databar.close), timeperiod = period) sma2 = talib.SMA(np.array(databar.high), timeperiod = period) return pd.DataFrame({'sma0':sma0, 'sma1':sma1, 'sma2':sma2}, index=pd.DatetimeIndex(databar.date)) except KeyError: print('Pls check databar whether is dataframe') return None elif indicator == 'atr': try: atr = talib.ATR(np.array(databar.high), np.array(databar.low), np.array(databar.close), timeperiod = period) return pd.DataFrame({'atr':atr}, index=pd.DatetimeIndex(databar.date)) except KeyError: print('Pls check databar whether is dataframe') return None elif indicator == 'macd': try: macd, macdsignal, macdhist = talib.MACD(databar, fastperiod = period['fastperiod'], slowperiod = period['slowperiod'], signalperiod = period['signalperiod']) return pd.DataFrame({'macd':macd, 'macdsignal':macdsignal, 'macdhist':macdhist}, index=pd.DatetimeIndex(databar.date)) except KeyError: print('Pls check databar whether is dataframe') return None
def __init__(self, marketevent): super(Indicator, self).__init__(marketevent) self.SMA = self.SimpleMovingAverage # shortcut
def SimpleMovingAverage(self, period, index=-1): data = self.get_preload(period, index, 'close') # ??period???????? sma = tb.SMA(data, period) return return_NaN(sma, index)
def adosc_bloomberg(my_close, my_high, my_low, my_volume, fastperiod, slowperiod): MFV = (((my_close - my_low)-(my_high - my_close)) / (my_high - my_low)) * my_volume ADL = np.cumsum(MFV) ADOSC = SMA(ADL.values, timeperiod=fastperiod) - SMA(ADL.values, timeperiod=slowperiod) return ADOSC
def overlap_process(event): print(event.widget.get()) overlap = event.widget.get() upperband, middleband, lowerband = ta.BBANDS(close, timeperiod=5, nbdevup=2, nbdevdn=2, matype=0) fig, axes = plt.subplots(2, 1, sharex=True) ax1, ax2 = axes[0], axes[1] axes[0].plot(close, 'rd-', markersize=3) axes[0].plot(upperband, 'y-') axes[0].plot(middleband, 'b-') axes[0].plot(lowerband, 'y-') axes[0].set_title(overlap, fontproperties="SimHei") if overlap == u'???': pass elif overlap == u'????????': real = ta.DEMA(close, timeperiod=30) axes[1].plot(real, 'r-') elif overlap == u'??????? ': real = ta.EMA(close, timeperiod=30) axes[1].plot(real, 'r-') elif overlap == u'??????——?????': real = ta.HT_TRENDLINE(close) axes[1].plot(real, 'r-') elif overlap == u'???????????': real = ta.KAMA(close, timeperiod=30) axes[1].plot(real, 'r-') elif overlap == u'?????': real = ta.MA(close, timeperiod=30, matype=0) axes[1].plot(real, 'r-') elif overlap == u'MESA???????': mama, fama = ta.MAMA(close, fastlimit=0, slowlimit=0) axes[1].plot(mama, 'r-') axes[1].plot(fama, 'g-') elif overlap == u'????????': real = ta.MAVP(close, periods, minperiod=2, maxperiod=30, matype=0) axes[1].plot(real, 'r-') elif overlap == u'???????': real = ta.SMA(close, timeperiod=30) axes[1].plot(real, 'r-') elif overlap == u'????????(T3)': real = ta.T3(close, timeperiod=5, vfactor=0) axes[1].plot(real, 'r-') elif overlap == u'????????': real = ta.TEMA(close, timeperiod=30) axes[1].plot(real, 'r-') elif overlap == u'?????? ': real = ta.TRIMA(close, timeperiod=30) axes[1].plot(real, 'r-') elif overlap == u'???????': real = ta.WMA(close, timeperiod=30) axes[1].plot(real, 'r-') plt.show() # ????
