momentum devandClaude Fable 5
e7c34f298f
Adds OperatorCommandEvent + a closed command vocabulary (flatten_all, flatten_symbol, pause_entries, resume_entries). POST /api/command emits it (token-required: 503 when DASHBOARD_TOKEN unset — commands never run unauthenticated); the live loop's new 5s "command" step consumes them via OperatorCommandConsumer with three replay-safety layers (in-session nonce set, persisted operator_command_done reload, 10-min staleness window). flatten_all routes through close_session DIRECTLY (never step_eod_flatten) so it cannot trip plan 003's daily-restart exit. Pause is checked at the top of BOTH request_approvals_for_ready_signals AND submit_breakout_now (missing either leaks pause on that path). qabot gains an authorized-chat-only /flatten slash command (prose "flatten" still refuses via _ACTION_WORDS). Snapshot exposes broker truth + command_bus + a defensive experiment ledger. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
422 lines
19 KiB
Python
422 lines
19 KiB
Python
#!/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
|
||
RETRIEVES the relevant data from Postgres (a desk snapshot, plus a focused
|
||
per-symbol dive when the question names a ticker) and asks the local Ollama model
|
||
to answer. Falls back to the raw snapshot if Ollama is down. Never accepts trade
|
||
commands. Sends a help menu so the user knows what it can answer (direction).
|
||
|
||
Also the DEAD-MAN'S SWITCH: the loop's own notifier lives inside the loop
|
||
process and cannot report the loop's death — this always-up bot watches the
|
||
event heartbeat (max(events.timestamp)) from outside and alerts when it goes
|
||
stale during market hours.
|
||
|
||
Env: TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID (required), DATABASE_URL,
|
||
OLLAMA_HOST / OLLAMA_MODEL / OLLAMA_TIMEOUT_SECONDS / OLLAMA_ENABLED,
|
||
DEADMAN_STALE_MIN / DEADMAN_DAILY_PING.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
import time
|
||
import urllib.request
|
||
from collections import Counter
|
||
from datetime import datetime, timedelta
|
||
from zoneinfo import ZoneInfo
|
||
|
||
from runtime.market_calendar import is_regular_hours
|
||
|
||
logger = logging.getLogger("telegram_bot")
|
||
|
||
_ET = ZoneInfo("America/New_York")
|
||
|
||
_API = "https://api.telegram.org"
|
||
_ACTION_WORDS = ("buy ", "sell ", "flatten", "close all", "cancel ", "short ",
|
||
"liquidate", "exit all", "place order", "submit order", "go long")
|
||
_HELP_WORDS = ("help", "/help", "/start", "what can you", "what do you", "commands", "menu")
|
||
_STOPWORDS = {"P&L", "PNL", "EOD", "VWAP", "ORB", "USD", "AI", "OK", "ET", "AM", "PM",
|
||
"WHY", "HOW", "THE", "AND", "ARE", "WAS", "DID", "ANY", "ALL", "NOT",
|
||
"YOU", "OUR", "WHAT", "WHEN", "BEST", "WORST", "TODAY", "NOW", "R"}
|
||
|
||
HELP_TEXT = (
|
||
"\U0001F916 I'm your momentum desk assistant (read-only). Ask me in plain English about:\n"
|
||
"• today's P&L / trades — \"how did we do?\", \"best and worst trade?\"\n"
|
||
"• the book — \"what's open?\", \"how's PLUG doing?\"\n"
|
||
"• the watchlist — \"what's on the watchlist?\", \"anything ready?\"\n"
|
||
"• why we are / aren't trading — \"why no entries?\", \"what got blocked?\"\n"
|
||
"• a specific ticker — \"what happened with VTAK?\", \"why did we skip AMC?\"\n"
|
||
"• all-time — \"how are we doing overall?\"\n"
|
||
"I can't place or change trades — questions only."
|
||
)
|
||
|
||
|
||
def is_help(q: str) -> bool:
|
||
ql = q.strip().lower()
|
||
return ql in ("help", "/help", "/start", "menu") or any(w in ql for w in _HELP_WORDS)
|
||
|
||
|
||
# --------------------------------------------------------------- data retrieval
|
||
def _recent_closed(store, limit=12):
|
||
rows = []
|
||
for e in store.query_events(event_type="position_closed", limit=None):
|
||
p = json.loads(e["payload_json"])
|
||
rows.append((str(e["timestamp"]), p))
|
||
rows.sort(reverse=True)
|
||
return [p for _, p in rows[:limit]]
|
||
|
||
|
||
def known_symbols(store) -> set:
|
||
syms = set()
|
||
try:
|
||
for w in store.query_events(event_type="symbol_state_changed", limit=None):
|
||
s = json.loads(w["payload_json"]).get("symbol")
|
||
if s:
|
||
syms.add(str(s).upper())
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
return syms
|
||
|
||
|
||
def gather_context(store) -> str:
|
||
from storage.projections import (query_account_positions_snapshot,
|
||
query_session_pnl, query_alltime_score,
|
||
query_watch_states_snapshot,
|
||
query_ready_signals_snapshot)
|
||
L: list[str] = []
|
||
try:
|
||
p = query_session_pnl(store)
|
||
L.append(f"TODAY (live session) P&L: total {p.get('total_pnl', 0):+.0f} "
|
||
f"(realized {p.get('realized_pnl', 0):+.0f}, unrealized {p.get('unrealized_pnl', 0):+.0f}); "
|
||
f"closed {p.get('closed_trades', 0)} W/L {p.get('wins', 0)}/{p.get('losses', 0)}.")
|
||
except Exception as exc: # noqa: BLE001
|
||
L.append(f"(today P&L unavailable: {exc})")
|
||
try:
|
||
rc = _recent_closed(store, 12)
|
||
if rc:
|
||
L.append("RECENT CLOSED TRADES (newest first):")
|
||
for t in rc:
|
||
L.append(f" {t.get('symbol')}: {t.get('realized_pnl', 0):+.0f} "
|
||
f"@entry {t.get('entry_price')} ({t.get('exit_reason', '')})")
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
try:
|
||
snap = query_account_positions_snapshot(store)
|
||
positions = snap[0]["positions"] if snap else []
|
||
if positions:
|
||
L.append(f"OPEN POSITIONS ({len(positions)}):")
|
||
for q in positions:
|
||
L.append(f" {q.get('symbol')}: qty {q.get('quantity')} @ {q.get('avg_entry_price')} "
|
||
f"-> {q.get('current_price')} (uPL {q.get('unrealized_pnl', 0):+.0f})")
|
||
else:
|
||
L.append("OPEN POSITIONS: none (book is flat).")
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
try:
|
||
ws = query_watch_states_snapshot(store)
|
||
if ws:
|
||
by_state = Counter(str(w.get("state") or "unknown") for w in ws)
|
||
L.append(f"WATCHLIST: {len(ws)} symbols — " +
|
||
", ".join(f"{n} {st}" for st, n in by_state.most_common()))
|
||
hot = [w for w in ws if str(w.get("state")) in ("ready", "armed")]
|
||
for w in sorted(hot, key=lambda x: -(x.get("last_score") or 0))[:8]:
|
||
L.append(f" {w.get('symbol')}: {w.get('state')} (score {w.get('last_score')})")
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
try:
|
||
rs = query_ready_signals_snapshot(store)
|
||
if rs:
|
||
L.append(f"READY SIGNALS ({len(rs)}):")
|
||
for s in rs[:8]:
|
||
av = s.get("above_vwap")
|
||
L.append(f" {s.get('symbol')}: entry {s.get('entry_price')} stop {s.get('stop_loss_price')} "
|
||
f"vwap {s.get('vwap')} {'(above VWAP)' if av else '(BELOW VWAP)' if av is False else ''}")
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
try: # why we are/aren't entering — the recent block tally
|
||
blocks = Counter()
|
||
for e in store.query_events(event_type="risk_rule_triggered", limit=300):
|
||
blocks[json.loads(e["payload_json"]).get("rule_type")] += 1
|
||
if blocks:
|
||
L.append("RECENT ENTRY BLOCKS (why entries were skipped): " +
|
||
", ".join(f"{rt}×{n}" for rt, n in blocks.most_common(8)))
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
try:
|
||
a = query_alltime_score(store)
|
||
L.append(f"ALL-TIME: realized {a.get('total_realized', 0):+.0f} over "
|
||
f"{a.get('trading_days', '?')} days, win% {(a.get('win_rate') or 0)*100:.0f}, "
|
||
f"trades {a.get('trades', '?')}.")
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
return "\n".join(L)
|
||
|
||
|
||
def symbol_context(store, sym: str) -> str:
|
||
from storage.projections import query_symbol_criteria
|
||
L = [f"\n=== DETAIL: {sym} ==="]
|
||
try:
|
||
for t in _recent_closed(store, 50):
|
||
if str(t.get("symbol")).upper() == sym:
|
||
L.append(f"last trade: {t.get('realized_pnl', 0):+.0f} @entry {t.get('entry_price')} "
|
||
f"stop {t.get('stop_loss_price')} ({t.get('exit_reason', '')})")
|
||
break
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
try:
|
||
c = query_symbol_criteria(store, sym)
|
||
crit = c.get("criteria") or []
|
||
if crit:
|
||
failed = [x.get("key") or x.get("label") for x in crit if not x.get("passed")]
|
||
passed = [x.get("key") or x.get("label") for x in crit if x.get("passed")]
|
||
L.append(f"setup score {c.get('score')} ({c.get('passed_count')} criteria passed); "
|
||
f"passed: {', '.join(passed) or 'none'}; failed: {', '.join(failed) or 'none'}")
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
return "\n".join(L) if len(L) > 1 else ""
|
||
|
||
|
||
# ---------------------------------------------------------------------- ollama
|
||
def ask_ollama(question: str, context: str, *, host: str, model: str,
|
||
timeout: int) -> str | None:
|
||
prompt = (
|
||
"You are the assistant for a small-cap momentum DAY-TRADING bot (opening-range "
|
||
"breakouts on $1-20 gappers). Answer the user's question using ONLY the DATA below, "
|
||
"which is the bot's live state. Be concrete and brief (a few sentences). Quote the "
|
||
"real numbers. If the data doesn't contain the answer, say so and tell the user what "
|
||
"you DO have (P&L, trades, the watchlist, ready signals, entry blocks, per-symbol "
|
||
"setup detail). Never invent tickers or numbers.\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": 500}}).encode(),
|
||
headers={"Content-Type": "application/json"})
|
||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||
text = (json.loads(resp.read().decode()).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:
|
||
if is_help(question):
|
||
return HELP_TEXT
|
||
if any(w in (" " + question.lower() + " ") for w in _ACTION_WORDS):
|
||
return ("I'm read-only — I answer questions about the data but won't place or "
|
||
"change trades. Try \"how did we do today?\" or send \"help\".")
|
||
context = gather_context(store)
|
||
# focused per-symbol retrieval: tickers in the question that the bot actually knows
|
||
tokens = set(re.findall(r"\b[A-Z]{1,5}\b", question.upper())) - _STOPWORDS
|
||
known = known_symbols(store)
|
||
for sym in [t for t in tokens if t in known][:3]:
|
||
context += symbol_context(store, sym)
|
||
if ollama_enabled:
|
||
out = ask_ollama(question, context, host=host, model=model, timeout=timeout)
|
||
if out:
|
||
return out
|
||
return f"(LLM unavailable — here's the raw snapshot)\n{context}"
|
||
|
||
|
||
# --------------------------------------------------------- operator kill switch
|
||
def parse_flatten_command(text: str):
|
||
"""Parse the ONE action the qabot performs: the explicit ``/flatten`` slash
|
||
command. Returns ``(command, symbol)`` or None. Prose like "sell everything"
|
||
is NOT matched here — it still hits the _ACTION_WORDS refusal in answer(). This
|
||
is the deliberate carve-out: an explicit slash command from the authorized chat
|
||
is allowed; free-text trade requests are not."""
|
||
parts = text.strip().split()
|
||
if not parts or parts[0].lower() != "/flatten":
|
||
return None
|
||
if len(parts) >= 2 and parts[1].strip():
|
||
return ("flatten_symbol", parts[1].strip().upper())
|
||
return ("flatten_all", None)
|
||
|
||
|
||
def emit_operator_command(store, command: str, symbol, requested_by: str) -> None:
|
||
"""Emit the same OperatorCommandEvent the dashboard does — the live loop's 5s
|
||
consumer executes it (single execution authority)."""
|
||
import uuid
|
||
from storage.event_schema import EventMode, OperatorCommandEvent
|
||
store.emit(OperatorCommandEvent(
|
||
timestamp=datetime.now(), mode=EventMode.PAPER,
|
||
message=f"telegram operator command {command}" + (f" {symbol}" if symbol else ""),
|
||
command=command, symbol=symbol, requested_by=requested_by,
|
||
nonce=uuid.uuid4().hex))
|
||
|
||
|
||
# ------------------------------------------------------------------- dead-man
|
||
def _now_et() -> datetime:
|
||
"""ET wall clock, tz-NAIVE — events.timestamp is ET-naive (RUNBOOK), so the
|
||
staleness math must be naive-vs-naive or it reads 4-5h stale from UTC."""
|
||
return datetime.now(_ET).replace(tzinfo=None)
|
||
|
||
|
||
def _max_event_ts(store) -> datetime | None:
|
||
"""Newest events.timestamp as a naive datetime (None on an empty store).
|
||
Raw query on purpose: query_events() orders ASC and stringifies timestamps."""
|
||
cur = store.con.cursor()
|
||
cur.execute("SELECT max(timestamp) FROM events")
|
||
row = cur.fetchone()
|
||
return row[0] if row else None
|
||
|
||
|
||
class DeadmanCheck:
|
||
"""Loop-death watchdog. During market hours (holiday-aware, from
|
||
runtime.market_calendar) a heartbeat older than DEADMAN_STALE_MIN minutes
|
||
raises ONE alert, re-raised at most every REALERT_MIN. After 20:30 ET a
|
||
once-daily ping (DEADMAN_DAILY_PING=1, default on) proves the watchdog
|
||
itself is alive. Clock/read/send are injectable for tests."""
|
||
|
||
CHECK_EVERY_S = 120
|
||
REALERT_MIN = 30
|
||
|
||
def __init__(self, read_max_ts, send, *, stale_min=None, daily_ping=None,
|
||
now_et=_now_et, is_market_hours=is_regular_hours):
|
||
self._read_max_ts = read_max_ts
|
||
self._send = send
|
||
self._stale_min = float(stale_min if stale_min is not None
|
||
else os.environ.get("DEADMAN_STALE_MIN", "5"))
|
||
dp = daily_ping if daily_ping is not None \
|
||
else os.environ.get("DEADMAN_DAILY_PING", "1")
|
||
self._daily_ping = str(dp) not in ("0", "false", "False")
|
||
self._now_et = now_et
|
||
self._is_market_hours = is_market_hours
|
||
self._next_check = 0.0
|
||
self._last_alerted: datetime | None = None
|
||
self._last_ping_date = None
|
||
|
||
def maybe_check(self) -> str | None:
|
||
"""Rate-limited entry point for the poll loop (wakes at least ~every 60s)."""
|
||
if time.monotonic() < self._next_check:
|
||
return None
|
||
self._next_check = time.monotonic() + self.CHECK_EVERY_S
|
||
return self.check()
|
||
|
||
def check(self) -> str | None:
|
||
"""Run one check now; returns the message sent, if any."""
|
||
now = self._now_et()
|
||
try:
|
||
max_ts = self._read_max_ts()
|
||
except Exception: # noqa: BLE001 — a flaky DB read must not kill the bot
|
||
logger.warning("deadman: heartbeat read failed", exc_info=True)
|
||
return None
|
||
stale_min = None if max_ts is None \
|
||
else (now - max_ts).total_seconds() / 60.0
|
||
msg = None
|
||
if self._is_market_hours(now) and (stale_min is None or stale_min > self._stale_min):
|
||
if (self._last_alerted is None
|
||
or now - self._last_alerted >= timedelta(minutes=self.REALERT_MIN)):
|
||
desc = "no events ever" if stale_min is None \
|
||
else f"no events for {stale_min:.0f} min"
|
||
msg = f"⚠️ DEAD-MAN: {desc} during market hours — loop may be down."
|
||
self._last_alerted = now
|
||
elif (self._daily_ping and (now.hour, now.minute) >= (20, 30)
|
||
and self._last_ping_date != now.date()):
|
||
self._last_ping_date = now.date()
|
||
if max_ts is not None and max_ts.date() == now.date():
|
||
msg = f"✅ heartbeat: loop wrote events today; last event {max_ts}."
|
||
else:
|
||
msg = f"⚠️ heartbeat: NO events today; last event {max_ts}."
|
||
if msg:
|
||
self._send(msg)
|
||
return msg
|
||
|
||
|
||
# ------------------------------------------------------------------- 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"))
|
||
from runtime.env_overrides import load as _load_overrides
|
||
_load_overrides()
|
||
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)", allowed, ollama_enabled, model)
|
||
_send(token, allowed, HELP_TEXT)
|
||
deadman = DeadmanCheck(lambda: _max_event_ts(store),
|
||
lambda text: _send(token, allowed, text))
|
||
|
||
offset = None
|
||
while True:
|
||
try:
|
||
deadman.maybe_check()
|
||
except Exception: # noqa: BLE001 — the watchdog must never kill the bot
|
||
logger.warning("deadman check failed", exc_info=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
|
||
# explicit /flatten slash command (authorized chat only) — the qabot's
|
||
# one action. Emits the guarded event; the loop executes it in ~5s.
|
||
cmd = parse_flatten_command(text)
|
||
if cmd is not None:
|
||
try:
|
||
emit_operator_command(store, cmd[0], cmd[1],
|
||
requested_by=f"telegram:{chat_id}")
|
||
_send(token, chat_id, "Command sent — loop executes within ~5s.")
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.warning("flatten command emit failed", exc_info=True)
|
||
_send(token, chat_id, f"(command failed to send: {exc})")
|
||
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()
|