From 76cde2edd082c56ccfe503284ae36d3a36eb4fcf Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 24 Aug 2026 13:23:06 +0000 Subject: [PATCH] feat: persist skipped ledger fills Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- .proofline/criteria/ac-0020.md | 16 + .proofline/criteria/ac-0021.md | 16 + .proofline/identities.json | 2 +- docs/runtime-flows.md | 4 +- .../persistence/test_ledger_skip_store.py | 89 ++++++ .../persistence/test_migration_and_drift.py | 1 + tests/application/test_execution_ledger.py | 177 +++++++++-- tests/conftest.py | 1 + tests/runtime/test_cli_ledger_backfill.py | 124 +++++++- tests/runtime/test_cli_ledger_skips.py | 132 +++++++++ trex_core/adapters/broker/kis.py | 54 ++-- trex_core/adapters/broker/paper.py | 16 +- .../adapters/persistence/ledger_skip_store.py | 247 ++++++++++++++++ .../sql/025_create_ledger_skip.sql | 30 ++ trex_core/application/broker_port.py | 1 + trex_core/application/execution.py | 112 ++++++- trex_core/application/ledger_skip.py | 98 +++++++ trex_core/runtime/cli.py | 274 +++++++++++++----- trex_core/runtime/daemon.py | 3 + 19 files changed, 1246 insertions(+), 151 deletions(-) create mode 100644 .proofline/criteria/ac-0020.md create mode 100644 .proofline/criteria/ac-0021.md create mode 100644 tests/adapters/persistence/test_ledger_skip_store.py create mode 100644 tests/runtime/test_cli_ledger_skips.py create mode 100644 trex_core/adapters/persistence/ledger_skip_store.py create mode 100644 trex_core/adapters/persistence/sql/025_create_ledger_skip.sql create mode 100644 trex_core/application/ledger_skip.py diff --git a/.proofline/criteria/ac-0020.md b/.proofline/criteria/ac-0020.md new file mode 100644 index 0000000..7d8a273 --- /dev/null +++ b/.proofline/criteria/ac-0020.md @@ -0,0 +1,16 @@ +--- +id: "ac-0020" +status: active +--- + +# 완료됐지만 원장 기록 불가인 fill의 durable skip 감사 기록 + +## Criterion + +core는 완료된 broker fill이 정상 `ledger_outbox` 경로로 들어갈 수 없을 때 WARN 로그만 남기고 끝내서는 안 된다. genuinely non-recordable 사례(예: `portfolio_id` 누락, `execution_id` 누락, 지원하지 않는 actual execution venue)마다 durable `ledger_skip` 감사 레코드 1건을 멱등하게 유지해야 하며, 이 레코드는 최소한 daemon instance, slot ID, portfolio ID, symbol, order ID, execution ID(있으면), side, qty, price, execution time, `order_venue`, `execution_venue` 또는 명시적 unknown 상태, machine-readable `skip_reason`, 생성/갱신/해결 timestamp, resolution state를 포함해야 한다. 이 경로는 broker·slot·portfolio 수량을 자동 보정하거나 mutate하지 않아야 하며, retryable ledger delivery failure는 계속 `ledger_outbox`와 `LedgerWriter` 재시도 의미론을 사용해야 한다. + +## Verification + +- 성공 조건: completed fill이 `missing_portfolio_id`, `missing_execution_id`, `unsupported_execution_venue` 등으로 정상 원장 기록이 불가능하면 slot transition 이후에도 `ledger_skip`에 동일 fill 식별자의 unresolved row가 남고, 같은 fill을 다시 관찰해도 duplicate row가 생기지 않는다. 운영자는 instance/date/portfolio/resolution state 조건으로 unresolved row를 조회할 수 있다. +- 실패·경계 조건: completed-but-unrecordable fill이 ephemeral log에만 남음, skip row가 retryable delivery failure row를 대체함, skip 처리 중 slot/broker/portfolio 수량이 암묵적으로 바뀜, unknown execution venue alone(SOR actual NULL)는 정상 outbox 경로 대신 skip으로 빠짐은 FAIL이다. +- 검사 방법: execution ledger unit tests, PostgreSQL `ledger_skip` store round-trip/filter/resolve tests, CLI skipped-ledger query tests, targeted Ruff/pytest와 ProofLine validate를 수행한다. diff --git a/.proofline/criteria/ac-0021.md b/.proofline/criteria/ac-0021.md new file mode 100644 index 0000000..fc2fa38 --- /dev/null +++ b/.proofline/criteria/ac-0021.md @@ -0,0 +1,16 @@ +--- +id: "ac-0021" +status: active +--- + +# skipped-ledger 알림과 해결 이벤트 + +## Criterion + +unresolved `ledger_skip` 레코드는 best-effort·non-blocking 방식으로 deduplicated `trex.alerts` 이벤트를 발행해야 하며, notifier 부재나 transport 오류가 fill lifecycle을 막아서는 안 된다. 동일 unresolved event는 중복 알림 없이 1회만 알리고, 이후 정상 ledger delivery가 성공하거나 operator가 명시적으로 reconcile/resolve 하면 해당 `ledger_skip` 레코드는 resolved 상태로 전이되고 distinct resolved alert event를 발행해야 한다. + +## Verification + +- 성공 조건: 새로운 unresolved skip 1건은 `trex.alerts`에 `ledger_skip_unresolved` alarm 1건을 만들고, manual backfill 또는 operator resolve 후에는 동일 skip에 대해 `ledger_skip_resolved` alarm이 별도로 발행된다. 알림 publish 실패는 swallow되며 durable state 전이는 유지된다. +- 실패·경계 조건: unresolved skip마다 중복 알림이 반복 발행됨, resolved 전이가 알림 없이 묵살됨, alert publish failure가 fill/backfill/resolve 경로를 abort시킴은 FAIL이다. +- 검사 방법: in-memory alarm transport tests, CLI ledger backfill resolution tests, skipped-ledger resolve command tests, PostgreSQL resolution-state verification을 실행한다. diff --git a/.proofline/identities.json b/.proofline/identities.json index 03e49b3..b98dc6e 100644 --- a/.proofline/identities.json +++ b/.proofline/identities.json @@ -1,5 +1,5 @@ { "schema_version": 1, "next_line_number": 7, - "next_ac_number": 20 + "next_ac_number": 22 } diff --git a/docs/runtime-flows.md b/docs/runtime-flows.md index c06f575..4d33d37 100644 --- a/docs/runtime-flows.md +++ b/docs/runtime-flows.md @@ -286,11 +286,11 @@ timeout 미지정·취소 실패로 잔여 고착된 pending 슬롯을 처분: 집행 루프 fill 전이 직후 `ExecutionService` 가 ledger 기록을 outbox 에 적재해 슬롯 전이 critical path 와 ledger write 를 분리한다(원장 기록 단일 주체 = 집행 경로, NFR-3). -- ledger skip 가시화(dcy-011/REQ-018, issue #11): `_record_ledger` 가 `portfolio_id` 또는 `execution_id` 누락으로 skip 하면 무성 반환하지 않고 **WARN 로그 + `ExecutionResult.ledger_skipped` 카운터**로 노출한다(슬롯 전이는 그대로 진행). portfolio_id 필수화(REQ-007)로 주원인은 제거됐고, 잔여 누락은 운영 점검 쿼리(REQ-055: closed 슬롯 ↔ `ledger_outbox` anti-join, `find_ledger_gap_slots`)로 영구 탐지한다. +- ledger skip 가시화(dcy-011/REQ-018, issue #11, issue #68): `_record_ledger` 가 completed fill을 정상 `ledger_outbox` 경로에 올릴 수 없으면 무성 반환하지 않고 **WARN 로그 + `ExecutionResult.ledger_skipped` 카운터 + durable `ledger_skip` row + best-effort `trex.alerts` alarm** 으로 남긴다(슬롯 전이는 그대로 진행). unresolved row는 `trex-core ledger skips list` 로 조회하고, manual backfill 또는 operator resolve 후 resolved 상태와 별도 resolved alarm을 남긴다. 1. fill 감지: `BrokerPort.poll_fill` 이 `state="filled"` 와 결정적 `execution_id`(paper: `paper:{broker_order_id}:{filled_qty}` / KIS: `kis-daily-ccld:{ord_dt}:{ord_gno_brno}:{odno}:{tot_ccld_qty}`), 계산된 `fee`(`calculate_fee(side, amount)`)·`tax`(`calculate_tax(side, amount)`)를 운반. fee/tax 는 단일 정책 지점(`broker_port.py`)에서 env 로 관리한다(dcy-026/REQ-018 CHG-20260612-01): `TREX_KIS_FEE_BPS`(기본 1.5 = 0.015%, 매수·매도 공통)·`TREX_KRX_TAX_SELL_BPS`(기본 20.0 = 0.2%, **매도 증권거래세**; 매수 tax=0). paper·KIS 가 동일 정책을 공유하며, KIS daily-ccld 응답은 per-fill fee/tax 필드를 제공하지 않아 정책 추정값을 사용한다(향후 노출 endpoint 도입 시 브로커 정본 우선). 2. 슬롯 전이: `pending_buy→holding`(avg_price/qty) 또는 `pending_sell→closed` 를 advisory lock 안에서 먼저 commit. 이 단계가 critical path. -3. ledger enqueue: 슬롯의 `portfolio_id`(== ledger `asset_id`, OQ-1 결정) 가 있으면 `LedgerTrade(asset_id, order_id, execution_id, side, qty, price, fee, tax, executed_at)` 을 `LedgerOutbox.enqueue` 로 적재. 행 식별자는 `correlation_id = {asset_id}:{order_id}:{execution_id}`(로컬 dedup). `portfolio_id` 부재면 ledger 기록 skip 후 슬롯 전이는 정상 유지. +3. ledger enqueue: 슬롯의 `portfolio_id`(== ledger `asset_id`, OQ-1 결정) 와 결정적 `execution_id`가 있으면 `LedgerTrade(asset_id, order_id, execution_id, side, qty, price, fee, tax, executed_at)` 을 `LedgerOutbox.enqueue` 로 적재한다. 행 식별자는 `correlation_id = {asset_id}:{order_id}:{execution_id}`(로컬 dedup)이다. completed fill이 `portfolio_id`/`execution_id` 누락 또는 unsupported actual venue 등 genuinely non-recordable 상태이면 `ledger_outbox` 대신 `ledger_skip` 에 unresolved 감사 row를 남기고 알람을 best-effort 발행한다. unknown actual venue alone(`order_venue=SOR`, `execution_venue=NULL`)는 계속 정상 outbox 경로다. 4. `clear_dispatch` 로 `broker_order_id` 해제. ## Flow: `ledger writer` (백그라운드, BAT-006) diff --git a/tests/adapters/persistence/test_ledger_skip_store.py b/tests/adapters/persistence/test_ledger_skip_store.py new file mode 100644 index 0000000..4235be0 --- /dev/null +++ b/tests/adapters/persistence/test_ledger_skip_store.py @@ -0,0 +1,89 @@ +from __future__ import annotations + +from datetime import UTC, date, datetime +from uuid import uuid4 + +from trex_core.adapters.persistence.ledger_skip_store import PostgreSQLLedgerSkipStore + + +def test_ledger_skip_store_round_trips_and_dedupes(pg_clean): + store = PostgreSQLLedgerSkipStore(pg_clean) + slot_id = uuid4() + + record, notify = store.record_skip( + instance_key="inst-1", + slot_id=slot_id, + portfolio_id="PF-1", + symbol="005930", + order_id="O-1", + execution_id=None, + side="buy", + qty=10.0, + price=1000.0, + executed_at=datetime(2026, 8, 24, 0, 0, tzinfo=UTC), + order_venue="SOR", + execution_venue="unknown", + skip_reason="missing_execution_id", + ) + again, notify_again = store.record_skip( + instance_key="inst-1", + slot_id=slot_id, + portfolio_id="PF-1", + symbol="005930", + order_id="O-1", + execution_id=None, + side="buy", + qty=10.0, + price=1000.0, + executed_at=datetime(2026, 8, 24, 0, 0, tzinfo=UTC), + order_venue="SOR", + execution_venue="unknown", + skip_reason="missing_execution_id", + ) + + assert notify is True + assert notify_again is False + assert again.skip_key == record.skip_key + listed = store.list_records(instance_key="inst-1", status="unresolved") + assert len(listed) == 1 + assert listed[0].skip_reason == "missing_execution_id" + + +def test_ledger_skip_store_filters_and_resolves(pg_clean): + store = PostgreSQLLedgerSkipStore(pg_clean) + slot_id = uuid4() + record, _ = store.record_skip( + instance_key="inst-1", + slot_id=slot_id, + portfolio_id="PF-2", + symbol="005930", + order_id="O-2", + execution_id="E-2", + side="sell", + qty=5.0, + price=2000.0, + executed_at=datetime(2026, 8, 24, 2, 0, tzinfo=UTC), + order_venue="KRX", + execution_venue="NXTX", + skip_reason="unsupported_execution_venue", + ) + + rows = store.list_records( + instance_key="inst-1", + portfolio_id="PF-2", + on_date=date(2026, 8, 24), + status="unresolved", + ) + assert [row.skip_key for row in rows] == [record.skip_key] + + resolved = store.resolve_for_fill( + instance_key="inst-1", + slot_id=slot_id, + side="sell", + order_id="O-2", + execution_id="E-2", + resolution_reason="manual_ledger_backfill", + ) + assert len(resolved) == 1 + assert resolved[0].status == "resolved" + assert resolved[0].resolution_reason == "manual_ledger_backfill" diff --git a/tests/adapters/persistence/test_migration_and_drift.py b/tests/adapters/persistence/test_migration_and_drift.py index 1d9fe08..e0aa8ca 100644 --- a/tests/adapters/persistence/test_migration_and_drift.py +++ b/tests/adapters/persistence/test_migration_and_drift.py @@ -26,6 +26,7 @@ _EXPECTED_TABLES = { "ledger_outbox", # BAT-006 / REQ-019 "snapshot_pause", # BAT-008 / REQ-032 "broker_account_snapshot", # dcy-041 / GitHub #45 + "ledger_skip", # GitHub #68 skipped-ledger durable audit } diff --git a/tests/application/test_execution_ledger.py b/tests/application/test_execution_ledger.py index ce15c44..8c7e164 100644 --- a/tests/application/test_execution_ledger.py +++ b/tests/application/test_execution_ledger.py @@ -6,12 +6,16 @@ critical path while the ledger write is eventually-consistent. """ from __future__ import annotations +from datetime import UTC + from trex_core.adapters.broker.kis import KISBroker from trex_core.adapters.broker.paper import PaperBroker from trex_core.adapters.ledger.in_memory import InMemoryLedger +from trex_core.adapters.messaging.in_memory import InMemoryTransport from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore -from trex_core.application.execution import ExecutionService +from trex_core.application.alarm_emitter import AlarmEmitter from trex_core.application.broker_port import FillStatus +from trex_core.application.execution import ExecutionService from trex_core.application.intents import BuyIntent, IntentService, SellIntent from trex_core.application.ledger_outbox import ( InMemoryLedgerOutbox, @@ -19,18 +23,62 @@ from trex_core.application.ledger_outbox import ( OutboxOutcome, WorkItemStatus, ) +from trex_core.application.ledger_skip import LedgerSkipRecord from trex_core.domain.slot import SlotService, SlotState from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository +class _SkipStore: + def __init__(self): + self.records = [] + self._by_key = {} -def _wire(broker, *, ledger=None, clock=None): + def record_skip(self, **kwargs): + record = LedgerSkipRecord( + skip_key=f"{kwargs['slot_id']}:{kwargs['side']}:{kwargs['skip_reason']}", + instance_key=kwargs["instance_key"], + slot_id=kwargs["slot_id"], + portfolio_id=kwargs["portfolio_id"], + symbol=kwargs["symbol"], + order_id=kwargs["order_id"], + execution_id=kwargs["execution_id"], + side=kwargs["side"], + qty=kwargs["qty"], + price=kwargs["price"], + executed_at=kwargs["executed_at"], + order_venue=kwargs["order_venue"], + execution_venue=kwargs["execution_venue"], + skip_reason=kwargs["skip_reason"], + status="unresolved", + resolution_reason=None, + resolved_at=None, + created_at=kwargs["executed_at"], + updated_at=kwargs["executed_at"], + ) + should_notify = record.skip_key not in self._by_key + self._by_key[record.skip_key] = record + self.records = list(self._by_key.values()) + return record, should_notify + + def resolve_for_fill(self, **_kwargs): + return [] + + + +def _wire(broker, *, ledger=None, clock=None, alarm_emitter=None, ledger_skip_store=None): repo = InMemorySlotRepository() svc = SlotService(repo) intents = IntentService(svc, repo, InMemoryIntentStore(), resolve_symbol=lambda pid: "005930") outbox = InMemoryLedgerOutbox(clock=clock) if clock else InMemoryLedgerOutbox() - execution = ExecutionService(svc, repo, broker, ledger_outbox=outbox) + execution = ExecutionService( + svc, + repo, + broker, + ledger_outbox=outbox, + alarm_emitter=alarm_emitter, + ledger_skip_store=ledger_skip_store, + ) writer = ( LedgerWriter(outbox, ledger or InMemoryLedger(), clock=clock) if clock else LedgerWriter(outbox, ledger or InMemoryLedger()) @@ -78,8 +126,10 @@ class MissingExecutionIdBroker(PaperBroker): def test_buy_fill_enqueues_one_ledger_item_and_writer_records_it(): ledger = InMemoryLedger() - repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0), - ledger=ledger) + repo, _, intents, execution, _, writer = _wire( + PaperBroker(fill_after_polls=0), + ledger=ledger, + ) sid = _buy(intents).slot_id r = execution.run_once() assert r.filled == 1 and r.ledger_enqueued == 1 @@ -127,8 +177,10 @@ def test_kis_sor_fill_without_actual_venue_enqueues_and_writes_once() -> None: def test_sell_fill_enqueues_sell_ledger_item(): ledger = InMemoryLedger() - repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0), - ledger=ledger) + repo, _, intents, execution, _, writer = _wire( + PaperBroker(fill_after_polls=0), + ledger=ledger, + ) sid = _buy(intents).slot_id execution.run_once() # -> holding + buy ledger item writer.run_once() # drain buy @@ -143,13 +195,16 @@ def test_sell_fill_enqueues_sell_ledger_item(): def test_portfolio_id_absent_skips_ledger_but_transitions_slot(): ledger = InMemoryLedger() - repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0), - ledger=ledger) + repo, svc, _, execution, outbox, _ = _wire( + PaperBroker(fill_after_polls=0), + ledger=ledger, + ) # dcy-011: buy admission now requires portfolio_id, so create the # pending_buy slot directly (bypass admission) to exercise the # execution-loop ledger skip + visibility (REQ-018 FR-4). - from trex_core.domain.slot.models import Slot from uuid import uuid4 + + from trex_core.domain.slot.models import Slot anon = Slot(slot_id=uuid4(), status=SlotState.INACTIVE) repo.add_slot(anon) svc.transit_slot(anon.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, @@ -176,14 +231,80 @@ def test_missing_execution_id_still_skips_malformed_ledger_work_item() -> None: assert outbox.all_items() == [] +def test_missing_portfolio_id_persists_skip_and_emits_alert() -> None: + transport = InMemoryTransport() + skip_store = _SkipStore() + repo, svc, _, execution, outbox, _ = _wire( + PaperBroker(fill_after_polls=0), + alarm_emitter=AlarmEmitter(transport), + ledger_skip_store=skip_store, + ) + from uuid import uuid4 + + from trex_core.domain.slot.models import Slot + + anon = Slot(slot_id=uuid4(), status=SlotState.INACTIVE) + repo.add_slot(anon) + svc.transit_slot( + anon.slot_id, + SlotState.INACTIVE, + SlotState.PENDING_BUY, + reason="buy", + actor="test", + symbol="005930", + portfolio_id=None, + ) + + result = execution.run_once() + + assert result.ledger_skipped == 1 + assert outbox.all_items() == [] + assert len(skip_store.records) == 1 + record = skip_store.records[0] + assert record.skip_reason == "missing_portfolio_id" + assert record.order_venue == "KRX" + alarms = transport.alarm_events() + assert len(alarms) == 1 + _, payload = alarms[0] + assert payload["code"] == "ledger_skip_unresolved" + assert payload["fields"]["skip_reason"] == "missing_portfolio_id" + + +def test_missing_execution_id_persists_skip_record() -> None: + transport = InMemoryTransport() + skip_store = _SkipStore() + repo, _, intents, execution, outbox, _ = _wire( + MissingExecutionIdBroker(), + alarm_emitter=AlarmEmitter(transport), + ledger_skip_store=skip_store, + ) + + sid = _buy(intents, iid="missing-execution-id").slot_id + assert sid is not None + result = execution.run_once() + + assert result.ledger_skipped == 1 + assert result.ledger_enqueued == 0 + assert repo.fetch_slot(sid).status is SlotState.HOLDING + assert outbox.all_items() == [] + assert len(skip_store.records) == 1 + record = skip_store.records[0] + assert record.skip_reason == "missing_execution_id" + assert record.portfolio_id == "PF-1" + assert record.execution_id is None + + def test_ledger_skip_emits_warning(caplog): """dcy-011 / REQ-018 FR-4 (S-3): ledger skip is logged, not silent.""" import logging ledger = InMemoryLedger() - repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0), - ledger=ledger) - from trex_core.domain.slot.models import Slot + repo, svc, _, execution, _, _ = _wire( + PaperBroker(fill_after_polls=0), + ledger=ledger, + ) from uuid import uuid4 + + from trex_core.domain.slot.models import Slot anon = Slot(slot_id=uuid4(), status=SlotState.INACTIVE) repo.add_slot(anon) svc.transit_slot(anon.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, @@ -197,8 +318,10 @@ def test_ledger_skip_emits_warning(caplog): def test_duplicate_execution_id_results_in_zero_duplicate_in_ledger(): """REQ-019 AC-1: same execution_id processed twice = 1 ledger entry.""" ledger = InMemoryLedger() - repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0), - ledger=ledger) + _, _, intents, execution, outbox, writer = _wire( + PaperBroker(fill_after_polls=0), + ledger=ledger, + ) _buy(intents) execution.run_once() writer.run_once() @@ -224,10 +347,10 @@ def test_retryable_failure_then_eventually_completes(): dcy-015: retryable rows back off, so advance a shared clock past each next_retry_at between writer passes. """ - from datetime import datetime, timedelta, timezone - now = {"t": datetime(2026, 6, 4, tzinfo=timezone.utc)} + from datetime import datetime, timedelta + now = {"t": datetime(2026, 6, 4, tzinfo=UTC)} ledger = InMemoryLedger(fail_n_then_succeed=2) - repo, svc, intents, execution, outbox, writer = _wire( + repo, _, intents, execution, outbox, writer = _wire( PaperBroker(fill_after_polls=0), ledger=ledger, clock=lambda: now["t"], ) sid = _buy(intents).slot_id @@ -253,8 +376,10 @@ def test_conflict_blocks_without_auto_retry(): # order id is PAPER-000001 / qty=10 => execution_id = paper:PAPER-000001:10. conflict_id = "paper:PAPER-000001:10" ledger = InMemoryLedger(conflict_executions={conflict_id}) - repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0), - ledger=ledger) + _, _, intents, execution, outbox, writer = _wire( + PaperBroker(fill_after_polls=0), + ledger=ledger, + ) _buy(intents) execution.run_once() stats = writer.run_once() @@ -271,8 +396,10 @@ def test_conflict_blocks_without_auto_retry(): def test_duplicate_completed_when_reprocessing_terminal_item(): """REQ-019 AC-4: a completed item reports duplicate_completed if re-listed.""" ledger = InMemoryLedger() - repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0), - ledger=ledger) + _, _, intents, execution, outbox, writer = _wire( + PaperBroker(fill_after_polls=0), + ledger=ledger, + ) _buy(intents) execution.run_once() writer.run_once() @@ -290,8 +417,10 @@ def test_duplicate_completed_when_reprocessing_terminal_item(): def test_slot_transition_proceeds_when_ledger_is_down(): """REQ-019 AC-5: portfolio-hub outage does not block the fill transition.""" ledger = InMemoryLedger(fail_n_then_succeed=100) - repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0), - ledger=ledger) + repo, _, intents, execution, outbox, writer = _wire( + PaperBroker(fill_after_polls=0), + ledger=ledger, + ) sid = _buy(intents).slot_id execution.run_once() assert repo.fetch_slot(sid).status is SlotState.HOLDING diff --git a/tests/conftest.py b/tests/conftest.py index efedc80..250933a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -18,6 +18,7 @@ from trex_core.adapters.broker.kis import KISBroker _PG_TABLES = ( "broker_account_snapshot", + "ledger_skip", "killswitch_exclusion", "snapshot_pause", "ledger_outbox", diff --git a/tests/runtime/test_cli_ledger_backfill.py b/tests/runtime/test_cli_ledger_backfill.py index 3c38c2c..e09282f 100644 --- a/tests/runtime/test_cli_ledger_backfill.py +++ b/tests/runtime/test_cli_ledger_backfill.py @@ -1,10 +1,13 @@ 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 @@ -63,6 +66,33 @@ def _evidence(): } +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([ @@ -80,7 +110,7 @@ def test_ledger_backfill_dry_run_emits_trade_and_writes_audit(monkeypatch, tmp_p 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"}) + 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) @@ -99,7 +129,7 @@ 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"}) + 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) @@ -115,7 +145,7 @@ def test_ledger_backfill_existing_completed_row_is_idempotent(monkeypatch, tmp_p 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"}) + 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", @@ -147,12 +177,23 @@ def test_ledger_backfill_existing_completed_row_is_idempotent(monkeypatch, tmp_p 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: _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) @@ -165,7 +206,7 @@ def test_ledger_backfill_live_enqueue_and_processes_exact_trade(monkeypatch, tmp 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"}) + 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: @@ -195,12 +236,23 @@ def test_ledger_backfill_live_enqueue_and_processes_exact_trade(monkeypatch, tmp 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: 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) @@ -210,3 +262,67 @@ def test_ledger_backfill_live_enqueue_and_processes_exact_trade(monkeypatch, tmp 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: _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" diff --git a/tests/runtime/test_cli_ledger_skips.py b/tests/runtime/test_cli_ledger_skips.py new file mode 100644 index 0000000..b67fc35 --- /dev/null +++ b/tests/runtime/test_cli_ledger_skips.py @@ -0,0 +1,132 @@ +from __future__ import annotations + +import json +from datetime import UTC, datetime +from types import SimpleNamespace +from uuid import UUID + +from trex_core.application.ledger_skip import LedgerSkipRecord +from trex_core.runtime import cli + + +def _cfg(): + return SimpleNamespace(dsn="postgresql://x/y", rabbitmq=None) + + +def _record(*, status="unresolved", resolution_reason=None): + return LedgerSkipRecord( + skip_key="ledger-skip:inst-1:slot-1:sell:O-1:E-1:missing_execution_id", + instance_key="inst-1", + slot_id=UUID("00000000-0000-0000-0000-000000000001"), + portfolio_id="PF-1", + symbol="005930", + order_id="O-1", + execution_id="E-1", + side="sell", + qty=10.0, + price=70000.0, + executed_at=datetime(2026, 8, 24, 0, 1, 2, tzinfo=UTC), + order_venue="SOR", + execution_venue="unknown", + skip_reason="missing_execution_id", + status=status, + resolution_reason=resolution_reason, + resolved_at=( + datetime(2026, 8, 24, 0, 5, 0, tzinfo=UTC) + if status == "resolved" + else None + ), + created_at=datetime(2026, 8, 24, 0, 1, 2, tzinfo=UTC), + updated_at=datetime(2026, 8, 24, 0, 5, 0, tzinfo=UTC), + ) + + +def test_ledger_skips_parser_accepts_list_and_resolve(): + parser = cli._build_parser() + args = parser.parse_args(["ledger", "skips", "list", "--status", "all"]) + assert args.ledger_cmd == "skips" + assert args.ledger_skips_cmd == "list" + args = parser.parse_args(["ledger", "skips", "resolve", "--skip-key", "abc"]) + assert args.ledger_skips_cmd == "resolve" + + +def test_ledger_skips_list_outputs_records(monkeypatch, capsys): + class _Store: + def __init__(self, *_args, **_kwargs): + pass + + def list_records(self, **_kwargs): + return [_record()] + + monkeypatch.setattr( + "trex_core.adapters.persistence.ledger_skip_store.PostgreSQLLedgerSkipStore", + _Store, + ) + args = SimpleNamespace( + cmd="ledger", + ledger_cmd="skips", + ledger_skips_cmd="list", + date="2026-08-24", + portfolio_id="PF-1", + status="unresolved", + limit=10, + instance="inst-1", + config=None, + config_path=None, + env_file=None, + json=True, + timeout=10.0, + ) + + rc = cli._cmd_ledger_skips_list(args, _cfg()) + + assert rc == 0 + out = json.loads(capsys.readouterr().out) + assert out["count"] == 1 + assert out["records"][0]["skip_reason"] == "missing_execution_id" + + +def test_ledger_skips_resolve_marks_record_and_emits_alert(monkeypatch, capsys): + class _Store: + def __init__(self, *_args, **_kwargs): + pass + + def resolve_by_key(self, skip_key, *, resolution_reason): + assert skip_key == "abc" + return _record(status="resolved", resolution_reason=resolution_reason) + + 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 + + monkeypatch.setattr( + "trex_core.adapters.persistence.ledger_skip_store.PostgreSQLLedgerSkipStore", + _Store, + ) + monkeypatch.setattr(cli, "_default_alarm_emitter", lambda cfg: (_Emitter(), _Transport())) + args = SimpleNamespace( + cmd="ledger", + ledger_cmd="skips", + ledger_skips_cmd="resolve", + skip_key="abc", + reason="operator_reconciled", + instance="inst-1", + config=None, + config_path=None, + env_file=None, + json=True, + timeout=10.0, + ) + + rc = cli._cmd_ledger_skips_resolve(args, _cfg()) + + assert rc == 0 + out = json.loads(capsys.readouterr().out) + assert out["result"] == "resolved" + assert published[0][0] == "ledger_skip_resolved" diff --git a/trex_core/adapters/broker/kis.py b/trex_core/adapters/broker/kis.py index a82af07..abb6982 100644 --- a/trex_core/adapters/broker/kis.py +++ b/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). diff --git a/trex_core/adapters/broker/paper.py b/trex_core/adapters/broker/paper.py index 9f04f71..bcf9683 100644 --- a/trex_core/adapters/broker/paper.py +++ b/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 diff --git a/trex_core/adapters/persistence/ledger_skip_store.py b/trex_core/adapters/persistence/ledger_skip_store.py new file mode 100644 index 0000000..aa08374 --- /dev/null +++ b/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], + ) diff --git a/trex_core/adapters/persistence/sql/025_create_ledger_skip.sql b/trex_core/adapters/persistence/sql/025_create_ledger_skip.sql new file mode 100644 index 0000000..d320ee3 --- /dev/null +++ b/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); diff --git a/trex_core/application/broker_port.py b/trex_core/application/broker_port.py index b56842a..297a217 100644 --- a/trex_core/application/broker_port.py +++ b/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 diff --git a/trex_core/application/execution.py b/trex_core/application/execution.py index ce84d14..5b427fd 100644 --- a/trex_core/application/execution.py +++ b/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). diff --git a/trex_core/application/ledger_skip.py b/trex_core/application/ledger_skip.py new file mode 100644 index 0000000..d2b3d96 --- /dev/null +++ b/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", + ) diff --git a/trex_core/runtime/cli.py b/trex_core/runtime/cli.py index c05eb83..37da86b 100644 --- a/trex_core/runtime/cli.py +++ b/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"): diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index 18a4ca2..a7c88a2 100644 --- a/trex_core/runtime/daemon.py +++ b/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