"""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 trex_core.application.broker_port import BrokerPort from trex_core.application.portfolio import SnapshotStore _log = logging.getLogger(__name__) 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