Refine Telegram alert behavior

app/services/alert_service.py

@@ -16,6 +16,7 @@ precedence DB > env; the bot token is write-only (never returned on read).
 from __future__ import annotations
 
 import logging
+import math
 from datetime import datetime, timedelta, timezone
 from types import SimpleNamespace
 
@@ -65,6 +66,11 @@ _BOOL_DEFAULTS = {
 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
@@ -100,6 +106,7 @@ QUAD_LABELS = {
 
 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:
@@ -226,6 +233,19 @@ async def _recently_alerted(
     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(
@@ -275,6 +295,25 @@ 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"
@@ -285,6 +324,17 @@ def _fmt_signed_move(from_price: float | int | None, to_price: float | int | Non
     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 "🔴"
@@ -301,7 +351,7 @@ def _format_qualified(s: dict) -> str:
 async def _collect_qualified(db: AsyncSession) -> list[tuple[str, str]]:
     out: list[tuple[str, str]] = []
     for s in await _qualified_setups(db):
-        key = f"qualified:{s['symbol']}:{s['direction']}"
+        key = _qualified_opportunity_key(s)
         out.append((key, _format_qualified(s)))
     return out
 
@@ -428,12 +478,10 @@ async def _collect_digest(db: AsyncSession) -> tuple[str, str] | None:
     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(f"{len(qualified)} qualified setup(s):")
+        lines.append("")
+        lines.append(f"<b>Qualified setups</b> ({len(qualified)})")
         for s in top:
-            lines.append(
-                f"• {s['symbol']} {s['direction'].upper()} "
-                f"R:R {s['rr_ratio']:.1f}:1, conf {(s.get('confidence_score') or 0):.0f}%"
-            )
+            lines.append(_format_qualified(s))
     else:
         lines.append("No qualified setups today.")
 
@@ -443,7 +491,7 @@ async def _collect_digest(db: AsyncSession) -> tuple[str, str] | None:
     open_trades = await paper_trade_service.list_trades(db, status="open")
     if open_trades:
         lines.append("")
-        lines.append(f"💼 <b>{len(open_trades)} open trade(s):</b>")
+        lines.append(f"<b>Open trades</b> ({len(open_trades)})")
         for t in open_trades:
             entry = t["entry_price"]
             cur = t.get("current_price")
@@ -451,7 +499,7 @@ async def _collect_digest(db: AsyncSession) -> tuple[str, str] | None:
             if cur and entry:
                 gain_pct = (cur - entry) / entry * 100.0 * sign
                 gain_usd = (cur - entry) * t["shares"] * sign
-                gain = f"{gain_pct:+.1f}% ({'+' if gain_usd >= 0 else '−'}${abs(gain_usd):.0f})"
+                gain = f"{gain_pct:+.1f}% ({_fmt_signed_money(gain_usd)})"
             else:
                 gain = "n/a"
             ts = t.get("trailing_stop")
@@ -460,7 +508,12 @@ async def _collect_digest(db: AsyncSession) -> tuple[str, str] | None:
                 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"• {t['symbol']} {t['direction'].upper()} {gain} · {stop_txt}")
+            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)
 
@@ -469,21 +522,29 @@ async def _collect_digest(db: AsyncSession) -> tuple[str, str] | None:
 # 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 = per_share * trade.shares
+    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 = f"{'+' if pnl_usd >= 0 else '−'}${abs(pnl_usd):.2f}"
+    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"}.get(
+    reason = {"trailing": "trailing stop", "stop": "stop-loss", "target": "target", "time": "max hold"}.get(
         trade.close_reason or "", trade.close_reason or "closed"
     )
     return (
@@ -493,8 +554,8 @@ def _format_closed_trade(trade: PaperTrade, symbol: str) -> str:
     )
 
 
-async def _collect_closed_trades(db: AsyncSession) -> list[tuple[str, str]]:
-    """One alert per auto-closed paper trade (trailing / stop / target). Manual
+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(
@@ -504,11 +565,63 @@ async def _collect_closed_trades(db: AsyncSession) -> list[tuple[str, str]]:
             PaperTrade.status == "closed",
             PaperTrade.closed_at.is_not(None),
             PaperTrade.closed_at > cutoff,
-            PaperTrade.close_reason.in_(("trailing", "stop", "target")),
+            PaperTrade.close_reason.in_(("trailing", "stop", "target", "time")),
         )
         .order_by(PaperTrade.closed_at.desc())
     )
-    return [(str(trade.id), _format_closed_trade(trade, symbol)) for trade, symbol in result.all()]
+    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."""
+    result = await db.execute(select(PaperTrade))
+    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))
 
 
 # ---------------------------------------------------------------------------
@@ -540,7 +653,26 @@ def _classify_quadrant(x: float, y: float, prev: str | None, margin: float = QUA
     return _bools_to_quadrant(x_high, y_high)
 
 
-async def _last_quadrant(db: AsyncSession) -> tuple[str | None, datetime | None]:
+def _quadrant_log_key(q: str, x: float, y: float) -> str:
+    return f"{q}:{x:.1f}:{y:.1f}"
+
+
+def _parse_quadrant_log_key(key: str | None) -> tuple[str | None, float | None, float | None]:
+    if not key:
+        return None, None, None
+    parts = key.split(":")
+    q = parts[0]
+    if q not in QUAD_LABELS:
+        return None, None, None
+    if len(parts) >= 3:
+        try:
+            return q, float(parts[1]), float(parts[2])
+        except ValueError:
+            pass
+    return q, None, None
+
+
+async def _last_quadrant(db: AsyncSession) -> tuple[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)
@@ -549,7 +681,10 @@ async def _last_quadrant(db: AsyncSession) -> tuple[str | None, datetime | None]
         .limit(1)
     )
     row = result.first()
-    return (row[0], row[1]) if row else (None, None)
+    if not row:
+        return None, None, None, None
+    prev_q, prev_x, prev_y = _parse_quadrant_log_key(row[0])
+    return prev_q, prev_x, prev_y, row[1]
 
 
 async def _collect_regime_quadrant(db: AsyncSession) -> list[tuple[str, str]]:
@@ -570,9 +705,9 @@ async def _collect_regime_quadrant(db: AsyncSession) -> list[tuple[str, str]]:
     if x is None or y is None:
         return []
 
-    prev, prev_time = await _last_quadrant(db)
+    prev, prev_x, prev_y, prev_time = await _last_quadrant(db)
     if prev is None:
-        _log_alert(db, QUAD_TYPE, _classify_quadrant(x, y, None))  # seed, no alert
+        _log_alert(db, QUAD_TYPE, _quadrant_log_key(_classify_quadrant(x, y, None), x, y))  # seed, no alert
         return []
 
     new_q = _classify_quadrant(x, y, prev)
@@ -585,12 +720,19 @@ async def _collect_regime_quadrant(db: AsyncSession) -> list[tuple[str, str]]:
         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"regime {prev_x:.0f} → {x:.0f} ({x - prev_x:+.0f}) · "
+            f"early-warning {prev_y:.0f} → {y:.0f} ({y - prev_y:+.0f})"
+        )
+    else:
+        metrics = f"regime {x:.0f} · early-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"regime {x:.0f} · early-warning {y:.0f}"
+        f"{metrics}"
     )
-    return [(new_q, text)]
+    return [(_quadrant_log_key(new_q, x, y), text)]
 
 
 # ---------------------------------------------------------------------------
@@ -650,11 +792,21 @@ async def dispatch_alerts(db: AsyncSession) -> dict:
 
     signal_outgoing: list[AlertItem] = []
     outgoing: list[AlertItem] = []
+    closed_outgoing: list[ClosedTradeItem] = []
+    qualified_inactive: list[str] = []
 
     if cfg["qualified"]:
-        for key, text in await _collect_qualified(db):
-            if not await _recently_alerted(db, "qualified", key):
+        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):
@@ -677,23 +829,43 @@ async def dispatch_alerts(db: AsyncSession) -> dict:
             outgoing.append((QUAD_TYPE, key, text))
 
     if cfg["trade_closed"]:
-        for key, text in await _collect_closed_trades(db):
+        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):
-                outgoing.append((TRADE_CLOSED_TYPE, key, text))
+                closed_outgoing.append((key, text, pnl_usd))
 
     sent = 0
-    candidates = len(signal_outgoing) + len(outgoing)
-    if signal_outgoing or outgoing:
+    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)
@@ -702,6 +874,9 @@ async def dispatch_alerts(db: AsyncSession) -> dict:
                 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}