diff --git a/tests/application/test_ledger_outbox.py b/tests/application/test_ledger_outbox.py index a244655..5867064 100644 --- a/tests/application/test_ledger_outbox.py +++ b/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 diff --git a/tests/runtime/test_cli_ledger_backfill.py b/tests/runtime/test_cli_ledger_backfill.py new file mode 100644 index 0000000..d13eee6 --- /dev/null +++ b/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 diff --git a/trex_core/adapters/broker/kis.py b/trex_core/adapters/broker/kis.py index 69cfd88..15d052f 100644 --- a/trex_core/adapters/broker/kis.py +++ b/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, diff --git a/trex_core/application/ledger_outbox.py b/trex_core/application/ledger_outbox.py index 539e28f..3e65490 100644 --- a/trex_core/application/ledger_outbox.py +++ b/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) diff --git a/trex_core/runtime/cli.py b/trex_core/runtime/cli.py index 344f57f..a41e6e9 100644 --- a/trex_core/runtime/cli.py +++ b/trex_core/runtime/cli.py @@ -11,6 +11,7 @@ Subcommands: cancel — 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)