diff --git a/api/main.py b/api/main.py index 3fc8ada..b271efd 100644 --- a/api/main.py +++ b/api/main.py @@ -12,6 +12,7 @@ POST /api/trading/approvals/approve {"order_id": ..., "approved_by"?: ...} POST /api/trading/approvals/reject {"order_id": ..., "reason"?: ...} POST /api/trading/exit-order {"symbol": ...} POST /api/watch/add {"symbol": ...} (manual watchlist inject) +POST /api/command {"command": ..., "symbol"?: ...} (operator kill switch; requires DASHBOARD_TOKEN) The server holds an EventStore for reads; trade actions route through a TradingExecutionService when one is attached (the orchestrator does this). @@ -26,11 +27,13 @@ import json import os import threading import time +import uuid from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, urlparse +from storage.event_schema import EventMode, OperatorCommandEvent, OPERATOR_COMMANDS from storage.event_store import EventStore from storage.projections import ( query_account_orders_snapshot, @@ -56,6 +59,27 @@ from storage.projections import ( STATIC_DIR = Path(__file__).parent / "static" +def _command_bus_enabled() -> bool: + """The operator command bus is live only when a token guards it AND it is + explicitly enabled — so a stray COMMAND_BUS_ENABLED never opens an + unauthenticated kill switch.""" + return bool(os.environ.get("DASHBOARD_TOKEN")) and ( + os.environ.get("COMMAND_BUS_ENABLED", "").strip().lower() + in {"1", "true", "yes", "on"} + ) + + +def _broker_block() -> dict: + """Broker truth for the LIVE badge, derived from the base URL the services + already have (not the execution_mode literal, which lied). Alpaca's paper host + is ``paper-api.alpaca.markets``; anything else is live money.""" + from urllib.parse import urlparse as _urlparse + + url = os.environ.get("ALPACA_BASE_URL") or "https://paper-api.alpaca.markets" + host = _urlparse(url).hostname or url + return {"base_url_host": host, "is_paper": host.startswith("paper-api")} + + def _eastern_session_date(): """Today's date in US/Eastern — matches the ingestion pipeline's flags.""" from zoneinfo import ZoneInfo @@ -150,6 +174,44 @@ class _ChangeSignal: return self._stamp +def _maybe_json(v): + """JSONB columns come back from psycopg2 as str OR already-parsed dict; make it + a dict either way so the ledger never double-encodes.""" + if isinstance(v, str): + try: + return json.loads(v) + except (ValueError, TypeError): + return {} + return v or {} + + +def _experiment_ledger(store) -> list[dict]: + """Latest verdict per experiment for the ledger panel (plan 006 tables). + + Returns [] on ANY error — the experiments/experiment_verdicts tables live in a + separate schema file applied by the registry, so on a bare/test database they + simply don't exist and the panel shows its empty state.""" + try: + rows = store.con.execute( + "SELECT v.experiment_id, v.as_of, v.n, v.verdict, v.metrics, " + " e.hypothesis, e.bars, e.status " + "FROM (SELECT DISTINCT ON (experiment_id) experiment_id, as_of, n, verdict, metrics " + " FROM experiment_verdicts ORDER BY experiment_id, as_of DESC) v " + "LEFT JOIN experiments e ON e.id = v.experiment_id " + "ORDER BY v.as_of DESC" + ).fetchall() + except Exception: # noqa: BLE001 — tables absent on a bare DB + return [] + out = [] + for r in rows: + out.append({ + "experiment": r[0], "as_of": str(r[1]), "n": r[2], "verdict": r[3], + "metrics": _maybe_json(r[4]), "hypothesis": r[5], + "bars": _maybe_json(r[6]), "status": r[7], + }) + return out + + class DashboardState: """Shared state: event store path + optional live execution service.""" @@ -201,6 +263,11 @@ class DashboardState: "generated_at": datetime.now(timezone.utc).isoformat(), "execution_mode": self.execution_mode, "has_execution": self.execution_service is not None, + # broker truth (LIVE badge) + operator command-bus availability + "broker": _broker_block(), + "command_bus": _command_bus_enabled(), + # experiment ledger (plan 006 tables) — [] until they exist + "experiments": _experiment_ledger(store) if is_live else [], "view_date": None if is_live else for_date, "available_dates": _available_dates(store), "pnl": query_session_pnl(store, for_date=for_date), @@ -316,6 +383,28 @@ class DashboardState: def exit_position(self, symbol: str) -> dict | None: return self._with_service(lambda svc: svc.submit_exit_order(symbol)) + def emit_command(self, command: str, symbol: str | None = None, + requested_by: str = "dashboard") -> dict: + """Emit a guarded operator command event (the loop consumes + executes it). + + Works even when execution_service is None — the standalone dashboard is a + thin emitter; the loop process is the single execution authority. Callers + gate on DASHBOARD_TOKEN before reaching here (commands never run + unauthenticated).""" + nonce = uuid.uuid4().hex + with self._lock: + store = self.open_store() + try: + store.emit(OperatorCommandEvent( + timestamp=datetime.now(), mode=EventMode.PAPER, + message=f"operator command {command}" + + (f" {symbol}" if symbol else ""), + command=command, symbol=(symbol or None), + requested_by=requested_by, nonce=nonce)) + finally: + store.close() + return {"ok": True, "command": command, "symbol": symbol, "nonce": nonce} + def add_watch(self, symbol: str) -> dict | None: """Manually inject a symbol onto the watchlist (if supported).""" if not self.watch_inject or not symbol: @@ -490,6 +579,29 @@ class DashboardHandler(BaseHTTPRequestHandler): result = self.state.exit_position(str(body.get("symbol") or "")) elif route == "/api/watch/add": result = self.state.add_watch(str(body.get("symbol") or "")) + elif route == "/api/command": + # The command bus is token-gated by construction: without + # DASHBOARD_TOKEN a command could run unauthenticated, so refuse + # outright (the top-of-do_POST gate only bites WHEN a token is set). + if not os.environ.get("DASHBOARD_TOKEN"): + self._send_json( + {"ok": False, "error": "command bus disabled: DASHBOARD_TOKEN not set"}, + 503) + return + command = str(body.get("command") or "") + if command not in OPERATOR_COMMANDS: + self._send_json( + {"ok": False, "error": f"unknown command {command!r}"}, 400) + return + symbol = body.get("symbol") + symbol = str(symbol).upper() if symbol else None + if command == "flatten_symbol" and not symbol: + self._send_json( + {"ok": False, "error": "flatten_symbol requires a symbol"}, 400) + return + self._send_json( + self.state.emit_command(command, symbol=symbol), 200) + return else: self._send_json({"error": "not found"}, 404) return diff --git a/run_live_paper.py b/run_live_paper.py index ea4b6fc..a9aed66 100644 --- a/run_live_paper.py +++ b/run_live_paper.py @@ -778,6 +778,22 @@ def main(argv: list[str] | None = None) -> int: f"{b['symbol']} ({b['reason']})" for b in backed ) + # Operator command bus (plan 007): consume kill-switch commands emitted by the + # dashboard / qabot on the same 5s cadence as the guard. flatten_all routes + # through close_session DIRECTLY (never step_eod_flatten) so it can't trip the + # plan-003 daily-restart exit. None in dry mode (no execution service). + from runtime.operator_commands import OperatorCommandConsumer + command_consumer = ( + OperatorCommandConsumer(rt["store"], rt["execution"], session_id=rt["session_id"]) + if rt["execution"] is not None else None + ) + + def step_commands(): + if command_consumer is None: + return None + done = command_consumer.poll() + return f"operator commands executed: {len(done)}" if done else None + def step_discover(): if not client: return "skipped (no keys)" @@ -1272,6 +1288,9 @@ def main(argv: list[str] | None = None) -> int: enabled=_flag("TRADING_EXECUTION_ENABLED", "1")) scheduler.add("guard", step_guard, float(os.environ.get("TRADING_ENTRY_GUARD_INTERVAL_SECONDS", "5")), enabled=_flag("TRADING_EXECUTION_ENABLED", "1")) + scheduler.add("command", step_commands, + float(os.environ.get("OPERATOR_COMMAND_INTERVAL_SECONDS", "5")), + enabled=(rt["execution"] is not None)) scheduler.add("exits", step_manage_exits, float(os.environ.get("TRADING_EXIT_MANAGE_INTERVAL_SECONDS", "12")), enabled=_flag("TRADING_EXECUTION_ENABLED", "1")) scheduler.add("eod", step_eod_flatten, float(os.environ.get("TRADING_EOD_INTERVAL_SECONDS", "30")), diff --git a/runtime/operator_commands.py b/runtime/operator_commands.py new file mode 100644 index 0000000..49d435e --- /dev/null +++ b/runtime/operator_commands.py @@ -0,0 +1,130 @@ +"""Operator command bus consumer (plan 007). + +The dashboard / qabot emit ``operator_command`` events (a guarded kill switch); +the live loop polls this consumer on its 5s guard cadence and executes each new +command through the execution service's existing protective-close helpers, then +echoes an ``operator_command_done`` event so a restart never re-runs it. + +Replay safety (three layers): + 1. an in-session consumed-nonce set (deduped within the process), + 2. persisted ``operator_command_done`` events reloaded on boot (deduped across + restarts), + 3. a staleness window — commands older than ``stale_seconds`` are ignored, so a + command issued while the loop was down does not fire on the next boot. + +Deliberately minimal: a closed command vocabulary (OPERATOR_COMMANDS), no +parameters beyond an optional symbol. flatten_all routes through +``close_session`` DIRECTLY (never ``step_eod_flatten``) so it can never trip +plan 003's daily-restart exit. +""" + +from __future__ import annotations + +import json +import logging +from datetime import datetime, timedelta + +from storage.event_schema import EventMode, OperatorCommandEvent, EventType + +logger = logging.getLogger(__name__) + +STALE_SECONDS = 600 # ignore commands older than 10 minutes (boot-replay guard) + + +class OperatorCommandConsumer: + def __init__(self, store, execution, *, session_id=None, now_fn=None, + stale_seconds: float = STALE_SECONDS, mode: EventMode = EventMode.PAPER): + self.store = store + self.execution = execution + self.session_id = session_id + self._now = now_fn or datetime.now + self._stale = float(stale_seconds) + self._mode = mode + # nonces already handled — seeded from persisted _done events so a restart + # doesn't re-execute a command the previous process already ran. + self._consumed: set[str] = self._load_done_nonces() + + def _cutoff_iso(self) -> str: + return (self._now() - timedelta(seconds=self._stale)).isoformat() + + def _load_done_nonces(self) -> set[str]: + done: set[str] = set() + try: + for e in self.store.query_events( + event_type=EventType.OPERATOR_COMMAND_DONE.value, + since=self._cutoff_iso(), limit=None, + ): + nonce = json.loads(e.get("payload_json") or "{}").get("nonce") + if nonce: + done.add(str(nonce)) + except Exception: # noqa: BLE001 — a rebuild failure must not block boot + logger.warning("operator-command done-nonce reload skipped", exc_info=True) + return done + + def poll(self) -> list[str]: + """Execute any new, in-window operator commands. Returns nonces executed.""" + try: + events = self.store.query_events( + event_type=EventType.OPERATOR_COMMAND.value, + since=self._cutoff_iso(), limit=None, + ) + except Exception: # noqa: BLE001 + logger.warning("operator-command poll failed", exc_info=True) + return [] + executed: list[str] = [] + for e in events: + p = json.loads(e.get("payload_json") or "{}") + nonce = str(p.get("nonce") or "") + if not nonce or nonce in self._consumed: + continue + self._consumed.add(nonce) # mark BEFORE executing so a raising command can't loop + result = self._execute(p) + self._emit_done(p, result) + executed.append(nonce) + logger.info("operator command %s (%s) -> %s", + p.get("command"), p.get("symbol") or "-", result) + return executed + + def _execute(self, p: dict) -> str: + cmd = p.get("command") + sym = p.get("symbol") + ex = self.execution + if ex is None: + return "error: no execution service" + try: + if cmd == "flatten_all": + # DIRECT close_session — never step_eod_flatten, so this cannot + # request plan 003's daily-restart exit. close_session also blocks + # re-entry for the session (the kill-switch semantic). + res = ex.close_session("operator_flatten") + errs = res.get("errors") if isinstance(res, dict) else None + closed = ", ".join(res.get("closed_positions") or []) if isinstance(res, dict) else "" + return f"error: {errs}" if errs else f"ok: flattened [{closed or 'none'}]" + if cmd == "flatten_symbol": + if not sym: + return "error: flatten_symbol requires a symbol" + res = ex.flatten_symbol(sym) + return "ok" if res.get("ok") else f"error: {res.get('error')}" + if cmd == "pause_entries": + ex.set_operator_paused(True) + return "ok: entries paused" + if cmd == "resume_entries": + ex.set_operator_paused(False) + return "ok: entries resumed" + return f"error: unknown command {cmd!r}" + except Exception as exc: # noqa: BLE001 — one bad command must not kill the loop + logger.warning("operator command %s failed", cmd, exc_info=True) + return f"error: {exc}" + + def _emit_done(self, p: dict, result: str) -> None: + try: + self.store.emit(OperatorCommandEvent( + timestamp=self._now(), mode=self._mode, correlation_id=self.session_id, + event_type=EventType.OPERATOR_COMMAND_DONE, + message=f"operator command {p.get('command')} done: {result}", + command=str(p.get("command") or ""), symbol=p.get("symbol"), + requested_by=str(p.get("requested_by") or "operator"), + nonce=str(p.get("nonce") or ""), result=result, + )) + except Exception: # noqa: BLE001 + logger.warning("operator-command done-echo emit failed", exc_info=True) diff --git a/storage/event_schema.py b/storage/event_schema.py index 902c8b4..5b8017a 100644 --- a/storage/event_schema.py +++ b/storage/event_schema.py @@ -44,6 +44,16 @@ class EventType(str, Enum): SYMBOL_EVALUATION_DETAIL = "symbol_evaluation_detail" ENTRY_GATE_EVALUATED = "entry_gate_evaluated" EXPERIMENT_VERDICT = "experiment_verdict" + OPERATOR_COMMAND = "operator_command" + OPERATOR_COMMAND_DONE = "operator_command_done" + + +# The complete operator command vocabulary — the ONLY commands the bus accepts. +# Deliberately closed: adding a parameterized command must redo the audit-trail + +# replay-safety analysis (see plan 007 maintenance notes). +OPERATOR_COMMANDS: frozenset[str] = frozenset( + {"flatten_all", "flatten_symbol", "pause_entries", "resume_entries"} +) class BaseEvent(BaseModel): @@ -294,6 +304,25 @@ class EntryGateEvaluatedEvent(BaseEvent): ) +class OperatorCommandEvent(BaseEvent): + """A guarded operator kill-switch command (plan 007). Emitted by the dashboard + command endpoint or the qabot ``/flatten``; consumed by the live loop's 5s + guard, which executes it through the existing protective-close helpers and + echoes an ``operator_command_done`` event (same schema, ``result`` set) so a + restart never re-runs it. ``nonce`` (a uuid) is the idempotency key.""" + + event_type: EventType = Field( + default=EventType.OPERATOR_COMMAND, description="Event type identifier" + ) + command: str = Field(description="One of OPERATOR_COMMANDS") + symbol: str | None = Field(default=None, description="Target symbol (flatten_symbol only)") + requested_by: str = Field(default="operator", description="Who issued the command") + nonce: str = Field(description="Idempotency key (uuid) — deduped by the consumer") + result: str | None = Field( + default=None, description="Set on the operator_command_done echo (ok / error text)" + ) + + class BrokerHealthChangedEvent(BaseEvent): """Event when broker health changes.""" diff --git a/telegram_bot.py b/telegram_bot.py index b8cb433..f32e4ee 100644 --- a/telegram_bot.py +++ b/telegram_bot.py @@ -227,6 +227,33 @@ def answer(store, question: str, *, host: str, model: str, timeout: int, 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 @@ -369,6 +396,18 @@ def main() -> None: 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) diff --git a/tests/unit/api/test_dashboard_command.py b/tests/unit/api/test_dashboard_command.py new file mode 100644 index 0000000..c716ede --- /dev/null +++ b/tests/unit/api/test_dashboard_command.py @@ -0,0 +1,109 @@ +"""POST /api/command — token gate, enum validation, event emission (plan 007). + +Drives DashboardHandler.do_POST directly (no loopback socket), against a real +DashboardState pinned to an isolated Postgres schema so emit_command actually +lands an event.""" + +import io +import json +import os +import uuid + +import psycopg2 +import pytest + +from api.main import DashboardHandler, DashboardState +from storage.event_schema import EventType +from storage.event_store import EventStore + + +@pytest.fixture(autouse=True) +def _isolated_pg_schema(monkeypatch): + schema = f"mem_{uuid.uuid4().hex[:12]}" + monkeypatch.setenv("MOMENTUM_PG_SCHEMA", schema) + yield + url = os.environ.get("DATABASE_URL") + if url: + conn = psycopg2.connect(url) + conn.autocommit = True + conn.cursor().execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE') + conn.close() + + +class CaptureHandler(DashboardHandler): + def __init__(self, state, path, body=b"", headers=None): + self.state = state + self.path = path + self.rfile = io.BytesIO(body) + self.headers = headers or {} + self.captured = [] + + def _send_json(self, payload, status=200): + self.captured.append((status, payload)) + + +def _post(state, body, headers=None): + raw = json.dumps(body).encode() + hdrs = {"Content-Length": str(len(raw))} + if headers: + hdrs.update(headers) + h = CaptureHandler(state, "/api/command", raw, hdrs) + h.do_POST() + return h.captured[-1] + + +def test_command_503_when_token_unset(monkeypatch): + monkeypatch.delenv("DASHBOARD_TOKEN", raising=False) + status, payload = _post(DashboardState("momentum"), {"command": "flatten_all"}) + assert status == 503 + assert payload["ok"] is False + + +def test_command_unknown_rejected(monkeypatch): + monkeypatch.setenv("DASHBOARD_TOKEN", "s3cret") + status, payload = _post(DashboardState("momentum"), {"command": "sell_everything"}, + headers={"X-Dashboard-Token": "s3cret"}) + assert status == 400 + assert "unknown command" in payload["error"] + + +def test_command_flatten_symbol_requires_symbol(monkeypatch): + monkeypatch.setenv("DASHBOARD_TOKEN", "s3cret") + status, payload = _post(DashboardState("momentum"), {"command": "flatten_symbol"}, + headers={"X-Dashboard-Token": "s3cret"}) + assert status == 400 + + +def test_command_bad_token_401(monkeypatch): + monkeypatch.setenv("DASHBOARD_TOKEN", "s3cret") + status, _ = _post(DashboardState("momentum"), {"command": "flatten_all"}, + headers={"X-Dashboard-Token": "wrong"}) + assert status == 401 + + +def test_command_valid_emits_event(monkeypatch): + monkeypatch.setenv("DASHBOARD_TOKEN", "s3cret") + state = DashboardState("momentum") + status, payload = _post(state, {"command": "flatten_all"}, + headers={"X-Dashboard-Token": "s3cret"}) + assert status == 200 and payload["ok"] is True and payload["nonce"] + store = EventStore("momentum") + try: + events = store.query_events(event_type=EventType.OPERATOR_COMMAND.value, limit=None) + assert len(events) == 1 + assert json.loads(events[0]["payload_json"])["command"] == "flatten_all" + finally: + store.close() + + +def test_command_emits_even_without_execution_service(monkeypatch): + # the standalone dashboard (execution_service=None) is a thin emitter + monkeypatch.setenv("DASHBOARD_TOKEN", "s3cret") + state = DashboardState("momentum", execution_service=None) + status, payload = _post(state, {"command": "pause_entries"}, + headers={"X-Dashboard-Token": "s3cret"}) + assert status == 200 and payload["ok"] is True + + +if __name__ == "__main__": # pragma: no cover + pytest.main([__file__, "-q"]) diff --git a/tests/unit/test_operator_commands.py b/tests/unit/test_operator_commands.py new file mode 100644 index 0000000..4359799 --- /dev/null +++ b/tests/unit/test_operator_commands.py @@ -0,0 +1,140 @@ +"""Operator command bus: consumer dedup/staleness, pause on BOTH entry paths, +flatten-does-not-self-restart, and the qabot slash-command carve-out (plan 007).""" + +import uuid +from datetime import datetime, timedelta + +from runtime.operator_commands import OperatorCommandConsumer +from storage.event_schema import EventMode, EventType, OperatorCommandEvent +from storage.event_store import EventStore +from tests.unit.test_trading_execution import FakeExecutor, emit_signal, make_service + + +class FakeExecution: + def __init__(self): + self.closed = [] + self.flattened = [] + self.paused = None + + def close_session(self, reason="eod_flatten"): + self.closed.append(reason) + return {"closed_positions": ["AAA"], "cancelled_entries": 0, "errors": []} + + def flatten_symbol(self, symbol, reason="operator_flatten_symbol"): + self.flattened.append(symbol) + return {"ok": True, "symbol": symbol} + + def set_operator_paused(self, paused): + self.paused = bool(paused) + + +def _emit_command(store, command, symbol=None, when=None, nonce=None): + store.emit(OperatorCommandEvent( + timestamp=when or datetime.now(), mode=EventMode.PAPER, + message=f"cmd {command}", command=command, symbol=symbol, + requested_by="test", nonce=nonce or uuid.uuid4().hex)) + + +def test_command_consumed_once_nonce_dedup(): + store = EventStore(":memory:") + ex = FakeExecution() + _emit_command(store, "flatten_all") + consumer = OperatorCommandConsumer(store, ex) + assert len(consumer.poll()) == 1 + assert ex.closed == ["operator_flatten"] + # second poll: already-consumed nonce -> no re-execution + assert consumer.poll() == [] + assert ex.closed == ["operator_flatten"] + store.close() + + +def test_done_event_prevents_reexecution_after_restart(): + store = EventStore(":memory:") + _emit_command(store, "flatten_all") + OperatorCommandConsumer(store, FakeExecution()).poll() # process A executes + echoes _done + ex2 = FakeExecution() + # process B boots fresh, reloads the _done nonce, must NOT re-run + assert OperatorCommandConsumer(store, ex2).poll() == [] + assert ex2.closed == [] + store.close() + + +def test_stale_command_ignored_on_boot(): + store = EventStore(":memory:") + ex = FakeExecution() + _emit_command(store, "flatten_all", when=datetime.now() - timedelta(minutes=20)) + _emit_command(store, "flatten_symbol", symbol="AAA") # fresh + consumer = OperatorCommandConsumer(store, ex, stale_seconds=600) + consumer.poll() + assert ex.closed == [] # the 20-min-old flatten_all was NOT run + assert ex.flattened == ["AAA"] # the fresh one was + store.close() + + +def test_flatten_all_routes_through_close_session_not_self_restart(): + import run_live_paper + run_live_paper._clean_exit_requested.clear() + store = EventStore(":memory:") + ex = FakeExecution() + _emit_command(store, "flatten_all") + OperatorCommandConsumer(store, ex).poll() + assert ex.closed == ["operator_flatten"] + # the operator flatten must NOT trip plan 003's daily-restart exit + assert not run_live_paper._clean_exit_requested.is_set() + store.close() + + +def test_pause_resume_dispatch(): + store = EventStore(":memory:") + ex = FakeExecution() + _emit_command(store, "pause_entries") + OperatorCommandConsumer(store, ex).poll() + assert ex.paused is True + _emit_command(store, "resume_entries") + OperatorCommandConsumer(store, ex).poll() + assert ex.paused is False + store.close() + + +def test_done_echo_records_result(): + store = EventStore(":memory:") + _emit_command(store, "flatten_symbol", symbol="BBB") + OperatorCommandConsumer(store, FakeExecution()).poll() + done = store.query_events(event_type=EventType.OPERATOR_COMMAND_DONE.value, limit=None) + assert len(done) == 1 + + +# ---- pause blocks BOTH entry paths (the drift class plan 005 warns about) ---- +def test_pause_blocks_both_entry_paths(): + store = EventStore(":memory:") + emit_signal(store) # a would-otherwise-pass ready signal + svc = make_service(store, FakeExecutor()) + svc.set_operator_paused(True) + assert svc.request_approvals_for_ready_signals() == [] # SLOW path blocked + fast = svc.submit_breakout_now("GOOD", 14.0, 13.45, last_price=14.1) + assert fast.get("skipped") == "paused" # FAST path blocked + # resume re-enables the slow path (the same signal now creates a request) + svc.set_operator_paused(False) + assert len(svc.request_approvals_for_ready_signals()) == 1 + store.close() + + +# ---- qabot slash-command carve-out + auth ---- +def test_qabot_parse_flatten_command(): + from telegram_bot import parse_flatten_command + assert parse_flatten_command("/flatten") == ("flatten_all", None) + assert parse_flatten_command("/flatten nvda") == ("flatten_symbol", "NVDA") + assert parse_flatten_command("please flatten everything") is None # prose still refuses + assert parse_flatten_command("how did we do?") is None + + +def test_qabot_unauthorized_chat_refused(): + from telegram_bot import is_authorized + assert is_authorized("999", "123") is False + assert is_authorized("123", "123") is True + assert is_authorized("123", None) is False # fail closed when unconfigured + + +if __name__ == "__main__": # pragma: no cover + import pytest + pytest.main([__file__, "-q"]) diff --git a/trading_execution.py b/trading_execution.py index 7918b1e..b4027e5 100644 --- a/trading_execution.py +++ b/trading_execution.py @@ -445,6 +445,11 @@ class TradingExecutionService: self._data_halt = False # session closed (end-of-day flatten): no new entries for the rest of the day self._session_closed = False + # operator PAUSE (plan 007 command bus): blocks new entries on BOTH the + # slow auto path and the fast trigger path until a resume command clears + # it. In-memory + recoverable — unlike _session_closed it does not flatten + # and a fresh command re-enables entries. + self._operator_paused = False # today's EOD flatten already SUCCEEDED (rebuilt from events on boot) — # the daily-restart loop reads this so a post-flatten fresh process # neither re-flattens nor re-exits (restart-churn guard) @@ -1082,11 +1087,20 @@ class TradingExecutionService: symbol=ctx.symbol, source=ctx.source, passed=decision.passed, reject_reason=decision.reason, gates=decision.gates)) + def set_operator_paused(self, paused: bool) -> None: + """Operator PAUSE/RESUME (command bus). Blocks new entries on both paths.""" + self._operator_paused = bool(paused) + def request_approvals_for_ready_signals(self, rank_map: dict | None = None) -> list[str]: """Turn fresh ready signals into approval requests. Returns order ids.""" if not self.settings.enabled: return [] + # operator pause — slow-path half of the BOTH-paths check (submit_breakout_now + # holds the other half; missing either would let pause leak on that path). + if self._operator_paused: + return [] + if self._daily_loss_breach(): return [] @@ -1350,6 +1364,10 @@ class TradingExecutionService: """ if not self.settings.enabled: return {"ok": False, "skipped": "disabled"} + # operator pause — fast-path half of the BOTH-paths check (see + # request_approvals_for_ready_signals for the slow-path half). + if self._operator_paused: + return {"ok": False, "skipped": "paused"} if self._daily_loss_breach(): return {"ok": False, "skipped": "halted"} if self._open_position_count() >= self.settings.max_concurrent_positions: @@ -1596,6 +1614,29 @@ class TradingExecutionService: ) return result + @_locked + def flatten_symbol(self, symbol: str, reason: str = "operator_flatten_symbol") -> dict: + """Operator single-name flatten (command bus): cancel the symbol's resting + protective legs, then market-close it (the naked-leg-safe helper shared with + the exit manager). Does NOT block re-entry / halt the session — that's the + whole-book ``close_session``'s job. Best-effort: never raises.""" + client = getattr(self.executor, "client", None) + if client is None or not hasattr(client, "close_position"): + return {"ok": False, "symbol": symbol, "error": "no broker client to flatten"} + try: + self._release_and_close(client, symbol) + except Exception as exc: # noqa: BLE001 + return {"ok": False, "symbol": symbol, "error": str(exc)} + self.store.emit( + RiskRuleTriggeredEvent( + timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id, + message=f"Operator flatten {symbol} ({reason})", + rule_type="operator_flatten_symbol", rule_value=0.0, + current_state={"symbol": symbol}, action_taken="flatten_symbol", + ) + ) + return {"ok": True, "symbol": symbol} + def flatten_overnight_carries(self, today_et, reason: str = "open_catchup") -> dict: """Catch-up flatten: close positions CARRIED from a prior session.