"""Email alerts — new approved signals and auto-closed paper trades. SMTP-based (Gmail-ready via app passwords). Enabled only when EMAIL_ALERTS_ENABLED=true and SMTP credentials are set. Alerts are deduped per (symbol, direction, day) via the AlertLog table so repeated scans of the same setup never spam your inbox. All failures are logged, never raised — alerting must never break the scan pipeline. """ import logging import os import smtplib from datetime import date, datetime from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from sqlalchemy.orm import Session from .. import models logger = logging.getLogger(__name__) def alerts_enabled() -> bool: return ( os.getenv("EMAIL_ALERTS_ENABLED", "false").lower() == "true" and bool(os.getenv("SMTP_USER")) and bool(os.getenv("SMTP_PASSWORD")) and bool(os.getenv("ALERT_EMAIL_TO")) ) def telegram_enabled() -> bool: return bool(os.getenv("TELEGRAM_BOT_TOKEN")) and bool(os.getenv("TELEGRAM_CHAT_ID")) def any_alerts_enabled() -> bool: return alerts_enabled() or telegram_enabled() def send_telegram(text: str) -> bool: """Push alert via Telegram bot (create one with @BotFather).""" if not telegram_enabled(): return False try: import httpx r = httpx.post( f"https://api.telegram.org/bot{os.getenv('TELEGRAM_BOT_TOKEN')}/sendMessage", json={"chat_id": os.getenv("TELEGRAM_CHAT_ID"), "text": text[:4000]}, timeout=10, ) r.raise_for_status() return True except Exception as exc: # noqa: BLE001 — alerting must never break the pipeline logger.error("Telegram alert failed: %s", exc) return False def _dispatch(subject: str, text: str, html: str | None = None) -> bool: """Send via every enabled channel; True if at least one succeeded.""" email_ok = send_email(subject, text, html) if alerts_enabled() else False tg_ok = send_telegram(f"{subject}\n\n{text}") if telegram_enabled() else False return email_ok or tg_ok def send_email(subject: str, text: str, html: str | None = None) -> bool: host = os.getenv("SMTP_HOST", "smtp.gmail.com") port = int(os.getenv("SMTP_PORT", "587")) user = os.getenv("SMTP_USER", "") password = os.getenv("SMTP_PASSWORD", "") to_addr = os.getenv("ALERT_EMAIL_TO", "") from_addr = os.getenv("ALERT_EMAIL_FROM", user) msg = MIMEMultipart("alternative") msg["Subject"] = subject msg["From"] = from_addr msg["To"] = to_addr msg.attach(MIMEText(text, "plain")) if html: msg.attach(MIMEText(html, "html")) try: with smtplib.SMTP(host, port, timeout=20) as server: server.starttls() server.login(user, password) server.sendmail(from_addr, [a.strip() for a in to_addr.split(",")], msg.as_string()) logger.info("Alert email sent: %s", subject) return True except Exception as exc: # noqa: BLE001 — alerting must never break the pipeline logger.error("Alert email failed: %s", exc) return False # ── signal alerts ──────────────────────────────────────── def _contract_line(s: models.Signal) -> str: """Human-readable 'which option to buy' line, e.g. 'Buy option: AUG 11 '26 252.5 CALL @ ~$4.20 (delta 0.55, OI 12,400)'.""" c = s.option_contract or {} if not c.get("strike"): return "" bid, ask = float(c.get("bid", 0)), float(c.get("ask", 0)) mid = float(c.get("mid") or 0) or round((bid + ask) / 2, 2) try: exp = datetime.fromisoformat(str(c["expiry"])[:10]).strftime("%b %d '%y").upper() except (KeyError, ValueError): exp = str(c.get("expiry", "?")) return ( f"Buy option: {exp} {c['strike']} {str(c.get('type', '')).upper()} @ ~${mid:.2f} " f"(delta {c.get('delta', '?')}, OI {int(c.get('open_interest', 0)):,})" ) def _signal_lines(s: models.Signal) -> str: lines = ( f"{s.symbol} {s.direction} — confidence {s.confidence:.0f}, " f"P(win) {s.probability:.0f}%, R:R {s.risk_reward:.1f}\n" f" Entry ${s.entry:.2f} | Stop ${s.stop_loss:.2f} | " f"Target 1 ${s.target1:.2f} | Target 2 ${s.target2:.2f}\n" f" Size {s.position_size} shares | Max risk ${s.max_risk_usd:,.0f} | {s.risk_rating} risk" ) contract = _contract_line(s) if contract: lines += f"\n ➜ {contract}" return lines def _signal_html_row(s: models.Signal) -> str: color = "#22c55e" if s.direction in ("CALL", "BUY") else "#ef4444" row = ( f"" f"{s.symbol}" f"{s.direction}" f"{s.confidence:.0f}" f"{s.probability:.0f}%" f"${s.entry:.2f}" f"${s.stop_loss:.2f}" f"${s.target1:.2f}" f"{s.risk_reward:.1f}" f"{s.position_size}" f"" ) contract = _contract_line(s) if contract: row += ( f"➜ {contract}" ) return row def build_signal_email(signals: list[models.Signal]) -> tuple[str, str, str]: """Returns (subject, text, html).""" n = len(signals) tickers = ", ".join(s.symbol for s in signals[:4]) + ("…" if n > 4 else "") subject = f"[Signal Lab] {n} new approved signal{'s' if n != 1 else ''}: {tickers}" text = ( f"AI Trading Signal Lab — {datetime.now().strftime('%b %d, %I:%M %p')}\n" f"{n} new setup(s) passed all risk gates:\n\n" + "\n\n".join(_signal_lines(s) for s in signals) + "\n\nPaper trading research only — not financial advice.\n" "Open http://localhost:3000/signals to review." ) html = ( "
" f"

⚡ {n} new approved signal{'s' if n != 1 else ''}

" f"

{datetime.now().strftime('%B %d, %Y %I:%M %p')} — " "all setups passed the risk gates.

" "" "" "" "" "" "" "" + "".join(_signal_html_row(s) for s in signals) + "
SymbolDirConfP(win)EntryStopTargetR:RSize
" "

" "Paper trading research only — not financial advice. " "Review in dashboard

" ) return subject, text, html def alert_new_signals(db: Session, signals: list[models.Signal]) -> int: """Alert newly approved signals not already alerted today. Returns count sent.""" if not any_alerts_enabled(): return 0 today = date.today() fresh: list[models.Signal] = [] for s in signals: if s.status != "approved": continue already = ( db.query(models.AlertLog) .filter( models.AlertLog.symbol == s.symbol, models.AlertLog.direction == s.direction, models.AlertLog.alert_date == today, models.AlertLog.kind == "signal", ) .first() ) if not already: fresh.append(s) if not fresh: return 0 subject, text, html = build_signal_email(fresh) if _dispatch(subject, text, html): for s in fresh: db.add(models.AlertLog(symbol=s.symbol, direction=s.direction, alert_date=today, kind="signal", signal_id=s.id)) db.commit() return len(fresh) return 0 # ── trade close alerts ─────────────────────────────────── def alert_closed_trades(db: Session, trades: list[models.PaperTrade]) -> int: """Alert trades auto-closed by stop/target/time/scale-out. Returns count sent.""" if not any_alerts_enabled() or not trades: return 0 outcome_labels = { "target": "🎯 TARGET HIT", "stop_loss": "🛑 STOPPED OUT", "time_stop": "⏱ TIME STOP", "scale_out": "💰 SCALED OUT (half banked)", } lines = [] for t in trades: outcome = outcome_labels.get(t.exit_reason or "", "CLOSED") line = ( f"{outcome}: {t.symbol} {t.direction} x{t.quantity} — " f"entry ${t.entry_price:.2f} → exit ${(t.exit_price or 0):.2f} — " f"P&L ${t.pnl:,.2f}" ) if t.instrument == "option" and t.option_contract: c = t.option_contract line += ( f"\n Option: {c.get('expiry')} {c.get('strike')} " f"{str(c.get('type', '')).upper()} x{t.option_contracts} — " f"${(t.option_entry_premium or 0):.2f} → ${(t.option_last_premium or 0):.2f} — " f"option P&L ${t.option_pnl:,.2f}" ) lines.append(line) total = sum(t.pnl for t in trades) subject = f"[Signal Lab] {len(trades)} paper trade(s) closed — P&L ${total:,.2f}" text = ( "Paper trades auto-closed:\n\n" + "\n".join(lines) + "\n\nPaper trading research only — no real money involved." ) if _dispatch(subject, text): today = date.today() for t in trades: db.add(models.AlertLog(symbol=t.symbol, direction=t.direction, alert_date=today, kind="trade_close", signal_id=t.signal_id)) db.commit() return len(trades) return 0