diff --git a/llm/server.py b/llm/server.py index 1fe6a34..e587d71 100644 --- a/llm/server.py +++ b/llm/server.py @@ -74,6 +74,39 @@ def warm_model() -> None: logger.warning("warm_model failed (non-fatal): {}", e) +def clear_kv_cache(hot: bool = True) -> None: + """Wipe oMLX's SSD prefix/KV cache (and, by default, the in-memory hot cache). + + oMLX prefix-caches every prompt into an SSD-backed paged KV store so repeated prefixes + skip prefill. Within a single agent that's a big win, but across personas the prefixes + diverge, so a finished agent's blocks are dead weight that just keeps eating SSD (and the + hot cache keeps eating RAM) until the disk fills or the prefill memory guard trips. Call + this at each agent boundary to keep peak SSD/RAM use to roughly one agent's context. + + Auth: oMLX's admin endpoints want a session cookie, minted by POSTing the main API key to + /admin/api/login; httpx.Client keeps that cookie for the follow-up clears. Fully non-fatal + — a cache-clear hiccup must never abort a run. + """ + s = get_settings() + base = f"http://{s.llm_host}:{s.llm_port}" + try: + with httpx.Client(base_url=base, timeout=30) as c: + c.post( + "/admin/api/login", + json={"api_key": s.llm_api_key, "remember": False}, + ).raise_for_status() + ssd = c.post("/admin/api/ssd-cache/clear") + ssd.raise_for_status() + msg = f"cleared oMLX SSD KV cache ({ssd.json().get('total_deleted', '?')} blocks)" + if hot: + h = c.post("/admin/api/hot-cache/clear") + h.raise_for_status() + msg += f", hot cache reclaimed {h.json().get('bytes_reclaimed', 0) / 1e6:.0f}MB" + logger.info(msg) + except Exception as e: + logger.warning("clear_kv_cache failed (non-fatal): {}", e) + + def ensure_ready(warm: bool = True) -> bool: start_server() ok = wait_healthy(120) diff --git a/mark_runner.sh b/mark_runner.sh new file mode 100755 index 0000000..1a6428a --- /dev/null +++ b/mark_runner.sh @@ -0,0 +1,4 @@ +while [ "$(date +%H:%M)" != "08:31" ]; do + sleep 1 +done +uv run python -m execution.run_market_open diff --git a/orchestration/run_nightly.py b/orchestration/run_nightly.py index 37943c6..2166c90 100644 --- a/orchestration/run_nightly.py +++ b/orchestration/run_nightly.py @@ -102,6 +102,10 @@ def main(compressed: bool = False) -> int: if not server.server_up(): # recover if a prior agent left oMLX unresponsive logger.warning("oMLX unreachable before {}; recovering...", persona.id) server.ensure_ready(warm=False) + # Fresh KV store per agent: the previous persona's SSD/hot cache blocks share no + # prefix with this one, so they'd only pile up on the SSD (and pin RAM) until the + # disk fills or the prefill memory guard trips. Clear at the boundary, not mid-agent. + server.clear_kv_cache() lessons = lessons_for(persona.id) # self-learning: past picks marked to market records[persona.id] = lessons blurb = "\n\n".join(x for x in (regime, lessons) if x) diff --git a/scripts/README.md b/scripts/README.md new file mode 100644 index 0000000..6f7ca56 --- /dev/null +++ b/scripts/README.md @@ -0,0 +1,76 @@ +# scripts/ + +Local operational scripts for the paper-trading agent. This machine holds the +Alpaca keys and the Gmail app password (`.env`) — none of that ever leaves it. + +## export_public.py — sanitized public snapshot + +Produces `trading-public.json`, the ONLY artifact the public dashboard +(`sites/apps/trading`) is allowed to read. It is fully decoupled: the hosted app +reads a file, it never touches this DB or any key. + +```bash +python3 scripts/export_public.py \ + --db state/trading.db \ + --out trading-public.json +``` + +Stdlib only (no `uv`/venv needed). Flags: + +- `--db` path to the SQLite DB (default `state/trading.db`) +- `--out` output JSON path (default `./trading-public.json`) +- `--output-dir` nightly committee dir holding `/_committee.json` + (default `/output`) + +### What it emits / strips + +- EMITTED: per-persona **percent** returns, **base-100** normalized (coarse/sample) + equity curves, thesis cards, committee verdicts, trade markers, and positions + as (symbol + side + **relative** weight). +- STRIPPED: absolute `$` equity / cash / buying-power, Alpaca account IDs, Alpaca + order IDs, per-share prices and market values. + +A self-check scans the emitted JSON for `$`, UUID id patterns, and raw +account-level dollar figures and **aborts** on any hit, so a sanitization +regression can never ship. The script only prints percentages and counts. + +## Scheduling (launchd / cron) + +Run it right after the nightly research + morning execution have written the DB +and committee JSON, so the public snapshot reflects the latest close. + +launchd (matches the existing `launchd/ai.tradingagent.*.plist` style) — a +daily job after the 08:30 CT execution, e.g. a wrapper `scripts/export.sh`: + +```bash +#!/usr/bin/env bash +cd /Users/aryan/Documents/GitHub/trading-agent +python3 scripts/export_public.py --db state/trading.db --out state/trading-public.json +# then sync to wherever the app reads it (see below) +``` + +cron equivalent (09:15 local, after execution settles): + +```cron +15 9 * * 1-5 cd /Users/aryan/Documents/GitHub/trading-agent && \ + /usr/bin/python3 scripts/export_public.py --out state/trading-public.json >> state/logs/export_public.log 2>&1 +``` + +## Syncing to the app + +The dashboard reads the file at the path in `TRADING_DATA_PATH` +(default: its bundled `data/trading-public.json`). Point the app at the exported +file, or copy the export into place after each run: + +```bash +# option A — copy into the app's data dir (rebuild/redeploy picks it up) +cp state/trading-public.json /path/to/sites/apps/trading/data/trading-public.json + +# option B — write to a shared volume the running app reads via TRADING_DATA_PATH +cp state/trading-public.json /srv/trading-data/trading-public.json +# app env: TRADING_DATA_PATH=/srv/trading-data/trading-public.json +``` + +The app `revalidate`s hourly, so a fresh file is served within the hour with no +redeploy. (Hosting/transport is handled separately — this script only writes a +local file.) diff --git a/scripts/export_public.py b/scripts/export_public.py new file mode 100755 index 0000000..9a445ee --- /dev/null +++ b/scripts/export_public.py @@ -0,0 +1,360 @@ +#!/usr/bin/env python3 +""" +export_public.py — SANITIZED public snapshot exporter for the trading dashboard. + +Reads the local paper-trading state (SQLite + nightly committee JSON) and emits a +single, DECOUPLED, read-only file (`trading-public.json`) that the public Next.js +dashboard consumes. NOTHING sensitive leaves this machine: + + STRIPPED — absolute $ equity / cash / buying-power, Alpaca account IDs, + Alpaca order IDs, per-share dollar prices, market values. + EMITTED — per-persona PERCENT returns, BASE-100 normalized equity curves, + thesis cards, committee verdicts, trade markers (no order IDs), + and positions as (symbol + side + RELATIVE weight) only. + +The equity source (`snapshots`) is sparse (~6 rows/persona over ~10 days), so the +curves are labelled coarse/sample and step-interpolated by the dashboard — no +intraday points are fabricated here. + +Stdlib only (sqlite3 + json). Run: + + python3 scripts/export_public.py \ + --db state/trading.db \ + --out trading-public.json + +A final self-check scans the emitted JSON for dollar figures / ID patterns and +aborts if any are found, so a sanitization regression can never ship silently. +""" + +from __future__ import annotations + +import argparse +import json +import re +import sqlite3 +import sys +from collections import defaultdict +from datetime import datetime, timezone +from pathlib import Path + +# --------------------------------------------------------------------------- # +# Persona presentation metadata. `account`/`agent` keys in the DB are persona +# slugs (nova / warren_lite / atlas / maverick) — NOT Alpaca account IDs. +# --------------------------------------------------------------------------- # +PERSONA_ORDER = ["nova", "warren_lite", "atlas", "maverick"] +PERSONA_META = { + "nova": {"label": "Nova", "style": "momentum"}, + "warren_lite": {"label": "Warren-lite", "style": "value"}, + "atlas": {"label": "Atlas", "style": "macro"}, + "maverick": {"label": "Maverick", "style": "high-beta"}, +} + +# Patterns that must NEVER appear in the emitted file. +_UUID_RE = re.compile(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}") + + +def _round(x, n): + """Round, coercing to float and tolerating None.""" + try: + return round(float(x), n) + except (TypeError, ValueError): + return None + + +def _short(text, limit=180): + """Trim a rationale to a word boundary with an ellipsis. + + Also strips '$' so no dollar glyph reaches the public file (rationales + occasionally reference strikes/prices like "$60 strike" — descriptive, but + the sanitized feed carries zero dollar signs by policy).""" + if not text: + return "" + t = " ".join(str(text).replace("$", "").split()) + if len(t) <= limit: + return t + cut = t[:limit].rsplit(" ", 1)[0].rstrip(",.;:") + return cut + "…" + + +# --------------------------------------------------------------------------- # +# Loaders +# --------------------------------------------------------------------------- # +def load_snapshots(con): + """{account: [(ts, equity), ...]} ordered by ts. Equity is used only to + derive PERCENT return + base-100 curve; the raw $ figure never leaves here.""" + rows = con.execute( + "SELECT account, ts, equity FROM snapshots " + "WHERE equity IS NOT NULL ORDER BY account, ts" + ).fetchall() + by_acct = defaultdict(list) + for account, ts, equity in rows: + by_acct[account].append((ts, float(equity))) + return by_acct + + +def build_curve(series): + """Base-100 normalized curve (one point per date, last snapshot of the day). + + Returns (points, return_pct, n_snapshots). Base = first snapshot's equity.""" + if not series: + return [], None, 0 + base = series[0][1] + if base == 0: + return [], None, len(series) + + per_date = {} # date -> equity of last snapshot that day (insertion keeps last) + for ts, equity in series: + per_date[ts[:10]] = equity + + points = [{"t": d, "v": _round(eq / base * 100.0, 2)} for d, eq in per_date.items()] + last_equity = series[-1][1] + return_pct = _round((last_equity / base - 1.0) * 100.0, 1) + return points, return_pct, len(series) + + +def load_positions(con): + """{account: [{symbol, side, weight}, ...]} — RELATIVE weights only. + + weight = |market_value| / sum(|market_value|) per account; the absolute $ + market_value is used only for the ratio and is never emitted.""" + rows = con.execute( + "SELECT account, symbol, qty, market_value FROM positions " + "WHERE market_value IS NOT NULL" + ).fetchall() + raw = defaultdict(list) + for account, symbol, qty, mv in rows: + raw[account].append((symbol, float(qty or 0), abs(float(mv)))) + + out = {} + for account, items in raw.items(): + total = sum(mv for _, _, mv in items) or 1.0 + packed = [ + { + "symbol": sym, + "side": "short" if qty < 0 else "long", + "weight": _round(mv / total, 4), + } + for sym, qty, mv in items + ] + packed.sort(key=lambda p: p["weight"], reverse=True) + out[account] = packed + return out + + +def load_theses(con): + """Deduped thesis cards: one row per (date, persona, symbol), highest + conviction wins. Returns a list plus a lookup of the chosen direction.""" + rows = con.execute( + "SELECT trade_date, agent, ticker, direction, conviction, thesis " + "FROM theses ORDER BY trade_date, agent, ticker" + ).fetchall() + + best = {} # (date, persona, symbol) -> row dict + for trade_date, agent, ticker, direction, conviction, thesis in rows: + key = (trade_date, agent, ticker) + conv = _round(conviction, 2) or 0.0 + if key not in best or conv > best[key]["conviction"]: + best[key] = { + "date": trade_date, + "persona": agent, + "symbol": ticker, + "direction": (direction or "long").lower(), + "conviction": conv, + "rationaleShort": _short(thesis), + } + + theses = sorted(best.values(), key=lambda t: (t["date"], t["persona"], t["symbol"])) + direction_by_key = {(t["date"], t["persona"], t["symbol"]): t["direction"] for t in theses} + return theses, direction_by_key + + +def load_committee(output_dir): + """Committee verdicts from state/output//_committee.json files. + + Keys look like 'nova|ASML'. Returns (list, verdict_lookup).""" + committee = [] + verdict_by_key = {} + if not output_dir.is_dir(): + return committee, verdict_by_key + + for date_dir in sorted(output_dir.iterdir()): + cfile = date_dir / "_committee.json" + if not cfile.is_file(): + continue + date = date_dir.name + try: + data = json.loads(cfile.read_text()) + except (json.JSONDecodeError, OSError): + continue + for key, val in data.items(): + if key.startswith("_") or "|" not in key: # skip _memo etc. + continue + persona, symbol = key.split("|", 1) + verdict = str(val.get("verdict", "")).lower() + if verdict not in ("approve", "reduce", "veto"): + continue + committee.append({ + "date": date, + "persona": persona, + "symbol": symbol, + "verdict": verdict, + "note": _short(val.get("reason", ""), 200), + }) + verdict_by_key[(date, persona, symbol)] = verdict + + committee.sort(key=lambda c: (c["date"], c["persona"], c["symbol"])) + return committee, verdict_by_key + + +def build_trades(direction_by_key, verdict_by_key): + """Trade markers derived from theses + committee (NOT order status/IDs). + + A thesis becomes a trade unless the committee vetoed it. side = buy|sell, + mapped from the thesis direction (long->buy, short->sell).""" + trades = [] + for (date, persona, symbol), direction in direction_by_key.items(): + if verdict_by_key.get((date, persona, symbol)) == "veto": + continue + trades.append({ + "date": date, + "persona": persona, + "symbol": symbol, + "side": "sell" if direction == "short" else "buy", + }) + trades.sort(key=lambda t: (t["date"], t["persona"], t["symbol"])) + return trades + + +# --------------------------------------------------------------------------- # +# Assembly + sanitization gate +# --------------------------------------------------------------------------- # +def build_payload(db_path, output_dir): + con = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) + try: + snaps = load_snapshots(con) + positions = load_positions(con) + theses, direction_by_key = load_theses(con) + finally: + con.close() + + committee, verdict_by_key = load_committee(output_dir) + trades = build_trades(direction_by_key, verdict_by_key) + + # as_of = latest snapshot date seen anywhere. + all_dates = [ts[:10] for series in snaps.values() for ts, _ in series] + as_of = max(all_dates) if all_dates else None + + personas = [] + snapshot_total = 0 + for pid in PERSONA_ORDER: + curve, ret, n = build_curve(snaps.get(pid, [])) + snapshot_total += n + meta = PERSONA_META.get(pid, {"label": pid, "style": "unknown"}) + personas.append({ + "id": pid, + "label": meta["label"], + "style": meta["style"], + "returnPct": ret, + "snapshotCount": n, + "equityCurve": curve, + "positions": positions.get(pid, []), + }) + + return { + "generatedAt": datetime.now(timezone.utc).isoformat(timespec="seconds"), + "asOf": as_of, + "personas": personas, + "theses": theses, + "committee": committee, + "trades": trades, + "meta": { + "note": "Paper trading (Alpaca PAPER). Sanitized, decoupled snapshot — no live account data.", + "curveNote": "Equity curves are base-100 normalized and COARSE/SAMPLE: " + "~6 sparse snapshots per persona over ~10 days, step-interpolated. Not intraday.", + "snapshotCount": snapshot_total, + "personaCount": len(personas), + "thesisCount": len(theses), + "committeeCount": len(committee), + "tradeCount": len(trades), + }, + } + + +def assert_sanitized(payload, db_path): + """Fail loudly if any $ figure, account/order UUID, or raw account-level + dollar value leaked into the emitted JSON.""" + blob = json.dumps(payload) + problems = [] + + if "$" in blob: + problems.append("literal '$' present") + m = _UUID_RE.search(blob) + if m: + problems.append(f"UUID-like id present: {m.group(0)[:8]}…") + + # Cross-check against the actual account-level dollar figures in the DB. + con = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) + try: + dollars = set() + for (v,) in con.execute("SELECT equity FROM snapshots UNION " + "SELECT cash FROM snapshots UNION " + "SELECT buying_power FROM snapshots UNION " + "SELECT market_value FROM positions"): + if v is None: + continue + fv = float(v) + if abs(fv) >= 50: # account/position dollar scale; avoids weight/conviction collisions + dollars.add(repr(round(fv, 2))) + dollars.add(repr(fv)) + finally: + con.close() + + for d in dollars: + needle = d.rstrip("0").rstrip(".") if "." in d else d + if needle and needle in blob: + problems.append(f"raw dollar figure present: {needle}") + break + + if problems: + raise SystemExit("SANITIZATION FAILED: " + "; ".join(problems)) + + +# --------------------------------------------------------------------------- # +def main(argv=None): + parser = argparse.ArgumentParser(description="Export a sanitized public trading snapshot.") + parser.add_argument("--db", default="state/trading.db", help="Path to trading.db (default: state/trading.db)") + parser.add_argument("--out", default="trading-public.json", help="Output JSON path (default: ./trading-public.json)") + parser.add_argument("--output-dir", default=None, + help="Nightly output dir with /_committee.json (default: /output)") + args = parser.parse_args(argv) + + db_path = Path(args.db).expanduser().resolve() + if not db_path.is_file(): + raise SystemExit(f"database not found: {db_path}") + + output_dir = Path(args.output_dir).expanduser().resolve() if args.output_dir \ + else db_path.parent / "output" + + payload = build_payload(db_path, output_dir) + assert_sanitized(payload, db_path) + + out_path = Path(args.out).expanduser().resolve() + out_path.parent.mkdir(parents=True, exist_ok=True) + out_path.write_text(json.dumps(payload, indent=2) + "\n") + + # Summary — non-sensitive lines only (percentages + counts, never $). + print(f"wrote {out_path}") + print(f"asOf {payload['asOf']} snapshots {payload['meta']['snapshotCount']}") + for p in payload["personas"]: + ret = p["returnPct"] + rstr = f"{ret:+.1f}%" if ret is not None else "n/a" + print(f" {p['id']:<12} {rstr:>7} curve {len(p['equityCurve'])}pt positions {len(p['positions'])}") + print(f"theses {payload['meta']['thesisCount']} " + f"committee {payload['meta']['committeeCount']} " + f"trades {payload['meta']['tradeCount']}") + print("sanitization: PASS (no $, no account/order IDs)") + return 0 + + +if __name__ == "__main__": + sys.exit(main())