feat: nightly event retention + external telemetry push

- runtime/retention.py: prune 8 high-volume debug event types older than
  RETENTION_DAYS (default 14) once/day at RETENTION_HOUR (default 03:00).
  Audit events (orders, fills, closes, risk rules) are never pruned. The
  events table hit 6 GB in 3 weeks at ~340k rows/day; this caps it.
- runtime/telemetry_push.py: POST a JSON snapshot (equity, day realized
  P&L, trades/wins, open positions, last-event age) to TELEMETRY_PUSH_URL
  every TELEMETRY_PUSH_INTERVAL_SECONDS (default 300). Unset URL = step
  disabled. Push failures are reported in the step result, never raised —
  an external outage can't touch the loop. Works as healthchecks.io
  dead-man ping + payload log in one.
- wired both as scheduler steps in run_live_paper (retention gated by
  RETENTION_ENABLED, telemetry by TELEMETRY_PUSH_URL presence)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

run_live_paper.py

@@ -1184,6 +1184,16 @@ def main(argv: list[str] | None = None) -> int:
         n = notifier.poll()
         return f"telegram: {n} alert(s)" if n else None
 
+    def step_retention():
+        from runtime.retention import maybe_prune
+        out = maybe_prune(rt["store"])
+        return None if out in ("idle", "done-today") else out
+
+    def step_telemetry():
+        from runtime.telemetry_push import push_snapshot
+        return push_snapshot(
+            rt["store"], broker=os.environ.get("BROKER", "alpaca").strip().lower())
+
     def step_manage_exits():
         if exit_mgr is None:
             return None
@@ -1368,6 +1378,15 @@ def main(argv: list[str] | None = None) -> int:
     scheduler.add("catchup", step_open_catchup_flatten,
                   float(os.environ.get("TRADING_OPEN_CATCHUP_INTERVAL_SECONDS", "30")),
                   enabled=_flag("TRADING_EXECUTION_ENABLED", "1"))
+    # nightly event-store prune (runtime/retention.py): cheap hourly check,
+    # actual DELETE once/day at RETENTION_HOUR
+    scheduler.add("retention", step_retention, 1800.0,
+                  enabled=_flag("RETENTION_ENABLED", "1"))
+    # external metrics push (runtime/telemetry_push.py): off unless
+    # TELEMETRY_PUSH_URL is set
+    scheduler.add("telemetry", step_telemetry,
+                  float(os.environ.get("TELEMETRY_PUSH_INTERVAL_SECONDS", "300")),
+                  enabled=bool(os.environ.get("TELEMETRY_PUSH_URL", "").strip()))
 
     def run_pass():
         # force every task due, in pipeline order

runtime/retention.py

@@ -0,0 +1,60 @@
+"""Event-store retention: prune high-volume debug event types.
+
+The events table grows ~340k rows/day and ~95% of it is per-tick diagnostic
+spam (criteria scores, heartbeats, account snapshots). Left alone it reached
+6 GB in three weeks and unindexed scans took minutes. Orders, fills, closes,
+and risk events are the audit trail — never pruned.
+"""
+
+from datetime import datetime, timedelta
+import os
+
+# High-volume diagnostic types safe to expire. Anything NOT listed is kept
+# forever (order_*, position_*, risk_rule_triggered, session_summary, ...).
+PRUNE_EVENT_TYPES = (
+    "criteria_evaluated",
+    "signal_blocked",
+    "module_tick",
+    "entry_gate_evaluated",
+    "account_summary_updated",
+    "account_positions_updated",
+    "account_orders_updated",
+    "symbol_evaluation_detail",
+)
+
+
+def prune_events(store, days: int = 14, event_types=PRUNE_EVENT_TYPES,
+                 now: datetime | None = None) -> dict:
+    """Delete prunable events older than ``days``. Idempotent."""
+    cutoff = (now or datetime.now()) - timedelta(days=days)
+    placeholders = ",".join("?" for _ in event_types)
+    with store._lock:
+        res = store.con.execute(
+            f"DELETE FROM events WHERE event_type IN ({placeholders})"
+            f" AND timestamp < ?",
+            [*event_types, cutoff],
+        )
+        store.con.commit()
+    deleted = getattr(getattr(res, "_cur", None), "rowcount", -1)
+    return {"deleted": deleted, "cutoff": cutoff.isoformat(timespec="seconds")}
+
+
+_last_run: str | None = None  # date of the last prune (process-local; restart re-runs, idempotent)
+
+
+def maybe_prune(store, now: datetime | None = None) -> str:
+    """Scheduler step: run prune_events once per day in the configured hour
+    (default 03:00 container-local — market quiet, premarket discovery not yet
+    running). Safe to call every scheduler pass."""
+    global _last_run
+    now = now or datetime.now()
+    hour = int(os.environ.get("RETENTION_HOUR", "3"))
+    days = int(os.environ.get("RETENTION_DAYS", "14"))
+    today = now.date().isoformat()
+    if now.hour != hour:
+        return "idle"
+    if _last_run == today:
+        return "done-today"
+    result = prune_events(store, days=days, now=now)
+    _last_run = today
+    return f"pruned {result['deleted']} rows older than {days}d"

runtime/telemetry_push.py

@@ -0,0 +1,82 @@
+"""Push a loop-health + day-P&L snapshot to an external webhook.
+
+Point TELEMETRY_PUSH_URL at anything that accepts a POST body — a
+healthchecks.io ping URL (gives dead-man alerting + a log of every payload),
+a Grafana Cloud / Loki push proxy, or any collector. Unset = disabled. A
+push failure never breaks the loop; the failure is the scheduler step's
+return string so it shows in the loop console.
+"""
+
+import json
+import os
+import urllib.request
+from datetime import datetime
+
+
+def _payload_of(event: dict) -> dict:
+    try:
+        return json.loads(event.get("payload_json") or "{}")
+    except (ValueError, TypeError):
+        return {}
+
+
+def build_payload(store, broker: str = "unknown",
+                  now: datetime | None = None) -> dict:
+    now = now or datetime.now()
+    today = now.date().isoformat()
+    out = {
+        "ts": now.isoformat(timespec="seconds"),
+        "broker": broker,
+        "equity": None,
+        "day_realized": 0.0,
+        "day_trades": 0,
+        "day_wins": 0,
+        "open_positions": None,
+        "last_event_age_s": None,
+    }
+    # newest account snapshot today (query_events is ASC — take the last row)
+    acct = store.query_events(event_type="account_summary_updated",
+                              since=today, limit=5000)
+    if acct:
+        out["equity"] = _payload_of(acct[-1]).get("total_equity")
+    pos = store.query_events(event_type="account_positions_updated",
+                             since=today, limit=5000)
+    if pos:
+        held = _payload_of(pos[-1]).get("positions")
+        out["open_positions"] = len(held) if isinstance(held, list) else None
+    for e in store.query_events(event_type="position_closed",
+                                since=today, limit=500):
+        pnl = float(_payload_of(e).get("realized_pnl") or 0.0)
+        out["day_realized"] = round(out["day_realized"] + pnl, 2)
+        out["day_trades"] += 1
+        out["day_wins"] += 1 if pnl > 0 else 0
+    latest = store.max_timestamp()
+    if latest is not None:
+        try:
+            out["last_event_age_s"] = int((now - latest).total_seconds())
+        except TypeError:  # tz-aware vs naive mismatch — skip rather than crash
+            pass
+    return out
+
+
+def push(url: str, payload: dict, timeout: float = 5.0) -> None:
+    req = urllib.request.Request(
+        url,
+        data=json.dumps(payload).encode(),
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    urllib.request.urlopen(req, timeout=timeout).read()
+
+
+def push_snapshot(store, broker: str = "unknown") -> str:
+    """Scheduler step: build + POST one snapshot. Never raises."""
+    url = os.environ.get("TELEMETRY_PUSH_URL", "").strip()
+    if not url:
+        return "disabled"
+    payload = build_payload(store, broker=broker)
+    try:
+        push(url, payload)
+    except Exception as exc:  # noqa: BLE001 — external outage must not kill the loop
+        return f"push failed: {exc}"
+    return f"pushed pnl={payload['day_realized']:+.0f} trades={payload['day_trades']}"

tests/unit/test_retention_and_telemetry.py

@@ -0,0 +1,79 @@
+"""Retention prune + telemetry push snapshot."""
+
+from datetime import datetime, timedelta
+
+from storage.event_store import EventStore
+from storage.event_schema import (
+    EventMode,
+    ModuleTickEvent,
+    PositionClosedEvent,
+)
+from runtime.retention import prune_events, maybe_prune
+from runtime.telemetry_push import build_payload, push_snapshot
+
+
+def _tick(ts):
+    return ModuleTickEvent(
+        timestamp=ts, mode=EventMode.PAPER, correlation_id="t",
+        message="tick", module="triggers", stage="completed",
+        duration_ms=1.0, input_count=0, output_count=0)
+
+
+def _close(ts, pnl):
+    return PositionClosedEvent(
+        timestamp=ts, mode=EventMode.PAPER, correlation_id="t",
+        message="closed", position_id="p1", symbol="AAA",
+        exit_reason="stop_loss", entry_price=1.0,
+        exit_price=1.0 + pnl, realized_pnl=pnl,
+        payload={"symbol": "AAA", "realized_pnl": pnl})
+
+
+def test_prune_deletes_old_spam_keeps_audit_and_recent():
+    store = EventStore(":memory:")
+    now = datetime.now()
+    old = now - timedelta(days=30)
+    store.emit(_tick(old))                # old spam -> pruned
+    store.emit(_tick(now))                # recent spam -> kept
+    store.emit(_close(old, 5.0))          # old AUDIT event -> kept forever
+    r = prune_events(store, days=14, now=now)
+    assert r["deleted"] == 1
+    assert len(store.query_events(event_type="module_tick", limit=10)) == 1
+    assert len(store.query_events(event_type="position_closed", limit=10)) == 1
+
+
+def test_maybe_prune_only_in_window_and_once_per_day():
+    import runtime.retention as retention
+    retention._last_run = None
+    store = EventStore(":memory:")
+    at_2am = datetime.now().replace(hour=2)
+    at_3am = datetime.now().replace(hour=3)
+    assert maybe_prune(store, now=at_2am) == "idle"
+    assert maybe_prune(store, now=at_3am).startswith("pruned")
+    assert maybe_prune(store, now=at_3am) == "done-today"
+
+
+def test_build_payload_day_stats():
+    store = EventStore(":memory:")
+    now = datetime.now()
+    store.emit(_close(now, 100.0))
+    store.emit(_close(now, -40.0))
+    store.emit(_close(now - timedelta(days=2), 999.0))  # not today -> excluded
+    p = build_payload(store, broker="tradier", now=now)
+    assert p["broker"] == "tradier"
+    assert p["day_trades"] == 2
+    assert p["day_wins"] == 1
+    assert p["day_realized"] == 60.0
+    assert p["last_event_age_s"] is not None
+
+
+def test_push_snapshot_disabled_without_url(monkeypatch):
+    monkeypatch.delenv("TELEMETRY_PUSH_URL", raising=False)
+    store = EventStore(":memory:")
+    assert push_snapshot(store) == "disabled"
+
+
+def test_push_snapshot_survives_unreachable_url(monkeypatch):
+    monkeypatch.setenv("TELEMETRY_PUSH_URL", "http://127.0.0.1:1/nope")
+    store = EventStore(":memory:")
+    out = push_snapshot(store)
+    assert out.startswith("push failed")