feat: capital-temperature position sizing (shadow-default)

decision_engine/config.py

@@ -117,6 +117,19 @@ class Settings(BaseServiceSettings):
         description="Publish checklist-BLOCKED BUY signals as non-actionable alerts (vs suppress)"
     )
 
+    # Capital-temperature position sizing: scale BUY size by the composite
+    # capital-temperature (0..1) from context-service. When disabled (shadow),
+    # the would-be multiplier is still computed and logged but NOT applied —
+    # mirrors feedback_accuracy_enabled. See market_context.get_capital_temperature.
+    capital_temperature_sizing_enabled: bool = Field(
+        False,
+        description="Apply capital-temperature size scaling (False = shadow: compute + log only)"
+    )
+    capital_temperature_max_reduction: float = Field(
+        0.50,
+        description="Max fractional size cut at full stress (temp=1.0): multiplier = 1 - temp*this"
+    )
+
     # Risk engine settings
     risk_engine_enabled: bool = Field(
         True,

decision_engine/market_context.py

@@ -77,6 +77,10 @@ def __init__(
         self._client: Optional[redis.Redis] = None
         self._regime: str = "UNKNOWN"
         self._regime_confidence: float = 0.0
+        # Composite capital-temperature (0.0 risk-on … 1.0 stressed), published by
+        # context-service inside market:context. 0.0 = no size reduction (fail-safe
+        # default when the field is absent).
+        self._capital_temperature: float = 0.0
         self._context_updated_at: Optional[float] = None  # epoch seconds from context payload
         self._lock = threading.RLock()
         self._thread: Optional[threading.Thread] = None
@@ -156,6 +160,15 @@ def get_regime_confidence(self) -> float:
         with self._lock:
             return self._regime_confidence
 
+    def get_capital_temperature(self) -> float:
+        """Return the composite capital-temperature (0.0 risk-on … 1.0 stressed).
+
+        Returns 0.0 (no size reduction) when context-service has not published a
+        temperature — a fail-safe: absence of the signal must not shrink sizing.
+        """
+        with self._lock:
+            return self._capital_temperature
+
     def get_multiplier(self, signal_type: str) -> float:
         """
         Return the confidence multiplier for a given signal type.
@@ -214,6 +227,16 @@ def _refresh(self) -> None:
             new_regime = str(data.get("regime", "UNKNOWN")).upper()
             new_confidence = float(data.get("regime_confidence", 0.0))
 
+            # Composite capital-temperature (nested object with a "value" 0..1).
+            # Absent/malformed → 0.0 (no size reduction).
+            new_temperature = 0.0
+            ct = data.get("capital_temperature")
+            if isinstance(ct, dict) and ct.get("value") is not None:
+                try:
+                    new_temperature = max(0.0, min(1.0, float(ct["value"])))
+                except (ValueError, TypeError):
+                    new_temperature = 0.0
+
             # Extract updated_at for staleness tracking.
             # context-service publishes this as an ISO timestamp string.
             new_updated_at: Optional[float] = None
@@ -242,4 +265,5 @@ def _refresh(self) -> None:
                 )
             self._regime = new_regime
             self._regime_confidence = new_confidence
+            self._capital_temperature = new_temperature
             self._context_updated_at = new_updated_at

decision_engine/service.py

@@ -417,10 +417,12 @@ def handle_indicator_event(self, event: dict):
                     and aggregated_signal.signal_type == SignalType.BUY
                 ):
                     try:
-                        # Fetch tier-based position size multiplier
+                        # Fetch tier-based position size multiplier, then fold in
+                        # the composite capital-temperature scaling (shadow-safe).
                         psm = 1.0
                         if self.tier_reader:
                             psm = self.tier_reader.get_position_size_multiplier(symbol)
+                        psm = self._apply_capital_temperature(symbol, psm)
                         trade_plan = self.trade_plan_engine.generate(
                             aggregated_signal, indicators,
                             position_size_multiplier=psm,
@@ -975,6 +977,31 @@ def _get_sector_for_symbol(self, symbol: str) -> Optional[str]:
                     self._symbol_to_sector[sym] = sector
         return self._symbol_to_sector.get(symbol)
 
+    def _apply_capital_temperature(self, symbol: str, tier_mult: float) -> float:
+        """Scale the position-size multiplier by the composite capital-temperature.
+
+        temp (0..1, 0=risk-on … 1=stressed) → temp_mult = 1 - temp*max_reduction.
+        The result is combined with the tier multiplier and re-clamped to 2% by the
+        trade planner. Always logs the would-be effect; only APPLIES it when
+        capital_temperature_sizing_enabled is True — otherwise it is shadow/log-only.
+        """
+        if self.market_context_reader is None:
+            return tier_mult
+        temp = self.market_context_reader.get_capital_temperature()
+        if temp <= 0.0:
+            return tier_mult  # risk-on, or no temperature published → no change
+
+        max_reduction = getattr(self.settings, "capital_temperature_max_reduction", 0.50)
+        temp_mult = 1.0 - temp * max_reduction
+        combined = tier_mult * temp_mult
+        enabled = getattr(self.settings, "capital_temperature_sizing_enabled", False)
+        logger.info(
+            f"Capital-temperature sizing {symbol}: temp={temp:.2f} → "
+            f"×{temp_mult:.2f} (tier ×{tier_mult:.2f} → combined ×{combined:.2f}) "
+            f"[{'APPLIED' if enabled else 'shadow'}]"
+        )
+        return combined if enabled else tier_mult
+
     def _should_publish(self, symbol: str, signal: AggregatedSignal) -> bool:
         """Check if we should publish this signal."""
         # Check confidence threshold — per-symbol override if configured

tests/test_capital_temperature_sizing.py

@@ -0,0 +1,60 @@
+"""
+Tests for capital-temperature position-size scaling.
+
+_apply_capital_temperature folds the composite capital-temperature (0..1) into the
+tier position-size multiplier: temp_mult = 1 - temp*max_reduction, combined =
+tier*temp_mult. It ALWAYS logs the would-be effect but only APPLIES it when
+capital_temperature_sizing_enabled is True (shadow otherwise).
+"""
+
+import unittest
+from unittest.mock import MagicMock
+
+from decision_engine.service import DecisionEngineService
+from decision_engine.config import Settings
+
+
+def _svc(enabled: bool, temp: float, max_reduction: float = 0.50) -> DecisionEngineService:
+    settings = MagicMock(spec=Settings)
+    settings.capital_temperature_sizing_enabled = enabled
+    settings.capital_temperature_max_reduction = max_reduction
+    svc = DecisionEngineService(settings)
+    svc.market_context_reader = MagicMock()
+    svc.market_context_reader.get_capital_temperature.return_value = temp
+    return svc
+
+
+class TestCapitalTemperatureSizing(unittest.TestCase):
+    def test_shadow_does_not_change_multiplier(self):
+        # Flag off: full stress, but the multiplier is unchanged (log-only).
+        svc = _svc(enabled=False, temp=1.0)
+        self.assertEqual(svc._apply_capital_temperature("AAPL", 1.0), 1.0)
+
+    def test_enabled_applies_reduction(self):
+        # Full stress → 1 - 1.0*0.50 = 0.50.
+        svc = _svc(enabled=True, temp=1.0)
+        self.assertAlmostEqual(svc._apply_capital_temperature("AAPL", 1.0), 0.50)
+
+    def test_enabled_combines_with_tier(self):
+        # temp 0.5 → temp_mult 0.75; tier 1.15 → combined.
+        svc = _svc(enabled=True, temp=0.5)
+        self.assertAlmostEqual(svc._apply_capital_temperature("X", 1.15), 1.15 * 0.75)
+
+    def test_zero_temperature_is_noop(self):
+        svc = _svc(enabled=True, temp=0.0)
+        self.assertEqual(svc._apply_capital_temperature("X", 1.15), 1.15)
+
+    def test_max_reduction_respected(self):
+        svc = _svc(enabled=True, temp=1.0, max_reduction=0.30)
+        self.assertAlmostEqual(svc._apply_capital_temperature("X", 1.0), 0.70)
+
+    def test_no_reader_is_noop(self):
+        settings = MagicMock(spec=Settings)
+        settings.capital_temperature_sizing_enabled = True
+        svc = DecisionEngineService(settings)
+        svc.market_context_reader = None
+        self.assertEqual(svc._apply_capital_temperature("X", 1.0), 1.0)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_market_context.py

@@ -149,6 +149,35 @@ def test_refresh_sideways_updates_regime(self):
         reader = self._refresh_with({"regime": "SIDEWAYS", "regime_confidence": 0.72})
         self.assertEqual(reader.get_regime(), "SIDEWAYS")
 
+    # --- capital_temperature parsing ---
+
+    def test_default_capital_temperature_is_zero(self):
+        self.assertEqual(_make_reader().get_capital_temperature(), 0.0)
+
+    def test_refresh_parses_capital_temperature(self):
+        reader = self._refresh_with({
+            "regime": "BEAR", "regime_confidence": 0.8,
+            "capital_temperature": {"value": 0.64, "direction": "HEATING"},
+        })
+        self.assertAlmostEqual(reader.get_capital_temperature(), 0.64)
+
+    def test_refresh_absent_capital_temperature_is_zero(self):
+        reader = self._refresh_with({"regime": "BULL", "regime_confidence": 0.9})
+        self.assertEqual(reader.get_capital_temperature(), 0.0)
+
+    def test_refresh_capital_temperature_clamped(self):
+        self.assertEqual(
+            self._refresh_with({"capital_temperature": {"value": 1.7}}).get_capital_temperature(), 1.0)
+        self.assertEqual(
+            self._refresh_with({"capital_temperature": {"value": -0.3}}).get_capital_temperature(), 0.0)
+
+    def test_refresh_malformed_capital_temperature_is_zero(self):
+        # Non-numeric value, and wrong-shaped (non-dict) field → 0.0, no crash.
+        self.assertEqual(
+            self._refresh_with({"capital_temperature": {"value": "hot"}}).get_capital_temperature(), 0.0)
+        self.assertEqual(
+            self._refresh_with({"capital_temperature": 0.5}).get_capital_temperature(), 0.0)
+
     def test_refresh_none_key_leaves_regime_unchanged(self):
         """If the Redis key doesn't exist yet, regime must stay at its previous value."""
         mock_redis = MagicMock()