""" VRP 4-regime book — live adapter over ``live_ibkr_trader.py``. Keeps execution parity with the monolithic script while fitting the platform orchestrator (shared IB session, portfolio snapshot, risk gates). """ from __future__ import annotations from pathlib import Path from typing import Any from RenTech.live.entry_ledger import ( check_and_reserve_entry, record_entry_submitted, release_entry_reservation, ) from RenTech.live.protocols import RiskAction, StrategyContext, StrategyCycleReport from RenTech.live.risk_gate import count_spy_option_positions from RenTech.strategy_stack.vrp_strategy_config import ( DEFAULT_STRATEGY_CONFIG_PATH, apply_strategy_params_to_vrp_backtester_module, load_strategy_config_file, sync_live_constants_from_vrp_backtester_module, ) import RenTech.strategy_stack.live_ibkr_trader as vrp class VrpIbkrStrategy: """Daily VRP options on SPY via IB (R1–R4 regimes).""" strategy_id = "vrp_core" def _resolve_paths(self, ctx: StrategyContext) -> tuple[Path, Path, Path]: state = Path( str(ctx.strategy_config.get("state_path", "RenTech/data/live_state/vrp_core.json")) ) if not state.is_absolute(): state = (ctx.repo_root / state).resolve() rec = Path( str( ctx.strategy_config.get( "recommend_out", "RenTech/data/logs/ibkr_trade_recommendation.json", ) ) ) if not rec.is_absolute(): rec = (ctx.repo_root / rec).resolve() cfg = Path( str(ctx.strategy_config.get("strategy_config_path", DEFAULT_STRATEGY_CONFIG_PATH)) ) if not cfg.is_absolute(): cfg = (ctx.repo_root / cfg).resolve() return state, rec, cfg def _apply_parity(self, cfg_path: Path) -> None: if cfg_path.is_file(): vrp._PARITY_CFG = load_strategy_config_file(cfg_path) # noqa: SLF001 apply_strategy_params_to_vrp_backtester_module(vrp._PARITY_CFG.strategy_params) # noqa: SLF001 sync_live_constants_from_vrp_backtester_module(vrp) print(f"[vrp_core][PARITY] {cfg_path}") else: vrp._PARITY_CFG = None # noqa: SLF001 print(f"[vrp_core][PARITY] missing {cfg_path} — built-in constants") async def _place_entry_leg( self, ctx: StrategyContext, state: dict[str, Any], regime: str, net: float, spy: Any, spy_live: float, *, recommend_out: Path, max_entry_qty: int | None, rec_only: bool, ignore_state_check: bool = False, report: StrategyCycleReport, ) -> bool: """Reserve ledger slot, place entry, record on success. Returns True if attempted.""" if rec_only: await vrp.place_entry_with_gtc( ctx.ib, state, regime, net, spy, spy_live, max_entry_qty=max_entry_qty, recommend_only=True, recommendation_out=recommend_out, ignore_state_check=ignore_state_check, ) report.entries_attempted += 1 return True allowed, reason = check_and_reserve_entry( ctx.state_dir, ctx.strategy_id, ctx.run_id, ctx.max_new_entries_per_day, today=ctx.today, ) if not allowed: report.messages.append(f"entry ledger: {reason}") return False n_before = len(state.get("strategies") or []) try: await vrp.place_entry_with_gtc( ctx.ib, state, regime, net, spy, spy_live, max_entry_qty=max_entry_qty, recommend_only=False, recommendation_out=recommend_out, ignore_state_check=ignore_state_check, ) except Exception: release_entry_reservation(ctx.state_dir, ctx.strategy_id, today=ctx.today) raise n_after = len(state.get("strategies") or []) if n_after <= n_before: release_entry_reservation(ctx.state_dir, ctx.strategy_id, today=ctx.today) report.messages.append(f"place_entry_with_gtc did not open {regime} (no state change)") return False record_entry_submitted( ctx.state_dir, ctx.strategy_id, ctx.run_id, today=ctx.today, regime=regime, source="vrp_ibkr", ) report.metadata["entry_ledger_recorded"] = True report.entries_attempted += 1 return True async def audit(self, ctx: StrategyContext) -> StrategyCycleReport: state_path, _, cfg_path = self._resolve_paths(ctx) self._apply_parity(cfg_path) code = await vrp.run_ib_state_audit(ctx.ib, state_path=state_path) return StrategyCycleReport( strategy_id=self.strategy_id, ok=code == 0, messages=[f"audit exit code {code}"], metadata={"state_path": str(state_path), "audit_code": code}, ) async def run_cycle(self, ctx: StrategyContext) -> StrategyCycleReport: state_path, recommend_out, cfg_path = self._resolve_paths(ctx) self._apply_parity(cfg_path) vrp.STATE_PATH = state_path # noqa: SLF001 report = StrategyCycleReport(strategy_id=self.strategy_id, ok=True) max_qty = ctx.strategy_config.get("max_entry_qty") max_entry_qty = int(max_qty) if max_qty is not None else None state = vrp._load_state_path(state_path) # noqa: SLF001 today = ctx.today rec_only = ctx.recommend_only if not rec_only and ctx.platform_risk.allow_exits: await vrp.manage_open_positions(ctx.ib, state, today) report.exits_attempted += 1 elif rec_only: report.messages.append("recommend_only: skipped exit management") from ib_insync import Index, Stock spy = Stock("SPY", "SMART", "USD") await ctx.ib.qualifyContractsAsync(spy) vix = Index("VIX", "CBOE", "USD") try: await ctx.ib.qualifyContractsAsync(vix) except Exception: vix = Index("VIX", exchange="CBOE", currency="USD") await ctx.ib.qualifyContractsAsync(vix) net = ctx.portfolio.net_liquidation_usd sma200, spy_hist = await vrp.get_spy_sma200(ctx.ib, spy) spy_live = await vrp.snapshot_price(ctx.ib, spy, "SPY", fallback=spy_hist) vix_live = await vrp.snapshot_price(ctx.ib, vix, "VIX") hh, mm = [int(x) for x in str(ctx.entry_after_et).split(":", 1)] gate = ctx.now_et.replace(hour=hh, minute=mm, second=0, microsecond=0) entry_time_ok = ctx.force_entry_now or ctx.now_et >= gate spy_bull = spy_live >= sma200 spy_opt_legs = count_spy_option_positions(ctx.portfolio) report.metadata.update( { "net_liq_usd": net, "spy_live": spy_live, "sma200": sma200, "vix_live": vix_live, "spy_gt_sma200": spy_bull, "entry_time_gate_pass": entry_time_ok, "capital_budget_usd": ctx.capital_budget_usd, "spy_option_legs_ib": spy_opt_legs, "max_open_positions": ctx.max_open_positions, } ) vrp._write_live_signal_state( # noqa: SLF001 { "platform_run_id": ctx.run_id, "strategy_id": self.strategy_id, "created_at": ctx.now_et.isoformat(), "recommend_only": rec_only, "net_liq_usd": net, "spy_live": spy_live, "sma200": sma200, "vix_live": vix_live, "regime_now": vrp.regime_from_vix(vix_live), "spy_gt_sma200": spy_bull, "entry_time_gate_pass": entry_time_ok, "state_path": str(state_path), } ) if not spy_bull: report.messages.append("SPY < SMA200 — cash/wait, no new entries") return report if not entry_time_ok: report.messages.append(f"before entry gate {ctx.entry_after_et} ET") return report if ctx.strategy_risk.action == RiskAction.HALT_ALL: report.messages.append("platform HALT — no entries") report.ok = False return report if not ctx.strategy_risk.allow_new_entries: report.messages.append(f"entries blocked: {ctx.strategy_risk.reasons}") return report if not ctx.enabled_for_entries: report.messages.append("enabled_for_entries=false") return report if spy_opt_legs >= ctx.max_open_positions and not state.get("strategies"): msg = ( f"IB has {spy_opt_legs} SPY option leg(s) but state is empty — " "reconciliation gap; block entry" ) print(f"[vrp_core][WARN] {msg}") report.messages.append(msg) return report if state.get("strategies") and not rec_only: report.messages.append("open positions in state — skip duplicate entry") return report reg = vrp.regime_from_vix(vix_live) report.metadata["regime"] = reg if reg == "diagonal": base = recommend_out rec_a = base.with_name(base.stem + "_r2a_diagonal" + base.suffix) rec_b = base.with_name(base.stem + "_r2b_spread" + base.suffix) await self._place_entry_leg( ctx, state, "diagonal", net, spy, spy_live, recommend_out=rec_a if rec_only else recommend_out, max_entry_qty=max_entry_qty, rec_only=rec_only, report=report, ) await self._place_entry_leg( ctx, state, "r2_spread", net, spy, spy_live, recommend_out=rec_b if rec_only else recommend_out, max_entry_qty=max_entry_qty, rec_only=rec_only, ignore_state_check=True, report=report, ) else: await self._place_entry_leg( ctx, state, reg, net, spy, spy_live, recommend_out=recommend_out, max_entry_qty=max_entry_qty, rec_only=rec_only, report=report, ) vrp._save_state(state) # noqa: SLF001 report.messages.append(f"regime={reg} entries_attempted={report.entries_attempted}") return report