#!/usr/bin/env python3 """ ForexBot2 — Telegram Bot ======================== Async Telegram interface with candle-boundary auto-scanner and real-time WebSocket level-breach monitor. Manual scan: XAU/USD × 3 timeframes (M15/H1/H4) via inline buttons. Auto-scan: Starts automatically on boot if TELEGRAM_CHAT_ID is set. /autoscan toggles for additional chats or to stop. M5 fires at :00/:05/:10/… + 10s buffer. M15 fires at :00/:15/:30/:45 + 10s buffer. H1 fires at :00 + 10s buffer. H4 fires at :00 every 4 hours + 10s buffer. WebSocket: Streams live ticks from the streaming provider — config.json `data_providers.streaming` when set (e.g. tiingo), else the active provider if it streams; auto-disabled otherwise. When price crosses a key structure level (swing S/R, BB, EMA200) cached from the last boundary scan, triggers an immediate M15 scan and sends a Level Break Alert. 5-minute cooldown prevents spam. News: /news replies with the net XAU/USD news sentiment score (display-only — news never gates signals). Owner lock: All commands and buttons only respond to TELEGRAM_CHAT_ID (the owner). Everyone else gets a polite refusal — the bot is a private control panel, not the product. Channel: If TELEGRAM_CHANNEL_ID is set (-100xxxxxxxxxx), auto-scan signal alerts and level-breach alerts are also posted to that channel. Pair it with a Telegram Stars subscription invite link to sell access — Telegram manages join/renewal/ kick; the bot just posts there (needs channel-admin + post-messages). Manual scans, status and cache replies stay owner-DM only. The channel feed runs even when the owner toggles their own DM auto-scan off. Usage: python telegram_bot.py TELEGRAM_TOKEN= TELEGRAM_CHAT_ID= \ TELEGRAM_CHANNEL_ID=<-100…> python telegram_bot.py """ import asyncio import functools import logging import logging.handlers import os import sys import threading import time from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone from typing import Dict, List, Optional, Set from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update from telegram.constants import ParseMode from telegram.error import BadRequest from telegram.ext import ( Application, CallbackQueryHandler, CommandHandler, ContextTypes, ) from config import ( PlatformConfig, SignalStrength, Timeframe, XAUUSD, PAIR_INFO, ) from app_config import ( get_app_config, telegram_token, telegram_chat_id, telegram_channel_id, ) from providers import BaseDataProvider, make_data_provider from news_engine import FundamentalContext, NewsEngine from pair_signals import PAIR_PRODUCERS from signal_aggregator import SignalAggregator, SignalReport _APP = get_app_config() # ── Force UTF-8 on Windows ──────────────────────────────────────────── if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") _LOG_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "logs") os.makedirs(_LOG_DIR, exist_ok=True) _log_formatter = logging.Formatter( "%(asctime)s [%(levelname)s] %(name)s: %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) _console_handler = logging.StreamHandler() _console_handler.setFormatter(_log_formatter) # 3-hour rotation × 240 backups = 30 days retention. Handlers sit on the # ROOT logger (force=True guarantees ownership even if something logged # first), so every module logger — data_provider, news_engine, # signal_aggregator, websockets, telegram.ext — lands in the same file. _file_handler = logging.handlers.TimedRotatingFileHandler( os.path.join(_LOG_DIR, "telegram.log"), when="h", interval=3, backupCount=240, encoding="utf-8", ) _file_handler.setFormatter(_log_formatter) logging.basicConfig(level=getattr(logging, _APP.log_level, logging.INFO), handlers=[_console_handler, _file_handler], force=True) logging.getLogger("httpx").setLevel(logging.WARNING) logging.getLogger("httpcore").setLevel(logging.WARNING) # Never let DEBUG runs write credentials into the 30-day rotating log: # the websockets client logger emits the connect URI (Twelve Data puts # the API key in the query string) and frame contents (Tiingo's auth # frame carries the key) — 2026-08-04 audit, finding K47. logging.getLogger("websockets").setLevel(logging.INFO) logging.getLogger("websockets.client").setLevel(logging.INFO) logger = logging.getLogger("telegram_bot") TOKEN = telegram_token() CHAT_ID = telegram_chat_id() # Owner lock: only this user/chat may drive the bot (commands + buttons). # The bot is the product's signal source — without the lock, anyone who # finds it could /autoscan and receive paid signals for free in DM. OWNER_ID: Optional[int] = int(CHAT_ID) if CHAT_ID else None # Subscriber channel (paid, Stars-subscription invite link managed by # Telegram). When set, auto-scan signal alerts and level-breach alerts # are also posted there. Manual scans, status, cache and boot messages # stay owner-DM only. Set TELEGRAM_CHANNEL_ID="" to disable the feed. _CHANNEL_ID_RAW = telegram_channel_id() CHANNEL_ID: Optional[int] = int(_CHANNEL_ID_RAW) if _CHANNEL_ID_RAW else None # ── Pairs & buttons ─────────────────────────────────────────────────── # The scanned pair list is config.json-driven (`scanning.pairs`, read via # app_config) and is the single switch for single- vs multi-pair operation. # Buttons, WS subscriptions, and scan loops all derive from it. SCAN_PAIRS = list(_APP.scan_pairs) _PAIRS_LABEL = " + ".join(PAIR_INFO[p]["display"] for p in SCAN_PAIRS) _LEVEL_KEYS = ( "swing_high_1", "swing_high_2", "swing_high_3", "swing_low_1", "swing_low_2", "swing_low_3", "ema_trend", "bb_upper", "bb_lower", "kc_upper", "kc_lower", ) _TF_BUTTON_LABELS = [ (Timeframe.M5, "5 Min"), (Timeframe.M15, "15 Min"), (Timeframe.H1, "1 Hour"), (Timeframe.H4, "4 Hour"), ] def _build_scan_buttons() -> List[dict]: """One row of TF buttons + a scan-all button per scanned pair.""" multi = len(SCAN_PAIRS) > 1 buttons = [] for symbol in SCAN_PAIRS: slug = symbol.replace("/", "").lower() base = PAIR_INFO[symbol]["display"].split("/")[0] # XAU prefix = f"{base} " if multi else "" for tf, label in _TF_BUTTON_LABELS: buttons.append({ "cb": f"scan_{slug}_{tf.value}", "symbol": symbol, "tf": tf, "label": f"{prefix}{label}", }) buttons.append({ "cb": f"scan_{slug}_all", "symbol": symbol, "tf": None, "label": f"Scan All TFs — {base}" if multi else "Scan All TFs", }) return buttons SCAN_BUTTONS = _build_scan_buttons() SCAN_LOOKUP = {b["cb"]: b for b in SCAN_BUTTONS} TF_CONFIRMATIONS = { Timeframe.M5: [Timeframe.M15, Timeframe.H1], Timeframe.M15: [Timeframe.M5, Timeframe.H1], Timeframe.H1: [Timeframe.M15, Timeframe.H4], Timeframe.H4: [Timeframe.H1, Timeframe.D1], } AUTOSCAN_CONFIRMATIONS = { Timeframe.M5: [Timeframe.M15, Timeframe.H1], Timeframe.M15: [Timeframe.M5, Timeframe.H1], Timeframe.H1: [Timeframe.M15, Timeframe.H4], Timeframe.H4: [Timeframe.H1, Timeframe.D1], } _executor = ThreadPoolExecutor(max_workers=4) # ── Markdown escaping ───────────────────────────────────────────────── def _escape_md(text: str) -> str: for ch in ("*", "_", "`", "[", "]"): text = text.replace(ch, "\\" + ch) return text def _fmt_price(price: float) -> str: return f"{price:.2f}" if price >= 10 else f"{price:.5f}" # ── Shared data provider (single cache across all scans) ───────────── _shared_dp: Optional[BaseDataProvider] = None def _get_shared_dp() -> BaseDataProvider: global _shared_dp if _shared_dp is None: _shared_dp = make_data_provider(PlatformConfig()) return _shared_dp _streaming_dp: Optional[BaseDataProvider] = None def _get_streaming_dp() -> BaseDataProvider: """Provider for the tick stream: config.json data_providers.streaming when set (a SEPARATE instance — the candle cache stays on the active provider), else the shared active provider. Reads app config at call time, not the import-time _APP snapshot, so tests can swap it via set_app_config().""" global _streaming_dp if _streaming_dp is None: app = get_app_config() if app.streaming_provider is None or \ app.streaming_provider == app.active_provider: _streaming_dp = _get_shared_dp() else: _streaming_dp = make_data_provider( PlatformConfig(), app.streaming_provider) return _streaming_dp # ── Signal formatting ───────────────────────────────────────────────── _SIGNAL_EMOJI = { SignalStrength.STRONG_BUY: "🟢", SignalStrength.BUY: "🟩", SignalStrength.NEUTRAL: "⬜", SignalStrength.SELL: "🟥", SignalStrength.STRONG_SELL: "🔴", } _QUALITY_EMOJI = { "A+": "🏆", "A": "🥇", "B": "🥈", "C": "🥉", "D": "⚠️", "F": "❌", } def _signal_label(sig: SignalStrength) -> str: return sig.name.replace("_", " ") def _mtf_line(report: SignalReport) -> str: if not report.mtf_signals: return "" parts = [] for tf, sig in report.mtf_signals.items(): score = report.mtf_scores.get(tf, 0) parts.append(f"{tf}: {_signal_label(sig)} {score:+.1f}") return " | ".join(parts) # Informational fixed-pip target shown on every trade setup (display-only, # not part of the backtested exit model). _INFO_TP_PIPS = 100 _IND_SHORT = { "ema_trend": "EMA", "rsi": "RSI", "macd": "MACD", "bollinger": "BB", "adx": "ADX", "roc": "ROC", "keltner": "KC", "divergence": "Div", "candlestick": "Cndl", } _IND_ARROW = { SignalStrength.STRONG_BUY: "⇈", SignalStrength.BUY: "↑", SignalStrength.NEUTRAL: "→", SignalStrength.SELL: "↓", SignalStrength.STRONG_SELL: "⇊", } def _indicator_summary(report: SignalReport) -> str: """One-line condensed indicator readout: 'EMA↑ RSI↑ MACD→ …'.""" if not report.primary_analysis: return "" parts = [] for name, ind in report.primary_analysis.indicators.items(): if name.startswith("_") or ind.weight == 0: continue label = _IND_SHORT.get(name, name) parts.append(f"{label}{_IND_ARROW.get(ind.signal, '→')}") return " ".join(parts) 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)" _DISCLAIMER = ( "⚠️ _Not financial advice. For educational and research purposes only. " "Trading forex/CFDs carries a high risk of loss — do your own analysis " "and never risk money you cannot afford to lose._" ) def format_signal_message(report: SignalReport) -> str: sig = report.final_signal emoji = _SIGNAL_EMOJI.get(sig, "⬜") qe = _QUALITY_EMOJI.get(report.signal_quality, "") label = _signal_label(sig) mtf_tag = "MTF ✅" if report.mtf_agreement else "MTF ❌" tf = report.primary_analysis.timeframe if report.primary_analysis else "" now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") regime_str = _escape_md(report.market_regime.replace("_", " ").title()) vol_str = _escape_md(str(report.volatility_regime)) lines: list[str] = [ f"{emoji} *{label} · {report.display_name} · {tf}* {qe}{report.signal_quality}", f"🕐 {now} · 💰 {_fmt_price(report.current_price)}", ] live = _live_price_line(report.symbol) if live: lines.append(live) lines += [ f"Score *{report.final_score:+.1f}* · Conf *{report.final_confidence:.0f}%* · {mtf_tag}", f"📊 Regime: {regime_str} · Vol: {vol_str}", ] if report.action_recommendation: lines.append(f"_{_escape_md(report.action_recommendation)}_") 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 += [ "", "🚀 *TRADE SETUP*", f"{dir_emoji} Direction: *{ts.direction}*", f"▶ Entry `{_fmt_price(ts.entry_price)}`", f"🛑 SL `{_fmt_price(ts.stop_loss)}` (${risk_usd:.2f} / {ts.risk_pips:.0f}p)", f"🎯 TP1 `{_fmt_price(ts.take_profit_1)}` (${reward_usd:.2f} / {ts.reward_pips:.0f}p)", ] if ts.take_profit_2: lines.append(f"🎯 TP2 `{_fmt_price(ts.take_profit_2)}`") # Informational fixed-pip target: where a 100-pip TP would sit for # this setup (display-only — not part of the backtested exit model). info_dist = _INFO_TP_PIPS * ts.pip_value if ts.direction == "BUY": info_price = ts.entry_price + info_dist else: info_price = ts.entry_price - info_dist risk_dist = abs(ts.entry_price - ts.stop_loss) info_r = f" · ≈{info_dist / risk_dist:.1f}R" if risk_dist > 0 else "" lines.append( f"🎯 TP {_INFO_TP_PIPS}p: `{_fmt_price(info_price)}` (${info_dist:.2f}{info_r})" ) lines.append(f"⚖️ RR 1:{ts.risk_reward:.1f}") # Trailing-stop management plan (matches backtested exit model: # TP1 is advisory, the trail is the exit). Params are stamped on # the setup by the producing config so per-TF factory overrides # (M15 arm 0.70, M5 dist 0.85) are what subscribers see; only a # legacy unstamped setup falls back to platform defaults. is_pb = getattr(ts, "strategy", "GENERIC").endswith("_PB") arm_r = getattr(ts, "trail_arm_r", None) dist_r = getattr(ts, "trail_dist_r", None) if arm_r is None or dist_r is None: sig_cfg = PlatformConfig().signals arm_r = sig_cfg.pb_trail_arm_r if is_pb else sig_cfg.trail_arm_r dist_r = sig_cfg.pb_trail_dist_r if is_pb else sig_cfg.trail_dist_r if arm_r > 0: if ts.direction == "BUY": arm_price = ts.entry_price + arm_r * risk_dist else: arm_price = ts.entry_price - arm_r * risk_dist trail_dist = dist_r * risk_dist tag = " (pullback)" if is_pb else "" lines.append( f"🔄 Trail{tag}: arm `{_fmt_price(arm_price)}` " f"(+{arm_r:.1f}R), trail SL ${trail_dist:.2f} " f"behind best price — TP1 is a guide, let it run" ) extra: list[str] = [] mtf = _mtf_line(report) if mtf: extra.append(f"📈 MTF: {mtf}") ind_summary = _indicator_summary(report) if ind_summary: extra.append(f"📋 {ind_summary}") for w in report.warnings: extra.append(f"⚠️ {_escape_md(w)}") if report.news_summary: extra.append(f"📰 {_escape_md(report.news_summary)}") if extra: lines += [""] + extra lines += ["", _DISCLAIMER] return "\n".join(lines) # ── Auto-scan summary format ───────────────────────────────────────── def _format_scan_summary(reports: List[SignalReport], tf: Timeframe) -> str: now = datetime.now(timezone.utc).strftime("%H:%M UTC") lines = [ f"📊 *{tf.value} Auto-Scan* | {now}", "━━━━━━━━━━━━━━━━━━━━", ] for r in reports: emoji = _SIGNAL_EMOJI.get(r.final_signal, "⬜") sig = _signal_label(r.final_signal) mtf = "✅" if r.mtf_agreement else "❌" lines.append( f"{emoji} *{r.display_name}* {sig} " f"{_QUALITY_EMOJI.get(r.signal_quality, '')}{r.signal_quality} " f"{r.final_score:+.1f} {r.final_confidence:.0f}% {mtf}MTF" ) lines.append("━━━━━━━━━━━━━━━━━━━━") actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B", "C")] if actionable: names = ", ".join(r.display_name for r in actionable) lines.append(f"🔔 Actionable: {names}") else: lines.append("No actionable signals this cycle.") lines += ["", _DISCLAIMER] return "\n".join(lines) # ── Keyboard helper ─────────────────────────────────────────────────── def _scan_button_rows() -> List[List[InlineKeyboardButton]]: """Per pair: one row of TF buttons + one scan-all row.""" rows = [] for symbol in SCAN_PAIRS: tf_buttons = [b for b in SCAN_BUTTONS if b["symbol"] == symbol and b["tf"] is not None] rows.append([InlineKeyboardButton(b["label"], callback_data=b["cb"]) for b in tf_buttons]) all_button = next((b for b in SCAN_BUTTONS if b["symbol"] == symbol and b["tf"] is None), None) if all_button: rows.append([InlineKeyboardButton(all_button["label"], callback_data=all_button["cb"])]) return rows def scan_keyboard() -> InlineKeyboardMarkup: return InlineKeyboardMarkup(_scan_button_rows()) def main_menu_keyboard() -> InlineKeyboardMarkup: """Full interactive menu: scans + settings + status in one keyboard.""" 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 = _scan_button_rows() 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( symbol: str, timeframe: Timeframe, confirmation_tfs: Optional[List[Timeframe]] = None, ) -> SignalReport: if confirmation_tfs is None: confirmation_tfs = TF_CONFIRMATIONS.get( timeframe, [Timeframe.M15, Timeframe.H4] ) # Per-pair tuned configs (regime weights, gates, sessions) come from # the pair_signals factories; a plain PlatformConfig would silently # run every pair with XAU defaults. factory = PAIR_PRODUCERS.get(symbol) if factory is not None: config = factory(timeframe, confirmation_tfs) else: config = PlatformConfig() config.primary_timeframe = timeframe config.confirmation_timeframes = confirmation_tfs aggregator = SignalAggregator(config) aggregator.data_provider = _get_shared_dp() return aggregator.analyze_pair(symbol=symbol) async def run_signal( symbol: str, timeframe: Timeframe, confirmation_tfs: Optional[List[Timeframe]] = None, ) -> SignalReport: loop = asyncio.get_running_loop() return await loop.run_in_executor( _executor, _run_signal_sync, symbol, timeframe, confirmation_tfs ) # ── 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 = _APP.alert_min_quality # 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] = {} _last_alert_setup: Dict[str, tuple] = {} # key → (entry, risk) of last ALERTED setup _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 _alert_fingerprint(report: SignalReport) -> tuple: """Alert identity for dedup: signal, quality, setup direction. Entry movement is checked separately in _should_alert against the last ALERTED entry/risk (a stable reference). Deriving a price bucket from the CURRENT SL distance here made the fingerprint churn on normal ATR/anchor drift — near-duplicate alerts every candle.""" setup = report.trade_setup return ( report.final_signal.name, report.signal_quality, setup.direction if setup else None, ) def _should_alert(key: str, report: SignalReport) -> bool: """True when this signal is new/changed for the key, or re-armed. A persisting identical signal dedups for the re-arm hour, but a setup whose entry moved >= half the last-alerted risk distance (new pullback/swing anchor) re-alerts immediately. Pure check — call _commit_alert only after the message actually went out, so a failed Telegram send doesn't consume the alert for an hour. """ fp = _alert_fingerprint(report) if _last_alert_fp.get(key) != fp: return True if time.time() - _last_alert_ts.get(key, 0) >= _ALERT_REARM_SECONDS: return True setup = report.trade_setup last = _last_alert_setup.get(key) if setup is not None and last is not None: last_entry, last_risk = last if abs(setup.entry_price - last_entry) >= 0.5 * max(last_risk, 1e-9): return True return False def _commit_alert(key: str, report: SignalReport) -> None: _last_alert_fp[key] = _alert_fingerprint(report) _last_alert_ts[key] = time.time() setup = report.trade_setup if setup is not None: _last_alert_setup[key] = ( setup.entry_price, abs(setup.entry_price - setup.stop_loss) ) else: _last_alert_setup.pop(key, None) # Per-key locks serializing the _should_alert → send → _commit_alert # window. The auto-scan loop and a concurrent breach handler share the # "XAU/USD_15min" dedup key; without the lock both pass _should_alert # while the other's send is still awaiting Telegram, and the same signal # goes out twice (2026-08-04 audit H36). Per-key rather than one global # lock so a slow send on one timeframe cannot head-of-line-block another # timeframe's alert. Keys are symbol × TF, so the dict stays tiny. _alert_locks: Dict[str, asyncio.Lock] = {} def _alert_lock(key: str) -> asyncio.Lock: lock = _alert_locks.get(key) if lock is None: lock = _alert_locks[key] = asyncio.Lock() return lock async def _send_alert(bot, chat_id: int, text: str) -> bool: """Markdown alert send with a plain-text fallback on parse failures, mirroring _safe_edit (2026-08-04 audit H35 — alert paths previously dropped the message on a BadRequest with only a warning logged). Returns True when a message reached the chat by either attempt.""" try: await bot.send_message(chat_id, text, parse_mode=ParseMode.MARKDOWN) return True except BadRequest as exc: logger.warning(f"Markdown alert failed for chat {chat_id} ({exc}); " f"retrying as plain text") try: await bot.send_message(chat_id, text) return True except Exception as exc2: logger.warning(f"Failed to send alert to chat {chat_id}: {exc2}") return False except Exception as exc: logger.warning(f"Failed to send alert to chat {chat_id}: {exc}") return False # ── Candle-boundary scheduler ──────────────────────────────────────── _autoscan_chats: Set[int] = set() _autoscan_tasks: list[asyncio.Task] = [] def _feed_active() -> bool: """Scan loops keep running while anyone is listening: owner DM auto-scan chats or the subscriber channel.""" return bool(_autoscan_chats) or CHANNEL_ID is not None def _signal_targets() -> List[int]: """Where signal/breach alerts go: owner auto-scan chats + channel.""" targets: List[int] = list(_autoscan_chats) if CHANNEL_ID is not None: targets.append(CHANNEL_ID) return targets def _seconds_until_boundary(interval_minutes: int, buffer: int = 10) -> float: """Seconds until the next candle close + buffer.""" now = datetime.now(timezone.utc) total_minutes = now.hour * 60 + now.minute elapsed_minutes = total_minutes % interval_minutes elapsed_seconds = elapsed_minutes * 60 + now.second + now.microsecond / 1e6 interval_seconds = interval_minutes * 60 remaining = interval_seconds - elapsed_seconds + buffer # NOTE: remaining > buffer always (elapsed < interval by construction); # a call at the exact boundary instant targets the NEXT boundary — # steady-state loops wake at boundary+buffer, so only a process # started in the boundary microsecond skips one scan (2026-08-04 # audit I42; the old `remaining <= buffer` guard was unreachable). return remaining async def _scan_timeframe( bot, tf: Timeframe, confirmation_tfs: List[Timeframe], ) -> None: """Scan all pairs for one timeframe and broadcast results.""" tasks = [run_signal(symbol, tf, confirmation_tfs) for symbol in SCAN_PAIRS] results = await asyncio.gather(*tasks, return_exceptions=True) reports: List[SignalReport] = [] for i, result in enumerate(results): if isinstance(result, Exception): # gather(return_exceptions=True) means there is no active # exception here — logger.exception would log "NoneType: None" # instead of the real stack; pass the instance explicitly. logger.error( f"Auto-scan failed for {SCAN_PAIRS[i]} {tf.value}: {result}", exc_info=result, ) else: reports.append(result) if not reports: return _update_level_cache(reports) candidates = [ r for r in reports if r.final_signal != SignalStrength.NEUTRAL # same gate as breach path and _passes_quality(r.signal_quality) ] for report in candidates: key = f"{report.symbol}_{tf.value}" # The lock spans check → send → commit so a concurrent breach # handler on the same key cannot double-alert (H36). async with _alert_lock(key): if not _should_alert(key, report): continue detail = format_signal_message(report) delivered = False for chat_id in _signal_targets(): if await _send_alert(bot, chat_id, detail): delivered = True # Commit only when the alert reached at least one target; # a fully failed send stays eligible for the next scan. if delivered: _commit_alert(key, report) # ── WebSocket level-breach monitor ────────────────────────────────── # Level cache keyed symbol → timeframe → {level_name: price}, with a # matching per-TF write timestamp. Keyed per TF (2026-08-04 audit H34) # because every scan path writes here — the M5/M15/H1/H4 boundary loops, # the breach rescan, the manual Scan-All button — and a single flat dict # meant the last writer wiped the other timeframes' swing/EMA levels # within minutes. Breach checks scan the union of all fresh TF entries. _level_cache: Dict[str, Dict[str, Dict[str, float]]] = {} _level_cache_ts: Dict[str, Dict[str, float]] = {} _LEVEL_CACHE_TTL = 3600 _last_prices: Dict[str, float] = {} _last_price_ts: Dict[str, float] = {} _scan_cooldowns: Dict[str, float] = {} # Breaches are edge-triggered (only the crossing tick fires), so one that # lands while the symbol is inside the rescan cooldown would otherwise be # dropped forever (H33b). Park it here; the tick loop fires it once the # cooldown expires. symbol → (price, level_name, recorded_at). _pending_breach: Dict[str, tuple] = {} # A failed breach rescan shrinks the just-set cooldown to this retry # window instead of silencing the symbol for the full cooldown (H33a). _BREACH_RETRY_SECONDS = 30 _cfg = PlatformConfig() _WS_COOLDOWN = _cfg.ws_cooldown_seconds _breach_tasks: Set[asyncio.Task] = set() def _update_level_cache(reports: List[SignalReport]) -> None: """Each report updates only its own timeframe's slot — an M5 scan can no longer overwrite the H1/H4 swing levels (H34).""" now = time.time() for r in reports: if r.primary_analysis and r.primary_analysis.key_levels: levels = { k: v for k, v in r.primary_analysis.key_levels.items() if k in _LEVEL_KEYS } if levels: tf = r.primary_analysis.timeframe _level_cache.setdefault(r.symbol, {})[tf] = levels _level_cache_ts.setdefault(r.symbol, {})[tf] = now def _check_level_breach(symbol: str, price: float) -> Optional[str]: """Scan the union of every fresh timeframe's levels; returns ' ' (e.g. '15min swing_high_1') on a crossing so the alert label says which timeframe's level broke.""" per_tf = _level_cache.get(symbol) if not per_tf: return None last = _last_prices.get(symbol) if last is None: return None now = time.time() ts_map = _level_cache_ts.get(symbol, {}) for tf, levels in per_tf.items(): if now - ts_map.get(tf, 0) > _LEVEL_CACHE_TTL: continue # stale TF entry — skip, fresher TFs still count for name, level in levels.items(): if (last < level <= price) or (last > level >= price): return f"{tf} {name}" return None def _cleanup_cooldowns() -> None: now = time.time() expired = [k for k, v in _scan_cooldowns.items() if now - v > _WS_COOLDOWN * 2] for k in expired: del _scan_cooldowns[k] async def _handle_breach(bot, pair_symbol: str, price: float, level_name: str) -> None: now = time.time() if now - _scan_cooldowns.get(pair_symbol, 0) < _WS_COOLDOWN: # Park the crossing instead of dropping it — price holding beyond # the level never re-fires the edge trigger, so without this the # breach would be lost forever (H33b). The tick loop fires it # once the cooldown expires; a newer crossing overwrites it. _pending_breach[pair_symbol] = (price, level_name, now) return # Early-set on purpose: this is what stops concurrent breach tasks # from stacking rescans. The failure path below rolls it back. _scan_cooldowns[pair_symbol] = now _pending_breach.pop(pair_symbol, None) # this crossing is being handled display = PAIR_INFO.get(pair_symbol, {}).get("display", pair_symbol) label = level_name.replace("_", " ").title() 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}") # Shrink the just-set cooldown to a short retry window so a # transient fetch error doesn't silence the symbol for the full # cooldown (H33a); the floor bounds the retry rate while the # provider stays down. Re-park the breach so the tick loop # retries it (setdefault: a newer crossing recorded meanwhile # wins over this stale one). retry = min(_BREACH_RETRY_SECONDS, _WS_COOLDOWN) _scan_cooldowns[pair_symbol] = now - _WS_COOLDOWN + retry _pending_breach.setdefault(pair_symbol, (price, level_name, now)) 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)): return key = f"{pair_symbol}_{Timeframe.M15.value}" # Lock spans check → send → commit; without it this handler and an # in-flight M15 auto-scan both pass _should_alert and double-alert # the same signal (H36). async with _alert_lock(key): if not _should_alert(key, 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) sent = False for chat_id in _signal_targets(): if await _send_alert(bot, chat_id, detail): sent = True if sent: _commit_alert(key, report) def _breach_task_done(task: asyncio.Task) -> None: """Fire-and-forget breach tasks must still surface their failures — without this, an exception sits unretrieved until GC and never reaches the log file.""" _breach_tasks.discard(task) if not task.cancelled() and task.exception() is not None: logger.error("Breach alert task failed", exc_info=task.exception()) async def _ws_monitor(bot) -> None: """Level-breach tick monitor. The provider owns the connection and protocol (`iter_ticks`); this loop owns reconnect/backoff and breach dispatch. Providers without streaming quotes simply disable it.""" dp = _get_streaming_dp() if not dp.supports_streaming: logger.info( "Data provider %r has no streaming quotes — " "level-breach monitor disabled", dp.name) return if getattr(dp, "_api_key", "unused") is None: # Guard, not a traceback loop: without a key every connect # attempt would raise and re-log a full traceback every ≤60s. logger.warning( "Streaming provider %r has no API key — " "level-breach monitor disabled", dp.name) return backoff = 1 while _feed_active(): got_tick = False try: async for pair_symbol, price in dp.iter_ticks(SCAN_PAIRS): if not got_tick: # Reset backoff only once the connection actually # produced data — resetting on connect let a # rejected key or accept-then-close server pin the # backoff at minimum forever (H38a/H38b). got_tick = True backoff = 1 if not _feed_active(): break breached = _check_level_breach(pair_symbol, price) _last_prices[pair_symbol] = price _last_price_ts[pair_symbol] = time.time() if breached: # A fresh crossing supersedes any parked one. _pending_breach.pop(pair_symbol, None) task = asyncio.create_task( _handle_breach(bot, pair_symbol, price, breached) ) _breach_tasks.add(task) task.add_done_callback(_breach_task_done) elif pair_symbol in _pending_breach: pend = _pending_breach[pair_symbol] if time.time() - pend[2] > _LEVEL_CACHE_TTL: # Ancient parked breach (feed gap/weekend) — an # alert about an hours-old crossing would only # mislead; the level cache it came from is stale # anyway. _pending_breach.pop(pair_symbol) elif (time.time() - _scan_cooldowns.get(pair_symbol, 0) >= _WS_COOLDOWN): # Crossing parked during the cooldown: fire it now # that the window expired — the edge trigger alone # would have dropped it forever (H33b). _pending_breach.pop(pair_symbol) task = asyncio.create_task( _handle_breach(bot, pair_symbol, pend[0], pend[1]) ) _breach_tasks.add(task) task.add_done_callback(_breach_task_done) _cleanup_cooldowns() except asyncio.CancelledError: break except Exception: logger.exception(f"WebSocket error, reconnecting in {backoff}s") await asyncio.sleep(backoff) backoff = min(backoff * 2, 60) else: # Clean server close: the async-for ended without raising. # This must back off and log like the error path — silently # reconnecting with zero delay hot-loops against a server # that accepts and immediately closes (H38a). if _feed_active(): logger.warning( f"WebSocket closed cleanly, reconnecting in {backoff}s") await asyncio.sleep(backoff) backoff = min(backoff * 2, 60) logger.info("WebSocket monitor stopped") async def _candle_loop(bot, tf: Timeframe, interval_minutes: int) -> None: label = tf.value try: while _feed_active(): wait = _seconds_until_boundary(interval_minutes, buffer=10) logger.info(f"{label} auto-scan: next in {wait:.0f}s") await asyncio.sleep(wait) if not _feed_active(): break logger.info(f"{label} auto-scan: firing") try: await _scan_timeframe( bot, tf, AUTOSCAN_CONFIRMATIONS[tf] ) except Exception: # One failed cycle must not kill the loop for good — # log it and wait for the next candle boundary. logger.exception(f"{label} auto-scan: cycle failed") except asyncio.CancelledError: pass finally: logger.info(f"{label} auto-scan loop stopped") _TF_INTERVAL_MINUTES = { Timeframe.M5: 5, Timeframe.M15: 15, Timeframe.M30: 30, Timeframe.H1: 60, Timeframe.H4: 240, Timeframe.D1: 1440, } _AUTOSCAN_LOOPS = [ (tf, _TF_INTERVAL_MINUTES[tf]) for tf in _APP.scan_timeframes ] _ws_task: Optional[asyncio.Task] = None def _start_scheduler(bot) -> None: global _ws_task # Per-TF respawn (2026-08-04 audit H37): drop finished loop tasks and # recreate only the missing ones. The old all-or-nothing alive check # left a single dead TF loop unmonitored forever while its siblings # lived — and with the channel feed active, /autoscan never reaches # _stop_scheduler, so nothing short of a process restart recovered it. _autoscan_tasks[:] = [t for t in _autoscan_tasks if not t.done()] alive_names = {t.get_name() for t in _autoscan_tasks} started = [] for tf, interval in _AUTOSCAN_LOOPS: name = f"candle_loop_{tf.name}" if name in alive_names: continue _autoscan_tasks.append( asyncio.create_task(_candle_loop(bot, tf, interval), name=name) ) started.append(tf.name) if started: logger.info("Candle-boundary scheduler started (%s)", "/".join(started)) if _ws_task is None or _ws_task.done(): _ws_task = asyncio.create_task(_ws_monitor(bot)) logger.info("WebSocket level-breach monitor started") def _stop_scheduler() -> None: global _ws_task for t in _autoscan_tasks: t.cancel() _autoscan_tasks.clear() if _ws_task and not _ws_task.done(): _ws_task.cancel() _ws_task = None for t in list(_breach_tasks): t.cancel() _breach_tasks.clear() logger.info("Scheduler and WebSocket monitor stopped") # ── Handlers ────────────────────────────────────────────────────────── def _owner_only(handler): """Reject commands/buttons from anyone but the owner. Signals are sold via the subscriber channel; the bot itself is the owner's control panel. Without this, any user who finds the bot could /autoscan and receive the paid feed for free in their DM. """ @functools.wraps(handler) async def wrapped(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: user = update.effective_user chat = update.effective_chat uid = user.id if user else None if OWNER_ID is not None and (uid == OWNER_ID or (chat and chat.id == OWNER_ID)): return await handler(update, context) logger.warning( f"Unauthorized access blocked: user={uid} " f"chat={chat.id if chat else None} handler={handler.__name__}" ) if update.callback_query: await update.callback_query.answer("🔒 Not authorized.") elif update.message: await update.message.reply_text( "🔒 This is a private control bot. Signals are published " "in the subscriber channel." ) return wrapped @_owner_only async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: scanning = "running" if _autoscan_chats else "off" text = ( f"👋 *ForexBot2 — {_PAIRS_LABEL} Signal Intelligence*\n\n" "Pick a timeframe to scan gold.\n" "Signals include a full trade setup with entry, SL and TP.\n\n" f"Auto-scan: *{scanning}* — /autoscan to toggle\n" "/scan — Quick scan buttons\n" "/status — Bot & WebSocket status\n" "/help — All commands" ) await update.message.reply_text( text, parse_mode=ParseMode.MARKDOWN, reply_markup=main_menu_keyboard(), ) _HELP_TEXT = ( "📖 *ForexBot2 Commands*\n\n" "/start — Welcome & interactive menu\n" "/scan — Quick scan menu\n" "/autoscan — Toggle auto-scan on/off\n" "/status — Bot status, WS connection, live prices\n" "/news — Net news sentiment score for XAU/USD\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).\n" "*Channel feed*: signal + breach alerts mirror to the subscriber " "channel when TELEGRAM_CHANNEL_ID is set; /autoscan only toggles " "your DM copy." ) @_owner_only async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: await update.message.reply_text(_HELP_TEXT, parse_mode=ParseMode.MARKDOWN) def _status_text() -> str: lines = ["📊 *ForexBot2 Status*\n"] if _autoscan_chats: lines.append(f"✅ Auto-scan: *running* ({len(_autoscan_chats)} chat)") tasks_alive = sum(1 for t in _autoscan_tasks if not t.done()) lines.append(f" Candle loops: {tasks_alive}/{len(_AUTOSCAN_LOOPS)}") else: lines.append("⏹ Auto-scan: *stopped*") if CHANNEL_ID is not None: lines.append(f"📢 Channel feed: *on* (`{CHANNEL_ID}`)") if _ws_task and not _ws_task.done(): provider = _streaming_dp.name if _streaming_dp else "?" lines.append(f"📡 WebSocket: *connected* ({provider})") else: lines.append("❌ WebSocket: *disconnected*") if _last_prices: lines.append("\n*Live Prices (WebSocket):*") for pair in SCAN_PAIRS: price = _last_prices.get(pair) display = PAIR_INFO.get(pair, {}).get("display", pair) age = time.time() - _last_price_ts.get(pair, 0) if price is None: lines.append(f" {display}: waiting...") else: lines.append(f" {display}: `{_fmt_price(price)}` ({age:.0f}s ago)") else: lines.append("\n_No live prices yet (WS not streaming)_") cached_pairs = len(_level_cache) cached_levels = sum( len(levels) for per_tf in _level_cache.values() for levels in per_tf.values() ) if cached_levels: all_ts = [ ts for per_tf in _level_cache_ts.values() for ts in per_tf.values() ] oldest_age = time.time() - min(all_ts) if all_ts else 0 lines.append(f"\n*Level Cache:* {cached_levels} levels across {cached_pairs} pairs (oldest {oldest_age:.0f}s)") else: lines.append("\n*Level Cache:* empty — breach alerts start after first candle scan") active_cooldowns = sum( 1 for v in _scan_cooldowns.values() if time.time() - v < _WS_COOLDOWN ) if active_cooldowns: lines.append(f"*Cooldowns:* {active_cooldowns} pair(s) in cooldown") active_breach = sum(1 for t in _breach_tasks if not t.done()) if active_breach: lines.append(f"*Breach tasks:* {active_breach} running") lines.append(f"\n*Alert quality:* {_alert_min_quality}+ " f"| re-arm {_ALERT_REARM_SECONDS // 60}m") return "\n".join(lines) @_owner_only async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: await update.message.reply_text(_status_text(), parse_mode=ParseMode.MARKDOWN) def _news_text(context: FundamentalContext, engine: NewsEngine) -> str: net, desc = engine.get_pair_sentiment(XAUUSD, context) if round(net, 2) == 0: net = 0.0 # a tiny negative mean would render as "-0.00" lines = [ "📰 *XAU/USD News Score*\n", f"*Net score:* `{net:+.2f}` — {desc}", ] # A cold/failed feed yields 0.00 "Neutral" — flag it so a data outage # can't be mistaken for a genuinely neutral read. if not context.news_sentiment: lines.append("_no scored headlines yet — score defaults to 0_") return "\n".join(lines) def _news_report_sync() -> str: engine = NewsEngine(PlatformConfig().news) return _news_text(engine.get_fundamental_context(), engine) @_owner_only async def cmd_news(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: msg = await update.message.reply_text("⏳ Fetching news…") loop = asyncio.get_running_loop() try: text = await loop.run_in_executor(_executor, _news_report_sync) except Exception as exc: logger.exception("News fetch failed") text = f"❌ News fetch failed: {_escape_md(str(exc))}" await msg.edit_text(text, parse_mode=ParseMode.MARKDOWN) @_owner_only async def cmd_cache(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: dp = _get_shared_dp() if not dp._cache_timestamps: await update.message.reply_text("📭 Cache is empty — no data fetched yet.") return now = time.time() lines = ["📦 *Data Cache Status*\n"] for pair in SCAN_PAIRS: display = PAIR_INFO.get(pair, {}).get("display", pair) pair_lines = [] for tf in [Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4, Timeframe.D1]: cache_key = f"{pair}_{tf.value}" ts = dp._cache_timestamps.get(cache_key) if ts is None: continue age_min = (now - ts) / 60 df = dp._cache.get(cache_key) last_bar = df.index[-1].strftime("%H:%M") if df is not None and not df.empty else "?" fresh = "✅" if age_min < dp.CANDLE_SECONDS.get(tf, 3600) / 60 else "⚠️" pair_lines.append(f" {tf.value:<5} {fresh} fetched `{age_min:.0f}m` ago | last bar `{last_bar}`") if pair_lines: lines.append(f"*{display}*") lines.extend(pair_lines) await update.message.reply_text("\n".join(lines), parse_mode=ParseMode.MARKDOWN) @_owner_only async def cmd_clearcache(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: dp = _get_shared_dp() count = dp.clear_cache() await update.message.reply_text( f"🗑 Cache cleared ({count} entries wiped).\nNext scan will fetch fresh data from API.", parse_mode=ParseMode.MARKDOWN, ) @_owner_only async def cmd_autoscan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: chat_id = update.effective_chat.id if chat_id in _autoscan_chats: _autoscan_chats.discard(chat_id) if not _feed_active(): _stop_scheduler() msg = "⏹ Auto-scan stopped." else: msg = ("⏹ Auto-scan stopped for this chat.\n" "📢 Subscriber channel feed continues.") await update.message.reply_text(msg, parse_mode=ParseMode.MARKDOWN) return _autoscan_chats.add(chat_id) _start_scheduler(context.bot) next_m5 = _seconds_until_boundary(5, buffer=10) next_m15 = _seconds_until_boundary(15, buffer=10) next_h1 = _seconds_until_boundary(60, buffer=10) next_h4 = _seconds_until_boundary(240, buffer=10) await update.message.reply_text( "✅ *Auto-scan enabled*\n\n" f"Scanning {_PAIRS_LABEL} on M5/M15/H1/H4 at candle close.\n" f"📡 Live price monitoring active.\n" f"Next: M5 {next_m5 / 60:.0f}m | M15 {next_m15 / 60:.0f}m | " f"H1 {next_h1 / 60:.0f}m | H4 {next_h4 / 60:.0f}m\n\n" "Use /autoscan again to stop.", parse_mode=ParseMode.MARKDOWN, ) @_owner_only async def cmd_scan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: await update.message.reply_text( f"🔍 *{_PAIRS_LABEL} — Pick a timeframe*", parse_mode=ParseMode.MARKDOWN, reply_markup=main_menu_keyboard(), ) @_owner_only async def cb_scan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: query = update.callback_query await query.answer() btn = SCAN_LOOKUP.get(query.data) if not btn: return if btn["tf"] is None: await _cb_scan_all(query, btn["symbol"]) return display = PAIR_INFO[btn["symbol"]]["display"] await _safe_edit(query, f"⏳ Scanning {display} {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: {_escape_md(str(exc))}" await _safe_edit(query, text, reply_markup=main_menu_keyboard()) @_owner_only 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 _feed_active(): _stop_scheduler() await query.answer("Auto-scan stopped") else: await query.answer("DM alerts off — channel feed continues") 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, symbol: str = XAUUSD) -> None: display = PAIR_INFO[symbol]["display"] await _safe_edit(query, f"⏳ Scanning {display} across M5 / M15 / H1 / H4…") tfs = [Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4] tasks = [run_signal(symbol, tf) for tf in tfs] results = await asyncio.gather(*tasks, return_exceptions=True) reports = [] errors = [] for tf, result in zip(tfs, results): if isinstance(result, Exception): errors.append(f"❌ {tf.value}: {result}") else: reports.append(result) if reports: _update_level_cache(reports) lines = [ f"📊 *{display} — All Timeframes*", f"🕐 {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}", "", ] for r in reports: emoji = _SIGNAL_EMOJI.get(r.final_signal, "⬜") qe = _QUALITY_EMOJI.get(r.signal_quality, "") tf_label = r.primary_analysis.timeframe if r.primary_analysis else "?" sig = _signal_label(r.final_signal) regime = r.market_regime.replace("_", " ").title() lines.append( f"{emoji} *{tf_label}* {sig} {qe}{r.signal_quality} " f"Score: {r.final_score:+.1f} {r.final_confidence:.0f}%" ) lines.append(f" _{_escape_md(regime)}_ | {'✅ MTF' if r.mtf_agreement else '❌ MTF'}") if r.trade_setup and r.final_signal != SignalStrength.NEUTRAL: ts = r.trade_setup d = "📈" if ts.direction == "BUY" else "📉" lines.append( f" {d} {ts.direction} `{_fmt_price(ts.entry_price)}` " f"SL `{_fmt_price(ts.stop_loss)}` TP `{_fmt_price(ts.take_profit_1)}` " f"RR 1:{ts.risk_reward:.1f}" ) lines.append("") for err in errors: lines.append(err) lines.append("_Use buttons below for detailed single-TF view._") lines += ["", _DISCLAIMER] await _safe_edit(query, "\n".join(lines), reply_markup=main_menu_keyboard()) # ── Boot-time auto-start ────────────────────────────────────────────── async def _post_init(app: Application) -> None: """Register command menu and start scheduler on boot.""" await app.bot.set_my_commands([ ("start", "Welcome & scan buttons"), ("scan", "Quick scan buttons"), ("autoscan", "Toggle auto-scan on/off"), ("status", "Bot status, WS connection, live prices"), ("news", "XAU/USD net news sentiment score"), ("cache", "Show data cache age per pair/TF"), ("clearcache", "Wipe data cache, force fresh fetch"), ("help", "All commands"), ]) if CHAT_ID: _autoscan_chats.add(int(CHAT_ID)) if not _feed_active(): logger.info("Neither TELEGRAM_CHAT_ID nor TELEGRAM_CHANNEL_ID set — " "auto-scan disabled on boot. Use /autoscan in chat to enable.") return if CHANNEL_ID is not None: logger.info(f"Subscriber channel feed enabled: {CHANNEL_ID}") logger.info("Running initial scan to populate level cache and signal stats...") try: tasks = [ asyncio.wait_for(run_signal(sym, Timeframe.M15), timeout=30) for sym in SCAN_PAIRS ] results = await asyncio.gather(*tasks, return_exceptions=True) init_reports = [r for r in results if isinstance(r, SignalReport)] for i, r in enumerate(results): if isinstance(r, Exception): logger.error(f"Initial scan failed for {SCAN_PAIRS[i]}: {r}") if init_reports: _update_level_cache(init_reports) n_levels = sum( len(levels) for per_tf in _level_cache.values() for levels in per_tf.values() ) logger.info(f"Level cache populated: {n_levels} levels") except Exception: logger.exception("Initial scan failed — WS will wait for first candle scan") _start_scheduler(app.bot) next_m5 = _seconds_until_boundary(5, buffer=10) next_m15 = _seconds_until_boundary(15, buffer=10) next_h1 = _seconds_until_boundary(60, buffer=10) next_h4 = _seconds_until_boundary(240, buffer=10) logger.info( f"Auto-scan started (chats={sorted(_autoscan_chats)}, channel={CHANNEL_ID}) — " f"M5 {next_m5:.0f}s, M15 {next_m15:.0f}s, H1 {next_h1:.0f}s, H4 {next_h4:.0f}s" ) # Boot message goes to the owner DM only — the subscriber channel # receives nothing but signal and breach alerts. if OWNER_ID is not None: channel_line = "📢 Subscriber channel feed active.\n" if CHANNEL_ID else "" try: await app.bot.send_message( OWNER_ID, f"🤖 *ForexBot2 started*\n\n" f"Auto-scanning {_PAIRS_LABEL} on M5/M15/H1/H4.\n" f"📡 Live price monitoring active.\n" f"{channel_line}" f"Next: M5 {next_m5 / 60:.0f}m | M15 {next_m15 / 60:.0f}m | " f"H1 {next_h1 / 60:.0f}m | H4 {next_h4 / 60:.0f}m\n\n" f"/autoscan to stop", parse_mode=ParseMode.MARKDOWN, ) except Exception: logger.exception("Failed to send boot message") # ── Graceful shutdown ───────────────────────────────────────────────── async def _post_shutdown(app: Application) -> None: logger.info("Shutting down ForexBot2...") _stop_scheduler() _executor.shutdown(wait=False) logger.info("Shutdown complete") # ── Entry point ─────────────────────────────────────────────────────── async def _on_error(update: object, context: ContextTypes.DEFAULT_TYPE) -> None: """python-telegram-bot routes every unhandled handler exception here; without a registered error handler PTB only half-logs them.""" logger.error( "Unhandled error in Telegram handler (update=%s)", getattr(update, "update_id", update), exc_info=context.error, ) def _install_excepthooks() -> None: """Route uncaught exceptions from the main thread and worker threads into the rotating log file instead of bare stderr, so a crash right before exit is still recorded.""" def hook(exc_type, exc, tb): if issubclass(exc_type, KeyboardInterrupt): sys.__excepthook__(exc_type, exc, tb) return logger.critical("Uncaught exception", exc_info=(exc_type, exc, tb)) def thread_hook(args): if issubclass(args.exc_type, SystemExit): return name = args.thread.name if args.thread else "?" logger.critical( f"Uncaught exception in thread {name}", exc_info=(args.exc_type, args.exc_value, args.exc_traceback)) sys.excepthook = hook threading.excepthook = thread_hook def main() -> None: if not TOKEN or not CHAT_ID: raise SystemExit( "telegram_bot: missing Telegram credentials. Set telegram.token " "and telegram.chat_id in config.json, or the TELEGRAM_TOKEN and " "TELEGRAM_CHAT_ID environment variables.") _install_excepthooks() logger.info("Starting ForexBot2 Telegram bot…") app = ( Application.builder() .token(TOKEN) .post_init(_post_init) .post_shutdown(_post_shutdown) .build() ) app.add_handler(CommandHandler("start", cmd_start)) app.add_handler(CommandHandler("scan", cmd_scan)) app.add_handler(CommandHandler("help", cmd_help)) app.add_handler(CommandHandler("status", cmd_status)) app.add_handler(CommandHandler("news", cmd_news)) app.add_handler(CommandHandler("autoscan", cmd_autoscan)) 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_")) app.add_error_handler(_on_error) logger.info("Bot running — press Ctrl+C to stop") app.run_polling(allowed_updates=Update.ALL_TYPES) if __name__ == "__main__": main()