213 lines
7.7 KiB
Python
213 lines
7.7 KiB
Python
"""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()
|