feat(regime): add ratio gate for soft rebalances

tests/test_config.py

@@ -46,6 +46,7 @@ class RegimeRebalanceConfigFactory(ModelFactory[RegimeRebalanceConfig]):
     flow_imbalance_tau = 0.70
     deficit_rail_start = 5000.0
     deficit_rail_stop = 2500.0
+    ratio_gate = None
 
 
 class ConfigFactory(ModelFactory[Config]):

tests/test_regime_rebalance.py

@@ -5,7 +5,7 @@
 from ib_async import IB, Stock
 
 import thetagang.portfolio_manager as pm_module
-from thetagang.config import RegimeRebalanceConfig, normalize_config
+from thetagang.config import RatioGateConfig, RegimeRebalanceConfig, normalize_config
 from thetagang.db import DataStore
 from thetagang.portfolio_manager import PortfolioManager
 
@@ -209,6 +209,100 @@ async def test_regime_rebalance_respects_regime_gate(portfolio_manager, mocker):
     assert orders == []
 
 
+@pytest.mark.asyncio
+async def test_regime_rebalance_ratio_gate_shadow_metrics_emitted(
+    portfolio_manager_with_db, mocker
+):
+    portfolio_manager_with_db.config.regime_rebalance.ratio_gate = SimpleNamespace(
+        enabled=False,
+        anchor="BBB",
+        drift_max=1.25,
+        var_min=0.0,
+    )
+    account_summary = {"NetLiquidation": SimpleNamespace(value="400")}
+    portfolio_positions = {
+        "AAA": [SimpleNamespace(contract=Stock("AAA", "SMART", "USD"), position=3)],
+        "BBB": [SimpleNamespace(contract=Stock("BBB", "SMART", "USD"), position=1)],
+    }
+
+    _mock_regime_tickers(portfolio_manager_with_db, mocker)
+    _mock_regime_history(
+        portfolio_manager_with_db, mocker, [100.0, 110.0, 100.0, 110.0]
+    )
+    portfolio_manager_with_db.ibkr.request_executions = mocker.AsyncMock(
+        return_value=[]
+    )
+
+    await portfolio_manager_with_db.check_regime_rebalance_positions(
+        account_summary, portfolio_positions
+    )
+
+    payload = portfolio_manager_with_db.data_store.get_last_event_payload(
+        "regime_rebalance_gate"
+    )
+    assert payload["ratio_gate"]["enabled"] is False
+    assert payload["ratio_gate"]["anchor"] == "BBB"
+    assert payload["ratio_gate"]["rest"] == ["AAA"]
+
+
+@pytest.mark.asyncio
+async def test_regime_rebalance_ratio_gate_blocks_soft_rebalance(
+    portfolio_manager, mocker
+):
+    portfolio_manager.config.regime_rebalance.choppiness_min = 0.0
+    portfolio_manager.config.regime_rebalance.efficiency_max = 1.0
+    portfolio_manager.config.regime_rebalance.ratio_gate = SimpleNamespace(
+        enabled=True,
+        anchor="BBB",
+        drift_max=1.25,
+        var_min=0.0,
+    )
+    account_summary = {"NetLiquidation": SimpleNamespace(value="400")}
+    portfolio_positions = {
+        "AAA": [SimpleNamespace(contract=Stock("AAA", "SMART", "USD"), position=3)],
+        "BBB": [SimpleNamespace(contract=Stock("BBB", "SMART", "USD"), position=1)],
+    }
+
+    _mock_regime_tickers(portfolio_manager, mocker)
+    _mock_regime_history(portfolio_manager, mocker, [100.0, 100.0, 100.0, 100.0])
+    portfolio_manager.ibkr.request_executions = mocker.AsyncMock(return_value=[])
+
+    _, orders = await portfolio_manager.check_regime_rebalance_positions(
+        account_summary, portfolio_positions
+    )
+
+    assert orders == []
+
+
+@pytest.mark.asyncio
+async def test_regime_rebalance_hard_band_ignores_ratio_gate(portfolio_manager, mocker):
+    portfolio_manager.config.regime_rebalance.soft_band = 0.30
+    portfolio_manager.config.regime_rebalance.hard_band = 0.10
+    portfolio_manager.config.regime_rebalance.choppiness_min = 10.0
+    portfolio_manager.config.regime_rebalance.efficiency_max = 0.01
+    portfolio_manager.config.regime_rebalance.ratio_gate = SimpleNamespace(
+        enabled=True,
+        anchor="BBB",
+        drift_max=1.25,
+        var_min=0.0,
+    )
+    account_summary = {"NetLiquidation": SimpleNamespace(value="400")}
+    portfolio_positions = {
+        "AAA": [SimpleNamespace(contract=Stock("AAA", "SMART", "USD"), position=3)],
+        "BBB": [SimpleNamespace(contract=Stock("BBB", "SMART", "USD"), position=1)],
+    }
+
+    _mock_regime_tickers(portfolio_manager, mocker)
+    _mock_regime_history(portfolio_manager, mocker, [100.0, 100.0, 100.0, 100.0])
+    portfolio_manager.ibkr.request_executions = mocker.AsyncMock(return_value=[])
+
+    _, orders = await portfolio_manager.check_regime_rebalance_positions(
+        account_summary, portfolio_positions
+    )
+
+    assert orders == [("AAA", "NYSE", -1), ("BBB", "NYSE", 1)]
+
+
 @pytest.mark.asyncio
 async def test_regime_rebalance_cooldown_blocks_trades(
     portfolio_manager, mocker, monkeypatch
@@ -601,6 +695,30 @@ def test_regime_rebalance_config_rejects_deficit_hysteresis_inversion():
         RegimeRebalanceConfig(deficit_rail_start=100.0, deficit_rail_stop=200.0)
 
 
+def test_regime_rebalance_config_rejects_ratio_gate_missing_anchor():
+    with pytest.raises(ValueError, match="ratio_gate.anchor must be set"):
+        RegimeRebalanceConfig(
+            symbols=["AAA", "BBB"],
+            ratio_gate=RatioGateConfig(enabled=True, anchor=""),
+        )
+
+
+def test_regime_rebalance_config_rejects_ratio_gate_anchor_not_in_symbols():
+    with pytest.raises(ValueError, match="ratio_gate.anchor must be in"):
+        RegimeRebalanceConfig(
+            symbols=["AAA", "BBB"],
+            ratio_gate=RatioGateConfig(enabled=True, anchor="CCC"),
+        )
+
+
+def test_regime_rebalance_config_rejects_ratio_gate_only_anchor_symbol():
+    with pytest.raises(ValueError, match="ratio_gate.anchor must leave"):
+        RegimeRebalanceConfig(
+            symbols=["AAA"],
+            ratio_gate=RatioGateConfig(enabled=True, anchor="AAA"),
+        )
+
+
 @pytest.mark.asyncio
 async def test_regime_rebalance_respects_no_trading(portfolio_manager, mocker):
     portfolio_manager.config.trading_is_allowed = mocker.Mock(

thetagang.toml

@@ -524,6 +524,14 @@ minimum_open_interest = 10
 # deficit_rail_stop = 2500  # hysteresis stop once within this band
 # order_history_lookback_days = 30  # look back this many calendar days for fills
 # shares_only = false  # when true, disables all option writes/rolls
+#
+# # Optional ratio gate (shadow mode when enabled=false):
+# # Uses log((rest basket) / anchor) to measure variance vs drift.
+# [regime_rebalance.ratio_gate]
+# enabled = false  # when true, gates soft rebalances only
+# anchor = "BTAL"  # hedge sleeve anchor; rest is symbols minus anchor
+# drift_max = 1.25  # t-stat threshold for mean drift of ratio returns
+# var_min = 0.0  # minimum variance of ratio returns
 
   [symbols.TLT]
   weight = 0.2

thetagang/config.py

@@ -536,6 +536,19 @@ class Puts(BaseModel):
     )
 
 
+class RatioGateConfig(BaseModel, DisplayMixin):
+    enabled: bool = Field(default=False)
+    anchor: str = Field(default="")
+    drift_max: float = Field(default=1.25, ge=0.0)
+    var_min: float = Field(default=0.0, ge=0.0)
+
+    def add_to_table(self, table: Table, section: str = "") -> None:
+        table.add_row("", "Ratio gate enabled", "=", f"{self.enabled}")
+        table.add_row("", "Ratio gate anchor", "=", self.anchor or "-")
+        table.add_row("", "Ratio gate drift max", "=", f"{ffmt(self.drift_max)}")
+        table.add_row("", "Ratio gate var min", "=", f"{ffmt(self.var_min)}")
+
+
 class RegimeRebalanceConfig(BaseModel, DisplayMixin):
     enabled: bool = Field(default=False)
     symbols: List[str] = Field(default_factory=list)
@@ -554,6 +567,7 @@ class RegimeRebalanceConfig(BaseModel, DisplayMixin):
     eps: float = Field(default=1e-8, gt=0.0)
     order_history_lookback_days: int = Field(default=30, ge=1)
     shares_only: bool = Field(default=False)
+    ratio_gate: Optional[RatioGateConfig] = None
 
     @model_validator(mode="after")
     def validate_bands(self) -> Self:
@@ -567,6 +581,18 @@ def validate_bands(self) -> Self:
             raise ValueError(
                 "regime_rebalance.deficit_rail_start must be >= deficit_rail_stop"
             )
+        if self.ratio_gate is not None:
+            if not self.ratio_gate.anchor:
+                raise ValueError("regime_rebalance.ratio_gate.anchor must be set")
+            if self.ratio_gate.anchor not in self.symbols:
+                raise ValueError(
+                    "regime_rebalance.ratio_gate.anchor must be in regime_rebalance.symbols"
+                )
+            rest_symbols = [s for s in self.symbols if s != self.ratio_gate.anchor]
+            if not rest_symbols:
+                raise ValueError(
+                    "regime_rebalance.ratio_gate.anchor must leave at least one non-anchor symbol"
+                )
         return self
 
     def add_to_table(self, table: Table, section: str = "") -> None:
@@ -592,6 +618,8 @@ def add_to_table(self, table: Table, section: str = "") -> None:
         table.add_row("", "Deficit rail start", "=", f"{dfmt(self.deficit_rail_start)}")
         table.add_row("", "Deficit rail stop", "=", f"{dfmt(self.deficit_rail_stop)}")
         table.add_row("", "Shares only", "=", f"{self.shares_only}")
+        if self.ratio_gate is not None:
+            self.ratio_gate.add_to_table(table, section)
 
 
 class ActionWhenClosedEnum(str, Enum):