"""Position reconciliation between portfolio-hub and broker (REQ-031). The broker account is single (invariant B) so portfolio-hub quantities are aggregated to `symbol` before being compared with `BrokerPort.get_balance()`. Mismatches flip the asset's `warning` overlay; a subsequent successful match clears it automatically (in-memory only — pause is the durable channel for manual holds). """ from __future__ import annotations import logging from typing import Optional from trex_core.application.alarm_emitter import AlarmEmitter from trex_core.application.broker_port import BrokerPort from trex_core.application.portfolio import SnapshotStore from trex_core.domain.slot.state import SlotState _log = logging.getLogger(__name__) def reconcile_core_holdings( *, slot_repo, slot_service, broker: BrokerPort, instance_key: str, alarm_emitter: Optional[AlarmEmitter] = None, ) -> dict[str, str]: """dcy-020 / REQ-062 (BAT-029): surface core-slot holding ↔ broker divergence. Broker balance is the reconcile ground truth. Decisions (no silent auto-fix, no auto-liquidation — issue #15): - A. core holding but broker flat (qty 0) -> slot to `error` + WARN (operator resolve) - C. both hold but qty differs -> WARN (mirror is non-authoritative) - B. broker holds, no core holding slot -> WARN (unmanaged position) Returns {key: 'error'|'qty_warning'|'ok'|'unmanaged_warning'} for telemetry. (Liquidation qty = broker authoritative in execution dispatch is BAT-δ.)""" broker_qty: dict[str, int] = { item.symbol: int(item.hldg_qty) for item in broker.get_balance() } decisions: dict[str, str] = {} matched: set[str] = set() for slot in slot_repo.list_holding_slots(instance_key): sym = slot.symbol matched.add(sym) bq = broker_qty.get(sym, 0) if bq == 0: # A — phantom holding try: slot_service.transit_slot( slot.slot_id, SlotState.HOLDING, SlotState.ERROR, reason="reconcile_mismatch:broker_flat", actor="reconcile", ) decisions[str(slot.slot_id)] = "error" _log.warning( "core holding but broker flat: slot=%s symbol=%s -> error", slot.slot_id, sym, ) # dcy-033 / REQ-100: CRITICAL alarm — position mismatch. if alarm_emitter is not None: alarm_emitter.publish_alarm( "core_holding_broker_flat", f"슬롯 보유인데 브로커 잔고 0 (slot={slot.slot_id} symbol={sym})", fields={"slot_id": str(slot.slot_id), "symbol": sym}, ) except Exception: # noqa: BLE001 - never let reconcile crash the loop _log.exception("failed to error-quarantine slot %s", slot.slot_id) elif slot.qty is not None and int(slot.qty) != bq: # C — qty mismatch decisions[str(slot.slot_id)] = "qty_warning" _log.warning( "qty mismatch: slot=%s symbol=%s core=%s broker=%d (mirror non-authoritative)", slot.slot_id, sym, slot.qty, bq, ) else: decisions[str(slot.slot_id)] = "ok" for sym, bq in broker_qty.items(): # B — unmanaged broker position if bq > 0 and sym not in matched: decisions[f"broker:{sym}"] = "unmanaged_warning" _log.warning( "broker holds %s qty=%d with no core holding slot (unmanaged position)", sym, bq, ) return decisions def reconcile_positions(*, store: SnapshotStore, broker: BrokerPort) -> dict[str, str]: """Run one reconciliation pass. Returns a `{portfolio_id: 'warning' | 'cleared' | 'ok'}` map for telemetry. The map is also reflected onto `store` via `set_warning(...)`. """ broker_qty: dict[str, int] = { item.symbol: int(item.hldg_qty) for item in broker.get_balance() } # Group portfolio-hub quantities by symbol so we can compare to broker's # per-symbol aggregate. by_symbol: dict[str, list[tuple[str, float]]] = {} assets = store.list_assets() for a in assets: sym = a.item.symbol by_symbol.setdefault(sym, []).append((a.item.portfolio_id, a.item.asset_quantity)) decisions: dict[str, str] = {} for symbol, entries in by_symbol.items(): hub_total = int(sum(qty for _, qty in entries)) broker_total = broker_qty.get(symbol, 0) match = hub_total == broker_total for pid, _ in entries: asset = store.get(pid) was_warned = bool(asset and asset.warning) if not match: if not was_warned: _log.warning( "position warning raised: portfolio_id=%s symbol=%s " "hub=%d broker=%d", pid, symbol, hub_total, broker_total, ) store.set_warning(pid, True) decisions[pid] = "warning" else: if was_warned: store.set_warning(pid, False) decisions[pid] = "cleared" else: decisions[pid] = "ok" return decisions