#!/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, timedelta, timezone from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set import pandas as pd 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, violent_watch_enabled, ) from providers import BaseDataProvider, make_data_provider from providers.base import _is_daily_maintenance from agent_panel import attach_panel, panel_health from llm_client import get_llm_client from news_engine import FundamentalContext, NewsEngine from pair_signals import PAIR_PRODUCERS from signal_aggregator import SignalAggregator, SignalReport from signal_log import gate_reason from violent_watch import attach_watch, should_send_headsup if TYPE_CHECKING: # type hints only — ViolentWatch is never needed at runtime here from violent_watch import ViolentWatch _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) # One file per (server-local) day × 30 backups = 30 days retention. # 2026-09-08: was 3-hourly × 240 — a 4-day review meant 18 files. The old # `telegram.log.YYYY-MM-DD_HH` files do not match the daily handler's # cleanup pattern and need one manual delete on the live box. # 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="midnight", backupCount=30, encoding="utf-8", ) _file_handler.setFormatter(_log_formatter) # Console only at IMPORT: `python test_all.py` imports this module, and an # import-time file handler wrote fixture alerts (2027 dates, 3300.00 prices) # into logs/telegram.log on whatever box ran the suite. main() attaches # _file_handler for real bot runs. logging.basicConfig(level=getattr(logging, _APP.log_level, logging.INFO), handlers=[_console_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], } _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}" # ── 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: # Pattern identifiers use underscores, which legacy Telegram Markdown # can misparse even when escaped inside this italicized action line. action = report.action_recommendation.replace("_", " ") lines.append(f"_{_escape_md(action)}_") 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)}") # Display-only violent-move watch (violent_watch.attach_watch stamps # it post-hoc; it is empty unless the feature is enabled). if report.watch_line: extra.append(_escape_md(report.watch_line)) if report.news_summary: extra.append(f"📰 {_escape_md(report.news_summary)}") if report.panel_verdict: mark = {"AGREE": "✅", "OPPOSE": "⛔", "NEUTRAL": "➖"}.get( report.panel_verdict, "") panel_label = ("AI macro view" if report.panel_verdict_source == "macro" else "AI Panel") extra.append( f"🤖 {panel_label} {mark} *{_escape_md(report.panel_verdict)}* " f"({report.panel_confidence:.0f}%) — " f"{_escape_md(report.panel_rationale)}") if report.panel_takes: votes = " · ".join( f"{t['persona']}: {t['stance']}" for t in report.panel_takes) extra.append(f" _{_escape_md(votes)}_") if extra: lines += [""] + extra 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 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; every scanned symbol is registered there # (SCAN_PAIRS is validated against PAIR_INFO at import). config = PAIR_PRODUCERS[symbol](timeframe, confirmation_tfs) aggregator = SignalAggregator(config) aggregator.data_provider = _get_shared_dp() return aggregator.analyze_pair(symbol=symbol, log_source="telegram") async def run_signal( symbol: str, timeframe: Timeframe, confirmation_tfs: Optional[List[Timeframe]] = None, ) -> SignalReport: loop = asyncio.get_running_loop() report = await loop.run_in_executor( _executor, _run_signal_sync, symbol, timeframe, confirmation_tfs ) _remember_scan(symbol, timeframe, report) return report # Most recent report per (symbol, timeframe) keyed by the closed-bar epoch # it was computed for. Closed-bar inputs are identical within a bar, so a # rescan inside the same bar (the WS level-breach handler fired up to # ~5 min apart all day, logging 3.5 M15 rows per candle — 2026-09 audit) # can reuse it instead of re-fetching, re-scoring and re-logging. _last_scan: Dict[str, tuple] = {} _TF_SECONDS_CACHE = { Timeframe.M1: 60, Timeframe.M5: 300, Timeframe.M15: 900, Timeframe.M30: 1800, Timeframe.H1: 3600, Timeframe.H4: 14400, Timeframe.D1: 86400, } def _bar_epoch(tf: Timeframe, now: Optional[float] = None) -> int: """Epoch of the most recent candle boundary for ``tf`` (UTC-anchored, the same grid _seconds_until_boundary sleeps to).""" secs = _TF_SECONDS_CACHE.get(tf, 900) t = int(time.time() if now is None else now) return t - t % secs def _remember_scan(symbol: str, tf: Timeframe, report: SignalReport) -> None: # File the report under the bar it actually ANALYSED, not the wall # clock at completion: a scan that straddles a boundary would land # under the new bar, so the boundary loop's own scan of that bar # found no cache entry and re-fetched/re-scored/re-logged the same # closed bar (15 duplicate M15 rows in five days — 2026-09-13 audit). epoch = _bar_epoch(tf) if report.primary_analysis is not None: try: # report.timestamp is the primary bar's OPEN label; the bar # closes (= the boundary this scan belongs to) one TF later. epoch = (int(pd.Timestamp(report.timestamp).timestamp()) + _TF_SECONDS_CACHE.get(tf, 900)) except (ValueError, TypeError): pass # unparseable label → wall-clock boundary, as before _last_scan[f"{symbol}_{tf.value}"] = (epoch, report) async def _cached_or_scan( symbol: str, timeframe: Timeframe, confirmation_tfs: Optional[List[Timeframe]] = None, ) -> SignalReport: """run_signal, unless this bar already has a report for the key.""" cached = _last_scan.get(f"{symbol}_{timeframe.value}") if cached is not None and cached[0] == _bar_epoch(timeframe): logger.info( f"{symbol} {timeframe.value}: reusing this bar's scan — no rescan") return cached[1] return await run_signal(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 _is_alert_candidate(report: SignalReport, tf_label: str) -> bool: """An alert is a trade call: non-neutral, grade at/above the threshold, AND a trade setup. Until 2026-09-08 the setup was not required, so a signal every generator had declined (spread gate, RR floor, swing geometry) still went out — 07:50 UTC that day sent a B-grade SELL with no setup under a "costs would eat the edge" warning. Skips are logged at INFO so the decision stays visible in telegram.log.""" if report.final_signal == SignalStrength.NEUTRAL: return False if not _passes_quality(report.signal_quality): return False if report.trade_setup is None: why = gate_reason(report.warnings, report.cleaned_warnings, report.signal_quality, False) logger.info( f"alert skipped: {report.display_name} {tf_label} " f"{report.final_signal.name} grade {report.signal_quality} " f"has no trade setup ({why})") return False return True def _alert_sent_line(report: SignalReport, tf_label: str, delivered_to: int, kind: str = "alert") -> str: """One INFO line per alert that reached a chat. Only failures were logged before 2026-09-08, so that day's log review had to reconstruct what was sent from panel rows and dedup skips.""" s = report.trade_setup setup = (f"{s.direction} entry {s.entry_price:.2f} sl {s.stop_loss:.2f} " f"tp1 {s.take_profit_1:.2f} [{s.strategy}]" if s else "no setup") rowid = getattr(report, "_signal_log_rowid", None) return (f"{kind} sent: {report.display_name} {tf_label} " f"{report.final_signal.name} grade {report.signal_quality} — " f"{setup} — signal-log row #{rowid} → {delivered_to} chat(s)") _last_alert_quality: Dict[str, str] = {} # key → grade of last ALERTED signal def _alert_fingerprint(report: SignalReport) -> tuple: """Alert identity for dedup: signal and 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. The grade is deliberately NOT part of the identity (2026-09 audit: an A↔B flicker on an unchanged idea re-sent the full alert ~4× in 45 min); a grade UPGRADE re-alerts via _should_alert instead.""" setup = report.trade_setup return ( report.final_signal.name, 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 # Same idea at a HIGHER grade than last alerted → worth re-sending; # a downgrade or flicker back is not. last_q = _last_alert_quality.get(key) if last_q is not None and (_QUALITY_RANK.get(report.signal_quality, 0) > _QUALITY_RANK.get(last_q, 0)): 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() _last_alert_quality[key] = report.signal_quality 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 async def _broadcast(bot, text: str) -> int: """Send `text` to every signal target; returns how many chats got it. Shared by the scan and breach paths so the delivery rule cannot drift between them.""" delivered = 0 for chat_id in _signal_targets(): if await _send_alert(bot, chat_id, text): delivered += 1 return delivered # ── 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 def _closed_bar_in_settlement(interval_minutes: int, now: Optional[datetime] = None) -> bool: """True when the bar that just closed (label = now - interval) falls in gold's daily settlement pause — no such bar exists, so a scan would only re-read the previous bar (61 stale rows in the 2026-09 audit DB).""" label_time = ((now or datetime.now(timezone.utc)) - timedelta(minutes=interval_minutes)) return _is_daily_maintenance(label_time) # ── Violent-move watch (display-only) ──────────────────────────────── # Per-process heads-up budget: {"day": date, "count": int, "sent": {...}}. _watch_state: Dict[str, Any] = {} def _format_headsup(report: SignalReport, w: "ViolentWatch") -> str: """Standalone owner heads-up. Advisory only — it never asks anyone to touch an open trade: every protective action tested lost money (sizing_campaign_2026-09.md), so the stop stays the protection.""" lines = [ f"🌪 *Heads-up — {_escape_md(report.display_name)}*: " f"violent-move conditions for the next hour.", _escape_md(" · ".join(w.triggers)), ] if w.likely_direction == "down": lines.append("If a big move comes, it has usually gone DOWN from " "readings like these.") elif w.likely_direction == "up": lines.append("If a big move comes, it has usually gone UP from " "readings like these.") lines.append( "If you have a trade open, expect big bars. Your stop is the " "protection — past tests show closing or tightening early loses " "money.") return "\n".join(lines) async def _scan_timeframe( bot, tf: Timeframe, confirmation_tfs: List[Timeframe], ) -> None: """Scan all pairs for one timeframe and broadcast results.""" # _cached_or_scan, not run_signal: the WS breach handler may already # have scanned this closed bar (and vice versa) — one fetch/score/log # per bar per TF (2026-09-13 audit). tasks = [_cached_or_scan(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) # Violent-move watch: DISPLAY-ONLY. Runs before the alert loop so # the stamped line is present when format_signal_message renders, # and can never break an alert (whole block is best-effort). try: vw_settings = get_app_config().violent_watch_settings if violent_watch_enabled(): now = datetime.now(timezone.utc) for report in reports: w = attach_watch(report, tf, vw_settings, now) if not (tf == _HEADSUP_TF and vw_settings.get("headsup", True) and OWNER_ID is not None): continue min_level = vw_settings.get("headsup_min_level", "hot") if w.level == "hot" or (min_level == "watch" and w.level == "watch"): ok, why = should_send_headsup( _watch_state, now, w.kinds, int(vw_settings.get("max_headsup_per_day", 12))) if ok: if await _send_alert(bot, OWNER_ID, _format_headsup(report, w)): logger.info( f"violent watch: heads-up sent ({w.level}: " f"{', '.join(w.kinds)}) signal-log row " f"#{getattr(report, '_signal_log_rowid', None)}") else: logger.warning( "violent watch: heads-up send failed " "(this hour's slot is consumed)") else: logger.info( f"violent watch: heads-up skipped ({why})") else: logger.debug( f"violent watch: {report.symbol} {tf.value} " f"level={w.level} (min {min_level}) — no heads-up") except Exception: logger.exception("violent watch failed") # Same gate as the breach path: non-neutral, grade ≥ threshold, and a # trade setup (_is_alert_candidate logs the no-setup skips). candidates = [r for r in reports if _is_alert_candidate(r, tf.value)] 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 # Panel AFTER the quality gate + dedup decision so no LLM # spend on suppressed alerts. Runs on the scan executor to # keep the event loop free; this key's alert lock is held # for the panel's wall time (≤~25 s), which only delays a # same-key duplicate that _should_alert would drop anyway. await asyncio.get_running_loop().run_in_executor( _executor, attach_panel, report) detail = format_signal_message(report) delivered = await _broadcast(bot, detail) # 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) logger.info(_alert_sent_line(report, tf.value, delivered)) # ── 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 # A breach landing inside this many seconds after an M15 boundary waits for # the candle loop's scan (boundary+10s) and reuses it — 2026-09-13 audit. _BREACH_DEFER_SECONDS = 20.0 _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: # A tick a second after an M15 boundary would scan a bar the feed # may still be lagging on, and the candle loop (boundary+10s) would # then scan the same bar again. Wait for the loop and reuse it. elapsed = time.time() - _bar_epoch(Timeframe.M15) if elapsed < _BREACH_DEFER_SECONDS: logger.debug("deferring breach rescan to the boundary scan") await asyncio.sleep(_BREACH_DEFER_SECONDS - elapsed) report = await _cached_or_scan(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 not _is_alert_candidate(report, Timeframe.M15.value): 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 # Same post-gate panel attach as the auto-scan path (see # _scan_timeframe for the spend/lock rationale). await asyncio.get_running_loop().run_in_executor( _executor, attach_panel, report) header = ( f"⚡ *LEVEL BREAK* — {_escape_md(display)} crossed " f"*{_escape_md(label)}* @ `{_fmt_price(price)}`\n\n" ) detail = header + format_signal_message(report) sent = await _broadcast(bot, detail) if sent: _commit_alert(key, report) logger.info(_alert_sent_line(report, Timeframe.M15.value, sent, kind="breach alert")) 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 if _closed_bar_in_settlement(interval_minutes): logger.info( f"{label} auto-scan: skipped — bar label inside the " f"daily settlement pause") continue logger.info(f"{label} auto-scan: firing") try: await _scan_timeframe( bot, tf, TF_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 ] # The violent-move heads-up is evaluated on the FASTEST scanned loop # only — it is a "next hour" warning, so once per fastest candle is the # most frequent it can be, and every slower loop would just repeat it. _HEADSUP_TF: Optional[Timeframe] = min( _APP.scan_timeframes, key=lambda tf: _TF_INTERVAL_MINUTES.get(tf, 10 ** 6), default=None, ) _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 _fmt_age(ts: Optional[float]) -> str: """Compact age like '45s', '12m', '3.2h' for status lines.""" if not ts: return "?" s = time.time() - ts if s < 90: return f"{s:.0f}s" if s < 5400: return f"{s / 60:.0f}m" return f"{s / 3600:.1f}h" def _news_health_lines() -> List[str]: """News / calendar / DeepSeek source health for /status. Reads the class-level `NewsEngine._source_health` written by the scan-path fetches plus the LLM client's health stamps — display only, never triggers a fetch. ⚠️ marks a degraded dependency; details live in the rotating log (transition-logged WARNINGs).""" lines = ["\n*News & AI:*"] # Snapshot: a scan thread may insert a source's FIRST entry while we # iterate (dict-changed-size RuntimeError); entry mutation is safe. for _ in range(3): try: sh = dict(NewsEngine._source_health) break except RuntimeError: continue else: sh = {} if not sh: lines.append(" _no news cycle yet (fills after first scan)_") else: cal = sh.get("calendar") ff = sh.get("calendar:forexfactory") if cal: if NewsEngine.source_is_failing(cal): lines.append(f" ⚠️ Calendar: unavailable — {cal['detail']}" f" ({_fmt_age(cal['err'])})") elif NewsEngine.source_is_failing(ff): lines.append(f" ⚠️ Calendar: {cal['extra']} — ForexFactory" f" down: {ff['detail']} ({_fmt_age(ff['err'])})") else: lines.append(f" ✅ Calendar: {cal['extra']}" f" ({_fmt_age(cal['ok'])} ago)") heads = sh.get("headlines") if heads: failing = NewsEngine.source_is_failing(heads) mark = "⚠️" if failing else "✅" detail = f" — {heads['detail']}" if failing else "" ts = heads["err"] if failing else heads["ok"] lines.append(f" {mark} Headlines: {heads['extra']}{detail}" f" ({_fmt_age(ts)} ago)") for name, entry in sorted(sh.items()): if name.startswith("rss:") and NewsEngine.source_is_failing(entry): lines.append(f" ⚠️ {name[4:]}: {entry['detail']}" f" ({_fmt_age(entry['err'])})") client = get_llm_client() if not client.available: lines.append(" DeepSeek: off — keyword scoring only") else: # The client's counter only resets on the first call of a new # UTC day — show 0 (not yesterday's total, or a phantom # cap-exhausted warning) until that reset happens. today = datetime.now(timezone.utc).date() used = client._count if client._count_day == today else 0 cap = client._daily_cap scored = len(NewsEngine._llm_sent_cache) err = client.last_error if used >= cap: lines.append(f" ⚠️ DeepSeek: daily cap {cap} exhausted —" " keyword fallback until UTC midnight") elif err and err[0] > client.last_ok: lines.append(f" ⚠️ DeepSeek: last call failed ({err[1]}," f" {_fmt_age(err[0])}) — keyword fallback") else: lines.append(f" ✅ DeepSeek: {used}/{cap} calls today," f" {scored} headlines scored") if get_app_config().agent_panel_settings.get("enabled", False): ph = panel_health() lines.append(f" 🤖 AI panel: {ph['used_today']}/{ph['cap']}" " panels today") return lines 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.extend(_news_health_lines()) 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.") # Only a real bot run writes the rotating log file: tests import this # module (an import-time handler logged their fixtures into # logs/telegram.log) and one of them calls main() to assert the refusal # above, so the attach sits after that guard — 2026-09-13 audit. if _file_handler not in logging.getLogger().handlers: logging.getLogger().addHandler(_file_handler) _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()