From 084e05a97db0a29d7453c1737ac4cffe5562a1e0 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 12 Sep 2026 13:33:12 +0000 Subject: [PATCH] feat(ops): track broker balance worker failure lifecycle and backoff Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- tests/adapters/broker/test_kis.py | 30 +++- tests/application/test_execution.py | 58 ++++++++ tests/application/test_snapshot_scheduler.py | 57 ++++++++ trex_core/adapters/broker/kis.py | 10 +- trex_core/application/account_snapshot.py | 7 +- trex_core/application/balance_health.py | 139 +++++++++++++++++++ trex_core/application/broker_port.py | 2 +- trex_core/application/execution.py | 80 ++++++++++- trex_core/application/snapshot_scheduler.py | 97 +++++++++++-- 9 files changed, 453 insertions(+), 27 deletions(-) create mode 100644 trex_core/application/balance_health.py diff --git a/tests/adapters/broker/test_kis.py b/tests/adapters/broker/test_kis.py index f2251bf..ff36ab6 100644 --- a/tests/adapters/broker/test_kis.py +++ b/tests/adapters/broker/test_kis.py @@ -59,7 +59,8 @@ def test_account_scope_hash_separates_endpoint_and_application_identity(): (403, "authorization_failed"), (422, "broker_rejected"), (429, "transport_failed"), - (500, "transport_failed"), + (500, "http_5xx"), + (504, "timeout"), (None, "broker_rejected"), ], ) @@ -350,8 +351,8 @@ def test_get_balance_fails_closed_on_later_transport_and_http_errors(monkeypatch request = httpx.Request("GET", "https://example.test/balance") response = httpx.Response(503, request=request) for failure, expected_code in [ - (httpx.ConnectTimeout("unavailable"), "transport_failed"), - (httpx.HTTPStatusError("bad gateway", request=request, response=response), "transport_failed"), + (httpx.ConnectTimeout("unavailable"), "timeout"), + (httpx.HTTPStatusError("bad gateway", request=request, response=response), "http_5xx"), ]: broker, calls = _balance_broker(monkeypatch, [ _BalanceResponse(_balance_payload([{"pdno": "005930", "hldg_qty": "2"}], fk="A", nk="1"), "F"), @@ -407,6 +408,29 @@ def test_get_available_cash_uses_cash_only_amount(monkeypatch): assert b.get_available_cash() == 695_591 +@pytest.mark.parametrize( + ("failure", "expected_code"), + [ + (httpx.ReadTimeout("timed out"), "timeout"), + ( + httpx.HTTPStatusError( + "bad gateway", + request=httpx.Request("GET", "https://example.test"), + response=httpx.Response(502, request=httpx.Request("GET", "https://example.test")), + ), + "http_5xx", + ), + ], +) +def test_get_available_cash_classifies_timeout_and_http_5xx(monkeypatch, failure, expected_code): + b = KISBroker("12345678-01", app_key="k", app_secret="s", base_url=_MOCK_URL) + monkeypatch.setattr(b, "_headers", lambda _tr: {}) + monkeypatch.setattr("trex_core.adapters.broker.kis.httpx.get", lambda *a, **kw: (_ for _ in ()).throw(failure)) + with pytest.raises(BrokerBalanceError) as raised: + b.get_available_cash() + assert raised.value.code == expected_code + + def test_submit_transport_failure_blocks_automatic_duplicate(monkeypatch): b = KISBroker("12345678-01", app_key="k", app_secret="s", base_url=_MOCK_URL) monkeypatch.setattr(b, "_headers", lambda _tr: {}) diff --git a/tests/application/test_execution.py b/tests/application/test_execution.py index 6b6ba80..e5df5c4 100644 --- a/tests/application/test_execution.py +++ b/tests/application/test_execution.py @@ -5,8 +5,11 @@ In-memory adapters; deterministic paper double drives the full async path. from __future__ import annotations +from datetime import UTC, datetime, timedelta + from trex_core.adapters.broker.paper import PaperBroker from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore +from trex_core.application.account_snapshot import AccountSnapshotUnavailable from trex_core.application.broker_port import ( BalanceItem, CancelOutcome, @@ -214,6 +217,61 @@ def test_pending_sell_uses_injected_bounded_balance_reader(): assert repo.fetch_slot(sid).status is SlotState.PENDING_SELL +def test_pending_sell_balance_failure_is_fail_closed_with_backoff_and_recovery(): + now = {"t": datetime(2026, 9, 12, tzinfo=UTC)} + broker = PaperBroker(fill_after_polls=0) + repo, svc, intents, _ = _wire(broker) + sid = _buy(intents, qty=3).slot_id + + execution = ExecutionService( + svc, + repo, + broker, + instance_key=INST, + clock=lambda: now["t"], + ) + execution.run_once() + svc.sell(sid, actor="t", reason="full") + broker._fill_after = 5 + + calls = {"n": 0} + reader_state = {"ok": False} + + def failing_reader(): + calls["n"] += 1 + if not reader_state["ok"]: + raise AccountSnapshotUnavailable("expired") + return [BalanceItem("005931", 3)] + + execution = ExecutionService( + svc, + repo, + broker, + instance_key=INST, + balance_reader=failing_reader, + clock=lambda: now["t"], + ) + first = execution.run_once() + assert first.sell_skipped_no_balance == 1 + assert calls["n"] == 1 + health = execution.balance_worker_health(stale_after_sec=600) + assert health["state"] == "unavailable" + assert health["failure_code"] == "stale_snapshot" + assert health["consecutive_failures"] == 1 + + second = execution.run_once() + assert second.sell_skipped_no_balance == 1 + assert calls["n"] == 1 # bounded backoff prevented another live read + + now["t"] += timedelta(seconds=16) + reader_state["ok"] = True + third = execution.run_once() + assert third.dispatched == 1 + health = execution.balance_worker_health(stale_after_sec=600) + assert health["state"] == "ok" + assert health["consecutive_failures"] == 0 + + def test_partial_sell_qty_is_min_intent_broker(): """dcy-021/REQ-062 (D4): partial sell dispatches min(intent qty, broker); remainder stays holding (broker-authoritative qty), cost basis preserved.""" diff --git a/tests/application/test_snapshot_scheduler.py b/tests/application/test_snapshot_scheduler.py index 6b0c7c2..e6aa00e 100644 --- a/tests/application/test_snapshot_scheduler.py +++ b/tests/application/test_snapshot_scheduler.py @@ -665,3 +665,60 @@ def test_build_kis_reinit_callback_returns_callable_for_kis(): result(new_cred) assert broker._account_ref == "new-acct" assert broker._app_key == "new-k" + + +def test_balance_worker_health_isolated_backoff_and_recovery(): + from trex_core.application.broker_port import BalanceItem, BrokerBalanceError + from trex_core.domain.slot import SlotService + from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository + + class SequencedBroker(PaperBroker): + def __init__(self): + super().__init__() + self.steps = [] + + def get_balance(self): + if not self.steps: + return [] + step = self.steps.pop(0) + if isinstance(step, Exception): + raise step + return step + + now = {"t": datetime(2026, 9, 12, 0, 0, tzinfo=timezone.utc)} + read = _make_stub() + store = SnapshotStore() + slot_repo = InMemorySlotRepository() + slot_service = SlotService(slot_repo, instance_key="default") + slot_repo.reconcile_universe("default", {"pf-X": "X"}) + broker = SequencedBroker() + sched = SnapshotScheduler( + read_port=read, + store=store, + broker=broker, + slot_repo=slot_repo, + slot_service=slot_service, + clock=lambda: now["t"], + ) + + broker.steps = [BrokerBalanceError("http_5xx"), [BalanceItem("X", 0)]] + assert sched.run_once() is True + state = sched.balance_worker_health(stale_after_sec=86_400) + assert state["position_reconciliation"]["state"] == "unavailable" + assert state["position_reconciliation"]["failure_code"] == "http_5xx" + assert state["position_reconciliation"]["consecutive_failures"] == 1 + assert state["core_holding_reconciliation"]["state"] == "ok" + + broker.steps = [[BalanceItem("X", 0)]] + assert sched.run_once() is True + assert broker.steps == [] # position worker stayed in backoff, core worker still ran + state = sched.balance_worker_health(stale_after_sec=86_400) + assert state["position_reconciliation"]["consecutive_failures"] == 1 + assert state["core_holding_reconciliation"]["state"] == "ok" + + now["t"] += timedelta(seconds=16) + broker.steps = [[BalanceItem("X", 0)], [BalanceItem("X", 0)]] + assert sched.run_once() is True + state = sched.balance_worker_health(stale_after_sec=86_400) + assert state["position_reconciliation"]["state"] == "ok" + assert state["position_reconciliation"]["consecutive_failures"] == 0 diff --git a/trex_core/adapters/broker/kis.py b/trex_core/adapters/broker/kis.py index 4c2e711..d936632 100644 --- a/trex_core/adapters/broker/kis.py +++ b/trex_core/adapters/broker/kis.py @@ -54,7 +54,11 @@ def _balance_http_error_code(status: int | None) -> str: return "authentication_failed" if status == 403: return "authorization_failed" - if status in {408, 425, 429} or (status is not None and status >= 500): + if status in {408, 504}: + return "timeout" + if status is not None and status >= 500: + return "http_5xx" + if status in {425, 429}: return "transport_failed" return "broker_rejected" @@ -1235,6 +1239,8 @@ class KISBroker: except httpx.HTTPStatusError as exc: status = exc.response.status_code if exc.response is not None else None raise BrokerBalanceError(_balance_http_error_code(status)) from None + except httpx.TimeoutException: + raise BrokerBalanceError("timeout") from None except httpx.HTTPError: raise BrokerBalanceError("transport_failed") from None except (ValueError, TypeError): @@ -1318,6 +1324,8 @@ class KISBroker: except httpx.HTTPStatusError as exc: status = exc.response.status_code if exc.response is not None else None raise BrokerBalanceError(_balance_http_error_code(status)) from None + except httpx.TimeoutException: + raise BrokerBalanceError("timeout") from None except httpx.HTTPError: raise BrokerBalanceError("transport_failed") from None except (KeyError, TypeError, ValueError): diff --git a/trex_core/application/account_snapshot.py b/trex_core/application/account_snapshot.py index c61c08c..c42439b 100644 --- a/trex_core/application/account_snapshot.py +++ b/trex_core/application/account_snapshot.py @@ -52,10 +52,11 @@ class AccountSnapshotUnavailable(RuntimeError): "read_failed", }) - def __init__(self, code: str) -> None: + def __init__(self, code: str, *, detail_code: str | None = None) -> None: if code not in self._CODES: raise ValueError(f"unsupported account snapshot error code: {code}") self.code = code + self.detail_code = detail_code super().__init__(f"account snapshot unavailable ({code})") @@ -217,7 +218,9 @@ class AccountSnapshotService: return self._broker.get_balance() except BrokerBalanceError as exc: if exc.code != "transport_failed": - raise AccountSnapshotUnavailable("live_read_failed") from None + raise AccountSnapshotUnavailable( + "live_read_failed", detail_code=exc.code + ) from None snapshot = self._load_current() age = self._age(snapshot.holdings_captured_at, self._sell_max_age) _log.warning( diff --git a/trex_core/application/balance_health.py b/trex_core/application/balance_health.py new file mode 100644 index 0000000..37194d5 --- /dev/null +++ b/trex_core/application/balance_health.py @@ -0,0 +1,139 @@ +"""Broker-balance worker health tracking with bounded backoff.""" +from __future__ import annotations + +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from threading import Lock +from typing import Callable + + +@dataclass(frozen=True) +class BalanceWorkerHealth: + worker: str + last_success_at: datetime | None + consecutive_failures: int + failure_code: str | None + failure_at: datetime | None + next_retry_at: datetime | None + + def as_dict(self, *, now: datetime, stale_after_sec: float) -> dict[str, object]: + stale = False + if self.last_success_at is not None and stale_after_sec >= 0: + stale = (now - self.last_success_at).total_seconds() > stale_after_sec + state = "ok" + if self.last_success_at is None: + state = "unavailable" + elif self.consecutive_failures > 0: + state = "unavailable" + elif stale: + state = "stale" + return { + "worker": self.worker, + "state": state, + "last_success_at": ( + self.last_success_at.astimezone(UTC).isoformat() + if self.last_success_at is not None + else None + ), + "consecutive_failures": self.consecutive_failures, + "failure_code": self.failure_code, + "failure_at": ( + self.failure_at.astimezone(UTC).isoformat() + if self.failure_at is not None + else None + ), + "next_retry_at": ( + self.next_retry_at.astimezone(UTC).isoformat() + if self.next_retry_at is not None + else None + ), + "stale": stale, + } + + +class BalanceHealthTracker: + def __init__( + self, + *, + clock: Callable[[], datetime] | None = None, + base_backoff_sec: float = 15.0, + max_backoff_sec: float = 300.0, + ) -> None: + self._clock = clock or (lambda: datetime.now(tz=UTC)) + self._base_backoff_sec = max(1.0, float(base_backoff_sec)) + self._max_backoff_sec = max(self._base_backoff_sec, float(max_backoff_sec)) + self._lock = Lock() + self._states: dict[str, BalanceWorkerHealth] = {} + + def _now(self) -> datetime: + return self._clock() + + def _get(self, worker: str) -> BalanceWorkerHealth: + return self._states.get( + worker, + BalanceWorkerHealth( + worker=worker, + last_success_at=None, + consecutive_failures=0, + failure_code=None, + failure_at=None, + next_retry_at=None, + ), + ) + + def record_success(self, worker: str, *, captured_at: datetime | None = None) -> str | None: + with self._lock: + prev = self._get(worker) + recovered = prev.consecutive_failures > 0 + now = self._now() + self._states[worker] = BalanceWorkerHealth( + worker=worker, + last_success_at=captured_at or now, + consecutive_failures=0, + failure_code=None, + failure_at=None, + next_retry_at=None, + ) + return "recovered" if recovered else None + + def record_failure(self, worker: str, *, code: str) -> tuple[str | None, float]: + with self._lock: + prev = self._get(worker) + now = self._now() + failures = prev.consecutive_failures + 1 + backoff = min( + self._base_backoff_sec * (2 ** max(0, failures - 1)), + self._max_backoff_sec, + ) + self._states[worker] = BalanceWorkerHealth( + worker=worker, + last_success_at=prev.last_success_at, + consecutive_failures=failures, + failure_code=code, + failure_at=now, + next_retry_at=now + timedelta(seconds=backoff), + ) + lifecycle: str | None = None + if prev.consecutive_failures == 0 or prev.failure_code != code: + lifecycle = "detected" + elif failures in {3, 6, 12}: + lifecycle = "ongoing" + return lifecycle, backoff + + def ready(self, worker: str) -> tuple[bool, float]: + with self._lock: + state = self._get(worker) + now = self._now() + retry_at = state.next_retry_at + if retry_at is None: + return True, 0.0 + remain = (retry_at - now).total_seconds() + if remain <= 0: + return True, 0.0 + return False, remain + + def status(self, worker: str, *, stale_after_sec: float) -> dict[str, object]: + with self._lock: + state = self._get(worker) + now = self._now() + return state.as_dict(now=now, stale_after_sec=stale_after_sec) diff --git a/trex_core/application/broker_port.py b/trex_core/application/broker_port.py index 297a217..d72a110 100644 --- a/trex_core/application/broker_port.py +++ b/trex_core/application/broker_port.py @@ -20,7 +20,7 @@ class BrokerBalanceError(RuntimeError): _CODES = frozenset({ "token_issuance_failed", "authentication_failed", "authorization_failed", - "transport_failed", "malformed_response", "broker_rejected", + "transport_failed", "timeout", "http_5xx", "malformed_response", "broker_rejected", "ownership_mismatch", }) diff --git a/trex_core/application/execution.py b/trex_core/application/execution.py index dc74257..3e2eab3 100644 --- a/trex_core/application/execution.py +++ b/trex_core/application/execution.py @@ -30,13 +30,16 @@ from dataclasses import dataclass, replace from datetime import UTC, datetime, timedelta from trex_core.application.alarm_emitter import AlarmEmitter +from trex_core.application.balance_health import BalanceHealthTracker from trex_core.application.broker_port import ( BalanceItem, + BrokerBalanceError, BrokerPort, CancelOutcome, CancelState, OrderRequest, ) +from trex_core.application.account_snapshot import AccountSnapshotUnavailable from trex_core.application.ledger_outbox import LedgerOutbox from trex_core.application.ledger_port import LedgerTrade from trex_core.application.ledger_skip import publish_ledger_skip_alarm @@ -139,6 +142,8 @@ class ExecutionService: # account snapshot only through this injected reader. Reconciliation # remains wired directly to the live broker. self._sell_dispatch_balance_reader = balance_reader or broker.get_balance + self._balance_health = BalanceHealthTracker(clock=self._clock) + self._balance_worker = "execution" # ------------------------------------------------------------ def run_once(self) -> ExecutionResult: @@ -171,7 +176,22 @@ class ExecutionService: # surfaces via reconcile_core_holdings (REQ-062 A). No mirror # fallback by design. if sell_balance is None: - sell_balance = self._sell_dispatch_balance_map() + ready, backoff_left = self._balance_health.ready(self._balance_worker) + if not ready: + r.sell_skipped_no_balance += 1 + _log.info( + "sell dispatch deferred: broker balance backoff active " + "instance_key=%s worker=%s backoff_left_sec=%.1f", + self._instance_key, + self._balance_worker, + backoff_left, + ) + return + try: + sell_balance = self._sell_dispatch_balance_map() + except (AccountSnapshotUnavailable, BrokerBalanceError): + r.sell_skipped_no_balance += 1 + return bal = sell_balance.get(slot.symbol or "", 0) # dcy-021 / REQ-062 (D4): full sell (slot.qty is None) => broker # balance; partial sell => min(order qty, broker balance). Broker @@ -420,10 +440,60 @@ class ExecutionService: def _sell_dispatch_balance_map(self) -> dict: """Balance used only to size a not-yet-submitted pending sell.""" - return { - b.symbol: int(b.hldg_qty) - for b in self._sell_dispatch_balance_reader() - } + try: + holdings = self._sell_dispatch_balance_reader() + except AccountSnapshotUnavailable as exc: + code = "stale_snapshot" if exc.code == "expired" else ( + getattr(exc, "detail_code", None) or exc.code + ) + lifecycle, backoff = self._balance_health.record_failure( + self._balance_worker, code=code + ) + if lifecycle is not None: + _log.warning( + "broker balance unavailable: lifecycle=%s worker=%s instance_key=%s " + "failure_code=%s consecutive_failures=%d backoff_sec=%.1f", + lifecycle, + self._balance_worker, + self._instance_key, + code, + self._balance_health.status(self._balance_worker, stale_after_sec=600)[ + "consecutive_failures" + ], + backoff, + ) + raise + except BrokerBalanceError as exc: + lifecycle, backoff = self._balance_health.record_failure( + self._balance_worker, code=exc.code + ) + if lifecycle is not None: + _log.warning( + "broker balance unavailable: lifecycle=%s worker=%s instance_key=%s " + "failure_code=%s consecutive_failures=%d backoff_sec=%.1f", + lifecycle, + self._balance_worker, + self._instance_key, + exc.code, + self._balance_health.status(self._balance_worker, stale_after_sec=600)[ + "consecutive_failures" + ], + backoff, + ) + raise + lifecycle = self._balance_health.record_success(self._balance_worker) + if lifecycle == "recovered": + state = self._balance_health.status(self._balance_worker, stale_after_sec=600) + _log.info( + "broker balance recovered: worker=%s instance_key=%s last_success_at=%s", + self._balance_worker, + self._instance_key, + state["last_success_at"], + ) + return {b.symbol: int(b.hldg_qty) for b in holdings} + + def balance_worker_health(self, *, stale_after_sec: float = 600.0) -> dict[str, object]: + return self._balance_health.status(self._balance_worker, stale_after_sec=stale_after_sec) def _resolve_fill_price(self, fs, slot) -> tuple[float | None, bool]: """Resolve the fill price from FillStatus, with limit_price backfill. diff --git a/trex_core/application/snapshot_scheduler.py b/trex_core/application/snapshot_scheduler.py index b6ef0bc..ca50da9 100644 --- a/trex_core/application/snapshot_scheduler.py +++ b/trex_core/application/snapshot_scheduler.py @@ -16,7 +16,8 @@ from datetime import date, datetime, time, timedelta, timezone from typing import Callable, Optional from trex_core.application.alarm_emitter import AlarmEmitter -from trex_core.application.broker_port import BrokerPort +from trex_core.application.balance_health import BalanceHealthTracker +from trex_core.application.broker_port import BrokerBalanceError, BrokerPort from trex_core.application.portfolio import ( BrokerCredentialPayload, DegradedReason, @@ -95,6 +96,7 @@ class SnapshotScheduler: # REQ-114: entry gate. True = new entries allowed (default). # credential_fetch_enabled=False keeps this always True (no gate). self._entry_allowed: bool = True + self._balance_health = BalanceHealthTracker(clock=self._clock) # ---- callable from a daemon thread loop -------------------------------- def run_forever(self, stop: threading.Event, *, tick_sec: float = 60.0) -> None: @@ -310,14 +312,18 @@ class SnapshotScheduler: self._store.swap(items) try: # BAT-050 / REQ-103: pass warn_store + slot_repo for C-case durable warning. - reconcile_positions( - store=self._store, - broker=self._broker, - slot_repo=self._slot_repo, - instance_key=self._instance_key, - warn_store=self._warn_store, - alarm_emitter=self._alarm_emitter, - notification_store=self._notification_store, + self._run_balance_worker( + worker="position_reconciliation", + stale_after_sec=24 * 60 * 60, + run=lambda: reconcile_positions( + store=self._store, + broker=self._broker, + slot_repo=self._slot_repo, + instance_key=self._instance_key, + warn_store=self._warn_store, + alarm_emitter=self._alarm_emitter, + notification_store=self._notification_store, + ), ) except Exception: # noqa: BLE001 _log.exception("reconciliation pass failed") @@ -330,18 +336,79 @@ class SnapshotScheduler: try: from trex_core.application.reconciliation import reconcile_core_holdings # BAT-050 / REQ-103: pass warn_store for B-case durable warning. - reconcile_core_holdings( - slot_repo=self._slot_repo, slot_service=self._slot_service, - broker=self._broker, instance_key=self._instance_key, - alarm_emitter=self._alarm_emitter, # dcy-033 / REQ-100 - warn_store=self._warn_store, # BAT-050 / REQ-103 - notification_store=self._notification_store, # req-0007 / ac-0024 + self._run_balance_worker( + worker="core_holding_reconciliation", + stale_after_sec=24 * 60 * 60, + run=lambda: reconcile_core_holdings( + slot_repo=self._slot_repo, slot_service=self._slot_service, + broker=self._broker, instance_key=self._instance_key, + alarm_emitter=self._alarm_emitter, # dcy-033 / REQ-100 + warn_store=self._warn_store, # BAT-050 / REQ-103 + notification_store=self._notification_store, # req-0007 / ac-0024 + ), ) except Exception: # noqa: BLE001 _log.exception("core-holding divergence reconcile failed") self._last_fired_date = self._clock() return True + def _run_balance_worker( + self, + *, + worker: str, + stale_after_sec: float, + run: Callable[[], object], + ) -> None: + ready, backoff_left = self._balance_health.ready(worker) + if not ready: + _log.info( + "broker balance worker deferred: worker=%s instance_key=%s backoff_left_sec=%.1f", + worker, + self._instance_key, + backoff_left, + ) + return + try: + run() + except BrokerBalanceError as exc: + lifecycle, backoff = self._balance_health.record_failure(worker, code=exc.code) + if lifecycle is not None: + state = self._balance_health.status(worker, stale_after_sec=stale_after_sec) + _log.warning( + "broker balance unavailable: lifecycle=%s worker=%s instance_key=%s " + "failure_code=%s consecutive_failures=%s backoff_sec=%.1f", + lifecycle, + worker, + self._instance_key, + exc.code, + state["consecutive_failures"], + backoff, + ) + return + lifecycle = self._balance_health.record_success(worker) + if lifecycle == "recovered": + state = self._balance_health.status(worker, stale_after_sec=stale_after_sec) + _log.info( + "broker balance recovered: worker=%s instance_key=%s last_success_at=%s", + worker, + self._instance_key, + state["last_success_at"], + ) + + def balance_worker_health( + self, + *, + stale_after_sec: float = 24 * 60 * 60, + ) -> dict[str, dict[str, object]]: + return { + "position_reconciliation": self._balance_health.status( + "position_reconciliation", stale_after_sec=stale_after_sec + ), + "core_holding_reconciliation": self._balance_health.status( + "core_holding_reconciliation", stale_after_sec=stale_after_sec + ), + } + # ---- asset-centric slot reconcile (dcy-020/REQ-058, BAT-028) ---------- def _reconcile_slots(self, items: list) -> None: universe = {