"""Utilitas bersama semua service NeoBDM. Aturan penting: jam market IDX SELALU WIB (UTC+7) — jangan pernah pakai jam lokal sistem (bisa WITA/WIT). Semua service wajib import datetime dari sini. """ import os import json import time import urllib.request import urllib.parse from datetime import datetime as _datetime, timezone, timedelta WIB = timezone(timedelta(hours=7)) KONGLO_TICKERS = tuple([ "BRPT", "TPIA", "BREN", "CUAN", "PTRO", "BNBR", "BUMI", "BRMS", "ENRG", "DEWA", "RAJA", "RATU", "WIFI", "BUVA", "MINA", "ADRO", "ADMR", "AADI", "EMAS", "MDKA", "TINS", "ANTM", "BIPI", "DSSA", "JARR", "PGUN", "TEBE", "PACK", # Anggota grup Konglo yang sebelumnya belum ikut di PA watchlist. "VKTR", "MDIA", "CDIA", "PSKT", "CBRE", "PADI", "UANG", "MBMA", "BFIN", "ESSA", "GOTO", ]) LQ45_TICKERS = tuple([ "AADI", "ADMR", "ADRO", "AKRA", "AMMN", "AMRT", "ANTM", "ASII", "BBCA", "BBNI", "BBRI", "BBTN", "BMRI", "BRPT", "BUMI", "CPIN", "CUAN", "DEWA", "EMTK", "ESSA", "EXCL", "GOTO", "HRTA", "ICBP", "INCO", "INDF", "INKP", "ISAT", "ITMG", "JPFA", "KLBF", "MAPI", "MBMA", "MDKA", "MEDC", "PGAS", "PGEO", "PTBA", "SCMA", "SMGR", "TLKM", "TOWR", "UNTR", "UNVR", "WIFI", ]) PREMIUM_WATCHLIST_TICKERS = tuple(sorted(set(KONGLO_TICKERS) | set(LQ45_TICKERS))) class datetime(_datetime): """datetime.now() otomatis WIB.""" @classmethod def now(cls, tz=None): return _datetime.now(tz or WIB) IDX_HOLIDAYS_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "idx_holidays.txt") _idx_holidays_cache = {"mtime": None, "dates": set()} def _load_idx_holidays(): """Libur bursa dari idx_holidays.txt (satu tanggal YYYY-MM-DD per baris).""" try: mtime = os.path.getmtime(IDX_HOLIDAYS_FILE) except OSError: return set() if _idx_holidays_cache["mtime"] != mtime: dates = set() with open(IDX_HOLIDAYS_FILE) as f: for line in f: line = line.split("#")[0].strip() if line: dates.add(line) _idx_holidays_cache.update(mtime=mtime, dates=dates) return _idx_holidays_cache["dates"] def market_sessions(now=None): """Sesi bursa IDX hari ini dalam menit-sejak-00:00 WIB. Senin-Kamis: 09:00-12:00 & 13:30-16:00. Jumat: 09:00-11:30 & 14:00-16:00.""" now = now or datetime.now() if now.weekday() == 4: # Jumat return (9 * 60, 11 * 60 + 30), (14 * 60, 16 * 60) return (9 * 60, 12 * 60), (13 * 60 + 30, 16 * 60) def market_total_minutes(now=None): """Total menit perdagangan hari ini (330 Senin-Kamis, 270 Jumat).""" (o1, c1), (o2, c2) = market_sessions(now) return float((c1 - o1) + (c2 - o2)) def market_elapsed_minutes(now=None): """Menit bursa yang sudah berjalan hari ini (jeda istirahat tidak dihitung). Sebelum buka / weekend / libur: kembalikan total menit penuh — volume yang terbaca dari scanner saat itu adalah volume full-day hari sebelumnya.""" now = now or datetime.now() total = market_total_minutes(now) if now.weekday() >= 5 or now.strftime("%Y-%m-%d") in _load_idx_holidays(): return total (o1, c1), (o2, c2) = market_sessions(now) mins = now.hour * 60 + now.minute if mins < o1: return total if mins <= c1: return float(mins - o1 + 1) if mins < o2: return float(c1 - o1) if mins <= c2: return float((c1 - o1) + (mins - o2) + 1) return total def is_market_open(now=None): """Jam bursa IDX (Jumat beda sesi), weekend & libur bursa = tutup.""" now = now or datetime.now() if now.weekday() >= 5: return False if now.strftime("%Y-%m-%d") in _load_idx_holidays(): return False (o1, c1), (o2, c2) = market_sessions(now) mins = now.hour * 60 + now.minute return (o1 <= mins <= c1) or (o2 <= mins <= c2) def round_idx_tick(price): """Bulatkan harga ke tick size IDX.""" if price <= 0: return 0.0 if price < 200: return float(int(price + 0.5)) elif price < 500: return float(int(price / 2 + 0.5) * 2) elif price < 2000: return float(int(price / 5 + 0.5) * 5) elif price < 5000: return float(int(price / 10 + 0.5) * 10) else: return float(int(price / 25 + 0.5) * 25) def get_idx_tick_offset(price, offset): """Geser harga N tick ke atas/bawah sesuai tick size IDX.""" curr = round_idx_tick(price) for _ in range(abs(offset)): if offset < 0: if curr <= 200: tick = 1 elif curr <= 500: tick = 2 elif curr <= 2000: tick = 5 elif curr <= 5000: tick = 10 else: tick = 25 curr -= tick else: if curr < 200: tick = 1 elif curr < 500: tick = 2 elif curr < 2000: tick = 5 elif curr < 5000: tick = 10 else: tick = 25 curr += tick return float(curr) def load_env(env_file="./.env"): env = {} if os.path.exists(env_file): with open(env_file, "r") as f: for line in f: line = line.strip() if line and not line.startswith("#") and "=" in line: k, v = line.split("=", 1) env[k.strip()] = v.strip() return env def load_pa_telegram_config(): """Load the dedicated Telegram destination for Price Action alerts. PA notifications must not silently fall back to the general NeoBDM bot. Environment variables override the project-local ``.env`` values so the same code works in the dashboard process and in a service environment. """ env_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env") env = load_env(env_file) raw_enabled = os.environ.get( "PA_TELEGRAM_ENABLED", env.get("PA_TELEGRAM_ENABLED", "false"), ) return { "enabled": str(raw_enabled).strip().lower() not in { "0", "false", "no", "off" }, "token": os.environ.get("PA_TELEGRAM_TOKEN") or env.get("PA_TELEGRAM_TOKEN", ""), "chat_id": os.environ.get("PA_TELEGRAM_CHAT_ID") or env.get("PA_TELEGRAM_CHAT_ID", ""), } def telegram_enabled(): """Return whether outbound Telegram notifications are enabled.""" raw = os.environ.get("TELEGRAM_ENABLED") if raw is None: env_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env") try: with open(env_file, encoding="utf-8") as stream: for line in stream: key, separator, value = line.partition("=") if separator and key.strip() == "TELEGRAM_ENABLED": raw = value.strip().strip("'\"") break except OSError: pass return str(raw if raw is not None else "true").strip().lower() not in { "0", "false", "no", "off" } def notify_telegram(token, chat_id, message, reply_markup=None, enabled=None, alert_mode=None): if enabled is None: enabled = telegram_enabled() if not enabled: print("[TELEGRAM OFF] Notifikasi dinonaktifkan", flush=True) return False if not token or not chat_id: return False # Notifikasi otomatis default hanya untuk READY BUY. Caller yang memang # meminta lifecycle alert (mis. outcome Swing TP/SL) dapat memilih mode # eksplisit tanpa mengubah kebijakan default seluruh notifier. selected_alert_mode = str( alert_mode or os.environ.get("TRADING_ALERT_MODE", "READY_BUY_ONLY") ).strip().upper() if selected_alert_mode == "READY_BUY_ONLY" and "[READY BUY" not in message.upper(): print("[READY BUY ONLY] Skip notifikasi otomatis non entry", flush=True) return False url = f"https://api.telegram.org/bot{token}/sendMessage" payload = { "chat_id": chat_id, "text": message, "parse_mode": "HTML", "disable_web_page_preview": "true", } if reply_markup: payload["reply_markup"] = json.dumps(reply_markup) data = urllib.parse.urlencode(payload).encode() # Retry sampai berhasil; jeda membesar (maks 5 menit), batas pengaman total 2 jam. deadline = time.time() + 2 * 3600 attempt = 0 while True: attempt += 1 req = urllib.request.Request(url, data=data) try: with urllib.request.urlopen(req, timeout=15) as resp: if attempt > 1: print(f"✅ Telegram terkirim setelah {attempt} percobaan", flush=True) return True except Exception as e: print(f"⚠️ Gagal kirim Telegram (percobaan {attempt}): {e}", flush=True) if time.time() >= deadline: print("❌ Menyerah kirim Telegram setelah 2 jam.", flush=True) return False time.sleep(min(5 * attempt, 300))