我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用talib.MACD。
def handle_bar(context, bar_dict): # ????????????? # ??????????history_bars??????ndarray???????????? prices = history_bars(context.s1, context.OBSERVATION, '1d', 'close') # ?Talib??MACD?????????????????macd,signal ? hist macd, signal, hist = talib.MACD(prices, context.SHORTPERIOD, context.LONGPERIOD, context.SMOOTHPERIOD) # macd ?????????signal?macd????????????????????????????????? if (macd[-1] - signal[-1] > 0 and macd[-2] - signal[-2] < 0): sell_qty = context.portfolio.positions[context.s1].sell_quantity # ????????????????????? if (sell_qty > 0): buy_close(context.s1, 1) # ???? buy_open(context.s1, 1) if (macd[-1] - signal[-1] < 0 and macd[-2] - signal[-2] > 0): buy_qty = context.portfolio.positions[context.s1].buy_quantity # ????????????????????? if (buy_qty > 0): sell_close(context.s1, 1) # ???? sell_open(context.s1, 1)
def get_indicator(df, indicator): ret_df = df if 'MACD' in indicator: macd, macdsignal, macdhist = ta.MACD(df.close.values, fastperiod=12, slowperiod=26, signalperiod=9) ret_df = KlineData._merge_dataframe(pd.DataFrame([macd, macdsignal, macdhist]).T.rename(columns={0: "macddif", 1: "macddem", 2: "macdhist"}), ret_df) ret_df = KlineData._merge_dataframe(line_intersections(ret_df, columns=['macddif', 'macddem']), ret_df) if 'MFI' in indicator: real = ta.MFI(df.high.values, df.low.values, df.close.values, df.volume.values, timeperiod=14) ret_df = KlineData._merge_dataframe(pd.DataFrame([real]).T.rename(columns={0: "mfi"}), ret_df) if 'ATR' in indicator: real = ta.NATR(df.high.values, df.low.values, df.close.values, timeperiod=14) ret_df = KlineData._merge_dataframe(pd.DataFrame([real]).T.rename(columns={0: "atr"}), ret_df) if 'ROCR' in indicator: real = ta.ROCR(df.close.values, timeperiod=10) ret_df = KlineData._merge_dataframe(pd.DataFrame([real]).T.rename(columns={0: "rocr"}), ret_df) ret_df['date'] = pd.to_datetime(ret_df['date'], format='%Y-%m-%d') return ret_df
def handle_bar(context, bar_dict): # ????????????? # ??????????history_bars??????ndarray???????????? prices = history_bars(context.s1, context.OBSERVATION, '1d', 'close') # ?Talib??MACD?????????????????macd,signal ? hist macd, signal, hist = talib.MACD(prices, context.SHORTPERIOD, context.LONGPERIOD, context.SMOOTHPERIOD) # macd ?????????signal?macd????????????????????????????????? if macd[-1] - signal[-1] > 0 and macd[-2] - signal[-2] < 0: sell_qty = context.portfolio.positions[context.s1].sell_quantity # ????????????????????? if sell_qty > 0: buy_close(context.s1, 1) # ???? buy_open(context.s1, 1) if macd[-1] - signal[-1] < 0 and macd[-2] - signal[-2] > 0: buy_qty = context.portfolio.positions[context.s1].buy_quantity # ????????????????????? if buy_qty > 0: sell_close(context.s1, 1) # ???? sell_open(context.s1, 1)
def handle_bar(context, bar_dict): # ?????????????x # ??????????history_bars??????ndarray???????????? prices = history_bars(context.s1, context.OBSERVATION, '1d', 'close') # ?Talib??MACD?????????????????macd,signal ? hist macd, signal, hist = talib.MACD(prices, context.SHORTPERIOD, context.LONGPERIOD, context.SMOOTHPERIOD) # macd ?????????signal?macd????????????????????????????????? if macd[-1] - signal[-1] > 0 and macd[-2] - signal[-2] < 0: sell_qty = context.portfolio.positions[context.s1].sell_quantity # ????????????????????? if sell_qty > 0: buy_close(context.s1, 1) # ???? buy_open(context.s1, 1) if macd[-1] - signal[-1] < 0 and macd[-2] - signal[-2] > 0: buy_qty = context.portfolio.positions[context.s1].buy_quantity # ????????????????????? if buy_qty > 0: sell_close(context.s1, 1) # ???? sell_open(context.s1, 1)
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 getStrategy_MACD(start_trading_day,end_trading_day,_time,_close): point = [] iday = _time.index(start_trading_day) eday = _time.index(end_trading_day) macd, signal,hist = ta.MACD(np.array(_close,dtype='f8'),fastperiod=12,slowperiod=26,signalperiod=9) macd = macd[iday:eday] signal = signal[iday:eday] point.append(0) for i in range(1,len(macd)): if (macd[i-1] <= signal[i-1]) and (macd[i] >= signal[i]): point.append(1) elif (macd[i-1] >= signal[i-1]) and (macd[i] <= signal[i]): point.append(-1) else: point.append(0) return point
def macd_rule(df): try: df = MACD(df) except: return False input_1 = 0.2 input_2 = -0.8 input_3 = 22 * 3 index = -1 df['macd_dif_1'] = df['macd_dif'].shift(1) df['macd_dea_1'] = df['macd_dea'].shift(1) return (abs(df['macd_dea'][index]) < input_1) & \ (abs(df['macd_dif'][index]) < input_1) & \ (df['macd_dif'][-input_3:].min() < input_2) & \ (df['macd_dif'][index] > df['macd_dea'][index]) & \ ((df['macd_dea_1'][index] > df['macd_dif_1'][index]) | (abs(df['macd_dea_1'][index] - df['macd_dif_1'][index]) < 0.007))
def macd_rule_1(df): try: df = MACD(df) except: return False input_1 = 0 input_2 = -0.8 input_3 = 0.05 dif_len = len(df['macd_dif']) if dif_len < 2: return False if abs(df['macd_dif'][-1]) > input_3: return False for idx in range(dif_len-1, 1, -1): if ((df['macd_dif'][idx] - df['macd_dif'][idx-1]) > input_1): continue if df['macd_dif'][idx] <= input_2: return True else: return False
def macd_rule(df, index = -1): try: if not {'macd_dif', 'macd_dea', 'macd'}.issubset(df.columns): df = MACD(df) except Exception as e: print(e) return False input_1 = -3 input_2 = 0.05 # df['macd_dif_1'] = df['macd_dif'].shift(1) # df['macd_dea_1'] = df['macd_dea'].shift(1) #(df['macd_dif'][-input_3:].min() < input_2) & \ return (df['macd_dif'][index] > input_1) & \ (df['macd_dif'][index] < input_2) & \ (df['macd_dif'][index] > df['macd_dea'][index]) & \ ((df['macd_dea'][index-1] > df['macd_dif'][index-1]) | (abs(df['macd_dea'][index-1] - df['macd_dif'][index-1]) < 0.007))
def Unittest_Kline(): """""" kline = Guider("600100", "") print(kline.getData(0).date, kline.getLastData().date) #kline.myprint() obv = kline.OBV() pl.figure pl.subplot(2,1,1) pl.plot(kline.getCloses()) pl.subplot(2,1,2) ma,m2,m3 = kline.MACD() pl.plot(ma) pl.plot(m2,'r') left = np.arange(0, len(m3)) pl.bar(left,m3) #pl.plot(obv, 'y') pl.show() #Unittest_Kstp() # #??????????? #----------------------------------------------------------------------
def compute(self, today, assets, out, close): sig_lines = [] for col in close.T: # get signal line only try: _, signal_line, _ = talib.MACD(col, fastperiod=12, slowperiod=26, signalperiod=10) sig_lines.append(signal_line[-1]) # if error calculating, return NaN except: sig_lines.append(np.nan) out[:] = sig_lines # 20-day Stochastic Oscillator
def __init__(self, selector): self.selector = selector self.supported = {"ROCP", "OROCP", "HROCP", "LROCP", "MACD", "RSI", "VROCP", "BOLL", "MA", "VMA", "PRICE_VOLUME"} self.feature = []
def init(context): context.s1 = "000001.XSHE" # ??MACD?????????macd?????? context.SHORTPERIOD = 12 context.LONGPERIOD = 26 context.SMOOTHPERIOD = 9 context.OBSERVATION = 100 # ???????????????????????????????????????????
def handle_bar(context, bar_dict): # ????????????? # bar_dict[order_book_id] ?????????bar?? # context.portfolio ??????????????? # ??order_shares(id_or_ins, amount)?????? # TODO: ?????????? # ?????????sma??????????????????????ema???????????????????????????? prices = history_bars(context.s1, context.OBSERVATION,'1d','close') # ?Talib??MACD?????????????????macd, signal ? hist macd, signal, hist = talib.MACD(prices, context.SHORTPERIOD, context.LONGPERIOD, context.SMOOTHPERIOD) plot("macd", macd[-1]) plot("macd signal", signal[-1]) # macd ?????????signal?macd??????macd?????????????????macd???signal?????? # ??macd??????macd_signal if macd[-1] - signal[-1] < 0 and macd[-2] - signal[-2] > 0: # ????portfolio?????? curPosition = context.portfolio.positions[context.s1].quantity #???? if curPosition > 0: order_target_value(context.s1, 0) # ???????????????????? if macd[-1] - signal[-1] > 0 and macd[-2] - signal[-2] < 0: # ???? order_target_percent(context.s1, 1)
def init(context): # context???????s1????????? context.s1 = 'IF1606' # ??MACD?????????macd?????? context.SHORTPERIOD = 12 context.LONGPERIOD = 26 context.SMOOTHPERIOD = 9 context.OBSERVATION = 50 # ??????????????????????handle_bar????? subscribe(context.s1) # ?????????????????????????????
def init(context): # context???????s1????????? context.s1 = 'IF1606' # ??MACD?????????macd?????? context.SHORTPERIOD = 12 context.LONGPERIOD = 26 context.SMOOTHPERIOD = 9 context.OBSERVATION = 50 #??????????????????????handle_bar????? subscribe(context.s1) # ?????????????????????????????
def init(context): # context???????s1????????? context.s1 = "IF88" # ??MACD?????????macd?????? context.SHORTPERIOD = 12 context.LONGPERIOD = 26 context.SMOOTHPERIOD = 9 context.OBSERVATION = 50 # ??????????????????????handle_bar????? subscribe(context.s1) # ?????????????????????????????
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 can_buy(stock, day_count=3): DIF, DEA, macd = MACD(stock) for i in range(1, day_count + 1): if (DIF[-i] > DEA[-i] and DIF[-i - 1] < DEA[-i - 1]): return True return False # ??????????
def can_buy_prior(stock, day_count=3): DIF, DEA, macd = MACD(stock) count = 0 for i in range(1, len(macd) - 2): if DIF[-i] > 0 or DEA[-i] > 0: return False if (DIF[-i] - DEA[-i] > 0 and DIF[-i - 1] - DEA[-i - 1] < 0): count += 1 if count >= 2: return True if i >= day_count and count == 0: return False # ?????????????????? ??
def can_sell(stock, day_count=3): DIF, DEA, macd = MACD(stock) if DIF[-1] < DEA[-1]: return True result = True for i in range(1, day_count): result = result and DIF[-i] < DIF[-i - 1] return result # ??MACD??????????
def MACD(stock): prices = attribute_history(stock, 130, '1d', ('close'), fq='pre')['close'].values # ????????? cur_prices = attribute_history(stock, 1, '1m', ('close'), fq='pre')['close'].values prices += cur_prices DIF, DEA, macd = talib.MACD(prices, fastperiod=12, slowperiod=26, signalperiod=9) return DIF, DEA, macd
def add_MACD(self, fastperiod=12, slowperiod=26, signalperiod=9, types=['line', 'line', 'histogram'], colors=['primary', 'tertiary', 'fill'], **kwargs): """Moving Average Convergence Divergence. Note that the first argument of types and colors refers to MACD, the second argument refers to MACD signal line and the third argument refers to MACD histogram. """ if not self.has_close: raise Exception() utils.kwargs_check(kwargs, VALID_TA_KWARGS) if 'kind' in kwargs: kwargs['type'] = kwargs['kind'] if 'kinds' in kwargs: types = kwargs['type'] if 'type' in kwargs: types = [kwargs['type']] * 3 if 'color' in kwargs: colors = [kwargs['color']] * 3 name = 'MACD({},{},{})'.format(str(fastperiod), str(slowperiod), str(signalperiod)) macd = name smacd = name + '[Sign]' hmacd = name + '[Hist]' self.sec[macd] = dict(type=types[0], color=colors[0]) self.sec[smacd] = dict(type=types[1], color=colors[1], on=macd) self.sec[hmacd] = dict(type=types[2], color=colors[2], on=macd) (self.ind[macd], self.ind[smacd], self.ind[hmacd]) = talib.MACD(self.df[self.cl].values, fastperiod, slowperiod, signalperiod)
def calculate_macd(self, period_name, closing_prices): macd, macd_sig, macd_hist = talib.MACD(closing_prices, fastperiod=12, slowperiod=26, signalperiod=9) self.current_indicators[period_name]['macd'] = macd[-1] self.current_indicators[period_name]['macd_sig'] = macd_sig[-1] self.current_indicators[period_name]['macd_hist'] = macd_hist[-1] self.current_indicators[period_name]['macd_hist_diff'] = Decimal(macd_hist[-1]) - Decimal(macd_hist[-2])
def calculate_vol_macd(self, period_name, volumes): macd, macd_sig, macd_hist = talib.MACD(volumes, fastperiod=50, slowperiod=200, signalperiod=14) self.current_indicators[period_name]['vol_macd'] = macd[-1] self.current_indicators[period_name]['vol_macd_sig'] = macd_sig[-1] self.current_indicators[period_name]['vol_macd_hist'] = macd_hist[-1]
def MACD(df, short_win=12, long_win=26, macd_win=9): # talib??MACD prices = np.array(df['close']) macd_tmp = talib.MACD(prices, fastperiod=short_win, slowperiod=long_win, signalperiod=macd_win) df['macd_dif'] = macd_tmp[0] df['macd_dea'] = macd_tmp[1] df['macd'] = macd_tmp[2] return df
def macd_rule_2(df, symbol): try: df = MACD(df) except: return False input_1 = -3 input_2 = -0.2 index = -1 return (df['macd_dif'][index] > input_1) & \ (df['macd_dif'][index] < input_2) & \ (df['macd_dif'][index] > df['macd_dea'][index]) & \ ((df['macd_dea'][index-1] > df['macd_dif'][index-1]) | (abs(df['macd_dea'][index-1] - df['macd_dif'][index-1]) < 0.007))
def macd_rule_1(df, index = -1): try: if not {'macd_dif', 'macd_dea', 'macd'}.issubset(df.columns): df = MACD(df) except Exception as e: print(e) return False return (df['macd_dif'][index] > df['macd_dea'][index]) & \ ((df['macd_dea'][index-1] > df['macd_dif'][index-1]) | (abs(df['macd_dea'][index-1] - df['macd_dif'][index-1]) < 0.007))
def macd_rule_2(df, index = -1): try: if not {'macd_dif', 'macd_dea', 'macd'}.issubset(df.columns): df = MACD(df) except Exception as e: print(e) return False input = 0.05 return (df['macd_dif'][index] < input) & (df['macd_dea'][index] < input)
def macd(self, fastPeriod, slowPeriod, signalPeriod, array=False): """MACD??""" macd, signal, hist = talib.MACD(self.close, fastPeriod, slowPeriod, signalPeriod) if array: return macd, signal, hist return macd[-1], signal[-1], hist[-1] # ----------------------------------------------------------------------
def MACD(self): """??????????????? MACD: (12-day EMA - 26-day EMA) ??? ?? Signal Line: 9-day EMA of MACD ??, ?? MACD Histogram: MACD - Signal Line ???, ??? return : macd, macdsignal, macdhist(??)""" closes = self.getCloses() return talib.MACD(closes)
def MACD(closes): """??????????????? MACD: (12-day EMA - 26-day EMA) ??? ?? Signal Line: 9-day EMA of MACD ??, ?? MACD Histogram: MACD - Signal Line ???, ??? return : macd, macdsignal, macdhist(??)""" return talib.MACD(closes)
