From ffb4e63f77f6770222a0bf620ab949a3a12fa28c Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 4 Jul 2026 14:12:14 +0800 Subject: [PATCH] Add opt-in dual-leg combo regime cap Co-Authored-By: Codex --- src/crypto_strategies/catalog.py | 8 + src/crypto_strategies/entrypoints/__init__.py | 8 + src/crypto_strategies/manifests/__init__.py | 8 + .../strategies/crypto_equity_combo.py | 191 ++++++++++++++++-- tests/test_crypto_equity_combo.py | 75 +++++++ 5 files changed, 268 insertions(+), 22 deletions(-) diff --git a/src/crypto_strategies/catalog.py b/src/crypto_strategies/catalog.py index 69a0f16..e183dda 100644 --- a/src/crypto_strategies/catalog.py +++ b/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, diff --git a/src/crypto_strategies/entrypoints/__init__.py b/src/crypto_strategies/entrypoints/__init__.py index c2c60c5..f2e9af8 100644 --- a/src/crypto_strategies/entrypoints/__init__.py +++ b/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)), diff --git a/src/crypto_strategies/manifests/__init__.py b/src/crypto_strategies/manifests/__init__.py index 307c49a..2a26fa2 100644 --- a/src/crypto_strategies/manifests/__init__.py +++ b/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, diff --git a/src/crypto_strategies/strategies/crypto_equity_combo.py b/src/crypto_strategies/strategies/crypto_equity_combo.py index 3ac8b09..8cdd800 100644 --- a/src/crypto_strategies/strategies/crypto_equity_combo.py +++ b/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), diff --git a/tests/test_crypto_equity_combo.py b/tests/test_crypto_equity_combo.py index 708a0a2..36c29bd 100644 --- a/tests/test_crypto_equity_combo.py +++ b/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}