feat(portfolio): introduce SleeveId as allocation key with instrument-aware policy lookup (RAD-67)

- Added SleeveId NewType and sleeve_name field on StrategySpec
- RiskAllocationPolicy.allocate() signature changed from list[SleeveId] to
  Mapping[SleeveId, Instrument] so policies resolve actual instrument history
- PortfolioRunner._apply_risk_budgets() builds sleeve->instrument mapping
- PortfolioRunner.strategies keyed by SleeveId; on_candle_close fans out to
  all sleeves matching the instrument
- Reconciliation derives epic from str(instrument), not sleeve key
- All portfolio tests updated

Co-Authored-By: Paperclip <[email protected]>

Author: 2 people

tests/portfolio/test_crash_recovery.py

@@ -20,7 +20,13 @@
 
 from tradedesk import Direction
 from tradedesk.execution import BrokerPosition, PositionTracker
-from tradedesk.portfolio import Instrument, JournalEntry, PositionJournal, ReconciliationManager
+from tradedesk.portfolio import (
+    Instrument,
+    JournalEntry,
+    PositionJournal,
+    ReconciliationManager,
+    SleeveId,
+)
 from tradedesk.types import Candle
 
 # ---------------------------------------------------------------------------
@@ -61,6 +67,7 @@ class _FakeStrategy:
 
     def __init__(self, epic: str = "") -> None:
         self.epic = epic
+        self.instrument = Instrument(epic)
         self.position = PositionTracker()
         self.entry_atr: float = 0.0
         self._on_position_change: Callable[[str], None] | None = None
@@ -101,7 +108,7 @@ def _build_manager(
 ) -> ReconciliationManager:
     if client is None:
         client = AsyncMock()
-    strategies = {Instrument(e): _FakeStrategy(e) for e in epics}
+    strategies = {SleeveId(e): _FakeStrategy(e) for e in epics}
     runner = MagicMock()
     runner.strategies = strategies
     mgr = ReconciliationManager(
@@ -111,13 +118,13 @@ def _build_manager(
         target_period="HOUR",
         enable_event_subscription=False,
     )
-    for inst, strat in strategies.items():
+    for strat in strategies.values():
         strat._on_position_change = mgr.persist_positions
     return mgr
 
 
 def _strat(mgr: ReconciliationManager, epic: str) -> _FakeStrategy:
-    return mgr._runner.strategies[Instrument(epic)]  # type: ignore[return-value]
+    return mgr._runner.strategies[SleeveId(epic)]  # type: ignore[return-value]
 
 
 @pytest.fixture

tests/portfolio/test_orchestrator_reconciliation.py

@@ -10,7 +10,7 @@
 from tradedesk import Direction
 from tradedesk.execution import BrokerPosition
 from tradedesk.execution.position import PositionTracker
-from tradedesk.portfolio import Instrument, JournalEntry, PositionJournal
+from tradedesk.portfolio import Instrument, JournalEntry, PositionJournal, SleeveId
 from tradedesk.portfolio.reconciliation import ReconciliationManager
 from tradedesk.types import Candle
 
@@ -68,6 +68,7 @@ class _FakeStrategy:
     def __init__(self, client=None, epic="", period="", **kwargs):
         self.client = client
         self.epic = epic
+        self.instrument = Instrument(epic)
         self.position = PositionTracker()
         self._on_position_change = None
         self.entry_atr = 0.0
@@ -126,7 +127,7 @@ def _build_manager(epics, *, journal, client=None):
     strategies = {}
     for epic in epics:
         strat = _FakeStrategy(client=client, epic=epic)
-        strategies[Instrument(epic)] = strat
+        strategies[SleeveId(epic)] = strat
 
     runner = MagicMock()
     runner.strategies = strategies
@@ -139,15 +140,15 @@ def _build_manager(epics, *, journal, client=None):
     )
 
     # Register position-change callbacks (mirrors real orchestrator wiring)
-    for inst, strat in strategies.items():
+    for strat in strategies.values():
         strat._on_position_change = mgr.persist_positions
 
     return mgr
 
 
 def _strat(mgr, epic):
     """Get strategy by epic string."""
-    return mgr._runner.strategies[Instrument(epic)]
+    return mgr._runner.strategies[SleeveId(epic)]
 
 
 @pytest.fixture

tests/portfolio/test_policy.py

@@ -1,21 +1,32 @@
 """Tests for portfolio risk allocation policies."""
 
 from tradedesk.portfolio.risk import EqualSplitRiskPolicy
-from tradedesk.portfolio.types import Instrument
+from tradedesk.portfolio.types import Instrument, SleeveId
 
 
-def test_equal_split_allocates_per_active_instrument():
-    """Test that EqualSplitRiskPolicy divides budget equally across active instruments."""
+def test_equal_split_allocates_per_active_sleeve():
+    """Test that EqualSplitRiskPolicy divides budget equally across active sleeves."""
     p = EqualSplitRiskPolicy(portfolio_risk_budget=10.0)
 
-    # No active instruments -> empty allocation
-    assert p.allocate([]) == {}
+    # No active sleeves -> empty allocation
+    assert p.allocate({}) == {}
 
-    # One active instrument -> gets full budget
-    a = p.allocate([Instrument("EURUSD")])
-    assert a[Instrument("EURUSD")] == 10.0
+    # One active sleeve -> gets full budget
+    a = p.allocate({SleeveId("AdaptiveFade_EURUSD"): Instrument("CS.D.EURUSD.TODAY.IP")})
+    assert a[SleeveId("AdaptiveFade_EURUSD")] == 10.0
 
-    # Two active instruments -> split equally
-    ab = p.allocate([Instrument("EURUSD"), Instrument("GBPUSD")])
-    assert ab[Instrument("EURUSD")] == 5.0
-    assert ab[Instrument("GBPUSD")] == 5.0
+    # Two active sleeves -> split equally
+    ab = p.allocate({
+        SleeveId("AdaptiveFade_EURUSD"): Instrument("CS.D.EURUSD.TODAY.IP"),
+        SleeveId("BollingerReversion_GBPUSD"): Instrument("CS.D.GBPUSD.TODAY.IP"),
+    })
+    assert ab[SleeveId("AdaptiveFade_EURUSD")] == 5.0
+    assert ab[SleeveId("BollingerReversion_GBPUSD")] == 5.0
+
+    # Two active sleeves on the same instrument -> each gets 5.0 independently
+    dual = p.allocate({
+        SleeveId("AdaptiveFade_AUDCAD"): Instrument("CS.D.AUDCAD.TODAY.IP"),
+        SleeveId("BollingerReversion_AUDCAD"): Instrument("CS.D.AUDCAD.TODAY.IP"),
+    })
+    assert dual[SleeveId("AdaptiveFade_AUDCAD")] == 5.0
+    assert dual[SleeveId("BollingerReversion_AUDCAD")] == 5.0

tests/portfolio/test_portfolio_runner.py

@@ -5,7 +5,7 @@
 from tradedesk.marketdata.events import CandleClosedEvent
 from tradedesk.portfolio.risk import EqualSplitRiskPolicy
 from tradedesk.portfolio.runner import PortfolioRunner
-from tradedesk.portfolio.types import Instrument
+from tradedesk.portfolio.types import Instrument, SleeveId
 
 
 class FakeStrategy:
@@ -33,16 +33,16 @@ async def evaluate_signals(self) -> None:
 
 @pytest.mark.asyncio
 async def test_runner_splits_risk_across_active_strategies():
-    """Test that PortfolioRunner splits risk budget across active strategies."""
+    """Test that PortfolioRunner splits risk budget across active sleeves."""
     s1 = FakeStrategy("EURUSD", active=True)
     s2 = FakeStrategy("GBPUSD", active=True)
     s3 = FakeStrategy("USDJPY", active=False)
 
     r = PortfolioRunner(
         strategies={
-            Instrument("EURUSD"): s1,
-            Instrument("GBPUSD"): s2,
-            Instrument("USDJPY"): s3,
+            SleeveId("s1"): s1,
+            SleeveId("s2"): s2,
+            SleeveId("s3"): s3,
         },
         policy=EqualSplitRiskPolicy(portfolio_risk_budget=10.0),
         default_risk_per_trade=10.0,
@@ -62,3 +62,39 @@ async def test_runner_splits_risk_across_active_strategies():
     assert s1.evaluate_signals_calls == 1
     assert s2.update_state_calls == 0
     assert s2.evaluate_signals_calls == 0
+
+
+@pytest.mark.asyncio
+async def test_runner_fans_out_to_dual_sleeve_same_instrument():
+    """Both sleeves on the same instrument receive the candle and share risk budget."""
+    fade = FakeStrategy("AUDCAD", active=True)
+    bollinger = FakeStrategy("AUDCAD", active=True)
+    other = FakeStrategy("EURUSD", active=False)
+
+    r = PortfolioRunner(
+        strategies={
+            SleeveId("AdaptiveFade_AUDCAD"): fade,
+            SleeveId("BollingerReversion_AUDCAD"): bollinger,
+            SleeveId("BollingerReversion_EURUSD"): other,
+        },
+        policy=EqualSplitRiskPolicy(portfolio_risk_budget=10.0),
+        default_risk_per_trade=10.0,
+    )
+
+    await r.on_candle_close(
+        CandleClosedEvent(instrument=Instrument("AUDCAD"), timeframe="15MINUTE", candle=None)
+    )
+
+    # Both AUDCAD sleeves active -> 5.0 each (2 active out of 3 sleeves)
+    assert fade._rpt == 5.0
+    assert bollinger._rpt == 5.0
+    # Inactive EURUSD sleeve gets default
+    assert other._rpt == 10.0
+    # Both AUDCAD sleeves should have processed the candle
+    assert fade.update_state_calls == 1
+    assert fade.evaluate_signals_calls == 1
+    assert bollinger.update_state_calls == 1
+    assert bollinger.evaluate_signals_calls == 1
+    # EURUSD sleeve should not have processed the AUDCAD candle
+    assert other.update_state_calls == 0
+    assert other.evaluate_signals_calls == 0

tests/portfolio/test_reconciliation.py

@@ -14,7 +14,7 @@
     _direction_matches,
     reconcile,
 )
-from tradedesk.portfolio.types import Instrument
+from tradedesk.portfolio.types import Instrument, SleeveId
 
 
 def _journal_entry(instrument="USDJPY", direction="long", size=1.0):
@@ -194,6 +194,7 @@ class TestReconciliationManager:
     @pytest.fixture
     def mock_strategy(self):
         strat = Mock()
+        strat.instrument = Instrument("USDJPY")
         strat.position = Mock()
         strat.position.is_flat.return_value = True
         strat.position.direction = None
@@ -209,8 +210,8 @@ def mock_strategy(self):
     @pytest.fixture
     def mock_runner(self, mock_strategy):
         runner = Mock()
-        # Populate with one default strategy
-        runner.strategies = {Instrument("USDJPY"): mock_strategy}
+        # Populate with one default strategy, keyed by SleeveId
+        runner.strategies = {SleeveId("USDJPY"): mock_strategy}
         return runner
 
     @pytest.fixture
@@ -260,7 +261,7 @@ async def test_startup_orphan_adoption(self, manager, mock_client, mock_runner):
         restored = await manager.reconcile_on_startup()
 
         assert "USDJPY" in restored
-        strat = mock_runner.strategies[Instrument("USDJPY")]
+        strat = mock_runner.strategies[SleeveId("USDJPY")]
         # Verify open called
         strat.position.open.assert_called_with(Direction.LONG, 1.0, 150.0)
         # Should save to persist the adoption
@@ -275,7 +276,7 @@ async def test_startup_broker_failure(self, manager, mock_client, mock_journal,
         restored = await manager.reconcile_on_startup()
 
         assert "USDJPY" in restored
-        strat = mock_runner.strategies[Instrument("USDJPY")]
+        strat = mock_runner.strategies[SleeveId("USDJPY")]
         strat.restore_from_journal.assert_called_once()
 
     @pytest.mark.asyncio
@@ -289,15 +290,15 @@ async def test_periodic_reconcile_correction(self, manager, mock_client, mock_ru
         # Force reconcile
         await manager.periodic_reconcile()
 
-        strat = mock_runner.strategies[Instrument("USDJPY")]
+        strat = mock_runner.strategies[SleeveId("USDJPY")]
         strat.position.open.assert_called_with(Direction.LONG, 1.0, 150.0)
         manager._journal.save.assert_called_once()
 
     @pytest.mark.asyncio
     async def test_post_warmup_check(self, manager, mock_client, mock_runner):
         """Verifies exit check on restored positions."""
         # Setup strategy to look like it has a position
-        strat = mock_runner.strategies[Instrument("USDJPY")]
+        strat = mock_runner.strategies[SleeveId("USDJPY")]
         strat.position.is_flat.return_value = False
 
         mock_client.get_historical_candles.return_value = [Mock(close=155.0)]

tradedesk/portfolio/__init__.py

@@ -19,6 +19,7 @@
     Instrument,
     PortfolioStrategy,
     ReconcilableStrategy,
+    SleeveId,
     StrategySpec,
 )
 
@@ -42,6 +43,7 @@
     "ReconciliationResult",
     "RiskAllocationPolicy",
     "SimplePortfolio",
+    "SleeveId",
     "StrategySpec",
     "WeightedRollingTracker",
     "atr_normalised_size",

tradedesk/portfolio/reconciliation.py

@@ -11,7 +11,7 @@
 from ..types import Direction
 from .journal import JournalEntry, PositionJournal
 from .runner import PortfolioRunner
-from .types import Instrument, ReconcilableStrategy
+from .types import ReconcilableStrategy
 
 log = logging.getLogger(__name__)
 
@@ -324,7 +324,7 @@ async def reconcile_on_startup(self) -> set[str]:
                 restored_instruments = self._restore_from_journal(journal_positions)
             return restored_instruments
 
-        managed_instruments = {str(inst) for inst in self._runner.strategies.keys()}
+        managed_instruments = {str(s.instrument) for s in self._runner.strategies.values()}
 
         result = reconcile(
             journal_positions=journal_positions,
@@ -398,9 +398,9 @@ async def reconcile_on_startup(self) -> set[str]:
     def _restore_from_journal(self, journal_positions: dict[str, JournalEntry]) -> set[str]:
         """Restore positions from journal only (broker unreachable)."""
         restored: set[str] = set()
-        for inst, s in self._runner.strategies.items():
+        for s in self._runner.strategies.values():
             strat = cast(ReconcilableStrategy, s)
-            epic = str(inst)
+            epic = str(s.instrument)
             entry = journal_positions.get(epic)
             if entry is None or entry.direction is None:
                 continue
@@ -428,9 +428,9 @@ def _apply_reconciliation(
         broker_by_instrument = {bp.instrument: bp for bp in broker_positions}
         restored: set[str] = set()
 
-        for inst, s in self._runner.strategies.items():
+        for s in self._runner.strategies.values():
             strat = cast(ReconcilableStrategy, s)
-            epic = str(inst)
+            epic = str(s.instrument)
             entry = next((e for e in result.entries if e.instrument == epic), None)
             if entry is None:
                 continue
@@ -511,9 +511,9 @@ async def post_warmup_check(self, restored_instruments: set[str]) -> None:
         or adopted might be stale (e.g. stop-loss breached, regime deactivated).
         This method checks each one and exits immediately if warranted.
         """
-        for inst, s in self._runner.strategies.items():
+        for s in self._runner.strategies.values():
             strat = cast(ReconcilableStrategy, s)
-            epic = str(inst)
+            epic = str(s.instrument)
             if epic not in restored_instruments:
                 continue
             if strat.position.is_flat():
@@ -558,9 +558,9 @@ def persist_positions(self, changed_epic: str = "") -> None:
             return
 
         entries = []
-        for inst, s in self._runner.strategies.items():
+        for s in self._runner.strategies.values():
             strat = cast(ReconcilableStrategy, s)
-            entries.append(strat.to_journal_entry(str(inst)))
+            entries.append(strat.to_journal_entry(str(s.instrument)))
 
         self._journal.save(entries)
 
@@ -578,12 +578,12 @@ async def periodic_reconcile(self) -> None:
 
         # Build current local state
         journal_positions: dict[str, JournalEntry] = {}
-        for inst, s in self._runner.strategies.items():
+        for s in self._runner.strategies.values():
             strat = cast(ReconcilableStrategy, s)
-            epic = str(inst)
+            epic = str(s.instrument)
             journal_positions[epic] = strat.to_journal_entry(epic)
 
-        managed_instruments = {str(inst) for inst in self._runner.strategies.keys()}
+        managed_instruments = {str(s.instrument) for s in self._runner.strategies.values()}
 
         # Exclude instruments with recent position changes to avoid settlement race
         skipped = self._recently_changed_instruments.copy()
@@ -613,7 +613,11 @@ async def periodic_reconcile(self) -> None:
             if entry.discrepancy == DiscrepancyType.MATCHED:
                 continue
 
-            maybe_strat = self._runner.strategies.get(Instrument(entry.instrument))
+            maybe_strat = next(
+                (s for s in self._runner.strategies.values()
+                 if str(s.instrument) == entry.instrument),
+                None,
+            )
             if maybe_strat is None:
                 continue
             strat = cast(ReconcilableStrategy, maybe_strat)