feat(ledger): dcy-018 option D cash/position write-back + 추정 현금 (BAT-023)

ledger write 응답의 portfolio-hub 권위 cash/quantity 를 스냅샷에 write-back 해
매수/매도 직후 "수량 live·현금 stale" 비대칭 제거(option D 활성화). portfolio-hub#8
(*_after 응답) 동작 중 확인으로 그간 deferred 됐던 REQ-030 을 닫음.

거버넌스: 전부 change-req (신규 REQ 없음, dcy-018 OQ-3 a).
- REQ-017: record_trade → LedgerWriteOutcome{result, cash_balance_after,
  asset_quantity_after}. 어댑터가 응답 root flat *_after 파싱(new+duplicate, 누락 None)
- REQ-019: LedgerWriter 에 SnapshotStore 주입 → new+duplicate write-back
  (best-effort, *_after 절대값이라 멱등). LedgerOutbox.pending_cash_delta()
- REQ-029: 추정 현금 오버레이(_apply_cash_estimate, cash_estimated 라벨, OQ-4 i)
  + 슬롯 오버레이 유지(OQ-1 B). 미정산 outbox delta 기반, write-back 수렴
- REQ-030: option D 활성화 → Approved→Implemented

best-effort 격리(NFR-1): write-back/추정 실패가 ledger 멱등·청산 무차단.
배치 번호: 동시 커밋된 BAT-022(slot-state, dcy-016)와 충돌해 BAT-023 으로 배정.

검증: 단위 350 passed/35 skipped, 인프라 게이트(PG+RabbitMQ) 385 passed, ruff clean.
신규 test_ledger_option_d + adapter/PG 테스트. BAT-023 release-candidate.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

trex_core/adapters/ledger/in_memory.py

@@ -8,6 +8,7 @@ from trex_core.application.ledger_port import (
     LedgerPermanentError,
     LedgerRetryableError,
     LedgerTrade,
+    LedgerWriteOutcome,
     LedgerWriteResult,
 )
 
@@ -32,6 +33,8 @@ class InMemoryLedger:
         fail_n_then_succeed: int = 0,
         conflict_executions: set[str] | None = None,
         permanent_executions: set[str] | None = None,
+        cash_after: float | None = None,
+        quantity_after: float | None = None,
     ) -> None:
         self._trades: dict[str, LedgerTrade] = {}
         self._order: list[LedgerTrade] = []
@@ -39,8 +42,12 @@ class InMemoryLedger:
         self._fail_remaining = fail_n_then_succeed
         self._conflicts = set(conflict_executions or set())
         self._permanent = set(permanent_executions or set())
+        # dcy-018: optional authoritative cash/quantity returned on the outcome
+        # (simulates portfolio-hub#8 option D response). None => write-back skip.
+        self._cash_after = cash_after
+        self._quantity_after = quantity_after
 
-    def record_trade(self, trade: LedgerTrade) -> LedgerWriteResult:
+    def record_trade(self, trade: LedgerTrade) -> LedgerWriteOutcome:
         with self._lock:
             if trade.execution_id in self._conflicts:
                 raise LedgerConflictError(
@@ -55,10 +62,16 @@ class InMemoryLedger:
                 raise LedgerRetryableError("simulated transient failure")
             if trade.execution_id in self._trades:
                 # idempotent: same execution_id => no new entry, report dedup.
-                return LedgerWriteResult.DUPLICATE_OK
-            self._trades[trade.execution_id] = trade
-            self._order.append(trade)
-            return LedgerWriteResult.NEW
+                result = LedgerWriteResult.DUPLICATE_OK
+            else:
+                self._trades[trade.execution_id] = trade
+                self._order.append(trade)
+                result = LedgerWriteResult.NEW
+            return LedgerWriteOutcome(
+                result=result,
+                cash_balance_after=self._cash_after,
+                asset_quantity_after=self._quantity_after,
+            )
 
     # --- introspection for tests ---
     def trades(self) -> list[LedgerTrade]:

trex_core/adapters/ledger/portfolio_hub.py

@@ -26,9 +26,20 @@ from trex_core.application.ledger_port import (
     LedgerPermanentError,
     LedgerRetryableError,
     LedgerTrade,
+    LedgerWriteOutcome,
     LedgerWriteResult,
 )
 
+
+def _opt_num(v):
+    """Parse an optional numeric (cash/quantity) field; None if absent/invalid."""
+    if v is None:
+        return None
+    try:
+        return float(v)
+    except (TypeError, ValueError):
+        return None
+
 # dcy-015 / REQ-017: client errors that retrying cannot resolve.
 _PERMANENT_STATUS = frozenset({400, 404, 422})
 
@@ -75,7 +86,7 @@ class PortfolioHubLedger:
         return httpx.post(url, json=json, headers=self._headers(),
                           timeout=self._timeout)
 
-    def record_trade(self, trade: LedgerTrade) -> LedgerWriteResult:
+    def record_trade(self, trade: LedgerTrade) -> LedgerWriteOutcome:
         url = f"{self._base_url}/v1/portfolios/{trade.asset_id}/ledger/trades"
         try:
             resp = self._post(url, json=self._payload(trade))
@@ -89,9 +100,19 @@ class PortfolioHubLedger:
                 body = resp.json() if resp.content else {}
             except ValueError:
                 body = {}
-            if isinstance(body, dict) and body.get("duplicate") is True:
-                return LedgerWriteResult.DUPLICATE_OK
-            return LedgerWriteResult.NEW
+            if not isinstance(body, dict):
+                body = {}
+            result = (
+                LedgerWriteResult.DUPLICATE_OK if body.get("duplicate") is True
+                else LedgerWriteResult.NEW
+            )
+            # dcy-018 / REQ-017 (option D, portfolio-hub#8): root flat authoritative
+            # cash/quantity after the write — present on new(201) + duplicate(200).
+            return LedgerWriteOutcome(
+                result=result,
+                cash_balance_after=_opt_num(body.get("cash_balance_after")),
+                asset_quantity_after=_opt_num(body.get("asset_quantity_after")),
+            )
         if resp.status_code == 409:
             raise LedgerConflictError(
                 f"portfolio-hub 409 conflict for execution_id={trade.execution_id}: "

trex_core/adapters/persistence/postgresql.py

@@ -369,6 +369,20 @@ class PostgreSQLLedgerOutbox:
                 )
                 return [self._row_to_item(r) for r in cur.fetchall()]
 
+    def pending_cash_delta(self) -> dict[str, float]:
+        # dcy-018 / REQ-029 (OQ-4): estimated cash delta of all pending rows by
+        # asset_id. SUM in SQL: sell → +(qty*price - fee - tax), buy → negative.
+        with self._psycopg.connect(self._dsn) as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    "SELECT asset_id, "
+                    "SUM(CASE WHEN side='sell' "
+                    "         THEN (qty*price - fee - tax) "
+                    "         ELSE -(qty*price + fee + tax) END) "
+                    "FROM ledger_outbox WHERE status='pending' GROUP BY asset_id"
+                )
+                return {r[0]: float(r[1]) for r in cur.fetchall() if r[1] is not None}
+
     def get(self, correlation_id: str) -> Optional[LedgerWorkItem]:
         with self._psycopg.connect(self._dsn) as conn:
             with conn.cursor() as cur:

trex_core/application/ledger_outbox.py

@@ -69,6 +69,19 @@ _DELIVERY_FAILED_OUTCOMES = frozenset(
 )
 
 
+def trade_cash_delta(trade: LedgerTrade) -> float:
+    """Estimated cash impact of a trade (dcy-018 / REQ-029 OQ-4 estimate).
+
+    buy → -(qty*price + fee + tax); sell → +(qty*price - fee - tax). The fee is
+    the core's approximation, so this is an estimate that converges to the
+    portfolio-hub authoritative value on option-D write-back."""
+    notional = float(trade.qty) * float(trade.price)
+    fees = float(trade.fee) + float(trade.tax)
+    if trade.side == "sell":
+        return notional - fees
+    return -(notional + fees)
+
+
 @dataclass(frozen=True)
 class LedgerWorkItem:
     """One outbox row.
@@ -102,6 +115,14 @@ class LedgerOutbox(Protocol):
         """Return pending work items in FIFO order, up to `limit`."""
         ...
 
+    def pending_cash_delta(self) -> dict[str, float]:
+        """Sum the estimated cash delta of ALL not-yet-settled (pending) rows,
+        grouped by asset_id (dcy-018 / REQ-029, OQ-4 estimate overlay).
+
+        buy → -(qty*price + fee + tax); sell → +(qty*price - fee - tax).
+        Used to show an estimated cash before option-D write-back lands."""
+        ...
+
     def get(self, correlation_id: str) -> Optional[LedgerWorkItem]:
         ...
 
@@ -142,6 +163,7 @@ class LedgerWriter:
         outbox: LedgerOutbox,
         ledger: LedgerPort,
         *,
+        snapshot_store=None,
         retry_base_sec: float = _DEFAULT_RETRY_BASE_SEC,
         retry_cap_sec: float = _DEFAULT_RETRY_CAP_SEC,
         max_attempts: int = _DEFAULT_MAX_ATTEMPTS,
@@ -150,6 +172,9 @@ class LedgerWriter:
     ) -> None:
         self._outbox = outbox
         self._ledger = ledger
+        # dcy-018 / REQ-019/030 (option D): SnapshotStore for cash/quantity
+        # write-back. None => write-back disabled (e.g. slice-1 / tests).
+        self._snapshot_store = snapshot_store
         # dcy-015 / REQ-019: exponential backoff + jitter + max-attempts cap.
         self._retry_base = retry_base_sec
         self._retry_cap = retry_cap_sec
@@ -189,7 +214,7 @@ class LedgerWriter:
             self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.DUPLICATE_COMPLETED)
             return OutboxOutcome.DUPLICATE_COMPLETED
         try:
-            result = self._ledger.record_trade(item.trade)
+            outcome = self._ledger.record_trade(item.trade)
         except LedgerConflictError as exc:
             self._outbox.mark_outcome(
                 item.correlation_id, OutboxOutcome.CONFLICT_BLOCKED, error=str(exc)
@@ -227,15 +252,39 @@ class LedgerWriter:
                 error=str(exc), next_retry_at=next_retry_at,
             )
             return OutboxOutcome.RETRYABLE_FAILURE
+        # dcy-018 / REQ-019/030 (option D): write back portfolio-hub authoritative
+        # cash/quantity to the snapshot store on new+duplicate (OQ-2 b). *_after
+        # is absolute, so applying it on duplicate is idempotent. best-effort —
+        # never let a write-back failure block the ledger outcome (NFR-1).
+        self._writeback(item.trade, outcome)
         # Adapter-level dedup (server says "already recorded"): outcome is
         # duplicate_completed; the ledger has exactly one entry for this
         # execution_id either way (REQ-019 AC-1, AC-4).
-        if result is LedgerWriteResult.DUPLICATE_OK:
+        if outcome.result is LedgerWriteResult.DUPLICATE_OK:
             self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.DUPLICATE_COMPLETED)
             return OutboxOutcome.DUPLICATE_COMPLETED
         self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.COMPLETED)
         return OutboxOutcome.COMPLETED
 
+    def _writeback(self, trade: LedgerTrade, outcome) -> None:
+        """Option D cash/quantity write-back (dcy-018). best-effort; skips when
+        no SnapshotStore is wired or the response omits authoritative values."""
+        if self._snapshot_store is None:
+            return
+        if outcome.cash_balance_after is None or outcome.asset_quantity_after is None:
+            return  # older hub / missing fields → leave snapshot as-is
+        try:
+            self._snapshot_store.update_cash_position(
+                trade.asset_id,
+                outcome.cash_balance_after,
+                outcome.asset_quantity_after,
+            )
+        except Exception:  # noqa: BLE001 - write-back must not block the ledger path
+            _log.warning(
+                "snapshot cash/position write-back failed: asset_id=%s",
+                trade.asset_id, exc_info=True,
+            )
+
 
 # --- in-memory outbox (tests + slice-1 runtime) ------------------------------
 
@@ -288,6 +337,15 @@ class InMemoryLedgerOutbox:
                         break
             return out
 
+    def pending_cash_delta(self) -> dict[str, float]:
+        with self._lock:
+            out: dict[str, float] = {}
+            for row in self._rows.values():
+                if row.status is WorkItemStatus.PENDING:
+                    aid = row.trade.asset_id
+                    out[aid] = out.get(aid, 0.0) + trade_cash_delta(row.trade)
+            return out
+
     def get(self, correlation_id: str) -> Optional[LedgerWorkItem]:
         with self._lock:
             return self._rows.get(correlation_id)

trex_core/application/ledger_port.py

@@ -29,6 +29,21 @@ class LedgerWriteResult(str, Enum):
     DUPLICATE_OK = "duplicate_ok"
 
 
+@dataclass(frozen=True)
+class LedgerWriteOutcome:
+    """Result of `record_trade` + portfolio-hub's authoritative cash/position
+    after the write (dcy-018 / REQ-017, option D).
+
+    portfolio-hub#8 returns `cash_balance_after`/`asset_quantity_after` (root
+    flat, absolute authoritative values) on `new`(201) and `duplicate`(200);
+    they are None when the response omits them (older hub) so write-back is
+    skipped. 409/422 raise instead of returning an outcome."""
+
+    result: LedgerWriteResult
+    cash_balance_after: Optional[float] = None
+    asset_quantity_after: Optional[float] = None
+
+
 @dataclass(frozen=True)
 class LedgerTrade:
     """One ledger trade record (dcy-004 FR-1 contract)."""
@@ -82,16 +97,18 @@ class LedgerPort(Protocol):
     stub mimics the same contract for deterministic tests.
     """
 
-    def record_trade(self, trade: LedgerTrade) -> LedgerWriteResult:
-        """Append a trade to the ledger.
+    def record_trade(self, trade: LedgerTrade) -> LedgerWriteOutcome:
+        """Append a trade to the ledger (dcy-018 / REQ-017).
 
         Returns:
-            LedgerWriteResult.NEW           on a fresh entry,
-            LedgerWriteResult.DUPLICATE_OK  when the adapter recognized the
-                                            execution_id as already recorded.
+            LedgerWriteOutcome — `.result` is NEW (fresh entry) or DUPLICATE_OK
+            (execution_id already recorded), plus `.cash_balance_after` /
+            `.asset_quantity_after` (portfolio-hub authoritative values, or None
+            when the response omits them, option D).
 
         Raises:
             LedgerConflictError: ledger reports an unrecoverable conflict (409).
+            LedgerPermanentError: non-retryable client error (404/400/422).
             LedgerRetryableError: transient failure; outbox should retry.
         """
         ...

trex_core/application/messaging_handlers.py

@@ -171,6 +171,7 @@ def make_snapshot_rpc_handler(
     watchdog_store: Optional[WatchdogStore] = None,
     watchdog_timeout_sec: int = 120,
     killswitch_store=None,
+    ledger_outbox=None,
 ):
     """Build the RPC handler for `trex.snapshot.req` (REQ-028).
 
@@ -229,6 +230,7 @@ def make_snapshot_rpc_handler(
             )
             payload = envelope.to_dict()
             _attach_watchdog(payload, slot_repo, day, watchdog_store, watchdog_timeout_sec)
+            _apply_cash_estimate(payload, ledger_outbox)
             return RpcResponse(payload=payload)
 
         # --- BAT-007 legacy stub-shape path ---
@@ -259,6 +261,27 @@ def make_snapshot_rpc_handler(
     return handler
 
 
+def _apply_cash_estimate(payload, ledger_outbox) -> None:
+    """Overlay an estimated cash for assets with unsettled (pending) ledger
+    writes (dcy-018 / REQ-029, OQ-4). cash_balance += pending cash delta and the
+    asset is flagged `cash_estimated=true` until option-D write-back lands.
+    Best-effort: any failure leaves the response unchanged (NFR-1)."""
+    if ledger_outbox is None:
+        return
+    try:
+        deltas = ledger_outbox.pending_cash_delta()
+        if not deltas:
+            return
+        for a in payload.get("assets", []):
+            d = deltas.get(a.get("asset_id"))
+            if d:
+                base = a.get("cash_balance") or 0.0
+                a["cash_balance"] = base + d
+                a["cash_estimated"] = True
+    except Exception:  # noqa: BLE001 - estimate must never break the response
+        return
+
+
 def _attach_watchdog(payload, slot_repo, day, watchdog_store, timeout_sec) -> None:
     """Attach holding-slot watchdog state to a snapshot response (dcy-017/REQ-034).
 

trex_core/runtime/daemon.py

@@ -147,15 +147,16 @@ class Daemon:
         )
         self.broker = _build_broker(config)
         self.outbox = PostgreSQLLedgerOutbox(config.dsn)
+        # BAT-008: portfolio-hub read + snapshot store + durable pause + scheduler.
+        # (Created before the ledger writer so option-D write-back can target it.)
+        self.snapshot_store = SnapshotStore()
         self.ledger_writer = LedgerWriter(
             self.outbox, _build_ledger(config),
+            snapshot_store=self.snapshot_store,  # dcy-018 / REQ-030: option D write-back
             retry_base_sec=config.ledger_retry_base_sec,
             retry_cap_sec=config.ledger_retry_cap_sec,
             max_attempts=config.ledger_max_attempts,
         )
-
-        # BAT-008: portfolio-hub read + snapshot store + durable pause + scheduler.
-        self.snapshot_store = SnapshotStore()
         # BAT-010/011: per-slot watchdog (snapshot RPC writer, safety-sell reader).
         self.watchdog = WatchdogStore()
         self.pause_store = PostgreSQLPauseStore(config.dsn)
@@ -273,6 +274,7 @@ class Daemon:
             watchdog_store=self.watchdog,
             watchdog_timeout_sec=self.config.safety_sell.watchdog_timeout_sec,
             killswitch_store=self.killswitch_store,
+            ledger_outbox=self.outbox,  # dcy-018 / REQ-029: estimated-cash overlay
         )
         transport.consume_rpc_forever(
             topology.snapshot_queue(self.config.instance_key), handler, self.stop)