Telegram bot: gated push alerts + standalone Ollama Q&A

C1 — runtime/notifier.py: gated, fire-and-forget push alerts that NEVER raise
into the loop. Decoupled by POLLING the event store (no notifier threaded
through breaker/exit/execution). Pushes on major events only: daily-loss
breaker, catastrophe-stop, naked-stop flatten, any |closed trade| > $300, and a
once-per-day EOD summary. Wired as a 45s scheduler step + EOD summary/failure
hooks in step_eod_flatten. Boot ping on start. Silent no-op when unconfigured.

C2 — telegram_bot.py: standalone (separate process) READ-ONLY Q&A listener,
locked to TELEGRAM_CHAT_ID. Long-polls getUpdates; gathers a Postgres data
snapshot (today P&L, open positions, recent trades, all-time) and asks the
local Ollama model. Falls back to the raw snapshot if Ollama is down/disabled;
refuses any action command (buy/sell/flatten) — questions only.

Tests: disabled no-op, major-event dedup, EOD-once, auth gate, action-refusal,
ollama-down + ollama-disabled fallback. 8/8 green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

run_live_paper.py

@@ -814,6 +814,20 @@ def main(argv: list[str] | None = None) -> int:
             "path — it will not affect live entries (see entry-path-fast-vs-auto)."
         )
 
+    # Telegram push alerts — gated, decoupled (polls the event store), and
+    # fire-and-forget so it can never impede trading.
+    from runtime.notifier import TelegramNotifier, send_telegram
+    notifier = TelegramNotifier(rt["store"])
+    if notifier.enabled:
+        print("[boot] telegram alerts: ON")
+        send_telegram("\U0001F916 Momentum loop started — push alerts armed.")
+    else:
+        print("[boot] telegram alerts: off (TELEGRAM_BOT_TOKEN/CHAT_ID unset)")
+
+    def step_notify():
+        n = notifier.poll()
+        return f"telegram: {n} alert(s)" if n else None
+
     def step_manage_exits():
         if exit_mgr is None:
             return None
@@ -866,7 +880,15 @@ def main(argv: list[str] | None = None) -> int:
             gap = days_to_next_session(now_et.date())
             sev = f"‼ NAKED-CARRY RISK ({gap}-day gap ahead): " if pre_closure else ""
             print(f"[{datetime.now():%H:%M:%S}] {sev}{tag} could not flatten: {res['errors']}", flush=True)
+            send_telegram(f"⚠️ {tag} FAILED to flatten: {res['errors']} — {sev or 'will retry'}")
             return f"{sev}{tag} closed [{closed}] errors={res['errors']} (will retry)"
+        # success — push the once-per-day EOD P&L summary
+        try:
+            from storage.projections import query_session_pnl
+            notifier.send_eod_summary(now_et.date().isoformat(),
+                                      query_session_pnl(rt["store"]))
+        except Exception:  # noqa: BLE001
+            pass
         return f"{tag} closed [{closed}]"
 
     # --- open catch-up flatten: a day-trading book should be FLAT by the open;
@@ -955,6 +977,8 @@ def main(argv: list[str] | None = None) -> int:
                   enabled=_flag("TRADING_EXECUTION_ENABLED", "1"))
     scheduler.add("eod", step_eod_flatten, float(os.environ.get("TRADING_EOD_INTERVAL_SECONDS", "30")),
                   enabled=_flag("TRADING_EXECUTION_ENABLED", "1"))
+    scheduler.add("notify", step_notify, float(os.environ.get("TELEGRAM_NOTIFY_INTERVAL_SECONDS", "45")),
+                  enabled=notifier.enabled)
     scheduler.add("catchup", step_open_catchup_flatten,
                   float(os.environ.get("TRADING_OPEN_CATCHUP_INTERVAL_SECONDS", "30")),
                   enabled=_flag("TRADING_EXECUTION_ENABLED", "1"))

runtime/notifier.py

@@ -0,0 +1,138 @@
+"""Telegram push notifier — gated alerts on MAJOR trading events.
+
+Fully gated like the Ollama integration: if TELEGRAM_BOT_TOKEN / TELEGRAM_CHAT_ID
+are unset, every call is a SILENT no-op that NEVER raises into the trading loop.
+
+Decoupled by design: it POLLS the event store for major events rather than
+threading a notifier through the breaker / exit / execution subsystems, so it
+cannot slow or break trading. One small scheduler step calls ``poll()``.
+
+Major events (the only things worth a phone buzz):
+  - daily-loss breaker trip          (risk_rule_triggered rule_type=daily_loss)
+  - catastrophe-stop exit            (rule_type=exit_catastrophe)
+  - naked-stop enforcement / flatten (rule_type=exit_naked_stop)
+  - any closed trade with |realized| > BIG_TRADE_USD
+  - a once-per-day EOD summary       (send_eod_summary, called by the loop)
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import urllib.request
+
+logger = logging.getLogger(__name__)
+
+MAJOR_RULES = {"daily_loss", "exit_catastrophe", "exit_naked_stop"}
+BIG_TRADE_USD = 300.0
+_API = "https://api.telegram.org"
+
+
+def telegram_enabled() -> bool:
+    return bool(os.environ.get("TELEGRAM_BOT_TOKEN")
+                and os.environ.get("TELEGRAM_CHAT_ID"))
+
+
+def send_telegram(text: str, *, token: str | None = None,
+                  chat_id: str | None = None, timeout: int = 10) -> bool:
+    """Send one message. Gated + fire-and-forget: returns False (never raises)
+    if unconfigured or on any error — trading must never be impeded by Telegram."""
+    token = token or os.environ.get("TELEGRAM_BOT_TOKEN")
+    chat_id = chat_id or os.environ.get("TELEGRAM_CHAT_ID")
+    if not token or not chat_id:
+        return False
+    try:
+        req = urllib.request.Request(
+            f"{_API}/bot{token}/sendMessage",
+            data=json.dumps({"chat_id": chat_id, "text": text,
+                             "disable_web_page_preview": True}).encode(),
+            headers={"Content-Type": "application/json"},
+        )
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            return bool(json.loads(resp.read().decode()).get("ok"))
+    except Exception:  # noqa: BLE001 — must never raise into the loop
+        logger.debug("telegram send failed", exc_info=True)
+        return False
+
+
+def _fmt_rule(p: dict) -> str:
+    rt = p.get("rule_type")
+    icon = {"exit_catastrophe": "\U0001F6D1", "exit_naked_stop": "⚠️",
+            "daily_loss": "\U0001F53B"}.get(rt, "❗")
+    return f"{icon} {p.get('message') or rt}"
+
+
+def _fmt_close(p: dict) -> str:
+    pnl = p.get("realized_pnl") or 0.0
+    icon = "\U0001F7E2" if pnl > 0 else "\U0001F534"
+    return f"{icon} {p.get('symbol')} closed {pnl:+.0f} ({p.get('exit_reason') or ''})"
+
+
+def fmt_eod(for_date: str, pnl: dict) -> str:
+    total = pnl.get("total_pnl", 0.0)
+    icon = "\U0001F4C8" if total >= 0 else "\U0001F4C9"
+    wr = pnl.get("win_rate")
+    wr_s = f"{wr*100:.0f}%" if wr is not None else "n/a"
+    return (f"{icon} EOD {for_date}: {total:+.0f}\n"
+            f"closed {pnl.get('closed_trades', '?')}  "
+            f"W/L {pnl.get('wins', '?')}/{pnl.get('losses', '?')}  win% {wr_s}")
+
+
+class TelegramNotifier:
+    """Polls the event store and pushes one message per NEW major event since it
+    started. In-memory dedup; primes 'seen' at startup so history is never
+    replayed. All methods are no-ops + non-raising when Telegram is unconfigured."""
+
+    def __init__(self, store, *, big_trade_usd: float = BIG_TRADE_USD):
+        self.store = store
+        self.big_trade_usd = big_trade_usd
+        self._seen: set = set()
+        self._eod_sent_date: str | None = None
+        self.enabled = telegram_enabled()
+        if self.enabled:
+            try:
+                self._scan(notify=False)  # prime: mark existing events as seen
+            except Exception:  # noqa: BLE001
+                logger.debug("notifier prime failed", exc_info=True)
+
+    def poll(self) -> int:
+        """One poll cycle. Returns messages sent. Never raises."""
+        if not self.enabled:
+            return 0
+        try:
+            return self._scan(notify=True)
+        except Exception:  # noqa: BLE001
+            logger.debug("notifier poll failed", exc_info=True)
+            return 0
+
+    def _scan(self, *, notify: bool) -> int:
+        sent = 0
+        for e in self.store.query_events(event_type="risk_rule_triggered", limit=200):
+            p = json.loads(e["payload_json"])
+            if p.get("rule_type") not in MAJOR_RULES:
+                continue
+            key = ("rule", p.get("rule_type"), str(e["timestamp"]), p.get("message"))
+            if key in self._seen:
+                continue
+            self._seen.add(key)
+            if notify and send_telegram(_fmt_rule(p)):
+                sent += 1
+        for e in self.store.query_events(event_type="position_closed", limit=200):
+            p = json.loads(e["payload_json"])
+            if abs(p.get("realized_pnl") or 0.0) < self.big_trade_usd:
+                continue
+            key = ("close", p.get("symbol"), str(e["timestamp"]))
+            if key in self._seen:
+                continue
+            self._seen.add(key)
+            if notify and send_telegram(_fmt_close(p)):
+                sent += 1
+        return sent
+
+    def send_eod_summary(self, for_date: str, pnl: dict) -> bool:
+        """One-per-day EOD summary. ``pnl`` = query_session_pnl(...) dict."""
+        if not self.enabled or self._eod_sent_date == for_date:
+            return False
+        self._eod_sent_date = for_date
+        return send_telegram(fmt_eod(for_date, pnl))

telegram_bot.py

@@ -0,0 +1,181 @@
+#!/usr/bin/env python3
+"""Standalone Telegram Q&A bot — answers free-form questions about the trading
+data, READ-ONLY, locked to the authorized chat, Ollama-backed.
+
+Runs as its OWN process (`python telegram_bot.py`) so it can NEVER impede the
+trading loop. It long-polls getUpdates; for each message from TELEGRAM_CHAT_ID it
+gathers a data snapshot from Postgres and asks the local Ollama model to answer.
+If Ollama is down/disabled it falls back to the raw snapshot. It never accepts
+trade commands — questions only (mirrors the view-only dashboard rule).
+
+Env: TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID (required), DATABASE_URL,
+OLLAMA_HOST / OLLAMA_MODEL / OLLAMA_TIMEOUT_SECONDS / OLLAMA_ENABLED.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import time
+import urllib.request
+
+logger = logging.getLogger("telegram_bot")
+
+_API = "https://api.telegram.org"
+_ACTION_WORDS = ("buy", "sell", "flatten", "close all", "cancel", "short",
+                 "liquidate", "exit all", "place order", "submit")
+
+
+# ---------------------------------------------------------------- data context
+def gather_context(store) -> str:
+    """A compact, read-only snapshot of the current trading state for the LLM."""
+    from storage.projections import (query_account_positions_snapshot,
+                                      query_session_pnl, query_alltime_score)
+    lines: list[str] = []
+    try:
+        p = query_session_pnl(store)
+        lines.append(
+            f"TODAY P&L: total {p.get('total_pnl', 0):+.0f} "
+            f"(realized {p.get('realized_pnl', 0):+.0f}, "
+            f"unrealized {p.get('unrealized_pnl', 0):+.0f}); "
+            f"closed {p.get('closed_trades', 0)} "
+            f"W/L {p.get('wins', 0)}/{p.get('losses', 0)}.")
+        for t in (p.get("trades") or [])[:12]:
+            lines.append(f"  closed {t.get('symbol')}: {t.get('pnl', t.get('realized_pnl', 0)):+.0f} "
+                         f"({t.get('exit_reason', '')})")
+    except Exception as exc:  # noqa: BLE001
+        lines.append(f"(P&L unavailable: {exc})")
+    try:
+        snap = query_account_positions_snapshot(store)
+        positions = snap[0]["positions"] if snap else []
+        if positions:
+            lines.append(f"OPEN POSITIONS ({len(positions)}):")
+            for q in positions:
+                lines.append(
+                    f"  {q.get('symbol')}: qty {q.get('quantity')} @ "
+                    f"{q.get('avg_entry_price')} -> {q.get('current_price')} "
+                    f"(uPL {q.get('unrealized_pnl', 0):+.0f})")
+        else:
+            lines.append("OPEN POSITIONS: none (flat).")
+    except Exception as exc:  # noqa: BLE001
+        lines.append(f"(positions unavailable: {exc})")
+    try:
+        a = query_alltime_score(store)
+        lines.append(f"ALL-TIME: realized {a.get('total_realized', 0):+.0f} over "
+                     f"{a.get('trading_days', '?')} days, win% "
+                     f"{(a.get('win_rate') or 0)*100:.0f}, trades {a.get('trades', '?')}.")
+    except Exception:  # noqa: BLE001
+        pass
+    return "\n".join(lines)
+
+
+# ---------------------------------------------------------------------- ollama
+def ask_ollama(question: str, context: str, *, host: str, model: str,
+               timeout: int) -> str | None:
+    """Free-form answer from the local model. None on any failure."""
+    prompt = (
+        "You are a concise, READ-ONLY assistant for a momentum day-trading bot. "
+        "Answer the user's question using ONLY the data below. If the data does "
+        "not contain the answer, say so. Do not invent numbers. Keep it short.\n\n"
+        f"=== DATA ===\n{context}\n\n=== QUESTION ===\n{question}\n\n=== ANSWER ===\n")
+    try:
+        req = urllib.request.Request(
+            f"{host}/api/generate",
+            data=json.dumps({"model": model, "prompt": prompt, "stream": False,
+                             "options": {"temperature": 0.2, "num_predict": 400}}).encode(),
+            headers={"Content-Type": "application/json"},
+        )
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            body = json.loads(resp.read().decode())
+        text = (body.get("response") or "").strip()
+        return text or None
+    except Exception:  # noqa: BLE001
+        logger.debug("ollama unavailable", exc_info=True)
+        return None
+
+
+def answer(store, question: str, *, host: str, model: str, timeout: int,
+           ollama_enabled: bool = True) -> str:
+    """Build the read-only answer. Refuses action commands; falls back to the raw
+    snapshot when Ollama is down/disabled (never crashes)."""
+    if any(w in question.lower() for w in _ACTION_WORDS):
+        return ("I'm read-only — I can answer questions about the data but I "
+                "won't place or change trades. Ask me about P&L, positions, or "
+                "today's trades.")
+    context = gather_context(store)
+    if ollama_enabled:
+        out = ask_ollama(question, context, host=host, model=model, timeout=timeout)
+        if out:
+            return out
+    return f"(LLM unavailable — raw snapshot)\n{context}"
+
+
+# ------------------------------------------------------------------- transport
+def is_authorized(chat_id, allowed) -> bool:
+    return allowed is not None and str(chat_id) == str(allowed)
+
+
+def _send(token: str, chat_id, text: str) -> None:
+    try:
+        req = urllib.request.Request(
+            f"{_API}/bot{token}/sendMessage",
+            data=json.dumps({"chat_id": chat_id, "text": text[:4000]}).encode(),
+            headers={"Content-Type": "application/json"})
+        urllib.request.urlopen(req, timeout=10).read()
+    except Exception:  # noqa: BLE001
+        logger.warning("reply send failed", exc_info=True)
+
+
+def main() -> None:
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
+    from dotenv import load_dotenv
+    load_dotenv(os.path.join(os.path.dirname(__file__), ".env"))
+    token = os.environ.get("TELEGRAM_BOT_TOKEN")
+    allowed = os.environ.get("TELEGRAM_CHAT_ID")
+    if not token or not allowed:
+        logger.error("TELEGRAM_BOT_TOKEN / TELEGRAM_CHAT_ID not set — Q&A bot disabled")
+        return
+    host = os.environ.get("OLLAMA_HOST", "http://localhost:11434")
+    model = os.environ.get("OLLAMA_MODEL", "qwen2.5:7b-instruct")
+    timeout = int(os.environ.get("OLLAMA_TIMEOUT_SECONDS", "60"))
+    ollama_enabled = os.environ.get("OLLAMA_ENABLED", "1") not in ("0", "false", "False")
+
+    from storage.event_store import EventStore
+    store = EventStore("momentum")
+    logger.info("Q&A bot online (chat %s, ollama=%s %s) — ask me about the data",
+                allowed, ollama_enabled, model)
+    _send(token, allowed, "\U0001F916 Q&A bot online — ask me about P&L, positions, or today's trades.")
+
+    offset = None
+    while True:
+        try:
+            url = f"{_API}/bot{token}/getUpdates?timeout=50"
+            if offset is not None:
+                url += f"&offset={offset}"
+            with urllib.request.urlopen(url, timeout=60) as resp:
+                updates = json.loads(resp.read().decode()).get("result", [])
+        except Exception:  # noqa: BLE001
+            time.sleep(3)
+            continue
+        for u in updates:
+            offset = u["update_id"] + 1
+            msg = u.get("message") or u.get("edited_message") or {}
+            chat_id = (msg.get("chat") or {}).get("id")
+            text = (msg.get("text") or "").strip()
+            if not text:
+                continue
+            if not is_authorized(chat_id, allowed):
+                logger.warning("ignoring message from unauthorized chat %s", chat_id)
+                continue
+            try:
+                reply = answer(store, text, host=host, model=model, timeout=timeout,
+                               ollama_enabled=ollama_enabled)
+            except Exception as exc:  # noqa: BLE001
+                logger.warning("answer failed", exc_info=True)
+                reply = f"(error answering: {exc})"
+            _send(token, chat_id, reply)
+
+
+if __name__ == "__main__":
+    main()

tests/unit/test_telegram.py

@@ -0,0 +1,102 @@
+"""Telegram notifier (push) + Q&A bot (interactive) — gating, dedup, auth, fallback."""
+
+import pytest
+
+import runtime.notifier as nf
+import telegram_bot as qa
+from storage.event_schema import EventMode, PositionClosedEvent, RiskRuleTriggeredEvent
+from storage.event_store import EventStore
+
+
+@pytest.fixture
+def store():
+    s = EventStore(":memory:")
+    yield s
+    s.close()
+
+
+def _rule(store, rule_type, message="m"):
+    store.emit(RiskRuleTriggeredEvent(
+        timestamp=__import__("datetime").datetime(2026, 6, 23, 10, 0),
+        mode=EventMode.PAPER, correlation_id="t", message=message,
+        rule_type=rule_type, rule_value=0.0, current_state={}, action_taken="x"))
+
+
+def _close(store, symbol, pnl):
+    store.emit(PositionClosedEvent(
+        timestamp=__import__("datetime").datetime(2026, 6, 23, 10, 1),
+        mode=EventMode.PAPER, correlation_id="t", message="c",
+        position_id=f"{symbol}-1", symbol=symbol, exit_price=10.0,
+        realized_pnl=pnl, exit_reason="stop_loss"))
+
+
+# ----------------------------------------------------------------- C1 notifier
+def test_send_telegram_noop_when_unconfigured(monkeypatch):
+    monkeypatch.delenv("TELEGRAM_BOT_TOKEN", raising=False)
+    monkeypatch.delenv("TELEGRAM_CHAT_ID", raising=False)
+    assert nf.send_telegram("hi") is False          # silent no-op, no network
+    assert nf.telegram_enabled() is False
+
+
+def test_notifier_disabled_is_inert(store, monkeypatch):
+    monkeypatch.delenv("TELEGRAM_BOT_TOKEN", raising=False)
+    monkeypatch.delenv("TELEGRAM_CHAT_ID", raising=False)
+    n = nf.TelegramNotifier(store)
+    _rule(store, "daily_loss")
+    assert n.poll() == 0                              # never sends, never raises
+
+
+def test_notifier_pushes_major_events_once(store, monkeypatch):
+    monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "x")
+    monkeypatch.setenv("TELEGRAM_CHAT_ID", "1")
+    sent = []
+    monkeypatch.setattr(nf, "send_telegram", lambda text, **kw: sent.append(text) or True)
+    n = nf.TelegramNotifier(store)                    # primes on empty store
+    _rule(store, "exit_catastrophe", "CAT")           # major
+    _rule(store, "entry_backout", "noise")            # routine -> ignored
+    _close(store, "BIG", 553)                          # > $300 -> major
+    _close(store, "SMALL", 40)                         # < $300 -> ignored
+    assert n.poll() == 2
+    assert n.poll() == 0                              # dedup: no re-send
+    joined = " ".join(sent)
+    assert "CAT" in joined and "BIG" in joined
+    assert "noise" not in joined and "SMALL" not in joined
+
+
+def test_eod_summary_once_per_day(store, monkeypatch):
+    monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "x")
+    monkeypatch.setenv("TELEGRAM_CHAT_ID", "1")
+    sent = []
+    monkeypatch.setattr(nf, "send_telegram", lambda text, **kw: sent.append(text) or True)
+    n = nf.TelegramNotifier(store)
+    pnl = {"total_pnl": -2464, "closed_trades": 12, "wins": 3, "losses": 9, "win_rate": 0.25}
+    assert n.send_eod_summary("2026-06-23", pnl) is True
+    assert n.send_eod_summary("2026-06-23", pnl) is False   # not twice
+    assert "2026-06-23" in sent[0] and "-2464" in sent[0]
+
+
+# ------------------------------------------------------------------- C2 Q&A bot
+def test_auth_gate():
+    assert qa.is_authorized(2082795871, "2082795871") is True
+    assert qa.is_authorized(999, "2082795871") is False
+    assert qa.is_authorized(2082795871, None) is False
+
+
+def test_answer_refuses_action_commands(store):
+    for cmd in ("buy 100 AAPL", "flatten everything", "sell PLUG now"):
+        out = qa.answer(store, cmd, host="h", model="m", timeout=1, ollama_enabled=True)
+        assert "read-only" in out.lower()
+
+
+def test_answer_falls_back_when_ollama_disabled(store):
+    out = qa.answer(store, "what's my P&L?", host="h", model="m", timeout=1,
+                    ollama_enabled=False)
+    assert "raw snapshot" in out.lower()
+    assert "TODAY P&L" in out                          # gather_context ran
+
+
+def test_answer_falls_back_when_ollama_down(store, monkeypatch):
+    monkeypatch.setattr(qa, "ask_ollama", lambda *a, **k: None)   # simulate down
+    out = qa.answer(store, "how are positions?", host="h", model="m", timeout=1,
+                    ollama_enabled=True)
+    assert "raw snapshot" in out.lower()