#!/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 TwelveData for XAU/USD.
             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.

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 json
import logging
import logging.handlers
import os
import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from typing import Dict, List, Optional, Set

import websockets

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, provider_api_key, telegram_token, telegram_chat_id,
    telegram_channel_id, tiingo_api_key,
)
from data_provider import ForexDataProvider, make_data_provider
from tiingo_provider import (
    Quote, TiingoQuoteProvider, WIDE_SPREAD_MULTIPLE, log_quote,
)
from pair_signals import PAIR_PRODUCERS
from signal_aggregator import SignalAggregator, SignalReport

_APP = get_app_config()

# ── Force UTF-8 on Windows ────────────────────────────────────────────
if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(encoding="utf-8")

_LOG_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "logs")
os.makedirs(_LOG_DIR, exist_ok=True)
_log_formatter = logging.Formatter(
    "%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
_console_handler = logging.StreamHandler()
_console_handler.setFormatter(_log_formatter)
# 3-hour rotation × 240 backups = 30 days retention. Handlers sit on the
# ROOT logger (force=True guarantees ownership even if something logged
# first), so every module logger — data_provider, tiingo_provider,
# news_engine, signal_aggregator, websockets, telegram.ext — lands in
# the same file.
_file_handler = logging.handlers.TimedRotatingFileHandler(
    os.path.join(_LOG_DIR, "telegram.log"),
    when="h", interval=3, backupCount=240, encoding="utf-8",
)
_file_handler.setFormatter(_log_formatter)
logging.basicConfig(level=getattr(logging, _APP.log_level, logging.INFO),
                    handlers=[_console_handler, _file_handler],
                    force=True)
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)
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)

_WS_TO_PAIR = {PAIR_INFO[p]["display"]: p for p in SCAN_PAIRS}
_WS_SYMBOLS = list(_WS_TO_PAIR.keys())

_LEVEL_KEYS = (
    "swing_high_1", "swing_high_2", "swing_high_3",
    "swing_low_1", "swing_low_2", "swing_low_3",
    "ema_trend", "bb_upper", "bb_lower",
    "kc_upper", "kc_lower",
)

_TF_BUTTON_LABELS = [
    (Timeframe.M5, "5 Min"),
    (Timeframe.M15, "15 Min"),
    (Timeframe.H1, "1 Hour"),
    (Timeframe.H4, "4 Hour"),
]


def _build_scan_buttons() -> List[dict]:
    """One row of TF buttons + a scan-all button per scanned pair."""
    multi = len(SCAN_PAIRS) > 1
    buttons = []
    for symbol in SCAN_PAIRS:
        slug = symbol.replace("/", "").lower()
        base = PAIR_INFO[symbol]["display"].split("/")[0]  # XAU
        prefix = f"{base} " if multi else ""
        for tf, label in _TF_BUTTON_LABELS:
            buttons.append({
                "cb": f"scan_{slug}_{tf.value}", "symbol": symbol,
                "tf": tf, "label": f"{prefix}{label}",
            })
        buttons.append({
            "cb": f"scan_{slug}_all", "symbol": symbol, "tf": None,
            "label": f"Scan All TFs — {base}" if multi else "Scan All TFs",
        })
    return buttons


SCAN_BUTTONS = _build_scan_buttons()

SCAN_LOOKUP = {b["cb"]: b for b in SCAN_BUTTONS}

TF_CONFIRMATIONS = {
    Timeframe.M5:  [Timeframe.M15, Timeframe.H1],
    Timeframe.M15: [Timeframe.M5, Timeframe.H1],
    Timeframe.H1:  [Timeframe.M15, Timeframe.H4],
    Timeframe.H4:  [Timeframe.H1, Timeframe.D1],
}

AUTOSCAN_CONFIRMATIONS = {
    Timeframe.M5:  [Timeframe.M15, Timeframe.H1],
    Timeframe.M15: [Timeframe.M5, Timeframe.H1],
    Timeframe.H1:  [Timeframe.M15, Timeframe.H4],
    Timeframe.H4:  [Timeframe.H1, Timeframe.D1],
}

_executor = ThreadPoolExecutor(max_workers=4)

# ── Markdown escaping ─────────────────────────────────────────────────

def _escape_md(text: str) -> str:
    for ch in ("*", "_", "`", "[", "]"):
        text = text.replace(ch, "\\" + ch)
    return text


def _fmt_price(price: float) -> str:
    return f"{price:.2f}" if price >= 10 else f"{price:.5f}"


# ── Shared data provider (single cache across all scans) ─────────────

_shared_dp: Optional[ForexDataProvider] = None


def _get_shared_dp() -> ForexDataProvider:
    global _shared_dp
    if _shared_dp is None:
        _shared_dp = make_data_provider(PlatformConfig())
    return _shared_dp


# ── Tiingo bid/ask spread monitor (optional; no key → feature off) ───
# Warn-only BY DESIGN: there is no historical bid/ask to replay, so a
# suppressing gate could never be validated (live == backtest rule).
# Signals never consume bid/ask — this is execution-cost visibility.

_TIINGO_KEY = tiingo_api_key()
_tiingo: Optional[TiingoQuoteProvider] = (
    TiingoQuoteProvider(_TIINGO_KEY) if _TIINGO_KEY else None
)
_last_quotes: Dict[str, Quote] = {}
_QUOTE_MAX_AGE = 600          # cached quote stays displayable this long
_SPREAD_SAMPLE_SECONDS = 300  # sampler cadence (→ logs/spread_log.csv)

# config.json telegram.channel_spread_detail: True while the channel is
# internal/testing → it gets raw bid/ask like the owner DM. MUST be
# flipped back to false before the channel goes public (Tiingo free
# tier is internal-use-only).
_CHANNEL_SPREAD_DETAIL = _APP.channel_spread_detail


def _spread_detail_for(chat_id: int) -> bool:
    """Owner DMs always get raw quote numbers; the channel only while
    config.json marks it internal."""
    return chat_id != CHANNEL_ID or _CHANNEL_SPREAD_DETAIL


# ── Signal formatting ─────────────────────────────────────────────────

_SIGNAL_EMOJI = {
    SignalStrength.STRONG_BUY:  "🟢",
    SignalStrength.BUY:         "🟩",
    SignalStrength.NEUTRAL:     "⬜",
    SignalStrength.SELL:        "🟥",
    SignalStrength.STRONG_SELL: "🔴",
}

_QUALITY_EMOJI = {
    "A+": "🏆", "A": "🥇", "B": "🥈", "C": "🥉", "D": "⚠️", "F": "❌",
}

_SIGNAL_TAG = {
    SignalStrength.STRONG_BUY:  "Strong Buy",
    SignalStrength.BUY:         "Bullish",
    SignalStrength.NEUTRAL:     "Neutral",
    SignalStrength.SELL:        "Bearish",
    SignalStrength.STRONG_SELL: "Strong Sell",
}


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)


def _indicator_line(name: str, ind) -> str:
    tag = _SIGNAL_TAG.get(ind.signal, "Neutral")
    short = ind.description.split(" | ")[0] if " | " in ind.description else ind.description
    return f"[{tag}] *{_escape_md(ind.name)}*: {_escape_md(short)}"


def _live_price_line(symbol: str) -> Optional[str]:
    """Fresh WebSocket tick for this symbol, if we have one (<2 min old)."""
    price = _last_prices.get(symbol)
    ts = _last_price_ts.get(symbol, 0)
    if price is None:
        return None
    age = time.time() - ts
    if age > 120:
        return None
    return f"📡 Live: *{_fmt_price(price)}*  ({age:.0f}s tick)"


def _spread_lines(symbol: str, include_quote: bool = True) -> List[str]:
    """Live bid/ask + wide-spread warning for signal messages.
    Empty when Tiingo is off or the cached quote is stale.

    include_quote=False is the subscriber-channel variant: Tiingo's
    free tier is licensed internal-use-only (no redistributing the
    data), so raw bid/ask/spread numbers stay in the owner DM — the
    channel gets only a number-free wide-spread caution."""
    q = _last_quotes.get(symbol)
    if q is None or q.age_seconds() > _QUOTE_MAX_AGE:
        return []
    assumed = PAIR_INFO.get(symbol, {}).get("default_spread", 0.0)
    wide = assumed > 0 and q.spread >= WIDE_SPREAD_MULTIPLE * assumed
    if not include_quote:
        return [
            "⚠️ Spread unusually wide right now — execution cost is "
            "elevated, consider waiting for it to normalize"
        ] if wide else []
    lines = [
        f"💱 Bid/Ask: {_fmt_price(q.bid)} / {_fmt_price(q.ask)}"
        f"  (spread ${q.spread:.2f})"
    ]
    if wide:
        lines.append(
            f"⚠️ Wide spread: ${q.spread:.2f} ≈ {q.spread / assumed:.1f}× the "
            f"${assumed:.2f} cost the strategy was validated at — "
            f"execution is expensive right now"
        )
    return lines


_DISCLAIMER = (
    "⚠️ _Not financial advice. For educational and research purposes only. "
    "Trading forex/CFDs carries a high risk of loss — do your own analysis "
    "and never risk money you cannot afford to lose._"
)


def format_signal_message(report: SignalReport, spread_detail: bool = True) -> str:
    sig = report.final_signal
    emoji = _SIGNAL_EMOJI.get(sig, "⬜")
    qe = _QUALITY_EMOJI.get(report.signal_quality, "")
    label = _signal_label(sig)
    mtf_tag = "✅ MTF aligned" if report.mtf_agreement else "❌ MTF diverged"

    lines: list[str] = []

    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 += [
        f"🔍 *{report.display_name} Signal Analysis*  ({report.primary_analysis.timeframe if report.primary_analysis else ''})",
        f"🕐 {now}",
        "",
        f"💰 Close: *{_fmt_price(report.current_price)}*",
    ]
    live = _live_price_line(report.symbol)
    if live:
        lines.append(live)
    lines += _spread_lines(report.symbol, include_quote=spread_detail)
    lines += [
        f"📊 Regime: {regime_str} | Vol: {vol_str}",
        "",
        "━━━━━━━━━━━━━━━━━━━━",
        f"{emoji} *{label}*  {qe}{report.signal_quality}  |  Score: {report.final_score:+.1f}  |  {report.final_confidence:.0f}% conf",
        f"_{_escape_md(report.action_recommendation)}_",
        f"{mtf_tag}",
    ]

    mtf = _mtf_line(report)
    if mtf:
        lines += ["", f"📈 *MTF*: {mtf}"]

    if report.primary_analysis:
        lines += ["", "━━━━━━━━━━━━━━━━━━━━", "📋 *Indicators*"]
        for name, ind in report.primary_analysis.indicators.items():
            if name.startswith("_") or ind.weight == 0:
                continue
            lines.append(_indicator_line(name, ind))

    if report.warnings:
        lines.append("")
        for w in report.warnings:
            lines.append(f"⚠️ {_escape_md(w)}")

    if report.trade_setup and sig not in (SignalStrength.NEUTRAL,):
        ts = report.trade_setup
        dir_emoji = "📈" if ts.direction == "BUY" else "📉"
        risk_usd = ts.risk_pips * ts.pip_value
        reward_usd = ts.reward_pips * ts.pip_value
        lines += [
            "",
            "━━━━━━━━━━━━━━━━━━━━",
            f"🚀 *TRADE SETUP*",
            f"{dir_emoji} Direction: *{ts.direction}*",
            f"▶ Entry:  `{_fmt_price(ts.entry_price)}`",
            f"🛑 SL:     `{_fmt_price(ts.stop_loss)}`  (${risk_usd:.2f} / {ts.risk_pips:.0f} pips)",
            f"🎯 TP1:    `{_fmt_price(ts.take_profit_1)}`  (${reward_usd:.2f} / {ts.reward_pips:.0f} pips)",
        ]
        if ts.take_profit_2:
            lines.append(f"🎯 TP2:    `{_fmt_price(ts.take_profit_2)}`")
        lines.append(f"⚖️  RR:     1:{ts.risk_reward:.1f}")

        # Trailing-stop management plan (matches backtested exit model:
        # TP1 is advisory, the trail is the exit). Params are stamped on
        # the setup by the producing config so per-TF factory overrides
        # (M15 arm 0.70, M5 dist 0.85) are what subscribers see; only a
        # legacy unstamped setup falls back to platform defaults.
        is_pb = getattr(ts, "strategy", "GENERIC").endswith("_PB")
        arm_r = getattr(ts, "trail_arm_r", None)
        dist_r = getattr(ts, "trail_dist_r", None)
        if arm_r is None or dist_r is None:
            sig_cfg = PlatformConfig().signals
            arm_r = sig_cfg.pb_trail_arm_r if is_pb else sig_cfg.trail_arm_r
            dist_r = sig_cfg.pb_trail_dist_r if is_pb else sig_cfg.trail_dist_r
        if arm_r > 0:
            risk_dist = abs(ts.entry_price - ts.stop_loss)
            if ts.direction == "BUY":
                arm_price = ts.entry_price + arm_r * risk_dist
            else:
                arm_price = ts.entry_price - arm_r * risk_dist
            trail_dist = dist_r * risk_dist
            tag = " (pullback)" if is_pb else ""
            lines.append(
                f"🔄 Trail{tag}:  once price hits `{_fmt_price(arm_price)}` "
                f"(+{arm_r:.1f}R), trail SL ${trail_dist:.2f} "
                f"behind best price — TP1 is a guide, let it run"
            )

    lines += ["", "━━━━━━━━━━━━━━━━━━━━", _DISCLAIMER]

    return "\n".join(lines)


# ── Auto-scan summary format ─────────────────────────────────────────

def _format_scan_summary(reports: List[SignalReport], tf: Timeframe) -> str:
    now = datetime.now(timezone.utc).strftime("%H:%M UTC")
    lines = [
        f"📊 *{tf.value} Auto-Scan* | {now}",
        "━━━━━━━━━━━━━━━━━━━━",
    ]
    for r in reports:
        emoji = _SIGNAL_EMOJI.get(r.final_signal, "⬜")
        sig = _signal_label(r.final_signal)
        mtf = "✅" if r.mtf_agreement else "❌"
        lines.append(
            f"{emoji} *{r.display_name}*  {sig}  "
            f"{_QUALITY_EMOJI.get(r.signal_quality, '')}{r.signal_quality}  "
            f"{r.final_score:+.1f}  {r.final_confidence:.0f}%  {mtf}MTF"
        )
    lines.append("━━━━━━━━━━━━━━━━━━━━")

    actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B", "C")]
    if actionable:
        names = ", ".join(r.display_name for r in actionable)
        lines.append(f"🔔 Actionable: {names}")
    else:
        lines.append("No actionable signals this cycle.")

    lines += ["", _DISCLAIMER]

    return "\n".join(lines)


# ── Keyboard helper ───────────────────────────────────────────────────

def _scan_button_rows() -> List[List[InlineKeyboardButton]]:
    """Per pair: one row of TF buttons + one scan-all row."""
    rows = []
    for symbol in SCAN_PAIRS:
        tf_buttons = [b for b in SCAN_BUTTONS
                      if b["symbol"] == symbol and b["tf"] is not None]
        rows.append([InlineKeyboardButton(b["label"], callback_data=b["cb"])
                     for b in tf_buttons])
        all_button = next((b for b in SCAN_BUTTONS
                           if b["symbol"] == symbol and b["tf"] is None), None)
        if all_button:
            rows.append([InlineKeyboardButton(all_button["label"],
                                              callback_data=all_button["cb"])])
    return rows


def scan_keyboard() -> InlineKeyboardMarkup:
    return InlineKeyboardMarkup(_scan_button_rows())


def main_menu_keyboard() -> InlineKeyboardMarkup:
    """Full interactive menu: scans + settings + status in one keyboard."""
    autoscan_label = "🔔 Auto-scan: ON" if _autoscan_chats else "🔕 Auto-scan: OFF"
    quality_label = f"🎚 Alerts: {_alert_min_quality}+" if _alert_min_quality != "A" else "🎚 Alerts: A only"
    rows = _scan_button_rows()
    rows.append([
        InlineKeyboardButton(autoscan_label, callback_data="menu_autoscan"),
        InlineKeyboardButton(quality_label, callback_data="menu_quality"),
    ])
    rows.append([
        InlineKeyboardButton("📊 Status", callback_data="menu_status"),
        InlineKeyboardButton("ℹ️ Help", callback_data="menu_help"),
    ])
    return InlineKeyboardMarkup(rows)


async def _safe_edit(query, text: str, reply_markup=None) -> None:
    """edit_message_text that survives 'message is not modified' and
    Markdown parse failures (falls back to plain text)."""
    try:
        await query.edit_message_text(
            text, parse_mode=ParseMode.MARKDOWN, reply_markup=reply_markup
        )
    except BadRequest as exc:
        if "not modified" in str(exc).lower():
            return
        logger.warning(f"Markdown edit failed ({exc}); retrying as plain text")
        try:
            await query.edit_message_text(text, reply_markup=reply_markup)
        except BadRequest:
            pass


# ── Signal runner (shared data provider for cache reuse) ──────────────

def _run_signal_sync(
    symbol: str,
    timeframe: Timeframe,
    confirmation_tfs: Optional[List[Timeframe]] = None,
) -> SignalReport:
    if confirmation_tfs is None:
        confirmation_tfs = TF_CONFIRMATIONS.get(
            timeframe, [Timeframe.M15, Timeframe.H4]
        )
    # Per-pair tuned configs (regime weights, gates, sessions) come from
    # the pair_signals factories; a plain PlatformConfig would silently
    # run every pair with XAU defaults.
    factory = PAIR_PRODUCERS.get(symbol)
    if factory is not None:
        config = factory(timeframe, confirmation_tfs)
    else:
        config = PlatformConfig()
        config.primary_timeframe = timeframe
        config.confirmation_timeframes = confirmation_tfs
    aggregator = SignalAggregator(config)
    aggregator.data_provider = _get_shared_dp()
    return aggregator.analyze_pair(symbol=symbol)


async def run_signal(
    symbol: str,
    timeframe: Timeframe,
    confirmation_tfs: Optional[List[Timeframe]] = None,
) -> SignalReport:
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(
        _executor, _run_signal_sync, symbol, timeframe, confirmation_tfs
    )


async def _refresh_quotes(symbols: List[str]) -> None:
    """Fetch fresh Tiingo quotes (blocking HTTP) off the event loop —
    called just before alerts go out so the spread block reflects the
    moment of entry, and by the 5-min sampler."""
    if _tiingo is None:
        return
    loop = asyncio.get_running_loop()
    for symbol in set(symbols):
        try:
            quote = await loop.run_in_executor(
                _executor, _tiingo.get_quote, symbol)
        except Exception as exc:
            logger.warning(f"Tiingo quote refresh failed for {symbol}: {exc}")
            continue
        if quote is not None:
            _last_quotes[symbol] = quote


# ── Alert gating (quality threshold + dedup) ─────────────────────────

_QUALITY_RANK = {"A+": 4, "A": 3, "B": 2, "C": 1}
_ALERT_QUALITY_CYCLE = ["C", "B", "A"]   # C+ → B+ → A-only
_alert_min_quality = _APP.alert_min_quality

# Fingerprint of the last alert per (symbol, timeframe) so a persisting
# signal doesn't re-send an identical message every candle.
_last_alert_fp: Dict[str, tuple] = {}
_last_alert_ts: Dict[str, float] = {}
_last_alert_setup: Dict[str, tuple] = {}   # key → (entry, risk) of last ALERTED setup
_ALERT_REARM_SECONDS = 3600   # identical signal re-alerts after 1h


def _passes_quality(quality: str) -> bool:
    return _QUALITY_RANK.get(quality, 0) >= _QUALITY_RANK.get(_alert_min_quality, 1)


def _alert_fingerprint(report: SignalReport) -> tuple:
    """Alert identity for dedup: signal, quality, setup direction.

    Entry movement is checked separately in _should_alert against the
    last ALERTED entry/risk (a stable reference). Deriving a price bucket
    from the CURRENT SL distance here made the fingerprint churn on
    normal ATR/anchor drift — near-duplicate alerts every candle."""
    setup = report.trade_setup
    return (
        report.final_signal.name,
        report.signal_quality,
        setup.direction if setup else None,
    )


def _should_alert(key: str, report: SignalReport) -> bool:
    """True when this signal is new/changed for the key, or re-armed.

    A persisting identical signal dedups for the re-arm hour, but a setup
    whose entry moved >= half the last-alerted risk distance (new
    pullback/swing anchor) re-alerts immediately.

    Pure check — call _commit_alert only after the message actually went
    out, so a failed Telegram send doesn't consume the alert for an hour.
    """
    fp = _alert_fingerprint(report)
    if _last_alert_fp.get(key) != fp:
        return True
    if time.time() - _last_alert_ts.get(key, 0) >= _ALERT_REARM_SECONDS:
        return True
    setup = report.trade_setup
    last = _last_alert_setup.get(key)
    if setup is not None and last is not None:
        last_entry, last_risk = last
        if abs(setup.entry_price - last_entry) >= 0.5 * max(last_risk, 1e-9):
            return True
    return False


def _commit_alert(key: str, report: SignalReport) -> None:
    _last_alert_fp[key] = _alert_fingerprint(report)
    _last_alert_ts[key] = time.time()
    setup = report.trade_setup
    if setup is not None:
        _last_alert_setup[key] = (
            setup.entry_price, abs(setup.entry_price - setup.stop_loss)
        )
    else:
        _last_alert_setup.pop(key, None)


# ── 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
    if remaining <= buffer:
        remaining += interval_seconds
    return remaining


async def _scan_timeframe(
    bot,
    tf: Timeframe,
    confirmation_tfs: List[Timeframe],
) -> None:
    """Scan all pairs for one timeframe and broadcast results."""
    tasks = [run_signal(symbol, tf, confirmation_tfs) for symbol in SCAN_PAIRS]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    reports: List[SignalReport] = []
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            logger.exception(f"Auto-scan failed for {SCAN_PAIRS[i]} {tf.value}: {result}")
        else:
            reports.append(result)

    if not reports:
        return

    _update_level_cache(reports)

    actionable = [
        r for r in reports
        if r.final_signal != SignalStrength.NEUTRAL   # same gate as breach path
        and _passes_quality(r.signal_quality)
        and _should_alert(f"{r.symbol}_{tf.value}", r)
    ]
    if actionable:
        await _refresh_quotes([r.symbol for r in actionable])
    sent_ok: Set[int] = set()
    for chat_id in _signal_targets():
        for i, report in enumerate(actionable):
            detail = format_signal_message(
                report, spread_detail=_spread_detail_for(chat_id))
            try:
                await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN)
                sent_ok.add(i)
            except Exception as exc:
                logger.warning(f"Failed to send signal for chat {chat_id}: {exc}")
    # Commit fingerprints only for alerts that reached at least one target;
    # a fully failed send stays eligible for the next scan.
    for i, report in enumerate(actionable):
        if i in sent_ok:
            _commit_alert(f"{report.symbol}_{tf.value}", report)


# ── WebSocket level-breach monitor ──────────────────────────────────

_level_cache: Dict[str, Dict[str, float]] = {}
_level_cache_ts: Dict[str, float] = {}
_LEVEL_CACHE_TTL = 3600
_last_prices: Dict[str, float] = {}
_last_price_ts: Dict[str, float] = {}
_scan_cooldowns: Dict[str, float] = {}

_cfg = PlatformConfig()
_WS_API_KEY = provider_api_key()
_WS_URL = f"wss://ws.twelvedata.com/v1/quotes/price?apikey={_WS_API_KEY}"
_WS_COOLDOWN = _cfg.ws_cooldown_seconds

_breach_tasks: Set[asyncio.Task] = set()


def _update_level_cache(reports: List[SignalReport]) -> None:
    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:
                _level_cache[r.symbol] = levels
                _level_cache_ts[r.symbol] = now


def _check_level_breach(symbol: str, price: float) -> Optional[str]:
    levels = _level_cache.get(symbol)
    if not levels:
        return None

    cache_age = time.time() - _level_cache_ts.get(symbol, 0)
    if cache_age > _LEVEL_CACHE_TTL:
        return None

    last = _last_prices.get(symbol)
    if last is None:
        return None

    for name, level in levels.items():
        if (last < level <= price) or (last > level >= price):
            return 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, ws_symbol: str, pair_symbol: str, price: float, level_name: str) -> None:
    now = time.time()
    if now - _scan_cooldowns.get(pair_symbol, 0) < _WS_COOLDOWN:
        return

    _scan_cooldowns[pair_symbol] = now
    display = PAIR_INFO.get(pair_symbol, {}).get("display", ws_symbol)
    label = level_name.replace("_", " ").title()
    logger.info(f"Level breach: {display} crossed {label} at {price} — rescanning M15")

    try:
        report = await run_signal(pair_symbol, Timeframe.M15)
        _update_level_cache([report])
    except Exception:
        logger.exception(f"Breach scan failed for {display}")
        return

    # Alert only when the rescan produced an actionable, changed signal —
    # same gates as the auto-scanner, so a breach can't spam the channel.
    if (report.final_signal == SignalStrength.NEUTRAL
            or not _passes_quality(report.signal_quality)
            or not _should_alert(f"{pair_symbol}_{Timeframe.M15.value}", report)):
        return

    header = (
        f"⚡ *LEVEL BREAK* — {_escape_md(display)} crossed "
        f"*{_escape_md(label)}* @ `{_fmt_price(price)}`\n\n"
    )
    await _refresh_quotes([pair_symbol])
    sent = False
    for chat_id in _signal_targets():
        detail = header + format_signal_message(
            report, spread_detail=_spread_detail_for(chat_id))
        try:
            await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN)
            sent = True
        except Exception as exc:
            logger.warning(f"Failed to send breach alert to chat {chat_id}: {exc}")
    if sent:
        _commit_alert(f"{pair_symbol}_{Timeframe.M15.value}", report)


def _breach_task_done(task: asyncio.Task) -> None:
    """Fire-and-forget breach tasks must still surface their failures —
    without this, an exception sits unretrieved until GC and never
    reaches the log file."""
    _breach_tasks.discard(task)
    if not task.cancelled() and task.exception() is not None:
        logger.error("Breach alert task failed", exc_info=task.exception())


async def _ws_heartbeat(ws) -> None:
    """Twelve Data drops idle WS connections; heartbeat every 10s keeps it
    alive through quiet periods (weekends, low-liquidity hours)."""
    while True:
        await asyncio.sleep(10)
        await ws.send(json.dumps({"action": "heartbeat"}))


async def _ws_monitor(bot) -> None:
    backoff = 1
    while _feed_active():
        hb_task: Optional[asyncio.Task] = None
        try:
            async with websockets.connect(_WS_URL) as ws:
                subscribe = {
                    "action": "subscribe",
                    "params": {"symbols": ",".join(_WS_SYMBOLS)},
                }
                await ws.send(json.dumps(subscribe))
                logger.info(f"WebSocket connected, subscribed to {len(_WS_SYMBOLS)} symbols")
                backoff = 1
                hb_task = asyncio.create_task(_ws_heartbeat(ws))

                async for raw in ws:
                    if not _feed_active():
                        break

                    try:
                        msg = json.loads(raw)
                    except json.JSONDecodeError:
                        logger.warning(f"WebSocket: malformed JSON, skipping")
                        continue

                    event = msg.get("event")
                    if event == "subscribe-status":
                        fails = msg.get("fails") or []
                        if fails:
                            logger.warning(f"WebSocket: failed to subscribe: {fails}")
                        continue
                    if event != "price":
                        continue

                    ws_symbol = msg.get("symbol")
                    pair_symbol = _WS_TO_PAIR.get(ws_symbol)
                    if not pair_symbol:
                        continue

                    try:
                        price = float(msg["price"])
                    except (KeyError, ValueError, TypeError) as e:
                        logger.warning(f"WebSocket: bad price data for {ws_symbol}: {e}")
                        continue

                    breached = _check_level_breach(pair_symbol, price)
                    _last_prices[pair_symbol] = price
                    _last_price_ts[pair_symbol] = time.time()

                    if breached:
                        task = asyncio.create_task(
                            _handle_breach(bot, ws_symbol, pair_symbol, price, breached)
                        )
                        _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)
        finally:
            if hb_task is not None:
                hb_task.cancel()

    logger.info("WebSocket monitor stopped")


async def _candle_loop(bot, tf: Timeframe, interval_minutes: int) -> None:
    label = tf.value
    try:
        while _feed_active():
            wait = _seconds_until_boundary(interval_minutes, buffer=10)
            logger.info(f"{label} auto-scan: next in {wait:.0f}s")
            await asyncio.sleep(wait)
            if not _feed_active():
                break
            logger.info(f"{label} auto-scan: firing")
            try:
                await _scan_timeframe(
                    bot, tf, AUTOSCAN_CONFIRMATIONS[tf]
                )
            except Exception:
                # One failed cycle must not kill the loop for good —
                # log it and wait for the next candle boundary.
                logger.exception(f"{label} auto-scan: cycle failed")
    except asyncio.CancelledError:
        pass
    finally:
        logger.info(f"{label} auto-scan loop stopped")


def _weekend_closed(now: datetime) -> bool:
    """Fri 22:00 → Sun 22:00 UTC — the window the platform treats as
    market-closed (same as the weekend filter)."""
    wd, hr = now.weekday(), now.hour
    return (wd == 4 and hr >= 22) or wd == 5 or (wd == 6 and hr < 22)


async def _spread_sampler_loop() -> None:
    """Every 5 min: fetch bid/ask for each scanned pair, cache it for
    the alert formatter, and append to logs/spread_log.csv — the
    dataset that validates (or refutes) the assumed $0.30 round-trip
    cost every backtest charges."""
    try:
        while True:
            try:
                if not _weekend_closed(datetime.now(timezone.utc)):
                    await _refresh_quotes(SCAN_PAIRS)
                    for symbol in SCAN_PAIRS:
                        q = _last_quotes.get(symbol)
                        # Age guard: never log a stale row after a failed fetch.
                        if q is not None and q.age_seconds() < _SPREAD_SAMPLE_SECONDS:
                            log_quote(q)
            except Exception:
                logger.exception("Spread sampler: cycle failed")
            await asyncio.sleep(_SPREAD_SAMPLE_SECONDS)
    except asyncio.CancelledError:
        pass
    finally:
        logger.info("Spread sampler stopped")


_TF_INTERVAL_MINUTES = {
    Timeframe.M5: 5, Timeframe.M15: 15, Timeframe.M30: 30,
    Timeframe.H1: 60, Timeframe.H4: 240, Timeframe.D1: 1440,
}
_AUTOSCAN_LOOPS = [
    (tf, _TF_INTERVAL_MINUTES[tf]) for tf in _APP.scan_timeframes
]


_ws_task: Optional[asyncio.Task] = None
_spread_task: Optional[asyncio.Task] = None


def _start_scheduler(bot) -> None:
    global _ws_task, _spread_task
    alive = [t for t in _autoscan_tasks if not t.done()]
    if not alive:
        _autoscan_tasks.clear()
        for tf, interval in _AUTOSCAN_LOOPS:
            _autoscan_tasks.append(
                asyncio.create_task(_candle_loop(bot, tf, interval))
            )
        logger.info("Candle-boundary scheduler started (%s)",
                    "/".join(tf.name for tf, _ in _AUTOSCAN_LOOPS))

    if _ws_task is None or _ws_task.done():
        _ws_task = asyncio.create_task(_ws_monitor(bot))
        logger.info("WebSocket level-breach monitor started")

    if _tiingo is not None and (_spread_task is None or _spread_task.done()):
        _spread_task = asyncio.create_task(_spread_sampler_loop())
        logger.info("Tiingo spread sampler started (every %ss)",
                    _SPREAD_SAMPLE_SECONDS)


def _stop_scheduler() -> None:
    global _ws_task, _spread_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
    if _spread_task and not _spread_task.done():
        _spread_task.cancel()
    _spread_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"
    "/cache — Show data cache age per pair/TF\n"
    "/clearcache — Wipe data cache, force fresh fetch\n"
    "/help — This message\n\n"
    "*Auto-scan* fires at candle close: M5/M15/H1/H4.\n"
    "*Alerts* button sets the minimum quality grade that gets sent.\n"
    "*WebSocket* streams live prices; key-level breaches trigger an "
    "instant M15 rescan and alert (5-min cooldown).\n"
    "*Channel feed*: signal + breach alerts mirror to the subscriber "
    "channel when TELEGRAM_CHANNEL_ID is set; /autoscan only toggles "
    "your DM copy."
)


@_owner_only
async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text(_HELP_TEXT, parse_mode=ParseMode.MARKDOWN)


def _status_text() -> str:
    lines = ["📊 *ForexBot2 Status*\n"]

    if _autoscan_chats:
        lines.append(f"✅ Auto-scan: *running* ({len(_autoscan_chats)} chat)")
        tasks_alive = sum(1 for t in _autoscan_tasks if not t.done())
        lines.append(f"   Candle loops: {tasks_alive}/{len(_AUTOSCAN_LOOPS)}")
    else:
        lines.append("⏹ Auto-scan: *stopped*")

    if CHANNEL_ID is not None:
        lines.append(f"📢 Channel feed: *on* (`{CHANNEL_ID}`)")

    if _ws_task and not _ws_task.done():
        lines.append("📡 WebSocket: *connected*")
    else:
        lines.append("❌ WebSocket: *disconnected*")

    if _tiingo is not None:
        for pair in SCAN_PAIRS:
            q = _last_quotes.get(pair)
            display = PAIR_INFO.get(pair, {}).get("display", pair)
            if q is None:
                lines.append(f"💱 Spread ({display}): waiting for first quote")
            else:
                lines.append(
                    f"💱 Spread ({display}): `${q.spread:.2f}` "
                    f"({q.age_seconds():.0f}s ago)")

    if _last_prices:
        lines.append("\n*Live Prices (WebSocket):*")
        for ws_sym, pair in _WS_TO_PAIR.items():
            price = _last_prices.get(pair)
            display = PAIR_INFO.get(pair, {}).get("display", ws_sym)
            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(v) for v in _level_cache.values())
    if cached_levels:
        oldest_age = time.time() - min(_level_cache_ts.values()) if _level_cache_ts else 0
        lines.append(f"\n*Level Cache:* {cached_levels} levels across {cached_pairs} pairs (oldest {oldest_age:.0f}s)")
    else:
        lines.append("\n*Level Cache:* empty — breach alerts start after first candle scan")

    active_cooldowns = sum(
        1 for v in _scan_cooldowns.values()
        if time.time() - v < _WS_COOLDOWN
    )
    if active_cooldowns:
        lines.append(f"*Cooldowns:* {active_cooldowns} pair(s) in cooldown")

    active_breach = sum(1 for t in _breach_tasks if not t.done())
    if active_breach:
        lines.append(f"*Breach tasks:* {active_breach} running")

    lines.append(f"\n*Alert quality:* {_alert_min_quality}+ "
                 f"| re-arm {_ALERT_REARM_SECONDS // 60}m")
    return "\n".join(lines)


@_owner_only
async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text(_status_text(), parse_mode=ParseMode.MARKDOWN)


@_owner_only
async def cmd_cache(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    dp = _get_shared_dp()
    if not dp._cache_timestamps:
        await update.message.reply_text("📭 Cache is empty — no data fetched yet.")
        return

    now = time.time()
    lines = ["📦 *Data Cache Status*\n"]

    for pair in SCAN_PAIRS:
        display = PAIR_INFO.get(pair, {}).get("display", pair)
        pair_lines = []
        for tf in [Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4, Timeframe.D1]:
            cache_key = f"{pair}_{tf.value}"
            ts = dp._cache_timestamps.get(cache_key)
            if ts is None:
                continue
            age_min = (now - ts) / 60
            df = dp._cache.get(cache_key)
            last_bar = df.index[-1].strftime("%H:%M") if df is not None and not df.empty else "?"
            fresh = "✅" if age_min < dp.CANDLE_SECONDS.get(tf, 3600) / 60 else "⚠️"
            pair_lines.append(f"  {tf.value:<5} {fresh} fetched `{age_min:.0f}m` ago | last bar `{last_bar}`")
        if pair_lines:
            lines.append(f"*{display}*")
            lines.extend(pair_lines)

    await update.message.reply_text("\n".join(lines), parse_mode=ParseMode.MARKDOWN)


@_owner_only
async def cmd_clearcache(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    dp = _get_shared_dp()
    count = dp.clear_cache()
    await update.message.reply_text(
        f"🗑 Cache cleared ({count} entries wiped).\nNext scan will fetch fresh data from API.",
        parse_mode=ParseMode.MARKDOWN,
    )


@_owner_only
async def cmd_autoscan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    chat_id = update.effective_chat.id

    if chat_id in _autoscan_chats:
        _autoscan_chats.discard(chat_id)
        if not _feed_active():
            _stop_scheduler()
            msg = "⏹ Auto-scan stopped."
        else:
            msg = ("⏹ Auto-scan stopped for this chat.\n"
                   "📢 Subscriber channel feed continues.")
        await update.message.reply_text(msg, parse_mode=ParseMode.MARKDOWN)
        return

    _autoscan_chats.add(chat_id)
    _start_scheduler(context.bot)

    next_m5 = _seconds_until_boundary(5, buffer=10)
    next_m15 = _seconds_until_boundary(15, buffer=10)
    next_h1 = _seconds_until_boundary(60, buffer=10)
    next_h4 = _seconds_until_boundary(240, buffer=10)
    await update.message.reply_text(
        "✅ *Auto-scan enabled*\n\n"
        f"Scanning {_PAIRS_LABEL} on M5/M15/H1/H4 at candle close.\n"
        f"📡 Live price monitoring active.\n"
        f"Next: M5 {next_m5 / 60:.0f}m | M15 {next_m15 / 60:.0f}m | "
        f"H1 {next_h1 / 60:.0f}m | H4 {next_h4 / 60:.0f}m\n\n"
        "Use /autoscan again to stop.",
        parse_mode=ParseMode.MARKDOWN,
    )


@_owner_only
async def cmd_scan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text(
        f"🔍 *{_PAIRS_LABEL} — Pick a timeframe*",
        parse_mode=ParseMode.MARKDOWN,
        reply_markup=main_menu_keyboard(),
    )


@_owner_only
async def cb_scan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    query = update.callback_query
    await query.answer()

    btn = SCAN_LOOKUP.get(query.data)
    if not btn:
        return

    if btn["tf"] is None:
        await _cb_scan_all(query, btn["symbol"])
        return

    display = PAIR_INFO[btn["symbol"]]["display"]
    await _safe_edit(query, f"⏳ Scanning {display} {btn['label']}…")

    try:
        report = await run_signal(btn["symbol"], btn["tf"])
        text = format_signal_message(report)
    except Exception as exc:
        logger.exception("Signal scan failed")
        text = f"❌ Scan failed: {_escape_md(str(exc))}"

    await _safe_edit(query, text, reply_markup=main_menu_keyboard())


@_owner_only
async def cb_menu(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Interactive menu actions: autoscan toggle, quality cycle, status, help."""
    global _alert_min_quality
    query = update.callback_query
    action = query.data

    if action == "menu_autoscan":
        chat_id = query.message.chat.id
        if chat_id in _autoscan_chats:
            _autoscan_chats.discard(chat_id)
            if not _feed_active():
                _stop_scheduler()
                await query.answer("Auto-scan stopped")
            else:
                await query.answer("DM alerts off — channel feed continues")
        else:
            _autoscan_chats.add(chat_id)
            _start_scheduler(context.bot)
            await query.answer("Auto-scan started")
        try:
            await query.edit_message_reply_markup(reply_markup=main_menu_keyboard())
        except BadRequest:
            pass

    elif action == "menu_quality":
        i = _ALERT_QUALITY_CYCLE.index(_alert_min_quality) if _alert_min_quality in _ALERT_QUALITY_CYCLE else 0
        _alert_min_quality = _ALERT_QUALITY_CYCLE[(i + 1) % len(_ALERT_QUALITY_CYCLE)]
        await query.answer(f"Alerting {_alert_min_quality}+ signals")
        try:
            await query.edit_message_reply_markup(reply_markup=main_menu_keyboard())
        except BadRequest:
            pass

    elif action == "menu_status":
        await query.answer()
        await query.message.reply_text(_status_text(), parse_mode=ParseMode.MARKDOWN)

    elif action == "menu_help":
        await query.answer()
        await query.message.reply_text(_HELP_TEXT, parse_mode=ParseMode.MARKDOWN)


async def _cb_scan_all(query, symbol: str = XAUUSD) -> None:
    display = PAIR_INFO[symbol]["display"]
    await _safe_edit(query, f"⏳ Scanning {display} across M5 / M15 / H1 / H4…")

    tfs = [Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4]
    tasks = [run_signal(symbol, tf) for tf in tfs]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    reports = []
    errors = []
    for tf, result in zip(tfs, results):
        if isinstance(result, Exception):
            errors.append(f"❌ {tf.value}: {result}")
        else:
            reports.append(result)

    if reports:
        _update_level_cache(reports)

    lines = [
        f"📊 *{display} — All Timeframes*",
        f"🕐 {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}",
        "",
    ]

    for r in reports:
        emoji = _SIGNAL_EMOJI.get(r.final_signal, "⬜")
        qe = _QUALITY_EMOJI.get(r.signal_quality, "")
        tf_label = r.primary_analysis.timeframe if r.primary_analysis else "?"
        sig = _signal_label(r.final_signal)
        regime = r.market_regime.replace("_", " ").title()
        lines.append(
            f"{emoji} *{tf_label}*  {sig}  {qe}{r.signal_quality}  "
            f"Score: {r.final_score:+.1f}  {r.final_confidence:.0f}%"
        )
        lines.append(f"   _{_escape_md(regime)}_ | {'✅ MTF' if r.mtf_agreement else '❌ MTF'}")
        if r.trade_setup and r.final_signal != SignalStrength.NEUTRAL:
            ts = r.trade_setup
            d = "📈" if ts.direction == "BUY" else "📉"
            lines.append(
                f"   {d} {ts.direction} `{_fmt_price(ts.entry_price)}` "
                f"SL `{_fmt_price(ts.stop_loss)}` TP `{_fmt_price(ts.take_profit_1)}` "
                f"RR 1:{ts.risk_reward:.1f}"
            )
        lines.append("")

    for err in errors:
        lines.append(err)

    lines.append("_Use buttons below for detailed single-TF view._")
    lines += ["", _DISCLAIMER]

    await _safe_edit(query, "\n".join(lines), reply_markup=main_menu_keyboard())


# ── Boot-time auto-start ──────────────────────────────────────────────

async def _post_init(app: Application) -> None:
    """Register command menu and start scheduler on boot."""
    await app.bot.set_my_commands([
        ("start", "Welcome & scan buttons"),
        ("scan", "Quick scan buttons"),
        ("autoscan", "Toggle auto-scan on/off"),
        ("status", "Bot status, WS connection, live prices"),
        ("cache", "Show data cache age per pair/TF"),
        ("clearcache", "Wipe data cache, force fresh fetch"),
        ("help", "All commands"),
    ])

    if CHAT_ID:
        _autoscan_chats.add(int(CHAT_ID))

    if not _feed_active():
        logger.info("Neither TELEGRAM_CHAT_ID nor TELEGRAM_CHANNEL_ID set — "
                    "auto-scan disabled on boot. Use /autoscan in chat to enable.")
        return

    if CHANNEL_ID is not None:
        logger.info(f"Subscriber channel feed enabled: {CHANNEL_ID}")

    logger.info("Running initial scan to populate level cache and signal stats...")
    try:
        tasks = [
            asyncio.wait_for(run_signal(sym, Timeframe.M15), timeout=30)
            for sym in SCAN_PAIRS
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        init_reports = [r for r in results if isinstance(r, SignalReport)]
        for i, r in enumerate(results):
            if isinstance(r, Exception):
                logger.error(f"Initial scan failed for {SCAN_PAIRS[i]}: {r}")
        if init_reports:
            _update_level_cache(init_reports)
            logger.info(f"Level cache populated: {sum(len(v) for v in _level_cache.values())} levels")
    except Exception:
        logger.exception("Initial scan failed — WS will wait for first candle scan")

    _start_scheduler(app.bot)

    next_m5 = _seconds_until_boundary(5, buffer=10)
    next_m15 = _seconds_until_boundary(15, buffer=10)
    next_h1 = _seconds_until_boundary(60, buffer=10)
    next_h4 = _seconds_until_boundary(240, buffer=10)
    logger.info(
        f"Auto-scan started (chats={sorted(_autoscan_chats)}, channel={CHANNEL_ID}) — "
        f"M5 {next_m5:.0f}s, M15 {next_m15:.0f}s, H1 {next_h1:.0f}s, H4 {next_h4:.0f}s"
    )

    # Boot message goes to the owner DM only — the subscriber channel
    # receives nothing but signal and breach alerts.
    if OWNER_ID is not None:
        channel_line = "📢 Subscriber channel feed active.\n" if CHANNEL_ID else ""
        try:
            await app.bot.send_message(
                OWNER_ID,
                f"🤖 *ForexBot2 started*\n\n"
                f"Auto-scanning {_PAIRS_LABEL} on M5/M15/H1/H4.\n"
                f"📡 Live price monitoring active.\n"
                f"{channel_line}"
                f"Next: M5 {next_m5 / 60:.0f}m | M15 {next_m15 / 60:.0f}m | "
                f"H1 {next_h1 / 60:.0f}m | H4 {next_h4 / 60:.0f}m\n\n"
                f"/autoscan to stop",
                parse_mode=ParseMode.MARKDOWN,
            )
        except Exception:
            logger.exception("Failed to send boot message")


# ── Graceful shutdown ─────────────────────────────────────────────────

async def _post_shutdown(app: Application) -> None:
    logger.info("Shutting down ForexBot2...")
    _stop_scheduler()
    _executor.shutdown(wait=False)
    logger.info("Shutdown complete")


# ── Entry point ───────────────────────────────────────────────────────

async def _on_error(update: object, context: ContextTypes.DEFAULT_TYPE) -> None:
    """python-telegram-bot routes every unhandled handler exception here;
    without a registered error handler PTB only half-logs them."""
    logger.error(
        "Unhandled error in Telegram handler (update=%s)",
        getattr(update, "update_id", update),
        exc_info=context.error,
    )


def _install_excepthooks() -> None:
    """Route uncaught exceptions from the main thread and worker threads
    into the rotating log file instead of bare stderr, so a crash right
    before exit is still recorded."""
    def hook(exc_type, exc, tb):
        if issubclass(exc_type, KeyboardInterrupt):
            sys.__excepthook__(exc_type, exc, tb)
            return
        logger.critical("Uncaught exception", exc_info=(exc_type, exc, tb))

    def thread_hook(args):
        if issubclass(args.exc_type, SystemExit):
            return
        name = args.thread.name if args.thread else "?"
        logger.critical(
            f"Uncaught exception in thread {name}",
            exc_info=(args.exc_type, args.exc_value, args.exc_traceback))

    sys.excepthook = hook
    threading.excepthook = thread_hook


def main() -> None:
    if not TOKEN or not CHAT_ID:
        raise SystemExit(
            "telegram_bot: missing Telegram credentials. Set telegram.token "
            "and telegram.chat_id in config.json, or the TELEGRAM_TOKEN and "
            "TELEGRAM_CHAT_ID environment variables.")

    _install_excepthooks()
    logger.info("Starting ForexBot2 Telegram bot…")

    app = (
        Application.builder()
        .token(TOKEN)
        .post_init(_post_init)
        .post_shutdown(_post_shutdown)
        .build()
    )

    app.add_handler(CommandHandler("start", cmd_start))
    app.add_handler(CommandHandler("scan", cmd_scan))
    app.add_handler(CommandHandler("help", cmd_help))
    app.add_handler(CommandHandler("status", cmd_status))
    app.add_handler(CommandHandler("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()