"""dcy-020 / REQ-062 (BAT-029): core holding ↔ broker divergence reconcile.""" from __future__ import annotations from dataclasses import replace from trex_core.application.broker_port import BalanceItem from trex_core.application.portfolio import InMemoryReconcileWarningStore from trex_core.application.reconciliation import reconcile_core_holdings from trex_core.domain.slot import SlotService, SlotState from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository INST = "inst-1" class FakeBroker: def __init__(self, balances): self._b = balances def name(self): # noqa: D401 return "fake" def submit_order(self, req): ... def poll_fill(self, *a, **k): ... def cancel_order(self, *a, **k): ... def get_balance(self): return list(self._b) def _hold(repo, svc, pid, symbol, qty): repo.reconcile_universe(INST, {pid: "SYM"}) slot = repo.find_slot_by_portfolio(INST, pid) svc.transit_slot(slot.slot_id, slot.status, SlotState.PENDING_BUY, reason="r", actor="a", symbol=symbol, qty=qty) svc.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, reason="r", actor="a", avg_price=70_000) return repo.find_slot_by_portfolio(INST, pid) def _wire(): repo = InMemorySlotRepository() return repo, SlotService(repo, instance_key=INST) class TrackingRepository(InMemorySlotRepository): """Records writes so reconciliation-only classification stays read-only.""" def __init__(self): super().__init__() self.update_calls = 0 def update_slot(self, slot): self.update_calls += 1 return super().update_slot(slot) def test_A_core_holding_broker_flat_goes_error(): repo, svc = _wire() slot = _hold(repo, svc, "pf-a", "005930", 10) broker = FakeBroker([]) # broker holds nothing d = reconcile_core_holdings(slot_repo=repo, slot_service=svc, broker=broker, instance_key=INST) assert d[str(slot.slot_id)] == "error" assert repo.fetch_slot(slot.slot_id).status is SlotState.ERROR # quarantined def test_C_qty_mismatch_warns_no_state_change(): repo, svc = _wire() slot = _hold(repo, svc, "pf-a", "005930", 10) broker = FakeBroker([BalanceItem(symbol="005930", hldg_qty=7)]) # mismatch d = reconcile_core_holdings(slot_repo=repo, slot_service=svc, broker=broker, instance_key=INST) assert d[str(slot.slot_id)] == "qty_warning" assert repo.fetch_slot(slot.slot_id).status is SlotState.HOLDING # unchanged def test_ok_when_qty_matches(): repo, svc = _wire() slot = _hold(repo, svc, "pf-a", "005930", 10) broker = FakeBroker([BalanceItem(symbol="005930", hldg_qty=10)]) d = reconcile_core_holdings(slot_repo=repo, slot_service=svc, broker=broker, instance_key=INST) assert d[str(slot.slot_id)] == "ok" def test_B_broker_holds_unmanaged_warns(): repo, svc = _wire() broker = FakeBroker([BalanceItem(symbol="000660", hldg_qty=5)]) # core has none d = reconcile_core_holdings(slot_repo=repo, slot_service=svc, broker=broker, instance_key=INST) assert d["broker:000660"] == "unmanaged_position" def test_broker_held_symbol_with_same_instance_non_holding_slot_is_read_only_divergence(): repo = TrackingRepository() svc = SlotService(repo, instance_key=INST) warn_store = InMemoryReconcileWarningStore() repo.reconcile_universe(INST, {"pf-a": "000660"}) slot = repo.find_slot_by_portfolio(INST, "pf-a") assert slot is not None writes_before = repo.update_calls decisions = reconcile_core_holdings( slot_repo=repo, slot_service=svc, broker=FakeBroker([BalanceItem(symbol="000660", hldg_qty=5)]), instance_key=INST, warn_store=warn_store, ) assert decisions == {"broker:000660": "managed_non_holding_divergence"} assert repo.fetch_slot(slot.slot_id) == slot assert repo.update_calls == writes_before assert warn_store.list_warned() == set() assert repo.list_slots(INST) == [slot] def test_broker_held_symbol_with_same_instance_error_slot_is_read_only_divergence(): repo = TrackingRepository() svc = SlotService(repo, instance_key=INST) warn_store = InMemoryReconcileWarningStore() repo.reconcile_universe(INST, {"pf-a": "000660"}) inactive = repo.find_slot_by_portfolio(INST, "pf-a") assert inactive is not None error = replace(inactive, status=SlotState.ERROR) repo.update_slot(error) writes_before = repo.update_calls decisions = reconcile_core_holdings( slot_repo=repo, slot_service=svc, broker=FakeBroker([BalanceItem(symbol="000660", hldg_qty=5)]), instance_key=INST, warn_store=warn_store, ) assert decisions == {"broker:000660": "managed_non_holding_divergence"} assert repo.fetch_slot(error.slot_id) == error assert repo.update_calls == writes_before assert warn_store.list_warned() == set() def test_other_instance_only_managed_slot_remains_unmanaged_position_without_writes(): repo = TrackingRepository() svc = SlotService(repo, instance_key=INST) warn_store = InMemoryReconcileWarningStore() repo.reconcile_universe("other-instance", {"pf-a": "000660"}) other = repo.find_slot_by_portfolio("other-instance", "pf-a") assert other is not None writes_before = repo.update_calls decisions = reconcile_core_holdings( slot_repo=repo, slot_service=svc, broker=FakeBroker([BalanceItem(symbol="000660", hldg_qty=5)]), instance_key=INST, warn_store=warn_store, ) assert decisions == {"broker:000660": "unmanaged_position"} assert repo.fetch_slot(other.slot_id) == other assert repo.update_calls == writes_before assert warn_store.list_warned() == set() assert repo.list_slots(INST) == []