diff --git a/run_live_paper.py b/run_live_paper.py index 6657ffb..c4a097b 100644 --- a/run_live_paper.py +++ b/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 diff --git a/runtime/retention.py b/runtime/retention.py new file mode 100644 index 0000000..5e83ddb --- /dev/null +++ b/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" diff --git a/runtime/telemetry_push.py b/runtime/telemetry_push.py new file mode 100644 index 0000000..874dbaa --- /dev/null +++ b/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']}" diff --git a/tests/unit/test_retention_and_telemetry.py b/tests/unit/test_retention_and_telemetry.py new file mode 100644 index 0000000..3a4b08a --- /dev/null +++ b/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")