Add allocation intent metadata for IBKR

Author: Pigbibi

decision_mapper.py

@@ -3,7 +3,11 @@
 from collections.abc import Mapping
 from typing import Any
 
-from quant_platform_kit.strategy_contracts import StrategyDecision
+from quant_platform_kit.strategy_contracts import (
+    StrategyDecision,
+    build_allocation_intent,
+    build_allocation_payload,
+)
 
 
 _EMERGENCY_FLAGS = frozenset({"emergency", "hard_defense"})
@@ -25,10 +29,15 @@ def _derive_target_weights(decision: StrategyDecision) -> dict[str, float]:
 def _derive_managed_symbols(
     decision: StrategyDecision,
     runtime_metadata: Mapping[str, Any],
+    *,
+    allocation_payload: Mapping[str, Any] | None = None,
 ) -> tuple[str, ...]:
     explicit = runtime_metadata.get("managed_symbols")
     if explicit:
         return tuple(str(symbol) for symbol in explicit)
+    allocation_symbols = allocation_payload.get("strategy_symbols") if allocation_payload else None
+    if allocation_symbols:
+        return tuple(str(symbol) for symbol in allocation_symbols)
     return tuple(position.symbol for position in decision.positions)
 
 
@@ -93,18 +102,33 @@ def map_strategy_decision(
     risk_flags = tuple(str(flag) for flag in decision.risk_flags)
     no_execute = bool(_NO_EXECUTE_FLAGS & set(risk_flags))
     target_weights = None if no_execute else _derive_target_weights(decision)
+    allocation_payload = None
+    if not no_execute and decision.positions:
+        allocation_payload = build_allocation_payload(
+            build_allocation_intent(
+                decision,
+                strategy_profile=strategy_profile,
+                strategy_symbols_order="risk_safe_income",
+            )
+        )
     signal_desc = _derive_signal_description(decision, runtime_metadata)
     status_desc = _derive_status_description(decision, runtime_metadata)
     is_emergency = bool(_EMERGENCY_FLAGS & set(risk_flags))
 
     metadata: dict[str, Any] = {**runtime_metadata, **diagnostics}
     metadata.setdefault("strategy_profile", strategy_profile)
     metadata.setdefault("status_icon", "🐤")
-    metadata.setdefault("managed_symbols", _derive_managed_symbols(decision, runtime_metadata))
+    metadata.setdefault(
+        "managed_symbols",
+        _derive_managed_symbols(decision, runtime_metadata, allocation_payload=allocation_payload),
+    )
     safe_haven_symbol = _derive_safe_haven_symbol(decision, runtime_metadata)
     if safe_haven_symbol:
         metadata.setdefault("safe_haven_symbol", safe_haven_symbol)
     metadata.setdefault("risk_flags", risk_flags)
     metadata.setdefault("actionable", not no_execute)
+    if allocation_payload:
+        metadata.setdefault("target_mode", allocation_payload["target_mode"])
+        metadata.setdefault("allocation", allocation_payload)
 
     return target_weights, signal_desc, is_emergency, status_desc, metadata

requirements.txt

@@ -1,7 +1,7 @@
 flask
 gunicorn
-quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@v0.7.5  # quote fallback hotfix
-us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@v0.7.5
+quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@v0.7.7
+us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@v0.7.7
 pandas
 numpy
 requests

tests/test_decision_mapper.py

@@ -28,6 +28,11 @@ def test_map_strategy_decision_maps_weight_positions_and_safe_haven():
     assert metadata["safe_haven_symbol"] == "BOXX"
     assert metadata["managed_symbols"] == ("AAA", "BOXX")
     assert metadata["status_icon"] == "🧲"
+    assert metadata["target_mode"] == "weight"
+    assert metadata["allocation"]["target_mode"] == "weight"
+    assert metadata["allocation"]["strategy_symbols"] == ("AAA", "BOXX")
+    assert metadata["allocation"]["targets"] == {"AAA": 0.6, "BOXX": 0.4}
+    assert metadata["allocation"]["positions"][1]["role"] == "safe_haven"
 
 
 def test_map_strategy_decision_returns_noop_when_flagged_no_execute():
@@ -55,3 +60,4 @@ def test_map_strategy_decision_returns_noop_when_flagged_no_execute():
     assert status_desc == "fail_closed | reason=feature_snapshot_path_missing"
     assert metadata["actionable"] is False
     assert metadata["snapshot_guard_decision"] == "fail_closed"
+    assert "allocation" not in metadata