feat: add bounded KIS account read recovery

trex_core/adapters/broker/kis.py

@@ -12,6 +12,7 @@ Endpoints: /oauth2/tokenP (token), /uapi/domestic-stock/v1/trading/order-cash
 from __future__ import annotations
 
 import base64
+import hashlib
 import json
 from dataclasses import replace
 from datetime import datetime, time, timedelta, timezone
@@ -157,6 +158,11 @@ class KISBroker:
     def name(self) -> str:
         return "kis-mock" if self._mock else "kis-live"
 
+    def account_scope_hash(self) -> str:
+        """Stable non-secret identity for account-scoped persisted state."""
+        cano, product = self._account()
+        return hashlib.sha256(f"{cano}|{product}".encode()).hexdigest()
+
     # --- credential hot-swap (dcy-036 / REQ-113 AC-2 / REQ-115 AC-1) ----------
     def reinitialize(self, credential) -> None:
         """Replace app_key/app_secret/account_ref with fetched day-start credential.
@@ -711,8 +717,20 @@ class KISBroker:
             )
             response.raise_for_status()
             payload = response.json()
-        except (httpx.HTTPError, KeyError, TypeError, ValueError):
+        except BrokerBalanceError:
+            raise
+        except httpx.HTTPStatusError as exc:
+            status = exc.response.status_code if exc.response is not None else None
+            code = (
+                "authentication_failed"
+                if status == 401
+                else ("authorization_failed" if status == 403 else "transport_failed")
+            )
+            raise BrokerBalanceError(code) from None
+        except httpx.HTTPError:
             raise BrokerBalanceError("transport_failed") from None
+        except (KeyError, TypeError, ValueError):
+            raise BrokerBalanceError("malformed_response") from None
         if not isinstance(payload, dict):
             raise BrokerBalanceError("malformed_response")
         if str(payload.get("rt_cd")) != "0":

trex_core/adapters/broker/ownership.py

@@ -159,5 +159,11 @@ class OwnershipAwareBroker:
     def get_balance(self) -> list[BalanceItem]:
         return self._store.project(self._delegate.get_balance())
 
+    def account_scope_hash(self) -> str:
+        return self._delegate.account_scope_hash()
+
+    def get_available_cash(self, *, price_check_code: str = "005930") -> int:
+        return self._delegate.get_available_cash(price_check_code=price_check_code)
+
     def get_current_price(self, symbol: str, market: str = "J"):
         return self._delegate.get_current_price(symbol, market)

trex_core/adapters/persistence/account_snapshot.py

@@ -0,0 +1,104 @@
+"""PostgreSQL persistence for bounded broker account snapshots."""
+from __future__ import annotations
+
+import json
+from contextlib import contextmanager
+from datetime import datetime
+from typing import Any, Optional
+
+from trex_core.application.account_snapshot import AccountSnapshot
+from trex_core.application.broker_port import BalanceItem
+
+
+def _require_psycopg(module: Any = None) -> Any:
+    if module is not None:
+        return module
+    try:
+        import psycopg  # type: ignore[import-not-found]
+    except ImportError as exc:  # pragma: no cover
+        raise RuntimeError("psycopg is required for account snapshot persistence") from exc
+    return psycopg
+
+
+class PostgreSQLAccountSnapshotStore:
+    def __init__(self, dsn: str, *, psycopg_module: Any = None) -> None:
+        self._dsn = dsn
+        self._psycopg = _require_psycopg(psycopg_module)
+
+    def save(self, snapshot: AccountSnapshot) -> None:
+        holdings = [
+            {"symbol": item.symbol, "hldg_qty": item.hldg_qty}
+            for item in snapshot.holdings
+        ]
+        with self._psycopg.connect(self._dsn, autocommit=False) as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    "INSERT INTO broker_account_snapshot "
+                    "(account_hash, holdings, holdings_captured_at, "
+                    "available_cash, cash_captured_at) "
+                    "VALUES (%s, %s::jsonb, %s, %s, %s) "
+                    "ON CONFLICT (account_hash) DO UPDATE SET "
+                    "holdings=excluded.holdings, "
+                    "holdings_captured_at=excluded.holdings_captured_at, "
+                    "available_cash=excluded.available_cash, "
+                    "cash_captured_at=excluded.cash_captured_at",
+                    (
+                        snapshot.account_hash,
+                        json.dumps(holdings, separators=(",", ":"), sort_keys=True),
+                        snapshot.holdings_captured_at,
+                        snapshot.available_cash,
+                        snapshot.cash_captured_at,
+                    ),
+                )
+                conn.commit()
+
+    def load(self, account_hash: str) -> Optional[AccountSnapshot]:
+        with self._psycopg.connect(self._dsn) as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    "SELECT holdings, holdings_captured_at, "
+                    "available_cash, cash_captured_at "
+                    "FROM broker_account_snapshot WHERE account_hash=%s",
+                    (account_hash,),
+                )
+                row = cur.fetchone()
+        if row is None:
+            return None
+        raw_holdings = row[0]
+        if isinstance(raw_holdings, str):
+            raw_holdings = json.loads(raw_holdings)
+        holdings = tuple(
+            BalanceItem(
+                symbol=str(item["symbol"]),
+                hldg_qty=int(item["hldg_qty"]),
+            )
+            for item in raw_holdings
+        )
+        return AccountSnapshot(
+            account_hash=account_hash,
+            holdings=holdings,
+            holdings_captured_at=(
+                row[1] if isinstance(row[1], datetime)
+                else (datetime.fromisoformat(row[1]) if row[1] else None)
+            ),
+            available_cash=int(row[2]) if row[2] is not None else None,
+            cash_captured_at=(
+                row[3] if isinstance(row[3], datetime)
+                else (datetime.fromisoformat(row[3]) if row[3] else None)
+            ),
+        )
+
+    @contextmanager
+    def refresh_guard(self, account_hash: str):
+        """One refresh owner per account across Core processes."""
+        lock_id = int(account_hash[:15], 16)
+        with self._psycopg.connect(self._dsn, autocommit=True) as conn:
+            with conn.cursor() as cur:
+                cur.execute("SELECT pg_try_advisory_lock(%s)", (lock_id,))
+                acquired = bool(cur.fetchone()[0])
+            try:
+                yield acquired
+            finally:
+                if acquired:
+                    with conn.cursor() as cur:
+                        cur.execute("SELECT pg_advisory_unlock(%s)", (lock_id,))

trex_core/adapters/persistence/sql/021_create_broker_account_snapshot.sql

@@ -0,0 +1,10 @@
+-- dcy-041 / GitHub #45: account-scoped last healthy KIS snapshot.
+-- account_hash is SHA-256(CANO|ACNT_PRDT_CD); no plaintext account identity,
+-- token, key, order response, or broker payload is persisted.
+CREATE TABLE IF NOT EXISTS broker_account_snapshot (
+    account_hash    TEXT PRIMARY KEY,
+    holdings        JSONB NOT NULL,
+    holdings_captured_at TIMESTAMPTZ,
+    available_cash  BIGINT CHECK (available_cash IS NULL OR available_cash >= 0),
+    cash_captured_at TIMESTAMPTZ
+);

trex_core/application/account_snapshot.py

@@ -0,0 +1,242 @@
+"""Bounded account snapshot recovery for transient KIS read failures.
+
+Holdings and cash capacity have independent capture times so a cash-read
+failure never invalidates a healthy holdings refresh (and vice versa). Fresh
+reconciliation remains live-only; snapshots are consumed solely by the
+new-buy admission guard and the pending-sell quantity path.
+"""
+from __future__ import annotations
+
+import logging
+import random
+import threading
+from contextlib import AbstractContextManager, contextmanager
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Callable, Optional, Protocol
+
+from trex_core.application.broker_port import BalanceItem, BrokerBalanceError
+
+_log = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class AccountSnapshot:
+    account_hash: str
+    holdings: tuple[BalanceItem, ...]
+    holdings_captured_at: Optional[datetime]
+    available_cash: Optional[int]
+    cash_captured_at: Optional[datetime]
+
+
+class AccountSnapshotStore(Protocol):
+    def save(self, snapshot: AccountSnapshot) -> None: ...
+    def load(self, account_hash: str) -> Optional[AccountSnapshot]: ...
+    def refresh_guard(self, account_hash: str) -> AbstractContextManager[bool]: ...
+
+
+class AccountSnapshotBroker(Protocol):
+    def account_scope_hash(self) -> str: ...
+    def get_balance(self) -> list[BalanceItem]: ...
+    def get_available_cash(self, *, price_check_code: str = "005930") -> int: ...
+
+
+class AccountSnapshotUnavailable(RuntimeError):
+    """A sanitized fail-closed result for snapshot consumers."""
+
+    _CODES = frozenset({
+        "not_ready",
+        "account_mismatch",
+        "expired",
+        "live_read_failed",
+    })
+
+    def __init__(self, code: str) -> None:
+        if code not in self._CODES:
+            raise ValueError(f"unsupported account snapshot error code: {code}")
+        self.code = code
+        super().__init__(f"account snapshot unavailable ({code})")
+
+
+class InMemoryAccountSnapshotStore:
+    def __init__(self) -> None:
+        self._items: dict[str, AccountSnapshot] = {}
+        self._items_lock = threading.Lock()
+        self._refresh_lock = threading.Lock()
+
+    def save(self, snapshot: AccountSnapshot) -> None:
+        with self._items_lock:
+            self._items[snapshot.account_hash] = snapshot
+
+    def load(self, account_hash: str) -> Optional[AccountSnapshot]:
+        with self._items_lock:
+            return self._items.get(account_hash)
+
+    @contextmanager
+    def refresh_guard(self, account_hash: str):
+        acquired = self._refresh_lock.acquire(blocking=False)
+        try:
+            yield acquired
+        finally:
+            if acquired:
+                self._refresh_lock.release()
+
+
+class AccountSnapshotService:
+    """Refresh, age-check, and consume account-scoped broker snapshots."""
+
+    def __init__(
+        self,
+        *,
+        broker: AccountSnapshotBroker,
+        store: AccountSnapshotStore,
+        buy_max_age_sec: int = 600,
+        sell_max_age_sec: int = 3600,
+        refresh_interval_sec: float = 300.0,
+        clock: Optional[Callable[[], datetime]] = None,
+        rng: Optional[Callable[[], float]] = None,
+    ) -> None:
+        if buy_max_age_sec < 0 or sell_max_age_sec < buy_max_age_sec:
+            raise ValueError("invalid account snapshot age policy")
+        if sell_max_age_sec > 21_600:
+            raise ValueError("account snapshot age exceeds six-hour ceiling")
+        if refresh_interval_sec <= 0:
+            raise ValueError("account snapshot refresh interval must be positive")
+        self._broker = broker
+        self._store = store
+        self._buy_max_age = buy_max_age_sec
+        self._sell_max_age = sell_max_age_sec
+        self._refresh_interval = refresh_interval_sec
+        self._clock = clock or (lambda: datetime.now(tz=timezone.utc))
+        self._rng = rng or random.random
+        self._refresh_lock = threading.Lock()
+        self._last_holdings_error: Optional[str] = "not_attempted"
+        self._last_cash_error: Optional[str] = "not_attempted"
+
+    def refresh(self) -> bool:
+        """Refresh holdings and cash independently under one account-scoped owner."""
+        account_hash = self._broker.account_scope_hash()
+        if not account_hash:
+            self._last_holdings_error = "ownership_mismatch"
+            self._last_cash_error = "ownership_mismatch"
+            return False
+        with self._refresh_lock, self._store.refresh_guard(account_hash) as acquired:
+            if not acquired:
+                return False
+            existing = self._store.load(account_hash)
+            holdings = existing.holdings if existing is not None else ()
+            holdings_captured_at = (
+                existing.holdings_captured_at if existing is not None else None
+            )
+            available_cash = existing.available_cash if existing is not None else None
+            cash_captured_at = existing.cash_captured_at if existing is not None else None
+            now = self._clock()
+            holdings_ok = cash_ok = False
+            try:
+                holdings = tuple(self._broker.get_balance())
+                holdings_captured_at = now
+                self._last_holdings_error = None
+                holdings_ok = True
+            except BrokerBalanceError as exc:
+                self._last_holdings_error = exc.code
+                _log.warning(
+                    "account holdings snapshot refresh failed: code=%s",
+                    exc.code,
+                )
+            try:
+                available_cash = int(self._broker.get_available_cash())
+                if available_cash < 0:
+                    raise BrokerBalanceError("malformed_response")
+                cash_captured_at = now
+                self._last_cash_error = None
+                cash_ok = True
+            except BrokerBalanceError as exc:
+                self._last_cash_error = exc.code
+                _log.warning(
+                    "account cash snapshot refresh failed: code=%s",
+                    exc.code,
+                )
+            if not account_hash or self._broker.account_scope_hash() != account_hash:
+                self._last_holdings_error = "ownership_mismatch"
+                self._last_cash_error = "ownership_mismatch"
+                return False
+            if not holdings_ok and not cash_ok:
+                return False
+            snapshot = AccountSnapshot(
+                account_hash=account_hash,
+                holdings=holdings,
+                holdings_captured_at=holdings_captured_at,
+                available_cash=available_cash,
+                cash_captured_at=cash_captured_at,
+            )
+            self._store.save(snapshot)
+            return holdings_ok and cash_ok
+
+    def run_forever(self, stop: threading.Event) -> None:
+        """Refresh continuously with bounded exponential backoff."""
+        failures = 0
+        while not stop.is_set():
+            ok = self.refresh()
+            failures = 0 if ok else min(failures + 1, 4)
+            if ok:
+                wait_sec = self._refresh_interval * (0.9 + 0.2 * self._rng())
+            else:
+                wait_sec = min(15.0 * (2 ** (failures - 1)), 300.0)
+            if stop.wait(wait_sec):
+                return
+
+    def _load_current(self) -> AccountSnapshot:
+        account_hash = self._broker.account_scope_hash()
+        if not account_hash:
+            raise AccountSnapshotUnavailable("account_mismatch")
+        snapshot = self._store.load(account_hash)
+        if snapshot is None or snapshot.account_hash != account_hash:
+            raise AccountSnapshotUnavailable("account_mismatch")
+        return snapshot
+
+    def _age(self, captured_at: Optional[datetime], max_age_sec: int) -> float:
+        if captured_at is None:
+            raise AccountSnapshotUnavailable("not_ready")
+        age = (self._clock() - captured_at).total_seconds()
+        if age < 0 or age > max_age_sec:
+            raise AccountSnapshotUnavailable("expired")
+        return age
+
+    def holdings_for_sell(self) -> list[BalanceItem]:
+        """Prefer a live balance; fallback only on transient transport failure."""
+        try:
+            return self._broker.get_balance()
+        except BrokerBalanceError as exc:
+            if exc.code != "transport_failed":
+                raise AccountSnapshotUnavailable("live_read_failed") from None
+        snapshot = self._load_current()
+        age = self._age(snapshot.holdings_captured_at, self._sell_max_age)
+        _log.warning(
+            "sell balance using account snapshot: age_sec=%.0f",
+            age,
+        )
+        return list(snapshot.holdings)
+
+    def buy_guard(
+        self,
+        *,
+        qty: Optional[float],
+        limit_price: Optional[float],
+        reserved_cash: float = 0.0,
+    ) -> Optional[str]:
+        """Return a stable rejection reason, or None when bounded admission passes."""
+        if self._last_cash_error not in {None, "transport_failed"}:
+            return "account_snapshot_unavailable"
+        try:
+            snapshot = self._load_current()
+            self._age(snapshot.cash_captured_at, self._buy_max_age)
+        except AccountSnapshotUnavailable:
+            return "account_snapshot_unavailable"
+        if qty is None or qty <= 0 or limit_price is None or limit_price <= 0:
+            return "account_snapshot_price_required"
+        requested = float(qty) * float(limit_price)
+        if snapshot.available_cash is None:
+            return "account_snapshot_unavailable"
+        if requested + max(0.0, reserved_cash) > snapshot.available_cash:
+            return "insufficient_snapshot_cash"
+        return None

trex_core/application/execution.py

@@ -30,6 +30,7 @@ from typing import Callable, Optional
 
 from trex_core.application.alarm_emitter import AlarmEmitter
 from trex_core.application.broker_port import (
+    BalanceItem,
     BrokerPort,
     CancelOutcome,
     CancelState,
@@ -84,6 +85,7 @@ class ExecutionService:
         stale_pending_threshold_sec: Optional[int] = None,
         instance_key: str = "default",
         alarm_emitter: Optional[AlarmEmitter] = None,
+        balance_reader: Optional[Callable[[], list[BalanceItem]]] = None,
     ) -> None:
         self._svc = slot_service
         self._repo = repo
@@ -104,6 +106,10 @@ class ExecutionService:
         )
         # dcy-033 / REQ-100: best-effort alarm emitter (None = no alarms sent).
         self._alarm_emitter = alarm_emitter
+        # dcy-041 / GitHub #45: pending-sell quantity may use a bounded
+        # 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
 
     # ------------------------------------------------------------
     def run_once(self) -> ExecutionResult:
@@ -136,7 +142,7 @@ class ExecutionService:
                 # surfaces via reconcile_core_holdings (REQ-062 A). No mirror
                 # fallback by design.
                 if sell_balance is None:
-                    sell_balance = self._balance_map()
+                    sell_balance = self._sell_dispatch_balance_map()
                 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
@@ -313,6 +319,13 @@ class ExecutionService:
         """Current broker balance as {symbol: hldg_qty} (REQ-062 ground truth)."""
         return {b.symbol: int(b.hldg_qty) for b in self._broker.get_balance()}
 
+    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()
+        }
+
     def _resolve_fill_price(self, fs, slot) -> tuple[float | None, bool]:
         """Resolve the fill price from FillStatus, with limit_price backfill.
 

trex_core/application/intents.py

@@ -115,6 +115,7 @@ class IntentService:
         default_order_timeout_sec: Optional[int] = None,
         warn_store=None,  # Optional[ReconcileWarningStore] — BAT-050 / REQ-104
         entry_allowed_fn: Optional[Callable[[], bool]] = None,
+        buy_guard_fn: Optional[Callable[[BuyIntent], Optional[str]]] = None,
     ) -> None:
         self._svc = slot_service
         self._repo = slot_repo
@@ -138,6 +139,10 @@ class IntentService:
         # When returns False, BuyIntent is rejected with reason 'day_start_load_failed'.
         # SellIntent is NOT gated by this (REQ-114 AC-5: 청산 경로 영향 없음).
         self._entry_allowed_fn: Optional[Callable[[], bool]] = entry_allowed_fn
+        # dcy-041 / GitHub #45: opt-in bounded cash-capacity admission guard.
+        # The callable returns a stable rejection reason or None. SellIntent is
+        # deliberately outside this gate.
+        self._buy_guard_fn = buy_guard_fn
 
     # ------------------------------------------------------------
     def lookup(self, intent_id: str) -> Optional[IntentDecision]:
@@ -270,6 +275,15 @@ class IntentService:
             symbol = _resolve(slot)
             if not symbol:
                 return IntentDecision(intent.intent_id, "rejected", None, "unknown_portfolio")
+            if self._buy_guard_fn is not None:
+                try:
+                    guard_reason = self._buy_guard_fn(intent)
+                except Exception:  # noqa: BLE001 - buy guard fails closed
+                    guard_reason = "account_snapshot_unavailable"
+                if guard_reason is not None:
+                    return IntentDecision(
+                        intent.intent_id, "rejected", None, guard_reason
+                    )
             # M cap applies only to NEW activations (initial buy). An increase on
             # an already-active asset does not change the active asset count.
             if (not is_increase and self._max_active_slots is not None

trex_core/application/operator_report.py

@@ -129,7 +129,7 @@ def render_text(rep: OperatorReport) -> str:
     # REQ-110 WAC 실현손익 섹션 (portfolio-hub 통합)
     if rep.wac_unavailable:
         lines.append(
-            f"=== WAC 실현 PnL: portfolio-hub unavailable — realized PnL unavailable ==="
+            "=== WAC 실현 PnL: portfolio-hub unavailable — realized PnL unavailable ==="
         )
         if rep.wac_unavailable_reason:
             lines.append(f"  ({rep.wac_unavailable_reason})")

trex_core/runtime/config.py

@@ -112,6 +112,28 @@ class SnapshotScheduleConfig:
     tick_sec: float = 60.0
 
 
+@dataclass(frozen=True)
+class AccountSnapshotConfig:
+    """Opt-in bounded KIS account-read recovery (dcy-041 / GitHub #45)."""
+
+    enabled: bool = False
+    buy_max_age_sec: int = 600
+    sell_max_age_sec: int = 3600
+    refresh_interval_sec: float = 300.0
+
+    def __post_init__(self) -> None:
+        if self.buy_max_age_sec < 0:
+            raise ValueError("account snapshot buy_max_age_sec must be >= 0")
+        if self.sell_max_age_sec < self.buy_max_age_sec:
+            raise ValueError(
+                "account snapshot sell_max_age_sec must be >= buy_max_age_sec"
+            )
+        if self.sell_max_age_sec > 21_600:
+            raise ValueError("account snapshot sell_max_age_sec exceeds six hours")
+        if self.refresh_interval_sec <= 0:
+            raise ValueError("account snapshot refresh_interval_sec must be > 0")
+
+
 @dataclass(frozen=True)
 class SafetySellConfig:
     """Safety-sell feature tuning (REQ-041 / BAT-010).
@@ -227,6 +249,9 @@ class DaemonConfig:
     rabbitmq: Optional[RabbitMQConfig] = None
     # REQ-030 (BAT-008): daily snapshot refresh.
     snapshot_schedule: SnapshotScheduleConfig = SnapshotScheduleConfig()
+    # dcy-041 / GitHub #45: opt-in account read snapshot. Disabled keeps the
+    # previous live-only balance path byte-for-byte compatible.
+    account_snapshot: AccountSnapshotConfig = AccountSnapshotConfig()
     # REQ-041 (BAT-010): safety-sell feature config.
     safety_sell: SafetySellConfig = SafetySellConfig()
     # dcy-013 / REQ-014 (BAT-017): Alert-only pending-order stale watchdog
@@ -301,6 +326,20 @@ class DaemonConfig:
             kst_time=os.environ.get("TREX_SNAPSHOT_SCHEDULE_KST", "07:30").strip() or "07:30",
             tick_sec=float(os.environ.get("TREX_SNAPSHOT_TICK_SEC", "60")),
         )
+        account_snapshot = AccountSnapshotConfig(
+            enabled=os.environ.get(
+                "TREX_ACCOUNT_SNAPSHOT_ENABLED", ""
+            ).lower() in ("1", "true", "yes"),
+            buy_max_age_sec=int(
+                os.environ.get("TREX_ACCOUNT_SNAPSHOT_BUY_MAX_AGE_SEC", "600")
+            ),
+            sell_max_age_sec=int(
+                os.environ.get("TREX_ACCOUNT_SNAPSHOT_SELL_MAX_AGE_SEC", "3600")
+            ),
+            refresh_interval_sec=float(
+                os.environ.get("TREX_ACCOUNT_SNAPSHOT_REFRESH_INTERVAL_SEC", "300")
+            ),
+        )
         _mc_time = os.environ.get("TREX_MARKET_CLOSE_LIQUIDATION_TIME", "").strip() or None
         safety_sell = SafetySellConfig(
             watchdog_timeout_sec=int(os.environ.get("TREX_WATCHDOG_TIMEOUT", "120")),
@@ -329,6 +368,7 @@ class DaemonConfig:
             ledger_interval_sec=float(os.environ.get("TREX_LEDGER_INTERVAL_SEC", "1.0")),
             rabbitmq=rabbitmq,
             snapshot_schedule=snapshot_schedule,
+            account_snapshot=account_snapshot,
             safety_sell=safety_sell,
             pending_stale_threshold_sec=int(
                 os.environ.get("TREX_PENDING_STALE_THRESHOLD", "0")

trex_core/runtime/daemon.py

@@ -32,6 +32,9 @@ from trex_core.adapters.messaging import (
     topology,
 )
 from trex_core.adapters.persistence.migrate import apply_migrations
+from trex_core.adapters.persistence.account_snapshot import (
+    PostgreSQLAccountSnapshotStore,
+)
 from trex_core.adapters.persistence.postgresql import (
     PgStrategyStateStore,
     PostgreSQLIntentStore,
@@ -46,9 +49,11 @@ from trex_core.adapters.portfolio import (
 )
 from trex_core.adapters.portfolio.killswitch_store import PostgreSQLKillSwitchStore
 from trex_core.application.alarm_emitter import AlarmEmitter
+from trex_core.application.account_snapshot import AccountSnapshotService
 from trex_core.application.broker_port import BrokerPort
 from trex_core.application.execution import ExecutionService
 from trex_core.application.intents import IntentService
+from trex_core.application.intents import BuyIntent
 from trex_core.application.ledger_outbox import LedgerWriter
 from trex_core.application.ledger_port import LedgerPort
 from trex_core.application.messaging import MessagingTransport
@@ -199,6 +204,28 @@ class Daemon:
         from trex_core.runtime.broker_provider import build_broker_provider
         self.broker_provider = build_broker_provider(config, read_port=self.portfolio_read)
         self.broker, self.broker_source = self.broker_provider.get_broker_for_read()
+        self.account_snapshot_service: AccountSnapshotService | None = None
+        if config.account_snapshot.enabled:
+            if config.broker != "kis":
+                raise RuntimeError(
+                    "account snapshot recovery is supported only with broker=kis"
+                )
+            required = (
+                "account_scope_hash",
+                "get_balance",
+                "get_available_cash",
+            )
+            if any(not callable(getattr(self.broker, name, None)) for name in required):
+                raise RuntimeError(
+                    "configured KIS broker lacks account snapshot capabilities"
+                )
+            self.account_snapshot_service = AccountSnapshotService(
+                broker=self.broker,  # type: ignore[arg-type]
+                store=PostgreSQLAccountSnapshotStore(config.dsn),
+                buy_max_age_sec=config.account_snapshot.buy_max_age_sec,
+                sell_max_age_sec=config.account_snapshot.sell_max_age_sec,
+                refresh_interval_sec=config.account_snapshot.refresh_interval_sec,
+            )
 
         # Event publisher transport — one dedicated instance for the
         # execution-loop thread (pika is not thread-safe).
@@ -263,6 +290,11 @@ class Daemon:
             default_order_timeout_sec=config.default_order_timeout_sec,  # dcy-022
             warn_store=self.warn_store,  # BAT-050 / REQ-104: reconcile-warning gate
             entry_allowed_fn=entry_allowed_fn,  # dcy-036 / REQ-114: day-start gate
+            buy_guard_fn=(
+                self._account_snapshot_buy_guard
+                if self.account_snapshot_service is not None
+                else None
+            ),
         )
         self.execution = ExecutionService(
             self.slot_service, self.slot_repo, self.broker,
@@ -271,6 +303,11 @@ class Daemon:
             stale_pending_threshold_sec=config.pending_stale_threshold_sec,
             instance_key=config.instance_key,
             alarm_emitter=self.alarm_emitter,  # dcy-033 / REQ-100
+            balance_reader=(
+                self.account_snapshot_service.holdings_for_sell
+                if self.account_snapshot_service is not None
+                else None
+            ),
         )
 
         # BAT-011: safety-sell trigger service (forced_close / SL-TP / market-close).
@@ -293,6 +330,35 @@ class Daemon:
         asset = self.snapshot_store.get(portfolio_id)
         return asset.item.symbol if asset is not None else None
 
+    def _account_snapshot_buy_guard(self, intent: BuyIntent) -> str | None:
+        """Bound one new buy against fresh cached cash plus pending reservations."""
+        service = self.account_snapshot_service
+        if service is None:
+            return None
+        reserved_cash = 0.0
+        for slot in self.slot_repo.list_slots(
+            self.config.instance_key, status_filter={"pending_buy"}
+        ):
+            if (
+                slot.qty is None
+                or slot.qty <= 0
+                or slot.limit_price is None
+                or slot.limit_price <= 0
+            ):
+                return "pending_buy_notional_unknown"
+            price = float(slot.limit_price)
+            if slot.slippage_pct:
+                price *= 1.0 + float(slot.slippage_pct) / 100.0
+            reserved_cash += float(slot.qty) * price
+        price = intent.entry_price
+        if price is not None and intent.slippage_pct:
+            price *= 1.0 + float(intent.slippage_pct) / 100.0
+        return service.buy_guard(
+            qty=intent.qty,
+            limit_price=price,
+            reserved_cash=reserved_cash,
+        )
+
     # ------------------------------------------------------------ event hook
     def _make_event_emitter(self):
         def emit(event_type: str, slot, routing_strategy_id: str | None = None) -> None:
@@ -424,9 +490,18 @@ class Daemon:
         except Exception:  # noqa: BLE001
             _log.exception("snapshot scheduler loop crashed")
 
+    def _account_snapshot_loop(self) -> None:
+        service = self.account_snapshot_service
+        if service is not None:
+            service.run_forever(self.stop)
+
     # ------------------------------------------------------------ lifecycle
     def setup(self) -> None:
         apply_migrations(self.config.dsn)
+        if self.account_snapshot_service is not None:
+            # First refresh is synchronous: buy admission never starts from a
+            # stale row without first attempting the current account.
+            self.account_snapshot_service.refresh()
         # dcy-020 (BAT-030): the daily pool preallocation is gone — permanent
         # asset slots are created/retired by the snapshot scheduler's universe
         # reconcile (boot fetch runs immediately in its thread).
@@ -453,6 +528,8 @@ class Daemon:
                 "safety_sell_enabled=false — trex-safety-sell worker not started. "
                 "Autonomous safety-sell will NOT run."
             )
+        if self.account_snapshot_service is not None:
+            workers.append(("trex-account-snapshot", self._account_snapshot_loop))
         for name, target in workers:
             t = threading.Thread(target=target, name=name, daemon=True)
             t.start()