Skip to content

[codex] guard crypto equity combo exits and rotation config #56

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Pigbibi merged 1 commit into from
Jul 4, 2026
Merged

[codex] guard crypto equity combo exits and rotation config #56

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions src/crypto_strategies/catalog.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
15 changes: 15 additions & 0 deletions src/crypto_strategies/entrypoints/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Comment on lines +467 to +468

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Forward the strategy date into the DCA overlay

When zscore_exit_allow_outside_execution_window is set to False, the downstream BTC DCA plan decides whether the Z-score overlay is in-window from its as_of argument (kwargs.get("as_of")); this entrypoint now forwards the flag but still never forwards ctx.as_of, so the DCA code falls back to the wall-clock date. Backtests/replays or delayed evaluations can therefore skip or apply the combo BTC exit based on today rather than the StrategyContext date; pass as_of=ctx.as_of with the DCA-related kwargs.

Useful? React with 👍 / 👎.

),
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] = []
Expand Down
7 changes: 7 additions & 0 deletions src/crypto_strategies/manifests/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
6 changes: 5 additions & 1 deletion src/crypto_strategies/strategies/crypto_btc_dca.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
88 changes: 68 additions & 20 deletions src/crypto_strategies/strategies/crypto_equity_combo.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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
Expand All @@ -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(
Expand All @@ -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 (
Expand All @@ -110,15 +146,19 @@ 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 {}

trend_pool = resolve_authoritative_rotation_pool(
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(
Expand All @@ -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,
)


Expand Down Expand Up @@ -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,
Expand All @@ -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)
Expand All @@ -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,
Expand Down
91 changes: 91 additions & 0 deletions tests/test_crypto_equity_combo.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
DEFAULT_TREND_WEIGHT,
PROFILE_NAME,
SIGNAL_SOURCE,
build_target_weights,
compute_signals,
extract_managed_symbols,
)
Expand All @@ -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}
Expand Down
Loading