Add opt-in dual-leg combo regime cap

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

src/crypto_strategies/catalog.py

@@ -143,7 +143,15 @@
     "btc_weight": 0.30,
     "trend_weight": 0.70,
     "dynamic_mode": True,
+    "dynamic_regime_mode": "legacy",
     "dynamic_regime_off_cut": 0.50,
+    "dynamic_hard_sma200_ratio": 0.97,
+    "dynamic_hard_ma200_slope": -0.015,
+    "dynamic_soft_sma200_ratio": 1.05,
+    "dynamic_hard_btc_weight": 0.30,
+    "dynamic_hard_trend_weight": 0.0,
+    "dynamic_soft_btc_weight": 0.45,
+    "dynamic_soft_trend_weight": 0.15,
     "smart_multiplier_enabled": True,
     "cycle_indicator_enabled": True,
     "zscore_exit_enabled": True,

src/crypto_strategies/entrypoints/__init__.py

@@ -460,7 +460,15 @@ def evaluate_crypto_equity_combo(ctx: StrategyContext) -> StrategyDecision:
         btc_weight=float(config.get("btc_weight", 0.30)),
         trend_weight=float(config.get("trend_weight", 0.70)),
         dynamic_mode=bool(config.get("dynamic_mode", True)),
+        dynamic_regime_mode=str(config.get("dynamic_regime_mode", "legacy")),
         dynamic_regime_off_cut=float(config.get("dynamic_regime_off_cut", 0.50)),
+        dynamic_hard_sma200_ratio=float(config.get("dynamic_hard_sma200_ratio", 0.97)),
+        dynamic_hard_ma200_slope=float(config.get("dynamic_hard_ma200_slope", -0.015)),
+        dynamic_soft_sma200_ratio=float(config.get("dynamic_soft_sma200_ratio", 1.05)),
+        dynamic_hard_btc_weight=float(config.get("dynamic_hard_btc_weight", 0.30)),
+        dynamic_hard_trend_weight=float(config.get("dynamic_hard_trend_weight", 0.0)),
+        dynamic_soft_btc_weight=float(config.get("dynamic_soft_btc_weight", 0.45)),
+        dynamic_soft_trend_weight=float(config.get("dynamic_soft_trend_weight", 0.15)),
         smart_multiplier_enabled=bool(config.get("smart_multiplier_enabled", True)),
         cycle_indicator_enabled=bool(config.get("cycle_indicator_enabled", True)),
         zscore_exit_enabled=bool(config.get("zscore_exit_enabled", True)),

src/crypto_strategies/manifests/__init__.py

@@ -146,7 +146,15 @@
         "btc_weight": 0.30,
         "trend_weight": 0.70,
         "dynamic_mode": True,
+        "dynamic_regime_mode": "legacy",
         "dynamic_regime_off_cut": 0.50,
+        "dynamic_hard_sma200_ratio": 0.97,
+        "dynamic_hard_ma200_slope": -0.015,
+        "dynamic_soft_sma200_ratio": 1.05,
+        "dynamic_hard_btc_weight": 0.30,
+        "dynamic_hard_trend_weight": 0.0,
+        "dynamic_soft_btc_weight": 0.45,
+        "dynamic_soft_trend_weight": 0.15,
         "smart_multiplier_enabled": True,
         "cycle_indicator_enabled": True,
         "zscore_exit_enabled": True,

src/crypto_strategies/strategies/crypto_equity_combo.py

@@ -6,8 +6,10 @@
 configured.
 
 Static mode: fixed weights per leg (default: 30/70 BTC/trend).
-Dynamic mode: regime-based adjustment — when BTC is below SMA200, reduce
-trend leg by 50 % and re-allocate to BTC.
+Dynamic legacy mode: regime-based adjustment — when BTC is below SMA200,
+reduce trend leg by 50 % and re-allocate to BTC.
+Dynamic dual-leg mode: opt-in tiered regime adjustment that can cap both BTC
+and trend legs, leaving the residual in cash.
 
 Usage
 -----
@@ -36,6 +38,8 @@
 DEFAULT_BTC_WEIGHT = 0.30
 DEFAULT_TREND_WEIGHT = 0.70
 DYNAMIC_REGIME_OFF_CUT = 0.50
+DYNAMIC_REGIME_MODE_LEGACY = "legacy"
+DYNAMIC_REGIME_MODE_DUAL_LEG = "dual_leg"
 
 TREND_ONLY_KWARGS = frozenset({
     "trend_pool_size",
@@ -58,6 +62,137 @@ def _clamp_ratio(value: float, *, default: float = 1.0) -> float:
     return min(1.0, max(0.0, numeric))
 
 
+def _normalized_regime_mode(value: object) -> str:
+    mode = str(value or DYNAMIC_REGIME_MODE_LEGACY).strip().lower().replace("-", "_")
+    if mode in {"dual", "dual_leg", "tiered", "cash_cap"}:
+        return DYNAMIC_REGIME_MODE_DUAL_LEG
+    return DYNAMIC_REGIME_MODE_LEGACY
+
+
+def _first_finite(payload: dict[str, Any] | None, *keys: str) -> float | None:
+    if not isinstance(payload, dict):
+        return None
+    for key in keys:
+        value = payload.get(key)
+        try:
+            numeric = float(value)
+        except (TypeError, ValueError):
+            continue
+        if numeric == numeric:
+            return numeric
+    return None
+
+
+def _btc_sma200_ratio(
+    prices: dict[str, float],
+    benchmark_snapshot: dict[str, Any] | None,
+    indicators_map: dict[str, Any] | None,
+) -> float | None:
+    gap = _first_finite(
+        benchmark_snapshot,
+        "sma200_gap",
+        "gap_vs_sma200",
+        "price_vs_sma200",
+    )
+    if gap is not None:
+        return 1.0 + gap
+
+    ratio = _first_finite(
+        benchmark_snapshot,
+        "price_sma200_ratio",
+        "sma200_ratio",
+        "mayer_multiple",
+    )
+    if ratio is not None:
+        return ratio
+
+    ma200 = _first_finite(benchmark_snapshot, "ma200", "sma200", "sma_200")
+    if ma200 is None and isinstance(indicators_map, dict):
+        btc_indicators = indicators_map.get("BTCUSDT")
+        ma200 = _first_finite(btc_indicators, "ma200", "sma200", "sma_200")
+
+    price = _first_finite(benchmark_snapshot, "close", "price")
+    if price is None:
+        price = _first_finite(prices, "BTCUSDT", "BTC")
+
+    if price is None or ma200 is None or ma200 <= 0.0:
+        return None
+    return price / ma200
+
+
+def _resolve_dynamic_weights(
+    prices: dict[str, float],
+    indicators_map: dict[str, Any] | None,
+    benchmark_snapshot: dict[str, Any] | None,
+    *,
+    btc_weight: float,
+    trend_weight: float,
+    dynamic_mode: bool,
+    dynamic_regime_mode: str,
+    dynamic_regime_off_cut: float,
+    dynamic_hard_sma200_ratio: float,
+    dynamic_hard_ma200_slope: float,
+    dynamic_soft_sma200_ratio: float,
+    dynamic_hard_btc_weight: float,
+    dynamic_hard_trend_weight: float,
+    dynamic_soft_btc_weight: float,
+    dynamic_soft_trend_weight: float,
+) -> tuple[float, float, dict[str, object]]:
+    ratio = _btc_sma200_ratio(prices, benchmark_snapshot, indicators_map)
+    ma200_slope = _first_finite(benchmark_snapshot, "ma200_slope", "sma200_slope")
+
+    regime_off = False
+    if isinstance(benchmark_snapshot, dict):
+        regime_on = benchmark_snapshot.get("regime_on")
+        if regime_on is not None:
+            regime_off = not bool(regime_on)
+        elif ratio is not None:
+            regime_off = ratio <= 1.0
+    else:
+        btc_snapshot = _extract_btc_snapshot(indicators_map or {})
+        regime_off = not btc_snapshot.get("regime_on", True)
+
+    mode = _normalized_regime_mode(dynamic_regime_mode)
+    regime_tier = "risk_on"
+    regime_off_cut = 0.0
+    effective_btc = btc_weight
+    effective_trend = trend_weight
+
+    if dynamic_mode and mode == DYNAMIC_REGIME_MODE_DUAL_LEG:
+        hard = regime_off and (
+            (ratio is None and ma200_slope is None)
+            or (ratio is not None and ratio < dynamic_hard_sma200_ratio)
+            or (ma200_slope is not None and ma200_slope < dynamic_hard_ma200_slope)
+        )
+        soft = (
+            regime_off
+            or (ratio is not None and ratio < dynamic_soft_sma200_ratio)
+            or (ma200_slope is not None and ma200_slope < 0.0)
+        )
+        if hard:
+            regime_tier = "hard"
+            effective_btc = _clamp_ratio(dynamic_hard_btc_weight, default=btc_weight)
+            effective_trend = _clamp_ratio(dynamic_hard_trend_weight, default=0.0)
+        elif soft:
+            regime_tier = "soft"
+            effective_btc = _clamp_ratio(dynamic_soft_btc_weight, default=btc_weight)
+            effective_trend = _clamp_ratio(dynamic_soft_trend_weight, default=trend_weight)
+    elif dynamic_mode and regime_off:
+        regime_tier = "legacy_regime_off"
+        regime_off_cut = _clamp_ratio(dynamic_regime_off_cut, default=DYNAMIC_REGIME_OFF_CUT)
+        effective_btc = btc_weight + trend_weight * regime_off_cut
+        effective_trend = trend_weight * (1.0 - regime_off_cut)
+
+    return effective_btc, effective_trend, {
+        "regime_off": regime_off,
+        "regime_mode": mode,
+        "regime_tier": regime_tier,
+        "dynamic_regime_off_cut": regime_off_cut,
+        "btc_sma200_ratio": ratio,
+        "ma200_slope": ma200_slope,
+    }
+
+
 def _compute_btc_leg(
     total_equity: float,
     btc_weight: float,
@@ -208,7 +343,15 @@ def build_target_weights(
     btc_weight: float = DEFAULT_BTC_WEIGHT,
     trend_weight: float = DEFAULT_TREND_WEIGHT,
     dynamic_mode: bool = True,
+    dynamic_regime_mode: str = DYNAMIC_REGIME_MODE_LEGACY,
     dynamic_regime_off_cut: float = DYNAMIC_REGIME_OFF_CUT,
+    dynamic_hard_sma200_ratio: float = 0.97,
+    dynamic_hard_ma200_slope: float = -0.015,
+    dynamic_soft_sma200_ratio: float = 1.05,
+    dynamic_hard_btc_weight: float = 0.30,
+    dynamic_hard_trend_weight: float = 0.0,
+    dynamic_soft_btc_weight: float = 0.45,
+    dynamic_soft_trend_weight: float = 0.15,
     translator=None,
     **kwargs: Any,
 ) -> tuple[dict[str, float], dict[str, object]]:
@@ -233,24 +376,23 @@ def build_target_weights(
 
     state = state or {}
 
-    # Dynamic regime adjustment
-    regime_off = False
-    if benchmark_snapshot:
-        regime_on = benchmark_snapshot.get("regime_on")
-        if regime_on is not None:
-            regime_off = not bool(regime_on)
-    else:
-        btc_snapshot = _extract_btc_snapshot(indicators_map or {})
-        regime_off = not btc_snapshot.get("regime_on", True)
-
-    if dynamic_mode and regime_off:
-        regime_off_cut = _clamp_ratio(dynamic_regime_off_cut, default=DYNAMIC_REGIME_OFF_CUT)
-        effective_btc = btc_weight + trend_weight * regime_off_cut
-        effective_trend = trend_weight * (1.0 - regime_off_cut)
-    else:
-        regime_off_cut = 0.0
-        effective_btc = btc_weight
-        effective_trend = trend_weight
+    effective_btc, effective_trend, regime_metadata = _resolve_dynamic_weights(
+        prices,
+        indicators_map,
+        benchmark_snapshot,
+        btc_weight=btc_weight,
+        trend_weight=trend_weight,
+        dynamic_mode=dynamic_mode,
+        dynamic_regime_mode=dynamic_regime_mode,
+        dynamic_regime_off_cut=dynamic_regime_off_cut,
+        dynamic_hard_sma200_ratio=dynamic_hard_sma200_ratio,
+        dynamic_hard_ma200_slope=dynamic_hard_ma200_slope,
+        dynamic_soft_sma200_ratio=dynamic_soft_sma200_ratio,
+        dynamic_hard_btc_weight=dynamic_hard_btc_weight,
+        dynamic_hard_trend_weight=dynamic_hard_trend_weight,
+        dynamic_soft_btc_weight=dynamic_soft_btc_weight,
+        dynamic_soft_trend_weight=dynamic_soft_trend_weight,
+    )
 
     # Compute legs
     btc_weights, btc_leg_metadata = _compute_btc_leg(
@@ -295,11 +437,16 @@ def build_target_weights(
             "trend_weight": effective_trend,
             "base_btc_weight": btc_weight,
             "base_trend_weight": trend_weight,
-            "dynamic_regime_off_cut": regime_off_cut,
+            "dynamic_regime_off_cut": regime_metadata["dynamic_regime_off_cut"],
+            "dynamic_regime_mode": regime_metadata["regime_mode"],
+            "regime_tier": regime_metadata["regime_tier"],
         },
         "btc_leg": {"weights": btc_weights, **btc_leg_metadata},
         "trend_leg": {"weights": trend_weights, **trend_metadata},
-        "regime_off": regime_off,
+        "regime_off": regime_metadata["regime_off"],
+        "regime_tier": regime_metadata["regime_tier"],
+        "btc_sma200_ratio": regime_metadata["btc_sma200_ratio"],
+        "ma200_slope": regime_metadata["ma200_slope"],
         "dynamic_mode": dynamic_mode,
         "gross_exposure": sum(combined.values()),
         "selected_count": len(combined),

tests/test_crypto_equity_combo.py

@@ -156,6 +156,81 @@ def test_dynamic_regime_off_cut_is_configurable(self) -> None:
         self.assertAlmostEqual(custom_metadata["combo"]["trend_weight"], 0.49)
         self.assertAlmostEqual(custom_metadata["combo"]["dynamic_regime_off_cut"], 0.30)
 
+    def test_dual_leg_regime_hard_caps_btc_and_trend_to_cash(self) -> None:
+        """Opt-in dual-leg mode should allow hard risk-off to leave residual cash."""
+        _, metadata = build_target_weights(
+            prices={"BTCUSDT": 94000.0},
+            indicators_map={},
+            universe_snapshot=[],
+            benchmark_snapshot={
+                "regime_on": False,
+                "ma200": 100000.0,
+                "ma200_slope": -0.02,
+            },
+            portfolio={"total_equity": 100000.0, "buying_power": 1000.0},
+            btc_weight=0.30,
+            trend_weight=0.70,
+            dynamic_regime_mode="dual_leg",
+            dynamic_hard_btc_weight=0.25,
+            dynamic_hard_trend_weight=0.0,
+            smart_multiplier_enabled=False,
+        )
+
+        combo = metadata["combo"]
+        self.assertAlmostEqual(combo["btc_weight"], 0.25)
+        self.assertAlmostEqual(combo["trend_weight"], 0.0)
+        self.assertEqual(combo["dynamic_regime_mode"], "dual_leg")
+        self.assertEqual(combo["regime_tier"], "hard")
+        self.assertAlmostEqual(metadata["btc_sma200_ratio"], 0.94)
+
+    def test_dual_leg_regime_soft_uses_neutral_cash_cap(self) -> None:
+        """Opt-in dual-leg mode should support a soft/neutral tier."""
+        _, metadata = build_target_weights(
+            prices={"BTCUSDT": 102000.0},
+            indicators_map={},
+            universe_snapshot=[],
+            benchmark_snapshot={
+                "regime_on": False,
+                "ma200": 100000.0,
+                "ma200_slope": -0.005,
+            },
+            portfolio={"total_equity": 100000.0, "buying_power": 1000.0},
+            btc_weight=0.30,
+            trend_weight=0.70,
+            dynamic_regime_mode="dual_leg",
+            dynamic_soft_btc_weight=0.45,
+            dynamic_soft_trend_weight=0.15,
+            smart_multiplier_enabled=False,
+        )
+
+        combo = metadata["combo"]
+        self.assertAlmostEqual(combo["btc_weight"], 0.45)
+        self.assertAlmostEqual(combo["trend_weight"], 0.15)
+        self.assertEqual(combo["regime_tier"], "soft")
+
+    def test_dual_leg_regime_keeps_base_weights_when_risk_on(self) -> None:
+        """Opt-in dual-leg mode should not alter base weights in risk-on conditions."""
+        _, metadata = build_target_weights(
+            prices={"BTCUSDT": 110000.0},
+            indicators_map={},
+            universe_snapshot=[],
+            benchmark_snapshot={
+                "regime_on": True,
+                "ma200": 100000.0,
+                "ma200_slope": 0.02,
+            },
+            portfolio={"total_equity": 100000.0, "buying_power": 1000.0},
+            btc_weight=0.30,
+            trend_weight=0.70,
+            dynamic_regime_mode="dual_leg",
+            smart_multiplier_enabled=False,
+        )
+
+        combo = metadata["combo"]
+        self.assertAlmostEqual(combo["btc_weight"], 0.30)
+        self.assertAlmostEqual(combo["trend_weight"], 0.70)
+        self.assertEqual(combo["regime_tier"], "risk_on")
+
     def test_compute_signals_returns_tuple(self) -> None:
         """compute_signals should return a 5-tuple with weights, signal_desc, cash_residual, status_desc, metadata."""
         prices = {"BTCUSDT": 60000.0}