"""Telegram alerts: notify on actionable signals so the dashboard isn't a
poll-only tool.

Triggers (each toggleable):
  - qualified setups: a (symbol, direction) setup that clears the activation gate
  - watchlist S/R proximity: a watched ticker's price entering a strong S/R zone
  - score deterioration: a watched ticker's composite dropping sharply vs a
    running watermark
  - daily digest: one end-of-day summary

Dedup is via the AlertLog table: cooldown-based for the first two and the digest,
watermark-based for score drops. Telegram credentials follow the usual
precedence DB > env; the bot token is write-only (never returned on read).
"""

from __future__ import annotations

import logging
import math
from collections import defaultdict
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace

import httpx
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings
from app.models.alert import AlertLog
from app.models.ohlcv import OHLCVRecord
from app.models.paper_trade import PaperTrade
from app.services.trade_policy import MANUAL_BOOK
from app.models.score import CompositeScore
from app.models.sr_level import SRLevel
from app.models.ticker import Ticker
from app.models.watchlist import WatchlistEntry
from app.services import settings_store
from app.services.admin_service import get_activation_config, update_setting
from app.services.qualification import best_target_probability, setup_qualifies
from app.services.rr_scanner_service import get_trade_setups
from app.services.sr_service import cluster_sr_zones

logger = logging.getLogger(__name__)

# SystemSetting keys
KEY_ENABLED = "alerts_enabled"
KEY_TOKEN = "alerts_telegram_bot_token"
KEY_CHAT_ID = "alerts_telegram_chat_id"
KEY_QUALIFIED = "alerts_qualified_enabled"
KEY_SR = "alerts_sr_proximity_enabled"
KEY_SCORE_DROP = "alerts_score_drop_enabled"
KEY_DIGEST = "alerts_digest_enabled"
KEY_REGIME_QUADRANT = "alerts_regime_quadrant_enabled"
KEY_TRADE_CLOSED = "alerts_trade_closed_enabled"

_BOOL_DEFAULTS = {
    KEY_ENABLED: False,
    KEY_QUALIFIED: True,
    KEY_SR: True,
    KEY_SCORE_DROP: True,
    KEY_DIGEST: True,
    # Experimental human-facing thermometer: opt in explicitly. Existing stored
    # true values remain true; only missing/reset configurations default off.
    KEY_REGIME_QUADRANT: False,
    KEY_TRADE_CLOSED: True,
}

# Paper-trade auto-close alert: catch every close at least once (the job runs
# hourly), then never re-send the same trade (a huge cooldown ≈ once-per-trade).
CLOSED_LOOKBACK_HOURS = 26
CLOSED_ALERT_COOLDOWN_HOURS = 24 * 365 * 5
TRADE_CLOSED_TYPE = "trade_closed"
PAPER_BOOK_STARTING_CAPITAL = 10_000.0
QUALIFIED_TARGET_ZONE_PCT = 1.0
QUALIFIED_STATE_TYPE = "qualified_state"
QUALIFIED_ACTIVE = 1.0
QUALIFIED_INACTIVE = 0.0

# Tunables (kept as constants for now; promote to settings if needed)
SR_PROXIMITY_PCT = 2.0      # within this % of a strong zone → alert
SR_MIN_STRENGTH = 60        # only strong zones are alert-worthy
SR_CLUSTER_TOLERANCE = 0.02  # merge levels within 2% into one zone (matches chart)
SCORE_DROP_POINTS = 15.0    # composite drop vs watermark that triggers an alert
COOLDOWN_HOURS = 72         # don't re-send the same key within this window
DIGEST_HOUR_UTC = 22        # send the daily digest on the first run at/after this hour

WATERMARK_TYPE = "score_watermark"
SIGNAL_BUNDLE_ALERT_TYPES = ("qualified", "sr_proximity", "score_drop")
SIGNAL_BUNDLE_SECTIONS = (
    ("qualified", "Qualified setups"),
    ("sr_proximity", "Near support/resistance"),
    ("score_drop", "Score drops"),
)
SIGNAL_BUNDLE_MAX_CHARS = 3900  # Telegram limit is 4096; keep room for HTML parsing

# Regime quadrant-change alert: (State x Warning) quadrant.
# Hysteresis (a deadband around each divider) stops a point sitting on a boundary
# from flip-flopping; the cooldown caps how often a genuine change can re-alert.
QUAD_TYPE = "regime_quadrant"
QUAD_X_DIV = 50.0           # v3 State divider (backend response is authoritative)
QUAD_Y_DIV = 40.0           # v3 Warning divider; the axes have different ranges
QUAD_MARGIN = 5.0           # half-width of the hysteresis deadband around each divider
QUAD_COOLDOWN_DAYS = 3      # min days between quadrant-change alerts
QUAD_LABELS = {
    "1": "Early warning",
    "2": "Active stress",
    "3": "Healthy",
    "4": "Stressed / stabilizing",
}

AlertItem = tuple[str, str, str]  # alert_type, dedup_key, text
AlertLogRef = tuple[str, str]     # alert_type, dedup_key
ClosedTradeItem = tuple[str, str, float]  # dedup_key, text, pnl_usd


def _as_bool(value: str | None, default: bool) -> bool:
    if value is None:
        return default
    return value.strip().lower() == "true"


async def _resolve(db: AsyncSession) -> dict:
    keys = [KEY_ENABLED, KEY_TOKEN, KEY_CHAT_ID, KEY_QUALIFIED, KEY_SR, KEY_SCORE_DROP, KEY_DIGEST, KEY_REGIME_QUADRANT, KEY_TRADE_CLOSED]
    stored = await settings_store.get_map(db, keys)

    db_token = (stored.get(KEY_TOKEN) or "").strip()
    if db_token:
        token, token_source = db_token, "database"
    elif settings.telegram_bot_token:
        token, token_source = settings.telegram_bot_token, "environment"
    else:
        token, token_source = "", "none"

    chat_id = (stored.get(KEY_CHAT_ID) or "").strip() or (settings.telegram_chat_id or "").strip()

    return {
        "enabled": _as_bool(stored.get(KEY_ENABLED), _BOOL_DEFAULTS[KEY_ENABLED]),
        "token": token,
        "token_source": token_source,
        "chat_id": chat_id,
        "qualified": _as_bool(stored.get(KEY_QUALIFIED), _BOOL_DEFAULTS[KEY_QUALIFIED]),
        "sr": _as_bool(stored.get(KEY_SR), _BOOL_DEFAULTS[KEY_SR]),
        "score_drop": _as_bool(stored.get(KEY_SCORE_DROP), _BOOL_DEFAULTS[KEY_SCORE_DROP]),
        "digest": _as_bool(stored.get(KEY_DIGEST), _BOOL_DEFAULTS[KEY_DIGEST]),
        "regime_quadrant": _as_bool(stored.get(KEY_REGIME_QUADRANT), _BOOL_DEFAULTS[KEY_REGIME_QUADRANT]),
        "trade_closed": _as_bool(stored.get(KEY_TRADE_CLOSED), _BOOL_DEFAULTS[KEY_TRADE_CLOSED]),
    }


async def get_alert_config(db: AsyncSession) -> dict:
    """Public config — never includes the raw bot token."""
    r = await _resolve(db)
    return {
        "enabled": r["enabled"],
        "telegram_chat_id": r["chat_id"],
        "bot_token_configured": bool(r["token"]),
        "bot_token_source": r["token_source"],
        "qualified_enabled": r["qualified"],
        "sr_proximity_enabled": r["sr"],
        "score_drop_enabled": r["score_drop"],
        "digest_enabled": r["digest"],
        "regime_quadrant_enabled": r["regime_quadrant"],
        "trade_closed_enabled": r["trade_closed"],
    }


async def update_alert_config(
    db: AsyncSession,
    *,
    enabled: bool | None = None,
    bot_token: str | None = None,
    telegram_chat_id: str | None = None,
    qualified_enabled: bool | None = None,
    sr_proximity_enabled: bool | None = None,
    score_drop_enabled: bool | None = None,
    digest_enabled: bool | None = None,
    regime_quadrant_enabled: bool | None = None,
    trade_closed_enabled: bool | None = None,
) -> dict:
    """Persist config. An empty/omitted bot_token leaves the stored token intact."""
    bool_updates = {
        KEY_ENABLED: enabled,
        KEY_QUALIFIED: qualified_enabled,
        KEY_SR: sr_proximity_enabled,
        KEY_SCORE_DROP: score_drop_enabled,
        KEY_DIGEST: digest_enabled,
        KEY_REGIME_QUADRANT: regime_quadrant_enabled,
        KEY_TRADE_CLOSED: trade_closed_enabled,
    }
    for key, val in bool_updates.items():
        if val is not None:
            await update_setting(db, key, "true" if val else "false")

    if telegram_chat_id is not None:
        await update_setting(db, KEY_CHAT_ID, telegram_chat_id.strip())

    if bot_token:  # only overwrite when a non-empty token is supplied
        await update_setting(db, KEY_TOKEN, bot_token.strip())

    return await get_alert_config(db)


# ---------------------------------------------------------------------------
# Telegram transport
# ---------------------------------------------------------------------------

async def _send(client: httpx.AsyncClient, token: str, chat_id: str, text: str) -> None:
    resp = await client.post(
        f"https://api.telegram.org/bot{token}/sendMessage",
        json={
            "chat_id": chat_id,
            "text": text,
            "parse_mode": "HTML",
            "disable_web_page_preview": True,
        },
    )
    resp.raise_for_status()


# ---------------------------------------------------------------------------
# Dedup helpers
# ---------------------------------------------------------------------------

async def _recently_alerted(
    db: AsyncSession, alert_type: str, key: str, cooldown_hours: int = COOLDOWN_HOURS
) -> bool:
    cutoff = datetime.now(timezone.utc) - timedelta(hours=cooldown_hours)
    result = await db.execute(
        select(AlertLog.id)
        .where(
            AlertLog.alert_type == alert_type,
            AlertLog.dedup_key == key,
            AlertLog.created_at > cutoff,
        )
        .limit(1)
    )
    return result.first() is not None


async def _latest_qualified_states(db: AsyncSession) -> dict[str, bool]:
    """Latest active/inactive state per qualified setup opportunity."""
    result = await db.execute(
        select(AlertLog.dedup_key, AlertLog.value)
        .where(AlertLog.alert_type == QUALIFIED_STATE_TYPE)
        .order_by(AlertLog.created_at.asc(), AlertLog.id.asc())
    )
    states: dict[str, bool] = {}
    for key, value in result.all():
        states[key] = bool(value and value > 0)
    return states


def _log_alert(db: AsyncSession, alert_type: str, key: str, value: float | None = None) -> None:
    db.add(
        AlertLog(
            alert_type=alert_type,
            dedup_key=key,
            value=value,
            created_at=datetime.now(timezone.utc),
        )
    )


# ---------------------------------------------------------------------------
# Trigger collectors
# ---------------------------------------------------------------------------

async def _watchlist_tickers(db: AsyncSession) -> list[tuple[int, str]]:
    """Distinct tickers across all watchlists (single-user app → one chat)."""
    result = await db.execute(
        select(WatchlistEntry.ticker_id, Ticker.symbol)
        .join(Ticker, WatchlistEntry.ticker_id == Ticker.id)
        .where(WatchlistEntry.entry_type != "dismissed")
        .distinct()
    )
    return [(tid, sym) for tid, sym in result.all()]


async def _qualified_setups(db: AsyncSession) -> list[dict]:
    # live_recommendation: gate and format on current score/sentiment context,
    # not the values frozen into the setup at scan time.
    setups = await get_trade_setups(
        db,
        live_recommendation=True,
        exclude_open_trade_tickers=True,
        exclude_reentry_gate_locked_tickers=True,
    )
    config = await get_activation_config(db)
    return [s for s in setups if setup_qualifies(SimpleNamespace(**s), config)]


def _fmt_price(value: float | int | None) -> str:
    return "n/a" if value is None else f"{float(value):.2f}"


def _fmt_money(value: float | int | None) -> str:
    if value is None:
        return "n/a"
    return f"${float(value):,.2f}"


def _fmt_signed_money(value: float | int | None) -> str:
    if value is None:
        return "n/a"
    amount = float(value)
    return f"{'+' if amount >= 0 else '-'}${abs(amount):,.2f}"


def _fmt_signed_pct(value: float | int | None) -> str:
    if value is None:
        return "n/a"
    return f"{float(value):+.1f}%"


def _fmt_signed_move(from_price: float | int | None, to_price: float | int | None) -> str:
    if from_price is None or to_price is None:
        return "n/a"
    from_float = float(from_price)
    if from_float == 0:
        return "n/a"
    pct = (float(to_price) - from_float) / from_float * 100.0
    return f"{pct:+.1f}%"


def _qualified_opportunity_key(s: dict) -> str:
    """Stable key for one alert per trade opportunity, not per scanner row."""
    target = s.get("target")
    if target is None or float(target) <= 0:
        zone = "unknown"
    else:
        step = math.log1p(QUALIFIED_TARGET_ZONE_PCT / 100.0)
        zone = str(round(math.log(float(target)) / step))
    return f"qualified:{s['symbol']}:{s['direction']}:target-zone:{zone}"


def _format_qualified(s: dict) -> str:
    prob = best_target_probability(SimpleNamespace(**s))
    arrow = "🟢" if s["direction"] == "long" else "🔴"
    current = s.get("current_price") or s.get("entry_price")
    return (
        f"{arrow} <b>{s['symbol']} {s['direction'].upper()}</b> | "
        f"now {_fmt_price(current)} | entry {_fmt_price(s['entry_price'])} | "
        f"target {_fmt_price(s['target'])} ({_fmt_signed_move(current, s['target'])}) | "
        f"stop {_fmt_price(s['stop_loss'])} | R:R {s['rr_ratio']:.1f} | "
        f"conf {(s.get('confidence_score') or 0):.0f}% | P(target) {prob:.0f}%"
    )


async def _collect_qualified(db: AsyncSession) -> list[tuple[str, str]]:
    out: list[tuple[str, str]] = []
    for s in await _qualified_setups(db):
        key = _qualified_opportunity_key(s)
        out.append((key, _format_qualified(s)))
    return out


async def _latest_close(db: AsyncSession, ticker_id: int) -> float | None:
    result = await db.execute(
        select(OHLCVRecord.close)
        .where(OHLCVRecord.ticker_id == ticker_id)
        .order_by(OHLCVRecord.date.desc())
        .limit(1)
    )
    row = result.first()
    return float(row[0]) if row else None


def _sr_zone_label(zone: dict) -> str:
    return (
        f"{zone['low']:.2f}–{zone['high']:.2f}"
        if zone["level_count"] > 1
        else f"{zone['midpoint']:.2f}"
    )


def _sr_touch_price(zone: dict, current_price: float) -> float:
    low = float(zone["low"])
    high = float(zone["high"])
    if current_price < low:
        return low
    if current_price > high:
        return high
    return float(zone["midpoint"])


def _format_sr_proximity(symbol: str, zone: dict, current_price: float) -> str:
    touch_price = _sr_touch_price(zone, current_price)
    return (
        f"📍 <b>{symbol}</b> {zone['type']} | "
        f"now {_fmt_price(current_price)} -> {_sr_zone_label(zone)} "
        f"({_fmt_signed_move(current_price, touch_price)}) | "
        f"strength {float(zone['strength']):.0f}"
    )


async def _collect_sr_proximity(db: AsyncSession) -> list[tuple[str, str]]:
    """One alert per watchlist ticker for the NEAREST strong S/R zone within range.

    Levels are merged into zones with the same clusterer the chart uses, so a
    cluster of near-duplicate levels (e.g. 183 + 185) is a single zone and a
    single alert. Scoped to the watchlist only — qualified tickers already get
    their own 'qualified setup' alert, so S/R on them would be redundant.
    """
    watchlist = await _watchlist_tickers(db)
    if not watchlist:
        return []

    ticker_ids = [ticker_id for ticker_id, _ in watchlist]
    latest_dates = (
        select(
            OHLCVRecord.ticker_id,
            func.max(OHLCVRecord.date).label("latest_date"),
        )
        .where(OHLCVRecord.ticker_id.in_(ticker_ids))
        .group_by(OHLCVRecord.ticker_id)
        .subquery()
    )
    prices_result = await db.execute(
        select(OHLCVRecord.ticker_id, OHLCVRecord.close).join(
            latest_dates,
            (OHLCVRecord.ticker_id == latest_dates.c.ticker_id)
            & (OHLCVRecord.date == latest_dates.c.latest_date),
        )
    )
    prices = {ticker_id: float(close) for ticker_id, close in prices_result.all()}

    levels_result = await db.execute(
        select(SRLevel).where(SRLevel.ticker_id.in_(ticker_ids))
    )
    levels_by_ticker: dict[int, list[dict]] = defaultdict(list)
    for level in levels_result.scalars():
        levels_by_ticker[level.ticker_id].append(
            {
                "price_level": level.price_level,
                "strength": level.strength,
                "type": level.type,
            }
        )

    out: list[tuple[str, str]] = []
    for tid, symbol in watchlist:
        price = prices.get(tid)
        if not price:
            continue

        levels = levels_by_ticker[tid]
        if not levels:
            continue

        zones = cluster_sr_zones(levels, price, tolerance=SR_CLUSTER_TOLERANCE)
        strong = [z for z in zones if z["strength"] >= SR_MIN_STRENGTH]
        if not strong:
            continue

        # Nearest strong zone only.
        nearest = min(strong, key=lambda z: abs(price - z["midpoint"]))
        dist_pct = abs(price - _sr_touch_price(nearest, price)) / price * 100
        if dist_pct > SR_PROXIMITY_PCT:
            continue

        key = f"sr:{symbol}:{nearest['type']}"  # one per side per ticker per cooldown
        out.append((key, _format_sr_proximity(symbol, nearest, price)))
    return out


async def _collect_score_drops(db: AsyncSession) -> list[tuple[str, str]]:
    """Returns drop messages and (as a side effect) advances watermarks.

    Watermark = the reference composite. Alert when current drops
    SCORE_DROP_POINTS below it, then rebaseline to current so a single slide
    doesn't re-fire; let the watermark rise with the score so the next drop is
    measured from the new high.
    """
    watchlist = await _watchlist_tickers(db)
    if not watchlist:
        return []

    ticker_ids = [ticker_id for ticker_id, _ in watchlist]
    symbols = [symbol for _, symbol in watchlist]
    scores_result = await db.execute(
        select(CompositeScore.ticker_id, CompositeScore.score).where(
            CompositeScore.ticker_id.in_(ticker_ids)
        )
    )
    scores = {ticker_id: float(score) for ticker_id, score in scores_result.all()}

    ranked_watermarks = (
        select(
            AlertLog.dedup_key,
            AlertLog.value,
            func.row_number()
            .over(
                partition_by=AlertLog.dedup_key,
                order_by=(AlertLog.created_at.desc(), AlertLog.id.desc()),
            )
            .label("rank"),
        )
        .where(
            AlertLog.alert_type == WATERMARK_TYPE,
            AlertLog.dedup_key.in_(symbols),
        )
        .subquery()
    )
    watermarks_result = await db.execute(
        select(ranked_watermarks.c.dedup_key, ranked_watermarks.c.value).where(
            ranked_watermarks.c.rank == 1
        )
    )
    watermarks = {
        symbol: float(value)
        for symbol, value in watermarks_result.all()
        if value is not None
    }

    out: list[tuple[str, str]] = []
    for tid, symbol in watchlist:
        current = scores.get(tid)
        if current is None:
            continue

        base = watermarks.get(symbol)
        if base is None:
            _log_alert(db, WATERMARK_TYPE, symbol, value=current)  # seed, no alert
            continue
        if current <= base - SCORE_DROP_POINTS:
            out.append((
                f"scoredrop:{symbol}",
                f"🔻 <b>{symbol}</b> composite score fell to {current:.0f} (from {base:.0f})",
            ))
            _log_alert(db, WATERMARK_TYPE, symbol, value=current)  # rebaseline
        elif current > base:
            _log_alert(db, WATERMARK_TYPE, symbol, value=current)  # track the rise
    return out


async def _collect_digest(db: AsyncSession) -> tuple[str, str] | None:
    now = datetime.now(timezone.utc)
    if now.hour < DIGEST_HOUR_UTC:
        return None
    key = f"digest:{now.date().isoformat()}"
    if await _recently_alerted(db, "digest", key, cooldown_hours=20):
        return None

    qualified = await _qualified_setups(db)
    lines = [f"📊 <b>Daily digest</b> — {now.date().isoformat()}"]
    if qualified:
        top = sorted(qualified, key=lambda s: s["rr_ratio"], reverse=True)[:5]
        lines.append("")
        lines.append(f"<b>Qualified setups</b> ({len(qualified)})")
        for s in top:
            lines.append(_format_qualified(s))
    else:
        lines.append("No qualified setups today.")

    # Open paper trades: unrealized gain + the live trailing stop and how far away.
    from app.services import paper_trade_service

    open_trades = await paper_trade_service.list_trades(db, status="open")
    if open_trades:
        lines.append("")
        lines.append(f"<b>Open trades</b> ({len(open_trades)})")
        for t in open_trades:
            entry = t["entry_price"]
            cur = t.get("current_price")
            sign = 1.0 if t["direction"] == "long" else -1.0
            if cur and entry:
                gain_pct = (cur - entry) / entry * 100.0 * sign
                gain_usd = (cur - entry) * t["shares"] * sign
                gain = f"{gain_pct:+.1f}% ({_fmt_signed_money(gain_usd)})"
            else:
                gain = "n/a"
            ts = t.get("trailing_stop")
            if ts is not None:
                dist = t.get("trailing_distance_pct")
                stop_txt = f"trail {ts:.2f}" + (f" ({dist:.1f}% away)" if dist is not None else "")
            else:
                stop_txt = f"stop {t['stop_loss']:.2f}"
            lines.append(
                f"💼 <b>{t['symbol']} {t['direction'].upper()} open</b> | "
                f"now {_fmt_price(cur)} | entry {_fmt_price(entry)} | "
                f"target {_fmt_price(t.get('target'))} ({_fmt_signed_move(cur, t.get('target'))}) | "
                f"{stop_txt} | P&L {gain}"
            )

    return key, "\n".join(lines)


# ---------------------------------------------------------------------------
# Paper-trade close trigger (one summary per auto-closed trade)
# ---------------------------------------------------------------------------

def _closed_trade_pnl(trade: PaperTrade) -> float:
    sign = 1.0 if trade.direction == "long" else -1.0
    entry = trade.entry_price
    exit_price = trade.close_price if trade.close_price is not None else entry
    per_share = (exit_price - entry) * sign
    return per_share * trade.shares


def _format_closed_trade(trade: PaperTrade, symbol: str) -> str:
    sign = 1.0 if trade.direction == "long" else -1.0
    entry = trade.entry_price
    exit_price = trade.close_price if trade.close_price is not None else entry
    per_share = (exit_price - entry) * sign
    pnl_pct = (per_share / entry * 100.0) if entry else 0.0
    pnl_usd = _closed_trade_pnl(trade)
    risk = abs(entry - trade.stop_loss)
    r_mult = (per_share / risk) if risk > 0 else None
    win = per_share > 0
    money = _fmt_signed_money(pnl_usd)
    r_txt = f" · {r_mult:+.2f}R" if r_mult is not None else ""
    days = (trade.closed_at - trade.opened_at).days if (trade.closed_at and trade.opened_at) else None
    held = f" · held {days}d" if days is not None else ""
    reason = {"trailing": "trailing stop", "stop": "stop-loss", "target": "target", "time": "max hold"}.get(
        trade.close_reason or "", trade.close_reason or "closed"
    )
    return (
        f"{'✅' if win else '🔴'} <b>{symbol} {trade.direction.upper()} closed</b> ({reason})\n"
        f"{pnl_pct:+.1f}% · {money}{r_txt}{held}\n"
        f"{entry:.2f} → {exit_price:.2f}"
    )


async def _collect_closed_trades(db: AsyncSession) -> list[ClosedTradeItem]:
    """One alert item per auto-closed paper trade. Manual
    closes are skipped — you already know about those. Dedup is by trade id."""
    cutoff = datetime.now(timezone.utc) - timedelta(hours=CLOSED_LOOKBACK_HOURS)
    result = await db.execute(
        select(PaperTrade, Ticker.symbol)
        .join(Ticker, PaperTrade.ticker_id == Ticker.id)
        .where(
            PaperTrade.status == "closed",
            PaperTrade.closed_at.is_not(None),
            PaperTrade.closed_at > cutoff,
            PaperTrade.close_reason.in_(("trailing", "stop", "target", "time")),
            # Your own positions only — shadow trades are a research record, not
            # something you hold, and mixing them in unlabelled reads as if you
            # were stopped out of a name you never took.
            PaperTrade.book == MANUAL_BOOK,
        )
        .order_by(PaperTrade.closed_at.desc())
    )
    return [
        (str(trade.id), _format_closed_trade(trade, symbol), _closed_trade_pnl(trade))
        for trade, symbol in result.all()
    ]


async def _paper_book_value(db: AsyncSession) -> float:
    """Paper-trade equity: fixed capital plus realized/unrealized P&L.

    Discretionary book only — the shadow book runs on its own notional equity
    and folding it in would report a number matching neither book.
    """
    result = await db.execute(
        select(PaperTrade).where(PaperTrade.book == MANUAL_BOOK)
    )
    trades = list(result.scalars().all())
    latest: dict[int, float | None] = {}
    for trade in trades:
        if trade.status == "open" and trade.ticker_id not in latest:
            latest[trade.ticker_id] = await _latest_close(db, trade.ticker_id)

    total_pnl = 0.0
    for trade in trades:
        ref = trade.close_price if trade.status == "closed" else latest.get(trade.ticker_id)
        if ref is None:
            ref = trade.entry_price
        sign = 1.0 if trade.direction == "long" else -1.0
        pnl = (float(ref) - trade.entry_price) * trade.shares * sign
        total_pnl += pnl
    return PAPER_BOOK_STARTING_CAPITAL + total_pnl


def _closed_trade_bundle(
    items: list[ClosedTradeItem],
    *,
    current_book_value: float | None,
) -> tuple[list[AlertLogRef], str] | None:
    if not items:
        return None
    total_pnl = sum(item[2] for item in items)
    previous_book_value = (
        current_book_value - total_pnl
        if current_book_value is not None
        else None
    )
    change_pct = (
        total_pnl / previous_book_value * 100.0
        if previous_book_value not in (None, 0)
        else None
    )
    lines = [f"💼 <b>Paper trades closed</b> — {len(items)} trade(s)"]
    if current_book_value is not None and previous_book_value is not None:
        lines.append(
            f"Paper book {_fmt_money(previous_book_value)} → {_fmt_money(current_book_value)} "
            f"({_fmt_signed_money(total_pnl)}, {_fmt_signed_pct(change_pct)})"
        )
    lines.append("")
    lines.append("\n\n".join(item[1] for item in items))
    return ([(TRADE_CLOSED_TYPE, item[0]) for item in items], "\n".join(lines))


# ---------------------------------------------------------------------------
# Regime quadrant-change trigger (hysteresis + cooldown)
# ---------------------------------------------------------------------------

def _bools_to_quadrant(x_high: bool, y_high: bool) -> str:
    if y_high:
        return "2" if x_high else "1"   # Active stress / Early warning
    return "4" if x_high else "3"       # Stressed/stabilizing / Healthy


def _quadrant_to_bools(q: str) -> tuple[bool, bool]:
    return {"1": (False, True), "2": (True, True), "3": (False, False), "4": (True, False)}[q]


def _classify_quadrant(
    x: float,
    y: float,
    prev: str | None,
    margin: float = QUAD_MARGIN,
    x_div: float = QUAD_X_DIV,
    y_div: float = QUAD_Y_DIV,
) -> str:
    """Quadrant of (State x, Warning y), with per-axis hysteresis.

    Each axis only flips once the value crosses its divider by ``margin`` in the
    new direction, so a point parked on a divider keeps its current quadrant
    instead of flip-flopping. ``prev`` None means a fresh (no-hysteresis) classify.
    """
    if prev is None:
        return _bools_to_quadrant(x >= x_div, y >= y_div)
    px, py = _quadrant_to_bools(prev)
    x_high = (x >= x_div - margin) if px else (x >= x_div + margin)
    y_high = (y >= y_div - margin) if py else (y >= y_div + margin)
    return _bools_to_quadrant(x_high, y_high)


def _quadrant_log_key(q: str, x: float, y: float, basket_hash: str | None = None) -> str:
    return f"{basket_hash or 'legacy'}:{q}:{x:.1f}:{y:.1f}"


def _parse_quadrant_log_key(
    key: str | None,
) -> tuple[str | None, str | None, float | None, float | None]:
    if not key:
        return None, None, None, None
    parts = key.split(":")
    if parts[0] in QUAD_LABELS:
        basket_hash, q, values = None, parts[0], parts[1:]
    elif len(parts) >= 2:
        basket_hash, q, values = parts[0], parts[1], parts[2:]
    else:
        return None, None, None, None
    if q not in QUAD_LABELS:
        return None, None, None, None
    if len(values) >= 2:
        try:
            return basket_hash, q, float(values[0]), float(values[1])
        except ValueError:
            pass
    return basket_hash, q, None, None


async def _last_quadrant(
    db: AsyncSession,
) -> tuple[str | None, str | None, float | None, float | None, datetime | None]:
    """Most recently logged quadrant (and when), our baseline for change + cooldown."""
    result = await db.execute(
        select(AlertLog.dedup_key, AlertLog.created_at)
        .where(AlertLog.alert_type == QUAD_TYPE)
        .order_by(AlertLog.created_at.desc())
        .limit(1)
    )
    row = result.first()
    if not row:
        return None, None, None, None, None
    basket_hash, prev_q, prev_x, prev_y = _parse_quadrant_log_key(row[0])
    return basket_hash, prev_q, prev_x, prev_y, row[1]


async def _collect_regime_quadrant(db: AsyncSession) -> list[tuple[str, str]]:
    """Alert once when the regime quadrant changes (hysteresis + cooldown).

    Seeds silently on first run. Thereafter alerts only when the
    hysteresis-confirmed quadrant differs from the last logged one AND the
    cooldown has elapsed. The dispatch loop logs the new quadrant on send, which
    becomes the next baseline and resets the cooldown clock.
    """
    from app.services.regime_monitor_service import get_regime_history, get_regime_monitor

    data = await get_regime_monitor(db)
    if not data.get("available"):
        return []
    state = data.get("state") or {}
    warning = data.get("warning") or {}
    x = state.get("score")
    y = warning.get("score")
    if x is None or y is None:
        return []

    quality = data.get("data_quality") or {}
    if (
        float(state.get("coverage") or 0) < 75
        or float(warning.get("coverage") or 0) < 75
        or not quality.get("is_fresh")
    ):
        return []

    quadrant_cfg = data.get("quadrant_config") or {}
    x_div = float(quadrant_cfg.get("state_divider", QUAD_X_DIV))
    y_div = float(quadrant_cfg.get("warning_divider", QUAD_Y_DIV))
    margin = float(quadrant_cfg.get("margin", QUAD_MARGIN))
    basket_hash = str((data.get("basket") or {}).get("hash") or "unknown")

    prev_hash, prev, prev_x, prev_y, prev_time = await _last_quadrant(db)
    if prev is None or prev_hash != basket_hash:
        seed = _classify_quadrant(x, y, None, margin, x_div, y_div)
        _log_alert(db, QUAD_TYPE, _quadrant_log_key(seed, x, y, basket_hash))
        return []

    new_q = _classify_quadrant(x, y, prev, margin, x_div, y_div)
    if new_q == prev:
        return []

    history = await get_regime_history(db, days=14)
    valid = [
        point for point in history
        if point.get("state") is not None
        and point.get("warning") is not None
        and float(point.get("state_coverage") or 0) >= 75
        and float(point.get("warning_coverage") or 0) >= 75
    ]
    if len(valid) < 2:
        return []
    prior = valid[-2]
    prior_q = _classify_quadrant(
        float(prior["state"]),
        float(prior["warning"]),
        prev,
        margin,
        x_div,
        y_div,
    )
    if prior_q != new_q:
        return []

    if prev_time is not None:
        if prev_time.tzinfo is None:
            prev_time = prev_time.replace(tzinfo=timezone.utc)
        if datetime.now(timezone.utc) - prev_time < timedelta(days=QUAD_COOLDOWN_DAYS):
            return []  # genuine change, but inside the cooldown — stay quiet

    if prev_x is not None and prev_y is not None:
        metrics = (
            f"State {prev_x:.0f} → {x:.0f} ({x - prev_x:+.0f}) · "
            f"Warning {prev_y:.0f} → {y:.0f} ({y - prev_y:+.0f})"
        )
    else:
        metrics = f"State {x:.0f} · Warning {y:.0f}"
    text = (
        f"🧭 <b>Regime quadrant change</b>\n"
        f"{QUAD_LABELS.get(prev, prev)} → {QUAD_LABELS.get(new_q, new_q)}\n"
        f"{metrics}\n"
        f"coverage: state {state.get('coverage'):.0f}% / warning {warning.get('coverage'):.0f}%\n"
        f"<i>Risk thermometer - not a trade signal.</i>"
    )
    return [(_quadrant_log_key(new_q, x, y, basket_hash), text)]


# ---------------------------------------------------------------------------
# Dispatch
# ---------------------------------------------------------------------------

def _signal_bundle_messages(items: list[AlertItem]) -> list[tuple[list[AlertLogRef], str]]:
    if not items:
        return []

    by_type: dict[str, list[AlertItem]] = {key: [] for key in SIGNAL_BUNDLE_ALERT_TYPES}
    for item in items:
        by_type.setdefault(item[0], []).append(item)

    total = sum(len(group) for group in by_type.values())
    header = f"📣 <b>Signal run</b> — {total} new alert(s)"
    bundles: list[tuple[list[AlertLogRef], str]] = []
    lines = [header]
    logs: list[AlertLogRef] = []
    current_section: str | None = None

    def flush() -> None:
        nonlocal lines, logs, current_section
        if logs:
            bundles.append((logs.copy(), "\n".join(lines)))
        lines = [f"{header} (continued)"]
        logs = []
        current_section = None

    for alert_type, section_title in SIGNAL_BUNDLE_SECTIONS:
        for item_type, key, text in by_type.get(alert_type, []):
            block: list[str] = []
            if current_section != alert_type:
                block.extend(["", f"<b>{section_title}</b>"])
            block.append(text)

            if logs and len("\n".join(lines + block)) > SIGNAL_BUNDLE_MAX_CHARS:
                flush()
                block = ["", f"<b>{section_title}</b>", text]

            lines.extend(block)
            logs.append((item_type, key))
            current_section = alert_type

    if logs:
        bundles.append((logs.copy(), "\n".join(lines)))
    return bundles


async def dispatch_alerts(db: AsyncSession) -> dict:
    """Gather all enabled triggers, dedup, and push to Telegram. Job entrypoint."""
    cfg = await _resolve(db)
    if not cfg["enabled"]:
        return {"status": "disabled", "sent": 0}
    if not cfg["token"] or not cfg["chat_id"]:
        return {"status": "no_credentials", "sent": 0}

    signal_outgoing: list[AlertItem] = []
    outgoing: list[AlertItem] = []
    closed_outgoing: list[ClosedTradeItem] = []
    qualified_inactive: list[str] = []

    if cfg["qualified"]:
        previous_qualified_states = await _latest_qualified_states(db)
        qualified_items = await _collect_qualified(db)
        current_qualified_keys = {key for key, _ in qualified_items}
        for key, text in qualified_items:
            if not previous_qualified_states.get(key, False):
                signal_outgoing.append(("qualified", key, text))
        qualified_inactive = [
            key
            for key, active in previous_qualified_states.items()
            if active and key not in current_qualified_keys
        ]

    if cfg["sr"]:
        for key, text in await _collect_sr_proximity(db):
            if not await _recently_alerted(db, "sr_proximity", key):
                signal_outgoing.append(("sr_proximity", key, text))

    if cfg["score_drop"]:
        # also seeds/advances watermarks as a side effect
        for key, text in await _collect_score_drops(db):
            signal_outgoing.append(("score_drop", key, text))

    if cfg["digest"]:
        digest = await _collect_digest(db)
        if digest is not None:
            outgoing.append(("digest", digest[0], digest[1]))

    if cfg["regime_quadrant"]:
        # cooldown/hysteresis handled in the collector (like score drops)
        for key, text in await _collect_regime_quadrant(db):
            outgoing.append((QUAD_TYPE, key, text))

    if cfg["trade_closed"]:
        for key, text, pnl_usd in await _collect_closed_trades(db):
            if not await _recently_alerted(db, TRADE_CLOSED_TYPE, key, cooldown_hours=CLOSED_ALERT_COOLDOWN_HOURS):
                closed_outgoing.append((key, text, pnl_usd))

    sent = 0
    candidates = len(signal_outgoing) + len(outgoing) + len(closed_outgoing)
    closed_bundle = (
        _closed_trade_bundle(
            closed_outgoing,
            current_book_value=await _paper_book_value(db),
        )
        if closed_outgoing
        else None
    )
    if signal_outgoing or outgoing or closed_bundle:
        async with httpx.AsyncClient(timeout=15) as client:
            for log_refs, text in _signal_bundle_messages(signal_outgoing):
                try:
                    await _send(client, cfg["token"], cfg["chat_id"], text)
                    for alert_type, key in log_refs:
                        _log_alert(db, alert_type, key)
                        if alert_type == "qualified":
                            _log_alert(db, QUALIFIED_STATE_TYPE, key, value=QUALIFIED_ACTIVE)
                    sent += 1
                except Exception:
                    logger.exception("Failed to send signal alert bundle")

            if closed_bundle is not None:
                log_refs, text = closed_bundle
                try:
                    await _send(client, cfg["token"], cfg["chat_id"], text)
                    for alert_type, key in log_refs:
                        _log_alert(db, alert_type, key)
                    sent += 1
                except Exception:
                    logger.exception("Failed to send trade-closed alert bundle")

            for alert_type, key, text in outgoing:
                try:
                    await _send(client, cfg["token"], cfg["chat_id"], text)
                    _log_alert(db, alert_type, key)
                    sent += 1
                except Exception:
                    logger.exception("Failed to send alert %s", key)

    for key in qualified_inactive:
        _log_alert(db, QUALIFIED_STATE_TYPE, key, value=QUALIFIED_INACTIVE)

    await db.commit()  # persist watermark seeds/advances and sent-logs
    return {"status": "ok", "sent": sent, "candidates": candidates}


async def send_test_alert(db: AsyncSession) -> dict:
    """Send a fixed message to verify Telegram credentials."""
    cfg = await _resolve(db)
    if not cfg["token"] or not cfg["chat_id"]:
        return {"ok": False, "error": "Bot token and chat ID must both be configured."}
    try:
        async with httpx.AsyncClient(timeout=15) as client:
            await _send(
                client, cfg["token"], cfg["chat_id"],
                "✅ <b>Signal Platform</b> — test alert. Notifications are wired up correctly.",
            )
        return {"ok": True}
    except Exception as exc:
        logger.warning("Test alert failed: %s", exc)
        return {"ok": False, "error": str(exc)}