""" Robot Analis Saham IDX - Multi-Indicator Swing Screener =========================================================== Scan semua saham IDX (dari tickers_idx.txt), cari kandidat rebound yang oversold TAPI dikonfirmasi beberapa indikator lain sekaligus (bukan cuma RSI < 30 doang) — untuk swing/position trading, bukan scalping. Indikator yang dipakai & kenapa: 1. RSI(14) < 30 -> basis oversold 2. MACD bullish crossover -> momentum udah mulai belok naik 3. Close > SMA trend -> filter trend jangka panjang (hindari saham yang emang downtrend struktural, "falling knife") 4. Bollinger Bands(20,2) -> harga secara statistik memang murah (dekat/di bawah lower band) 5. Volume > 1.5x rata-rata -> oversold + volume naik = tanda akumulasi, bukan cuma sepi transaksi 6. Stochastic RSI -> lebih sensitif dari RSI biasa, dipakai untuk nangkep titik balik (oversold + %K cross %D) 7. MA Golden Cross (20/50) -> konfirmasi momentum jangka menengah mulai naik Setiap indikator kasih poin -> total SCORE (maks MAX_POSSIBLE_SCORE). Semakin tinggi score, semakin banyak konfirmasi yang align. Ini jauh lebih valid daripada filter RSI tunggal. Catatan: "Foreign Flow" (net asing) SENGAJA belum dimasukkan karena datanya gak tersedia gratis di yfinance - itu data khusus dari RTI Business/Stockbit/ API berbayar seperti DataSectors atau Invezgo. Kalau nanti kamu punya akses salah satu dari itu, strukturnya bisa ditambah sebagai indikator ke-8. Cara pakai: 1. pip install yfinance pandas matplotlib mplfinance requests --break-system-packages 2. Pastikan tickers_idx.txt ada di folder yang sama 3. Isi TELEGRAM_BOT_TOKEN dan TELEGRAM_CHAT_ID 4. Jalankan: python rsi_screener.py 5. Chart candlestick otomatis tersimpan di folder charts/ dan dikirim ke Telegram """ import os import sys import time import warnings from datetime import datetime, timedelta from zoneinfo import ZoneInfo # Suppress pandas warnings warnings.filterwarnings('ignore', category=FutureWarning) warnings.filterwarnings('ignore', category=UserWarning) import requests import pandas as pd import numpy as np import yfinance as yf import matplotlib matplotlib.use("Agg") # biar bisa jalan tanpa display/GUI (server/headless) import matplotlib.lines as mlines import mplfinance as mpf # AlphaX Pattern Detection sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) try: from alphax import detect_all_bullish_patterns ALPHAX_AVAILABLE = True except ImportError as e: print(f"[WARN] AlphaX module tidak tersedia: {e}") ALPHAX_AVAILABLE = False # ============ KONFIGURASI ============ TELEGRAM_BOT_TOKEN = "ISI_TOKEN_BOT_TELEGRAM_KAMU" TELEGRAM_CHAT_ID = "ISI_CHAT_ID_KAMU" # MARKET HOURS DETECTION MARKET_DAYS = [0, 1, 2, 3, 4] # Senin-Jumat (0=Monday, 4=Friday) MARKET_HOUR_START = 9 # 09:00 WIB MARKET_HOUR_END = 16 # 16:00 WIB (bursa tutup 16:15) MARKET_TZ = "Asia/Jakarta" TICKER_FILE = "tickers_idx.txt" MIN_VOLUME = 100_000 # filter saham terlalu tipis transaksinya BATCH_SIZE = 50 BATCH_DELAY_SEC = 2 # --- TIMEFRAME --- # "1h" = candle per jam, trend filter pakai SMA200 (~1 bulan lebih data jam). # Catatan: makin pendek timeframe, makin banyak sinyal tapi makin # "berisik" (noise) - cocok kalau kamu mau screening lebih sering, # tapi tetap bukan scalping selama MIN_SCORE & filter likuiditas dijaga ketat. # "1d" = candle harian, trend filter pakai SMA200 (cocok untuk swing mingguan-bulanan) # "1wk" = candle mingguan, trend filter pakai SMA50 (~1 tahun), cocok posisi lebih panjang TIMEFRAME = "1d" TIMEFRAME_CONFIG = { "1h": {"interval": "60m", "period": "3mo", "trend_sma": 200}, "1d": {"interval": "1d", "period": "2y", "trend_sma": 200}, "1wk": {"interval": "1wk", "period": "5y", "trend_sma": 50}, } # --- Bobot skor tiap indikator (bisa kamu tuning) --- SCORE_RSI_OVERSOLD = 2 SCORE_MACD_CROSSOVER = 2 SCORE_ABOVE_TREND_SMA = 2 SCORE_NEAR_BB_LOWER = 1 SCORE_VOLUME_SPIKE = 1 SCORE_STOCH_RSI_BULLISH = 2 # Stochastic RSI oversold + baru cross naik SCORE_MA_GOLDEN_CROSS = 1 # MA cepat (20) baru cross naik MA lambat (50) SCORE_ALPHAX_PATTERN = 5 # AlphaX pattern detection (max bonus dari confluence score) # --- Fibonacci Retracement --- SHOW_FIBONACCI = True # Tampilkan Fibonacci di chart dan analisis FIBONACCI_PERIOD = 60 # Period untuk cari high/low (candle) FIBONACCI_LEVELS = [0, 0.236, 0.382, 0.5, 0.618, 0.786, 1] # Fibonacci levels FIBONACCI_COLORS = ["#ff1744", "#ff9800", "#ffeb3b", "#9e9e9e", "#ffeb3b", "#ff9800", "#ff1744"] MAX_POSSIBLE_SCORE = (SCORE_RSI_OVERSOLD + SCORE_MACD_CROSSOVER + SCORE_ABOVE_TREND_SMA + SCORE_NEAR_BB_LOWER + SCORE_VOLUME_SPIKE + SCORE_STOCH_RSI_BULLISH + SCORE_MA_GOLDEN_CROSS + SCORE_ALPHAX_PATTERN) MIN_SCORE = 5 # minimal skor total biar masuk hasil (dari maks MAX_POSSIBLE_SCORE) RSI_THRESHOLD = 30 BB_PROXIMITY = 1.02 # dianggap "dekat lower band" kalau close <= lower_band * 1.02 VOLUME_SPIKE_MULT = 1.5 STOCH_RSI_PERIOD = 14 STOCH_SMOOTH_K = 3 STOCH_SMOOTH_D = 3 STOCH_OVERSOLD = 20 # StochRSI %K di bawah ini dianggap oversold (skala 0-100) MA_FAST_PERIOD = 20 MA_SLOW_PERIOD = 50 # --- AlphaX Pattern Detection Configuration --- ALPHAX_ENABLED = True ALPHAX_MIN_CONFLUENCE = 5 # Minimum confluence score (0-10) ALPHAX_HTF_TIMEFRAME = "1wk" # Higher timeframe untuk bias check (1wk untuk daily chart) ALPHAX_HTF_FAST_EMA = 21 ALPHAX_HTF_SLOW_EMA = 55 ALPHAX_CHOP_MAX = 62.0 # Chop index threshold (di atas ini = choppy, avoid) ALPHAX_PIVOT_LEFT_LB = 10 ALPHAX_PIVOT_RIGHT_LB = 10 ALPHAX_MAX_PIVOTS = 500 ALPHAX_MIN_RR = 1.5 # Minimum Risk:Reward ratio # --- FILTER LIKUIDITAS & FREE FLOAT --- # Ini FILTER KERAS (hard filter) - bukan penambah skor, tapi syarat wajib # supaya saham layak dianalisa sama sekali (saham gak likuid/free float kecil # gampang digoreng & susah exit posisi walau sinyal teknikalnya bagus). MIN_AVG_VALUE_TRADED = 1_000_000_000 # rata-rata nilai transaksi 20 hari (Rupiah), pakai data harga yang sudah ada MIN_FREE_FLOAT_PCT = 25 # minimal free float (%), dari cache free_float_idx.csv FREE_FLOAT_FILE = "free_float_idx.csv" MAX_CLOSE_PRICE = 300 # maksimal harga close (Rupiah), kalau None = tanpa batas MIN_CLOSE_PRICE = 50 # minimal harga close (Rupiah), kalau None = tanpa batas # Kalau True: saham TANPA data free float di cache otomatis di-skip (aman, tapi # bisa buang banyak saham kalau cache belum lengkap). # Kalau False: saham tanpa data free float TETAP diproses (filter ini dilewati # khusus buat saham itu), cocok dipakai sebelum cache lengkap. REQUIRE_FREE_FLOAT_DATA = False # --- JADWAL: kapan robot boleh jalan --- RUN_DAYS = [0, 1, 2, 3, 4] # 0=Senin ... 4=Jumat RUN_HOUR_START = 9 # jam mulai boleh screening (WIB) RUN_HOUR_END = 16 # jam terakhir boleh screening (WIB) ENFORCE_SCHEDULE = True # Weekend/after-hours mode: tetap screening walau market tutup # Kalau True: screening jalan 7 hari, tidak peduli RUN_DAYS # Kalau False: ikut RUN_DAYS (Senin-Jumat saja) ALLOW_WEEKEND_SCREENING = True # --- SELF-LOOP MODE --- # True = script jalan terus sebagai proses lokal, otomatis screening # tiap jam bulat (13:00, 14:00, dst) selama masih di jendela # RUN_HOUR_START-RUN_HOUR_END, tanpa perlu cron/scheduler eksternal. # False = jalan sekali lalu exit (dipakai kalau kamu tetap mau pakai cron). SELF_LOOP = True # --- CHART OTOMATIS --- CHARTS_DIR = "charts" # folder lokal buat simpan PNG SEND_CHART_TO_TELEGRAM = True # kirim chart sebagai foto ke Telegram juga MAX_CHARTS_PER_RUN = 15 # batasi biar gak spam Telegram kalau hasil banyak CHART_LOOKBACK_CANDLES = 90 # berapa candle terakhir yang ditampilkan di chart # ======================================= def is_market_open() -> tuple[bool, str]: """ Check apakah IDX market sedang buka. Return (is_open, message) """ now = datetime.now(ZoneInfo(MARKET_TZ)) # Check hari if now.weekday() not in MARKET_DAYS: return False, f"Market tutup (hari {now.strftime('%A')})" # Check jam if not (MARKET_HOUR_START <= now.hour < MARKET_HOUR_END): status = "sebelum buka" if now.hour < MARKET_HOUR_START else "sudah tutup" return False, f"Market {status} (jam {now.hour}:00 WIB)" return True, "Market buka" def is_within_schedule() -> bool: if not ENFORCE_SCHEDULE: return True # Weekend/after-hours mode: kalau True, skip hari check if ALLOW_WEEKEND_SCREENING: # Cuma check jam (kalau mau), tapi tetap jalan walau weekend if RUN_HOUR_START <= datetime.now(ZoneInfo(MARKET_TZ)).hour < RUN_HOUR_END: return True # Jam pas, jalan # Di luar jam, tetap return True biar bisa jalan manual/weekend return True # Original logic: ikut RUN_DAYS (Senin-Jumat) now = datetime.now(ZoneInfo("Asia/Jakarta")) - timedelta(hours=7) if now.weekday() not in RUN_DAYS: print(f"[SKIP] Hari ini ({now.strftime('%A')}) di luar RUN_DAYS.") return False if not (RUN_HOUR_START <= now.hour < RUN_HOUR_END): print(f"[SKIP] Jam sekarang ({now.hour}:00 WIB) di luar jadwal.") return False return True def load_tickers() -> list[str]: with open(TICKER_FILE) as f: codes = [line.strip() for line in f if line.strip()] return [f"{code}.JK" for code in codes] def load_free_float() -> dict: """ Baca cache free_float_idx.csv (dari update_free_float.py). Return dict {kode_saham: persen_free_float}. Kalau file belum ada, return dict kosong (semua saham diperlakukan 'data gak tersedia'). """ if not os.path.exists(FREE_FLOAT_FILE): print(f"[WARN] {FREE_FLOAT_FILE} belum ada - jalankan update_free_float.py dulu " f"kalau mau filter free float aktif penuh.") return {} free_float_map = {} with open(FREE_FLOAT_FILE) as f: next(f) # skip header for line in f: parts = line.strip().split(",") if len(parts) == 2 and parts[1]: try: free_float_map[parts[0]] = float(parts[1]) except ValueError: continue return free_float_map # ============ INDIKATOR ============ def calc_rsi(close: pd.Series, period: int = 14) -> pd.Series: delta = close.diff() gain = delta.clip(lower=0) loss = -delta.clip(upper=0) avg_gain = gain.ewm(alpha=1 / period, min_periods=period, adjust=False).mean() avg_loss = loss.ewm(alpha=1 / period, min_periods=period, adjust=False).mean() rs = avg_gain / avg_loss return 100 - (100 / (1 + rs)) def calc_macd(close: pd.Series, fast=12, slow=26, signal=9): ema_fast = close.ewm(span=fast, adjust=False).mean() ema_slow = close.ewm(span=slow, adjust=False).mean() macd_line = ema_fast - ema_slow signal_line = macd_line.ewm(span=signal, adjust=False).mean() histogram = macd_line - signal_line return macd_line, signal_line, histogram def calc_bollinger(close: pd.Series, period=20, num_std=2): sma = close.rolling(window=period).mean() std = close.rolling(window=period).std() upper = sma + num_std * std lower = sma - num_std * std return sma, upper, lower def calc_fibonacci_levels(df: pd.DataFrame, period: int = FIBONACCI_PERIOD) -> dict: """ Calculate Fibonacci Retracement levels dari highest high dan lowest low dalam period tertentu. Return dict dengan levels dan harga """ if len(df) < period: return {} # Ambil data dari period terakhir lookback_df = df.tail(period) high_price = lookback_df['High'].max() low_price = lookback_df['Low'].min() diff = high_price - low_price if diff <= 0: return {} # Calculate Fibonacci levels fib_levels = {} for level_pct in FIBONACCI_LEVELS: # Level dari atas ke bawah (0% di high, 100% di low) level_price = high_price - (diff * level_pct) fib_levels[f"{level_pct*100:.1f}%"] = level_price return { 'high': high_price, 'low': low_price, 'period': period, 'levels': fib_levels } def calc_stochastic_rsi(rsi: pd.Series, period=STOCH_RSI_PERIOD, smooth_k=STOCH_SMOOTH_K, smooth_d=STOCH_SMOOTH_D): """ Stochastic RSI: 'stochastic dari RSI' - lebih sensitif daripada RSI biasa, bagus buat nangkep titik balik oversold/overbought yang lebih presisi. Skala 0-100 (bukan 0-1) biar konsisten sama style StochRSI di TradingView. """ min_rsi = rsi.rolling(window=period).min() max_rsi = rsi.rolling(window=period).max() stoch = (rsi - min_rsi) / (max_rsi - min_rsi) * 100 k = stoch.rolling(window=smooth_k).mean() d = k.rolling(window=smooth_d).mean() return k, d def analyze(df: pd.DataFrame, trend_sma_period: int, ticker: str = None, free_float_map: dict = None) -> dict | None: if len(df) < trend_sma_period + 5: return None # data belum cukup untuk trend SMA panjang # Set ticker attribute untuk AlphaX (butuh untuk HTF data download) if ticker is not None: df.attrs['ticker'] = f"{ticker}.JK" close = df["Close"] df["rsi"] = calc_rsi(close) macd_line, signal_line, hist = calc_macd(close) df["macd"], df["macd_signal"], df["macd_hist"] = macd_line, signal_line, hist df["sma_trend"] = close.rolling(window=trend_sma_period).mean() _, bb_upper, bb_lower = calc_bollinger(close) df["bb_lower"] = bb_lower df["vol_sma20"] = df["Volume"].rolling(window=20).mean() df["stoch_k"], df["stoch_d"] = calc_stochastic_rsi(df["rsi"]) df["ma_fast"] = close.rolling(window=MA_FAST_PERIOD).mean() df["ma_slow"] = close.rolling(window=MA_SLOW_PERIOD).mean() df["value_traded"] = df["Close"] * df["Volume"] df["avg_value_20"] = df["value_traded"].rolling(window=20).mean() latest = df.iloc[-1] prev = df.iloc[-2] # Calculate Fibonacci levels fib_levels = calc_fibonacci_levels(df, FIBONACCI_PERIOD) if SHOW_FIBONACCI else {} required_cols = ["rsi", "macd", "sma_trend", "bb_lower", "vol_sma20", "stoch_k", "stoch_d", "ma_fast", "ma_slow", "avg_value_20"] if latest[required_cols].isna().any(): return None # --- HARD FILTER: harga close (min/max price) --- if MAX_CLOSE_PRICE is not None and latest["Close"] > MAX_CLOSE_PRICE: return None if MIN_CLOSE_PRICE is not None and latest["Close"] < MIN_CLOSE_PRICE: return None # --- HARD FILTER: likuiditas (rata-rata nilai transaksi 20 hari) --- if latest["avg_value_20"] < MIN_AVG_VALUE_TRADED: return None # --- HARD FILTER: free float --- free_float_pct = None if free_float_map is not None and ticker is not None: free_float_pct = free_float_map.get(ticker) if free_float_pct is not None and free_float_pct < MIN_FREE_FLOAT_PCT: return None # free float ada datanya, tapi di bawah syarat -> tolak if free_float_pct is None and REQUIRE_FREE_FLOAT_DATA: return None # data gak ada & wajib ada -> tolak score = 0 reasons = [] # 1. RSI oversold if latest["rsi"] < RSI_THRESHOLD: score += SCORE_RSI_OVERSOLD reasons.append("RSI oversold") # 2. MACD bullish crossover (baru cross di candle terakhir) macd_crossed_up = (prev["macd"] <= prev["macd_signal"]) and (latest["macd"] > latest["macd_signal"]) if macd_crossed_up: score += SCORE_MACD_CROSSOVER reasons.append("MACD bullish crossover") # 3. Trend filter: harga masih di atas SMA jangka panjang if latest["Close"] > latest["sma_trend"]: score += SCORE_ABOVE_TREND_SMA reasons.append(f"Close > SMA{trend_sma_period} (trend intact)") # 4. Dekat/di bawah lower Bollinger Band if latest["Close"] <= latest["bb_lower"] * BB_PROXIMITY: score += SCORE_NEAR_BB_LOWER reasons.append("Dekat lower Bollinger Band") # 5. Volume spike vs rata-rata 20 hari if latest["Volume"] > latest["vol_sma20"] * VOLUME_SPIKE_MULT: score += SCORE_VOLUME_SPIKE reasons.append("Volume spike") # 6. Stochastic RSI oversold + baru cross naik (%K cross di atas %D) stoch_oversold = latest["stoch_k"] < STOCH_OVERSOLD stoch_crossed_up = (prev["stoch_k"] <= prev["stoch_d"]) and (latest["stoch_k"] > latest["stoch_d"]) if stoch_oversold and stoch_crossed_up: score += SCORE_STOCH_RSI_BULLISH reasons.append("Stochastic RSI oversold + cross naik") # 7. MA Golden Cross (MA cepat baru potong ke atas MA lambat) ma_crossed_up = (prev["ma_fast"] <= prev["ma_slow"]) and (latest["ma_fast"] > latest["ma_slow"]) if ma_crossed_up: score += SCORE_MA_GOLDEN_CROSS reasons.append(f"Golden Cross MA{MA_FAST_PERIOD}/{MA_SLOW_PERIOD}") # ===== NEW: AlphaX Pattern Detection ===== alphax_pattern = None if ALPHAX_ENABLED and ALPHAX_AVAILABLE: try: pattern = detect_all_bullish_patterns( df, left_lb=ALPHAX_PIVOT_LEFT_LB, right_lb=ALPHAX_PIVOT_RIGHT_LB, max_pivots=ALPHAX_MAX_PIVOTS, min_confluence=ALPHAX_MIN_CONFLUENCE, htf_timeframe=ALPHAX_HTF_TIMEFRAME, htf_fast=ALPHAX_HTF_FAST_EMA, htf_slow=ALPHAX_HTF_SLOW_EMA, chop_max=ALPHAX_CHOP_MAX, min_rr=ALPHAX_MIN_RR ) if pattern.detected: confluence = min(ALPHAX_MIN_CONFLUENCE, 10) # Get confluence score alphax_score = min(int(confluence), SCORE_ALPHAX_PATTERN) score += alphax_score reasons.append(f"{pattern.name} (conf: {confluence}/10)") alphax_pattern = { "name": pattern.name, "is_bullish": pattern.is_bullish, "entry_price": pattern.entry_price, "stop_price": pattern.stop_price, "target_price": pattern.target_price, "confluence": int(confluence) } except Exception as e: print(f"[WARN] AlphaX detection error untuk {ticker}: {e}") if latest["Volume"] < MIN_VOLUME: return None if score < MIN_SCORE: return None return { "score": score, "rsi": round(float(latest["rsi"]), 2), "stoch_k": round(float(latest["stoch_k"]), 2), "close": round(float(latest["Close"]), 2), "volume": int(latest["Volume"]), "avg_value_traded": float(latest["avg_value_20"]), "free_float_pct": free_float_pct, "reasons": reasons, "alphax_pattern": alphax_pattern, # AlphaX pattern info (optional) "fibonacci": fib_levels if fib_levels else None, # Fibonacci levels } # ============ CHART GENERATOR ============ def generate_chart(ticker: str, df: pd.DataFrame, result: dict, trend_sma_period: int) -> str | None: """ Bikin chart candlestick PNG (pakai mplfinance) + panel volume, dengan overlay SMA trend, BB lower, Fibonacci levels, dan anotasi titik sinyal (RSI oversold, MACD crossover) - mirip gaya analisa manual TradingView. """ try: os.makedirs(CHARTS_DIR, exist_ok=True) plot_df = df.tail(CHART_LOOKBACK_CANDLES).copy() # Marker: cuma ada nilai di titik sinyal, NaN di titik lain (biar mplfinance # cuma gambar scatter dot di titik itu saja) oversold_marker = plot_df["Close"].where(plot_df["rsi"] < RSI_THRESHOLD) macd_cross_mask = (plot_df["macd"].shift(1) <= plot_df["macd_signal"].shift(1)) & \ (plot_df["macd"] > plot_df["macd_signal"]) macd_cross_marker = plot_df["Close"].where(macd_cross_mask) ma_cross_mask = (plot_df["ma_fast"].shift(1) <= plot_df["ma_slow"].shift(1)) & \ (plot_df["ma_fast"] > plot_df["ma_slow"]) ma_cross_marker = plot_df["Close"].where(ma_cross_mask) addplots = [ mpf.make_addplot(plot_df["sma_trend"], color="#888888", linestyle="--", width=1.2), mpf.make_addplot(plot_df["bb_lower"], color="#d62728", linestyle=":", width=0.8), mpf.make_addplot(plot_df["ma_fast"], color="#9467bd", linestyle="-", width=1.0), mpf.make_addplot(plot_df["ma_slow"], color="#8c564b", linestyle="-", width=1.0), mpf.make_addplot(plot_df["vol_sma20"], panel=1, color="#555555", width=1), mpf.make_addplot(plot_df["stoch_k"], panel=2, color="#1f77b4", width=1.1, ylabel="StochRSI"), mpf.make_addplot(plot_df["stoch_d"], panel=2, color="#ff7f0e", width=1.0), ] has_oversold = oversold_marker.notna().any() has_macd_cross = macd_cross_marker.notna().any() has_ma_cross = ma_cross_marker.notna().any() if has_oversold: addplots.append(mpf.make_addplot( oversold_marker, type="scatter", markersize=70, marker="v", color="orange")) if has_macd_cross: addplots.append(mpf.make_addplot( macd_cross_marker, type="scatter", markersize=90, marker="^", color="blue")) if has_ma_cross: addplots.append(mpf.make_addplot( ma_cross_marker, type="scatter", markersize=90, marker="*", color="gold")) # Dark mode style dengan background hitam style = mpf.make_mpf_style( base_mpf_style="nightclouds", # Gunakan dark theme sebagai base rc={ "font.size": 8, "axes.facecolor": "#1a1a1a", # Background utama hitam "figure.facecolor": "#0d0d0d", # Figure background hitam "axes.edgecolor": "#333333", # Border abu-abu gelap "text.color": "#cccccc", # Text abu-abu terang "ytick.color": "#cccccc", "xtick.color": "#cccccc", "grid.color": "#333333", # Grid abu-abu gelap "grid.linestyle": "--", "grid.linewidth": 0.5, } ) title_reasons = ", ".join(result["reasons"]) ff_text = f"{result['free_float_pct']}%" if result.get("free_float_pct") is not None else "N/A" value_b = result["avg_value_traded"] / 1_000_000_000 info_line = f"Free Float: {ff_text} | Avg Value 20D: Rp {value_b:.2f} Miliar" fig, axes = mpf.plot( plot_df[["Open", "High", "Low", "Close", "Volume"]], type="candle", style=style, volume=True, addplot=addplots, panel_ratios=(3, 1, 1), title=f"\n\n{ticker} — Score {result['score']}/{MAX_POSSIBLE_SCORE}\n{title_reasons}{info_line}", figsize=(11, 8.5), returnfig=True, datetime_format="%d %b", xrotation=45, ) # Add Fibonacci horizontal lines if result.get("fibonacci") and result["fibonacci"].get("levels"): fib_ax = axes[0] # Price panel fib_data = result["fibonacci"] # Add horizontal lines untuk tiap Fibonacci level for i, (level_name, level_price) in enumerate(fib_data["levels"].items()): if level_name in ["0.0%", "100.0%"]: color = FIBONACCI_COLORS[0] if level_name == "0.0%" else FIBONACCI_COLORS[-1] width = 2.0 linestyle = "-" else: # Find index dengan approximate matching (untuk handle floating point precision) level_value = float(level_name.rstrip("%")) / 100 idx = None for j, fib_level in enumerate(FIBONACCI_LEVELS): if abs(fib_level - level_value) < 0.001: # Tolerance untuk floating point idx = j break if idx is None: idx = i # Fallback ke iteration index color = FIBONACCI_COLORS[idx] width = 1.0 linestyle = "--" fib_ax.axhline(y=level_price, color=color, linestyle=linestyle, linewidth=width, alpha=0.7) # Add label untuk level fib_ax.text(0.01, level_price, f" {level_name} ({level_price:,.0f})", transform=fib_ax.get_yaxis_transform(), fontsize=7, verticalalignment='center', color=color, weight='bold') # Garis referensi oversold/overbought di panel Stochastic RSI (panel index 2 -> axes[4] karena # tiap panel punya twin-axis, jadi urutannya: [price, price2, volume, volume2, stoch, stoch2]) try: stoch_ax = axes[4] stoch_ax.axhline(STOCH_OVERSOLD, color="gray", linestyle=":", linewidth=0.8) stoch_ax.axhline(100 - STOCH_OVERSOLD, color="gray", linestyle=":", linewidth=0.8) stoch_ax.set_ylim(0, 100) except (IndexError, AttributeError): pass # mplfinance gak auto-legend addplot, jadi bikin manual legend_elems = [ mlines.Line2D([], [], color="#888888", linestyle="--", label=f"SMA{trend_sma_period}"), mlines.Line2D([], [], color="#d62728", linestyle=":", label="BB Lower"), mlines.Line2D([], [], color="#9467bd", label=f"MA{MA_FAST_PERIOD}"), mlines.Line2D([], [], color="#8c564b", label=f"MA{MA_SLOW_PERIOD}"), ] if has_oversold: legend_elems.append(mlines.Line2D( [], [], color="orange", marker="v", linestyle="None", markersize=8, label="RSI Oversold")) if has_macd_cross: legend_elems.append(mlines.Line2D( [], [], color="blue", marker="^", linestyle="None", markersize=8, label="MACD Cross ↑")) if has_ma_cross: legend_elems.append(mlines.Line2D( [], [], color="gold", marker="*", linestyle="None", markersize=10, label=f"Golden Cross MA{MA_FAST_PERIOD}/{MA_SLOW_PERIOD}")) axes[0].legend(handles=legend_elems, loc="upper left", fontsize=7.5, ncol=2) # Info harga terakhir + kapan data ini diambil (jam candle terakhir vs jam script jalan) last_candle_time = plot_df.index[-1] last_close = plot_df["Close"].iloc[-1] fetch_time = datetime.now(ZoneInfo(MARKET_TZ)) # Check market status market_open, market_msg = is_market_open() # Check apakah data candle terakhir adalah hari ini today = fetch_time.date() last_candle_date = last_candle_time.date() if hasattr(last_candle_time, 'date') else last_candle_time data_stale = last_candle_date < today is_intraday = (plot_df.index.hour != 0).any() or (plot_df.index.minute != 0).any() time_fmt = "%d %b %Y %H:%M" if is_intraday else "%d %b %Y" # Market status indicator market_status_text = "[MARKET BUKA]" if market_open else "[MARKET TUTUP]" if data_stale and not market_open: market_status_text += f" (Data: {last_candle_time.strftime(time_fmt)})" # Fibonacci info fib_info = "" if result.get("fibonacci") and result["fibonacci"].get("levels"): fib = result["fibonacci"] current_close = result["close"] # Use result close price # Cek posisi harga relatif terhadap Fibonacci levels fib_levels = fib["levels"] near_level = None min_dist = float('inf') for level_name, level_price in fib_levels.items(): dist = abs(current_close - level_price) / current_close * 100 if dist < min_dist: min_dist = dist near_level = level_name if near_level and min_dist < 5: # Kalau dalam 5% dari suatu level fib_info = f"\nNear Fibo {near_level} ({min_dist:.1f}% away)" info_text = ( f"{market_status_text}{fib_info}\n" f"Harga Terakhir: {last_close:,.0f}\n" f"Candle Terakhir: {last_candle_time.strftime(time_fmt)}\n" f"Analisa: {fetch_time.strftime('%d %b %Y %H:%M')} WIB" ) axes[0].text( 0.99, 0.02, info_text, transform=axes[0].transAxes, fontsize=8, ha="right", va="bottom", bbox=dict(boxstyle="round", facecolor="#0d0d0d", alpha=0.85, edgecolor="#333333", pad=5) ) filename = f"{ticker}_{datetime.now().strftime('%Y%m%d_%H%M')}.png" filepath = os.path.join(CHARTS_DIR, filename) fig.savefig(filepath, dpi=130, bbox_inches="tight") import matplotlib.pyplot as plt plt.close(fig) return filepath except Exception as e: print(f"[WARN] Gagal bikin chart {ticker}: {e}") return None def chunk(lst, size): for i in range(0, len(lst), size): yield lst[i:i + size] def screen_batch(tickers: list[str], interval: str, period: str, trend_sma: int, free_float_map: dict) -> list[dict]: results = [] try: data = yf.download(tickers, period=period, interval=interval, group_by="ticker", threads=True, progress=False, auto_adjust=True) except Exception as e: print(f"[WARN] Gagal download batch: {e}") return results for ticker in tickers: try: df = data[ticker] if len(tickers) > 1 else data df = df.dropna(subset=["Close"]).copy() clean_ticker = ticker.replace(".JK", "") result = analyze(df, trend_sma, ticker=clean_ticker, free_float_map=free_float_map) if result: result["ticker"] = clean_ticker result["chart_path"] = generate_chart(clean_ticker, df, result, trend_sma) results.append(result) except Exception: continue return results def screen_all(tickers: list[str], timeframe: str, free_float_map: dict) -> list[dict]: cfg = TIMEFRAME_CONFIG[timeframe] all_results = [] batches = list(chunk(tickers, BATCH_SIZE)) for i, batch in enumerate(batches, 1): print(f"[INFO] Batch {i}/{len(batches)} ({len(batch)} ticker)...") all_results.extend(screen_batch(batch, cfg["interval"], cfg["period"], cfg["trend_sma"], free_float_map)) if i < len(batches): time.sleep(BATCH_DELAY_SEC) return sorted(all_results, key=lambda x: (-x["score"], x["rsi"])) # ============ OUTPUT ============ def format_telegram_message(results: list[dict], timeframe: str, market_status: tuple = None) -> str: # Check market status if market_status is None: market_status = is_market_open() is_open, market_msg = market_status market_indicator = "[MARKET BUKA]" if is_open else "[MARKET TUTUP]" data_note = "" if is_open else f"\n_⚠️ Data analisa: harga penutupan terakhir ({market_msg})_" # Build filter description price_filter = "" if MAX_CLOSE_PRICE is not None or MIN_CLOSE_PRICE is not None: min_p = f"≥Rp{MIN_CLOSE_PRICE}" if MIN_CLOSE_PRICE is not None else "" max_p = f"≤Rp{MAX_CLOSE_PRICE}" if MAX_CLOSE_PRICE is not None else "" price_range = f"{min_p} {max_p}".strip() price_filter = f", harga {price_range}" header = (f"{market_indicator}\n" f"📊 *Swing Screener* (timeframe {timeframe}, min score {MIN_SCORE}/{MAX_POSSIBLE_SCORE})\n" f"_Filter: free float ≥{MIN_FREE_FLOAT_PCT}%, avg value ≥Rp{MIN_AVG_VALUE_TRADED/1e9:.1f}M/hari{price_filter}" f" (M = Miliar Rupiah)_{data_note}\n") if not results: return header + "\nTidak ada kandidat yang lolos semua filter hari ini." lines = [header] for r in results[:30]: reasons_str = ", ".join(r["reasons"]) ff_str = f"{r['free_float_pct']}%" if r.get("free_float_pct") is not None else "N/A" value_str = f"Rp{r['avg_value_traded']/1e9:.2f}M" # Highlight AlphaX pattern jika ada pattern_info = "" if r.get("alphax_pattern"): pat = r["alphax_pattern"] pattern_info = f"\n 📐 *Pattern:* {pat['name']} (confluence: {pat['confluence']}/10)" # Fibonacci info fib_info = "" if r.get("fibonacci") and r["fibonacci"].get("levels"): fib = r["fibonacci"] current_close = r["close"] # Cek posisi harga relatif terhadap Fibonacci levels fib_levels = fib["levels"] near_level = None min_dist = float('inf') for level_name, level_price in fib_levels.items(): dist = abs(current_close - level_price) / current_close * 100 if dist < min_dist: min_dist = dist near_level = level_name if near_level and min_dist < 5: level_price = fib_levels[near_level] fib_info = f"\n 📊 Fibo: {near_level} (Rp{level_price:,.0f})" lines.append( f"• `{r['ticker']}` — Score: *{r['score']}/{MAX_POSSIBLE_SCORE}* | RSI: {r['rsi']} | " f"Close: {r['close']} | FF: {ff_str} | Val: {value_str}\n _{reasons_str}_{pattern_info}{fib_info}" ) if len(results) > 30: lines.append(f"\n...dan {len(results) - 30} saham lainnya.") lines.append( "\n⚠️ Ini hasil screening otomatis, bukan rekomendasi beli. " "Tetap cek fundamental, berita terkini, dan kondisi market secara keseluruhan sebelum entry." ) return "\n".join(lines) def send_telegram_alert(message: str): if "ISI_TOKEN" in TELEGRAM_BOT_TOKEN or "ISI_CHAT_ID" in TELEGRAM_CHAT_ID: print("[INFO] Telegram belum dikonfigurasi. Hasil:\n") print(message) return url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage" try: resp = requests.post(url, json={ "chat_id": TELEGRAM_CHAT_ID, "text": message, "parse_mode": "Markdown" }, timeout=10) resp.raise_for_status() print("[INFO] Alert terkirim ke Telegram.") except Exception as e: print(f"[ERROR] Gagal kirim ke Telegram: {e}") def send_telegram_photo(photo_path: str, caption: str): if "ISI_TOKEN" in TELEGRAM_BOT_TOKEN or "ISI_CHAT_ID" in TELEGRAM_CHAT_ID: return # Telegram belum dikonfigurasi, skip diam-diam (pesan teks sudah cukup kasih tau) url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendPhoto" try: with open(photo_path, "rb") as photo_file: resp = requests.post( url, data={"chat_id": TELEGRAM_CHAT_ID, "caption": caption, "parse_mode": "Markdown"}, files={"photo": photo_file}, timeout=20, ) resp.raise_for_status() except Exception as e: print(f"[ERROR] Gagal kirim chart {photo_path}: {e}") def run_screening(): tickers = load_tickers() free_float_map = load_free_float() # Check market status market_open, market_msg = is_market_open() print(f"[INFO] Market status: {market_msg}") price_info = "" if MAX_CLOSE_PRICE is not None or MIN_CLOSE_PRICE is not None: min_p = f"≥{MIN_CLOSE_PRICE}" if MIN_CLOSE_PRICE is not None else "" max_p = f"≤{MAX_CLOSE_PRICE}" if MAX_CLOSE_PRICE is not None else "" price_info = f", price {min_p} {max_p}".strip() print(f"[INFO] Scanning {len(tickers)} saham, timeframe={TIMEFRAME}, min_score={MIN_SCORE}, " f"free_float_data={len(free_float_map)} saham, min_avg_value={MIN_AVG_VALUE_TRADED:,.0f}{price_info}...") results = screen_all(tickers, TIMEFRAME, free_float_map) message = format_telegram_message(results, TIMEFRAME, (market_open, market_msg)) send_telegram_alert(message) if SEND_CHART_TO_TELEGRAM and results: sent = 0 for r in results: if sent >= MAX_CHARTS_PER_RUN: print(f"[INFO] Sudah kirim {MAX_CHARTS_PER_RUN} chart, sisanya cek folder {CHARTS_DIR}/ lokal.") break if r.get("chart_path"): caption = f"`{r['ticker']}` — Score {r['score']}/{MAX_POSSIBLE_SCORE} | RSI {r['rsi']} | Close {r['close']}" send_telegram_photo(r["chart_path"], caption) sent += 1 print(f"[DONE] {len(results)} saham lolos filter. Chart tersimpan di folder {CHARTS_DIR}/.") def main_once(): """Jalan sekali lalu exit. Cocok dipakai kalau kamu pakai cron.""" if not is_within_schedule(): return run_screening() def seconds_until_next_hour_mark() -> float: """Hitung berapa detik lagi sampai jam bulat berikutnya (contoh: sekarang 13:23 -> tunggu sampai 14:00).""" now = datetime.now(ZoneInfo("Asia/Jakarta")) next_hour = (now.replace(minute=0, second=0, microsecond=0) + pd.Timedelta(hours=1)) return (next_hour - now).total_seconds() def main_loop(): """ Self-loop lokal: jalan terus di terminal, screening otomatis TEPAT tiap jam bulat (13:00, 14:00, 15:00, dst), bukan interval ngambang dari kapan pun kamu start script-nya. Catatan buat local run: - Selama terminal/proses ini kebuka, dia jalan terus. - Kalau laptop sleep/mati atau terminal ditutup, prosesnya berhenti (beda dengan VPS yang nyala 24/7) - tinggal jalankan ulang kalau gitu. """ print(f"[INFO] Self-loop aktif (align ke jam bulat). " f"Jendela jalan: {RUN_HOUR_START}:00-{RUN_HOUR_END}:00 WIB, hari {RUN_DAYS}.") while True: # now = datetime.now(ZoneInfo("Asia/Jakarta")) now = datetime.now(ZoneInfo("Asia/Jakarta")) - timedelta(hours=7) if is_within_schedule(): try: run_screening() except Exception as e: print(f"[ERROR] Screening gagal: {e}") else: print(f"[IDLE] {now.strftime('%Y-%m-%d %H:%M:%S')} WIB - di luar jendela jadwal.") sleep_sec = seconds_until_next_hour_mark() next_run = now + pd.Timedelta(seconds=sleep_sec) print(f"[INFO] Tidur sampai {next_run.strftime('%H:%M:%S')} WIB ({int(sleep_sec)}s)...") time.sleep(sleep_sec) if __name__ == "__main__": if SELF_LOOP: main_loop() else: main_once() # ============ CARA MENJALANKAN (LOCAL) ============ # Mode default sekarang SELF_LOOP=True, jadi cukup: # # python rsi_screener.py # # ...lalu biarkan terminal itu terbuka. Dia akan: # - Screening otomatis tiap jam bulat (13:00, 14:00, dst) # - Cuma benar-benar jalan kalau masuk RUN_HOUR_START-RUN_HOUR_END # dan hari termasuk RUN_DAYS # - Kalau di luar jam/hari itu, dia idle & nunggu jam bulat berikutnya # # Supaya tetap jalan walau terminal ditutup (Mac/Linux): # nohup python3 rsi_screener.py > screener.log 2>&1 & # (cek proses: ps aux | grep rsi_screener | stop: kill ) # # Kalau lebih suka pakai Task Scheduler (Windows) atau cron (Mac/Linux) # daripada self-loop, set SELF_LOOP = False lalu jadwalkan eksekusinya # dari situ sesuai kebutuhan. # =========================================