Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
48 changes: 47 additions & 1 deletion tests/test_market_regime_control_consumption.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,9 @@

from quant_platform_kit.strategy_contracts import StrategyContext

from us_equity_strategies.entrypoints._common import apply_market_regime_control_to_weights
from us_equity_strategies.entrypoints._common import (
apply_market_regime_control_to_weights,
)


def _market_regime_payload(route: str, scalar: float, *, authorized: bool = True) -> dict[str, object]:
Expand Down Expand Up @@ -50,6 +52,31 @@ def _market_regime_payload(route: str, scalar: float, *, authorized: bool = True
}


def _core_only_unavailable_regime() -> dict[str, object]:
unavailable = {"enabled": False, "available": False}
return {
"profile": "market_regime_control",
"schema_version": "soxl_core_only_market_regime_unavailable.v1",
"as_of": "2026-05-28",
"candidate_id": "SOXL_P3_CORE_ONLY_9_INPUT_V1",
"market_regime_control_enabled": False,
"component_signals": {
name: dict(unavailable)
for name in (
"crisis",
"macro",
"taco",
"panic_reversal",
"volatility_delever_price_rebound",
)
},
"execution_controls": {
"position_control_allowed": False,
"consumption_evidence_status": "static_research_only",
},
}


def test_market_regime_control_weight_consumer_scales_risk_weight_to_safe_haven() -> None:
weights, diagnostics = apply_market_regime_control_to_weights(
{"AAPL": 0.80, "BOXX": 0.20},
Expand Down Expand Up @@ -120,3 +147,22 @@ def test_market_regime_control_weight_consumer_does_not_apply_unapproved_positio
assert diagnostics["market_regime_control_position_control_authorized"] is False
assert diagnostics["market_regime_control_consumption_evidence_status"] == "notification_only"
assert diagnostics["market_regime_control_applied"] is False


def test_core_only_unavailable_regime_cannot_change_weights_when_consumer_is_enabled() -> None:
weights, diagnostics = apply_market_regime_control_to_weights(
{"SOXL": 0.70, "SOXX": 0.20, "BOXX": 0.10},
market_regime_control_config={"market_regime_control_enabled": True},
ctx=StrategyContext(
as_of="2026-05-28",
artifacts={"market_regime_control": _core_only_unavailable_regime()},
),
safe_haven="BOXX",
)

assert weights == {"SOXL": 0.70, "SOXX": 0.20, "BOXX": 0.10}
assert diagnostics["market_regime_control_found"] is True
assert diagnostics["market_regime_control_active"] is False
assert diagnostics["market_regime_control_position_control_allowed"] is False
assert diagnostics["market_regime_control_position_control_authorized"] is False
assert diagnostics["market_regime_control_applied"] is False
76 changes: 76 additions & 0 deletions tests/test_strategy_plans.py
Original file line number Diff line number Diff line change
Expand Up @@ -1887,6 +1887,82 @@ def test_soxl_soxx_trend_income_can_disable_market_regime_control_position_effec
self.assertAlmostEqual(plan["targets"]["BOXX"], 100000.0 * 0.10)
self.assertFalse(plan["notification_context"]["risk_controls"]["market_regime_control"]["enabled"])

def test_soxl_soxx_trend_income_unavailable_regime_cannot_change_targets(self):
_skip_if_missing_numeric_stack()
from us_equity_strategies.strategies.soxl_soxx_trend_income import (
SOXX_GATE_TIERED_BLEND_MODE,
build_rebalance_plan as build_soxl_soxx_plan,
)

account_state = {
"available_cash": 5000.0,
"market_values": {"SOXL": 0.0, "SOXX": 0.0, "BOXX": 100000.0, "QQQI": 0.0, "SPYI": 0.0},
"quantities": {"SOXL": 0, "SOXX": 0, "BOXX": 1000, "QQQI": 0, "SPYI": 0},
"sellable_quantities": {"SOXL": 0, "SOXX": 0, "BOXX": 1000, "QQQI": 0, "SPYI": 0},
"total_strategy_equity": 100000.0,
"metadata": {
"market_regime_control": {
"profile": "market_regime_control",
"schema_version": "soxl_core_only_market_regime_unavailable.v1",
"candidate_id": "SOXL_P3_CORE_ONLY_9_INPUT_V1",
"market_regime_control_enabled": False,
Comment thread
Pigbibi marked this conversation as resolved.
"component_signals": {
name: {"enabled": False, "available": False}
for name in (
"crisis",
"macro",
"taco",
"panic_reversal",
"volatility_delever_price_rebound",
)
},
"execution_controls": {
"position_control_allowed": False,
"consumption_evidence_status": "static_research_only",
},
}
},
}

plan = build_soxl_soxx_plan(
{
"soxl": {"price": 50.0, "ma_trend": 45.0},
"soxx": {"price": 109.0, "ma_trend": 100.0},
},
account_state,
trend_ma_window=140,
translator=_translator,
cash_reserve_ratio=0.03,
min_trade_ratio=0.01,
min_trade_floor=100.0,
rebalance_threshold_ratio=0.01,
income_layer_start_usd=150000.0,
income_layer_max_ratio=0.15,
income_layer_qqqi_weight=0.70,
income_layer_spyi_weight=0.30,
attack_allocation_mode=SOXX_GATE_TIERED_BLEND_MODE,
blend_gate_trend_source="SOXX",
trend_entry_buffer=0.08,
trend_mid_buffer=0.06,
trend_exit_buffer=0.02,
blend_gate_soxl_weight=0.70,
blend_gate_mid_soxl_weight=0.65,
blend_gate_active_soxx_weight=0.20,
blend_gate_defensive_soxx_weight=0.15,
market_regime_control_enabled=True,
market_regime_control_apply_risk_reduced=True,
)

self.assertTrue(plan["market_regime_control_found"])
self.assertFalse(plan["market_regime_control_active"])
self.assertFalse(plan["market_regime_control_applied"])
self.assertFalse(plan["market_regime_control_position_control_allowed"])
self.assertFalse(plan["market_regime_control_position_control_authorized"])
self.assertEqual(plan["active_risk_asset"], "SOXX+SOXL")
self.assertAlmostEqual(plan["targets"]["SOXL"], 100000.0 * 0.70)
self.assertAlmostEqual(plan["targets"]["SOXX"], 100000.0 * 0.20)
self.assertAlmostEqual(plan["targets"]["BOXX"], 100000.0 * 0.10)

def test_soxl_soxx_trend_income_legacy_crisis_adapter_maps_to_market_regime_risk_off(self):
_skip_if_missing_numeric_stack()
from us_equity_strategies.strategies.soxl_soxx_trend_income import (
Expand Down