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
10 changes: 10 additions & 0 deletions decision_mapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
StrategyDecision,
build_allocation_intent,
build_allocation_payload,
resolve_decision_target_mode,
translate_decision_to_target_mode,
)
from strategy_registry import IBKR_PLATFORM, resolve_strategy_definition
Expand Down Expand Up @@ -142,6 +143,15 @@ def map_strategy_decision(
diagnostics = dict(decision.diagnostics)
risk_flags = tuple(str(flag) for flag in decision.risk_flags)
no_execute = bool(_NO_EXECUTE_FLAGS & set(risk_flags))
target_mode = resolve_decision_target_mode(decision)
total_equity_value = runtime_metadata.get("portfolio_total_equity")
if not no_execute and target_mode == "value" and total_equity_value is not None:
total_equity = float(total_equity_value)
if total_equity <= 0.0:
no_execute = True
risk_flags = tuple(dict.fromkeys((*risk_flags, "no_execute")))
diagnostics.setdefault("execution_blocked_reason", "non_positive_total_equity")
diagnostics.setdefault("portfolio_total_equity", total_equity)
normalized_decision = decision if no_execute else _normalize_to_weight_decision(decision, runtime_metadata)
target_weights = None if no_execute else _derive_target_weights(normalized_decision)
allocation_payload = None
Expand Down
32 changes: 32 additions & 0 deletions tests/test_decision_mapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -105,3 +105,35 @@ def test_map_strategy_decision_translates_value_targets_for_semiconductor_strate
assert metadata["allocation"]["target_mode"] == "weight"
assert metadata["allocation"]["strategy_symbols"] == ("SOXL", "SOXX", "QQQI", "SPYI", "BOXX")
assert metadata["allocation"]["targets"]["SOXL"] == 0.6


def test_map_strategy_decision_no_executes_value_targets_when_total_equity_zero():
decision = StrategyDecision(
positions=(
PositionTarget(symbol="SOXL", target_value=30000.0),
PositionTarget(symbol="BOXX", target_value=15000.0, role="safe_haven"),
),
diagnostics={
"signal_description": "risk on",
"status_description": "soxl>ma150",
},
)

target_weights, signal_desc, is_emergency, status_desc, metadata = map_strategy_decision(
decision,
strategy_profile="soxl_soxx_trend_income",
runtime_metadata={
"portfolio_total_equity": 0.0,
"status_icon": "blocked",
"managed_symbols": ("SOXL", "BOXX"),
},
)

assert target_weights is None
assert signal_desc == "risk on"
assert is_emergency is False
assert status_desc == "soxl>ma150"
assert metadata["actionable"] is False
assert metadata["risk_flags"] == ("no_execute",)
assert metadata["execution_blocked_reason"] == "non_positive_total_equity"
assert "allocation" not in metadata