Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- # File: signal_strategy_tv_mt5.py
- import MetaTrader5 as mt5
- import pandas as pd
- from tradingview_scraper.symbols.technicals import Indicators
- import mplfinance as mpf
- import matplotlib.pyplot as plt
- from matplotlib.patches import Rectangle
- # === Kelas Deteksi Sinyal ===
- class SignalStrategyTvMt5:
- def __init__(self, symbol="BTCUSD", timeframe=mt5.TIMEFRAME_D1, bars=100):
- self.symbol = symbol
- self.timeframe = timeframe
- self.bars = bars
- self.df = None
- self.signal_df = None
- def connect_mt5(self):
- if not mt5.initialize():
- raise Exception("MT5 gagal di-inisialisasi")
- print("[✓] MT5 connected")
- def shutdown_mt5(self):
- mt5.shutdown()
- print("[✓] MT5 shutdown selesai")
- def deteksi_ob(self, indikator):
- rsi = indikator.get("RSI", 0)
- stoch_k = indikator.get("Stoch.K", 0)
- if rsi > 70:
- return "OB - RSI"
- elif stoch_k > 80:
- return "OB - Stoch"
- return "Normal"
- def deteksi_fvg(self, df):
- result = {}
- for i in range(2, len(df)):
- t = df['time'].iloc[i]
- if df['low'].iloc[i] > df['high'].iloc[i - 2]:
- result[t] = "Bullish FVG"
- elif df['high'].iloc[i] < df['low'].iloc[i - 2]:
- result[t] = "Bearish FVG"
- else:
- result[t] = ""
- return result
- def deteksi_bos(self, df, period=5):
- result = {}
- for i in range(period, len(df) - period):
- t = df['time'].iloc[i]
- swing_high = max(df['high'].iloc[i - period:i])
- swing_low = min(df['low'].iloc[i - period:i])
- if df['close'].iloc[i] > swing_high:
- result[t] = "BOS Bullish"
- elif df['close'].iloc[i] < swing_low:
- result[t] = "BOS Bearish"
- else:
- result[t] = ""
- return result
- def deteksi_choch(self, df, period=5):
- result = {}
- for i in range(period + 1, len(df)):
- t = df['time'].iloc[i]
- prev_high = max(df['high'].iloc[i - period - 1:i - 1])
- prev_low = min(df['low'].iloc[i - period - 1:i - 1])
- curr = df['close'].iloc[i]
- prev = df['close'].iloc[i - 1]
- if prev > prev_high and curr < prev_low:
- result[t] = "CHoCH Bearish"
- elif prev < prev_low and curr > prev_high:
- result[t] = "CHoCH Bullish"
- else:
- result[t] = ""
- return result
- def deteksi_engulfing(self, df):
- result = {}
- for i in range(1, len(df)):
- prev_open = df['open'].iloc[i - 1]
- prev_close = df['close'].iloc[i - 1]
- curr_open = df['open'].iloc[i]
- curr_close = df['close'].iloc[i]
- if prev_close < prev_open and curr_close > curr_open and curr_close > prev_open and curr_open < prev_close:
- result[df['time'].iloc[i]] = "Bullish Engulfing"
- elif prev_close > prev_open and curr_close < curr_open and curr_close < prev_open and curr_open > prev_close:
- result[df['time'].iloc[i]] = "Bearish Engulfing"
- else:
- result[df['time'].iloc[i]] = ""
- return result
- def konfirmasi_sinyal_buy_sell(self, df, signal_df):
- result = {}
- for i in range(2, len(signal_df)):
- row = signal_df.iloc[i]
- t = row['time']
- fvg = row['FVG']
- ob = row['OB']
- bos = row['BOS']
- choch = row['CHoCH']
- engulf = row.get('ENGULFING', "")
- harga = df['close'].iloc[i]
- arah = None
- if "Bullish" in fvg and "Bullish" in engulf and ("Bullish" in bos or "Bullish" in choch):
- arah = "BUY"
- elif "Bearish" in fvg and "Bearish" in engulf and ("Bearish" in bos or "Bearish" in choch):
- arah = "SELL"
- if arah == "BUY":
- high_fvg = df['high'].iloc[i - 2]
- low_fvg = df['low'].iloc[i]
- fibo_0618 = low_fvg + (high_fvg - low_fvg) * 0.618
- if harga > fibo_0618 and harga < high_fvg:
- result[t] = f"Konfirmasi {arah} dengan SL TP 1:2"
- else:
- result[t] = ""
- elif arah == "SELL":
- low_fvg = df['low'].iloc[i - 2]
- high_fvg = df['high'].iloc[i]
- fibo_0618 = low_fvg + (high_fvg - low_fvg) * 0.618
- if harga < fibo_0618 and harga > low_fvg:
- result[t] = f"Konfirmasi {arah} dengan SL TP 1:2"
- else:
- result[t] = ""
- else:
- result[t] = ""
- return result
- def get_data(self):
- self.connect_mt5()
- data = mt5.copy_rates_from_pos(self.symbol, self.timeframe, 0, self.bars)
- if data is None:
- raise Exception("Gagal ambil data dari MT5")
- df = pd.DataFrame(data)
- df['time'] = pd.to_datetime(df['time'], unit='s')
- self.df = df
- indikator = Indicators().scrape(
- exchange="BINANCE",
- symbol=self.symbol,
- timeframe="1d",
- indicators=["RSI", "Stoch.K"]
- )
- ob_status = self.deteksi_ob(indikator)
- fvg = self.deteksi_fvg(df)
- bos = self.deteksi_bos(df)
- choch = self.deteksi_choch(df)
- engulfing = self.deteksi_engulfing(df)
- signal_df = pd.DataFrame({
- 'time': df['time'],
- 'OB': [ob_status] * len(df),
- 'FVG': df['time'].map(fvg),
- 'BOS': df['time'].map(bos),
- 'CHoCH': df['time'].map(choch),
- 'ENGULFING': df['time'].map(engulfing),
- })
- konfirmasi = self.konfirmasi_sinyal_buy_sell(df, signal_df)
- signal_df['KONFIRMASI'] = df['time'].map(konfirmasi)
- self.shutdown_mt5()
- self.signal_df = signal_df
- return df, signal_df
- # === Kelas Console & Visual ===
- class ShowConsoleSignal:
- def __init__(self):
- self.strategy = SignalStrategyTvMt5()
- self.df = None
- self.sinyal = None
- def run(self):
- print("📊 === ALGO SIGNAL CONSOLE ===")
- try:
- self.df, self.sinyal = self.strategy.get_data()
- except Exception as e:
- print(f"[!] ERROR: {e}")
- return
- while True:
- print("\nPilih menu:")
- print("[1] Lihat OHLC MT5 terakhir")
- print("[2] Lihat sinyal terakhir")
- print("[3] Tampilkan grafik sinyal (mplfinance)")
- print("[4] Keluar")
- pilihan = input(">> Pilih [1-4]: ").strip()
- if pilihan == "1":
- print(self.df.tail())
- elif pilihan == "2":
- print(self.sinyal.tail())
- elif pilihan == "3":
- self.mplft()
- elif pilihan == "4":
- break
- else:
- print("Pilihan tidak valid.")
- def mplft(self):
- if self.df is None or self.sinyal is None:
- print("[!] Data belum tersedia.")
- return
- df_plot = self.df.copy()
- df_plot.set_index('time', inplace=True)
- df_plot = df_plot[['open', 'high', 'low', 'close']]
- fig, axlist = mpf.plot(
- df_plot,
- type='candle',
- style='yahoo',
- title='Sinyal Deteksi (OB, FVG, BOS, CHoCH, ENGULFING)',
- ylabel='Harga',
- volume=False,
- returnfig=True
- )
- ax = axlist[0]
- for i in range(2, len(self.df)):
- row = self.sinyal.iloc[i]
- time = self.df['time'].iloc[i]
- close = self.df['close'].iloc[i]
- if isinstance(row['KONFIRMASI'], str) and row['KONFIRMASI'] != "":
- arah = "BUY" if "BUY" in row['KONFIRMASI'] else "SELL"
- if arah == "BUY":
- high_fvg = self.df['high'].iloc[i - 2]
- low_fvg = self.df['low'].iloc[i]
- fibo_0 = high_fvg
- fibo_1 = low_fvg
- fibo_0618 = fibo_1 + (fibo_0 - fibo_1) * 0.618
- sl = fibo_1
- tp = fibo_0618 + (fibo_0618 - sl) * 2
- ax.text(time, close * 0.99, "BUY", color='green', fontsize=8, ha='center', va='top')
- else:
- low_fvg = self.df['low'].iloc[i - 2]
- high_fvg = self.df['high'].iloc[i]
- fibo_0 = low_fvg
- fibo_1 = high_fvg
- fibo_0618 = fibo_0 + (fibo_1 - fibo_0) * 0.618
- sl = fibo_1
- tp = fibo_0618 - (sl - fibo_0618) * 2
- ax.text(time, close * 1.01, "SELL", color='white', fontsize=8, ha='center', va='bottom')
- ax.axhline(fibo_0, color='gray', linestyle='--', linewidth=0.8)
- ax.axhline(fibo_0618, color='blue', linestyle='--', linewidth=1.2)
- ax.axhline(fibo_1, color='gray', linestyle='--', linewidth=0.8)
- box_color = 'green' if arah == 'BUY' else 'red'
- y_bottom = min(sl, tp)
- y_top = max(sl, tp)
- height = y_top - y_bottom
- rect = Rectangle((time, y_bottom), pd.Timedelta(days=0.5), height, color=box_color, alpha=0.2)
- ax.add_patch(rect)
- plt.tight_layout()
- plt.show()
- # === Entry Point ===
- if __name__ == "__main__":
- ShowConsoleSignal().run()
Advertisement
Add Comment
Please, Sign In to add comment