diff --git a/orchestration/__init__.py b/orchestration/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/orchestration/budget.py b/orchestration/budget.py new file mode 100644 index 0000000..7e89166 --- /dev/null +++ b/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() diff --git a/orchestration/checkpoint.py b/orchestration/checkpoint.py new file mode 100644 index 0000000..9a1b6e7 --- /dev/null +++ b/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//.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)) diff --git a/orchestration/run_nightly.py b/orchestration/run_nightly.py new file mode 100644 index 0000000..11c5a75 --- /dev/null +++ b/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//. 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))