From 091b4657ef1d52ae44a74c170d51ec57921e8d94 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 4 Jul 2026 11:11:45 +0800 Subject: [PATCH] guard crypto equity combo exits and rotation config Co-Authored-By: Codex --- src/crypto_strategies/catalog.py | 8 ++ src/crypto_strategies/entrypoints/__init__.py | 15 +++ src/crypto_strategies/manifests/__init__.py | 7 ++ .../strategies/crypto_btc_dca.py | 6 +- .../strategies/crypto_equity_combo.py | 88 ++++++++++++++---- tests/test_crypto_equity_combo.py | 91 +++++++++++++++++++ 6 files changed, 194 insertions(+), 21 deletions(-) diff --git a/src/crypto_strategies/catalog.py b/src/crypto_strategies/catalog.py index 5aed5cc..27e6fed 100644 --- a/src/crypto_strategies/catalog.py +++ b/src/crypto_strategies/catalog.py @@ -143,11 +143,19 @@ "btc_weight": 0.30, "trend_weight": 0.70, "dynamic_mode": True, + "smart_multiplier_enabled": True, + "cycle_indicator_enabled": True, + "zscore_exit_enabled": True, + "zscore_exit_parking_symbol": "USDT", + "zscore_exit_risk_reduced_exposure": 0.50, + "zscore_exit_risk_off_exposure": 0.25, + "zscore_exit_allow_outside_execution_window": True, "circuit_breaker_enabled": True, "btc_drawdown_threshold": 0.30, "vol_scaling_enabled": True, } + STRATEGY_DEFINITIONS: dict[str, StrategyDefinition] = { CRYPTO_LIVE_POOL_ROTATION_PROFILE: StrategyDefinition( profile=CRYPTO_LIVE_POOL_ROTATION_PROFILE, diff --git a/src/crypto_strategies/entrypoints/__init__.py b/src/crypto_strategies/entrypoints/__init__.py index 7189c08..5f3cf46 100644 --- a/src/crypto_strategies/entrypoints/__init__.py +++ b/src/crypto_strategies/entrypoints/__init__.py @@ -459,8 +459,23 @@ def evaluate_crypto_equity_combo(ctx: StrategyContext) -> StrategyDecision: 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)), + zscore_exit_parking_symbol=str(config.get("zscore_exit_parking_symbol", "USDT")), + zscore_exit_risk_reduced_exposure=float( + config.get("zscore_exit_risk_reduced_exposure", 0.50) + ), + zscore_exit_risk_off_exposure=float(config.get("zscore_exit_risk_off_exposure", 0.25)), + zscore_exit_allow_outside_execution_window=bool( + config.get("zscore_exit_allow_outside_execution_window", True) + ), + trend_pool_size=int(config.get("trend_pool_size", 5)), + rotation_top_n=int(config.get("rotation_top_n", 2)), + weight_mode=str(config.get("weight_mode", "inverse_vol")), + allow_rotation_refresh=bool(config.get("allow_rotation_refresh", True)), circuit_breaker_enabled=bool(config.get("circuit_breaker_enabled", True)), + btc_drawdown_threshold=float(config.get("btc_drawdown_threshold", 0.30)), vol_scaling_enabled=bool(config.get("vol_scaling_enabled", True)), + target_vol=float(config.get("target_vol", 0.40)), + max_leverage=float(config.get("max_leverage", 1.0)), ) positions: list[PositionTarget] = [] diff --git a/src/crypto_strategies/manifests/__init__.py b/src/crypto_strategies/manifests/__init__.py index acfb639..e5543fc 100644 --- a/src/crypto_strategies/manifests/__init__.py +++ b/src/crypto_strategies/manifests/__init__.py @@ -146,6 +146,13 @@ "btc_weight": 0.30, "trend_weight": 0.70, "dynamic_mode": True, + "smart_multiplier_enabled": True, + "cycle_indicator_enabled": True, + "zscore_exit_enabled": True, + "zscore_exit_parking_symbol": "USDT", + "zscore_exit_risk_reduced_exposure": 0.50, + "zscore_exit_risk_off_exposure": 0.25, + "zscore_exit_allow_outside_execution_window": True, "circuit_breaker_enabled": True, "btc_drawdown_threshold": 0.30, "vol_scaling_enabled": True, diff --git a/src/crypto_strategies/strategies/crypto_btc_dca.py b/src/crypto_strategies/strategies/crypto_btc_dca.py index 07db034..f6e5262 100644 --- a/src/crypto_strategies/strategies/crypto_btc_dca.py +++ b/src/crypto_strategies/strategies/crypto_btc_dca.py @@ -940,7 +940,11 @@ def compute_signals( base_investment_usd=base_amount, derived_indicators=derived_indicators, translator=translator, - **{k: v for k, v in kwargs.items() if k not in ("as_of",)}, + **{ + k: v + for k, v in kwargs.items() + if k not in {"as_of", "smart_multiplier_enabled", "base_investment_usd"} + }, ) except (ValueError, TypeError) as exc: logger.warning("btc_dca build_rebalance_plan failed, falling back: %s", exc) diff --git a/src/crypto_strategies/strategies/crypto_equity_combo.py b/src/crypto_strategies/strategies/crypto_equity_combo.py index b07ecbd..de64421 100644 --- a/src/crypto_strategies/strategies/crypto_equity_combo.py +++ b/src/crypto_strategies/strategies/crypto_equity_combo.py @@ -37,6 +37,26 @@ DEFAULT_TREND_WEIGHT = 0.70 DYNAMIC_REGIME_OFF_CUT = 0.50 +TREND_ONLY_KWARGS = frozenset({ + "trend_pool_size", + "rotation_top_n", + "weight_mode", + "allow_rotation_refresh", + "circuit_breaker_enabled", + "btc_drawdown_threshold", + "vol_scaling_enabled", + "target_vol", + "max_leverage", +}) + + +def _clamp_ratio(value: float, *, default: float = 1.0) -> float: + try: + numeric = float(value) + except (TypeError, ValueError): + return float(default) + return min(1.0, max(0.0, numeric)) + def _compute_btc_leg( total_equity: float, @@ -47,7 +67,7 @@ def _compute_btc_leg( derived_indicators: dict[str, Any] | None = None, translator=None, **kwargs: Any, -) -> dict[str, float]: +) -> tuple[dict[str, float], dict[str, object]]: """Compute BTC leg target using the enhanced smart DCA strategy. Delegates to ``crypto_btc_dca.compute_signals`` to incorporate @@ -59,35 +79,46 @@ def _compute_btc_leg( get_dynamic_btc_target_ratio, ) - smart_ratio: float | None = None + base_ratio = get_dynamic_btc_target_ratio(total_equity) + smart_ratio = base_ratio + dca_metadata: dict[str, Any] = {} + zscore_target_exposure = 1.0 try: + btc_kwargs = {k: v for k, v in kwargs.items() if k not in TREND_ONLY_KWARGS} result = compute_signals( prices=prices or {}, portfolio=portfolio, total_equity=total_equity, derived_indicators=derived_indicators, translator=translator, - **kwargs, + **btc_kwargs, ) - metadata = result.get("metadata", {}) if isinstance(result, dict) else {} - regime = str(metadata.get("regime", "")) - # Use smart multiplier only when the strategy is in an accumulation regime + dca_metadata = result.get("metadata", {}) if isinstance(result, dict) else {} + regime = str(dca_metadata.get("regime", "")) + # Use accumulation multipliers to scale the DCA leg, but do not let + # valuation-skip regimes force a full target-weight sell in combo mode. if regime and regime not in ("ordinary_dca", "ahr999_expensive"): - multiplier = float(metadata.get("multiplier", 1.0)) - base_ratio = get_dynamic_btc_target_ratio(total_equity) - # Scale ratio by multiplier, clamped to [base_ratio, base_ratio * 3] + multiplier = float(dca_metadata.get("multiplier", 1.0)) smart_ratio = min(base_ratio * max(1.0, multiplier), base_ratio * 3.0) smart_ratio = min(0.65, max(0.0, smart_ratio)) + + zscore_exit = dca_metadata.get("zscore_exit") + if isinstance(zscore_exit, dict) and zscore_exit.get("applied"): + zscore_target_exposure = _clamp_ratio( + zscore_exit.get("target_btc_exposure"), + default=1.0, + ) + smart_ratio *= zscore_target_exposure except (ValueError, TypeError) as exc: logger.debug("btc_dca smart signals unavailable (non-critical): %s", exc) - if smart_ratio is None: - from crypto_strategies.strategies.crypto_btc_dca import ( - get_dynamic_btc_target_ratio, - ) - smart_ratio = get_dynamic_btc_target_ratio(total_equity) - - return {"BTCUSDT": smart_ratio * float(btc_weight)} + target_weight = smart_ratio * float(btc_weight) + return {"BTCUSDT": target_weight}, { + "base_ratio": base_ratio, + "smart_ratio": smart_ratio, + "zscore_target_exposure": zscore_target_exposure, + "dca_metadata": dca_metadata, + } def _compute_trend_leg( @@ -100,6 +131,11 @@ def _compute_trend_leg( rotation_top_n: int = 2, weight_mode: str = "inverse_vol", vol_scaling_enabled: bool = True, + allow_rotation_refresh: bool = True, + circuit_breaker_enabled: bool = True, + btc_drawdown_threshold: float = 0.30, + target_vol: float = 0.40, + max_leverage: float = 1.0, ) -> dict[str, float]: """Compute trend leg targets using rotation logic.""" from crypto_strategies.strategies.crypto_live_pool_rotation.core import ( @@ -110,7 +146,11 @@ def _compute_trend_leg( ) btc_snapshot = _extract_btc_snapshot(indicators_map) - blocked, _ = _check_circuit_breaker(btc_snapshot) + blocked, _ = _check_circuit_breaker( + btc_snapshot, + circuit_breaker_enabled=circuit_breaker_enabled, + btc_drawdown_threshold=btc_drawdown_threshold, + ) if blocked: return {} @@ -118,7 +158,7 @@ def _compute_trend_leg( state, trend_universe_symbols=list(universe_snapshot), trend_pool_size=trend_pool_size, - allow_refresh=True, + allow_refresh=allow_rotation_refresh, ) candidates = select_rotation_weights( @@ -135,6 +175,8 @@ def _compute_trend_leg( return _apply_volatility_scaling( raw_weights, indicators_map, vol_scaling_enabled=vol_scaling_enabled, + target_vol=target_vol, + max_leverage=max_leverage, ) @@ -191,7 +233,7 @@ def build_target_weights( effective_trend = trend_weight # Compute legs - btc_weights = _compute_btc_leg( + btc_weights, btc_leg_metadata = _compute_btc_leg( total_equity, effective_btc, prices=prices, portfolio=portfolio, derived_indicators=indicators_map, @@ -205,7 +247,13 @@ def build_target_weights( indicators_map, prices, universe_symbols, state, effective_trend, trend_pool_size=int(kwargs.get("trend_pool_size", 5)), rotation_top_n=int(kwargs.get("rotation_top_n", 2)), + weight_mode=str(kwargs.get("weight_mode", "inverse_vol")), vol_scaling_enabled=bool(kwargs.get("vol_scaling_enabled", True)), + allow_rotation_refresh=bool(kwargs.get("allow_rotation_refresh", True)), + circuit_breaker_enabled=bool(kwargs.get("circuit_breaker_enabled", True)), + btc_drawdown_threshold=float(kwargs.get("btc_drawdown_threshold", 0.30)), + target_vol=float(kwargs.get("target_vol", 0.40)), + max_leverage=float(kwargs.get("max_leverage", 1.0)), ) except (ValueError, TypeError, KeyError) as exc: logger.warning("trend_leg failed, using empty weights: %s", exc) @@ -227,7 +275,7 @@ def build_target_weights( "base_btc_weight": btc_weight, "base_trend_weight": trend_weight, }, - "btc_leg": {"weights": btc_weights}, + "btc_leg": {"weights": btc_weights, **btc_leg_metadata}, "trend_leg": {"weights": trend_weights}, "regime_off": regime_off, "dynamic_mode": dynamic_mode, diff --git a/tests/test_crypto_equity_combo.py b/tests/test_crypto_equity_combo.py index 8fb8564..3cf6d48 100644 --- a/tests/test_crypto_equity_combo.py +++ b/tests/test_crypto_equity_combo.py @@ -9,6 +9,7 @@ DEFAULT_TREND_WEIGHT, PROFILE_NAME, SIGNAL_SOURCE, + build_target_weights, compute_signals, extract_managed_symbols, ) @@ -34,6 +35,96 @@ def test_extract_managed_symbols(self) -> None: symbols = extract_managed_symbols() self.assertEqual(symbols, ("BTCUSDT",)) + + def test_zscore_exit_reduces_final_btc_leg_target(self) -> None: + """Z-Score exit should reduce the combo BTC target, not stay in metadata only.""" + base_weights, _ = build_target_weights( + prices={"BTCUSDT": 60000.0}, + indicators_map={}, + universe_snapshot=[], + benchmark_snapshot={"regime_on": True}, + portfolio={"total_equity": 100000.0, "buying_power": 1000.0}, + btc_weight=1.0, + trend_weight=0.0, + smart_multiplier_enabled=False, + as_of="2026-05-26", + ) + reduced_weights, _ = build_target_weights( + prices={"BTCUSDT": 60000.0}, + indicators_map={}, + universe_snapshot=[], + benchmark_snapshot={"regime_on": True}, + portfolio={"total_equity": 100000.0, "buying_power": 1000.0}, + btc_weight=1.0, + trend_weight=0.0, + smart_multiplier_enabled=False, + as_of="2026-05-26", + zscore_exit_context={ + "plugin": "btc_zscore_exit", + "canonical_route": "risk_reduced", + "position_control": { + "final_route": "risk_reduced", + "target_allocations": {"BTCUSDT": 0.50, "USDT": 0.50}, + }, + }, + ) + + self.assertGreater(base_weights["BTCUSDT"], 0.0) + self.assertAlmostEqual(reduced_weights["BTCUSDT"], base_weights["BTCUSDT"] * 0.50) + + def test_trend_leg_honors_rotation_refresh_lock(self) -> None: + """Combo trend leg should pass allow_rotation_refresh into pool resolution.""" + indicators_map = { + "BTCUSDT": { + "close": 100000.0, + "sma200": 80000.0, + "regime_on": True, + "roc20": 0.05, + "roc60": 0.10, + "roc120": 0.20, + }, + "ETHUSDT": { + "close": 3000.0, + "sma20": 2800.0, + "sma60": 2600.0, + "sma200": 2200.0, + "roc20": 0.20, + "roc60": 0.35, + "roc120": 0.60, + "vol20": 0.25, + }, + "SOLUSDT": { + "close": 180.0, + "sma20": 170.0, + "sma60": 160.0, + "sma200": 120.0, + "roc20": 0.48, + "roc60": 0.65, + "roc120": 0.95, + "vol20": 0.30, + }, + } + + weights, metadata = build_target_weights( + prices={"BTCUSDT": 100000.0, "ETHUSDT": 3000.0, "SOLUSDT": 180.0}, + indicators_map=indicators_map, + universe_snapshot=["ETHUSDT", "SOLUSDT"], + benchmark_snapshot={"regime_on": True}, + portfolio={"total_equity": 100000.0, "buying_power": 1000.0}, + state={"rotation_pool_symbols": ["ETHUSDT"]}, + btc_weight=0.0, + trend_weight=1.0, + dynamic_mode=False, + allow_rotation_refresh=False, + rotation_top_n=1, + weight_mode="equal", + vol_scaling_enabled=False, + ) + + positive_weights = {symbol for symbol, weight in weights.items() if weight > 0.0} + self.assertEqual(positive_weights, {"ETHUSDT"}) + self.assertEqual(set(metadata["trend_leg"]["weights"]), {"ETHUSDT"}) + 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}