From 59e296a73637487417ecbed206d2defa674ece75 Mon Sep 17 00:00:00 2001 From: momentum dev Date: Tue, 23 Jun 2026 16:29:15 -0400 Subject: [PATCH] Telegram bot: gated push alerts + standalone Ollama Q&A MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- run_live_paper.py | 24 +++++ runtime/notifier.py | 138 +++++++++++++++++++++++++++ telegram_bot.py | 181 ++++++++++++++++++++++++++++++++++++ tests/unit/test_telegram.py | 102 ++++++++++++++++++++ 4 files changed, 445 insertions(+) create mode 100644 runtime/notifier.py create mode 100644 telegram_bot.py create mode 100644 tests/unit/test_telegram.py diff --git a/run_live_paper.py b/run_live_paper.py index a0b226e..c4ecc8f 100644 --- a/run_live_paper.py +++ b/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")) diff --git a/runtime/notifier.py b/runtime/notifier.py new file mode 100644 index 0000000..3f2c697 --- /dev/null +++ b/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)) diff --git a/telegram_bot.py b/telegram_bot.py new file mode 100644 index 0000000..d6545b4 --- /dev/null +++ b/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() diff --git a/tests/unit/test_telegram.py b/tests/unit/test_telegram.py new file mode 100644 index 0000000..942ddf6 --- /dev/null +++ b/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()