"""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
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
import httpx
from sqlalchemy import 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.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,
KEY_REGIME_QUADRANT: True,
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"
# 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"
# Regime quadrant-change alert: (regime index x early-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 = 40.0 # regime index divider (matches the frontend quadrant)
QUAD_Y_DIV = 60.0 # early-warning divider
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": "① Hot & brittle",
"2": "② Transition",
"3": "③ Healthy & broad",
"4": "④ Real downturn",
}
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
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),
)
)
async def _watermark(db: AsyncSession, symbol: str) -> float | None:
result = await db.execute(
select(AlertLog.value)
.where(AlertLog.alert_type == WATERMARK_TYPE, AlertLog.dedup_key == symbol)
.order_by(AlertLog.created_at.desc())
.limit(1)
)
row = result.first()
return row[0] if row else None
# ---------------------------------------------------------------------------
# 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]:
setups = await get_trade_setups(db)
config = await get_activation_config(db)
return [s for s in setups if setup_qualifies(SimpleNamespace(**s), config)]
def _format_qualified(s: dict) -> str:
prob = best_target_probability(SimpleNamespace(**s))
arrow = "🟢" if s["direction"] == "long" else "🔴"
return (
f"{arrow} {s['symbol']} {s['direction'].upper()} — qualified setup\n"
f"entry {s['entry_price']:.2f} → target {s['target']:.2f} "
f"(R:R {s['rr_ratio']:.1f}:1)\n"
f"confidence {(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 = f"qualified:{s['symbol']}:{s['direction']}"
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
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.
"""
out: list[tuple[str, str]] = []
for tid, symbol in await _watchlist_tickers(db):
price = await _latest_close(db, tid)
if not price:
continue
levels_result = await db.execute(select(SRLevel).where(SRLevel.ticker_id == tid))
levels = [
{"price_level": lv.price_level, "strength": lv.strength, "type": lv.type}
for lv in levels_result.scalars().all()
]
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 - nearest["midpoint"]) / price * 100
if dist_pct > SR_PROXIMITY_PCT:
continue
label = (
f"{nearest['low']:.2f}–{nearest['high']:.2f}"
if nearest["level_count"] > 1
else f"{nearest['midpoint']:.2f}"
)
key = f"sr:{symbol}:{nearest['type']}" # one per side per ticker per cooldown
out.append((
key,
f"📍 {symbol} approaching {nearest['type']} {label} "
f"(now {price:.2f}, {dist_pct:.1f}% away)",
))
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.
"""
out: list[tuple[str, str]] = []
for tid, symbol in await _watchlist_tickers(db):
comp_result = await db.execute(
select(CompositeScore.score).where(CompositeScore.ticker_id == tid)
)
row = comp_result.first()
if row is None or row[0] is None:
continue
current = float(row[0])
base = await _watermark(db, 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"🔻 {symbol} 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"📊 Daily digest — {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):")
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}%"
)
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"💼 {len(open_trades)} open trade(s):")
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}% ({'+' if gain_usd >= 0 else '−'}${abs(gain_usd):.0f})"
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"• {t['symbol']} {t['direction'].upper()} {gain} · {stop_txt}")
return key, "\n".join(lines)
# ---------------------------------------------------------------------------
# Paper-trade close trigger (one summary per auto-closed trade)
# ---------------------------------------------------------------------------
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
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}"
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(
trade.close_reason or "", trade.close_reason or "closed"
)
return (
f"{'✅' if win else '🔴'} {symbol} {trade.direction.upper()} closed ({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[tuple[str, str]]:
"""One alert per auto-closed paper trade (trailing / stop / target). 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")),
)
.order_by(PaperTrade.closed_at.desc())
)
return [(str(trade.id), _format_closed_trade(trade, symbol)) for trade, symbol in result.all()]
# ---------------------------------------------------------------------------
# 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" # ② Transition / ① Hot & brittle
return "4" if x_high else "3" # ④ Real downturn / ③ Healthy & broad
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) -> str:
"""Quadrant of (regime index x, early 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 >= QUAD_X_DIV, y >= QUAD_Y_DIV)
px, py = _quadrant_to_bools(prev)
x_high = (x >= QUAD_X_DIV - margin) if px else (x >= QUAD_X_DIV + margin)
y_high = (y >= QUAD_Y_DIV - margin) if py else (y >= QUAD_Y_DIV + margin)
return _bools_to_quadrant(x_high, y_high)
async def _last_quadrant(db: AsyncSession) -> tuple[str | 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()
return (row[0], row[1]) if row else (None, None)
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_monitor
data = await get_regime_monitor(db)
if not data.get("available"):
return []
x = data.get("total_score")
y = (data.get("early_warning") or {}).get("score")
if x is None or y is None:
return []
prev, prev_time = await _last_quadrant(db)
if prev is None:
_log_alert(db, QUAD_TYPE, _classify_quadrant(x, y, None)) # seed, no alert
return []
new_q = _classify_quadrant(x, y, prev)
if new_q == prev:
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
text = (
f"🧭 Regime quadrant change\n"
f"{QUAD_LABELS.get(prev, prev)} → {QUAD_LABELS.get(new_q, new_q)}\n"
f"regime {x:.0f} · early-warning {y:.0f}"
)
return [(new_q, text)]
# ---------------------------------------------------------------------------
# Dispatch
# ---------------------------------------------------------------------------
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}
outgoing: list[tuple[str, str, str]] = [] # (alert_type, key, text)
if cfg["qualified"]:
for key, text in await _collect_qualified(db):
if not await _recently_alerted(db, "qualified", key):
outgoing.append(("qualified", key, text))
if cfg["sr"]:
for key, text in await _collect_sr_proximity(db):
if not await _recently_alerted(db, "sr_proximity", key):
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):
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 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))
sent = 0
if outgoing:
async with httpx.AsyncClient(timeout=15) as client:
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)
await db.commit() # persist watermark seeds/advances and sent-logs
return {"status": "ok", "sent": sent, "candidates": len(outgoing)}
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"],
"✅ Signal Platform — 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)}