feat(orchestration): nightly runner, rollover budgets, incremental checkpoints

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016DvdRZkyCy9NJpHYDdPqmS

orchestration/budget.py

@@ -0,0 +1,43 @@
+"""Time budgeting for the nightly run.
+
+Each agent gets a wall-clock deadline = its fair share of the remaining window, capped by
+the per-agent config, and never past the hard stop. If an agent finishes early the freed
+time rolls over to those remaining (bounded by the cap), and the hard stop always holds.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+from zoneinfo import ZoneInfo
+
+
+def local_hard_stop_ts(hard_stop_hhmm: str, local_tz: str) -> float:
+    """Epoch seconds of the next HH:MM in local time (today if still ahead, else tomorrow)."""
+    tz = ZoneInfo(local_tz)
+    now = datetime.now(tz)
+    hh, mm = (int(x) for x in hard_stop_hhmm.split(":"))
+    target = now.replace(hour=hh, minute=mm, second=0, microsecond=0)
+    if target <= now:
+        target += timedelta(days=1)
+    return target.timestamp()
+
+
+def next_deadline(
+    now_ts: float, hard_stop_ts: float, agents_remaining: int, per_agent_cap_s: float
+) -> float:
+    """Deadline for the current agent: min(fair share, cap), never past hard stop."""
+    if agents_remaining <= 0 or now_ts >= hard_stop_ts:
+        return hard_stop_ts
+    fair_share = (hard_stop_ts - now_ts) / agents_remaining
+    span = min(fair_share, per_agent_cap_s) if per_agent_cap_s else fair_share
+    return now_ts + max(span, 60.0)
+
+
+def next_market_trade_date(market_tz: str) -> str:
+    """Best-effort next trading date (ET), rolling weekends to Monday. Holidays handled in
+    Phase 3 via Alpaca's calendar; this is only for naming the nightly output folder."""
+    tz = ZoneInfo(market_tz)
+    d = datetime.now(tz).date()
+    while d.weekday() >= 5:  # 5 = Sat, 6 = Sun
+        d += timedelta(days=1)
+    return d.isoformat()

orchestration/checkpoint.py

@@ -0,0 +1,32 @@
+"""Incremental, crash-safe JSON output. Each agent's output is (over)written atomically
+after every research round and again at finalize, so a hang or power loss mid-run still
+leaves the best-so-far result on disk under state/output/<trade_date>/<persona>.json.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+from agents.schemas import AgentOutput, RunManifest
+
+
+def _atomic_write(path: Path, text: str) -> None:
+    tmp = path.with_suffix(path.suffix + ".tmp")
+    tmp.write_text(text)
+    os.replace(tmp, path)  # atomic on the same filesystem
+
+
+class Checkpointer:
+    def __init__(self, output_dir: Path, trade_date: str):
+        self.dir = Path(output_dir) / trade_date
+        self.dir.mkdir(parents=True, exist_ok=True)
+
+    def agent_path(self, persona_id: str) -> Path:
+        return self.dir / f"{persona_id}.json"
+
+    def write(self, output: AgentOutput) -> None:
+        _atomic_write(self.agent_path(output.persona_id), output.model_dump_json(indent=2))
+
+    def write_manifest(self, manifest: RunManifest) -> None:
+        _atomic_write(self.dir / "_manifest.json", manifest.model_dump_json(indent=2))

orchestration/run_nightly.py

@@ -0,0 +1,86 @@
+"""ENTRYPOINT #1 — the nightly research run.
+
+Brings up oMLX, then runs the 3 personas strictly sequentially (one model, time-sliced),
+each under a rollover wall-clock budget bounded by the hard stop. Every agent's output is
+checkpointed to state/output/<trade_date>/. Run with --compressed for a fast end-to-end
+test (short budgets).
+
+    uv run python -m orchestration.run_nightly [--compressed]
+"""
+
+from __future__ import annotations
+
+import sys
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from loguru import logger  # noqa: E402
+
+from agents.graph import run_agent  # noqa: E402
+from agents.schemas import RunManifest  # noqa: E402
+from config.personas import PERSONAS  # noqa: E402
+from config.settings import get_settings  # noqa: E402
+from llm import server  # noqa: E402
+from llm.client import resolve_model_id  # noqa: E402
+from orchestration.budget import (  # noqa: E402
+    local_hard_stop_ts,
+    next_deadline,
+    next_market_trade_date,
+)
+from orchestration.checkpoint import Checkpointer  # noqa: E402
+
+
+def main(compressed: bool = False) -> int:
+    s = get_settings()
+    s.ensure_dirs()
+    trade_date = next_market_trade_date(s.market_tz)
+    logger.add(s.log_dir / f"nightly_{trade_date}.log", rotation="10 MB", retention="14 days")
+    logger.info("=== nightly run trade_date={} profile={} compressed={} ===",
+                trade_date, s.model_profile, compressed)
+
+    if not server.ensure_ready():
+        logger.error("oMLX not ready; aborting.")
+        return 1
+
+    model_name = resolve_model_id()
+    run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+    cp = Checkpointer(s.output_dir, trade_date)
+    manifest = RunManifest(
+        run_id=run_id,
+        trade_date=trade_date,
+        model=model_name,
+        started_at=datetime.now(timezone.utc),
+        status="running",
+    )
+    cp.write_manifest(manifest)
+
+    hard_stop = local_hard_stop_ts(s.hard_stop, s.local_tz)
+    per_agent_cap_s = 5 * 60 if compressed else s.per_agent_wallclock_min * 60
+    step_budget = 12 if compressed else s.per_agent_step_budget
+    min_steps = 4 if compressed else s.per_agent_min_steps
+
+    for i, persona in enumerate(PERSONAS):
+        now = time.time()
+        if now >= hard_stop:
+            logger.warning("hard stop reached; skipping {} remaining agents", len(PERSONAS) - i)
+            break
+        deadline = next_deadline(now, hard_stop, len(PERSONAS) - i, per_agent_cap_s)
+        logger.info("--- {} until {} ---", persona.id,
+                    datetime.fromtimestamp(deadline).strftime("%H:%M:%S"))
+        out = run_agent(persona, deadline_ts=deadline, step_budget=step_budget,
+                        min_steps=min_steps, checkpointer=cp, model_name=model_name)
+        cp.write(out)
+        manifest.agents.append(persona.id)
+
+    manifest.status = "complete"
+    manifest.ended_at = datetime.now(timezone.utc)
+    cp.write_manifest(manifest)
+    logger.info("=== nightly run complete; output in {} ===", cp.dir)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(compressed="--compressed" in sys.argv))