"""dcy-020 / REQ-062 (BAT-029): core holding ↔ broker divergence reconcile.""" from __future__ import annotations from dataclasses import replace from unittest.mock import Mock import pytest from trex_core.application.broker_port import BalanceItem, BrokerBalanceError 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_later_page_snapshot_symbol_keeps_holding_slot_ok(kis_balance_broker_factory): repo, svc = _wire() slot = _hold(repo, svc, "pf-later", "000660", 4) broker, calls = kis_balance_broker_factory([ {"rows": [], "tr_cont": "F", "fk": "A", "nk": "1"}, {"rows": [{"pdno": "000660", "hldg_qty": "4"}], "tr_cont": "E"}, ]) decisions = reconcile_core_holdings( slot_repo=repo, slot_service=svc, broker=broker, instance_key=INST, ) assert decisions == {str(slot.slot_id): "ok"} assert repo.fetch_slot(slot.slot_id).status is SlotState.HOLDING assert len(calls) == 2 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) == [] def test_reconcile_core_holdings_propagates_later_page_failure_without_write( kis_balance_broker_factory, monkeypatch ): repo = TrackingRepository() svc = SlotService(repo, instance_key=INST) slot = _hold(repo, svc, "pf-a", "005930", 10) writes_before = repo.update_calls warn_store = InMemoryReconcileWarningStore() set_warned = Mock(wraps=warn_store.set_warned) clear_warned = Mock(wraps=warn_store.clear_warned) monkeypatch.setattr(warn_store, "set_warned", set_warned) monkeypatch.setattr(warn_store, "clear_warned", clear_warned) broker, calls = kis_balance_broker_factory([ { "rows": [{"pdno": "005930", "hldg_qty": "10"}], "tr_cont": "F", "fk": "A", "nk": "1", }, {"rows": [], "tr_cont": "X"}, ]) with pytest.raises(BrokerBalanceError, match="malformed_response"): reconcile_core_holdings( slot_repo=repo, slot_service=svc, broker=broker, instance_key=INST, warn_store=warn_store, ) assert len(calls) == 2 assert repo.fetch_slot(slot.slot_id).status is SlotState.HOLDING assert repo.update_calls == writes_before set_warned.assert_not_called() clear_warned.assert_not_called()