From 282c753aaff4196c21e9add71de6f78731e1997c Mon Sep 17 00:00:00 2001 From: Elmehdi Aitbrahim Date: Fri, 24 Jul 2026 01:20:26 -0400 Subject: [PATCH] =?UTF-8?q?feat(cli):=20keel=20tui=20=E2=80=94=20live=20re?= =?UTF-8?q?ad-only=20operator=20dashboard?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A full-screen, auto-refreshing terminal dashboard built directly on the pure `gather_status(...) -> StatusReport` that `keel status` (PR #138) was left as the substrate for. It is strictly a *view* over the same report: mode, kill-switch, autonomy, Rail 11 drawdown/equity, open positions, rule counts, per-product data freshness and subscriptions, colour-coded and refreshed on an interval so an operator can watch the funded paper-forward at a glance. - Pure, testable core (`build_screen`/`render_plain`/`_freshness_style`); never re-derives Rail 11 / freshness / autonomy logic. - stdlib `curses` only — no new dependency (in keeping with the project's hand-rolled ethos). `curses` imported lazily so the module stays importable and the pure-core tests stay portable. - Strictly read-only: NEVER calls the broker or touches the network, and cannot confirm/kill/arm anything. - `--once` renders a single frame to stdout (pipes/CI); `--interval N` sets the refresh cadence; the live loop re-opens the repo each poll (so it reflects a separate `keel agent`'s committed writes) and survives a transient DB read error by painting an alert instead of crashing. 39 new tests. Design spec: docs/superpowers/specs/2026-07-24-tui-dashboard-design.md Co-Authored-By: Claude Opus 4.8 --- .../specs/2026-07-24-tui-dashboard-design.md | 119 ++++ keel/cli.py | 8 + keel/commands/tui.py | 387 ++++++++++++ tests/commands/test_tui.py | 597 ++++++++++++++++++ 4 files changed, 1111 insertions(+) create mode 100644 docs/superpowers/specs/2026-07-24-tui-dashboard-design.md create mode 100644 keel/commands/tui.py create mode 100644 tests/commands/test_tui.py diff --git a/docs/superpowers/specs/2026-07-24-tui-dashboard-design.md b/docs/superpowers/specs/2026-07-24-tui-dashboard-design.md new file mode 100644 index 00000000..edbccce0 --- /dev/null +++ b/docs/superpowers/specs/2026-07-24-tui-dashboard-design.md @@ -0,0 +1,119 @@ +# `keel tui` — live read-only operator dashboard + +**Status:** design → implement (this branch). **Date:** 2026-07-24. + +## Motivation + +The funded paper-forward (5-trend Turtle, $10k + $500/mo) is now live and will accrue trades +toward the n=100 evidence floor over *months*. Watching it means re-running `keel status` by hand. +`keel status` (PR #138) was deliberately built as the *substrate* for a TUI: `gather_status(...) +-> StatusReport` is a pure, broker-free report and `keel status --json` is its forward-compatible +shape. This spec adds the auto-refreshing full-screen dashboard that substrate was for. + +The user named "TUI" explicitly as the next feature to build. It is unblocked *right now* — it does +not require the monorepo split or a `keel-client` protocol; a read-only, single-process dashboard +sits directly on `gather_status`. + +## Non-goals + +- **No actions.** The TUI is strictly read-only, exactly like `keel status` — it NEVER touches the + broker or network and cannot confirm/kill/arm anything. Acting from a TUI (confirm a pending + order, kill/resume) is a separate, larger feature that needs its own gating design. +- **No streaming / event push.** Poll-and-repaint on an interval, not `stream_events()`. +- **No new runtime dependency.** The project is deliberately stdlib-conservative (hand-rolled + indicators over numpy, hand-rolled Decimal metrics, declined the quant stack). The TUI uses + stdlib **`curses`** only. No `rich`/`textual`. +- **No multi-process health.** A TUI that can tell a hung `keel agent` from a healthy one needs the + `app_health` table from the monorepo spec, which does not exist. Out of scope. + +## Design + +`keel/commands/tui.py`, mirroring the two-layer shape of `keel/commands/status.py`: + +### 1. Pure screen model (the testable core) + +```python +@dataclass(frozen=True) +class ScreenLine: + text: str + style: str # one of: "heading" | "normal" | "ok" | "alert" | "warn" | "muted" + +def build_screen(report: StatusReport, now_ts: int) -> list[ScreenLine]: ... +``` + +`build_screen` turns a `StatusReport` into styled rows. It reuses the *report* from +`gather_status` — it must not re-derive any status logic (Rail 11, freshness, etc.). Sections, +in order, each with semantic styling: + +- **Title / mode** — `keel · mode` + a formatted `now_ts`. `heading`. +- **Kill switch** — `alert` (red) when engaged, `ok` (green) when clear. +- **Autonomy** — `alert` when ON (orders placed without asking), `muted` when off; the + lapsed/lapses-at sub-line as `muted`; the profile-unreadable warning as `warn`. +- **Equity / drawdown / Rail 11** — HWM, total & weekly drawdown vs their ceilings, and the + Rail 11 line styled by `report.rail11_status`: `HALTED`→`alert`, `unknown`→`warn`, `ok`→`ok`. + In paper mode, the `paper_cash_usdc` line. +- **Open positions** — a header line then one row per position (id, product, qty, entry, + age, rule); a position with `has_bracket=False` renders its bracket note as `warn`. +- **Rules** — the status counts line, then each live rule (`live` rules styled `alert`-ish/normal + since a live rule means real money can move). +- **Data freshness** — one row per product; each styled by staleness via `_freshness_style` + (see below): fresh→`ok`, stale→`warn`, no-data→`warn`. +- **Subscriptions** — one row per venue. +- **Footer** — `q quit · refreshing every Ns · read-only (no broker)`. `muted`. + +`_freshness_style(granularity: str | None, age_sec: int | None) -> str`: a pure helper. No local +data or unknown granularity → `warn`. Otherwise compare `age_sec` to the granularity's own period +(a daily series older than ~2 days is stale): `age_sec > 2 * period_seconds(granularity)` → `warn`, +else `ok`. Keep the period lookup a small dict keyed by `Granularity.value`. + +### 2. Rendering + loop (thin I/O) + +- `render_plain(report, now_ts) -> list[str]` — the `ScreenLine.text` values only (styles + dropped). Drives `--once` and any non-tty use; directly testable. +- `_paint(stdscr, lines: list[ScreenLine]) -> None` — paint styled lines into a curses window, + mapping each style to a curses attribute (bold/colour), truncating to the window width and + clipping to its height so a small terminal never raises. Tested against a *fake* stdscr that + records `addstr` calls — no real terminal needed. +- `run_once(open_state, now_fn, echo) -> None` — `report = gather_status(*open_state())`; echo + `render_plain`. `open_state: Callable[[], tuple[Repository, Config]]`, `now_fn: Callable[[], + int]`, `echo: Callable[[str], None]` — all injectable, so `run_once` is testable with fakes and + no CliRunner/terminal. +- `run_live(open_state, now_fn, interval) -> None` — `import curses` *lazily inside the function* + (keeps the module importable where curses is absent, and keeps `build_screen` tests portable), + then `curses.wrapper` a loop: poll `gather_status`, `_paint`, `getch` with a timeout of + `interval` seconds; quit on `q`/`Q`/Ctrl-C. **Re-open the repo each poll** (via `open_state`) so + the dashboard reflects writes committed by a separate `keel agent` process. + +### 3. CLI + +`keel tui`, registered in `keel/cli.py` via `cli.add_command(tui_cmd)`: + +``` +--interval FLOAT seconds between refreshes (default 5.0; must be > 0) +--once render a single frame to stdout and exit (no curses; for pipes/CI) +``` + +`open_state` closes over `_open_repo(ctx)` / `_load_cfg(ctx)` from `keel.commands._common`, called +fresh each poll. Default (interactive) path calls `run_live`; `--once` calls `run_once` with +`click.echo`. No disclaimer footer in the live loop (it owns the screen); `--once` may print the +disclaimer after the frame, matching `status`'s scripting-friendliness. + +## Testing (TDD) + +Unit tests in `tests/commands/test_tui.py`, driven by `StatusReport` fixtures (reuse the shapes in +`tests/commands/test_status.py`): + +1. `build_screen` includes mode, kill-switch, HWM/drawdown, each open position, rule counts, each + freshness row, subscriptions. +2. Style logic: kill-switch engaged → `alert`; Rail 11 `HALTED`→`alert`, `unknown`→`warn`, + `ok`→`ok`; autonomy ON → `alert`; a bracket-less position → a `warn` line; stale freshness → + `warn` via `_freshness_style` (parametrised over granularity/age). +3. `render_plain` returns the same text as `build_screen`'s lines (styles stripped). +4. `_paint` against a fake stdscr: does not raise on a tiny window (clips), maps styles to attrs, + writes each visible line. +5. `run_once` with a fake `open_state`/`now_fn` echoes a full frame. +6. CLI: `CliRunner` invokes `keel tui --once --db --config ` against a real temp DB and + prints a frame; `--interval 0` is rejected. + +Acceptance: `uv run pytest -q` green (test count up), `uv run ruff check` clean, and `keel tui +--once` prints a coherent dashboard against the real `keel.db`. diff --git a/keel/cli.py b/keel/cli.py index 5181b90b..2e2002a8 100644 --- a/keel/cli.py +++ b/keel/cli.py @@ -83,6 +83,7 @@ from keel.commands.status import status_cmd from keel.commands.subscription import subscription_group from keel.commands.trials import trials_group +from keel.commands.tui import tui_cmd from keel.commands.withdrawals import withdrawals_group from keel.compliance import purification as purification_mod from keel.compliance import screen as screen_mod @@ -1639,6 +1640,13 @@ def simulate( cli.add_command(status_cmd) +# -- tui (live, read-only, full-screen operator dashboard, no broker call) ----------------------- + +# `keel status` was built as the substrate for this: `tui_cmd` is a curses view over the same +# `gather_status` report, defined in `keel.commands.tui` and registered here. +cli.add_command(tui_cmd) + + # -- kill / resume ------------------------------------------------------------------------------ diff --git a/keel/commands/tui.py b/keel/commands/tui.py new file mode 100644 index 00000000..120659b2 --- /dev/null +++ b/keel/commands/tui.py @@ -0,0 +1,387 @@ +"""`keel tui` -- a live, read-only, full-screen operator dashboard. + +`keel status` (`keel/commands/status.py`) was deliberately built as the substrate for this: its +`gather_status(repo, config, now_ts) -> StatusReport` is a pure, broker-free report, and +`keel status --json` is its forward-compatible shape. `keel tui` is strictly a *view* over that +same report -- it never re-derives Rail 11, freshness, or autonomy logic, only styles it. + +Like `keel status`, this NEVER calls the broker or touches the network, and it is strictly +read-only: it cannot confirm, kill, or arm anything. Acting from the dashboard (confirming a +pending order, kill/resume) is a separate, larger feature with its own gating design. + +Two layers, mirroring `status.py`'s split: + +- `build_screen` is a PURE function of `(StatusReport, now_ts)` -> `list[ScreenLine]`, directly + unit-testable without curses or a CliRunner. `render_plain` and `_freshness_style` are pure + helpers built the same way. +- `_paint` (curses rendering), `run_once` (single-frame, `--once`/pipes/CI), and `run_live` + (the auto-refreshing `curses.wrapper` loop) are the thin I/O layer. `curses` is imported + lazily inside the functions that need it, so this module stays importable -- and + `build_screen`/`render_plain` tests stay portable -- even where a real terminal is absent. +""" + +from __future__ import annotations + +import time +from collections.abc import Callable +from dataclasses import dataclass +from typing import Any + +import click + +from keel.commands._common import DISCLAIMER, _load_cfg, _open_repo +from keel.commands.status import StatusReport, _human_age, gather_status +from keel.config import Config +from keel.data.repository import Repository +from keel.types import Granularity + +# -- the pure screen model (the testable core) -------------------------------------------------- + + +@dataclass(frozen=True) +class ScreenLine: + text: str + style: str # one of: "heading" | "normal" | "ok" | "alert" | "warn" | "muted" + + +# Period, in seconds, of each configured candle granularity -- keyed by `Granularity.value` (the +# same strings `ProductFreshness.granularity` stores) so `_freshness_style` never has to import +# the enum member itself, just compare strings. +_GRANULARITY_PERIOD_SEC: dict[str, int] = { + Granularity.ONE_MINUTE.value: 60, + Granularity.FIVE_MINUTE.value: 300, + Granularity.FIFTEEN_MINUTE.value: 900, + Granularity.ONE_HOUR.value: 3600, + Granularity.SIX_HOUR.value: 21600, + Granularity.ONE_DAY.value: 86400, +} + + +def _freshness_style(granularity: str | None, age_sec: int | None) -> str: + """`"ok"` when a product's newest candle is within 2x its own granularity's period, `"warn"` + when it is staler than that -- or when there is no local data / unknown granularity to begin + with (a daily series a couple of days old is fine; a couple of *periods* old is stale).""" + if granularity is None or age_sec is None: + return "warn" + period = _GRANULARITY_PERIOD_SEC.get(granularity) + if period is None: + return "warn" + return "warn" if age_sec > 2 * period else "ok" + + +def _blank() -> ScreenLine: + return ScreenLine("", "normal") + + +def _title_lines(report: StatusReport, now_ts: int) -> list[ScreenLine]: + return [ScreenLine(f"keel · {report.mode} mode · now={now_ts}", "heading")] + + +def _kill_switch_lines(report: StatusReport) -> list[ScreenLine]: + if report.kill_switch_engaged: + return [ScreenLine("kill_switch: ENGAGED (halted)", "alert")] + return [ScreenLine("kill_switch: clear", "ok")] + + +def _autonomy_lines(report: StatusReport) -> list[ScreenLine]: + a = report.autonomy + lines: list[ScreenLine] = [] + if not a.profile_readable: + lines.append( + ScreenLine( + " WARNING: profile row unreadable -- reporting autonomy as OFF (safe reading).", + "warn", + ) + ) + if a.live: + lines.append(ScreenLine("autonomy: ON -- orders placed WITHOUT asking", "alert")) + else: + lines.append(ScreenLine("autonomy: off", "muted")) + if a.autonomous and not a.live: + lines.append(ScreenLine(f" (was ON but LAPSED at {a.autonomous_until})", "muted")) + elif a.live and a.autonomous_until is not None: + lines.append(ScreenLine(f" lapses at {a.autonomous_until}", "muted")) + return lines + + +def _rail11_style(status: str) -> str: + if status == "HALTED": + return "alert" + if status == "unknown": + return "warn" + return "ok" + + +def _equity_lines(report: StatusReport) -> list[ScreenLine]: + lines: list[ScreenLine] = [] + mode_text = f"equity_state_mode: {report.equity_state_mode or 'unknown'}" + lines.append(ScreenLine(mode_text, "normal")) + hwm = report.high_water_mark if report.high_water_mark is not None else "unknown" + lines.append(ScreenLine(f"high_water_mark: {hwm}", "normal")) + dd_total = report.drawdown_total_pct if report.drawdown_total_pct is not None else "unknown" + dd_weekly = report.drawdown_weekly_pct if report.drawdown_weekly_pct is not None else "unknown" + lines.append( + ScreenLine( + f"drawdown: total={dd_total} (ceiling {report.max_total_dd_pct}) " + f"weekly={dd_weekly} (ceiling {report.max_weekly_dd_pct})", + "normal", + ) + ) + rail11_text = f"rail11 (drawdown breaker): {report.rail11_status}" + lines.append(ScreenLine(rail11_text, _rail11_style(report.rail11_status))) + if report.mode == "paper": + lines.append(ScreenLine(f"paper_cash_usdc: {report.paper_cash_usdc}", "normal")) + return lines + + +def _open_position_lines(report: StatusReport) -> list[ScreenLine]: + lines: list[ScreenLine] = [] + if not report.open_positions: + lines.append(ScreenLine("open positions: no open positions", "normal")) + return lines + lines.append(ScreenLine(f"open positions ({len(report.open_positions)}):", "normal")) + for pos in report.open_positions: + bracket_note = "bracketed" if pos.has_bracket else "NO bracket" + row_style = "normal" if pos.has_bracket else "warn" + lines.append( + ScreenLine( + f" [{pos.id}] {pos.product_id} qty={pos.qty} entry={pos.entry_price} " + f"opened_at={pos.opened_at} rule={pos.rule_name} ({bracket_note})", + row_style, + ) + ) + return lines + + +def _rule_lines(report: StatusReport) -> list[ScreenLine]: + lines: list[ScreenLine] = [] + counts = " ".join(f"{status}={count}" for status, count in sorted(report.rule_counts.items())) + lines.append(ScreenLine(f"rules: {counts or 'none'}", "normal")) + for rule in report.live_rules: + lines.append( + ScreenLine( + f" live [{rule.id}] {rule.kind} product={rule.product_id} params={rule.params}", + "alert", + ) + ) + return lines + + +def _freshness_lines(report: StatusReport) -> list[ScreenLine]: + lines: list[ScreenLine] = [ScreenLine("data freshness:", "normal")] + for f in report.data_freshness: + style = _freshness_style(f.granularity, f.age_sec) + if f.last_ts is None: + lines.append(ScreenLine(f" {f.product_id}: no data", style)) + else: + age_text = f" {f.product_id} ({f.granularity}): {_human_age(f.age_sec or 0)}" + lines.append(ScreenLine(age_text, style)) + return lines + + +def _subscription_lines(report: StatusReport) -> list[ScreenLine]: + if not report.subscriptions: + return [] + lines: list[ScreenLine] = [ScreenLine("subscriptions:", "normal")] + for s in report.subscriptions: + cap = "unlimited" if s.effective_cap is None else str(s.effective_cap) + sub_text = f" {s.venue}: tier={s.tier_name} status={s.effective_status} cap={cap}" + lines.append(ScreenLine(sub_text, "normal")) + return lines + + +def build_screen(report: StatusReport, now_ts: int) -> list[ScreenLine]: + """Turn a `StatusReport` into styled rows -- a PURE function of the report, reusing every + logic decision (Rail 11, freshness, autonomy) `gather_status` already made. Never re-derives + status; only styles it.""" + lines: list[ScreenLine] = [] + lines.extend(_title_lines(report, now_ts)) + lines.extend(_kill_switch_lines(report)) + lines.extend(_autonomy_lines(report)) + lines.append(_blank()) + lines.extend(_equity_lines(report)) + lines.append(_blank()) + lines.extend(_open_position_lines(report)) + lines.append(_blank()) + lines.extend(_rule_lines(report)) + lines.append(_blank()) + lines.extend(_freshness_lines(report)) + sub_lines = _subscription_lines(report) + if sub_lines: + lines.append(_blank()) + lines.extend(sub_lines) + lines.append(_blank()) + # Deliberately interval-independent: `build_screen` doesn't know the poll interval, so it + # cannot say "refreshing every Ns" without threading that through its signature. The live + # loop is free to show its own interval-bearing status line if desired. + lines.append(ScreenLine("q quit · read-only (no broker)", "muted")) + return lines + + +def render_plain(report: StatusReport, now_ts: int) -> list[str]: + """The `.text` of each `build_screen` line, styles dropped -- drives `--once` and any + non-tty use.""" + return [line.text for line in build_screen(report, now_ts)] + + +# -- render + loop (thin I/O) -------------------------------------------------------------------- + + +def _style_attrs() -> dict[str, int]: + """Map each `ScreenLine.style` to a curses attribute bitmask. Uses only attribute constants + that are safe to read without a real terminal having called `initscr()` (`A_BOLD`, `A_DIM`, + ...); colour pairs are layered on top only when `curses.has_colors()` can be queried without + raising (i.e. a real terminal did initialise), so this stays callable against a fake stdscr + in tests. Any `curses.error` while querying/initialising colour support is swallowed -- the + attribute-only styling below is still a coherent, if colourless, rendering.""" + import curses + + attrs: dict[str, int] = { + "heading": curses.A_BOLD, + "alert": curses.A_BOLD | curses.A_REVERSE, + "warn": curses.A_BOLD | curses.A_UNDERLINE, + "ok": curses.A_NORMAL, + "normal": curses.A_NORMAL, + "muted": curses.A_DIM, + } + try: + if curses.has_colors(): + curses.start_color() + curses.use_default_colors() + curses.init_pair(1, curses.COLOR_RED, -1) + curses.init_pair(2, curses.COLOR_YELLOW, -1) + curses.init_pair(3, curses.COLOR_GREEN, -1) + attrs["alert"] |= curses.color_pair(1) + attrs["warn"] |= curses.color_pair(2) + attrs["ok"] |= curses.color_pair(3) + except curses.error: + pass + return attrs + + +def _paint(stdscr: Any, lines: list[ScreenLine]) -> None: + """Paint styled `lines` into a curses window, clipped to its current size so a tiny terminal + never raises `curses.error`. Testable against a fake `stdscr` (records `addstr(y, x, text, + attr)`, has `getmaxyx()`) -- no real terminal required.""" + import curses + + height, width = stdscr.getmaxyx() + attrs = _style_attrs() + stdscr.erase() + for y, line in enumerate(lines): + if y >= height: + break + max_width = max(width - 1, 0) + text = line.text[:max_width] + attr = attrs.get(line.style, curses.A_NORMAL) + try: + stdscr.addstr(y, 0, text, attr) + except curses.error: + # Classic bottom-right-corner write: some terminals raise when the cursor would + # advance past the last cell. Never fatal to the dashboard. + pass + stdscr.refresh() + + +OpenState = Callable[[], "tuple[Repository, Config]"] +NowFn = Callable[[], int] +Echo = Callable[[str], None] + + +def run_once(open_state: OpenState, now_fn: NowFn, echo: Echo) -> None: + """Render a single frame and hand each line to `echo` -- drives `--once` (pipes/CI) and is + directly testable with fakes, no CliRunner or terminal needed.""" + repo, config = open_state() + now_ts = now_fn() + report = gather_status(repo, config, now_ts) + for line in render_plain(report, now_ts): + echo(line) + + +def run_live(open_state: OpenState, now_fn: NowFn, interval: float) -> None: + """The auto-refreshing dashboard: `curses.wrapper` a loop that re-opens the repo (via + `open_state`) every poll -- so it reflects writes committed by a separate `keel agent` + process -- gathers a fresh report, paints it, then waits up to `interval` seconds for a + keypress. Quits on `q`/`Q`; a `KeyboardInterrupt` (Ctrl-C) exits gracefully rather than + dumping a traceback onto a terminal `curses.wrapper` may not have fully restored.""" + import curses + + def _loop(stdscr: Any) -> None: + try: + curses.curs_set(0) + except curses.error: + # Not every terminal has a hideable cursor -- never fatal to the dashboard. + pass + while True: + now_ts = now_fn() + try: + # `Repository` exposes no public connection handle or `close()` (only the + # private `_conn`), so there is nothing safe to close here each poll -- `repo` + # simply falls out of scope and is garbage-collected. + repo, config = open_state() + report = gather_status(repo, config, now_ts) + _paint(stdscr, build_screen(report, now_ts)) + except Exception as exc: + # A transient read error (e.g. `sqlite3.OperationalError: database is locked` + # from a concurrent `keel agent` writer) must never kill the dashboard -- + # paint an alert line and keep polling. `KeyboardInterrupt` is not caught here + # (it isn't an `Exception`) so Ctrl-C still reaches the outer handler below. + _paint(stdscr, [ScreenLine(f"status read failed: {exc} -- retrying...", "alert")]) + stdscr.timeout(int(interval * 1000)) + ch = stdscr.getch() + if ch in (ord("q"), ord("Q")): + break + + try: + curses.wrapper(_loop) + except KeyboardInterrupt: + pass + + +# -- the command ---------------------------------------------------------------------------- + + +@click.command("tui") +@click.option( + "--interval", + type=float, + default=5.0, + show_default=True, + help="Seconds between refreshes.", +) +@click.option( + "--once", + is_flag=True, + default=False, + help="Render a single frame to stdout and exit (no curses; for pipes/CI).", +) +@click.pass_context +def tui_cmd(ctx: click.Context, interval: float, once: bool) -> None: + """Live, read-only, full-screen operator dashboard -- never calls the broker. + + A view over the same `gather_status` report `keel status` prints once: mode, kill-switch, + autonomy, Rail 11 drawdown/equity state, open positions, rule counts, per-product data + freshness, and subscriptions, auto-refreshing on an interval. Strictly read-only: it cannot + confirm, kill, or arm anything -- for that, use the dedicated commands. + + `--once` renders a single frame to stdout and exits without touching curses, for pipes/CI, + matching `status`'s scripting-friendliness (and prints the disclaimer footer after the + frame). The default, interactive path owns the whole screen via `curses.wrapper` and re-opens + the repo every poll so it reflects writes committed by a separate `keel agent` process; quit + with `q`. + """ + if interval <= 0: + raise click.ClickException("--interval must be > 0") + + def open_state() -> tuple[Repository, Config]: + return _open_repo(ctx), _load_cfg(ctx) + + now_fn: NowFn = lambda: int(time.time()) # noqa: E731 + + if once: + run_once(open_state, now_fn, click.echo) + click.echo("") + click.echo(DISCLAIMER) + return + + run_live(open_state, now_fn, interval) diff --git a/tests/commands/test_tui.py b/tests/commands/test_tui.py new file mode 100644 index 00000000..4b4e30ce --- /dev/null +++ b/tests/commands/test_tui.py @@ -0,0 +1,597 @@ +"""Tests for `keel tui` -- the live, read-only, full-screen operator dashboard. + +`keel tui` is a *view* over `keel status`'s own report: it must not re-derive Rail 11, +freshness, or autonomy logic, only style `StatusReport` into `ScreenLine`s. Mirrors +`tests/commands/test_status.py`'s fixture style (in-memory `Repository`, `_config` helper, +`NOW_TS` constant), plus the pure `build_screen`/`_freshness_style`/`render_plain`/`_paint`/ +`run_once` seams that make the interactive `run_live` loop thin, untested I/O. +""" + +from __future__ import annotations + +import sqlite3 +import sys +from decimal import Decimal +from types import SimpleNamespace +from typing import Any + +import pytest +from click.testing import CliRunner + +from keel.cli import cli +from keel.commands.status import ( + AutonomyStatus, + OpenPositionStatus, + ProductFreshness, + RuleSummary, + StatusReport, + SubscriptionStatusRow, +) +from keel.commands.tui import ( + ScreenLine, + _freshness_style, + _paint, + _style_attrs, + build_screen, + render_plain, + run_live, + run_once, +) +from keel.config import ( + AutoTradeConfig, + Caps, + Config, + DcaConfig, + MarketDataConfig, + MoneyMgmtConfig, +) +from keel.data.db import connect, migrate +from keel.data.repository import Repository +from keel.types import Granularity + +NOW_TS = 1_800_000_000 + + +@pytest.fixture +def repo() -> Repository: + conn = connect(":memory:") + migrate(conn) + r = Repository(conn) + r.set_state("kill_switch", False) + return r + + +def _config(**overrides: Any) -> Config: + base: dict[str, Any] = dict( + allowlist=["BTC", "ETH"], + target_weights={}, + risk_pct=Decimal("0.01"), + caps=Caps( + max_per_order_usd=Decimal("100000"), + max_per_day_usd=Decimal("300000"), + max_exposure_usd=Decimal("1000000"), + max_per_asset_pct=Decimal("1"), + ), + market_data=MarketDataConfig( + granularities=[Granularity.ONE_DAY, Granularity.ONE_HOUR], history_days=365 + ), + auto_trade=AutoTradeConfig(mode="paper", interval_sec=900), + money_mgmt=MoneyMgmtConfig( + max_total_dd_pct=Decimal("0.20"), max_weekly_dd_pct=Decimal("0.08") + ), + dca=DcaConfig(budget_usd=Decimal("50"), cadence_days=7), + ) + base.update(overrides) + return Config(**base) + + +def _base_report(**overrides: Any) -> StatusReport: + base: dict[str, Any] = dict( + now_ts=NOW_TS, + mode="paper", + kill_switch_engaged=False, + autonomy=AutonomyStatus( + live=False, + autonomous=False, + autonomous_until=None, + updated_ts=None, + profile_readable=True, + ), + equity_state_mode="paper", + high_water_mark=Decimal("10000"), + drawdown_total_pct=Decimal("0.05"), + drawdown_weekly_pct=Decimal("0.01"), + max_total_dd_pct=Decimal("0.20"), + max_weekly_dd_pct=Decimal("0.08"), + rail11_status="ok", + paper_cash_usdc=Decimal("955.25"), + open_positions=[], + rule_counts={}, + live_rules=[], + data_freshness=[], + subscriptions=[], + ) + base.update(overrides) + return StatusReport(**base) + + +# -- build_screen: sections present ------------------------------------------------------------ + + +def test_build_screen_includes_mode_and_now() -> None: + report = _base_report(mode="paper") + lines = build_screen(report, NOW_TS) + texts = [line.text for line in lines] + assert any("paper" in t for t in texts) + + +def test_build_screen_includes_kill_switch() -> None: + report = _base_report(kill_switch_engaged=True) + lines = build_screen(report, NOW_TS) + texts = [line.text.lower() for line in lines] + assert any("kill" in t and "engaged" in t for t in texts) + + +def test_build_screen_includes_hwm_and_drawdown() -> None: + report = _base_report(high_water_mark=Decimal("12345.6")) + lines = build_screen(report, NOW_TS) + texts = " ".join(line.text for line in lines) + assert "12345.6" in texts + assert "drawdown" in texts.lower() + + +def test_build_screen_includes_paper_cash_in_paper_mode() -> None: + report = _base_report(mode="paper", paper_cash_usdc=Decimal("42.00")) + lines = build_screen(report, NOW_TS) + texts = " ".join(line.text for line in lines) + assert "42.00" in texts + + +def test_build_screen_omits_paper_cash_outside_paper_mode() -> None: + report = _base_report(mode="confirm", paper_cash_usdc=None) + lines = build_screen(report, NOW_TS) + texts = " ".join(line.text.lower() for line in lines) + assert "paper_cash" not in texts + + +def test_build_screen_includes_each_open_position() -> None: + pos = OpenPositionStatus( + id=1, + product_id="BTC-USD", + rule_name="turtle_breakout", + qty=Decimal("0.01"), + entry_price=Decimal("65000"), + opened_at=NOW_TS - 3600, + has_bracket=True, + ) + report = _base_report(open_positions=[pos]) + lines = build_screen(report, NOW_TS) + texts = " ".join(line.text for line in lines) + assert "BTC-USD" in texts + assert "turtle_breakout" in texts + assert "0.01" in texts + assert "65000" in texts + + +def test_build_screen_includes_rule_counts_and_live_rules() -> None: + rule = RuleSummary( + id=7, kind="turtle_breakout", status="live", product_id="BTC-USD", params={"lookback": 20} + ) + report = _base_report(rule_counts={"live": 1, "candidate": 2}, live_rules=[rule]) + lines = build_screen(report, NOW_TS) + texts = " ".join(line.text for line in lines) + assert "live=1" in texts + assert "candidate=2" in texts + assert "turtle_breakout" in texts + + +def test_build_screen_includes_each_freshness_row() -> None: + freshness = [ + ProductFreshness("BTC-USD", "ONE_HOUR", NOW_TS - 3600, 3600), + ProductFreshness("ETH-USD", None, None, None), + ] + report = _base_report(data_freshness=freshness) + lines = build_screen(report, NOW_TS) + texts = " ".join(line.text for line in lines) + assert "BTC-USD" in texts + assert "ETH-USD" in texts + + +def test_build_screen_includes_subscriptions() -> None: + sub = SubscriptionStatusRow( + venue="coinbase", + tier_name="Preferred", + pacing="opportunistic", + stored_status="active", + effective_status="active", + effective_cap=Decimal("1000"), + ) + report = _base_report(subscriptions=[sub]) + lines = build_screen(report, NOW_TS) + texts = " ".join(line.text for line in lines) + assert "coinbase" in texts + assert "Preferred" in texts + + +def test_build_screen_footer_is_present_and_interval_independent() -> None: + report = _base_report() + lines = build_screen(report, NOW_TS) + footer = lines[-1] + assert footer.style == "muted" + assert "quit" in footer.text.lower() + assert "read-only" in footer.text.lower() + + +# -- style logic -------------------------------------------------------------------------------- + + +def test_kill_switch_engaged_is_alert_style() -> None: + report = _base_report(kill_switch_engaged=True) + lines = build_screen(report, NOW_TS) + kill_line = next(line for line in lines if "kill" in line.text.lower()) + assert kill_line.style == "alert" + + +def test_kill_switch_clear_is_ok_style() -> None: + report = _base_report(kill_switch_engaged=False) + lines = build_screen(report, NOW_TS) + kill_line = next(line for line in lines if "kill" in line.text.lower()) + assert kill_line.style == "ok" + + +@pytest.mark.parametrize( + "rail11_status,expected_style", + [("HALTED", "alert"), ("unknown", "warn"), ("ok", "ok")], +) +def test_rail11_style_matches_status(rail11_status: str, expected_style: str) -> None: + report = _base_report(rail11_status=rail11_status) + lines = build_screen(report, NOW_TS) + rail_line = next(line for line in lines if "rail11" in line.text.lower()) + assert rail_line.style == expected_style + + +def test_autonomy_live_is_alert_style() -> None: + autonomy = AutonomyStatus( + live=True, autonomous=True, autonomous_until=None, updated_ts=NOW_TS, profile_readable=True + ) + report = _base_report(autonomy=autonomy) + lines = build_screen(report, NOW_TS) + autonomy_line = next(line for line in lines if "autonomy" in line.text.lower()) + assert autonomy_line.style == "alert" + + +def test_autonomy_off_is_muted_style() -> None: + autonomy = AutonomyStatus( + live=False, autonomous=False, autonomous_until=None, updated_ts=None, profile_readable=True + ) + report = _base_report(autonomy=autonomy) + lines = build_screen(report, NOW_TS) + autonomy_line = next(line for line in lines if "autonomy" in line.text.lower()) + assert autonomy_line.style == "muted" + + +def test_autonomy_unreadable_profile_is_warn_style() -> None: + autonomy = AutonomyStatus( + live=False, autonomous=False, autonomous_until=None, updated_ts=None, profile_readable=False + ) + report = _base_report(autonomy=autonomy) + lines = build_screen(report, NOW_TS) + warn_line = next(line for line in lines if "unreadable" in line.text.lower()) + assert warn_line.style == "warn" + + +def test_bracketless_position_has_warn_line() -> None: + pos = OpenPositionStatus( + id=2, + product_id="ETH-USD", + rule_name="dca", + qty=Decimal("1"), + entry_price=Decimal("3000"), + opened_at=NOW_TS, + has_bracket=False, + ) + report = _base_report(open_positions=[pos]) + lines = build_screen(report, NOW_TS) + bracket_lines = [line for line in lines if "bracket" in line.text.lower()] + assert any(line.style == "warn" for line in bracket_lines) + + +# -- _freshness_style (pure, parametrised) ----------------------------------------------------- + + +@pytest.mark.parametrize( + "granularity,age_sec,expected", + [ + ("ONE_HOUR", 60, "ok"), + ("ONE_HOUR", 3600, "ok"), + ("ONE_HOUR", 3600 * 3, "warn"), + ("ONE_DAY", 86400, "ok"), + ("ONE_DAY", 86400 * 3, "warn"), + (None, 10, "warn"), + ("ONE_HOUR", None, "warn"), + (None, None, "warn"), + ], +) +def test_freshness_style(granularity: str | None, age_sec: int | None, expected: str) -> None: + assert _freshness_style(granularity, age_sec) == expected + + +# -- render_plain ----------------------------------------------------------------------------- + + +def test_render_plain_matches_build_screen_text() -> None: + report = _base_report() + lines = build_screen(report, NOW_TS) + plain = render_plain(report, NOW_TS) + assert plain == [line.text for line in lines] + + +class _FakeCursesError(Exception): + pass + + +def _fake_curses(*, has_colors: bool = True) -> SimpleNamespace: + """A stand-in `curses` module -- distinct attribute-constant ints, `has_colors()`/ + `color_pair()`/`init_pair()` recorded on `.calls` (in call order), no real terminal + required. Installed via `monkeypatch.setitem(sys.modules, "curses", ...)` since both + `_style_attrs` and `run_live` do `import curses` lazily inside the function body, so the + patched module is what they bind.""" + calls: list[str] = [] + fake = SimpleNamespace( + A_BOLD=1 << 0, + A_DIM=1 << 1, + A_UNDERLINE=1 << 2, + A_REVERSE=1 << 3, + A_NORMAL=0, + COLOR_RED=1, + COLOR_YELLOW=2, + COLOR_GREEN=3, + error=_FakeCursesError, + has_colors=lambda: has_colors, + start_color=lambda: calls.append("start_color"), + use_default_colors=lambda: calls.append("use_default_colors"), + init_pair=lambda n, fg, bg: calls.append(f"init_pair:{n}"), + color_pair=lambda n: 1 << (10 + n), + curs_set=lambda visibility: None, + calls=calls, + ) + return fake + + +# -- _style_attrs (fake curses module, no real terminal) --------------------------------------- + + +def test_style_attrs_calls_use_default_colors_before_init_pair( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Regression guard: `curses.wrapper` never calls `use_default_colors()`, so an `init_pair` + background of `-1` is illegal and raises `curses.error` -- caught, but silently dropping ALL + colour. `_style_attrs` must call `use_default_colors()` itself, before the first + `init_pair`.""" + fake_curses = _fake_curses() + monkeypatch.setitem(sys.modules, "curses", fake_curses) + + attrs = _style_attrs() + + assert "use_default_colors" in fake_curses.calls + assert fake_curses.calls.index("use_default_colors") < fake_curses.calls.index("init_pair:1") + assert attrs["alert"] & fake_curses.color_pair(1) + assert attrs["warn"] & fake_curses.color_pair(2) + assert attrs["ok"] & fake_curses.color_pair(3) + + +# -- _paint (fake stdscr, no real terminal) ----------------------------------------------------- + + +class _FakeStdscr: + def __init__(self, height: int, width: int) -> None: + self._height = height + self._width = width + self.calls: list[tuple[int, int, str, int]] = [] + + def getmaxyx(self) -> tuple[int, int]: + return (self._height, self._width) + + def addstr(self, y: int, x: int, text: str, attr: int = 0) -> None: + self.calls.append((y, x, text, attr)) + + def erase(self) -> None: + pass + + def refresh(self) -> None: + pass + + +def test_paint_does_not_raise_on_tiny_window() -> None: + lines = [ + ScreenLine("keel · paper mode", "heading"), + ScreenLine("kill_switch: clear", "ok"), + ScreenLine("autonomy: off", "muted"), + ScreenLine("open positions: none", "normal"), + ScreenLine("q quit · read-only (no broker)", "muted"), + ] + stdscr = _FakeStdscr(height=3, width=10) + _paint(stdscr, lines) # must not raise even though window is smaller than content + + +def test_paint_records_addstr_per_visible_line() -> None: + lines = [ + ScreenLine("line one", "normal"), + ScreenLine("line two", "ok"), + ] + stdscr = _FakeStdscr(height=24, width=80) + _paint(stdscr, lines) + assert len(stdscr.calls) == 2 + ys = [call[0] for call in stdscr.calls] + assert ys == [0, 1] + + +def test_paint_truncates_to_window_width() -> None: + lines = [ScreenLine("x" * 200, "normal")] + stdscr = _FakeStdscr(height=24, width=20) + _paint(stdscr, lines) + assert len(stdscr.calls) == 1 + text = stdscr.calls[0][2] + assert len(text) <= 20 + + +def test_paint_applies_distinct_attrs_by_style() -> None: + lines = [ + ScreenLine("alert line", "alert"), + ScreenLine("normal line", "normal"), + ] + stdscr = _FakeStdscr(height=24, width=80) + _paint(stdscr, lines) + attrs = [call[3] for call in stdscr.calls] + # Not asserting exact bit values (curses colour init may be unavailable off a real terminal) + # -- just that the two differently-styled lines don't collapse to the same attr. + assert attrs[0] != attrs[1] + + +# -- run_once ------------------------------------------------------------------------------- + + +def test_run_once_captures_full_frame(repo: Repository) -> None: + repo.set_state("drawdown_total_pct", Decimal("0.05")) + repo.set_state("drawdown_weekly_pct", Decimal("0.01")) + config = _config() + + echoed: list[str] = [] + + def open_state() -> tuple[Repository, Any]: + return repo, config + + run_once(open_state, lambda: NOW_TS, echoed.append) + + assert echoed # a full frame was produced + joined = "\n".join(echoed) + assert "paper mode" in joined + assert "read-only" in joined.lower() + + +# -- run_live (fake curses module, no real terminal) --------------------------------------------- + + +class _ScriptedStdscr(_FakeStdscr): + """Like `_FakeStdscr`, but `getch()` returns `-1` (no key) until `quit_after` polls have + happened, then returns `q` so the loop under test terminates deterministically.""" + + def __init__(self, height: int, width: int, quit_after: int) -> None: + super().__init__(height, width) + self._quit_after = quit_after + self._polls = 0 + + def timeout(self, ms: int) -> None: + pass + + def getch(self) -> int: + self._polls += 1 + return ord("q") if self._polls >= self._quit_after else -1 + + +def test_run_live_survives_transient_read_error_and_keeps_polling( + repo: Repository, monkeypatch: pytest.MonkeyPatch +) -> None: + config = _config() + stdscr = _ScriptedStdscr(height=24, width=80, quit_after=2) + + fake_curses = _fake_curses() + fake_curses.wrapper = lambda fn: fn(stdscr) + monkeypatch.setitem(sys.modules, "curses", fake_curses) + + opens: list[int] = [] + + def open_state() -> tuple[Repository, Any]: + opens.append(1) + if len(opens) == 1: + raise sqlite3.OperationalError("database is locked") + return repo, config + + run_live(open_state, lambda: NOW_TS, interval=0.01) + + # The first poll's read error didn't kill the loop -- a second poll (the happy path) ran too. + assert len(opens) >= 2 + painted_texts = [call[2] for call in stdscr.calls] + assert any("status read failed" in t for t in painted_texts) + assert any("paper" in t for t in painted_texts) + + +def test_run_live_read_error_does_not_swallow_keyboard_interrupt( + repo: Repository, monkeypatch: pytest.MonkeyPatch +) -> None: + """`except Exception`, not `except BaseException` -- Ctrl-C during the per-poll read must + still propagate out to `run_live`'s own `try/except KeyboardInterrupt`, which swallows it.""" + stdscr = _ScriptedStdscr(height=24, width=80, quit_after=100) + + fake_curses = _fake_curses() + fake_curses.wrapper = lambda fn: fn(stdscr) + monkeypatch.setitem(sys.modules, "curses", fake_curses) + + def open_state() -> tuple[Repository, Any]: + raise KeyboardInterrupt + + run_live(open_state, lambda: NOW_TS, interval=0.01) # must not raise + + +# -- CLI ---------------------------------------------------------------------------------------- + + +def _repo_at(db_path) -> Repository: + conn = connect(str(db_path)) + migrate(conn) + return Repository(conn) + + +def test_tui_once_command_exits_zero_and_prints_frame(tmp_path, valid_config_path) -> None: + db_path = tmp_path / "keel.db" + _repo_at(db_path).set_state("kill_switch", False) + + result = CliRunner().invoke( + cli, ["--db", str(db_path), "--config", str(valid_config_path), "tui", "--once"] + ) + + assert result.exit_code == 0, result.output + assert "paper mode" in result.output + assert "read-only" in result.output.lower() + + +def test_tui_zero_interval_without_once_is_rejected(tmp_path, valid_config_path) -> None: + db_path = tmp_path / "keel.db" + _repo_at(db_path).set_state("kill_switch", False) + + result = CliRunner().invoke( + cli, + [ + "--db", + str(db_path), + "--config", + str(valid_config_path), + "tui", + "--interval", + "0", + "--once", + ], + ) + + assert result.exit_code != 0 + + +def test_tui_negative_interval_is_rejected(tmp_path, valid_config_path) -> None: + db_path = tmp_path / "keel.db" + _repo_at(db_path).set_state("kill_switch", False) + + result = CliRunner().invoke( + cli, + [ + "--db", + str(db_path), + "--config", + str(valid_config_path), + "tui", + "--interval", + "-1", + "--once", + ], + ) + + assert result.exit_code != 0