from __future__ import annotations import json from datetime import datetime from types import SimpleNamespace from uuid import UUID from trex_core.application.ledger_outbox import OutboxOutcome from trex_core.application.ledger_port import LedgerTrade from trex_core.application.ledger_skip import LedgerSkipRecord 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 _skip_record(): return LedgerSkipRecord( skip_key="ledger-skip:inst-1:slot-1:sell:12345678:0001234567:kis-daily-ccld:20260824:12345678:0001234567:10:missing_execution_id", instance_key="inst-1", slot_id=UUID(_SLOT_ID), portfolio_id="PF-1", symbol="005930", order_id="12345678:0001234567", execution_id="kis-daily-ccld:20260824:12345678:0001234567:10", side="sell", qty=10.0, price=70000.0, executed_at=datetime.fromisoformat("2026-08-24T00:01:02+00:00"), order_venue="SOR", execution_venue="unknown", skip_reason="missing_execution_id", status="resolved", resolution_reason="manual_ledger_backfill", resolved_at=datetime.fromisoformat("2026-08-24T00:03:02+00:00"), created_at=datetime.fromisoformat("2026-08-24T00:01:02+00:00"), updated_at=datetime.fromisoformat("2026-08-24T00:03:02+00:00"), ) _SLOT_ID = "00000000-0000-0000-0000-000000000001" 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_ID, "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_ID, "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_ID, "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") class _Store: def __init__(self, *_args, **_kwargs): pass def resolve_for_fill(self, **_kwargs): return [] import trex_core.adapters.ledger.portfolio_hub as hub_mod import trex_core.adapters.persistence.postgresql as pg_mod monkeypatch.setattr( pg_mod, "PostgreSQLLedgerOutbox", lambda dsn, *, instance_key: _Outbox(), ) monkeypatch.setattr(hub_mod, "PortfolioHubLedger", lambda *args, **kwargs: object()) monkeypatch.setattr(cli, "LedgerWriter", _Writer) monkeypatch.setattr( "trex_core.adapters.persistence.ledger_skip_store.PostgreSQLLedgerSkipStore", _Store, ) 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_ID, "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 class _Store: def __init__(self, *_args, **_kwargs): pass def resolve_for_fill(self, **_kwargs): return [] import trex_core.adapters.ledger.portfolio_hub as hub_mod import trex_core.adapters.persistence.postgresql as pg_mod monkeypatch.setattr( pg_mod, "PostgreSQLLedgerOutbox", lambda dsn, *, instance_key: outbox, ) monkeypatch.setattr(hub_mod, "PortfolioHubLedger", lambda *args, **kwargs: object()) monkeypatch.setattr(cli, "LedgerWriter", _Writer) monkeypatch.setattr( "trex_core.adapters.persistence.ledger_skip_store.PostgreSQLLedgerSkipStore", _Store, ) 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 def test_ledger_backfill_resolves_matching_skip(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_ID, "portfolio_id": "PF-1", "status": "inactive", "symbol": "005930"}) monkeypatch.setattr(cli, "_lookup_outbox_by_execution_id", lambda dsn, **kwargs: []) class _Outbox: def enqueue(self, trade: LedgerTrade): self.trade = trade 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): return OutboxOutcome.COMPLETED class _Store: def __init__(self, *_args, **_kwargs): pass def resolve_for_fill(self, **_kwargs): return [_skip_record()] published = [] class _Emitter: def publish_alarm(self, code, title, *, fields=None, severity="critical"): published.append((code, severity, fields)) class _Transport: def close(self): return None import trex_core.adapters.ledger.portfolio_hub as hub_mod import trex_core.adapters.persistence.postgresql as pg_mod monkeypatch.setattr( pg_mod, "PostgreSQLLedgerOutbox", lambda dsn, *, instance_key: _Outbox(), ) monkeypatch.setattr(hub_mod, "PortfolioHubLedger", lambda *args, **kwargs: object()) monkeypatch.setattr(cli, "LedgerWriter", _Writer) monkeypatch.setattr( "trex_core.adapters.persistence.ledger_skip_store.PostgreSQLLedgerSkipStore", _Store, ) monkeypatch.setattr(cli, "_default_alarm_emitter", lambda cfg: (_Emitter(), _Transport())) rc = cli._cmd_ledger_backfill(_args(tmp_path), _cfg(), lambda _: True) assert rc == 0 out = json.loads(capsys.readouterr().out) assert len(out["resolved_skips"]) == 1 assert out["resolved_skips"][0]["resolution_reason"] == "manual_ledger_backfill" assert published[0][0] == "ledger_skip_resolved"