"""Run one live cycle: connect, snapshot portfolio, risk gates, strategy plugins.""" from __future__ import annotations import json import uuid from datetime import date, datetime from pathlib import Path from typing import Any from zoneinfo import ZoneInfo from RenTech.live.entry_ledger import ( acquire_run_lock, record_entry_submitted, release_run_lock, touch_run_ledger, ) from RenTech.live.ibkr_session import connect_ib, disconnect_ib, ensure_positions_loaded, ping_account from RenTech.live.platform_config import PlatformConfig, StrategySlot, load_platform_config from RenTech.live.portfolio import fetch_portfolio_snapshot from RenTech.live.protocols import StrategyContext, StrategyCycleReport from RenTech.live.registry import load_all_enabled from RenTech.live.risk_gate import evaluate_platform_risk, evaluate_strategy_risk, strategy_capital_budget NY = ZoneInfo("America/New_York") _REPO = Path(__file__).resolve().parents[2] async def run_platform_cycle( cfg: PlatformConfig, *, recommend_only: bool | None = None, force_entry_now: bool = False, entry_after_et: str | None = None, audit_only: bool = False, strategy_filter: str | None = None, ) -> dict[str, Any]: rec_only = cfg.risk.recommend_only if recommend_only is None else bool(recommend_only) run_id = uuid.uuid4().hex[:12] today = date.today() now_et = datetime.now(NY) cfg.log_dir.mkdir(parents=True, exist_ok=True) cfg.state_dir.mkdir(parents=True, exist_ok=True) print("=" * 72) print(f" Live platform [{cfg.name}] run_id={run_id}") print(f" Config: {cfg.path}") print(f" Mode: {'AUDIT' if audit_only else ('RECOMMEND' if rec_only else 'LIVE')}") print("=" * 72) ib = await connect_ib(cfg.broker) summary: dict[str, Any] = { "run_id": run_id, "platform": cfg.name, "as_of": now_et.isoformat(), "recommend_only": rec_only, "audit_only": audit_only, "strategies": [], } try: print(f"[IB] Connected {cfg.broker.host}:{cfg.broker.port} clientId={cfg.broker.client_id}") if not await ping_account(ib): raise RuntimeError("IB account summary ping failed") await ensure_positions_loaded(ib) portfolio = await fetch_portfolio_snapshot(ib) print( f"[PORTFOLIO] NetLiq=${portfolio.net_liquidation_usd:,.2f} " f"Avail=${portfolio.available_funds_usd:,.2f} " f"MarginUtil={portfolio.margin_utilization:.1%} " f"Positions={len(portfolio.positions)} " f"OptLegs={len(portfolio.open_option_legs())}" ) platform_risk = evaluate_platform_risk( cfg, portfolio, today, recommend_only=rec_only, ib=ib ) print(f"[RISK] platform action={platform_risk.action.value} reasons={platform_risk.reasons or ['ok']}") slots = cfg.enabled_strategies() if strategy_filter: slots = tuple(s for s in slots if s.id == strategy_filter) if not slots: raise ValueError(f"No enabled strategy matching filter {strategy_filter!r}") loaded = load_all_enabled(slots) if not loaded: print("[WARN] No enabled strategies in config") summary["platform_risk"] = { "action": platform_risk.action.value, "reasons": platform_risk.reasons, } return summary global_entry_et = entry_after_et or "15:40" for slot, strategy in loaded: budget = strategy_capital_budget(slot, portfolio) entry_after = str(slot.config.get("entry_after_et", global_entry_et)) strat_risk = evaluate_strategy_risk( slot, portfolio, platform_risk, capital_budget_usd=budget, state_dir=cfg.state_dir, today=today, now_et=now_et, entry_after_et=entry_after, force_entry_now=force_entry_now, ) ctx = StrategyContext( run_id=run_id, today=today, now_et=now_et, ib=ib, portfolio=portfolio, platform_risk=platform_risk, strategy_risk=strat_risk, recommend_only=rec_only, force_entry_now=force_entry_now, entry_after_et=entry_after, entry_before_et=slot.entry_before_et, strategy_id=slot.id, strategy_config=dict(slot.config), capital_budget_usd=budget, repo_root=_REPO, log_dir=cfg.log_dir, state_dir=cfg.state_dir, max_new_entries_per_day=slot.max_new_entries_per_day, max_open_positions=slot.max_open_positions, enabled_for_entries=slot.enabled_for_entries, ) print( f"\n--- Strategy {slot.id} (priority={slot.priority}) " f"budget=${budget:,.0f} risk={strat_risk.action.value} ---" ) acquire_run_lock(cfg.state_dir, slot.id, run_id) touch_run_ledger(cfg.state_dir, slot.id, run_id, today) report: StrategyCycleReport try: if audit_only: report = await strategy.audit(ctx) else: report = await strategy.run_cycle(ctx) except Exception as e: report = StrategyCycleReport( strategy_id=slot.id, ok=False, messages=[f"{type(e).__name__}: {e}"], ) print(f"[ERROR] {slot.id}: {e}") finally: release_run_lock(cfg.state_dir, slot.id, run_id) if ( not rec_only and report.entries_attempted > 0 and not report.metadata.get("entry_ledger_recorded") ): for _ in range(report.entries_attempted): record_entry_submitted( cfg.state_dir, slot.id, run_id, today=today, source="orchestrator", ) summary["strategies"].append( { "id": slot.id, "ok": report.ok, "entries": report.entries_attempted, "exits": report.exits_attempted, "messages": report.messages, "metadata": report.metadata, } ) summary["platform_risk"] = { "action": platform_risk.action.value, "reasons": platform_risk.reasons, "net_liq_usd": portfolio.net_liquidation_usd, "margin_utilization": portfolio.margin_utilization, } out_path = cfg.log_dir / f"platform_run_{run_id}.json" tmp = out_path.with_suffix(".tmp") tmp.write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8") tmp.replace(out_path) print(f"\n[LOG] Wrote {out_path}") return summary finally: await disconnect_ib(ib) print("[IB] Disconnected.") def run_from_config_path( config_path: str | Path, **kwargs: Any, ) -> dict[str, Any]: from ib_insync import util cfg = load_platform_config(config_path) return util.run(run_platform_cycle(cfg, **kwargs))