"""Load and validate live platform JSON configuration.""" from __future__ import annotations import json import math from dataclasses import dataclass from pathlib import Path from typing import Any _REPO = Path(__file__).resolve().parents[2] DEFAULT_PLATFORM_CONFIG = _REPO / "RenTech" / "live" / "config" / "live_platform_default.json" @dataclass(frozen=True) class BrokerConfig: adapter: str host: str port: int client_id: int connect_timeout_sec: float market_data_type: int @dataclass(frozen=True) class RiskConfig: max_daily_loss_usd: float | None max_daily_loss_pct_nav: float | None min_net_liq_usd: float max_margin_utilization: float halt_on_kill_switch: bool recommend_only: bool max_drawdown_pct_from_peak: float | None min_excess_liquidity_usd: float | None block_if_pending_orders: bool @dataclass(frozen=True) class StrategySlot: id: str enabled: bool module: str priority: int capital_budget_mode: str capital_budget_nav_pct: float | None capital_budget_usd: float | None max_new_entries_per_day: int max_open_positions: int entry_before_et: str | None enabled_for_entries: bool config: dict[str, Any] @dataclass(frozen=True) class PlatformConfig: path: Path name: str log_dir: Path kill_switch_path: Path state_dir: Path broker: BrokerConfig risk: RiskConfig strategies: tuple[StrategySlot, ...] def enabled_strategies(self) -> tuple[StrategySlot, ...]: return tuple(s for s in self.strategies if s.enabled) def _req_float(d: dict[str, Any], key: str, *, lo: float | None = None, hi: float | None = None) -> float: v = float(d[key]) if not math.isfinite(v): raise ValueError(f"{key} must be finite") if lo is not None and v < lo: raise ValueError(f"{key} must be >= {lo}") if hi is not None and v > hi: raise ValueError(f"{key} must be <= {hi}") return v def load_platform_config(path: str | Path | None = None) -> PlatformConfig: p = Path(path).expanduser().resolve() if path is not None else DEFAULT_PLATFORM_CONFIG if not p.is_file(): raise FileNotFoundError(f"Platform config not found: {p}") raw = json.loads(p.read_text(encoding="utf-8")) if not isinstance(raw, dict): raise ValueError("Platform config must be a JSON object") plat = raw.get("platform", {}) if not isinstance(plat, dict): raise ValueError("'platform' must be an object") name = str(plat.get("name", "live_platform")) log_dir = (_REPO / str(plat.get("log_dir", "RenTech/data/logs/live_platform"))).resolve() kill_switch = (_REPO / str(plat.get("kill_switch_path", "RenTech/live/config/KILL_SWITCH"))).resolve() state_dir = (_REPO / str(plat.get("state_dir", "RenTech/data/live_state"))).resolve() br = raw.get("broker", {}) if not isinstance(br, dict): raise ValueError("'broker' must be an object") broker = BrokerConfig( adapter=str(br.get("adapter", "ibkr")).lower(), host=str(br.get("host", "127.0.0.1")), port=int(br.get("port", 7497)), client_id=int(br.get("client_id", 2)), connect_timeout_sec=float(br.get("connect_timeout_sec", 15)), market_data_type=int(br.get("market_data_type", 3)), ) if broker.adapter != "ibkr": raise ValueError(f"Only broker adapter 'ibkr' is implemented; got {broker.adapter!r}") rk = raw.get("risk", {}) if not isinstance(rk, dict): raise ValueError("'risk' must be an object") mdl = rk.get("max_daily_loss_usd") mdlp = rk.get("max_daily_loss_pct_nav") mdd_peak = rk.get("max_drawdown_pct_from_peak") min_excess = rk.get("min_excess_liquidity_usd") risk = RiskConfig( max_daily_loss_usd=float(mdl) if mdl is not None else None, max_daily_loss_pct_nav=float(mdlp) if mdlp is not None else None, min_net_liq_usd=_req_float(rk, "min_net_liq_usd", lo=0), max_margin_utilization=_req_float(rk, "max_margin_utilization", lo=0, hi=1.5), halt_on_kill_switch=bool(rk.get("halt_on_kill_switch", True)), recommend_only=bool(rk.get("recommend_only", False)), max_drawdown_pct_from_peak=float(mdd_peak) if mdd_peak is not None else None, min_excess_liquidity_usd=float(min_excess) if min_excess is not None else None, block_if_pending_orders=bool(rk.get("block_if_pending_orders", True)), ) slots_raw = raw.get("strategies", []) if not isinstance(slots_raw, list): raise ValueError("'strategies' must be a list") slots: list[StrategySlot] = [] seen: set[str] = set() for i, row in enumerate(slots_raw): if not isinstance(row, dict): raise ValueError(f"strategies[{i}] must be an object") sid = str(row.get("id", "")).strip() if not sid: raise ValueError(f"strategies[{i}] missing id") if sid in seen: raise ValueError(f"Duplicate strategy id: {sid}") seen.add(sid) cfg = row.get("config", {}) if cfg is not None and not isinstance(cfg, dict): raise ValueError(f"strategies[{i}].config must be an object") mode = str(row.get("capital_budget_mode", "nav_pct")).lower() nav_pct = row.get("capital_budget_nav_pct") cap_usd = row.get("capital_budget_usd") entry_before = row.get("entry_before_et") slots.append( StrategySlot( id=sid, enabled=bool(row.get("enabled", True)), module=str(row.get("module", "")).strip(), priority=int(row.get("priority", 100)), capital_budget_mode=mode, capital_budget_nav_pct=float(nav_pct) if nav_pct is not None else None, capital_budget_usd=float(cap_usd) if cap_usd is not None else None, max_new_entries_per_day=int(row.get("max_new_entries_per_day", 1)), max_open_positions=int(row.get("max_open_positions", 1)), entry_before_et=str(entry_before).strip() if entry_before else None, enabled_for_entries=bool(row.get("enabled_for_entries", True)), config=dict(cfg or {}), ) ) for s in slots: if not s.module: raise ValueError(f"Strategy {s.id!r} missing module (e.g. RenTech.live.strategies.vrp_ibkr:VrpIbkrStrategy)") slots.sort(key=lambda x: x.priority) return PlatformConfig( path=p, name=name, log_dir=log_dir, kill_switch_path=kill_switch, state_dir=state_dir, broker=broker, risk=risk, strategies=tuple(slots), )