""" Core Analysis Function """ import sys import os from datetime import datetime, timedelta from zoneinfo import ZoneInfo import pandas as pd import numpy as np import yfinance as yf # Import indicator functions try: from .indicators import calc_rsi, calc_macd, calc_bollinger, calc_fibonacci_levels, calc_stochastic_rsi except ImportError: from indicators import calc_rsi, calc_macd, calc_bollinger, calc_fibonacci_levels, calc_stochastic_rsi def analyze(df: pd.DataFrame, trend_sma_period: int, ticker: str = None, free_float_map: dict = None, config: dict = None) -> dict | None: """ Analyze stock data dan return result jika memenuhi kriteria. Args: df: DataFrame dengan OHLCV data trend_sma_period: Period untuk SMA trend filter ticker: Ticker symbol (tanpa .JK suffix) free_float_map: Dict mapping ticker ke free float percentage config: Configuration dict dengan semua parameter Returns: Dict dengan analysis result atau None jika tidak memenuhi kriteria """ if config is None: config = {} # Extract configuration values rsi_threshold = config.get('RSI_THRESHOLD', 30) bb_proximity = config.get('BB_PROXIMITY', 1.02) volume_spike_mult = config.get('VOLUME_SPIKE_MULT', 1.5) stoch_oversold = config.get('STOCH_OVERSOLD', 20) ma_fast_period = config.get('MA_FAST_PERIOD', 20) ma_slow_period = config.get('MA_SLOW_PERIOD', 50) min_volume = config.get('MIN_VOLUME', 100_000) min_avg_value_traded = config.get('MIN_AVG_VALUE_TRADED', 1_000_000_000) min_free_float_pct = config.get('MIN_FREE_FLOAT_PCT', 25) max_free_float_pct = config.get('MAX_FREE_FLOAT_PCT', None) require_free_float_data = config.get('REQUIRE_FREE_FLOAT_DATA', False) max_close_price = config.get('MAX_CLOSE_PRICE', None) min_close_price = config.get('MIN_CLOSE_PRICE', None) min_score = config.get('MIN_SCORE', 5) # Scoring weights score_rsi_oversold = config.get('SCORE_RSI_OVERSOLD', 2) score_macd_crossover = config.get('SCORE_MACD_CROSSOVER', 2) score_above_trend_sma = config.get('SCORE_ABOVE_TREND_SMA', 2) score_near_bb_lower = config.get('SCORE_NEAR_BB_LOWER', 1) score_volume_spike = config.get('SCORE_VOLUME_SPIKE', 1) score_stoch_rsi_bullish = config.get('SCORE_STOCH_RSI_BULLISH', 2) score_ma_golden_cross = config.get('SCORE_MA_GOLDEN_CROSS', 1) # Fibonacci configuration show_fibonacci = config.get('SHOW_FIBONACCI', True) fibonacci_period = config.get('FIBONACCI_PERIOD', 60) fibonacci_levels = config.get('FIBONACCI_LEVELS', [0, 0.236, 0.382, 0.5, 0.618, 0.786, 1]) # Stochastic RSI configuration stoch_rsi_period = config.get('STOCH_RSI_PERIOD', 14) stoch_smooth_k = config.get('STOCH_SMOOTH_K', 3) stoch_smooth_d = config.get('STOCH_SMOOTH_D', 3) if len(df) < trend_sma_period + 5: return None # data belum cukup untuk trend SMA panjang # Set ticker attribute 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"], stoch_rsi_period, stoch_smooth_k, stoch_smooth_d) 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, fibonacci_levels) 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: MA Bullish Trend (MA_fast > MA_slow) --- # Filter out saham yang sedang death cross/bearish (MA_fast < MA_slow) ma_fast_val = latest["ma_fast"].item() if hasattr(latest["ma_fast"], 'item') else latest["ma_fast"] ma_slow_val = latest["ma_slow"].item() if hasattr(latest["ma_slow"], 'item') else latest["ma_slow"] if ma_fast_val < ma_slow_val: return None # Skip saham dengan death cross / downtrend # --- HARD FILTER: harga close (min/max price) --- close_val = latest["Close"].item() if hasattr(latest["Close"], 'item') else latest["Close"] if max_close_price is not None and close_val > max_close_price: return None if min_close_price is not None and close_val < min_close_price: return None # --- HARD FILTER: likuiditas (rata-rata nilai transaksi 20 hari) --- avg_val = latest["avg_value_20"].item() if hasattr(latest["avg_value_20"], 'item') else latest["avg_value_20"] if avg_val < 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 not None and max_free_float_pct is not None and free_float_pct > max_free_float_pct: return None # free float terlalu tinggi (saham terlalu likuid) -> tolak if free_float_pct is None and require_free_float_data: return None # data gak ada & wajib ada -> tolak score = 0 reasons = [] # Helper function untuk convert pandas Series ke scalar def to_scalar(val): return val.item() if hasattr(val, 'item') else val # 1. RSI oversold rsi_val = to_scalar(latest["rsi"]) if rsi_val < rsi_threshold: score += score_rsi_oversold reasons.append("RSI oversold") # 2. MACD bullish crossover (baru cross di candle terakhir) macd_crossed_up = (to_scalar(prev["macd"]) <= to_scalar(prev["macd_signal"])) and \ (to_scalar(latest["macd"]) > to_scalar(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 (REMOVED - breakout dari bawah SMA juga valid) # Filter ini dihapus karena breakout biasanya terjadi ketika close < SMA200 # if close_val > to_scalar(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 close_val <= to_scalar(latest["bb_lower"]) * bb_proximity: score += score_near_bb_lower reasons.append("Dekat lower Bollinger Band") # 5. Volume spike vs rata-rata 20 hari (CHECK 5 CANDLE TERAKHIR) # Logic baru: Cari spike BUY terbesar dalam 5 candle terakhir # TAPI kalau setelahnya ada spike SELL yang lebih besar → skip volume_lookback = 5 recent_5 = df.tail(volume_lookback) # Cari volume BUY (candle hijau) terbesar dalam 5 candle terakhir max_buy_vol = 0 max_buy_idx = None max_buy_ratio = 0 for i, (idx, row) in enumerate(recent_5.iterrows()): row_close = to_scalar(row["Close"]) row_open = to_scalar(row["Open"]) is_green = row_close > row_open vol = to_scalar(row["Volume"]) vol_avg = to_scalar(row["vol_sma20"]) if not pd.isna(vol_avg) and vol_avg > 0: vol_ratio = vol / vol_avg if is_green and vol_ratio > max_buy_ratio: max_buy_ratio = vol_ratio max_buy_vol = vol max_buy_idx = i # Cek apakah setelah spike BUY ada spike SELL yang lebih besar has_bigger_sell_after = False if max_buy_idx is not None and max_buy_idx + 1 < len(recent_5): # Cek candle setelah spike buy terbesar candles_after_buy = recent_5.iloc[max_buy_idx + 1:] for idx, row in candles_after_buy.iterrows(): row_close = to_scalar(row["Close"]) row_open = to_scalar(row["Open"]) is_red = row_close <= row_open vol = to_scalar(row["Volume"]) vol_avg = to_scalar(row["vol_sma20"]) if not pd.isna(vol_avg) and vol_avg > 0: vol_ratio = vol / vol_avg if is_red and vol_ratio > max_buy_ratio: has_bigger_sell_after = True break # Berikan poin kalau ada spike buy dan tidak ada sell yang lebih besar setelahnya if max_buy_idx is not None and max_buy_ratio >= 0.8 and not has_bigger_sell_after: score += score_volume_spike reasons.append(f"Volume spike ({max_buy_ratio:.1f}x, buy)") elif max_buy_idx is not None and max_buy_ratio >= 0.8 and has_bigger_sell_after: # Ada spike buy tapi diikuti sell yang lebih besar - distribusi pass elif max_buy_idx is None: # Tidak ada spike buy dalam 5 candle terakhir pass # 6. Stochastic RSI oversold + baru cross naik (%K cross di atas %D) stoch_oversold = to_scalar(latest["stoch_k"]) < stoch_oversold stoch_crossed_up = (to_scalar(prev["stoch_k"]) <= to_scalar(prev["stoch_d"])) and \ (to_scalar(latest["stoch_k"]) > to_scalar(latest["stoch_d"])) if stoch_oversold and stoch_crossed_up: score += score_stoch_rsi_bullish reasons.append("Stochastic RSI oversold + cross naik") # 6b. Price Action: Check selling pressure (bearish divergence) # Cek 3 candle terakhir - kalau semua merah, heavy selling, skip recent_3 = df.tail(3) red_candle_count = ((recent_3["Close"] <= recent_3["Open"]).sum()).item() if hasattr((recent_3["Close"] <= recent_3["Open"]).sum(), 'item') else (recent_3["Close"] <= recent_3["Open"]).sum() if red_candle_count >= 3: return None # Skip - 3 candle merah berturut-turut = heavy selling pressure # 7. MA Golden Cross (MA cepat baru potong ke atas MA lambat) # Cukup MA_fast > MA_slow (bullish), tidak peduli kapan cross terjadi ma_bullish_now = ma_fast_val > ma_slow_val if ma_bullish_now: score += score_ma_golden_cross reasons.append(f"MA{ma_fast_period} > MA{ma_slow_period} (Bullish)") if to_scalar(latest["Volume"]) < min_volume: return None # --- HARD FILTER: volume hari ini (minimal MIN_VOL_DAILY) --- min_vol_daily = config.get('MIN_VOL_DAILY', 1_000_000) if to_scalar(latest["Volume"]) < min_vol_daily: return None # Volume hari ini terlalu kecil, rawan nyangkut # Wajib: Minimal harus punya Volume spike + MA golden cross has_volume_spike = max_buy_idx is not None and max_buy_ratio >= 0.8 and not has_bigger_sell_after has_ma_golden_cross = ma_bullish_now if not (has_volume_spike and has_ma_golden_cross): return None # Wajib punya kedua-duanya # Status label berdasarkan score # Score 2 → PANTAU (hanya 2 indikator wajib) # Score 3-4 → MENARIK (ada bonus indikator) # Score 5-7 → EKSEKUSI (banyak konfirmasi) if score == 2: status = "PANTAU" elif score <= 4: status = "MENARIK" elif score <= 7: status = "EKSEKUSI" else: status = "Unknown" if score < min_score: return None # Build indicator breakdown untuk display di Telegram indicator_breakdown = { "rsi_oversold": rsi_val < rsi_threshold, "macd_bullish_cross": macd_crossed_up, "near_bb_lower": close_val <= to_scalar(latest["bb_lower"]) * bb_proximity, "volume_spike": has_volume_spike, "stoch_rsi_bullish": (to_scalar(latest["stoch_k"]) < stoch_oversold) and stoch_crossed_up, "ma_golden_cross": has_ma_golden_cross, } return { "score": score, "status": status, # EKSEKUSI / PANTAU / MENARIK "rsi": round(to_scalar(latest["rsi"]), 2), "stoch_k": round(to_scalar(latest["stoch_k"]), 2), "close": round(close_val, 2), "volume": int(to_scalar(latest["Volume"])), "avg_value_traded": to_scalar(latest["avg_value_20"]), "free_float_pct": free_float_pct, "reasons": reasons, "fibonacci": fib_levels if fib_levels else None, # Fibonacci levels "indicator_breakdown": indicator_breakdown, # Breakdown ✅/❌ tiap indikator }