"""Position reconciliation between portfolio-hub and broker (REQ-031, REQ-103/105/106). 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). BAT-050 / REQ-103: B-case (slot qty != broker qty) and C-case (ledger/hub qty != broker qty) now also set a durable reconcile-warning via ReconcileWarningStore (slot_id keyed). The in-memory warning overlay is kept for backward compat; the durable store is additive and Optional (None => warn-log-only, backward compat). REQ-103 case_code values in slot_reconcile_warning: 'B' — slot has a holding but slot.qty differs from broker qty (set by reconcile_core_holdings) 'C' — portfolio-hub (ledger) qty differs from broker qty (set by reconcile_positions) BAT-051 / REQ-105: sync_slot_to_broker — operator CLI sync path. BAT-051 / REQ-106: verify_slot_reconcile — operator CLI verify path (C-only, read-centric). """ 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 ReconcileWarningStore, 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, warn_store: Optional[ReconcileWarningStore] = None, ) -> dict[str, str]: """dcy-020 / REQ-062 (BAT-029) + REQ-103 (BAT-050): surface core-slot holding vs broker divergence and set durable reconcile-warning for B case. 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 + CRITICAL alarm - C. both hold but qty differs (B in REQ-103 naming) -> WARN + durable reconcile-warning - B. broker holds, no core holding slot -> WARN (unmanaged position) REQ-103 case_code values (stored in slot_reconcile_warning.case_code): - case_code='B': slot has a holding but slot.qty differs from broker qty → set by THIS function (reconcile_core_holdings) - case_code='C': portfolio-hub (ledger) qty differs from broker qty → set by reconcile_positions warn_store: Optional — when None, WARN-log-only (backward compat, REQ-103 AC-5). Returns {key: 'error'|'qty_warning'|'ok'|'unmanaged_position'| 'managed_non_holding_divergence'} for telemetry. """ 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: # B case (REQ-103) — qty mismatch slot_id_str = str(slot.slot_id) decisions[slot_id_str] = "qty_warning" # REQ-107: structured WARNING log for B case. _log.warning( "reconcile_warning set: slot_id=%s portfolio_id=%s symbol=%s " "case_code=B broker_qty=%d slot_qty=%s actor=reconcile", slot_id_str, slot.portfolio_id, sym, bq, slot.qty, ) # REQ-103 / BAT-050: durable reconcile-warning (case_code='B'). if warn_store is not None: try: warn_store.set_warned( slot_id_str, portfolio_id=slot.portfolio_id or "", case_code="B", actor="reconcile", reason=f"slot_qty={slot.qty} broker_qty={bq} symbol={sym}", ) except Exception: # noqa: BLE001 - never let reconcile crash the loop _log.exception( "failed to set reconcile_warning for slot %s", slot.slot_id ) else: decisions[str(slot.slot_id)] = "ok" for sym, bq in broker_qty.items(): # unmatched broker position if bq > 0 and sym not in matched: managed = slot_repo.find_managed_slot_by_symbol(instance_key, sym) if managed is not None: decisions[f"broker:{sym}"] = "managed_non_holding_divergence" _log.warning( "managed non-holding divergence: slot_id=%s portfolio_id=%s " "symbol=%s status=%s instance_key=%s", managed.slot_id, managed.portfolio_id, sym, managed.status.value, instance_key, ) else: decisions[f"broker:{sym}"] = "unmanaged_position" _log.warning( "unmanaged position: symbol=%s qty=%d instance_key=%s", sym, bq, instance_key, ) return decisions def reconcile_positions( *, store: SnapshotStore, broker: BrokerPort, slot_repo=None, instance_key: str = "default", warn_store: Optional[ReconcileWarningStore] = None, ) -> dict[str, str]: """Run one reconciliation pass (REQ-031 + REQ-103 BAT-050). Returns a `{portfolio_id: 'warning' | 'cleared' | 'ok'}` map for telemetry. The map is also reflected onto `store` via `set_warning(...)`. BAT-050 / REQ-103: when warn_store and slot_repo are provided, C-case mismatches (ledger/hub qty vs broker) also set durable reconcile-warning on the corresponding slot (portfolio_id -> slot lookup via slot_repo). When either is None, falls back to in-memory warning only (backward compat, REQ-103 AC-5). """ 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" # REQ-103 / BAT-050: durable reconcile-warning for C case (case_code='C'). if warn_store is not None and slot_repo is not None: try: slot = slot_repo.find_slot_by_portfolio(instance_key, pid) if slot is not None: slot_id_str = str(slot.slot_id) # REQ-107: structured WARNING log for C case. _log.warning( "reconcile_warning set: slot_id=%s portfolio_id=%s " "symbol=%s case_code=C hub_qty=%d broker_qty=%d actor=reconcile", slot_id_str, pid, symbol, hub_total, broker_total, ) warn_store.set_warned( slot_id_str, portfolio_id=pid, case_code="C", actor="reconcile", reason=( f"hub_qty={hub_total} broker_qty={broker_total} " f"symbol={symbol}" ), ) except Exception: # noqa: BLE001 _log.exception( "failed to set reconcile_warning for portfolio_id=%s", pid ) else: if was_warned: store.set_warning(pid, False) decisions[pid] = "cleared" else: decisions[pid] = "ok" return decisions # --------------------------------------------------------------------------- # BAT-051 / REQ-105: operator sync command — slot qty ← broker # --------------------------------------------------------------------------- def sync_slot_to_broker( *, slot_id: str, slot_repo, slot_service, broker: BrokerPort, instance_key: str, warn_store: "ReconcileWarningStore", cost_basis_port=None, # Optional[PortfolioHubCostBasisPort] dry_run: bool = False, ) -> dict: """Sync one slot's qty to the broker ground truth (REQ-105, BAT-051). Operator-only path (actor="sync"). No silent auto-fix — caller must confirm before invoking (CLI --yes gate). Returns a summary dict for before/after output. INACTIVE + broker qty > 0 → INACTIVE→HOLDING transition (OQ-A actor=sync). HOLDING + qty mismatch → qty field updated to broker qty. PENDING_* → skipped (in-flight, do not touch). Broker qty == 0 for slot → skipped with WARN (slot-side A-case guard). Broker-only (no slot) → NOT handled here; caller filters (FR-3). After update, if slot.qty == broker_qty: clear_warned(slot_id). """ from uuid import UUID try: slot_uuid = UUID(slot_id) except ValueError: return {"slot_id": slot_id, "error": "invalid_slot_id", "skipped": True} slot = slot_repo.fetch_slot(slot_uuid) if slot is None: return {"slot_id": slot_id, "error": "slot_not_found", "skipped": True} sym = slot.symbol or "" pid = slot.portfolio_id or "" # Fetch broker balance for this symbol broker_qty = 0 for item in broker.get_balance(): if item.symbol == sym: broker_qty = int(item.hldg_qty) break before = { "qty": float(slot.qty) if slot.qty is not None else None, "status": slot.status.value, "avg_price": float(slot.avg_price) if slot.avg_price is not None else None, } if broker_qty == 0: _log.warning( "sync skipped: broker_qty=0 for slot=%s symbol=%s (A-case guard — " "use SlotService.transit_slot or manual review)", slot_id, sym, ) return { "slot_id": slot_id, "symbol": sym, "portfolio_id": pid, "before": before, "after": before, "warning_cleared": False, "dry_run": dry_run, "skipped": True, "skip_reason": "broker_qty_zero", } from trex_core.domain.slot.state import SlotState if slot.status in (SlotState.PENDING_BUY, SlotState.PENDING_SELL): _log.warning( "sync skipped: slot=%s symbol=%s status=%s in-flight — " "cannot sync while order is pending", slot_id, sym, slot.status.value, ) return { "slot_id": slot_id, "symbol": sym, "portfolio_id": pid, "before": before, "after": before, "warning_cleared": False, "dry_run": dry_run, "skipped": True, "skip_reason": f"status_in_flight:{slot.status.value}", } # Try to seed avg_price from portfolio-hub cost-basis port (FR-7, graceful degrade). avg_price = slot.avg_price seeded_avg_price = False if cost_basis_port is not None and pid: try: wac = cost_basis_port.get_cost_basis(pid) if wac is not None: avg_price = wac seeded_avg_price = True except Exception: # noqa: BLE001 — graceful degrade _log.warning( "sync: cost_basis_port.get_cost_basis failed for portfolio_id=%s slot=%s " "— skipping avg_price seed (graceful degrade)", pid, slot_id, ) elif cost_basis_port is None and pid: _log.warning( "sync: no cost_basis_port configured for slot=%s portfolio_id=%s " "— skipping avg_price seed (graceful degrade)", slot_id, pid, ) if dry_run: # Compute what *would* happen without writing anything. new_status = SlotState.HOLDING if slot.status is SlotState.INACTIVE else slot.status after = { "qty": float(broker_qty), "status": new_status.value, "avg_price": float(avg_price) if avg_price is not None else None, } return { "slot_id": slot_id, "symbol": sym, "portfolio_id": pid, "before": before, "after": after, "warning_cleared": False, # dry-run never clears "dry_run": True, "skipped": False, "seeded_avg_price": seeded_avg_price, } # --- Live write --- try: if slot.status is SlotState.INACTIVE: # OQ-A: INACTIVE→HOLDING transition (actor=sync, reason=onboarding). slot_service.transit_slot( slot_uuid, SlotState.INACTIVE, SlotState.HOLDING, reason="onboarding", actor="sync", qty=float(broker_qty), avg_price=avg_price, portfolio_id=pid or None, symbol=sym or None, ) else: # HOLDING: update qty (and optionally avg_price) in place. # Use update_slot directly to avoid a state-machine transition. from dataclasses import replace as _replace updated = _replace(slot, qty=float(broker_qty), avg_price=avg_price) slot_repo.update_slot(updated) _log.info( "sync updated slot qty: slot=%s symbol=%s qty %s→%d actor=sync", slot_id, sym, slot.qty, broker_qty, ) except Exception: # noqa: BLE001 _log.exception("sync_slot_to_broker: write failed for slot=%s", slot_id) return { "slot_id": slot_id, "symbol": sym, "portfolio_id": pid, "before": before, "after": before, "warning_cleared": False, "dry_run": False, "skipped": False, "error": "write_failed", } # Refresh to confirm refreshed = slot_repo.fetch_slot(slot_uuid) after_qty = float(refreshed.qty) if (refreshed and refreshed.qty is not None) else None after_status = refreshed.status.value if refreshed else slot.status.value after_avg = float(refreshed.avg_price) if (refreshed and refreshed.avg_price is not None) else None after = {"qty": after_qty, "status": after_status, "avg_price": after_avg} # Clear warning when qty now matches broker warning_cleared = False if after_qty is not None and int(after_qty) == broker_qty: try: warn_store.clear_warned(slot_id) warning_cleared = True _log.info( "sync: reconcile-warning cleared slot=%s symbol=%s broker_qty=%d actor=sync", slot_id, sym, broker_qty, ) except Exception: # noqa: BLE001 _log.exception("sync: failed to clear_warned for slot=%s", slot_id) return { "slot_id": slot_id, "symbol": sym, "portfolio_id": pid, "before": before, "after": after, "warning_cleared": warning_cleared, "dry_run": False, "skipped": False, "seeded_avg_price": seeded_avg_price, } # --------------------------------------------------------------------------- # BAT-051 / REQ-106: operator verify command — C-case re-check (read-centric) # --------------------------------------------------------------------------- def verify_slot_reconcile( *, portfolio_id: str, store: SnapshotStore, broker: BrokerPort, slot_repo, instance_key: str, warn_store: "ReconcileWarningStore", ) -> dict: """Re-check C-case reconcile-warning for a portfolio (REQ-106, BAT-051). Read-centric: does NOT modify slot qty. Clears warning only when re-check confirms ledger qty == broker qty (no force-clear path, OQ-C confirmed). result values: "cleared" — re-check passed, warning cleared. "rejected" — re-check failed (still mismatched), warning NOT cleared. "not_c_case" — slot is warned but not case_code='C' (use sync for B case). "not_warned" — slot has no active warning (idempotent, exit 0). "slot_not_found" — no slot for this portfolio_id in this instance. "no_warning_info" — is_warned but get_warning_info returned None (unexpected). """ slot = slot_repo.find_slot_by_portfolio(instance_key, portfolio_id) if slot is None: return { "portfolio_id": portfolio_id, "slot_id": None, "result": "slot_not_found", "hub_qty": None, "broker_qty": None, "reason": f"no slot for portfolio_id={portfolio_id} in instance={instance_key}", } slot_id_str = str(slot.slot_id) if not warn_store.is_warned(slot_id_str): return { "portfolio_id": portfolio_id, "slot_id": slot_id_str, "result": "not_warned", "hub_qty": None, "broker_qty": None, "reason": "no active reconcile-warning (already cleared or never set)", } # Check case_code — verify is C-only (REQ-106 OQ-C). info = warn_store.get_warning_info(slot_id_str) if info is None: return { "portfolio_id": portfolio_id, "slot_id": slot_id_str, "result": "no_warning_info", "hub_qty": None, "broker_qty": None, "reason": "warning exists but get_warning_info returned None (unexpected)", } case_code = info.get("case_code", "") if case_code != "C": return { "portfolio_id": portfolio_id, "slot_id": slot_id_str, "result": "not_c_case", "hub_qty": None, "broker_qty": None, "reason": ( f"warning case_code='{case_code}' is not 'C' — " "use 'trex-core sync' to resolve B-case (slot↔broker qty mismatch)" ), } # Re-run the C-case check: hub qty vs broker qty for this symbol. sym = slot.symbol or "" broker_qty_map: dict[str, int] = { item.symbol: int(item.hldg_qty) for item in broker.get_balance() } broker_qty = broker_qty_map.get(sym, 0) # Hub qty: sum asset_quantity for this portfolio_id from snapshot. hub_qty: int = 0 snap = store.get(portfolio_id) if snap is not None: hub_qty = int(snap.item.asset_quantity) match = hub_qty == broker_qty if match: try: warn_store.clear_warned(slot_id_str) store.set_warning(portfolio_id, False) _log.info( "verify: reconcile-warning cleared (C-case resolved) " "slot=%s portfolio_id=%s symbol=%s hub_qty=%d broker_qty=%d", slot_id_str, portfolio_id, sym, hub_qty, broker_qty, ) except Exception: # noqa: BLE001 _log.exception("verify: failed to clear_warned for slot=%s", slot_id_str) return { "portfolio_id": portfolio_id, "slot_id": slot_id_str, "result": "cleared", "hub_qty": hub_qty, "broker_qty": broker_qty, "reason": None, } else: _log.warning( "verify: C-case still mismatched — slot=%s portfolio_id=%s symbol=%s " "hub_qty=%d broker_qty=%d diff=%d — warning NOT cleared", slot_id_str, portfolio_id, sym, hub_qty, broker_qty, broker_qty - hub_qty, ) return { "portfolio_id": portfolio_id, "slot_id": slot_id_str, "result": "rejected", "hub_qty": hub_qty, "broker_qty": broker_qty, "reason": ( f"hub_qty={hub_qty} broker_qty={broker_qty} " f"diff={broker_qty - hub_qty} symbol={sym}" ), }