feat: persist skipped ledger fills

Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>

trex_core/adapters/broker/kis.py

@@ -14,11 +14,11 @@ from __future__ import annotations
 import base64
 import hashlib
 import json
+from collections.abc import Callable
 from dataclasses import replace
-from datetime import datetime, timedelta, timezone
+from datetime import UTC, datetime, timedelta, timezone
 from decimal import Decimal, InvalidOperation
 from time import monotonic
-from typing import Callable, Optional
 
 import httpx
 
@@ -47,7 +47,7 @@ class KISEvidenceLookupError(RuntimeError):
     """Read-only daily-fill evidence lookup failed."""
 
 
-def _balance_http_error_code(status: Optional[int]) -> str:
+def _balance_http_error_code(status: int | None) -> str:
     if status == 401:
         return "authentication_failed"
     if status == 403:
@@ -69,7 +69,7 @@ def resolve_order_venue(venue: str, at: datetime) -> str:
     return normalized
 
 
-def _safe_int(v: object) -> Optional[int]:
+def _safe_int(v: object) -> int | None:
     try:
         return int(str(v).strip())
     except (TypeError, ValueError):
@@ -92,7 +92,7 @@ def _encode_submission_unknown(req: OrderRequest, submitted_at: datetime) -> str
     return f"{_SUBMISSION_UNKNOWN_PREFIX}{encoded}"
 
 
-def _decode_submission_unknown(broker_order_id: str) -> Optional[dict]:
+def _decode_submission_unknown(broker_order_id: str) -> dict | None:
     if not broker_order_id.startswith(_SUBMISSION_UNKNOWN_PREFIX):
         return None
     encoded = broker_order_id[len(_SUBMISSION_UNKNOWN_PREFIX) :]
@@ -138,15 +138,15 @@ class KISBroker:
         timeout: float = 5.0,
         allow_live: bool = False,
         token_cache=None,
-        clock: Optional[Callable[[], datetime]] = None,
+        clock: Callable[[], datetime] | None = None,
     ) -> None:
         self._account_ref = account_ref
         self._app_key = app_key
         self._app_secret = app_secret
         self._base_url = base_url.rstrip("/")
         self._timeout = timeout
-        self._token: Optional[str] = None
-        self._token_exp: datetime = datetime.now(tz=timezone.utc)
+        self._token: str | None = None
+        self._token_exp: datetime = datetime.now(tz=UTC)
         self._token_cache = token_cache  # optional TokenCache (L2, REQ-016)
         self._clock = clock or (lambda: datetime.now(tz=_KST))
         self._fill_poll_not_before: dict[str, float] = {}
@@ -197,12 +197,12 @@ class KISBroker:
         self._app_secret = credential.app_secret
         # Invalidate L1 in-memory token so next call re-issues.
         self._token = None
-        self._token_exp = datetime.min.replace(tzinfo=timezone.utc)
+        self._token_exp = datetime.min.replace(tzinfo=UTC)
 
     # --- auth ---
     def _ensure_token(self) -> str:
         # L1: in-memory
-        if self._token and datetime.now(tz=timezone.utc) < self._token_exp:
+        if self._token and datetime.now(tz=UTC) < self._token_exp:
             return self._token
         # L2: shared/persistent cache (Redis) — survives restart / multi-process (REQ-016)
         if self._token_cache is not None:
@@ -224,7 +224,7 @@ class KISBroker:
         data = resp.json()
         self._token = data["access_token"]
         ttl = int(data.get("expires_in", 3600))
-        self._token_exp = datetime.now(tz=timezone.utc) + timedelta(seconds=max(60, ttl - 300))
+        self._token_exp = datetime.now(tz=UTC) + timedelta(seconds=max(60, ttl - 300))
         if self._token_cache is not None:
             self._token_cache.store(self._account_ref, self._app_key, self._token, self._token_exp)
         return self._token
@@ -364,8 +364,8 @@ class KISBroker:
         *,
         side: str,
         symbol: str,
-        order_date: Optional[str] = None,
-        submitted_at: Optional[datetime] = None,
+        order_date: str | None = None,
+        submitted_at: datetime | None = None,
     ) -> FillStatus:
         if broker_order_id.startswith(_SUBMISSION_UNKNOWN_PREFIX):
             return self._poll_unknown_submission(broker_order_id, side=side, symbol=symbol)
@@ -424,7 +424,7 @@ class KISBroker:
         if row is None:
             visible_grace = (
                 submitted_at is None
-                or (self._clock().astimezone(timezone.utc) - submitted_at.astimezone(timezone.utc))
+                or (self._clock().astimezone(UTC) - submitted_at.astimezone(UTC))
                 .total_seconds()
                 < _SUBMISSION_LOOKUP_GRACE_SECONDS
             )
@@ -639,11 +639,21 @@ class KISBroker:
     def _executed_at_from_row(row: dict, fallback_date: str) -> datetime:
         order_date = str(row.get("ord_dt") or fallback_date).strip()
         order_time = str(row.get("ord_tmd") or "").strip()[:6]
+        year = int(order_date[0:4])
+        month = int(order_date[4:6])
+        day = int(order_date[6:8])
         if len(order_time) == 6 and order_time.isdigit():
-            executed_at = datetime.strptime(order_date + order_time, "%Y%m%d%H%M%S")
-            return executed_at.replace(tzinfo=_KST).astimezone(timezone.utc)
-        executed_at = datetime.strptime(order_date, "%Y%m%d")
-        return executed_at.replace(tzinfo=_KST).astimezone(timezone.utc)
+            executed_at = datetime(
+                year,
+                month,
+                day,
+                int(order_time[0:2]),
+                int(order_time[2:4]),
+                int(order_time[4:6]),
+                tzinfo=_KST,
+            )
+            return executed_at.astimezone(UTC)
+        return datetime(year, month, day, tzinfo=_KST).astimezone(UTC)
 
     def lookup_fill_evidence(self, *, trading_date: str, execution_id: str) -> dict:
         """Read one exact filled/partial-cancel evidence row for a historical execution."""
@@ -737,7 +747,7 @@ class KISBroker:
             if cancelled:
                 return FillStatus(state="cancelled", raw=row)  # full external cancel
             return FillStatus(state="accepted", raw=row)
-        if ccld_qty >= ord_qty and ord_qty > 0:
+        if ccld_qty >= ord_qty > 0:
             # Deterministic execution_id (dcy-004 OQ-3, orbit kis.py rule):
             #   kis-daily-ccld:{ord_dt}:{ord_gno_brno}:{odno}:{tot_ccld_qty}
             ord_dt = str(row.get("ord_dt") or today)
@@ -760,6 +770,7 @@ class KISBroker:
                 fee=fee,
                 tax=tax,
                 execution_venue=execution_venue,
+                executed_at=KISBroker._executed_at_from_row(row, ord_dt),
                 raw=row,
             )
         # 0 < ccld_qty < ord_qty
@@ -784,6 +795,7 @@ class KISBroker:
                 fee=fee,
                 tax=tax,
                 execution_venue=execution_venue,
+                executed_at=KISBroker._executed_at_from_row(row, ord_dt),
                 cancelled=True,
                 raw=row,
             )
@@ -827,7 +839,7 @@ class KISBroker:
 
         quantities: dict[str, int] = {}
         seen_continuations: set[tuple[str, str]] = set()
-        continuation: Optional[tuple[str, str]] = None
+        continuation: tuple[str, str] | None = None
         while True:
             request_headers = headers
             if continuation is not None:
@@ -944,7 +956,7 @@ class KISBroker:
             raise BrokerBalanceError("malformed_response")
         return amount
 
-    def get_current_price(self, symbol: str, market: str = "J") -> Optional[Decimal]:
+    def get_current_price(self, symbol: str, market: str = "J") -> Decimal | None:
         """Current traded price `stck_prpr` via KIS inquire-price (REQ-038, REQ-118).
 
         tr_id: `FHKST01010100` per REQ-038 AC-7 (live and mock share the same tr_id).

trex_core/adapters/broker/paper.py

@@ -8,9 +8,8 @@ execution code path that production (KIS) takes.
 from __future__ import annotations
 
 import threading
+from datetime import UTC, datetime
 from decimal import Decimal
-from datetime import datetime
-from typing import Optional
 
 from trex_core.adapters.broker.seq_store import SeqStore
 from trex_core.application.broker_port import (
@@ -52,9 +51,9 @@ class PaperBroker:
         self,
         *,
         fill_after_polls: int = 0,
-        qty_per_poll: Optional[int] = None,
-        reject_symbols: Optional[set[str]] = None,
-        seq_store: Optional[SeqStore] = None,
+        qty_per_poll: int | None = None,
+        reject_symbols: set[str] | None = None,
+        seq_store: SeqStore | None = None,
     ) -> None:
         self._fill_after = fill_after_polls
         self._qty_per_poll = qty_per_poll
@@ -90,8 +89,8 @@ class PaperBroker:
         *,
         side: str,
         symbol: str,
-        order_date: Optional[str] = None,
-        submitted_at: Optional[datetime] = None,
+        order_date: str | None = None,
+        submitted_at: datetime | None = None,
     ) -> FillStatus:
         with self._lock:
             o = self._orders.get(broker_order_id)
@@ -122,6 +121,7 @@ class PaperBroker:
                     state="filled", filled_qty=o.filled, avg_price=o.price,
                     execution_id=execution_id, fee=fee, tax=tax,
                     execution_venue=o.execution_venue,
+                    executed_at=datetime.now(tz=UTC),
                 )
             return FillStatus(
                 state="partially_filled",
@@ -172,7 +172,7 @@ class PaperBroker:
         with self._lock:
             self._prices[symbol] = Decimal(str(price))
 
-    def get_current_price(self, symbol: str, market: str = "J") -> Optional[Decimal]:
+    def get_current_price(self, symbol: str, market: str = "J") -> Decimal | None:
         """Return the programmed current price for symbol, or None if unset.
 
         market accepted for BrokerPort protocol conformance (REQ-118) but

trex_core/adapters/persistence/ledger_skip_store.py

@@ -0,0 +1,247 @@
+from __future__ import annotations
+
+from datetime import UTC, date, datetime, time, timedelta
+from typing import Any
+from uuid import UUID
+
+from trex_core.adapters.persistence.postgresql import _require_psycopg
+from trex_core.application.ledger_skip import LedgerSkipRecord, make_ledger_skip_key
+
+
+class PostgreSQLLedgerSkipStore:
+    def __init__(self, dsn: str, *, psycopg_module: Any = None) -> None:
+        self._psycopg = _require_psycopg(psycopg_module)
+        self._dsn = dsn
+
+    def record_skip(
+        self,
+        *,
+        instance_key: str,
+        slot_id: UUID,
+        portfolio_id: str | None,
+        symbol: str | None,
+        order_id: str | None,
+        execution_id: str | None,
+        side: str,
+        qty: float,
+        price: float,
+        executed_at: datetime,
+        order_venue: str | None,
+        execution_venue: str | None,
+        skip_reason: str,
+    ) -> tuple[LedgerSkipRecord, bool]:
+        skip_key = make_ledger_skip_key(
+            instance_key=instance_key,
+            slot_id=slot_id,
+            side=side,
+            order_id=order_id,
+            execution_id=execution_id,
+            skip_reason=skip_reason,
+        )
+        now = datetime.now(tz=UTC)
+        notify = False
+        with self._psycopg.connect(self._dsn, autocommit=False) as conn, conn.cursor() as cur:
+            cur.execute(
+                "SELECT status FROM ledger_skip WHERE skip_key = %s FOR UPDATE",
+                (skip_key,),
+            )
+            row = cur.fetchone()
+            if row is None:
+                notify = True
+                cur.execute(
+                    "INSERT INTO ledger_skip "
+                    "(skip_key, instance_key, slot_id, portfolio_id, symbol, order_id, "
+                    " execution_id, side, qty, price, executed_at, order_venue, "
+                    " execution_venue, skip_reason, status, created_at, updated_at) "
+                    "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, "
+                    "        'unresolved', %s, %s)",
+                    (
+                        skip_key,
+                        instance_key,
+                        slot_id,
+                        portfolio_id,
+                        symbol,
+                        order_id,
+                        execution_id,
+                        side,
+                        qty,
+                        price,
+                        executed_at,
+                        order_venue or "unknown",
+                        execution_venue or "unknown",
+                        skip_reason,
+                        now,
+                        now,
+                    ),
+                )
+            else:
+                notify = row[0] == "resolved"
+                cur.execute(
+                    "UPDATE ledger_skip SET portfolio_id=%s, symbol=%s, order_id=%s, "
+                    "execution_id=%s, qty=%s, price=%s, executed_at=%s, "
+                    "order_venue=%s, execution_venue=%s, status='unresolved', "
+                    "resolution_reason=NULL, resolved_at=NULL, updated_at=%s "
+                    "WHERE skip_key=%s",
+                    (
+                        portfolio_id,
+                        symbol,
+                        order_id,
+                        execution_id,
+                        qty,
+                        price,
+                        executed_at,
+                        order_venue or "unknown",
+                        execution_venue or "unknown",
+                        now,
+                        skip_key,
+                    ),
+                )
+            conn.commit()
+        record = self.get(skip_key)
+        assert record is not None
+        return record, notify
+
+    def get(self, skip_key: str) -> LedgerSkipRecord | None:
+        with self._psycopg.connect(self._dsn) as conn, conn.cursor() as cur:
+            cur.execute(
+                "SELECT skip_key, instance_key, slot_id, portfolio_id, symbol, order_id, "
+                "execution_id, side, qty, price, executed_at, order_venue, execution_venue, "
+                "skip_reason, status, resolution_reason, resolved_at, created_at, updated_at "
+                "FROM ledger_skip WHERE skip_key = %s",
+                (skip_key,),
+            )
+            row = cur.fetchone()
+        return self._row_to_record(row) if row else None
+
+    def list_records(
+        self,
+        *,
+        instance_key: str | None = None,
+        portfolio_id: str | None = None,
+        on_date: date | None = None,
+        status: str = "unresolved",
+        limit: int = 100,
+    ) -> list[LedgerSkipRecord]:
+        clauses: list[str] = []
+        params: list[object] = []
+        if status != "all":
+            clauses.append("status = %s")
+            params.append(status)
+        if instance_key is not None:
+            clauses.append("instance_key = %s")
+            params.append(instance_key)
+        if portfolio_id is not None:
+            clauses.append("portfolio_id = %s")
+            params.append(portfolio_id)
+        if on_date is not None:
+            start = datetime.combine(on_date, time.min, tzinfo=UTC)
+            end = start + timedelta(days=1)
+            clauses.append("created_at >= %s AND created_at < %s")
+            params.extend([start, end])
+        where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
+        with self._psycopg.connect(self._dsn) as conn, conn.cursor() as cur:
+            cur.execute(
+                "SELECT skip_key, instance_key, slot_id, portfolio_id, symbol, order_id, "
+                "execution_id, side, qty, price, executed_at, order_venue, execution_venue, "
+                "skip_reason, status, resolution_reason, resolved_at, created_at, updated_at "
+                f"FROM ledger_skip {where} ORDER BY created_at DESC LIMIT %s",
+                (*params, limit),
+            )
+            rows = cur.fetchall()
+        return [self._row_to_record(row) for row in rows]
+
+    def resolve_for_fill(
+        self,
+        *,
+        instance_key: str,
+        slot_id: UUID,
+        side: str,
+        order_id: str | None,
+        execution_id: str | None,
+        resolution_reason: str,
+    ) -> list[LedgerSkipRecord]:
+        now = datetime.now(tz=UTC)
+        with self._psycopg.connect(self._dsn, autocommit=False) as conn, conn.cursor() as cur:
+            if execution_id:
+                cur.execute(
+                    "UPDATE ledger_skip SET status='resolved', resolution_reason=%s, "
+                    "resolved_at=%s, updated_at=%s "
+                    "WHERE status='unresolved' AND instance_key=%s AND slot_id=%s "
+                    "AND side=%s AND execution_id=%s "
+                    "RETURNING skip_key, instance_key, slot_id, portfolio_id, symbol, "
+                    "order_id, execution_id, side, qty, price, executed_at, order_venue, "
+                    "execution_venue, skip_reason, status, resolution_reason, resolved_at, "
+                    "created_at, updated_at",
+                    (
+                        resolution_reason,
+                        now,
+                        now,
+                        instance_key,
+                        slot_id,
+                        side,
+                        execution_id,
+                    ),
+                )
+            else:
+                cur.execute(
+                    "UPDATE ledger_skip SET status='resolved', resolution_reason=%s, "
+                    "resolved_at=%s, updated_at=%s "
+                    "WHERE status='unresolved' AND instance_key=%s AND slot_id=%s "
+                    "AND side=%s AND order_id=%s "
+                    "RETURNING skip_key, instance_key, slot_id, portfolio_id, symbol, "
+                    "order_id, execution_id, side, qty, price, executed_at, order_venue, "
+                    "execution_venue, skip_reason, status, resolution_reason, resolved_at, "
+                    "created_at, updated_at",
+                    (
+                        resolution_reason,
+                        now,
+                        now,
+                        instance_key,
+                        slot_id,
+                        side,
+                        order_id,
+                    ),
+                )
+            rows = cur.fetchall()
+            conn.commit()
+        return [self._row_to_record(row) for row in rows]
+
+    def resolve_by_key(self, skip_key: str, *, resolution_reason: str) -> LedgerSkipRecord | None:
+        now = datetime.now(tz=UTC)
+        with self._psycopg.connect(self._dsn, autocommit=False) as conn, conn.cursor() as cur:
+            cur.execute(
+                "UPDATE ledger_skip SET status='resolved', resolution_reason=%s, "
+                "resolved_at=%s, updated_at=%s "
+                "WHERE skip_key=%s AND status='unresolved' "
+                "RETURNING skip_key, instance_key, slot_id, portfolio_id, symbol, order_id, "
+                "execution_id, side, qty, price, executed_at, order_venue, execution_venue, "
+                "skip_reason, status, resolution_reason, resolved_at, created_at, updated_at",
+                (resolution_reason, now, now, skip_key),
+            )
+            row = cur.fetchone()
+            conn.commit()
+        return self._row_to_record(row) if row else None
+
+    @staticmethod
+    def _row_to_record(row: tuple) -> LedgerSkipRecord:
+        return LedgerSkipRecord(
+            skip_key=row[0],
+            instance_key=row[1],
+            slot_id=row[2],
+            portfolio_id=row[3],
+            symbol=row[4],
+            order_id=row[5],
+            execution_id=row[6],
+            side=row[7],
+            qty=float(row[8]),
+            price=float(row[9]),
+            executed_at=row[10],
+            order_venue=row[11],
+            execution_venue=row[12],
+            skip_reason=row[13],
+            status=row[14],
+            resolution_reason=row[15],
+            resolved_at=row[16],
+            created_at=row[17],
+            updated_at=row[18],
+        )

trex_core/adapters/persistence/sql/025_create_ledger_skip.sql

@@ -0,0 +1,30 @@
+CREATE TABLE IF NOT EXISTS ledger_skip (
+    skip_key TEXT PRIMARY KEY,
+    instance_key TEXT NOT NULL,
+    slot_id UUID NOT NULL,
+    portfolio_id TEXT NULL,
+    symbol TEXT NULL,
+    order_id TEXT NULL,
+    execution_id TEXT NULL,
+    side TEXT NOT NULL CHECK (side IN ('buy', 'sell')),
+    qty DOUBLE PRECISION NOT NULL,
+    price DOUBLE PRECISION NOT NULL,
+    executed_at TIMESTAMPTZ NOT NULL,
+    order_venue TEXT NOT NULL,
+    execution_venue TEXT NOT NULL,
+    skip_reason TEXT NOT NULL,
+    status TEXT NOT NULL CHECK (status IN ('unresolved', 'resolved')),
+    resolution_reason TEXT NULL,
+    resolved_at TIMESTAMPTZ NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_ledger_skip_status_created
+    ON ledger_skip (status, created_at);
+
+CREATE INDEX IF NOT EXISTS idx_ledger_skip_instance_portfolio_created
+    ON ledger_skip (instance_key, portfolio_id, created_at);
+
+CREATE INDEX IF NOT EXISTS idx_ledger_skip_execution_id
+    ON ledger_skip (execution_id);

trex_core/application/broker_port.py

@@ -107,6 +107,7 @@ class FillStatus:
     # Broker-authoritative execution exchange. For SOR orders this is KRX or
     # NXT; None means the polling surface did not establish it.
     execution_venue: Optional[str] = None
+    executed_at: Optional[datetime] = None
     # dcy-013 / REQ-012: an external cancel was detected.
     # order had a partial fill before the cancel, state stays "partially_filled"
     # and this flag is set so execution records the filled portion and returns

trex_core/application/execution.py

@@ -25,9 +25,9 @@ from __future__ import annotations
 
 import inspect
 import logging
+from collections.abc import Callable
 from dataclasses import dataclass, replace
-from datetime import datetime, timedelta, timezone
-from typing import Callable, Optional
+from datetime import UTC, datetime, timedelta
 
 from trex_core.application.alarm_emitter import AlarmEmitter
 from trex_core.application.broker_port import (
@@ -39,6 +39,7 @@ from trex_core.application.broker_port import (
 )
 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
 from trex_core.domain.pricing import ceil_to_tick, floor_to_tick
 from trex_core.domain.slot import SlotService, SlotState, SlotStateMismatchError
 from trex_core.runtime.logging_setup import set_correlation_id
@@ -57,7 +58,7 @@ _log = logging.getLogger(__name__)
 # `slot.strategy_id` so that slot_filled_sell events reach the correct
 # per-strategy queue instead of being NO_ROUTEd to the DLQ.
 SlotEventEmitter = Callable[
-    [str, object, Optional[str]], None
+    [str, object, str | None], None
 ]  # (event_type, slot, routing_strategy_id) -> None
 
 _MAX_DISPATCH_PER_TICK = 64
@@ -81,13 +82,14 @@ class ExecutionService:
         slot_service: SlotService,
         repo,
         broker: BrokerPort,
-        ledger_outbox: Optional[LedgerOutbox] = None,
-        event_emitter: Optional[SlotEventEmitter] = None,
-        stale_pending_threshold_sec: Optional[int] = None,
+        ledger_outbox: LedgerOutbox | None = None,
+        event_emitter: SlotEventEmitter | None = None,
+        stale_pending_threshold_sec: int | None = None,
         instance_key: str = "default",
-        alarm_emitter: Optional[AlarmEmitter] = None,
-        balance_reader: Optional[Callable[[], list[BalanceItem]]] = None,
-        clock: Optional[Callable[[], datetime]] = None,
+        alarm_emitter: AlarmEmitter | None = None,
+        ledger_skip_store=None,
+        balance_reader: Callable[[], list[BalanceItem]] | None = None,
+        clock: Callable[[], datetime] | None = None,
     ) -> None:
         self._svc = slot_service
         self._repo = repo
@@ -108,11 +110,12 @@ class ExecutionService:
         )
         # dcy-033 / REQ-100: best-effort alarm emitter (None = no alarms sent).
         self._alarm_emitter = alarm_emitter
+        self._ledger_skip_store = ledger_skip_store
         # One emission per durable unresolved state/attempt in this process.
         # A restart intentionally emits once again, covering a crash between a
         # durable cancel reservation and best-effort alarm publication.
         self._order_alarm_keys: set[tuple[str, str, int]] = set()
-        self._clock = clock or (lambda: datetime.now(tz=timezone.utc))
+        self._clock = clock or (lambda: datetime.now(tz=UTC))
         # 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.
@@ -137,7 +140,7 @@ class ExecutionService:
 
     # ------------------------------------------------------------
     def _dispatch(self, r: ExecutionResult) -> None:
-        sell_balance: Optional[dict[str, int]] = None  # lazy, once per tick
+        sell_balance: dict[str, int] | None = None  # lazy, once per tick
         for slot in self._repo.list_undispatched_pending(
             self._instance_key, _MAX_DISPATCH_PER_TICK
         ):
@@ -266,7 +269,7 @@ class ExecutionService:
                 r.rolled_back += 1
 
     def _poll(self, r: ExecutionResult) -> None:
-        now = self._clock().astimezone(timezone.utc)
+        now = self._clock().astimezone(UTC)
         for slot in self._repo.list_dispatched_pending(self._instance_key):
             side = "buy" if slot.status is SlotState.PENDING_BUY else "sell"
             set_correlation_id(str(slot.slot_id))  # REQ-084: same key as submit
@@ -716,7 +719,7 @@ class ExecutionService:
                 },
             )
 
-    def _emit_event(self, event_type: str, slot, routing_strategy_id: Optional[str] = None) -> None:
+    def _emit_event(self, event_type: str, slot, routing_strategy_id: str | None = None) -> None:
         if self._event_emitter is None:
             return
         try:
@@ -754,6 +757,14 @@ class ExecutionService:
                 slot.symbol,
                 side,
             )
+            self._persist_ledger_skip(
+                slot,
+                fs,
+                side,
+                broker_order_id,
+                skip_reason="missing_portfolio_id",
+                price=resolved_price,
+            )
             return
         if fs.execution_id is None:
             # Broker contract regression — slot already transitioned; surface for ops.
@@ -765,6 +776,14 @@ class ExecutionService:
                 slot.portfolio_id,
                 side,
             )
+            self._persist_ledger_skip(
+                slot,
+                fs,
+                side,
+                broker_order_id,
+                skip_reason="missing_execution_id",
+                price=resolved_price,
+            )
             return
         # REQ-095 KD-3: use pre-resolved price when available; fall back to
         # fs.avg_price for any direct callers that do not pass resolved_price.
@@ -787,6 +806,15 @@ class ExecutionService:
                 execution_venue,
                 slot.venue,
             )
+            self._persist_ledger_skip(
+                slot,
+                fs,
+                side,
+                broker_order_id,
+                skip_reason="unsupported_execution_venue",
+                price=price,
+                execution_venue=execution_venue,
+            )
             return
         trade = LedgerTrade(
             asset_id=str(slot.portfolio_id),
@@ -797,12 +825,68 @@ class ExecutionService:
             price=price,
             fee=float(fs.fee),
             tax=float(fs.tax),
-            executed_at=datetime.now(tz=timezone.utc),
+            executed_at=fs.executed_at or self._clock(),
             venue=execution_venue,
             order_venue=slot.venue,
         )
         self._ledger_outbox.enqueue(trade)
         r.ledger_enqueued += 1
+        self._resolve_ledger_skip(slot, side, broker_order_id, str(fs.execution_id))
+
+    def _persist_ledger_skip(
+        self,
+        slot,
+        fs,
+        side: str,
+        broker_order_id: str,
+        *,
+        skip_reason: str,
+        price: float | None,
+        execution_venue: str | None = None,
+    ) -> None:
+        if self._ledger_skip_store is None:
+            return
+        record, should_notify = self._ledger_skip_store.record_skip(
+            instance_key=self._instance_key,
+            slot_id=slot.slot_id,
+            portfolio_id=slot.portfolio_id,
+            symbol=slot.symbol,
+            order_id=str(broker_order_id) if broker_order_id else None,
+            execution_id=str(fs.execution_id) if fs.execution_id is not None else None,
+            side=side,
+            qty=float(fs.filled_qty),
+            price=(
+                float(price)
+                if price is not None
+                else (float(fs.avg_price) if fs.avg_price is not None else 0.0)
+            ),
+            executed_at=fs.executed_at or self._clock(),
+            order_venue=getattr(slot, "venue", None) or "unknown",
+            execution_venue=execution_venue or getattr(fs, "execution_venue", None) or "unknown",
+            skip_reason=skip_reason,
+        )
+        if should_notify:
+            publish_ledger_skip_alarm(self._alarm_emitter, record, resolved=False)
+
+    def _resolve_ledger_skip(
+        self,
+        slot,
+        side: str,
+        broker_order_id: str,
+        execution_id: str | None,
+    ) -> None:
+        if self._ledger_skip_store is None:
+            return
+        records = self._ledger_skip_store.resolve_for_fill(
+            instance_key=self._instance_key,
+            slot_id=slot.slot_id,
+            side=side,
+            order_id=str(broker_order_id) if broker_order_id else None,
+            execution_id=execution_id,
+            resolution_reason="ledger_enqueued",
+        )
+        for record in records:
+            publish_ledger_skip_alarm(self._alarm_emitter, record, resolved=True)
 
     def cancel_pending_slot(self, slot_id) -> str:
         """Operator backstop: send cancel_order for a pending slot (dcy-022/REQ-071).

trex_core/application/ledger_skip.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime
+from uuid import UUID
+
+from trex_core.application.alarm_emitter import AlarmEmitter
+
+
+@dataclass(frozen=True)
+class LedgerSkipRecord:
+    skip_key: str
+    instance_key: str
+    slot_id: UUID
+    portfolio_id: str | None
+    symbol: str | None
+    order_id: str | None
+    execution_id: str | None
+    side: str
+    qty: float
+    price: float
+    executed_at: datetime
+    order_venue: str
+    execution_venue: str
+    skip_reason: str
+    status: str
+    resolution_reason: str | None
+    resolved_at: datetime | None
+    created_at: datetime
+    updated_at: datetime
+
+
+def make_ledger_skip_key(
+    *,
+    instance_key: str,
+    slot_id: UUID,
+    side: str,
+    order_id: str | None,
+    execution_id: str | None,
+    skip_reason: str,
+) -> str:
+    return ":".join(
+        [
+            "ledger-skip",
+            instance_key,
+            str(slot_id),
+            side,
+            order_id or "-",
+            execution_id or "-",
+            skip_reason,
+        ]
+    )
+
+
+def ledger_skip_alarm_fields(record: LedgerSkipRecord) -> dict[str, object]:
+    return {
+        "skip_key": record.skip_key,
+        "instance_key": record.instance_key,
+        "slot_id": str(record.slot_id),
+        "portfolio_id": record.portfolio_id,
+        "symbol": record.symbol,
+        "order_id": record.order_id,
+        "execution_id": record.execution_id,
+        "side": record.side,
+        "qty": record.qty,
+        "price": record.price,
+        "executed_at": record.executed_at.isoformat(),
+        "order_venue": record.order_venue,
+        "execution_venue": record.execution_venue,
+        "skip_reason": record.skip_reason,
+        "status": record.status,
+        "resolution_reason": record.resolution_reason,
+        "resolved_at": record.resolved_at.isoformat() if record.resolved_at else None,
+        "created_at": record.created_at.isoformat(),
+        "updated_at": record.updated_at.isoformat(),
+    }
+
+
+def publish_ledger_skip_alarm(
+    alarm_emitter: AlarmEmitter | None,
+    record: LedgerSkipRecord,
+    *,
+    resolved: bool,
+) -> None:
+    if alarm_emitter is None:
+        return
+    code = "ledger_skip_resolved" if resolved else "ledger_skip_unresolved"
+    title = (
+        f"원장 skip 해소: slot={record.slot_id} reason={record.resolution_reason or 'resolved'}"
+        if resolved
+        else f"원장 skip 발생: slot={record.slot_id} reason={record.skip_reason}"
+    )
+    alarm_emitter.publish_alarm(
+        code,
+        title,
+        fields=ledger_skip_alarm_fields(record),
+        severity="info" if resolved else "critical",
+    )

trex_core/runtime/cli.py

@@ -12,6 +12,8 @@ Subcommands:
   sync                    — sync slot qty to broker ground truth (direct-DB; REQ-105)
   verify                  — re-check C-case reconcile-warning (direct-DB; REQ-106)
   ledger backfill         — evidence-backed historical ledger recovery
+  ledger skips list       — query durable skipped-ledger records
+  ledger skips resolve    — explicitly resolve a skipped-ledger record
 
 universe/slots/enter/exit go through the daemon's RabbitMQ RPC surface so the
 single-writer admission + invariants are preserved (no direct slot writes,
@@ -28,24 +30,27 @@ live accounts require `--yes --live` (double ack); `--dry-run` previews
 before/after without writing. sync is direct-DB (no RPC).
 verify (REQ-106): read-centric (no --yes/--live gate needed).
 ledger backfill: verifies KIS daily-ccld evidence, reconstructs one ledger
-trade, enqueues it through ledger_outbox, and immediately drives the
-LedgerWriter path with an audit JSON report.
+trade, enqueues it through ledger_outbox, immediately drives the
+LedgerWriter path, and resolves matching durable skipped-ledger records.
 """
 from __future__ import annotations
 
 import argparse
 import json as _json
+import logging
 import re
 import sys
 import uuid
-from datetime import datetime, timedelta, timezone
+from datetime import UTC, datetime, timedelta, timezone
 from pathlib import Path
 from zoneinfo import ZoneInfo
 
 from trex_core.adapters.messaging import topology
 from trex_core.adapters.portfolio.pause_store import PostgreSQLPauseStore
+from trex_core.application.alarm_emitter import AlarmEmitter
 from trex_core.application.ledger_outbox import LedgerWriter, OutboxOutcome
 from trex_core.application.ledger_port import LedgerTrade
+from trex_core.application.ledger_skip import ledger_skip_alarm_fields, publish_ledger_skip_alarm
 from trex_core.runtime.config import (
     DaemonConfig,
     DaemonInstanceConfig,
@@ -55,6 +60,7 @@ from trex_core.runtime.config import (
 
 _DEFAULT_TIMEOUT = 10.0
 _DEFAULT_LEDGER_BACKFILL_AUDIT_DIR = "artifacts/validation/ledger_backfill"
+_log = logging.getLogger(__name__)
 
 
 # --------------------------------------------------------------------------- #
@@ -91,6 +97,14 @@ def _default_rpc(cfg: DaemonConfig):
     return _rpc
 
 
+def _default_alarm_emitter(cfg: DaemonConfig):
+    from trex_core.adapters.messaging import InMemoryTransport, RabbitMQTransport
+
+    mq = getattr(cfg, "rabbitmq", None)
+    transport = RabbitMQTransport(mq.url, timeout=mq.timeout) if mq else InMemoryTransport()
+    return AlarmEmitter(transport), transport
+
+
 def _default_confirm(prompt: str) -> bool:
     try:
         return input(f"{prompt} [y/N] ").strip().lower() in ("y", "yes")
@@ -376,7 +390,7 @@ def _sanitize_path_component(value: str, *, limit: int = 80) -> str:
 
 def _ledger_backfill_audit_path(args, execution_id: str) -> Path:
     base = Path(getattr(args, "audit_dir", _DEFAULT_LEDGER_BACKFILL_AUDIT_DIR))
-    stamp = datetime.now(tz=timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+    stamp = datetime.now(tz=UTC).strftime("%Y%m%dT%H%M%SZ")
     name = (
         f"{stamp}--{_sanitize_path_component(_resolve_instance(args))}"
         f"--{_sanitize_path_component(execution_id)}.json"
@@ -393,37 +407,30 @@ def _write_ledger_backfill_audit(args, report: dict) -> str:
 
 
 def _lookup_ledger_backfill_evidence(cfg: DaemonConfig, *, trading_date: str, execution_id: str):
-    from trex_core.runtime.broker_provider import (
-        BrokerCredentialUnavailable,
-        build_broker_provider,
-    )
+    from trex_core.runtime.broker_provider import build_broker_provider
 
     broker, source = build_broker_provider(cfg).get_broker_for_read()
     lookup = getattr(broker, "lookup_fill_evidence", None)
     if not callable(lookup):
-        raise RuntimeError("configured broker does not support KIS ledger backfill evidence lookup")
-    try:
-        evidence = lookup(trading_date=trading_date, execution_id=execution_id)
-    except BrokerCredentialUnavailable:
-        raise
+        raise TypeError("configured broker does not support KIS ledger backfill evidence lookup")
+    evidence = lookup(trading_date=trading_date, execution_id=execution_id)
     return evidence, source
 
 
 def _resolve_ledger_backfill_slot(dsn: str, *, instance: str, symbol: str) -> dict:
     import psycopg
 
-    with psycopg.connect(dsn) as conn:
-        with conn.cursor() as cur:
-            cur.execute(
-                "SELECT slot_id::text, portfolio_id, status "
-                "FROM daily_slot "
-                "WHERE instance_key = %s AND symbol = %s "
-                "AND portfolio_id IS NOT NULL AND retired_at IS NULL "
-                "ORDER BY updated_at DESC NULLS LAST, slot_id ASC "
-                "LIMIT 2",
-                (instance, symbol),
-            )
-            rows = cur.fetchall()
+    with psycopg.connect(dsn) as conn, conn.cursor() as cur:
+        cur.execute(
+            "SELECT slot_id::text, portfolio_id, status "
+            "FROM daily_slot "
+            "WHERE instance_key = %s AND symbol = %s "
+            "AND portfolio_id IS NOT NULL AND retired_at IS NULL "
+            "ORDER BY updated_at DESC NULLS LAST, slot_id ASC "
+            "LIMIT 2",
+            (instance, symbol),
+        )
+        rows = cur.fetchall()
     if not rows:
         raise RuntimeError(f"no daemon-owned slot found for instance={instance} symbol={symbol}")
     if len(rows) != 1:
@@ -439,16 +446,15 @@ def _resolve_ledger_backfill_slot(dsn: str, *, instance: str, symbol: str) -> di
 def _lookup_outbox_by_execution_id(dsn: str, *, execution_id: str) -> list[dict]:
     import psycopg
 
-    with psycopg.connect(dsn) as conn:
-        with conn.cursor() as cur:
-            cur.execute(
-                "SELECT correlation_id, asset_id, order_id, execution_id, status, "
-                "attempts, last_outcome, last_error "
-                "FROM ledger_outbox WHERE execution_id = %s "
-                "ORDER BY created_at, correlation_id",
-                (execution_id,),
-            )
-            rows = cur.fetchall()
+    with psycopg.connect(dsn) as conn, conn.cursor() as cur:
+        cur.execute(
+            "SELECT correlation_id, asset_id, order_id, execution_id, status, "
+            "attempts, last_outcome, last_error "
+            "FROM ledger_outbox WHERE execution_id = %s "
+            "ORDER BY created_at, correlation_id",
+            (execution_id,),
+        )
+        rows = cur.fetchall()
     return [
         {
             "correlation_id": str(cid),
@@ -630,6 +636,33 @@ def _build_parser() -> argparse.ArgumentParser:
     )
     _add_common(p_backfill)
 
+    p_skips = p_ledger_sub.add_parser(
+        "skips",
+        help="Query or resolve durable skipped-ledger records",
+    )
+    p_skips_sub = p_skips.add_subparsers(dest="ledger_skips_cmd", required=True)
+    p_skips_list = p_skips_sub.add_parser(
+        "list",
+        help="List skipped-ledger records by instance/date/portfolio/state",
+    )
+    p_skips_list.add_argument("--date", default=None, help="UTC date filter (YYYY-MM-DD)")
+    p_skips_list.add_argument("--portfolio-id", dest="portfolio_id", default=None)
+    p_skips_list.add_argument(
+        "--status",
+        choices=("unresolved", "resolved", "all"),
+        default="unresolved",
+    )
+    p_skips_list.add_argument("--limit", type=int, default=100)
+    _add_common(p_skips_list)
+
+    p_skips_resolve = p_skips_sub.add_parser(
+        "resolve",
+        help="Mark one skipped-ledger record resolved and emit a resolved alert",
+    )
+    p_skips_resolve.add_argument("--skip-key", required=True)
+    p_skips_resolve.add_argument("--reason", default="operator_reconciled")
+    _add_common(p_skips_resolve)
+
     return parser
 
 
@@ -696,14 +729,13 @@ def _cmd_sync(args, cfg, confirm) -> int:
     # Determine target slot_ids
     import psycopg
     if args.portfolio_id:
-        with psycopg.connect(cfg.dsn) as conn:
-            with conn.cursor() as cur:
-                cur.execute(
-                    "SELECT slot_id FROM daily_slot "
-                    "WHERE instance_key = %s AND portfolio_id = %s LIMIT 1",
-                    (instance, args.portfolio_id),
-                )
-                row = cur.fetchone()
+        with psycopg.connect(cfg.dsn) as conn, conn.cursor() as cur:
+            cur.execute(
+                "SELECT slot_id FROM daily_slot "
+                "WHERE instance_key = %s AND portfolio_id = %s LIMIT 1",
+                (instance, args.portfolio_id),
+            )
+            row = cur.fetchone()
         if row is None:
             _fail(f"no slot found for portfolio_id={args.portfolio_id} instance={instance}", args.json)
             return 1
@@ -711,14 +743,13 @@ def _cmd_sync(args, cfg, confirm) -> int:
     else:
         # --all: target all warned slots (B case) plus all inactive slots with broker holding
         warned = warn_store.list_warned()
-        with psycopg.connect(cfg.dsn) as conn:
-            with conn.cursor() as cur:
-                cur.execute(
-                    "SELECT slot_id FROM daily_slot "
-                    "WHERE instance_key = %s AND status IN ('inactive', 'holding')",
-                    (instance,),
-                )
-                all_ids = {str(r[0]) for r in cur.fetchall()}
+        with psycopg.connect(cfg.dsn) as conn, conn.cursor() as cur:
+            cur.execute(
+                "SELECT slot_id FROM daily_slot "
+                "WHERE instance_key = %s AND status IN ('inactive', 'holding')",
+                (instance,),
+            )
+            all_ids = {str(r[0]) for r in cur.fetchall()}
         slot_ids = sorted(all_ids | warned)
 
     results = []
@@ -820,7 +851,7 @@ def _cmd_verify(args, cfg) -> int:
             items = hub.get_portfolios(daemon_id)
             store.swap(items)
         except Exception:  # noqa: BLE001 — degrade: verify with empty store
-            pass
+            _log.info("verify snapshot preload skipped", exc_info=True)
 
     if args.portfolio_id:
         target_pids = [args.portfolio_id]
@@ -903,30 +934,29 @@ def _cmd_report(args, cfg) -> int:
     holdings: list[Holding] = []
     pending: list[PendingOrder] = []
     trades: list[TradeRow] = []
-    with psycopg.connect(cfg.dsn) as conn:
-        with conn.cursor() as cur:
-            cur.execute(
-                "SELECT symbol, portfolio_id, qty, avg_price FROM daily_slot "
-                "WHERE status = 'holding' ORDER BY symbol"
-            )
-            for symbol, pid, qty, avg in cur.fetchall():
-                holdings.append(Holding(symbol or "", pid or "",
-                                        float(qty or 0), float(avg or 0)))
-            cur.execute(
-                "SELECT symbol, portfolio_id, status FROM daily_slot "
-                "WHERE status IN ('pending_buy','pending_sell') ORDER BY symbol"
-            )
-            for symbol, pid, status in cur.fetchall():
-                pending.append(PendingOrder(symbol or "", pid or "", status))
-            cur.execute(
-                "SELECT asset_id, side, qty, price, fee, tax FROM ledger_outbox "
-                "WHERE COALESCE(executed_at, created_at) >= %s "
-                "AND COALESCE(executed_at, created_at) < %s",
-                (start_utc, end_utc),
-            )
-            for asset_id, side, qty, price, fee, tax in cur.fetchall():
-                trades.append(TradeRow(asset_id, side, float(qty), float(price),
-                                       float(fee or 0), float(tax or 0)))
+    with psycopg.connect(cfg.dsn) as conn, conn.cursor() as cur:
+        cur.execute(
+            "SELECT symbol, portfolio_id, qty, avg_price FROM daily_slot "
+            "WHERE status = 'holding' ORDER BY symbol"
+        )
+        for symbol, pid, qty, avg in cur.fetchall():
+            holdings.append(Holding(symbol or "", pid or "",
+                                    float(qty or 0), float(avg or 0)))
+        cur.execute(
+            "SELECT symbol, portfolio_id, status FROM daily_slot "
+            "WHERE status IN ('pending_buy','pending_sell') ORDER BY symbol"
+        )
+        for symbol, pid, status in cur.fetchall():
+            pending.append(PendingOrder(symbol or "", pid or "", status))
+        cur.execute(
+            "SELECT asset_id, side, qty, price, fee, tax FROM ledger_outbox "
+            "WHERE COALESCE(executed_at, created_at) >= %s "
+            "AND COALESCE(executed_at, created_at) < %s",
+            (start_utc, end_utc),
+        )
+        for asset_id, side, qty, price, fee, tax in cur.fetchall():
+            trades.append(TradeRow(asset_id, side, float(qty), float(price),
+                                   float(fee or 0), float(tax or 0)))
 
     rep = build_report(holdings, pending, trades)
 
@@ -989,9 +1019,69 @@ def _cmd_report(args, cfg) -> int:
     return 0
 
 
+def _render_ledger_skip_record(record) -> dict[str, object]:
+    return ledger_skip_alarm_fields(record)
+
+
+def _cmd_ledger_skips_list(args, cfg) -> int:
+    from trex_core.adapters.persistence.ledger_skip_store import PostgreSQLLedgerSkipStore
+
+    instance = _resolve_instance(args)
+    on_date = _parse_iso_date(args.date) if args.date else None
+    records = PostgreSQLLedgerSkipStore(cfg.dsn).list_records(
+        instance_key=instance,
+        portfolio_id=args.portfolio_id,
+        on_date=on_date,
+        status=args.status,
+        limit=args.limit,
+    )
+    payload = {
+        "instance": instance,
+        "status": args.status,
+        "count": len(records),
+        "records": [_render_ledger_skip_record(record) for record in records],
+    }
+    if args.json:
+        _emit(payload, as_json=True)
+    else:
+        print(
+            f"instance={instance} status={args.status} count={len(records)} "
+            f"portfolio_id={args.portfolio_id or '*'} date={args.date or '*'}"
+        )
+        for record in records:
+            print(
+                f"  {record.skip_key}\t{record.status}\t{record.skip_reason}\t"
+                f"slot={record.slot_id}\tportfolio={record.portfolio_id or '-'}\t"
+                f"execution={record.execution_id or '-'}"
+            )
+    return 0
+
+
+def _cmd_ledger_skips_resolve(args, cfg) -> int:
+    from trex_core.adapters.persistence.ledger_skip_store import PostgreSQLLedgerSkipStore
+
+    store = PostgreSQLLedgerSkipStore(cfg.dsn)
+    record = store.resolve_by_key(args.skip_key, resolution_reason=args.reason)
+    if record is None:
+        _fail(f"unresolved ledger skip not found: {args.skip_key}", args.json)
+        return 1
+    emitter, transport = _default_alarm_emitter(cfg)
+    try:
+        publish_ledger_skip_alarm(emitter, record, resolved=True)
+    finally:
+        transport.close()
+    payload = {"result": "resolved", "record": _render_ledger_skip_record(record)}
+    if args.json:
+        _emit(payload, as_json=True)
+    else:
+        print(f"resolved {record.skip_key} reason={record.resolution_reason}")
+    return 0
+
+
 def _cmd_ledger_backfill(args, cfg, confirm) -> int:
     from trex_core.adapters.broker.kis import KISEvidenceLookupError
     from trex_core.adapters.ledger.portfolio_hub import PortfolioHubLedger
+    from trex_core.adapters.persistence.ledger_skip_store import PostgreSQLLedgerSkipStore
     from trex_core.adapters.persistence.postgresql import PostgreSQLLedgerOutbox
     from trex_core.runtime.broker_provider import BrokerCredentialUnavailable
 
@@ -1020,6 +1110,7 @@ def _cmd_ledger_backfill(args, cfg, confirm) -> int:
         "trade": None,
         "outbox_precheck": None,
         "result": None,
+        "resolved_skips": [],
         "audit_path": None,
     }
 
@@ -1084,7 +1175,7 @@ def _cmd_ledger_backfill(args, cfg, confirm) -> int:
             "execution_id": trade.execution_id,
             "existing": existing,
         }
-    except (BrokerCredentialUnavailable, KISEvidenceLookupError, RuntimeError) as exc:
+    except (BrokerCredentialUnavailable, KISEvidenceLookupError, RuntimeError, TypeError) as exc:
         report["result"] = {"status": "rejected", "reason": str(exc)}
         report["audit_path"] = _write_ledger_backfill_audit(args, report)
         if args.json:
@@ -1172,6 +1263,23 @@ def _cmd_ledger_backfill(args, cfg, confirm) -> int:
         "last_outcome": item.last_outcome.value if item and item.last_outcome else None,
         "last_error": item.last_error if item is not None else None,
     }
+    if exit_code == 0:
+        resolved = PostgreSQLLedgerSkipStore(cfg.dsn).resolve_for_fill(
+            instance_key=instance,
+            slot_id=uuid.UUID(str(slot_binding["slot_id"])),
+            side=trade.side,
+            order_id=trade.order_id,
+            execution_id=trade.execution_id,
+            resolution_reason="manual_ledger_backfill",
+        )
+        report["resolved_skips"] = [_render_ledger_skip_record(record) for record in resolved]
+        if resolved:
+            emitter, transport = _default_alarm_emitter(cfg)
+            try:
+                for record in resolved:
+                    publish_ledger_skip_alarm(emitter, record, resolved=True)
+            finally:
+                transport.close()
     report["audit_path"] = _write_ledger_backfill_audit(args, report)
     if args.json:
         _emit(report, as_json=True)
@@ -1219,6 +1327,18 @@ def main(argv: list[str] | None = None, *, rpc=None, confirm=None) -> int:
 
     if args.cmd == "ledger" and getattr(args, "ledger_cmd", None) == "backfill":
         return _cmd_ledger_backfill(args, cfg, confirm)
+    if (
+        args.cmd == "ledger"
+        and getattr(args, "ledger_cmd", None) == "skips"
+        and getattr(args, "ledger_skips_cmd", None) == "list"
+    ):
+        return _cmd_ledger_skips_list(args, cfg)
+    if (
+        args.cmd == "ledger"
+        and getattr(args, "ledger_cmd", None) == "skips"
+        and getattr(args, "ledger_skips_cmd", None) == "resolve"
+    ):
+        return _cmd_ledger_skips_resolve(args, cfg)
 
     # direct-DB control commands (pause/resume stay direct-DB, OQ-3).
     if args.cmd in ("pause", "resume", "paused"):

trex_core/runtime/daemon.py

@@ -35,6 +35,7 @@ from trex_core.adapters.persistence.migrate import apply_migrations
 from trex_core.adapters.persistence.account_snapshot import (
     PostgreSQLAccountSnapshotStore,
 )
+from trex_core.adapters.persistence.ledger_skip_store import PostgreSQLLedgerSkipStore
 from trex_core.adapters.persistence.postgresql import (
     PgStrategyStateStore,
     PostgreSQLIntentStore,
@@ -188,6 +189,7 @@ class Daemon:
         )
 
         self.outbox = PostgreSQLLedgerOutbox(config.dsn)
+        self.ledger_skip_store = PostgreSQLLedgerSkipStore(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()
@@ -303,6 +305,7 @@ 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
+            ledger_skip_store=self.ledger_skip_store,
             balance_reader=(
                 self.account_snapshot_service.holdings_for_sell
                 if self.account_snapshot_service is not None