| """Processes: every worker thread with what it does, its state, start/stop, autostart and recent activity; the event log. |
| |
| The workers are threads inside the console process (ops/supervisor.py). Every control here goes through ``runtime.sup()``, |
| so an observer console (``fly-trader ui --observe``) refuses them all; the guards the page shows by disabling a button |
| (already running, running outside the console, already stopping) are enforced again in the actions.""" |
| from __future__ import annotations |
|
|
| import json |
| import threading |
| from contextlib import nullcontext |
|
|
| from ... import config |
| from ...db.apilog import record_event |
| from ...db.connection import transaction |
| from ...db.queries import q, q1 |
| from ...logging_setup import scrub |
| from ...ops.supervisor import WORKERS |
| from .. import runtime |
| from ..hub import ActionError, TopicError, action, topic |
| from ..state import ROLE_LABEL, WORKER_INFO, WORKER_TYPE, jv, worker_order |
|
|
| GROUPS = ("memecoins", "prediction markets", "system") |
| LEVELS = ("info", "warning", "error") |
| ENGINES = ("runner", "kalshi_runner") |
| STOP_NOTE = { |
| "runner": "While it is stopped nothing is bought or sold: open positions stay open until it runs again.", |
| "pumpstream": "Without it the Solana feed goes stale within 3 minutes and the trading engine holds: no Solana entries or exits until it runs again.", |
| "rh_stream": "Without it the Robinhood Chain feed goes stale within 3 minutes and the engine's Robinhood books hold: no entries or exits there until it runs again.", |
| "kalshi_runner": "No new Kalshi entries while it is stopped; open positions still settle when their markets close.", |
| "kalshi_stream": "Without it the Kalshi feed goes stale and the Kalshi engine holds: no entries until it runs again.", |
| "vault": "The vault's jobs (flows, claims, the trading float, NAV, weekly settlement, public stats) pause until it runs again.", |
| } |
| _AUTOSTART = threading.Lock() |
|
|
|
|
| def _title(name: str) -> str: |
| return WORKER_INFO.get(name, (name,))[0] |
|
|
|
|
| def _state(s: dict, external: bool = False) -> str: |
| """running / stopping / crashed / stopped for a console thread; ``external`` when only a CLI process runs the worker.""" |
| if s.get("alive"): |
| return "stopping" if s.get("stopping") else "running" |
| return "external" if external else "crashed" if s.get("error") else "stopped" |
|
|
|
|
| def _start_note(name: str) -> str | None: |
| """What starting this worker sets off that the operator should know first (resets, real money).""" |
| notes = [] |
| if name == "runner": |
| if config.RESET_ON_START: |
| notes.append("RESET_ON_START is on: it first archives and clears the books other than the race books and the live books.") |
| if config.LIVE_ENABLED: |
| notes.append("LIVE_ENABLED=1: once the fly holds the selector's seat it also trades the bot wallet with real SOL.") |
| if config.RH_ENABLED and config.RH_LIVE_ENABLED: |
| notes.append("RH_LIVE_ENABLED=1: once the RH fly holds the seat it also trades the Robinhood Chain wallet with real ETH.") |
| elif name == "kalshi_runner" and config.KALSHI_LIVE_ENABLED and not config.kalshi_live_prerequisites_missing(): |
| notes.append(f"Kalshi live is on: both arms mirror onto subaccount {config.KALSHI_SUBACCOUNT} with real money (cap ${config.KALSHI_CAPITAL_USD:g}).") |
| return " ".join(notes) or None |
|
|
|
|
| def _host_pid(sup) -> int | None: |
| """The pid of the console that hosts the workers (an observer's own pid is not it).""" |
| if not isinstance(sup, runtime.Observer): |
| return sup.pid |
| r = q1("SELECT pid FROM processes WHERE cmd[1] = 'thread' AND stopped_at IS NULL ORDER BY started_at DESC LIMIT 1") |
| return int(r["pid"]) if r else None |
|
|
|
|
| def _hosting(sup) -> None: |
| """Worker controls belong to the console that hosts the workers (procs.stop and the autostart row do not go through the |
| Supervisor, so they would slip past the observer's refusal).""" |
| if isinstance(sup, runtime.Observer): |
| raise runtime.ObserverMode("this console only observes (fly-trader ui --observe): start and stop workers from the console that hosts them") |
|
|
|
|
| def _worker(body: dict, shown: bool = True) -> str: |
| """The worker an action names: one of the cards on the page (``shown``), or any worker the Supervisor knows.""" |
| name = body.get("name") |
| if not isinstance(name, str) or name not in (worker_order() if shown else WORKERS): |
| raise ActionError(f"no worker {name!r} on this page", 400) |
| return name |
|
|
|
|
| |
| @topic("processes.workers", every=3) |
| def workers(params: dict) -> dict: |
| """One card per worker (grouped as the menu is), the external CLI processes, and which console hosts the threads.""" |
| sup = runtime.sup(); status = sup.status(); ext = sup.external() |
| auto = jv((q1("SELECT value FROM ui_settings WHERE key = 'autostart'") or {}).get("value")) |
| cards = [] |
| for name in worker_order(): |
| title, icon, desc, role = WORKER_INFO[name]; s = status.get(name) or {} |
| err = (s.get("error") or "").splitlines() |
| cards.append({"name": name, "title": title, "icon": icon, "desc": desc, "role": role, "role_label": ROLE_LABEL[role], |
| "group": WORKER_TYPE.get(name, "system"), "state": _state(s, name in ext), "alive": bool(s.get("alive")), "stopping": bool(s.get("stopping")), |
| "started_at": s.get("started_at"), "stopped_at": s.get("stopped_at"), |
| "error": err[0][:300] if err and not s.get("alive") else None, "external": name in ext, "autostart": bool(auto.get(name)), |
| "ext_pid": (ext.get(name) or {}).get("pid"), "ext_started_at": (ext.get(name) or {}).get("started_at"), |
| "start_note": _start_note(name), "stop_note": STOP_NOTE.get(name)}) |
| groups = [{"key": g, "title": g.capitalize(), "workers": [c for c in cards if c["group"] == g]} for g in GROUPS] |
| counts = {k: sum(c["state"] == k for c in cards) for k in ("running", "stopping", "crashed", "stopped", "external")} |
| return {"pid": sup.pid, "observer": isinstance(sup, runtime.Observer), "host_pid": _host_pid(sup), |
| "external": [{"name": n, "title": _title(n), "pid": a.get("pid"), "started_at": a.get("started_at")} for n, a in ext.items()], |
| "groups": [g for g in groups if g["workers"]], |
| "counts": {**counts, "total": len(cards), "autostart": sum(c["autostart"] for c in cards), |
| "trade_needed": sum(c["role"] == "trade" for c in cards), "trade_up": sum(c["role"] == "trade" and c["state"] in ("running", "external") for c in cards)}} |
|
|
|
|
| @topic("processes.activity", every=3) |
| def activity(params: dict) -> dict: |
| """A worker's last 12 log lines, newest first (the in-memory ring of this console, or logs/<name>.log for an observer).""" |
| name = params.get("worker") |
| if name not in WORKERS: |
| raise TopicError(f"no worker {name!r}") |
| rows = [] |
| for ln in [ln for ln in runtime.sup().log_tail(name, 200).splitlines() if ln.strip()][-12:]: |
| try: |
| d = json.loads(ln) |
| except ValueError: |
| d = None |
| if isinstance(d, dict): |
| rows.append({"time": str(d.get("ts", ""))[11:19], "level": str(d.get("level", "")), "what": scrub(str(d.get("msg", "")))[:300]}) |
| else: |
| rows.append({"time": "", "level": "", "what": scrub(ln)[:300]}) |
| return {"name": name, "rows": rows[::-1]} |
|
|
|
|
| @topic("processes.events", every=10) |
| def events(params: dict) -> dict: |
| """The last 200 events at the chosen levels (none chosen: errors).""" |
| lv = params.get("levels") |
| levels = [x for x in LEVELS if isinstance(lv, list) and x in lv] or ["error"] |
| rows = q("SELECT ts, level, source, message FROM events WHERE level = ANY(%s) ORDER BY id DESC LIMIT 200", (levels,)) |
| return {"levels": levels, "rows": rows} |
|
|
|
|
| @topic("processes.event_stats", every=30) |
| def event_stats(params: dict) -> dict: |
| """The last 24 hours of events: how many per level, and warnings and errors per hour (24 buckets, oldest first).""" |
| by_level = {r["level"]: int(r["n"]) for r in q("SELECT level, count(*) AS n FROM events WHERE ts > now() - interval '24 hours' GROUP BY level")} |
| rows = q("""SELECT h.h, COALESCE(sum(CASE WHEN e.level = 'warning' THEN 1 END), 0) AS warning, COALESCE(sum(CASE WHEN e.level = 'error' THEN 1 END), 0) AS error |
| FROM generate_series(date_trunc('hour', now()) - interval '23 hours', date_trunc('hour', now()), interval '1 hour') AS h(h) |
| LEFT JOIN events e ON e.ts >= h.h AND e.ts < h.h + interval '1 hour' AND e.level IN ('warning', 'error') |
| GROUP BY h.h ORDER BY h.h""") |
| return {"levels": {lv: by_level.get(lv, 0) for lv in LEVELS} | {k: v for k, v in by_level.items() if k not in LEVELS}, |
| "hours": [r["h"] for r in rows], "warning": [int(r["warning"]) for r in rows], "error": [int(r["error"]) for r in rows]} |
|
|
|
|
| |
| @action("processes.start") |
| def start(body: dict) -> dict: |
| name = _worker(body); sup = runtime.sup(); _hosting(sup); title = _title(name) |
| if sup.alive(name): |
| raise ActionError(f"{title} is already running.") |
| ext = sup.external() |
| if name in ext: |
| raise ActionError(f"{title} is running outside the console (pid {ext[name].get('pid')}): stop the external process first.") |
| from . import safety |
| with safety.no_reset() if name in ENGINES else nullcontext(): |
| try: |
| started = sup.start(name, started_by="console") |
| except RuntimeError as e: |
| raise ActionError(str(e)) from None |
| if not started: |
| raise ActionError(f"{title} is already running.") |
| return {"message": f"Started {title}"} |
|
|
|
|
| @action("processes.stop") |
| def stop(body: dict) -> dict: |
| name = _worker(body, shown=False); sup = runtime.sup(); _hosting(sup); title = _title(name) |
| s = sup.status().get(name) or {} |
| if not s.get("alive"): |
| raise ActionError(f"{title} is not running.") |
| if s.get("stopping"): |
| raise ActionError(f"{title} is already stopping: it finishes its current step first.") |
| if not sup.request_stop(name, by="console"): |
| raise ActionError(f"{title} is not running.") |
| return {"message": f"Stop requested: {title} finishes its current step, then stops."} |
|
|
|
|
| @action("processes.stop_external") |
| def stop_external(body: dict) -> dict: |
| """A worker started from the command line (`fly-trader worker start`): SIGTERM, 30 s, then SIGKILL (ops/procs.stop).""" |
| name = _worker(body, shown=False); sup = runtime.sup(); _hosting(sup) |
| a = sup.external().get(name) |
| if not a: |
| raise ActionError(f"No external {name} process is running.") |
| from ...ops import procs |
| if not procs.stop(name): |
| return {"message": f"The external {name} process had already exited"} |
| return {"message": f"Stopped the external {name} process (pid {a.get('pid')})"} |
|
|
|
|
| @action("processes.autostart") |
| def autostart(body: dict) -> dict: |
| """Whether a worker starts with the console: the whole ``autostart`` map is read, changed and written back.""" |
| name = _worker(body); on = body.get("on") |
| if not isinstance(on, bool): |
| raise ActionError("on must be true or false", 400) |
| _hosting(runtime.sup()) |
| with _AUTOSTART: |
| r = q1("SELECT value FROM ui_settings WHERE key = 'autostart'"); val = jv(r["value"]) if r else {} |
| val[name] = on |
| with transaction() as conn: |
| conn.execute("INSERT INTO ui_settings (key, value) VALUES ('autostart', %s) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now()", |
| (json.dumps(val),)) |
| record_event("info", "console", "autostart changed", val) |
| return {"message": f"Autostart {'on' if on else 'off'} for {_title(name)}"} |
|
|
