From 09ee5208d4c4033730ef4cd69c1f992e93c7adfc Mon Sep 17 00:00:00 2001 From: harwinder Date: Wed, 8 Jul 2026 21:30:16 -0400 Subject: [PATCH] Telegram bot: WS heartbeat, breach alerts, alert dedup, interactive menu MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit WebSocket: - Send heartbeat every 10s — Twelve Data drops idle connections, so the monitor previously died on weekends/quiet hours and churned reconnects - Log failed symbol subscriptions from subscribe-status events - Stamp fresh live tick price + age into every alert message Alerts: - Re-enable level-breach alerts: WS breach triggers M15 rescan and sends the report when actionable (was scanning and discarding the result) - Dedup: identical (signal, quality, direction) per pair/TF is sent once, re-armed after 1h — no more repeat alerts every candle - Configurable minimum alert quality (C+/B+/A), cycled from the menu - Trade setup shows dollar risk/reward alongside pips Interactive menu (/start, /scan): - Timeframe scans, Scan All, auto-scan ON/OFF toggle, alert-quality cycle, status, and help as inline buttons - Markdown-safe edits: 'message not modified' swallowed, parse failures fall back to plain text; exception text escaped --- telegram_bot.py | 256 ++++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 214 insertions(+), 42 deletions(-) diff --git a/telegram_bot.py b/telegram_bot.py index f81412a..ef268dc 100755 --- a/telegram_bot.py +++ b/telegram_bot.py @@ -38,6 +38,7 @@ import websockets from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update from telegram.constants import ParseMode +from telegram.error import BadRequest from telegram.ext import ( Application, CallbackQueryHandler, @@ -177,6 +178,18 @@ def _indicator_line(name: str, ind) -> str: return f"[{tag}] *{_escape_md(ind.name)}*: {_escape_md(short)}" +def _live_price_line(symbol: str) -> Optional[str]: + """Fresh WebSocket tick for this symbol, if we have one (<2 min old).""" + price = _last_prices.get(symbol) + ts = _last_price_ts.get(symbol, 0) + if price is None: + return None + age = time.time() - ts + if age > 120: + return None + return f"📡 Live: *{_fmt_price(price)}* ({age:.0f}s tick)" + + def format_signal_message(report: SignalReport) -> str: sig = report.final_signal emoji = _SIGNAL_EMOJI.get(sig, "âŦœ") @@ -193,7 +206,12 @@ def format_signal_message(report: SignalReport) -> str: f"🔍 *{report.display_name} Signal Analysis* ({report.primary_analysis.timeframe if report.primary_analysis else ''})", f"🕐 {now}", "", - f"💰 Price: *{_fmt_price(report.current_price)}*", + f"💰 Close: *{_fmt_price(report.current_price)}*", + ] + live = _live_price_line(report.symbol) + if live: + lines.append(live) + lines += [ f"📊 Regime: {regime_str} | Vol: {vol_str}", "", "━━━━━━━━━━━━━━━━━━━━", @@ -221,14 +239,16 @@ def format_signal_message(report: SignalReport) -> str: if report.trade_setup and sig not in (SignalStrength.NEUTRAL,): ts = report.trade_setup dir_emoji = "📈" if ts.direction == "BUY" else "📉" + risk_usd = ts.risk_pips * ts.pip_value + reward_usd = ts.reward_pips * ts.pip_value lines += [ "", "━━━━━━━━━━━━━━━━━━━━", f"🚀 *TRADE SETUP*", f"{dir_emoji} Direction: *{ts.direction}*", f"â–ļ Entry: `{_fmt_price(ts.entry_price)}`", - f"🛑 SL: `{_fmt_price(ts.stop_loss)}` ({ts.risk_pips:.1f} pips)", - f"đŸŽ¯ TP1: `{_fmt_price(ts.take_profit_1)}` ({ts.reward_pips:.1f} pips)", + f"🛑 SL: `{_fmt_price(ts.stop_loss)}` (${risk_usd:.2f} / {ts.risk_pips:.0f} pips)", + f"đŸŽ¯ TP1: `{_fmt_price(ts.take_profit_1)}` (${reward_usd:.2f} / {ts.reward_pips:.0f} pips)", ] if ts.take_profit_2: lines.append(f"đŸŽ¯ TP2: `{_fmt_price(ts.take_profit_2)}`") @@ -283,6 +303,45 @@ def scan_keyboard() -> InlineKeyboardMarkup: return InlineKeyboardMarkup(rows) +def main_menu_keyboard() -> InlineKeyboardMarkup: + """Full interactive menu: scans + settings + status in one keyboard.""" + tf_buttons = [b for b in SCAN_BUTTONS if b["tf"] is not None] + all_button = next((b for b in SCAN_BUTTONS if b["tf"] is None), None) + autoscan_label = "🔔 Auto-scan: ON" if _autoscan_chats else "🔕 Auto-scan: OFF" + quality_label = f"🎚 Alerts: {_alert_min_quality}+" if _alert_min_quality != "A" else "🎚 Alerts: A only" + rows = [ + [InlineKeyboardButton(b["label"], callback_data=b["cb"]) for b in tf_buttons], + ] + if all_button: + rows.append([InlineKeyboardButton(all_button["label"], callback_data=all_button["cb"])]) + rows.append([ + InlineKeyboardButton(autoscan_label, callback_data="menu_autoscan"), + InlineKeyboardButton(quality_label, callback_data="menu_quality"), + ]) + rows.append([ + InlineKeyboardButton("📊 Status", callback_data="menu_status"), + InlineKeyboardButton("ℹī¸ Help", callback_data="menu_help"), + ]) + return InlineKeyboardMarkup(rows) + + +async def _safe_edit(query, text: str, reply_markup=None) -> None: + """edit_message_text that survives 'message is not modified' and + Markdown parse failures (falls back to plain text).""" + try: + await query.edit_message_text( + text, parse_mode=ParseMode.MARKDOWN, reply_markup=reply_markup + ) + except BadRequest as exc: + if "not modified" in str(exc).lower(): + return + logger.warning(f"Markdown edit failed ({exc}); retrying as plain text") + try: + await query.edit_message_text(text, reply_markup=reply_markup) + except BadRequest: + pass + + # ── Signal runner (shared data provider for cache reuse) ────────────── def _run_signal_sync( @@ -313,6 +372,40 @@ async def run_signal( ) +# ── Alert gating (quality threshold + dedup) ───────────────────────── + +_QUALITY_RANK = {"A+": 4, "A": 3, "B": 2, "C": 1} +_ALERT_QUALITY_CYCLE = ["C", "B", "A"] # C+ → B+ → A-only +_alert_min_quality = "C" + +# Fingerprint of the last alert per (symbol, timeframe) so a persisting +# signal doesn't re-send an identical message every candle. +_last_alert_fp: Dict[str, tuple] = {} +_last_alert_ts: Dict[str, float] = {} +_ALERT_REARM_SECONDS = 3600 # identical signal re-alerts after 1h + + +def _passes_quality(quality: str) -> bool: + return _QUALITY_RANK.get(quality, 0) >= _QUALITY_RANK.get(_alert_min_quality, 1) + + +def _should_alert(key: str, report: SignalReport) -> bool: + """True when this signal is new/changed for the key, or re-armed.""" + setup = report.trade_setup + fp = ( + report.final_signal.name, + report.signal_quality, + setup.direction if setup else None, + ) + now = time.time() + if (_last_alert_fp.get(key) == fp + and now - _last_alert_ts.get(key, 0) < _ALERT_REARM_SECONDS): + return False + _last_alert_fp[key] = fp + _last_alert_ts[key] = now + return True + + # ── Candle-boundary scheduler ──────────────────────────────────────── _autoscan_chats: Set[int] = set() @@ -352,7 +445,11 @@ async def _scan_timeframe( _update_level_cache(reports) - actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B", "C")] + actionable = [ + r for r in reports + if _passes_quality(r.signal_quality) + and _should_alert(f"{r.symbol}_{tf.value}", r) + ] for chat_id in list(_autoscan_chats): for report in actionable: detail = format_signal_message(report) @@ -427,18 +524,46 @@ async def _handle_breach(bot, ws_symbol: str, pair_symbol: str, price: float, le _scan_cooldowns[pair_symbol] = now display = PAIR_INFO.get(pair_symbol, {}).get("display", ws_symbol) label = level_name.replace("_", " ").title() - logger.info(f"Level breach: {display} crossed {label} at {price} (alert disabled, cache updated)") + logger.info(f"Level breach: {display} crossed {label} at {price} — rescanning M15") try: report = await run_signal(pair_symbol, Timeframe.M15) _update_level_cache([report]) except Exception: logger.exception(f"Breach scan failed for {display}") + return + + # Alert only when the rescan produced an actionable, changed signal — + # same gates as the auto-scanner, so a breach can't spam the channel. + if (report.final_signal == SignalStrength.NEUTRAL + or not _passes_quality(report.signal_quality) + or not _should_alert(f"{pair_symbol}_{Timeframe.M15.value}", report)): + return + + header = ( + f"⚡ *LEVEL BREAK* — {_escape_md(display)} crossed " + f"*{_escape_md(label)}* @ `{_fmt_price(price)}`\n\n" + ) + detail = header + format_signal_message(report) + for chat_id in list(_autoscan_chats): + try: + await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN) + except Exception as exc: + logger.warning(f"Failed to send breach alert to chat {chat_id}: {exc}") + + +async def _ws_heartbeat(ws) -> None: + """Twelve Data drops idle WS connections; heartbeat every 10s keeps it + alive through quiet periods (weekends, low-liquidity hours).""" + while True: + await asyncio.sleep(10) + await ws.send(json.dumps({"action": "heartbeat"})) async def _ws_monitor(bot) -> None: backoff = 1 while _autoscan_chats: + hb_task: Optional[asyncio.Task] = None try: async with websockets.connect(_WS_URL) as ws: subscribe = { @@ -448,6 +573,7 @@ async def _ws_monitor(bot) -> None: await ws.send(json.dumps(subscribe)) logger.info(f"WebSocket connected, subscribed to {len(_WS_SYMBOLS)} symbols") backoff = 1 + hb_task = asyncio.create_task(_ws_heartbeat(ws)) async for raw in ws: if not _autoscan_chats: @@ -459,7 +585,13 @@ async def _ws_monitor(bot) -> None: logger.warning(f"WebSocket: malformed JSON, skipping") continue - if msg.get("event") != "price": + event = msg.get("event") + if event == "subscribe-status": + fails = msg.get("fails") or [] + if fails: + logger.warning(f"WebSocket: failed to subscribe: {fails}") + continue + if event != "price": continue ws_symbol = msg.get("symbol") @@ -492,6 +624,9 @@ async def _ws_monitor(bot) -> None: logger.exception(f"WebSocket error, reconnecting in {backoff}s") await asyncio.sleep(backoff) backoff = min(backoff * 2, 60) + finally: + if hb_task is not None: + hb_task.cancel() logger.info("WebSocket monitor stopped") @@ -572,27 +707,31 @@ async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: await update.message.reply_text( text, parse_mode=ParseMode.MARKDOWN, - reply_markup=scan_keyboard(), + reply_markup=main_menu_keyboard(), ) +_HELP_TEXT = ( + "📖 *ForexBot2 Commands*\n\n" + "/start — Welcome & interactive menu\n" + "/scan — Quick XAU/USD scan menu\n" + "/autoscan — Toggle auto-scan on/off\n" + "/status — Bot status, WS connection, live prices\n" + "/cache — Show data cache age per pair/TF\n" + "/clearcache — Wipe data cache, force fresh fetch\n" + "/help — This message\n\n" + "*Auto-scan* fires at candle close: M5/M15/H1/H4.\n" + "*Alerts* button sets the minimum quality grade that gets sent.\n" + "*WebSocket* streams live prices; key-level breaches trigger an " + "instant M15 rescan and alert (5-min cooldown)." +) + + async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: - text = ( - "📖 *ForexBot2 Commands*\n\n" - "/start — Welcome & scan buttons\n" - "/scan — Quick XAU/USD scan buttons\n" - "/autoscan — Toggle auto-scan on/off\n" - "/status — Bot status, WS connection, live prices\n" - "/cache — Show data cache age per pair/TF\n" - "/clearcache — Wipe data cache, force fresh fetch\n" - "/help — This message\n\n" - "*Auto-scan* fires at candle close: M5/M15/H1/H4.\n" - "*WebSocket* monitors live prices and triggers breach scans." - ) - await update.message.reply_text(text, parse_mode=ParseMode.MARKDOWN) + await update.message.reply_text(_HELP_TEXT, parse_mode=ParseMode.MARKDOWN) -async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: +def _status_text() -> str: lines = ["📊 *ForexBot2 Status*\n"] if _autoscan_chats: @@ -639,7 +778,13 @@ async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None if active_breach: lines.append(f"*Breach tasks:* {active_breach} running") - await update.message.reply_text("\n".join(lines), parse_mode=ParseMode.MARKDOWN) + lines.append(f"\n*Alert quality:* {_alert_min_quality}+ " + f"| re-arm {_ALERT_REARM_SECONDS // 60}m") + return "\n".join(lines) + + +async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + await update.message.reply_text(_status_text(), parse_mode=ParseMode.MARKDOWN) async def cmd_cache(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: @@ -715,7 +860,7 @@ async def cmd_scan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: await update.message.reply_text( "🔍 *XAU/USD — Pick a timeframe*", parse_mode=ParseMode.MARKDOWN, - reply_markup=scan_keyboard(), + reply_markup=main_menu_keyboard(), ) @@ -731,30 +876,60 @@ async def cb_scan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: await _cb_scan_all(query) return - await query.edit_message_text( - f"âŗ Scanning XAU/USD {btn['label']}â€Ļ", - parse_mode=ParseMode.MARKDOWN, - ) + await _safe_edit(query, f"âŗ Scanning XAU/USD {btn['label']}â€Ļ") try: report = await run_signal(btn["symbol"], btn["tf"]) text = format_signal_message(report) except Exception as exc: logger.exception("Signal scan failed") - text = f"❌ Scan failed: {exc}" + text = f"❌ Scan failed: {_escape_md(str(exc))}" - await query.edit_message_text( - text, - parse_mode=ParseMode.MARKDOWN, - reply_markup=scan_keyboard(), - ) + await _safe_edit(query, text, reply_markup=main_menu_keyboard()) + + +async def cb_menu(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + """Interactive menu actions: autoscan toggle, quality cycle, status, help.""" + global _alert_min_quality + query = update.callback_query + action = query.data + + if action == "menu_autoscan": + chat_id = query.message.chat.id + if chat_id in _autoscan_chats: + _autoscan_chats.discard(chat_id) + if not _autoscan_chats: + _stop_scheduler() + await query.answer("Auto-scan stopped") + else: + _autoscan_chats.add(chat_id) + _start_scheduler(context.bot) + await query.answer("Auto-scan started") + try: + await query.edit_message_reply_markup(reply_markup=main_menu_keyboard()) + except BadRequest: + pass + + elif action == "menu_quality": + i = _ALERT_QUALITY_CYCLE.index(_alert_min_quality) if _alert_min_quality in _ALERT_QUALITY_CYCLE else 0 + _alert_min_quality = _ALERT_QUALITY_CYCLE[(i + 1) % len(_ALERT_QUALITY_CYCLE)] + await query.answer(f"Alerting {_alert_min_quality}+ signals") + try: + await query.edit_message_reply_markup(reply_markup=main_menu_keyboard()) + except BadRequest: + pass + + elif action == "menu_status": + await query.answer() + await query.message.reply_text(_status_text(), parse_mode=ParseMode.MARKDOWN) + + elif action == "menu_help": + await query.answer() + await query.message.reply_text(_HELP_TEXT, parse_mode=ParseMode.MARKDOWN) async def _cb_scan_all(query) -> None: - await query.edit_message_text( - "âŗ Scanning XAU/USD across M5 / M15 / H1 / H4â€Ļ", - parse_mode=ParseMode.MARKDOWN, - ) + await _safe_edit(query, "âŗ Scanning XAU/USD across M5 / M15 / H1 / H4â€Ļ") tfs = [Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4] tasks = [run_signal(XAUUSD, tf) for tf in tfs] @@ -803,11 +978,7 @@ async def _cb_scan_all(query) -> None: lines.append("_Use buttons below for detailed single-TF view._") - await query.edit_message_text( - "\n".join(lines), - parse_mode=ParseMode.MARKDOWN, - reply_markup=scan_keyboard(), - ) + await _safe_edit(query, "\n".join(lines), reply_markup=main_menu_keyboard()) # ── Boot-time auto-start ────────────────────────────────────────────── @@ -908,6 +1079,7 @@ def main() -> None: app.add_handler(CommandHandler("cache", cmd_cache)) app.add_handler(CommandHandler("clearcache", cmd_clearcache)) app.add_handler(CallbackQueryHandler(cb_scan, pattern="^scan_")) + app.add_handler(CallbackQueryHandler(cb_menu, pattern="^menu_")) logger.info("Bot running — press Ctrl+C to stop") app.run_polling(allowed_updates=Update.ALL_TYPES)