""" Regime intersection matrix and action mapping (SPX Regime State Space v2). ``classify_regime`` maps **every** discretized 6-tuple to a regime + action (exhaustive partition; no ``none`` bucket). """ from __future__ import annotations import itertools from dataclasses import dataclass from typing import Literal from RenTech.strategy_stack.spx_regime_state_space.features import ( FeatureState, HurstState, MaDistState, SkewState, SpotVolState, TermState, VrpState, ) RegimeId = Literal[ # v2 core (widened triggers) "D_structural_damage", "C_crash_up", "A_mean_reversion", "A1_extended_above", "A2_extended_below", "B_momentum", "B1_bullish", "B2_bearish", # Spot-vol squeeze extensions "C1_squeeze_skewed", "C2_squeeze_building", "C3_vol_rising_neutral", "C4_vol_rising_bear", # Backwardation / stress "E1_backwardation_bear_mom", "E2_backwardation_bear_fear", "E3_backwardation_bull_trend", "E4_backwardation_extended", "E5_backwardation_high_vol", "E6_backwardation_chop", # High VRP harvest (partial A) "F1_high_vrp_flat_above", "F2_high_vrp_flat_below", "F3_high_vrp_flat_neutral", "F4_high_vrp_trend_above", "F5_high_vrp_trend_neutral", "F6_high_vrp_random", # Low VRP / choppy "G1_choppy_bull_low_vrp", "G2_choppy_bear_low_vrp", "G3_choppy_neutral_low_vrp", "G4_low_vrp_random", "B3_momentum_steep_skew", "B4_momentum_normal_skew", # Normal VRP catch-all "H1_range_bound", "H2_chop_normal_skew", "H3_trend_steep_skew", "H4_trend_normal_skew", "H5_random_steep_skew", "H6_random_flat_skew", "H7_random_normal_skew", ] ActionId = Literal[ "cash", "iron_condor_neutral", "iron_condor_neg_delta", "iron_condor_pos_delta", "put_credit_spread", "call_debit_spread", "put_debit_spread", "long_call_calendar", "long_straddle", ] VRP_STATES: tuple[VrpState, ...] = ("low", "normal", "high") HURST_STATES: tuple[HurstState, ...] = ("anti", "random", "persistent") SKEW_STATES: tuple[SkewState, ...] = ("flat", "normal", "steep") TERM_STATES: tuple[TermState, ...] = ("contango", "backwardation") SPOT_VOL_STATES: tuple[SpotVolState, ...] = ("negative", "positive") MA_STATES: tuple[MaDistState, ...] = ("below", "neutral", "above") @dataclass(frozen=True) class RegimeState: regime: RegimeId action: ActionId feature_state: FeatureState rationale: str def classify_regime(st: FeatureState) -> RegimeState: """Map any 6-tuple to regime + action (exhaustive tree).""" vrp, hurst, skew, term, spot_vol, ma = st # --- Tier 0: structural damage override --- if vrp == "high" and hurst == "persistent" and ma == "below": return _rs( st, "D_structural_damage", "cash", "High VRP + below MA + persistent trend → capital preservation", ) # --- Tier 1: positive spot-vol (squeeze / vol-rising) --- if spot_vol == "positive": if ma == "above": if hurst == "persistent" and skew == "flat": return _rs( st, "C_crash_up", "long_call_calendar", "Crash-up: spot-vol+ / flat skew / persistent / above MA", ) if hurst == "persistent": return _rs( st, "C1_squeeze_skewed", "call_debit_spread", "Vol squeeze with skew → ride upside", ) return _rs( st, "C2_squeeze_building", "long_call_calendar", "Spot-vol+ / above MA → calendar long vol", ) if ma == "neutral": return _rs( st, "C3_vol_rising_neutral", "long_straddle", "Spot-vol+ / neutral MA → long gamma", ) return _rs( st, "C4_vol_rising_bear", "put_debit_spread", "Spot-vol+ / below MA → defensive put debit", ) # --- Tier 2: backwardation (term stress) --- if term == "backwardation": if ma == "below": action = "put_debit_spread" if vrp == "low": return _rs( st, "E1_backwardation_bear_mom", action, "Backwardation + below MA + low VRP → bear momentum", ) return _rs( st, "E2_backwardation_bear_fear", action, "Backwardation + below MA → defensive bear", ) if ma == "above": if hurst == "persistent" and vrp == "low": return _rs( st, "E3_backwardation_bull_trend", "call_debit_spread", "Backwardation bull: low VRP + persistent + above MA", ) return _rs( st, "E4_backwardation_extended", "iron_condor_neg_delta", "Backwardation + extended above MA → harvest call skew", ) if vrp == "high": return _rs( st, "E5_backwardation_high_vol", "long_straddle", "Backwardation + high VRP + neutral MA → long vol", ) return _rs( st, "E6_backwardation_chop", "iron_condor_neutral", "Backwardation + neutral MA → neutral harvest", ) # --- Tier 3: contango + high VRP (vol harvest family) --- if vrp == "high": if hurst == "anti" and skew in ("steep", "normal"): return _mean_reversion_harvest(st, ma, skew) if hurst == "anti" and skew == "flat": if ma == "above": return _rs( st, "F1_high_vrp_flat_above", "iron_condor_neg_delta", "High VRP chop + flat skew + above MA", ) if ma == "below": return _rs( st, "F2_high_vrp_flat_below", "put_credit_spread", "High VRP chop + flat skew + below MA", ) return _rs( st, "F3_high_vrp_flat_neutral", "iron_condor_neutral", "High VRP chop + flat skew + neutral MA", ) if hurst == "persistent": if ma == "above": return _rs( st, "F4_high_vrp_trend_above", "iron_condor_neg_delta", "High VRP + persistent trend + above MA", ) return _rs( st, "F5_high_vrp_trend_neutral", "iron_condor_neutral", "High VRP + persistent + neutral/below MA", ) act, _ = _ic_by_ma(ma) return _rs( st, "F6_high_vrp_random", act, f"High VRP + random walk → harvest by MA ({ma})", ) # --- Tier 4: contango + low VRP (momentum / expansion) --- if vrp == "low": if hurst == "persistent": if skew == "flat": return _momentum_trend(st, ma) if skew == "steep": act, _ = _debit_by_ma(ma) return _rs( st, "B3_momentum_steep_skew", act, f"Low VRP momentum + steep skew → directional ({ma})", ) act, _ = _debit_by_ma(ma) return _rs( st, "B4_momentum_normal_skew", act, f"Low VRP momentum + normal skew → directional ({ma})", ) if hurst == "anti": if ma == "above": return _rs( st, "G1_choppy_bull_low_vrp", "call_debit_spread", "Low VRP + anti-persistent + above MA", ) if ma == "below": return _rs( st, "G2_choppy_bear_low_vrp", "put_debit_spread", "Low VRP + anti-persistent + below MA", ) return _rs( st, "G3_choppy_neutral_low_vrp", "long_straddle", "Low VRP + choppy + neutral MA → long gamma", ) act, _ = _debit_by_ma(ma) return _rs( st, "G4_low_vrp_random", act, f"Low VRP + random walk → directional ({ma})", ) # --- Tier 5: contango + normal VRP (catch-all) --- if hurst == "anti": if skew == "steep": return _mean_reversion_harvest(st, ma, skew) if skew == "flat": return _rs( st, "H1_range_bound", "iron_condor_neutral", "Normal VRP + choppy + flat skew → range harvest", ) act, _ = _ic_by_ma(ma) return _rs( st, "H2_chop_normal_skew", act, f"Normal VRP + choppy + normal skew ({ma})", ) if hurst == "persistent": if skew == "flat": return _momentum_trend(st, ma) if skew == "steep": act, _ = _debit_by_ma(ma) return _rs( st, "H3_trend_steep_skew", act, f"Normal VRP + trend + steep skew ({ma})", ) act, _ = _debit_by_ma(ma) return _rs( st, "H4_trend_normal_skew", act, f"Normal VRP + trend + normal skew ({ma})", ) if skew == "steep": act, _ = _ic_by_ma(ma) return _rs( st, "H5_random_steep_skew", act, f"Normal VRP + random + steep skew ({ma})", ) if skew == "flat": return _rs( st, "H6_random_flat_skew", "iron_condor_neutral", "Normal VRP + random + flat skew", ) act, _ = _ic_by_ma(ma) return _rs( st, "H7_random_normal_skew", act, f"Normal VRP + random + normal skew ({ma})", ) def map_action(regime: RegimeId) -> ActionId: """Lookup action for a regime id (first matching lattice point).""" for combo in itertools.product( VRP_STATES, HURST_STATES, SKEW_STATES, TERM_STATES, SPOT_VOL_STATES, MA_STATES, ): rs = classify_regime(combo) # type: ignore[arg-type] if rs.regime == regime: return rs.action return "iron_condor_neutral" def verify_full_coverage() -> tuple[int, list[FeatureState]]: """Return (n_states, duplicates) — duplicates should be empty; n_states == 324.""" seen: dict[FeatureState, RegimeId] = {} for combo in itertools.product( VRP_STATES, HURST_STATES, SKEW_STATES, TERM_STATES, SPOT_VOL_STATES, MA_STATES, ): st: FeatureState = combo # type: ignore[assignment] rs = classify_regime(st) if rs.action == "cash" and rs.regime != "D_structural_damage": raise AssertionError(f"unexpected cash for {st} -> {rs.regime}") seen[st] = rs.regime return len(seen), [] def regime_distribution_summary() -> dict[str, int]: """Count how many of 324 lattice points map to each regime.""" counts: dict[str, int] = {} for combo in itertools.product( VRP_STATES, HURST_STATES, SKEW_STATES, TERM_STATES, SPOT_VOL_STATES, MA_STATES, ): rs = classify_regime(combo) # type: ignore[arg-type] counts[rs.regime] = counts.get(rs.regime, 0) + 1 return counts def _rs( st: FeatureState, regime: RegimeId, action: ActionId, rationale: str, ) -> RegimeState: return RegimeState(regime=regime, action=action, feature_state=st, rationale=rationale) def _ic_by_ma(ma: MaDistState) -> tuple[ActionId, str]: if ma == "above": return "iron_condor_neg_delta", "skewed IC (neg delta)" if ma == "below": return "iron_condor_pos_delta", "skewed IC (pos delta)" return "iron_condor_neutral", "delta-neutral IC" def _debit_by_ma(ma: MaDistState) -> tuple[ActionId, str]: if ma == "above": return "call_debit_spread", "bull debit" if ma == "below": return "put_debit_spread", "bear debit" return "iron_condor_neutral", "neutral IC" def _mean_reversion_harvest( st: FeatureState, ma: MaDistState, skew: SkewState, ) -> RegimeState: skew_lbl = "steep" if skew == "steep" else "normal" if ma == "above": return _rs( st, "A1_extended_above", "iron_condor_neg_delta", f"Mean-reversion harvest ({skew_lbl} skew) + above MA", ) if ma == "below": return _rs( st, "A2_extended_below", "put_credit_spread", f"Mean-reversion harvest ({skew_lbl} skew) + below MA", ) return _rs( st, "A_mean_reversion", "iron_condor_neutral", f"Mean-reversion harvest ({skew_lbl} skew) + neutral MA", ) def _momentum_trend(st: FeatureState, ma: MaDistState) -> RegimeState: if ma == "above": return _rs( st, "B1_bullish", "call_debit_spread", "Momentum: low/normal VRP + persistent + flat skew + above MA", ) if ma == "below": return _rs( st, "B2_bearish", "put_debit_spread", "Momentum: low/normal VRP + persistent + flat skew + below MA", ) return _rs( st, "B_momentum", "iron_condor_neutral", "Momentum trend but neutral MA anchor", )