feat: add ledger backfill command

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

tests/application/test_ledger_outbox.py

@@ -68,3 +68,18 @@ def test_writer_retries_until_success():
     now["t"] += timedelta(seconds=600)       # past the backoff window
     s2 = writer.run_once()
     assert s2.completed == 1 and ledger.count() == 1
+
+
+def test_process_correlation_id_only_drains_requested_row():
+    outbox = InMemoryLedgerOutbox()
+    ledger = InMemoryLedger()
+    writer = LedgerWriter(outbox, ledger)
+    first = outbox.enqueue(_trade(execution_id="e-1"))
+    second = outbox.enqueue(_trade(execution_id="e-2"))
+
+    outcome = writer.process_correlation_id(first.correlation_id)
+
+    assert outcome is OutboxOutcome.COMPLETED
+    assert ledger.count() == 1
+    assert outbox.get(first.correlation_id).status is WorkItemStatus.COMPLETED
+    assert outbox.get(second.correlation_id).status is WorkItemStatus.PENDING

tests/runtime/test_cli_ledger_backfill.py

@@ -0,0 +1,212 @@
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+
+from trex_core.application.ledger_outbox import OutboxOutcome
+from trex_core.application.ledger_port import LedgerTrade
+from trex_core.runtime import cli
+
+
+def _cfg(*, live: bool = False):
+    return SimpleNamespace(
+        dsn="postgresql://x/y",
+        broker="kis",
+        kis=SimpleNamespace(allow_live=live),
+        portfolio_hub=SimpleNamespace(base_url="http://pf.local", token="tok", timeout=5.0),
+    )
+
+
+def _args(tmp_path, **overrides):
+    data = {
+        "date": "2026-08-24",
+        "execution_id": "kis-daily-ccld:20260824:12345678:0001234567:10",
+        "dry_run": False,
+        "yes": True,
+        "live": True,
+        "json": True,
+        "audit_dir": str(tmp_path),
+        "instance": "inst-1",
+        "config": None,
+        "config_path": None,
+        "env_file": None,
+        "timeout": 10.0,
+        "ledger_cmd": "backfill",
+        "cmd": "ledger",
+    }
+    data.update(overrides)
+    return SimpleNamespace(**data)
+
+
+def _evidence():
+    return {
+        "symbol": "005930",
+        "side": "sell",
+        "qty": 10.0,
+        "price": 70000.0,
+        "fee": 105.0,
+        "tax": 1400.0,
+        "order_id": "12345678:0001234567",
+        "order_venue": "SOR",
+        "execution_venue": None,
+        "executed_at": "2026-08-24T00:01:02+00:00",
+        "raw": {
+            "pdno": "005930",
+            "sll_buy_dvsn_cd": "01",
+            "tot_ccld_qty": "10",
+            "avg_prvs": "70000",
+            "ord_dt": "20260824",
+            "ord_gno_brno": "12345678",
+            "odno": "0001234567",
+            "excg_id_dvsn_cd": "SOR",
+        },
+    }
+
+
+def test_ledger_backfill_parser_accepts_required_flags():
+    parser = cli._build_parser()
+    args = parser.parse_args([
+        "ledger", "backfill",
+        "--date", "2026-08-24",
+        "--execution-id", "kis-daily-ccld:20260824:12345678:0001234567:10",
+        "--dry-run",
+    ])
+    assert args.cmd == "ledger"
+    assert args.ledger_cmd == "backfill"
+    assert args.dry_run is True
+
+
+def test_ledger_backfill_dry_run_emits_trade_and_writes_audit(monkeypatch, tmp_path, capsys):
+    monkeypatch.setattr(cli, "_lookup_ledger_backfill_evidence",
+                        lambda cfg, **kwargs: (_evidence(), SimpleNamespace(mode="env", identity_label="kis-config")))
+    monkeypatch.setattr(cli, "_resolve_ledger_backfill_slot",
+                        lambda dsn, **kwargs: {"slot_id": "slot-1", "portfolio_id": "PF-1", "status": "inactive", "symbol": "005930"})
+    monkeypatch.setattr(cli, "_lookup_outbox_by_execution_id", lambda dsn, **kwargs: [])
+
+    rc = cli._cmd_ledger_backfill(_args(tmp_path, dry_run=True), _cfg(), lambda _: True)
+
+    assert rc == 0
+    out = json.loads(capsys.readouterr().out)
+    assert out["result"]["status"] == "dry_run"
+    assert out["trade"]["correlation_id"] == "PF-1:12345678:0001234567:kis-daily-ccld:20260824:12345678:0001234567:10"
+    assert out["trade"]["order_venue"] == "SOR"
+    assert out["trade"]["execution_venue"] is None
+    audit = json.loads(next(tmp_path.glob("*.json")).read_text(encoding="utf-8"))
+    assert audit["request"]["execution_id"] == _args(tmp_path).execution_id
+
+
+def test_ledger_backfill_live_requires_live_flag(monkeypatch, tmp_path, capsys):
+    monkeypatch.setattr(cli, "_lookup_ledger_backfill_evidence",
+                        lambda cfg, **kwargs: (_evidence(), SimpleNamespace(mode="env", identity_label="kis-config")))
+    monkeypatch.setattr(cli, "_resolve_ledger_backfill_slot",
+                        lambda dsn, **kwargs: {"slot_id": "slot-1", "portfolio_id": "PF-1", "status": "inactive", "symbol": "005930"})
+    monkeypatch.setattr(cli, "_lookup_outbox_by_execution_id", lambda dsn, **kwargs: [])
+
+    rc = cli._cmd_ledger_backfill(_args(tmp_path, live=False), _cfg(live=True), lambda _: True)
+
+    assert rc == 2
+    err = capsys.readouterr().err
+    assert "--live" in err
+    audit = json.loads(next(tmp_path.glob("*.json")).read_text(encoding="utf-8"))
+    assert audit["result"]["status"] == "rejected"
+
+
+def test_ledger_backfill_existing_completed_row_is_idempotent(monkeypatch, tmp_path, capsys):
+    monkeypatch.setattr(cli, "_lookup_ledger_backfill_evidence",
+                        lambda cfg, **kwargs: (_evidence(), SimpleNamespace(mode="env", identity_label="kis-config")))
+    monkeypatch.setattr(cli, "_resolve_ledger_backfill_slot",
+                        lambda dsn, **kwargs: {"slot_id": "slot-1", "portfolio_id": "PF-1", "status": "inactive", "symbol": "005930"})
+    existing = [{
+        "correlation_id": "PF-1:12345678:0001234567:kis-daily-ccld:20260824:12345678:0001234567:10",
+        "asset_id": "PF-1",
+        "order_id": "12345678:0001234567",
+        "execution_id": "kis-daily-ccld:20260824:12345678:0001234567:10",
+        "status": "completed",
+        "attempts": 1,
+        "last_outcome": "completed",
+        "last_error": None,
+    }]
+    monkeypatch.setattr(cli, "_lookup_outbox_by_execution_id", lambda dsn, **kwargs: existing)
+
+    class _Outbox:
+        def enqueue(self, trade):
+            raise AssertionError("completed execution must not be re-enqueued")
+
+        def get(self, correlation_id):
+            return SimpleNamespace(
+                status=SimpleNamespace(value="completed"),
+                attempts=1,
+                last_outcome=SimpleNamespace(value="completed"),
+                last_error=None,
+            )
+
+    class _Writer:
+        def __init__(self, *_args, **_kwargs):
+            pass
+
+        def process_correlation_id(self, correlation_id):
+            raise AssertionError("completed execution must not be re-processed")
+
+    import trex_core.adapters.persistence.postgresql as pg_mod
+    import trex_core.adapters.ledger.portfolio_hub as hub_mod
+
+    monkeypatch.setattr(pg_mod, "PostgreSQLLedgerOutbox", lambda dsn: _Outbox())
+    monkeypatch.setattr(hub_mod, "PortfolioHubLedger", lambda *args, **kwargs: object())
+    monkeypatch.setattr(cli, "LedgerWriter", _Writer)
+
+    rc = cli._cmd_ledger_backfill(_args(tmp_path), _cfg(), lambda _: True)
+
+    assert rc == 0
+    out = json.loads(capsys.readouterr().out)
+    assert out["result"]["status"] == "already_recorded"
+
+
+def test_ledger_backfill_live_enqueue_and_processes_exact_trade(monkeypatch, tmp_path, capsys):
+    monkeypatch.setattr(cli, "_lookup_ledger_backfill_evidence",
+                        lambda cfg, **kwargs: (_evidence(), SimpleNamespace(mode="env", identity_label="kis-config")))
+    monkeypatch.setattr(cli, "_resolve_ledger_backfill_slot",
+                        lambda dsn, **kwargs: {"slot_id": "slot-1", "portfolio_id": "PF-1", "status": "inactive", "symbol": "005930"})
+    monkeypatch.setattr(cli, "_lookup_outbox_by_execution_id", lambda dsn, **kwargs: [])
+
+    class _Outbox:
+        def __init__(self):
+            self.trade = None
+
+        def enqueue(self, trade: LedgerTrade):
+            self.trade = trade
+
+        def get(self, correlation_id):
+            assert self.trade is not None
+            return SimpleNamespace(
+                status=SimpleNamespace(value="completed"),
+                attempts=1,
+                last_outcome=SimpleNamespace(value="completed"),
+                last_error=None,
+            )
+
+    outbox = _Outbox()
+
+    class _Writer:
+        def __init__(self, seen_outbox, *_args, **_kwargs):
+            self.seen_outbox = seen_outbox
+
+        def process_correlation_id(self, correlation_id):
+            assert self.seen_outbox.trade is not None
+            assert correlation_id == self.seen_outbox.trade.correlation_id
+            return OutboxOutcome.COMPLETED
+
+    import trex_core.adapters.persistence.postgresql as pg_mod
+    import trex_core.adapters.ledger.portfolio_hub as hub_mod
+
+    monkeypatch.setattr(pg_mod, "PostgreSQLLedgerOutbox", lambda dsn: outbox)
+    monkeypatch.setattr(hub_mod, "PortfolioHubLedger", lambda *args, **kwargs: object())
+    monkeypatch.setattr(cli, "LedgerWriter", _Writer)
+
+    rc = cli._cmd_ledger_backfill(_args(tmp_path), _cfg(), lambda _: True)
+
+    assert rc == 0
+    out = json.loads(capsys.readouterr().out)
+    assert out["result"]["status"] == "completed"
+    assert outbox.trade is not None
+    assert outbox.trade.order_venue == "SOR"
+    assert outbox.trade.venue is None

trex_core/adapters/broker/kis.py

@@ -43,6 +43,10 @@ _SUBMISSION_LOOKUP_GRACE_SECONDS = 90.0
 _SUPPORTED_ORDER_VENUES = frozenset({"KRX", "NXT", "SOR"})
 
 
+class KISEvidenceLookupError(RuntimeError):
+    """Read-only daily-fill evidence lookup failed."""
+
+
 def _balance_http_error_code(status: Optional[int]) -> str:
     if status == 401:
         return "authentication_failed"
@@ -594,6 +598,116 @@ class KISBroker:
         # SOR is routing provenance, not an execution exchange.
         return None
 
+    @staticmethod
+    def _order_venue_from_row(row: dict) -> str | None:
+        """Parse original routing venue (KRX/NXT/SOR) without time inference."""
+        for key in ("excg_id_dvsn_cd", "excg_id_dvsn_Cd", "EXCG_ID_DVSN_CD", "order_venue"):
+            value = str(row.get(key) or "").strip().upper()
+            if value in _SUPPORTED_ORDER_VENUES:
+                return value
+            if value in {"SOR-KRX", "SOR-NXT"}:
+                return "SOR"
+        return None
+
+    @staticmethod
+    def _side_from_row(row: dict) -> str | None:
+        code = str(row.get("sll_buy_dvsn_cd") or row.get("SLL_BUY_DVSN_CD") or "").strip()
+        if code == "01":
+            return "sell"
+        if code == "02":
+            return "buy"
+        return None
+
+    @staticmethod
+    def _parse_daily_ccld_execution_id(execution_id: str) -> dict[str, str]:
+        parts = str(execution_id or "").split(":")
+        if len(parts) != 5 or parts[0] != "kis-daily-ccld":
+            raise KISEvidenceLookupError("unsupported execution_id format")
+        order_date, orgno, odno, qty = parts[1], parts[2], parts[3], parts[4]
+        if len(order_date) != 8 or not order_date.isdigit():
+            raise KISEvidenceLookupError("execution_id missing order date")
+        if not odno or not qty.isdigit() or int(qty) <= 0:
+            raise KISEvidenceLookupError("execution_id missing order number or qty")
+        return {
+            "order_date": order_date,
+            "orgno": orgno,
+            "odno": odno,
+            "qty": qty,
+        }
+
+    @staticmethod
+    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]
+        if len(order_time) == 6 and order_time.isdigit():
+            ordered_at = datetime.strptime(order_date + order_time, "%Y%m%d%H%M%S")
+            return ordered_at.replace(tzinfo=_KST).astimezone(timezone.utc)
+        ordered_at = datetime.strptime(order_date, "%Y%m%d")
+        return ordered_at.replace(tzinfo=_KST).astimezone(timezone.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."""
+        parsed = self._parse_daily_ccld_execution_id(execution_id)
+        if trading_date != parsed["order_date"]:
+            raise KISEvidenceLookupError("execution_id date does not match --date")
+        tr_id, params = self._daily_fill_request(parsed["order_date"], order_number=parsed["odno"])
+        try:
+            resp = httpx.get(
+                f"{self._base_url}/uapi/domestic-stock/v1/trading/inquire-daily-ccld",
+                headers=self._headers(tr_id),
+                params=params,
+                timeout=self._timeout,
+            )
+            payload = resp.json()
+        except (httpx.HTTPError, ValueError) as exc:
+            raise KISEvidenceLookupError(
+                f"daily fill evidence lookup failed: {exc.__class__.__name__}"
+            ) from exc
+        if getattr(resp, "status_code", 200) >= 400 or str(payload.get("rt_cd")) != "0":
+            raise KISEvidenceLookupError("daily fill evidence lookup rejected")
+        matches: list[dict] = []
+        normalized_odno = parsed["odno"].lstrip("0") or "0"
+        for row in payload.get("output1") or []:
+            row_odno = str(row.get("odno") or "").strip().lstrip("0") or "0"
+            if row_odno != normalized_odno:
+                continue
+            side = self._side_from_row(row)
+            if side is None:
+                continue
+            status = self._fill_status_from_row(
+                row, side=side, today=parsed["order_date"], fallback_odno=parsed["odno"]
+            )
+            if status.execution_id != execution_id:
+                continue
+            order_venue = self._order_venue_from_row(row)
+            if order_venue is None:
+                continue
+            orgno = str(row.get("ord_gno_brno") or "").strip()
+            odno = str(row.get("odno") or parsed["odno"]).strip()
+            symbol = str(row.get("pdno") or "").strip()
+            if not symbol:
+                continue
+            matches.append(
+                {
+                    "symbol": symbol.zfill(6),
+                    "side": side,
+                    "qty": float(status.filled_qty),
+                    "price": float(status.avg_price or 0.0),
+                    "fee": float(status.fee),
+                    "tax": float(status.tax),
+                    "order_date": parsed["order_date"],
+                    "execution_id": execution_id,
+                    "order_id": f"{orgno}:{odno}" if orgno else odno,
+                    "order_venue": order_venue,
+                    "execution_venue": status.execution_venue,
+                    "executed_at": self._executed_at_from_row(row, parsed["order_date"]),
+                    "raw": dict(row),
+                }
+            )
+        if len(matches) != 1:
+            raise KISEvidenceLookupError("daily fill evidence not found or ambiguous")
+        return matches[0]
+
     @staticmethod
     def _fill_status_from_row(
         row: dict,

trex_core/application/ledger_outbox.py

@@ -205,6 +205,13 @@ class LedgerWriter:
                 stats.delivery_failed += 1
         return stats
 
+    def process_correlation_id(self, correlation_id: str) -> OutboxOutcome | None:
+        """Process one specific outbox row immediately."""
+        item = self._outbox.get(correlation_id)
+        if item is None:
+            return None
+        return self._process(item)
+
     def _backoff_delay(self, attempts: int) -> float:
         """Exponential backoff with jitter: min(base*2^(n-1), cap) + up to 25%."""
         base_delay = min(self._retry_base * (2 ** (attempts - 1)), self._retry_cap)

trex_core/runtime/cli.py

@@ -11,6 +11,7 @@ Subcommands:
   cancel <slot_id>        — cancel a pending order (control RPC; safeguarded)
   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
 
 universe/slots/enter/exit go through the daemon's RabbitMQ RPC surface so the
 single-writer admission + invariants are preserved (no direct slot writes,
@@ -26,18 +27,25 @@ sync safeguards (REQ-105): `--yes` required (skip confirmation prompt);
 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.
 """
 from __future__ import annotations
 
 import argparse
 import json as _json
+import re
 import sys
 import uuid
+from pathlib import Path
 from datetime import datetime, timedelta, timezone
 from zoneinfo import ZoneInfo
 
 from trex_core.adapters.messaging import topology
 from trex_core.adapters.portfolio.pause_store import PostgreSQLPauseStore
+from trex_core.application.ledger_outbox import LedgerWriter, OutboxOutcome
+from trex_core.application.ledger_port import LedgerTrade
 from trex_core.runtime.config import (
     DaemonConfig,
     DaemonInstanceConfig,
@@ -46,6 +54,7 @@ from trex_core.runtime.config import (
 )
 
 _DEFAULT_TIMEOUT = 10.0
+_DEFAULT_LEDGER_BACKFILL_AUDIT_DIR = "artifacts/validation/ledger_backfill"
 
 
 # --------------------------------------------------------------------------- #
@@ -101,6 +110,10 @@ def _emit(data: dict, *, as_json: bool) -> None:
         print(_json.dumps(data, ensure_ascii=False, default=str))
 
 
+def _emit_pretty_json(data: dict) -> None:
+    print(_json.dumps(data, ensure_ascii=False, indent=2, default=str))
+
+
 # --------------------------------------------------------------------------- #
 # RPC subcommand handlers
 # --------------------------------------------------------------------------- #
@@ -352,6 +365,135 @@ class _CachedBalanceBroker:
         return getattr(self._broker, name)
 
 
+def _parse_iso_date(raw: str) -> str:
+    return datetime.fromisoformat(f"{raw}T00:00:00+00:00").date().isoformat()
+
+
+def _sanitize_path_component(value: str, *, limit: int = 80) -> str:
+    cleaned = re.sub(r"[^A-Za-z0-9._-]+", "_", str(value or "").strip()).strip("._")
+    return (cleaned or "value")[:limit]
+
+
+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")
+    name = (
+        f"{stamp}--{_sanitize_path_component(_resolve_instance(args))}"
+        f"--{_sanitize_path_component(execution_id)}.json"
+    )
+    return base.resolve() / name
+
+
+def _write_ledger_backfill_audit(args, report: dict) -> str:
+    path = _ledger_backfill_audit_path(args, report["request"]["execution_id"])
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(_json.dumps(report, ensure_ascii=False, indent=2, default=str) + "\n",
+                    encoding="utf-8")
+    return str(path)
+
+
+def _lookup_ledger_backfill_evidence(cfg: DaemonConfig, *, trading_date: str, execution_id: str):
+    from trex_core.runtime.broker_provider import (
+        BrokerCredentialUnavailable, 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
+    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()
+    if not rows:
+        raise RuntimeError(f"no daemon-owned slot found for instance={instance} symbol={symbol}")
+    if len(rows) != 1:
+        raise RuntimeError(f"ambiguous daemon-owned slot mapping for instance={instance} symbol={symbol}")
+    return {
+        "slot_id": str(rows[0][0]),
+        "portfolio_id": str(rows[0][1]),
+        "status": str(rows[0][2]),
+        "symbol": symbol,
+    }
+
+
+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()
+    return [
+        {
+            "correlation_id": str(cid),
+            "asset_id": str(asset_id),
+            "order_id": str(order_id),
+            "execution_id": str(exec_id),
+            "status": str(status),
+            "attempts": int(attempts or 0),
+            "last_outcome": str(last_outcome) if last_outcome else None,
+            "last_error": str(last_error) if last_error else None,
+        }
+        for cid, asset_id, order_id, exec_id, status, attempts, last_outcome, last_error in rows
+    ]
+
+
+def _normalize_existing_execution(existing_rows: list[dict], trade: LedgerTrade) -> dict | None:
+    if not existing_rows:
+        return None
+    if len(existing_rows) != 1:
+        raise RuntimeError(f"execution_id={trade.execution_id} already maps to multiple outbox rows")
+    existing = existing_rows[0]
+    if existing["correlation_id"] != trade.correlation_id:
+        raise RuntimeError(
+            f"execution_id={trade.execution_id} already belongs to {existing['correlation_id']}"
+        )
+    return existing
+
+
+def _ledger_outcome_result(outcome: OutboxOutcome | None, existing_status: str | None = None) -> tuple[str, int]:
+    if outcome is OutboxOutcome.COMPLETED:
+        return "completed", 0
+    if outcome is OutboxOutcome.DUPLICATE_COMPLETED:
+        return "duplicate_completed", 0
+    if outcome is OutboxOutcome.RETRYABLE_FAILURE:
+        return "retryable_failure", 1
+    if outcome is OutboxOutcome.CONFLICT_BLOCKED:
+        return "conflict_blocked", 1
+    if outcome in {OutboxOutcome.PERMANENT_FAILURE, OutboxOutcome.DEAD_LETTERED}:
+        return "delivery_failed", 1
+    if existing_status == "completed":
+        return "already_recorded", 0
+    if existing_status:
+        return f"already_present:{existing_status}", 0
+    return "no_action", 1
+
+
 # --------------------------------------------------------------------------- #
 # arg parsing
 # --------------------------------------------------------------------------- #
@@ -464,6 +606,29 @@ def _build_parser() -> argparse.ArgumentParser:
     )
     _add_common(p_verify)
 
+    p_ledger = sub.add_parser(
+        "ledger",
+        help="Ledger operator tools (direct-DB enqueue + immediate writer path)",
+    )
+    p_ledger_sub = p_ledger.add_subparsers(dest="ledger_cmd", required=True)
+    p_backfill = p_ledger_sub.add_parser(
+        "backfill",
+        help="Evidence-backed historical ledger backfill (KIS daily-ccld)",
+    )
+    p_backfill.add_argument("--date", required=True, help="KST trading date (YYYY-MM-DD)")
+    p_backfill.add_argument("--execution-id", dest="execution_id", required=True)
+    p_backfill.add_argument("--yes", action="store_true", help="skip confirmation")
+    p_backfill.add_argument("--live", action="store_true",
+                            help="confirm a LIVE ledger mutation (with --yes)")
+    p_backfill.add_argument("--dry-run", dest="dry_run", action="store_true",
+                            help="preview the exact ledger trade, do not write")
+    p_backfill.add_argument(
+        "--audit-dir",
+        default=_DEFAULT_LEDGER_BACKFILL_AUDIT_DIR,
+        help="directory for persistent JSON audit reports",
+    )
+    _add_common(p_backfill)
+
     return parser
 
 
@@ -821,6 +986,197 @@ def _cmd_report(args, cfg) -> int:
     return 0
 
 
+def _cmd_ledger_backfill(args, cfg, confirm) -> int:
+    from trex_core.adapters.ledger.portfolio_hub import PortfolioHubLedger
+    from trex_core.adapters.persistence.postgresql import PostgreSQLLedgerOutbox
+    from trex_core.adapters.broker.kis import KISEvidenceLookupError
+    from trex_core.runtime.broker_provider import BrokerCredentialUnavailable
+
+    instance = _resolve_instance(args)
+    broker_name, is_live = _broker_mode(cfg)
+    if broker_name != "kis":
+        _fail("ledger backfill is only supported with the KIS broker", args.json)
+        return 1
+    try:
+        trading_date = _parse_iso_date(args.date)
+    except ValueError:
+        _fail(f"invalid --date (expected YYYY-MM-DD): {args.date!r}", args.json)
+        return 1
+
+    report = {
+        "request": {
+            "instance": instance,
+            "date": trading_date,
+            "execution_id": str(args.execution_id),
+            "dry_run": bool(args.dry_run),
+            "broker": broker_name,
+            "live": bool(is_live),
+        },
+        "evidence": None,
+        "slot_binding": None,
+        "trade": None,
+        "outbox_precheck": None,
+        "result": None,
+        "audit_path": None,
+    }
+
+    try:
+        evidence, broker_source = _lookup_ledger_backfill_evidence(
+            cfg, trading_date=trading_date.replace("-", ""), execution_id=str(args.execution_id)
+        )
+        report["broker_source"] = {
+            "mode": broker_source.mode,
+            "identity_label": broker_source.identity_label,
+        }
+        report["evidence"] = {
+            "symbol": evidence["symbol"],
+            "side": evidence["side"],
+            "qty": evidence["qty"],
+            "price": evidence["price"],
+            "fee": evidence["fee"],
+            "tax": evidence["tax"],
+            "order_date": trading_date,
+            "order_id": evidence["order_id"],
+            "order_venue": evidence["order_venue"],
+            "execution_venue": evidence["execution_venue"],
+            "executed_at": evidence["executed_at"],
+            "raw": evidence["raw"],
+        }
+        slot_binding = _resolve_ledger_backfill_slot(
+            cfg.dsn, instance=instance, symbol=str(evidence["symbol"])
+        )
+        report["slot_binding"] = slot_binding
+        trade = LedgerTrade(
+            asset_id=str(slot_binding["portfolio_id"]),
+            order_id=str(evidence["order_id"]),
+            execution_id=str(args.execution_id),
+            side=str(evidence["side"]),
+            qty=float(evidence["qty"]),
+            price=float(evidence["price"]),
+            fee=float(evidence["fee"]),
+            tax=float(evidence["tax"]),
+            executed_at=evidence["executed_at"],
+            venue=evidence["execution_venue"],
+            order_venue=str(evidence["order_venue"]),
+        )
+        report["trade"] = {
+            "correlation_id": trade.correlation_id,
+            "asset_id": trade.asset_id,
+            "order_id": trade.order_id,
+            "execution_id": trade.execution_id,
+            "side": trade.side,
+            "qty": trade.qty,
+            "price": trade.price,
+            "fee": trade.fee,
+            "tax": trade.tax,
+            "executed_at": trade.executed_at,
+            "order_venue": trade.order_venue,
+            "execution_venue": trade.venue,
+        }
+        existing = _normalize_existing_execution(
+            _lookup_outbox_by_execution_id(cfg.dsn, execution_id=trade.execution_id),
+            trade,
+        )
+        report["outbox_precheck"] = {
+            "execution_id": trade.execution_id,
+            "existing": existing,
+        }
+    except (BrokerCredentialUnavailable, KISEvidenceLookupError, RuntimeError) as exc:
+        report["result"] = {"status": "rejected", "reason": str(exc)}
+        report["audit_path"] = _write_ledger_backfill_audit(args, report)
+        if args.json:
+            _emit(report, as_json=True)
+        else:
+            _emit_pretty_json(report)
+        return 1
+
+    if args.dry_run:
+        report["result"] = {
+            "status": "dry_run",
+            "reason": "no writes performed",
+        }
+        report["audit_path"] = _write_ledger_backfill_audit(args, report)
+        if args.json:
+            _emit(report, as_json=True)
+        else:
+            _emit_pretty_json(report)
+        return 0
+
+    if is_live and not getattr(args, "live", False):
+        report["result"] = {"status": "rejected", "reason": "live account requires --live (with --yes)"}
+        report["audit_path"] = _write_ledger_backfill_audit(args, report)
+        _fail("live account requires --live (with --yes)", args.json)
+        if args.json:
+            _emit(report, as_json=True)
+        else:
+            _emit_pretty_json(report)
+        return 2
+    if not getattr(args, "yes", False):
+        if args.json:
+            report["result"] = {"status": "rejected", "reason": "--json ledger backfill requires --yes"}
+            report["audit_path"] = _write_ledger_backfill_audit(args, report)
+            _fail("--json ledger backfill requires --yes", True)
+            _emit(report, as_json=True)
+            return 2
+        prompt = (
+            f"Backfill ledger trade on instance={instance} broker={broker_name}"
+            f"{' LIVE' if is_live else ''}: {report['trade']}?"
+        )
+        if not confirm(prompt):
+            report["result"] = {"status": "aborted", "reason": "operator declined confirmation"}
+            report["audit_path"] = _write_ledger_backfill_audit(args, report)
+            print("aborted")
+            _emit_pretty_json(report)
+            return 1
+
+    outbox = PostgreSQLLedgerOutbox(cfg.dsn)
+    if not cfg.portfolio_hub or not cfg.portfolio_hub.base_url:
+        report["result"] = {"status": "rejected", "reason": "portfolio-hub ledger adapter is not configured"}
+        report["audit_path"] = _write_ledger_backfill_audit(args, report)
+        if args.json:
+            _emit(report, as_json=True)
+        else:
+            _emit_pretty_json(report)
+        return 1
+    writer = LedgerWriter(
+        outbox,
+        PortfolioHubLedger(
+            cfg.portfolio_hub.base_url,
+            token=cfg.portfolio_hub.token or None,
+            timeout=cfg.portfolio_hub.timeout,
+        ),
+    )
+
+    existing = (report.get("outbox_precheck") or {}).get("existing")
+    if existing is None:
+        outbox.enqueue(trade)
+        outcome = writer.process_correlation_id(trade.correlation_id)
+    elif existing.get("status") == "pending":
+        outcome = writer.process_correlation_id(trade.correlation_id)
+    else:
+        outcome = None
+    item = outbox.get(trade.correlation_id)
+    result_status, exit_code = _ledger_outcome_result(
+        outcome,
+        existing_status=str(existing["status"]) if isinstance(existing, dict) else None,
+    )
+    report["result"] = {
+        "status": result_status,
+        "correlation_id": trade.correlation_id,
+        "outcome": outcome.value if outcome is not None else None,
+        "work_item_status": item.status.value if item is not None else None,
+        "attempts": item.attempts if item is not None else None,
+        "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,
+    }
+    report["audit_path"] = _write_ledger_backfill_audit(args, report)
+    if args.json:
+        _emit(report, as_json=True)
+    else:
+        _emit_pretty_json(report)
+    return exit_code
+
+
 def main(argv: list[str] | None = None, *, rpc=None, confirm=None) -> int:
     """CLI entrypoint.
 
@@ -858,6 +1214,9 @@ def main(argv: list[str] | None = None, *, rpc=None, confirm=None) -> int:
     if args.cmd == "verify":
         return _cmd_verify(args, cfg)
 
+    if args.cmd == "ledger" and getattr(args, "ledger_cmd", None) == "backfill":
+        return _cmd_ledger_backfill(args, cfg, confirm)
+
     # direct-DB control commands (pause/resume stay direct-DB, OQ-3).
     if args.cmd in ("pause", "resume", "paused"):
         store = PostgreSQLPauseStore(cfg.dsn)