"""Snapshot selection + masking + view composition (REQ-029, REQ-028).""" from __future__ import annotations from datetime import date from trex_core.application.portfolio import ( DegradedReason, InMemoryPauseStore, PortfolioItem, SnapshotStore, View, build_snapshot_envelope, ) from trex_core.domain.slot import SlotService, SlotState from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository DAY = date(2026, 6, 5) def _universe(): return [ PortfolioItem(portfolio_id="pf-A", symbol="005930", cash_balance=1_000_000, asset_quantity=0, active_strategy_id="target-value-v1"), PortfolioItem(portfolio_id="pf-B", symbol="000660", cash_balance=2_000_000, asset_quantity=10, active_strategy_id="trend-follow-v1"), PortfolioItem(portfolio_id="pf-C", symbol="035720", cash_balance=3_000_000, asset_quantity=0, active_strategy_id="target-value-v1"), ] def _wire(*, slots_setup=None): store = SnapshotStore() store.swap(_universe()) pause = InMemoryPauseStore() repo = InMemorySlotRepository() svc = SlotService(repo) svc.preallocate_daily_slots(DAY, 5) if slots_setup: slots_setup(svc, repo) return store, pause, repo, svc def _take_slot(svc, repo, *, portfolio_id, symbol, strategy_id, status): s = repo.find_idle_slot(DAY) assert s is not None svc.transit_slot( s.slot_id, SlotState.IDLE, SlotState.PENDING_BUY, reason="buy_intent", actor="t", symbol=symbol, portfolio_id=portfolio_id, strategy_id=strategy_id, qty=10, ) if status is SlotState.PENDING_BUY: return s.slot_id svc.transit_slot( s.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, reason="filled", actor="t", avg_price=100.0, qty=10, ) return s.slot_id def _env(store, pause, repo, **kw): """Helper: always pin trading_date to the test's hardcoded DAY.""" return build_snapshot_envelope( store=store, pause_store=pause, slot_repo=repo, trading_date=DAY, **kw, ) def test_view_all_includes_holding_excludes_pending_for_same_symbol(): def setup(svc, repo): _take_slot(svc, repo, portfolio_id="pf-A", symbol="005930", strategy_id="target-value-v1", status=SlotState.HOLDING) _take_slot(svc, repo, portfolio_id="pf-B", symbol="000660", strategy_id="trend-follow-v1", status=SlotState.PENDING_BUY) store, pause, repo, svc = _wire(slots_setup=setup) env = _env(store, pause, repo, available_slots=3, view=View.ALL) pids = {a["asset_id"] for a in env.assets} assert "pf-A" in pids # holding -> included assert "pf-B" not in pids # pending -> masked assert "pf-C" in pids # universe-only -> included def test_view_buyable_with_zero_available_slots_returns_empty(): store, pause, repo, svc = _wire() env = _env(store, pause, repo, available_slots=0, view=View.BUYABLE) assert env.assets == [] assert env.available_slots == 0 def test_view_buyable_excludes_occupied_portfolios(): def setup(svc, repo): _take_slot(svc, repo, portfolio_id="pf-A", symbol="005930", strategy_id="target-value-v1", status=SlotState.PENDING_BUY) _take_slot(svc, repo, portfolio_id="pf-B", symbol="000660", strategy_id="trend-follow-v1", status=SlotState.HOLDING) store, pause, repo, svc = _wire(slots_setup=setup) env = _env(store, pause, repo, available_slots=3, view=View.BUYABLE) pids = {a["asset_id"] for a in env.assets} assert pids == {"pf-C"} # A occupied (pending), B occupied (holding) def test_view_held_returns_only_holding_slot_assets(): def setup(svc, repo): _take_slot(svc, repo, portfolio_id="pf-A", symbol="005930", strategy_id="target-value-v1", status=SlotState.HOLDING) _take_slot(svc, repo, portfolio_id="pf-B", symbol="000660", strategy_id="trend-follow-v1", status=SlotState.PENDING_BUY) store, pause, repo, svc = _wire(slots_setup=setup) env = _env(store, pause, repo, available_slots=3, view=View.HELD) pids = {a["asset_id"] for a in env.assets} assert pids == {"pf-A"} def test_strategy_key_orthogonal_filter(): store, pause, repo, svc = _wire() env = _env(store, pause, repo, available_slots=3, view=View.ALL, strategy_key="target-value-v1") assert {a["active_strategy_id"] for a in env.assets} == {"target-value-v1"} def test_strategy_key_missing_includes_everything_unmasked(): store, pause, repo, svc = _wire() env = _env(store, pause, repo, available_slots=3, view=View.ALL, strategy_key=None) assert {a["asset_id"] for a in env.assets} == {"pf-A", "pf-B", "pf-C"} def test_warning_overlay_masks_asset_uniformly_across_views(): store, pause, repo, svc = _wire() store.set_warning("pf-A", True) for view in (View.ALL, View.BUYABLE): env = _env(store, pause, repo, available_slots=3, view=view) pids = {a["asset_id"] for a in env.assets} assert "pf-A" not in pids def test_pause_overlay_masks_asset(): store, pause, repo, svc = _wire() pause.pause("pf-B") env = _env(store, pause, repo, available_slots=3, view=View.ALL) assert all(a["asset_id"] != "pf-B" for a in env.assets) def test_envelope_shape_no_price_field(): store, pause, repo, svc = _wire() env = _env(store, pause, repo, available_slots=3, view=View.ALL) d = env.to_dict() assert set(d.keys()) == {"assets", "view", "available_slots", "degraded", "reason", "message"} assert isinstance(d["assets"], list) for asset in d["assets"]: assert set(asset.keys()) == {"symbol", "asset_id", "cash_balance", "asset_quantity", "active_strategy_id", "strategy_params"} assert "price" not in asset def test_empty_envelope_assets_is_list_not_null(): store, pause, repo, svc = _wire() pause.pause("pf-A") pause.pause("pf-B") pause.pause("pf-C") env = _env(store, pause, repo, available_slots=3, view=View.ALL) d = env.to_dict() assert d["assets"] == [] assert d["assets"] is not None def test_degraded_signal_in_envelope(): store, pause, repo, svc = _wire() store.set_degraded(DegradedReason.PORTFOLIO_HUB_UNAVAILABLE, "auth.daemon get failed (Bearer abc123)") env = _env(store, pause, repo, available_slots=3, view=View.ALL) d = env.to_dict() assert d["degraded"] is True assert d["reason"] == "portfolio_hub_unavailable" # sanitized — credentials stripped assert "Bearer abc123" not in d["message"] assert "" in d["message"] or "auth" in d["message"]