ForexBot2/telegram_bot.py

#!/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=<token> TELEGRAM_CHAT_ID=<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 exposure import attach_exposure, size_shared
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, last_alerted_bars, mark_alerted,
                        unmark_alerted)
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)
# ONE worker for an alert's database steps (refresh the "already open" line,
# stamp the row as alerted, undo the stamp). A single worker is the mutual
# exclusion: one alert's read and write run back to back, in arrival order,
# so of two alerts sent at the same candle close the second counts the
# first. It also keeps SQLite off the event loop (the logger's lock can be
# held for seconds by the outcome labeller) and never queues behind the
# scan pool above. Not an asyncio.Lock: on Python 3.9 one created at
# import binds to a loop that is not the running one.
_alert_db_executor = ThreadPoolExecutor(
    max_workers=1, thread_name_prefix="alert-db")

# ── 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).

_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 _panel_odds_word(p: float, usual: Optional[float]) -> str:
    """Plain-language headline: the estimate against the usual odds for
    this kind of setup (8 points either way), or against 60 / 40 when no
    base rate is known. Never a verdict word."""
    if usual is not None:
        d = p - usual
        if d >= 8:
            return "the odds look better than usual"
        if d <= -8:
            return "the odds look worse than usual"
        return "the odds look about usual"
    if p >= 60:
        return "the odds look good"
    if p <= 40:
        return "the odds look poor"
    return "the odds look about even"


def _panel_lines(report: SignalReport) -> list[str]:
    """The AI review for a new trader (owner rule 2026-09-26: simple and
    clear). One headline in everyday words, the chance "in 100" against
    the usual odds for this kind of setup, one reason for and one
    against, and a plain warning. Used whenever the verdict carries a
    probability (the "mean" source, and "macro" since 2026-09-26); no
    verdict word, no percentages that are not probabilities, no persona
    names unless several personas ran (research configurations)."""
    p = float(report.panel_probability or 0.0)
    base = (report.panel_context or {}).get("base_rate") or {}
    usual = base.get("win_rate")
    head = f"🤖 AI check: {_panel_odds_word(p, usual)}."
    if report.panel_rationale:
        head += f" {_escape_md(report.panel_rationale)}"
    chance = f"   Chance it pays off: about *{p:.0f} in 100*"
    if usual is not None:
        chance += f" (usual for this kind of setup: {usual:.0f} in 100)"
    out = [head, chance + "."]
    takes = [t for t in report.panel_takes if t.get("p_win") is not None]
    lead = max(takes, key=lambda t: abs(float(t["p_win"]) - 50.0), default=None)
    if lead is not None:
        if lead.get("reason_for"):
            out.append(f"   For: {_escape_md(lead['reason_for'])}")
        if lead.get("reason_against"):
            out.append(f"   Against: {_escape_md(lead['reason_against'])}")
    if len(takes) > 1:      # one persona: the headline says it all
        votes = " · ".join(f"{t['persona']} {float(t['p_win']):.0f}%" for t in takes)
        out.append(f"   _{_escape_md(votes)}_")
    out.append("   This check is an experiment. Do not trade on it alone.")
    return out


def _book_rate_line(report: SignalReport) -> Optional[str]:
    """One line of validated-book statistics for the setup's class
    (timeframe × strategy × grade × direction, falling back to wider
    classes exactly as the agent panel does). None when no table or no
    primary analysis is available; never raises."""
    ts = report.trade_setup
    if ts is None or report.primary_analysis is None:
        return None
    try:
        from agent_panel import _base_rate
        rate = _base_rate(report.primary_analysis.timeframe,
                          getattr(ts, "strategy", None) or "GENERIC",
                          report.signal_quality, ts.direction)
    except Exception:
        logger.debug("book rate line unavailable", exc_info=True)
        return None
    if not rate or "win_rate" not in rate or "mean_r" not in rate:
        return None
    scope = {"class": "this setup class",
             "timeframe_direction": f"{ts.direction} setups on this timeframe",
             "timeframe": "this timeframe"}.get(rate.get("level"), "this class")
    return (f"📚 Book ({scope}): {rate['n']} trades, "
            f"{rate['win_rate']:.0f}% winners, avg {rate['mean_r']:+.2f}R per trade")


def _plain_words_line(report: SignalReport) -> str:
    """Grade first, then the two raw numbers in words (owner request
    2026-09-27): "Direction: strong buy · Reliability: good · timeframes
    agree" replaces "Score +38.2 · Conf 62% · MTF ✅". The words come from
    main.plain_words so every surface says the same thing."""
    from main import direction_words, reliability_words  # display helper, no side effects
    agree = "timeframes agree" if report.mtf_agreement else "timeframes disagree"
    return (f"📐 Direction: *{direction_words(report.final_score)}* · "
            f"Reliability: *{reliability_words(report.final_confidence)}* · {agree}")


def _daily_trend_line(report: SignalReport) -> str:
    """"🧭 Daily trend: rising — this buy goes with it" (main.daily_trend_words,
    the same words on every surface), or "" when the scan carries no daily
    trend. On M15 a plan against a clear daily trend never reaches an
    alert (the core removes it); on the other timeframes the line is
    information only."""
    from main import daily_trend_words  # display helper, no side effects
    words = daily_trend_words(report)
    return f"🧭 {_escape_md(words)}" if words else ""


def format_signal_message(report: SignalReport, channel: bool = False) -> str:
    """The alert / scan text. `channel=True` builds the copy for every
    audience but the owner's own chat (the subscriber channel, a group
    the bot was switched on in): identical, except that the lot-size line
    (which reveals the owner's risk per trade) is left out unless
    config.json sets trade_size.telegram_channel."""
    sig = report.final_signal
    emoji = _SIGNAL_EMOJI.get(sig, "⬜")
    qe = _QUALITY_EMOJI.get(report.signal_quality, "")
    label = _signal_label(sig)
    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} *Grade {report.signal_quality} · {label} · {report.display_name} · {tf}*  {qe}",
        f"🕐 {now} · 💰 {_fmt_price(report.current_price)}",
    ]
    live = _live_price_line(report.symbol)
    if live:
        lines.append(live)
    lines += [
        _plain_words_line(report),
        f"📊 Regime: {regime_str} · Vol: {vol_str}",
    ]
    trend_line = _daily_trend_line(report)
    if trend_line:
        lines.append(trend_line)
    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
        # Order and wording (2026-09-26): the trail is the tested exit
        # (the backtester never exits at TP1 — `_simulate_trailing` reads
        # only the stop and the trailing rule), so it is printed first
        # and TP1/TP2 are labelled guides with their R. The RR line is
        # the PLANNED TP1 ratio, not the expected payoff: the validated
        # books earn their R from the trailed runners (M5 average win
        # +1.03R vs loss −0.76R; M15 +0.87R vs −0.93R).
        risk_dist = abs(ts.entry_price - ts.stop_loss)
        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 = 1R)",
        ]
        # Lot size for the owner's risk limit and the alerted plans still
        # open in this direction (exposure.attach_exposure stamps both
        # post-hoc; empty unless configured / unless something is open).
        # The size is private: the channel copy carries it only when
        # config.json says so.
        if report.size_line and (not channel or size_shared("telegram_channel")):
            lines.append(f"📏 {_escape_md(report.size_line)}")
        if report.open_line:
            lines.append(f"🧱 {_escape_md(report.open_line)}")

        # 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"🔄 Exit rule{tag} (tested): once price reaches "
                f"`{_fmt_price(arm_price)}` (+{arm_r:.1f}R), trail the SL "
                f"${trail_dist:.2f} behind the best price, moving it at "
                f"candle close only. Otherwise the SL is the exit."
            )

        tp1_r = (abs(ts.take_profit_1 - ts.entry_price) / risk_dist
                 if risk_dist > 0 else 0.0)
        lines.append(
            f"🎯 TP1 guide `{_fmt_price(ts.take_profit_1)}`  "
            f"(${reward_usd:.2f} / {ts.reward_pips:.0f}p = {tp1_r:.1f}R)"
        )
        if ts.take_profit_2:
            tp2_r = (abs(ts.take_profit_2 - ts.entry_price) / risk_dist
                     if risk_dist > 0 else 0.0)
            lines.append(
                f"🎯 TP2 guide `{_fmt_price(ts.take_profit_2)}`  ({tp2_r:.1f}R)")

        lines.append(f"⚖️ Planned TP1 RR 1:{ts.risk_reward:.1f} (guide only — the trail is the exit)")

        # What this setup class actually paid in the validated books
        # (agent_panel_base_rates.json, the table the AI estimate is
        # anchored to): the honest expectation for a discretionary reader.
        rate_line = _book_rate_line(report)
        if rate_line:
            lines.append(rate_line)

    extra: list[str] = []
    mtf = _mtf_line(report)
    if mtf:
        extra.append(f"📈 Other timeframes: {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 and (report.panel_probability or 0) > 0:
        extra.extend(_panel_lines(report))
    elif 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()
    report = aggregator.analyze_pair(symbol=symbol, log_source="telegram")
    # Display-only lot-size and "already open" lines (never raises; one
    # read-only query on signals.db). Stamped here, on the scan executor,
    # for the manual scans. An ALERT refreshes them just before it is sent
    # (_deliver_alert): every timeframe scans at the same candle close, so
    # the line stamped here cannot know about an alert going out beside it.
    return attach_exposure(report, surface="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
_last_alert_bar: Dict[str, pd.Timestamp] = {}   # key → bar label of last ALERTED setup
# Entry cooldown, in primary bars, after an ALERTED setup (2026-09-26):
# the backtester's `_apply_cooldown` (cooldown_bars=5) never enters within
# 5 bars of an accepted entry, so every validated book counts one trade per
# move. Live had no such rule: the entry-moved re-alert below re-sent the
# same idea on consecutive bars while price ran (2026-09-15 08:00–08:45,
# 2026-09-18 06:00–06:30, 2026-09-22 18:00–19:40 …); over 2026-09-04 →
# 2026-09-25 those re-entries were 42 of 99 unique setups and −25.1R of
# the −37.8R raw advisory book (backtest/research/rr_and_sell_campaign_
# 2026-09-26.md). Must equal backtest.cooldown_bars in gpu_backtest.py.
_ALERT_COOLDOWN_BARS = 5


def _key_bar_seconds(key: str) -> Optional[int]:
    """Candle length for an alert key ("XAU/USD_15min" → 900). None when
    the key carries no known timeframe (test keys), which disables the
    bar cooldown for that key."""
    tf_value = key.rsplit("_", 1)[-1]
    for tf, secs in BaseDataProvider.CANDLE_SECONDS.items():
        if tf.value == tf_value:
            return secs
    return None


def _bar_label_ts(label: Any) -> Optional[pd.Timestamp]:
    """A bar label as a UTC timestamp; None when it cannot be parsed."""
    try:
        ts = pd.Timestamp(label)
    except Exception:
        return None
    if pd.isna(ts):
        return None
    return ts.tz_localize("UTC") if ts.tzinfo is None else ts.tz_convert("UTC")


def _report_bar_ts(report: SignalReport) -> Optional[pd.Timestamp]:
    """The report's primary bar label as a UTC timestamp; None when it
    cannot be parsed (the cooldown then does not apply)."""
    return _bar_label_ts(report.timestamp)


def _primary_bars_since(key: str, report: SignalReport, last: pd.Timestamp,
                        now_bar: pd.Timestamp) -> Optional[int]:
    """How many primary bars lie after `last`, up to and including the
    report's own bar, in the frame the scan analysed (report._frames):
    the backtester's position difference `pos - last_entry_pos`. None when
    the report carries no frame for the key's timeframe or its bar is not
    in it; the caller then falls back to clock time."""
    try:
        tf = Timeframe(key.rsplit("_", 1)[-1])
        df = (getattr(report, "_frames", None) or {}).get(tf)
        if df is None or len(df) == 0:
            return None
        idx = df.index
        if getattr(idx, "tz", None) is None:
            idx = idx.tz_localize("UTC")
        if now_bar not in idx:
            return None
        return int(((idx > last) & (idx <= now_bar)).sum())
    except Exception:
        return None


def _inside_alert_cooldown(key: str, report: SignalReport) -> bool:
    """True while this key sits within _ALERT_COOLDOWN_BARS primary bars
    of its last ALERTED setup's bar (the backtester's entry cooldown).

    Bars are COUNTED in the scan's own primary frame, like the
    backtester's bar positions. Clock time is not the same thing: across
    the daily settlement pause or the weekend an hour or two days pass
    while no bar prints, and a clock rule released the 22:00 UTC bar that
    no book ever enters (six M15 and three M5 entries in three years,
    2026-10-01 review). Clock time remains the fallback for a report
    without frames."""
    last = _last_alert_bar.get(key)
    if last is None or report.trade_setup is None:
        return False
    secs = _key_bar_seconds(key)
    now_bar = _report_bar_ts(report)
    if secs is None or now_bar is None:
        return False
    if now_bar < last:
        return False
    bars = _primary_bars_since(key, report, last, now_bar)
    if bars is not None:
        return bars < _ALERT_COOLDOWN_BARS
    return (now_bar - last).total_seconds() < _ALERT_COOLDOWN_BARS * secs


def _restore_alert_cooldowns(bars: Dict[Any, str]) -> int:
    """Seed the cooldown memory from signal_log.last_alerted_bars() after a
    start-up: {(symbol, timeframe): bar label} of the newest alerted row.
    Without it every restart forgot the last alert and the next bar could
    alert again inside the 5 bars. Only the cooldown is restored; the
    dedup fingerprints start empty, as before. A key already in memory is
    left alone. Returns how many keys were seeded."""
    restored = 0
    for (symbol, timeframe), label in (bars or {}).items():
        bar = _bar_label_ts(label)
        key = f"{symbol}_{timeframe}"
        if bar is None or key in _last_alert_bar:
            continue
        _last_alert_bar[key] = bar
        restored += 1
    return restored



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.
    """
    if _inside_alert_cooldown(key, report):
        # Same rule as the backtester: no second entry within 5 bars of
        # the last one, whatever the grade or the entry drift did.
        logger.info(
            f"alert skipped: {key} {report.final_signal.name} grade "
            f"{report.signal_quality} bar {report.timestamp} is inside the "
            f"{_ALERT_COOLDOWN_BARS}-bar entry cooldown (backtester rule)")
        return False
    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:
    # In-memory dedup state only. The signal-log stamp of the alert is
    # written (and, for an undelivered alert, undone) by _deliver_alert on
    # the alert-db worker: a database write does not belong on the event
    # loop this function runs on.
    _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)
        )
        bar = _report_bar_ts(report)
        if bar is not None:
            _last_alert_bar[key] = bar
    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


def _is_owner_chat(chat_id: Any) -> bool:
    """True for the owner's own chat with the bot, the one audience of the
    private lot-size line. The owner lock is per USER, so /autoscan or a
    scan button used inside a group makes that group a target: it is not
    the owner's chat and gets the copy without the size."""
    return OWNER_ID is not None and chat_id == OWNER_ID


async def _broadcast(bot, text: str, channel_text: Optional[str] = None) -> 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. `channel_text`, when given, is what every target
    except the owner's own chat receives instead (the copy without the
    owner's private lot-size line)."""
    delivered = 0
    for chat_id in _signal_targets():
        body = (text if channel_text is None or _is_owner_chat(chat_id)
                else channel_text)
        if await _send_alert(bot, chat_id, body):
            delivered += 1
    return delivered


def _stamp_exposure_and_mark(report: SignalReport) -> Optional[str]:
    """On the alert-db worker: refresh the lot-size and "already open"
    lines, then stamp the report's signal-log row as alerted. Read first,
    write second, nothing in between: the next alert on the worker sees
    this stamp. Returns the stamp to undo if no chat receives the alert
    (None when nothing was written). Never raises."""
    attach_exposure(report, surface="telegram")
    return mark_alerted(report)


def _undo_cancelled_stamp(report: SignalReport, done: Any) -> None:
    """Done-callback of the stamp job when the alert's task was cancelled
    while that job was queued or running (a shutdown, or auto-scan switched
    off, at the wrong moment). Cancelling a task cannot stop a job the
    worker has already started, so its stamp lands although nothing will be
    sent: clear it. Runs on the alert-db worker right after the job, also
    when the pool has already been told to shut down. Never raises."""
    try:
        if done.cancelled():
            return                      # never ran: nothing was written
        stamp = done.result()
        if stamp:
            unmark_alerted(report, stamp)
    except Exception:
        logger.warning("alert: could not clear the stamp of a cancelled "
                       "alert", exc_info=True)


async def _undo_alert_mark(report: SignalReport, stamp: Optional[str]) -> None:
    """The alert reached no chat: clear the stamp this attempt wrote (and
    only that one), so the row does not count as a plan the reader holds.
    Never raises."""
    if not stamp:
        return
    try:
        await asyncio.get_running_loop().run_in_executor(
            _alert_db_executor, unmark_alerted, report, stamp)
    except Exception:
        logger.warning("alert: could not clear the alerted stamp of an "
                       "undelivered alert", exc_info=True)


async def _deliver_alert(bot, key: str, report: SignalReport, tf_label: str,
                         header: str = "", kind: str = "alert") -> int:
    """Send one alert: panel → exposure lines + alerted stamp → both
    copies → commit. The ONE delivery path of the candle scan and the
    level break; the caller holds _alert_lock(key) and has passed
    _should_alert. Returns how many chats received it.

    The row is stamped BEFORE the send so that an alert going out at the
    same candle close on another timeframe already counts this plan in its
    "already open" line (2026-10-01: stamped after the send, two alerts at
    one boundary never counted each other). A send that reaches no chat
    undoes the stamp and commits nothing, so the setup stays eligible for
    the next scan; so does a cancellation before the send has started.
    Residue: if this send fails completely after another alert has already
    counted it, that alert's line is one too high."""
    loop = asyncio.get_running_loop()
    # 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 loop.run_in_executor(_executor, attach_panel, report)
    # submit(), not run_in_executor(): the job's own future is kept so that
    # a cancellation HERE (nothing has been sent yet) can still undo the
    # stamp the job writes.
    job = _alert_db_executor.submit(_stamp_exposure_and_mark, report)
    try:
        stamp = await asyncio.wrap_future(job, loop=loop)
    except asyncio.CancelledError:
        job.add_done_callback(
            functools.partial(_undo_cancelled_stamp, report))
        raise
    delivered = 0
    try:
        detail = header + format_signal_message(report)
        delivered = await _broadcast(
            bot, detail, header + format_signal_message(report, channel=True))
    except Exception:
        # (A cancellation is not an Exception: on shutdown mid-send a chat
        # may already hold the message, so the stamp stays.)
        await _undo_alert_mark(report, stamp)
        raise
    if not delivered:
        await _undo_alert_mark(report, stamp)
        return 0
    # Commit only when the alert reached at least one target; a fully
    # failed send stays eligible for the next scan.
    _commit_alert(key, report)
    logger.info(_alert_sent_line(report, tf_label, delivered, kind=kind))
    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
            await _deliver_alert(bot, key, report, tf.value)


# ── 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
    '<tf> <level_name>' (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:
    # No M15 bar exists to read while the bar that "just closed" falls in
    # gold's daily settlement pause (21:00–22:00 UTC in summer) — the
    # first post-pause bar closes 15 min after the pause ends. A rescan
    # here only re-reads the last pre-pause bar: the 2026-09-26 review of
    # signals.db found the 20:45 UTC bar logged again at 22:01, 22:06 and
    # 22:11 every evening from these tick-driven rescans (the candle loop
    # already skips the same window, `_closed_bar_in_settlement`).
    if _closed_bar_in_settlement(15):
        logger.info(f"Level breach: {pair_symbol} crossed {level_name} at {price} — "
                    f"skipped, the M15 bar to scan is inside the settlement pause")
        return
    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

        # The same delivery as the auto-scan path (panel, the refreshed
        # "already open" line, stamp, send, commit), under the level-break
        # header. The cached M15 report this path reuses can be up to a
        # bar old; _deliver_alert refreshes its exposure lines.
        header = (
            f"⚡ *LEVEL BREAK* — {_escape_md(display)} crossed "
            f"*{_escape_md(label)}* @ `{_fmt_price(price)}`\n\n"
        )
        await _deliver_alert(bot, key, report, Timeframe.M15.value,
                             header=header, 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).
    Failure details are exception text, so they are Markdown-escaped: an
    unpaired underscore in one would make Telegram reject the reply."""
    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("  ⚠️ Calendar: unavailable — "
                             f"{_escape_md(str(cal['detail']))}"
                             f" ({_fmt_age(cal['err'])})")
            elif NewsEngine.source_is_failing(ff):
                lines.append(f"  ⚠️ Calendar: {cal['extra']} — ForexFactory"
                             f" down: {_escape_md(str(ff['detail']))}"
                             f" ({_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" — {_escape_md(str(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)")
        # The local headline scorer (news_model modes fallback / primary).
        # Present only in a process that ran it; a failing batch leaves the
        # cycle's headlines on keyword scores, which nothing else showed.
        model = sh.get("news_model")
        if model:
            if NewsEngine.source_is_failing(model):
                lines.append("  ⚠️ News model: not scoring — "
                             f"{_escape_md(str(model['detail']))}"
                             f" ({_fmt_age(model['err'])})")
            else:
                lines.append(f"  ✅ News model: {model['extra']}"
                             f" ({_fmt_age(model['ok'])} ago)")
        for name, entry in sorted(sh.items()):
            if name.startswith("rss:") and NewsEngine.source_is_failing(entry):
                lines.append(f"  ⚠️ {_escape_md(name[4:])}: "
                             f"{_escape_md(str(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"])
        # The reply lands in the chat the button was pressed in. Only the
        # owner's own chat gets the private lot-size line; the owner
        # pressing it inside a group must not show the group their risk.
        chat = update.effective_chat
        text = format_signal_message(
            report, channel=not _is_owner_chat(chat.id if chat else None))
    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')}",
        "",
    ]

    from main import direction_words, reliability_words  # display helper (2026-09-27)
    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}*  Grade {r.signal_quality} {qe} · {sig}"
        )
        lines.append(
            f"   {direction_words(r.final_score)}, {reliability_words(r.final_confidence)} reliability, "
            f"{'timeframes agree' if r.mtf_agreement else 'timeframes disagree'}"
        )
        lines.append(f"   _{_escape_md(regime)}_")
        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))

    # The 5-bar entry cooldown lives in memory; seed it from the signal log
    # so a restart (a deploy, the weekly timer) cannot re-alert inside it.
    # Before the feed check: /autoscan can start the loops later.
    try:
        bars = await asyncio.get_running_loop().run_in_executor(
            _alert_db_executor, last_alerted_bars)
        restored = _restore_alert_cooldowns(bars)
        logger.info(f"Alert cooldown restored from the signal log: "
                    f"{restored} timeframe(s)")
    except Exception:
        logger.exception("Alert cooldown restore failed — starting without it")

    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)
    _alert_db_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()