""" Telegram and Output Formatting Functions """ import requests def format_indicator_breakdown(breakdown: dict) -> str: """ Format indicator breakdown dengan ✅/❌ untuk setiap indikator. Args: breakdown: Dict dengan status setiap indikator (True/False) Returns: Formatted string dengan emoji checklist """ labels = { "rsi_oversold": "RSI Oversold", "macd_bullish_cross": "MACD Bullish Cross", "near_bb_lower": "Near BB Lower", "volume_spike": "Volume Spike (Buy)", "stoch_rsi_bullish": "Stoch RSI Bullish", "ma_golden_cross": "MA9 > MA21 (Bullish)", } lines = [] for key, label in labels.items(): status = breakdown.get(key, False) emoji = "✅" if status else "❌" lines.append(f" {emoji} {label}") return "\n".join(lines) def format_telegram_message(results: list[dict], timeframe: str, config: dict = None, market_status: tuple = None) -> str: """ Format analysis results untuk Telegram message. Args: results: List of analysis result dicts timeframe: Timeframe yang digunakan config: Configuration dict market_status: Tuple of (is_open, message) untuk market status Returns: Formatted message string """ if config is None: config = {} # Extract configuration values min_score = config.get('MIN_SCORE', 5) max_possible_score = config.get('MAX_POSSIBLE_SCORE', 13) min_free_float_pct = config.get('MIN_FREE_FLOAT_PCT', 25) min_avg_value_traded = config.get('MIN_AVG_VALUE_TRADED', 1_000_000_000) max_close_price = config.get('MAX_CLOSE_PRICE', None) min_close_price = config.get('MIN_CLOSE_PRICE', None) # Use provided market status or default if market_status is None: is_open, market_msg = True, "Market buka" # Default fallback else: 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 i, r in enumerate(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" # Format indicator breakdown breakdown_info = "" if r.get("indicator_breakdown"): breakdown_info = format_indicator_breakdown(r['indicator_breakdown']) # 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" 📊 Fibo: {near_level} (Rp{level_price:,.0f})" # Tambahkan status jika ada status_info = f" | Status: *{r.get('status', 'Unknown')}*" if r.get('status') else "" # Liquidity status dari ADTV liq_str = "" if r.get("adtv_info") and r["adtv_info"].get("Status Likuiditas"): liq_status = r["adtv_info"]["Status Likuiditas"] liq_str = f" | Liq: {liq_status}" # Volume candle terakhir vol_str = "" if r.get("volume"): vol = r["volume"] if vol >= 1_000_000: vol_str = f"{vol/1_000_000:.1f}M" elif vol >= 1_000: vol_str = f"{vol/1_000:.0f}K" else: vol_str = str(vol) # Format: ticker dulu, baru breakdown + fibo di bawahnya ticker_line = f"• `{r['ticker']}` — Score: *{r['score']}/{max_possible_score}*{status_info} | RSI: {r['rsi']} | " \ f"Close: {r['close']} (Vol: {vol_str}) | FF: {ff_str} | Val: {value_str}{liq_str}" # Tambahkan breakdown dan fibo di bawah ticker entry_lines = [ticker_line] if breakdown_info: entry_lines.append(breakdown_info) if fib_info: entry_lines.append(fib_info) # Entry & Stop Loss suggestions trading_info = [] if r.get("entry_suggestion"): entry = r["entry_suggestion"] method_text = entry["method"] # Biarkan method name as-is (fibonacci_0.5, ma_pullback, dll) trading_info.append(f" 📈 Entry: Rp{entry['entry_price']:,.0f} ({method_text})") else: # Tidak ada entry aman - kondisi berbahaya (stoch overbought, dekat resistance, dll) trading_info.append(f" ⚠️ Entry: Tunggu pullback (stoch overbought/resistance)") if r.get("stop_loss_suggestion"): sl = r["stop_loss_suggestion"] # Format SL seperti neurobro: "SL: Rp214 (1.0%)" trading_info.append(f" 🛡️ SL: Rp{sl['sl_price']:,.0f} ({sl['distance_pct']:.1f}%)") # Tambahkan warning jika SL terlalu jauh if sl.get("too_far") or sl.get("out_of_normal_range"): trading_info.append(f" ⚠️ {sl.get('note', 'SL di luar normal range')}") # ADTV info (optional - untuk kualitas likuiditas) if r.get("adtv_info"): adtv = r["adtv_info"] if adtv.get("Status") and "Error" not in adtv["Status"]: status_text = adtv["Status Likuiditas"] trading_info.append(f" 💧 ADTV: {adtv['ADTV']} ({status_text})") if trading_info: entry_lines.append("\n".join(trading_info)) # Gabungkan semua dengan newline entry_text = "\n".join(entry_lines) lines.append(entry_text) # Tambahkan jarak antar ticker (kecuali setelah ticker terakhir) if i < len(results[:30]) - 1 and len(results[:30]) > 1: lines.append("") # Empty line untuk spacing antar ticker 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 save_summary_to_file(message: str, filepath: str = "src/summary.md"): """ Save analysis summary ke markdown file. Args: message: Message text untuk disimpan filepath: Path ke output file """ try: with open(filepath, 'w', encoding='utf-8') as f: f.write(message) print(f"[INFO] Summary disimpan ke {filepath}") except Exception as e: print(f"[ERROR] Gagal simpan summary: {e}") def send_telegram_alert(message: str, bot_token: str, chat_id: str): """ Kirim alert message ke Telegram. Args: message: Message text untuk dikirim bot_token: Telegram bot token chat_id: Telegram chat ID """ if "ISI_TOKEN" in bot_token or "ISI_CHAT_ID" in chat_id: print("[INFO] Telegram belum dikonfigurasi. Hasil:\n") print(message) return url = f"https://api.telegram.org/bot{bot_token}/sendMessage" try: resp = requests.post(url, json={ "chat_id": 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, bot_token: str, chat_id: str): """ Kirim photo ke Telegram. Args: photo_path: Path ke photo file caption: Caption untuk photo bot_token: Telegram bot token chat_id: Telegram chat ID """ if "ISI_TOKEN" in bot_token or "ISI_CHAT_ID" in chat_id: return # Telegram belum dikonfigurasi, skip diam-diam (pesan teks sudah cukup kasih tau) url = f"https://api.telegram.org/bot{bot_token}/sendPhoto" try: with open(photo_path, "rb") as photo_file: resp = requests.post( url, data={"chat_id": 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}")