""" 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 pandas as pd import numpy as np import yfinance as yf import yfinance as tf # Import modular components try: from .ticker import load_tickers, load_free_float from .analyze import analyze from .chart import generate_chart from .telegram import format_telegram_message, send_telegram_alert, send_telegram_photo, save_summary_to_file except ImportError: from ticker import load_tickers, load_free_float from analyze import analyze from chart import generate_chart from telegram import format_telegram_message, send_telegram_alert, send_telegram_photo, save_summary_to_file # ============ 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 = 1_000_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 = 1 SCORE_NEAR_BB_LOWER = 1 SCORE_VOLUME_SPIKE = 1 SCORE_STOCH_RSI_BULLISH = 1 SCORE_MA_GOLDEN_CROSS = 1 # --- 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_NEAR_BB_LOWER + SCORE_VOLUME_SPIKE + SCORE_STOCH_RSI_BULLISH + SCORE_MA_GOLDEN_CROSS) MIN_SCORE = 2 # 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 = 9 MA_SLOW_PERIOD = 21 # --- 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 = 20 # minimal free float (%), dari cache free_float_idx.csv MAX_FREE_FLOAT_PCT = 50 # maksimal free float (%), dari cache free_float_idx.csv MIN_VOL_DAILY = 1_000_000 # minimal volume hari ini (latest candle) - hindari saham illiquid/nyangkut # --- LIKUIDITAS CHECK (Optional) --- CHECK_ADTV = True # True = cek ADTV 30 hari untuk kualitas likuiditas (tambah processing time) ADTV_PERIOD_DAYS = 30 # periode untuk hitung ADTV FREE_FLOAT_FILE = "free_float_idx.csv" SUMMARY_FILE = "src/summary.md" # file untuk simpan hasil analisa MAX_CLOSE_PRICE = None # maksimal harga close (Rupiah), kalau None = tanpa batas MIN_CLOSE_PRICE = None # 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 = True # --- 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 = "src/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 # --- ENTRY & STOP LOSS --- SR_MIN_TOUCH_USABLE = 2 # minimal touch untuk support/resistance zone dianggap valid FIB_LOOKBACK_BARS = 60 # lookback untuk Fibonacci calculation FIB_SCALP_LEVEL = 0.382 # Fibonacci level untuk scalping (38.2%) FIB_LONG_LEVELS = [0.5, 0.618] # Fibonacci levels untuk long term (50%, 61.8%) SL_SCALP_LOOKBACK_BARS = 5 # lookback untuk stop loss scalping (cari swing low terdekat) SL_SCALP_MAX_PCT = 2.0 # maximal distance stop loss scalping (2% dari entry) SL_LONG_ZONE_BUFFER_PCT = 0.5 # buffer di bawah demand zone untuk stop loss long term (0.5%) SL_LONG_PCT_RANGE = [3, 8] # normal range stop loss long term (3-8% dari entry) # ======================================= 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 # Import functions from modules have been done above # The following functions are now in their respective modules: # - load_tickers(), load_free_float() -> ticker.py # - calc_rsi(), calc_macd(), calc_bollinger(), calc_fibonacci_levels(), calc_stochastic_rsi() -> indicators.py # - analyze() -> analyze.py # - generate_chart() -> chart.py # - format_telegram_message(), send_telegram_alert(), send_telegram_photo() -> telegram.py 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, market_status: tuple = None) -> list[dict]: """ Screen a batch of tickers. Args: tickers: List of ticker symbols dengan .JK suffix interval: Data interval period: Data period trend_sma: Trend SMA period free_float_map: Dict mapping ticker ke free float percentage market_status: Tuple of (is_open, message) for market status Returns: List of analysis results """ # Get market status if market_status is None: market_status = is_market_open() # Build configuration dict untuk modules config = { # Scoring thresholds 'RSI_THRESHOLD': RSI_THRESHOLD, 'BB_PROXIMITY': BB_PROXIMITY, 'VOLUME_SPIKE_MULT': VOLUME_SPIKE_MULT, 'STOCH_OVERSOLD': STOCH_OVERSOLD, 'MA_FAST_PERIOD': MA_FAST_PERIOD, 'MA_SLOW_PERIOD': MA_SLOW_PERIOD, 'MIN_VOLUME': MIN_VOLUME, 'MIN_VOL_DAILY': MIN_VOL_DAILY, 'MIN_SCORE': MIN_SCORE, # Scoring weights 'SCORE_RSI_OVERSOLD': SCORE_RSI_OVERSOLD, 'SCORE_MACD_CROSSOVER': SCORE_MACD_CROSSOVER, 'SCORE_NEAR_BB_LOWER': SCORE_NEAR_BB_LOWER, 'SCORE_VOLUME_SPIKE': SCORE_VOLUME_SPIKE, 'SCORE_STOCH_RSI_BULLISH': SCORE_STOCH_RSI_BULLISH, 'SCORE_MA_GOLDEN_CROSS': SCORE_MA_GOLDEN_CROSS, # Hard filters 'MIN_AVG_VALUE_TRADED': MIN_AVG_VALUE_TRADED, 'MIN_FREE_FLOAT_PCT': MIN_FREE_FLOAT_PCT, 'MAX_FREE_FLOAT_PCT': MAX_FREE_FLOAT_PCT, 'REQUIRE_FREE_FLOAT_DATA': REQUIRE_FREE_FLOAT_DATA, 'MAX_CLOSE_PRICE': MAX_CLOSE_PRICE, 'MIN_CLOSE_PRICE': MIN_CLOSE_PRICE, # Fibonacci configuration 'SHOW_FIBONACCI': SHOW_FIBONACCI, 'FIBONACCI_PERIOD': FIBONACCI_PERIOD, 'FIBONACCI_LEVELS': FIBONACCI_LEVELS, # Stochastic RSI configuration 'STOCH_RSI_PERIOD': STOCH_RSI_PERIOD, 'STOCH_SMOOTH_K': STOCH_SMOOTH_K, 'STOCH_SMOOTH_D': STOCH_SMOOTH_D, # Chart configuration 'CHARTS_DIR': CHARTS_DIR, 'CHART_LOOKBACK_CANDLES': CHART_LOOKBACK_CANDLES, 'MAX_POSSIBLE_SCORE': MAX_POSSIBLE_SCORE, 'MARKET_TZ': MARKET_TZ, 'FIBONACCI_COLORS': FIBONACCI_COLORS, # Market status for chart generation 'market_open': market_status[0], 'market_msg': market_status[1], } 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: # Try accessing ticker data - if KeyError, ticker is delisted if len(tickers) > 1: try: df = data[ticker] except KeyError: # Ticker not found in download results = delisted clean_ticker = ticker.replace(".JK", "") print(f"[WARN] {ticker} appears to be delisted (KeyError - no data returned)") remove_delisted_ticker(clean_ticker) continue else: df = data # Check if df is valid DataFrame and has data if not isinstance(df, pd.DataFrame) or df.empty or len(df) == 0: clean_ticker = ticker.replace(".JK", "") print(f"[WARN] {ticker} has no valid data (empty DataFrame, likely delisted)") remove_delisted_ticker(clean_ticker) continue # Check if DataFrame has all required columns required_cols = ["Open", "High", "Low", "Close", "Volume"] if not all(col in df.columns for col in required_cols): clean_ticker = ticker.replace(".JK", "") print(f"[WARN] {ticker} missing required columns (likely delisted)") remove_delisted_ticker(clean_ticker) continue df = df.dropna(subset=["Close"]).copy() # Check again after dropna - if still empty, delisted if df.empty or len(df) == 0: clean_ticker = ticker.replace(".JK", "") print(f"[WARN] {ticker} became empty after dropna (no valid price data)") remove_delisted_ticker(clean_ticker) continue clean_ticker = ticker.replace(".JK", "") # Call analyze function with config result = analyze(df, trend_sma, ticker=clean_ticker, free_float_map=free_float_map, config=config) if result: result["ticker"] = clean_ticker # Add entry & stop loss suggestions entry_suggestion = suggest_entry(df) if entry_suggestion: # Hanya tambahkan jika ada setup yang aman result["entry_suggestion"] = entry_suggestion sl_suggestion = suggest_stop_loss(df, mode="long", entry_price=result.get("close")) if sl_suggestion: result["stop_loss_suggestion"] = sl_suggestion # Cek likuiditas tambahan (optional) if CHECK_ADTV: adtv_info = hitung_adtv_dan_status(ticker, periode_hari=ADTV_PERIOD_DAYS) result["adtv_info"] = adtv_info # Call generate_chart with config result["chart_path"] = generate_chart(clean_ticker, df, result, trend_sma, config) results.append(result) except Exception as e: error_msg = str(e).lower() # Check for delisted errors if "possibly delisted" in error_msg or "no data found" in error_msg or "symbol may be delisted" in error_msg or "keyerror" in error_msg: clean_ticker = ticker.replace(".JK", "") print(f"[WARN] {ticker} appears to be delisted: {e}") remove_delisted_ticker(clean_ticker) else: print(f"[WARN] Error processing {ticker}: {e}") continue return results def remove_delisted_ticker(ticker: str, free_float_file: str = FREE_FLOAT_FILE, ticker_file: str = TICKER_FILE): """ Hapus ticker yang sudah delisted dari free_float_idx.csv dan tickers_idx.txt. Args: ticker: Ticker symbol (tanpa .JK suffix) free_float_file: Path ke free float CSV file ticker_file: Path ke ticker list file """ import os # Remove from free_float_idx.csv if os.path.exists(free_float_file): try: with open(free_float_file, 'r', encoding='utf-8') as f: lines = f.readlines() # Filter out the delisted ticker new_lines = [] for line in lines: if line.startswith(f"{ticker},"): print(f"[INFO] Menghapus {ticker} dari {free_float_file}") continue new_lines.append(line) with open(free_float_file, 'w', encoding='utf-8') as f: f.writelines(new_lines) except Exception as e: print(f"[ERROR] Gagal menghapus dari {free_float_file}: {e}") # Remove from tickers_idx.txt if os.path.exists(ticker_file): try: with open(ticker_file, 'r', encoding='utf-8') as f: lines = f.readlines() # Filter out the delisted ticker new_lines = [] for line in lines: if line.strip() == ticker: print(f"[INFO] Menghapus {ticker} dari {ticker_file}") continue new_lines.append(line) with open(ticker_file, 'w', encoding='utf-8') as f: f.writelines(new_lines) except Exception as e: print(f"[ERROR] Gagal menghapus dari {ticker_file}: {e}") # ============ ENTRY & STOP LOSS ============ def estimate_tick_size(price: float) -> float: """ Estimate tick size berdasarkan harga saham IDX. Args: price: Harga saham Returns: Estimasi tick size """ if price < 100: return 1.0 elif price < 500: return 2.0 elif price < 2000: return 5.0 elif price < 5000: return 10.0 else: return 25.0 def hitung_adtv_dan_status(ticker_symbol, periode_hari=30): """ Menghitung ADTV saham dan menentukan status likuiditasnya. Args: ticker_symbol: Symbol ticker (contoh: BBCA.JK) periode_hari: Jumlah hari untuk hitung rata-rata (default 30) Returns: Dict dengan ADTV dan status likuiditas """ try: # Mengunduh data historis (ambil cadangan hari lebih banyak untuk mengantisipasi hari libur) saham = tf.Ticker(ticker_symbol) df = saham.history(period=f"{periode_hari * 2}d") if df.empty: return {"Ticker": ticker_symbol, "Status": "Data Tidak Ditemukan"} # Ambil data N hari perdagangan terakhir yang valid df = df.tail(periode_hari) # 1. Hitung Turnover Harian (Volume Lembar * Harga Penutupan) df['Daily_Turnover'] = df['Volume'] * df['Close'] # 2. Hitung Rata-rata Nilai Transaksi Harian (ADTV) adtv = df['Daily_Turnover'].mean() # 3. Ambil data harga terakhir untuk referensi harga_terakhir = df['Close'].iloc[-1] rata_volume_lembar = df['Volume'].mean() # 4. Tentukan Kategori Likuiditas berdasarkan standar ADTV (dalam Rupiah) if adtv > 20_000_000_000: status = "Sangat Likuid (>20M)" elif adtv >= 5_000_000_000: status = "Likuiditas Baik (5M-20M)" elif adtv >= 1_000_000_000: status = "Likuiditas Sedang (1M-5M)" else: status = "Sangat Tidak Likuid (<1M)" return { "Ticker": ticker_symbol, "Harga Terakhir": f"Rp{harga_terakhir:,.0f}", "Rata-rata Volume (Lembar)": f"{rata_volume_lembar:,.0f}", "ADTV": f"Rp{adtv:,.0f}", "Status Likuiditas": status } except Exception as e: return {"Ticker": ticker_symbol, "Status": f"Error: {str(e)}"} def suggest_entry(df: pd.DataFrame): """ Sarankan entry price berdasarkan support zone, MA pullback, atau Fibonacci. HANYA kasih entry jika kondisi AMAN. Return None jika kondisi berbahaya. Args: df: DataFrame dengan OHLCV dan indicators Returns: Dict dengan entry suggestion atau None jika kondisi tidak aman untuk entry """ if len(df) < 3: return None close = float(df["Close"].iloc[-1]) tick = estimate_tick_size(close) # --- SAFETY CHECKS: Jangan kasih entry jika kondisi berbahaya --- stoch_k = df["stoch_k"].iloc[-1] if "stoch_k" in df.columns else None bb_upper = df["bb_upper"].iloc[-1] if "bb_upper" in df.columns else None is_green = bool(df["Close"].iloc[-1] > df["Open"].iloc[-1]) # 1. Stochastic RSI overbought (>80) = bahaya entry if pd.notna(stoch_k) and stoch_k > 80: return None # Terlalu panas, tunggu cooldown ke area oversold # 2. Harga dekat upper BB/resistance = bahaya entry if pd.notna(bb_upper) and close >= float(bb_upper) * 0.98: return None # Harga di area resistance, tunggu pullback # 3. Bar merah/merosot = tidak aman entry if not is_green: return None # Sedang selling pressure, tunggu konfirmasi bullish # --- AMAN UNTUK ENTRY: Cari setup terbaik --- # Prioritas 1: Support zone teruji (dekat lower BB atau area support kuat) bb_lower = df["bb_lower"].iloc[-1] if "bb_lower" in df.columns else None if pd.notna(bb_lower) and close <= float(bb_lower) * 1.02: entry_price = close + tick return { "priority": 1, "method": "support_bb_lower", "entry_price": round(entry_price, 2), "bb_lower": round(float(bb_lower), 2) } # Prioritas 2: MA pullback (MA9/21 untuk long term) ma_fast = df["ma_fast"].iloc[-1] if "ma_fast" in df.columns else None ma_slow = df["sma_trend"].iloc[-1] if "sma_trend" in df.columns else None rsi_now = df["rsi"].iloc[-1] if "rsi" in df.columns else None rsi_prev = df["rsi"].iloc[-2] if "rsi" in df.columns and len(df) > 1 else None rsi_rising = pd.notna(rsi_now) and pd.notna(rsi_prev) and rsi_now > rsi_prev # Use MA fast/slow untuk pullback check ma_value = ma_fast if pd.notna(ma_fast) else ma_slow ma_name = "MA9" if ma_value == ma_fast or ma_fast is not None else "SMA50" if pd.notna(ma_value): near_ma = abs(close - float(ma_value)) / float(ma_value) * 100 <= 1.0 if near_ma and rsi_rising: return { "priority": 2, "method": "ma_pullback", "entry_price": round(close, 2), "ma_value": round(float(ma_value), 2), "ma_name": ma_name } # Prioritas 3: Fibonacci retracement if df.get("fibonacci") and df["fibonacci"].get("levels"): fib_levels = df["fibonacci"]["levels"] close_val = close # Cek dekat fibonacci level manapun for level_name, level_price in fib_levels.items(): if abs(close_val - level_price) / level_price * 100 <= 1.5: return { "priority": 3, "method": f"fibonacci_{level_name}", "entry_price": round(close, 2), "fib_level": round(level_price, 2) } # Tidak ada setup yang aman -> Jangan kasih entry (lebih baik skip) return None def suggest_stop_loss(df: pd.DataFrame, mode: str = "long", entry_price: float = None): """ Sarankan stop loss berdasarkan swing low atau support zone. Args: df: DataFrame dengan OHLCV dan indicators mode: "scalp" atau "long" entry_price: Entry price (kalau None, pakai close price terakhir) Returns: Dict dengan stop loss suggestion """ if len(df) < 5: return None close_price = entry_price if entry_price else float(df["Close"].iloc[-1]) tick = estimate_tick_size(close_price) if mode == "scalp": # Cari swing low terdekat (5 candle terakhir) lookback = df.tail(SL_SCALP_LOOKBACK_BARS) swing_low_idx = lookback["Low"].idxmin() wick_low = float(lookback.loc[swing_low_idx, "Low"]) sl = wick_low - tick dist_pct = (close_price - sl) / close_price * 100 too_far = dist_pct > SL_SCALP_MAX_PCT return { "sl_price": round(sl, 2), "basis": "wick_low_swing", "distance_pct": round(dist_pct, 2), "max_allowed_pct": SL_SCALP_MAX_PCT, "too_far": too_far, "note": "Jarak SL > 2%, tunggu pullback lebih dekat dulu" if too_far else None } else: # long # Gunakan BB lower sebagai referensi support bb_lower = df["bb_lower"].iloc[-1] if "bb_lower" in df.columns else None if pd.notna(bb_lower): sl = float(bb_lower) * (1 - SL_LONG_ZONE_BUFFER_PCT / 100) dist_pct = (close_price - sl) / close_price * 100 lo, hi = SL_LONG_PCT_RANGE out_of_range = not (lo <= dist_pct <= hi) return { "sl_price": round(sl, 2), "basis": "bb_lower_support", "distance_pct": round(dist_pct, 2), "normal_range_pct": SL_LONG_PCT_RANGE, "out_of_normal_range": out_of_range, "bb_lower": round(float(bb_lower), 2) } # Fallback: gunakan swing low dari lookback lookback = df.tail(20) swing_low_idx = lookback["Low"].idxmin() wick_low = float(lookback.loc[swing_low_idx, "Low"]) sl = wick_low - tick dist_pct = (close_price - sl) / close_price * 100 return { "sl_price": round(sl, 2), "basis": "swing_low_fallback", "distance_pct": round(dist_pct, 2), "note": "Gunakan swing low sebagai fallback" } def screen_all(tickers: list[str], timeframe: str, free_float_map: dict) -> list[dict]: """ Screen all tickers dengan batching. Args: tickers: List of ticker symbols tanpa .JK suffix timeframe: Timeframe configuration key free_float_map: Dict mapping ticker ke free float percentage Returns: List of analysis results sorted by score dan RSI """ cfg = TIMEFRAME_CONFIG[timeframe] # Get market status once for all batches market_status = is_market_open() 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, market_status)) if i < len(batches): time.sleep(BATCH_DELAY_SEC) return sorted(all_results, key=lambda x: (-x["score"], x["rsi"])) def run_screening(single_ticker: str = None): """ Main screening function - coordinates all modules. Args: single_ticker: If provided, only analyze this specific ticker (e.g., "AHAP.JK") """ if single_ticker: # Single ticker mode - clean and validate single_ticker = single_ticker.strip() # Remove $ prefix if present if single_ticker.startswith("$"): single_ticker = single_ticker[1:] # Ensure .JK suffix if not single_ticker.endswith(".JK"): single_ticker = f"{single_ticker}.JK" tickers = [single_ticker] print(f"[INFO] Single ticker mode: {single_ticker}") else: # Normal mode - load all tickers tickers = load_tickers(TICKER_FILE) free_float_map = load_free_float(FREE_FLOAT_FILE) # 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) # Build config dict for Telegram module config = { 'MIN_SCORE': MIN_SCORE, 'MAX_POSSIBLE_SCORE': MAX_POSSIBLE_SCORE, 'MIN_FREE_FLOAT_PCT': MIN_FREE_FLOAT_PCT, 'MIN_AVG_VALUE_TRADED': MIN_AVG_VALUE_TRADED, 'MAX_CLOSE_PRICE': MAX_CLOSE_PRICE, 'MIN_CLOSE_PRICE': MIN_CLOSE_PRICE, } message = format_telegram_message(results, TIMEFRAME, config, (market_open, market_msg)) save_summary_to_file(message, SUMMARY_FILE) send_telegram_alert(message, TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID) 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, TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID) sent += 1 print(f"[DONE] {len(results)} saham lolos filter. Chart tersimpan di folder {CHARTS_DIR}/.") def main_once(single_ticker: str = None): """Jalan sekali lalu exit. Cocok dipakai kalau kamu pakai cron.""" if single_ticker: # Single ticker mode - skip schedule check run_screening(single_ticker) elif not is_within_schedule(): return else: 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(single_ticker: str = None): """ 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. Args: single_ticker: If provided, run single ticker mode once and exit (no loop) 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. """ if single_ticker: # Single ticker mode - run once and exit (no loop) print(f"[INFO] Single ticker mode: {single_ticker}") run_screening(single_ticker) return 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__": import sys # Parse command line arguments single_ticker = None if len(sys.argv) > 1: single_ticker = sys.argv[1] if SELF_LOOP: main_loop(single_ticker) else: main_once(single_ticker) # ============ 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. # =========================================