diff --git a/.proofline/criteria/ac-0023.md b/.proofline/criteria/ac-0023.md index 1ab13c4..dd61d79 100644 --- a/.proofline/criteria/ac-0023.md +++ b/.proofline/criteria/ac-0023.md @@ -7,7 +7,7 @@ status: active ## Criterion -`case_code=B`(core slot↔KIS 수량 불일치) 또는 `case_code=C`(portfolio-hub↔KIS 수량 불일치) reconcile-warning이 활성인 `holding` slot은 `trex.slots.req.` 응답에서 제외해야 한다. 동일 slot의 B/C 원인은 서로 덮어쓰지 않고 동시에 보존하며, 응답에는 제외된 slot 수와 활성 case별 count를 포함한다. 신규 buy/sell intent는 같은 durable warning 원인 집합을 재확인하여 `reconcile_warned` 사유로 거부한다. warn-store를 읽지 못하거나 warning metadata가 비어 있거나 지원하지 않는 case를 포함하면 안전한 eligibility를 입증할 수 없으므로 slots RPC는 `degraded=true` / `reason="warn_store_unavailable"`, intent는 `reconcile_warning_unavailable`(조회 실패) 또는 `reconcile_warned`(empty/unsupported metadata)로 fail-closed 처리해야 한다. `sync`는 수량 일치 확인 후 B-case만 해제하고, `verify`는 instance+symbol 단위 KIS와 portfolio-hub 합계가 다시 일치한 경우 C-case만 해제하여 다른 활성 원인을 보존한다. C-case 수량 대조는 **instance_key + symbol 단위 합계**로 수행하며, 개별 portfolio 수량을 KIS 종목 합계와 직접 비교해서는 안 된다. 이 경로는 주문 submit/cancel, ledger write, slot lifecycle mutation을 유발하지 않아야 한다. +`case_code=B`(core slot↔KIS 수량 불일치) 또는 `case_code=C`(portfolio-hub↔KIS 수량 불일치) reconcile-warning이 활성인 `holding` slot은 `trex.slots.req.` 응답에서 제외해야 한다. 동일 slot의 B/C 원인은 서로 덮어쓰지 않고 동시에 보존하며, 응답에는 제외된 slot 수와 활성 case별 count를 포함한다. 신규 buy/sell intent는 같은 durable warning 원인 집합을 재확인하여 `reconcile_warned` 사유로 거부한다. warn-store를 읽지 못하거나 warning metadata가 비어 있거나 지원하지 않는 case를 포함하면 안전한 eligibility를 입증할 수 없으므로 slots RPC는 `degraded=true` / `reason="warn_store_unavailable"`, intent는 `reconcile_warning_unavailable`(조회 실패) 또는 `reconcile_warned`(empty/unsupported metadata)로 fail-closed 처리해야 한다. B-case는 수량 일치 확인을 마친 operator `sync` 또는 AC-0032~AC-0034를 모두 만족해 qty·avg_price·audit·B warning clear가 한 transaction으로 commit된 `auto_reconcile`만 해제할 수 있다. `verify`는 instance+symbol 단위 KIS와 portfolio-hub 합계가 다시 일치한 경우 C-case만 해제하여 다른 활성 원인을 보존한다. C-case 수량 대조는 **instance_key + symbol 단위 합계**로 수행하며, 개별 portfolio 수량을 KIS 종목 합계와 직접 비교해서는 안 된다. 이 경로는 주문 submit/cancel, portfolio-hub 또는 ledger write, slot lifecycle mutation을 유발하지 않아야 한다. ## Verification @@ -23,5 +23,6 @@ status: active - 해제 대상: HOLDING 뿐 아니라 모든 상태의 slot에 걸린 C-case 경고를 포함한다. - 해제 실패: 해제 도중 예외가 발생하면 `result='clear_failed'`를 반환하고 완전 성공으로 보고하지 않아야 한다(부분 성공 = 실패). - 시간이 지났다는 이유만으로 자동 해제하지 않음. +- 자동 해제: AC-0032의 authoritative evidence와 eligibility를 통과하고 AC-0033 transaction이 성공한 경우에만 B-case를 해제한다. CAS 실패·불완전 evidence·write 실패에서는 B-case를 유지한다. - 비침습 조건: slots RPC 호출 전후로 slot 상태 전이, broker mutation, ledger write side effect가 없어야 한다. - AC-0022 우선순위: B 또는 C reconcile warning으로 제외된 slot에는 broker-authoritative `qty` 응답 형식(AC-0022)을 적용하지 않는다. diff --git a/.proofline/criteria/ac-0024.md b/.proofline/criteria/ac-0024.md index 8a084e3..4cf1972 100644 --- a/.proofline/criteria/ac-0024.md +++ b/.proofline/criteria/ac-0024.md @@ -7,11 +7,11 @@ status: active ## Criterion -`trex-core`는 B-case(core slot quantity와 broker quantity 불일치)를 stable slot별로, C-case(같은 instance·symbol의 portfolio-hub `asset_quantity` 합계와 broker quantity 불일치)를 `instance_key + symbol`별로 하나의 durable notification lifecycle로 추적해야 한다. 새 mismatch와 동일 identity의 case별 비교 quantity tuple 변경은 sanitized `trex.alerts` detected/changed event를 각각 한 번 발행하고, 성공한 publish fingerprint를 process restart 뒤에도 보존하여 unchanged scheduler tick을 억제해야 한다. B-case는 operator `sync`, C-case는 operator `verify`의 authoritative durable warning clear에서만 resolved event를 한 번 발행해야 하며 scheduler 관찰만으로 recovery를 발행해서는 안 된다. Event는 lifecycle state, `instance_key`, `symbol`, `case_code`, broker quantity와 case별 비교 quantity를 포함하고 account number/hash, credential, token, DSN, raw payload와 exception 본문을 포함해서는 안 된다. Publish 성공 뒤에만 fingerprint를 전진시키고 publish 또는 notification-state persistence 실패는 reconciliation, warning set/clear, sync/verify, scheduler와 slot lifecycle을 중단하거나 바꾸지 않아야 한다. Resolved publish 실패 시 notification state가 writable이면 pending transition을 durable하게 보존하여 후속 retry opportunity에서 재발행한다. Notification-state persistence 자체가 실패하면 authoritative warning clear를 유지하고 event 유실 가능성을 sanitized log로 남긴다. +`trex-core`는 B-case(core slot quantity와 broker quantity 불일치)를 stable slot별로, C-case(같은 instance·symbol의 portfolio-hub `asset_quantity` 합계와 broker quantity 불일치)를 `instance_key + symbol`별로 하나의 durable notification lifecycle로 추적해야 한다. 새 mismatch와 동일 identity의 case별 비교 quantity tuple 변경은 sanitized `trex.alerts` detected/changed event를 각각 한 번 발행하고, 성공한 publish fingerprint를 process restart 뒤에도 보존하여 unchanged scheduler tick을 억제해야 한다. B-case는 operator `sync` 또는 AC-0033 transaction이 성공한 `auto_reconcile`, C-case는 operator `verify`의 authoritative durable warning clear에서만 resolved event를 한 번 발행해야 하며 수량 일치의 단순 scheduler 관찰만으로 recovery를 발행해서는 안 된다. Event는 lifecycle state, `instance_key`, `symbol`, `case_code`, broker quantity와 case별 비교 quantity를 포함하고 account number/hash, credential, token, DSN, raw payload와 exception 본문을 포함해서는 안 된다. Publish 성공 뒤에만 fingerprint를 전진시키고 publish 또는 notification-state persistence 실패는 reconciliation, warning set/clear, sync/verify/auto_reconcile, scheduler와 slot lifecycle을 중단하거나 바꾸지 않아야 한다. Resolved publish 실패 시 notification state가 writable이면 pending transition을 durable하게 보존하여 후속 retry opportunity에서 재발행한다. Notification-state persistence 자체가 실패하면 authoritative warning clear를 유지하고 event 유실 가능성을 sanitized log로 남긴다. ## Verification -- 성공 조건: 새로운 B-case slot mismatch는 B detected event 한 건, 새로운 C-case symbol aggregate mismatch는 portfolio 수와 무관하게 C detected event 한 건을 만든다. 동일 fingerprint의 반복 tick은 추가 event를 만들지 않고 quantity tuple 변경은 changed event 한 건을 만든다. process restart 모사 뒤 unchanged mismatch는 억제되며, B `sync`와 C aggregate `verify` clear는 각 identity의 resolved event 한 건만 만든다. +- 성공 조건: 새로운 B-case slot mismatch는 B detected event 한 건, 새로운 C-case symbol aggregate mismatch는 portfolio 수와 무관하게 C detected event 한 건을 만든다. 동일 fingerprint의 반복 tick은 추가 event를 만들지 않고 quantity tuple 변경은 changed event 한 건을 만든다. process restart 모사 뒤 unchanged mismatch는 억제되며, B `sync` 또는 성공한 `auto_reconcile`과 C aggregate `verify` clear는 각 identity의 resolved event 한 건만 만든다. - publish 실패 조건: transport failure에서는 detected/changed/resolved fingerprint가 전진하지 않아야 한다. detected/changed는 다음 동일 detection에서 재시도하고, resolved는 notification state가 writable한 경우 warning clear를 유지하면서 pending transition을 durable하게 보존해 후속 retry opportunity에서 재시도해야 한다. Notification-state persistence 실패도 business path를 중단하지 않아야 하며 resolved event 유실 가능성을 credential·DSN·raw exception 없는 log로 남겨야 한다. - 경계 조건: C-case 동일 symbol의 여러 portfolio/slot이 중복 event를 만들거나, scheduler의 단순 quantity 일치가 durable warning clear·resolved event를 만들거나, B slot 간 fingerprint가 섞이거나, restart 뒤 unchanged event가 재발행되거나, publish 전 fingerprint 기록으로 event가 유실되면 FAIL이다. - 보안·비변경 조건: 모든 lifecycle event의 allowlist fields와 rendered payload에 account number/hash, credential, token, DSN, raw KIS/portfolio-hub payload 및 exception 본문이 없어야 한다. 알림 전후 broker call mutation, portfolio-hub write, slot lifecycle/quantity mutation과 기존 AC-0023 blocking/clear semantics 변화가 없어야 한다. diff --git a/.proofline/criteria/ac-0032.md b/.proofline/criteria/ac-0032.md new file mode 100644 index 0000000..617f9b4 --- /dev/null +++ b/.proofline/criteria/ac-0032.md @@ -0,0 +1,18 @@ +--- +id: "ac-0032" +status: active +--- + +# B-case auto-reconcile authoritative evidence와 eligibility + +## Criterion + +`reconcile_core_holdings`는 하나의 성공한 broker balance snapshot에서 `broker_qty > 0`이고, 같은 scheduler refresh의 portfolio-hub item이 slot의 `instance_key`·`portfolio_id`·`symbol`과 일치하며 `asset_quantity == broker_qty`일 때만 B-case 자동 동기화를 검토한다. 별도 canonical cost-basis 응답은 같은 portfolio·symbol·current quantity를 증명하고, fetch 시각이 60초 이내이며, `data_completeness_warning`이 없고, 양의 finite `cost_basis_per_unit`과 수량×평단에 일치하는 양의 finite `total_cost_basis`를 제공해야 한다. 대상 slot은 `holding`, 미퇴역, broker dispatch/pending metadata 없음이어야 한다. 첫 구현에서 `inactive`, `pending_buy`, `pending_sell`, `error`, retired slot은 자동 동기화하지 않는다. 어느 evidence든 missing·stale·invalid·불일치이거나 read/set-warning가 실패하면 fail-closed하여 slot과 warning을 변경하지 않는다. Operator `sync`의 legacy/graceful-degrade 계약은 변경하지 않는다. + +## Verification + +- hub·broker 수량과 canonical cost-basis identity/current quantity가 모두 일치하는 holding만 통과한다. +- hub/broker mismatch, cost-basis warning·stale·legacy-only·missing·non-finite·0 이하·identity mismatch·total mismatch는 모두 자동 변경 없이 B warning을 유지한다. +- broker balance는 reconcile pass당 한 번만 읽고 판단과 write가 같은 snapshot을 사용한다. +- pending/dispatch metadata, retired 또는 holding 외 상태는 fail-closed한다. +- 기존 `sync_slot_to_broker`의 canonical/legacy diagnostics와 inactive onboarding 회귀 테스트는 그대로 통과한다. \ No newline at end of file diff --git a/.proofline/criteria/ac-0033.md b/.proofline/criteria/ac-0033.md new file mode 100644 index 0000000..403cf5a --- /dev/null +++ b/.proofline/criteria/ac-0033.md @@ -0,0 +1,18 @@ +--- +id: "ac-0033" +status: active +--- + +# Holding mirror qty·avg_price·B warning·audit 원자적 갱신 + +## Criterion + +AC-0032를 통과한 자동 동기화는 instance advisory lock과 database transaction 안에서 관측한 slot identity/status/qty/avg_price 및 dispatch metadata를 CAS 재검증하고, `qty=broker_qty`, `avg_price=portfolio-hub cost_basis_per_unit` 갱신, `case_code=B` warning만 삭제, `actor=auto_reconcile`·`reason=hub_broker_matched_core_mirror_stale` transition audit 추가를 하나의 commit으로 수행해야 한다. Audit extra에는 allowlist된 before/after qty·avg_price와 hub/broker quantity, portfolio_id, symbol만 포함한다. CAS 실패, B warning 부재 또는 어느 write든 실패하면 네 변경 모두 없어야 한다. C/unknown warning과 다른 instance·symbol·slot은 보존한다. 같은 evidence 재실행은 no-op이며 audit를 중복 추가하지 않는다. + +## Verification + +- 성공 시 slot status는 holding으로 유지되고 qty·avg_price, B clear, audit 한 건이 함께 반영된다. +- C와 기타 warning은 보존된다. +- CAS race와 강제 예외에서 transaction rollback 후 slot·warning·audit가 모두 이전 상태다. +- PostgreSQL 실제 transaction 테스트와 in-memory parity 테스트로 commit, rollback, isolation, idempotency를 검증한다. +- broker order submit/cancel, portfolio-hub write, ledger/outbox write는 발생하지 않는다. \ No newline at end of file diff --git a/.proofline/criteria/ac-0034.md b/.proofline/criteria/ac-0034.md new file mode 100644 index 0000000..d56ad2a --- /dev/null +++ b/.proofline/criteria/ac-0034.md @@ -0,0 +1,17 @@ +--- +id: "ac-0034" +status: active +--- + +# Auto-reconcile scheduler·daemon wiring과 오류 격리 + +## Criterion + +Production daemon은 기존 portfolio-hub read adapter, authoritative reconciliation broker, slot persistence, durable warning/notification store를 `SnapshotScheduler`의 B-case auto-reconcile 경로에 명시적으로 연결해야 한다. Scheduler는 성공한 universe refresh의 items와 core-holding worker가 한 번 읽은 complete broker snapshot을 사용하며 ad-hoc broker/Hub write client 또는 legacy cost-basis fallback을 만들지 않는다. Auto-reconcile evidence/read/CAS/transaction 오류는 해당 slot을 `qty_warning`으로 유지하고 scheduler·다른 slot·C-case reconciliation을 중단하지 않는다. 성공 commit 뒤에만 B resolved notification을 발행하며 publish 실패의 pending retry·fingerprint·instance isolation 및 sanitized payload 계약은 AC-0024를 유지한다. + +## Verification + +- daemon assembly test가 read port, reconciliation broker, atomic store, warn/notification store의 production 연결을 확인한다. +- scheduler test가 현재 refresh items 전달, broker 단일 read, auto-sync 성공/실패 격리와 다음 tick idempotency를 확인한다. +- resolved notification은 commit/B clear 뒤 한 번 발생하고 publish 실패는 business commit을 되돌리지 않으며 pending retry된다. +- synthetic 회귀: symbol `019685`, hub=broker=726, core=1452, canonical WAC≈2172.46829187에서 core만 726/WAC로 갱신되고 B만 해제되며 주문·Hub·ledger side effect가 없다. \ No newline at end of file diff --git a/.proofline/identities.json b/.proofline/identities.json index 7ce36f9..7ca9e3f 100644 --- a/.proofline/identities.json +++ b/.proofline/identities.json @@ -1,5 +1,5 @@ { "schema_version": 1, "next_line_number": 10, - "next_ac_number": 32 + "next_ac_number": 35 } diff --git a/tests/adapters/persistence/test_auto_reconcile.py b/tests/adapters/persistence/test_auto_reconcile.py new file mode 100644 index 0000000..cd4ebe5 --- /dev/null +++ b/tests/adapters/persistence/test_auto_reconcile.py @@ -0,0 +1,275 @@ +"""PostgreSQL auto-reconcile transaction sequencing tests.""" + +from __future__ import annotations + +from datetime import UTC, datetime +from types import SimpleNamespace +from uuid import uuid4 + +import psycopg +import pytest + +from trex_core.adapters.persistence.auto_reconcile import PostgreSQLAutoReconcileStore +from trex_core.adapters.persistence.postgresql import PostgreSQLSlotRepository +from trex_core.adapters.portfolio.reconcile_warning_store import ( + PostgreSQLReconcileWarningStore, +) +from trex_core.application.auto_reconcile import AutoReconcileCommand +from trex_core.domain.slot import SlotService, SlotState + + +class FakeCursor: + def __init__(self, rowcounts=(1, 1), fail_on_insert=False): + self.rowcounts = iter(rowcounts) + self.rowcount = -1 + self.fail_on_insert = fail_on_insert + self.calls = [] + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def execute(self, sql, params=None): + normalized = " ".join(sql.split()) + self.calls.append((normalized, params)) + if normalized.startswith("UPDATE daily_slot"): + self.rowcount = next(self.rowcounts) + elif normalized.startswith("DELETE FROM slot_reconcile_warning"): + self.rowcount = next(self.rowcounts) + elif normalized.startswith("INSERT INTO slot_transition_log") and self.fail_on_insert: + raise RuntimeError("audit insert failed") + + +class FakeConnection: + def __init__(self, cursor): + self._cursor = cursor + self.commits = 0 + self.rollbacks = 0 + + def cursor(self): + return self._cursor + + def commit(self): + self.commits += 1 + + def rollback(self): + self.rollbacks += 1 + + def close(self): + pass + + +class FakeConnectionContext: + def __init__(self, conn): + self.conn = conn + + def __enter__(self): + return self.conn + + def __exit__(self, exc_type, exc, tb): + return False + + +def _command(): + return AutoReconcileCommand( + slot_id=uuid4(), + instance_key="inst-a", + portfolio_id="pf-a", + symbol="019685", + expected_qty=1452.0, + expected_avg_price=2100.0, + target_qty=726.0, + target_avg_price=2172.46829187, + hub_quantity=726.0, + broker_quantity=726, + evidence_observed_at=datetime.now(tz=UTC), + ) + + +def _install(monkeypatch, conn): + module = SimpleNamespace(connect=lambda dsn, autocommit=False: FakeConnectionContext(conn)) + monkeypatch.setattr( + "trex_core.adapters.persistence.auto_reconcile._require_psycopg", + lambda _module=None: module, + ) + + +def test_postgresql_auto_reconcile_commits_update_b_clear_and_audit(monkeypatch): + cursor = FakeCursor(rowcounts=(1, 1)) + conn = FakeConnection(cursor) + _install(monkeypatch, conn) + + assert PostgreSQLAutoReconcileStore("postgresql://unused").apply_holding_mirror(_command()) + + assert conn.commits == 1 + assert conn.rollbacks == 0 + statements = [sql for sql, _params in cursor.calls] + assert statements[0].startswith("SELECT pg_advisory_xact_lock") + assert "status='holding'" in statements[1] + assert "avg_price IS NOT DISTINCT FROM" in statements[1] + assert "broker_order_id IS NULL" in statements[1] + assert statements[2].startswith("DELETE FROM slot_reconcile_warning") + assert cursor.calls[2][1][1] == "B" + assert statements[3].startswith("INSERT INTO slot_transition_log") + assert cursor.calls[3][1][2:4] == ( + "hub_broker_matched_core_mirror_stale", + "auto_reconcile", + ) + + +def test_postgresql_auto_reconcile_cas_failure_rolls_back_without_warning_or_audit(monkeypatch): + cursor = FakeCursor(rowcounts=(0,)) + conn = FakeConnection(cursor) + _install(monkeypatch, conn) + + assert not PostgreSQLAutoReconcileStore("postgresql://unused").apply_holding_mirror(_command()) + + assert conn.commits == 0 + assert conn.rollbacks == 1 + assert len(cursor.calls) == 2 + + +def test_postgresql_auto_reconcile_audit_failure_rolls_back_slot_and_warning(monkeypatch): + cursor = FakeCursor(rowcounts=(1, 1), fail_on_insert=True) + conn = FakeConnection(cursor) + _install(monkeypatch, conn) + + with pytest.raises(RuntimeError, match="audit insert failed"): + PostgreSQLAutoReconcileStore("postgresql://unused").apply_holding_mirror(_command()) + + assert conn.commits == 0 + assert conn.rollbacks == 1 + assert len(cursor.calls) == 4 + + +def _seed_real_transaction_state(pg_clean): + repo = PostgreSQLSlotRepository(pg_clean) + service = SlotService(repo, instance_key="inst-a") + warning = PostgreSQLReconcileWarningStore(pg_clean) + repo.reconcile_universe("inst-a", {"pf-a": "019685"}) + slot = repo.find_slot_by_portfolio("inst-a", "pf-a") + assert slot is not None + pending = service.transit_slot( + slot.slot_id, + SlotState.INACTIVE, + SlotState.PENDING_BUY, + reason="buy", + actor="test", + symbol="019685", + ) + holding = service.transit_slot( + pending.slot_id, + SlotState.PENDING_BUY, + SlotState.HOLDING, + reason="filled", + actor="test", + qty=1452.0, + avg_price=2100.0, + ) + warning.set_warned( + str(holding.slot_id), + portfolio_id="pf-a", + case_code="B", + actor="reconcile", + ) + warning.set_warned( + str(holding.slot_id), + portfolio_id="pf-a", + case_code="C", + actor="reconcile", + ) + command = AutoReconcileCommand( + slot_id=holding.slot_id, + instance_key="inst-a", + portfolio_id="pf-a", + symbol="019685", + expected_qty=1452.0, + expected_avg_price=2100.0, + target_qty=726.0, + target_avg_price=2172.46829187, + hub_quantity=726.0, + broker_quantity=726, + evidence_observed_at=datetime.now(tz=UTC), + ) + return repo, warning, holding, command + + +def test_postgresql_auto_reconcile_real_transaction_commit(pg_clean): + repo, warning, holding, command = _seed_real_transaction_state(pg_clean) + + assert PostgreSQLAutoReconcileStore(pg_clean).apply_holding_mirror(command) + + current = repo.fetch_slot(holding.slot_id) + assert current is not None + assert current.status is SlotState.HOLDING + assert current.qty == 726.0 + assert current.avg_price == pytest.approx(2172.46829187) + info = warning.get_warning_info(str(holding.slot_id)) + assert info is not None + assert info["case_codes"] == ["C"] + with psycopg.connect(pg_clean) as conn: + row = conn.execute( + "SELECT actor, reason FROM slot_transition_log " + "WHERE slot_id = %s ORDER BY ts DESC LIMIT 1", + (holding.slot_id,), + ).fetchone() + assert row == ("auto_reconcile", "hub_broker_matched_core_mirror_stale") + + +def test_postgresql_auto_reconcile_real_transaction_audit_failure_rolls_back(pg_clean): + repo, warning, holding, command = _seed_real_transaction_state(pg_clean) + with psycopg.connect(pg_clean) as conn: + before_audit_row = conn.execute( + "SELECT count(*) FROM slot_transition_log WHERE slot_id = %s", + (holding.slot_id,), + ).fetchone() + assert before_audit_row is not None + before_audit_count = before_audit_row[0] + + with psycopg.connect(pg_clean, autocommit=True) as conn: + conn.execute( + """ + CREATE FUNCTION trex_test_fail_auto_reconcile_audit() + RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + RAISE EXCEPTION 'forced auto-reconcile audit failure'; + END + $$ + """ + ) + conn.execute( + """ + CREATE TRIGGER trex_test_fail_auto_reconcile_audit + BEFORE INSERT ON slot_transition_log + FOR EACH ROW WHEN (NEW.actor = 'auto_reconcile') + EXECUTE FUNCTION trex_test_fail_auto_reconcile_audit() + """ + ) + + try: + with pytest.raises(psycopg.errors.RaiseException, match="forced auto-reconcile"): + PostgreSQLAutoReconcileStore(pg_clean).apply_holding_mirror(command) + + current = repo.fetch_slot(holding.slot_id) + assert current is not None + assert current.qty == 1452.0 + assert current.avg_price == 2100.0 + info = warning.get_warning_info(str(holding.slot_id)) + assert info is not None + assert info["case_codes"] == ["B", "C"] + with psycopg.connect(pg_clean) as conn: + after_audit_row = conn.execute( + "SELECT count(*) FROM slot_transition_log WHERE slot_id = %s", + (holding.slot_id,), + ).fetchone() + assert after_audit_row is not None + after_audit_count = after_audit_row[0] + assert after_audit_count == before_audit_count + finally: + with psycopg.connect(pg_clean, autocommit=True) as conn: + conn.execute( + "DROP TRIGGER IF EXISTS trex_test_fail_auto_reconcile_audit ON slot_transition_log" + ) + conn.execute("DROP FUNCTION IF EXISTS trex_test_fail_auto_reconcile_audit()") diff --git a/tests/adapters/portfolio/test_hub_read.py b/tests/adapters/portfolio/test_hub_read.py index c719501..69d0b5f 100644 --- a/tests/adapters/portfolio/test_hub_read.py +++ b/tests/adapters/portfolio/test_hub_read.py @@ -117,6 +117,58 @@ def test_get_portfolios_network_error_raises_unavailable(): # REQ-123: cost-basis parsing (canonical + diagnostics) # --------------------------------------------------------------------------- + +def test_get_cost_basis_evidence_parses_identity_quantity_and_complete_basis(): + def handler(req: httpx.Request) -> httpx.Response: + assert req.url.path == "/v1/portfolios/pf-1/holdings/cost-basis" + return httpx.Response(200, json={ + "portfolio_id": "pf-1", + "symbol": "019685", + "current_quantity": "726", + "cost_basis_per_unit": "2172.46829187", + "total_cost_basis": str(726 * 2172.46829187), + "data_completeness_warning": None, + }) + + adapter = PortfolioHubReadAdapter("https://pf.test", token="tok", client=_client(handler)) + evidence = adapter.get_cost_basis_evidence("pf-1") + assert evidence.portfolio_id == "pf-1" + assert evidence.symbol == "019685" + assert evidence.current_quantity == 726.0 + assert evidence.cost_basis_per_unit == pytest.approx(2172.46829187) + assert evidence.data_completeness_warning is None + assert evidence.observed_at.tzinfo is not None + + +@pytest.mark.parametrize("payload", [ + {"cost_basis_per_unit": 1, "total_cost_basis": 1}, + { + "portfolio_id": "wrong", "symbol": "019685", "current_quantity": 1, + "cost_basis_per_unit": 1, "total_cost_basis": 1, + }, + { + "portfolio_id": "pf-1", "symbol": "019685", "current_quantity": 1, + "cost_basis_per_unit": "NaN", "total_cost_basis": 1, + }, +]) +def test_get_cost_basis_evidence_rejects_missing_mismatched_or_nonfinite(payload): + def handler(req: httpx.Request) -> httpx.Response: + return httpx.Response(200, json=payload) + + adapter = PortfolioHubReadAdapter("https://pf.test", token="tok", client=_client(handler)) + with pytest.raises(PortfolioHubUnavailableError): + adapter.get_cost_basis_evidence("pf-1") + + +def test_get_cost_basis_evidence_never_uses_legacy_fields(): + def handler(req: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"avg_cost": 1234, "wac": 1234}) + + adapter = PortfolioHubReadAdapter("https://pf.test", token="tok", client=_client(handler)) + with pytest.raises(PortfolioHubUnavailableError): + adapter.get_cost_basis_evidence("pf-1") + + def test_get_cost_basis_with_diagnostics_prefers_canonical_when_valid(): """REQ-123 AC-1: cost_basis_per_unit가 있으면 우선 사용.""" diff --git a/tests/application/test_auto_reconcile.py b/tests/application/test_auto_reconcile.py new file mode 100644 index 0000000..9695e17 --- /dev/null +++ b/tests/application/test_auto_reconcile.py @@ -0,0 +1,326 @@ +"""Issue #94 holding mirror auto-reconciliation tests.""" + +from __future__ import annotations + +from dataclasses import replace +from datetime import UTC, datetime, timedelta + +import pytest + +from trex_core.adapters.persistence.auto_reconcile import InMemoryAutoReconcileStore +from trex_core.application.auto_reconcile import ( + AutoReconcileCommand, + HoldingsCostBasisEvidence, +) +from trex_core.application.broker_port import BalanceItem +from trex_core.application.portfolio import ( + InMemoryReconcileNotificationStore, + InMemoryReconcileWarningStore, + PortfolioItem, +) +from trex_core.application.reconciliation import reconcile_core_holdings +from trex_core.domain.slot import SlotService, SlotState +from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository + + +class CountingBroker: + def __init__(self, quantity: int = 726) -> None: + self.quantity = quantity + self.balance_reads = 0 + self.order_calls = 0 + + def name(self): + return "counting" + + def get_balance(self): + self.balance_reads += 1 + return [BalanceItem(symbol="019685", hldg_qty=self.quantity)] + + def get_current_price(self, symbol): + return 1.0 + + def submit_order(self, request): + self.order_calls += 1 + raise AssertionError("auto reconcile must not submit orders") + + def poll_fill(self, *args, **kwargs): + raise AssertionError("auto reconcile must not poll orders") + + def cancel_order(self, *args, **kwargs): + raise AssertionError("auto reconcile must not cancel orders") + + +class EvidencePort: + def __init__(self, evidence: HoldingsCostBasisEvidence) -> None: + self.evidence = evidence + self.calls = 0 + + def get_cost_basis_evidence(self, portfolio_id: str) -> HoldingsCostBasisEvidence: + self.calls += 1 + assert portfolio_id == "pf-019685" + return self.evidence + + +class RecordingAlarm: + def __init__(self) -> None: + self.calls: list[tuple[str, dict]] = [] + + def publish_alarm(self, code, title, *, fields=None, severity="critical"): + self.calls.append((code, fields or {})) + return True + + +def _holding(repo: InMemorySlotRepository, svc: SlotService): + repo.reconcile_universe("default", {"pf-019685": "019685"}) + slot = repo.find_slot_by_portfolio("default", "pf-019685") + assert slot is not None + pending = svc.transit_slot( + slot.slot_id, + SlotState.INACTIVE, + SlotState.PENDING_BUY, + reason="buy", + actor="test", + symbol="019685", + ) + return svc.transit_slot( + pending.slot_id, + SlotState.PENDING_BUY, + SlotState.HOLDING, + reason="filled", + actor="test", + avg_price=2100.0, + qty=1452.0, + ) + + +def _evidence(**overrides) -> HoldingsCostBasisEvidence: + values = { + "portfolio_id": "pf-019685", + "symbol": "019685", + "current_quantity": 726.0, + "cost_basis_per_unit": 2172.46829187, + "total_cost_basis": 726.0 * 2172.46829187, + "data_completeness_warning": None, + "observed_at": datetime.now(tz=UTC), + } + values.update(overrides) + return HoldingsCostBasisEvidence(**values) + + +def _run( + *, + evidence=None, + mutate_slot=None, + auto_store_wrapper=None, + notification=False, + portfolio_items=None, +): + if portfolio_items is None: + portfolio_items = [PortfolioItem("pf-019685", "019685", 0, 726.0)] + repo = InMemorySlotRepository() + svc = SlotService(repo) + warn = InMemoryReconcileWarningStore() + slot = _holding(repo, svc) + if mutate_slot is not None: + repo.update_slot(mutate_slot(slot)) + slot = repo.fetch_slot(slot.slot_id) + assert slot is not None + warn.set_warned( + str(slot.slot_id), + portfolio_id="pf-019685", + case_code="C", + actor="test", + reason="preserve-me", + ) + base_store = InMemoryAutoReconcileStore(repo, warn) + auto_store = auto_store_wrapper(base_store, repo) if auto_store_wrapper else base_store + broker = CountingBroker() + alarm = RecordingAlarm() if notification else None + notifications = InMemoryReconcileNotificationStore() if notification else None + port = EvidencePort(evidence or _evidence()) + decisions = reconcile_core_holdings( + slot_repo=repo, + slot_service=svc, + broker=broker, # type: ignore[arg-type] + instance_key="default", + warn_store=warn, + alarm_emitter=alarm, # type: ignore[arg-type] + notification_store=notifications, + portfolio_items=portfolio_items, + cost_basis_port=port, + auto_reconcile_store=auto_store, + ) + return repo, warn, slot, broker, port, decisions, alarm + + +def test_auto_reconcile_updates_qty_avg_audit_and_only_clears_b(): + repo, warn, slot, broker, port, decisions, _alarm = _run() + + current = repo.fetch_slot(slot.slot_id) + assert current is not None + assert decisions[str(slot.slot_id)] == "auto_synced" + assert current.qty == 726.0 + assert current.avg_price == pytest.approx(2172.46829187) + assert current.status is SlotState.HOLDING + assert warn.get_warning_info(str(slot.slot_id))["case_codes"] == ["C"] + audit = repo.transition_log_for(slot.slot_id)[-1] + assert audit.actor == "auto_reconcile" + assert audit.reason == "hub_broker_matched_core_mirror_stale" + assert audit.extra["before_qty"] == 1452.0 + assert audit.extra["after_qty"] == 726.0 + assert broker.balance_reads == 1 + assert broker.order_calls == 0 + assert port.calls == 1 + + +def test_auto_reconcile_repeat_is_idempotent(): + repo, warn, slot, broker, port, decisions, _alarm = _run() + audit_count = len(repo.transition_log_for(slot.slot_id)) + + second = reconcile_core_holdings( + slot_repo=repo, + slot_service=SlotService(repo), + broker=broker, # type: ignore[arg-type] + instance_key="default", + warn_store=warn, + portfolio_items=[PortfolioItem("pf-019685", "019685", 0, 726.0)], + cost_basis_port=port, + auto_reconcile_store=InMemoryAutoReconcileStore(repo, warn), + ) + + assert second[str(slot.slot_id)] == "ok" + assert len(repo.transition_log_for(slot.slot_id)) == audit_count + assert decisions[str(slot.slot_id)] == "auto_synced" + + +@pytest.mark.parametrize( + "evidence", + [ + _evidence(data_completeness_warning="history incomplete"), + _evidence(symbol="WRONG"), + _evidence(current_quantity=725.0, total_cost_basis=725.0 * 2172.46829187), + _evidence(observed_at=datetime.now(tz=UTC) - timedelta(minutes=2)), + _evidence(cost_basis_per_unit=float("nan"), total_cost_basis=float("nan")), + ], +) +def test_auto_reconcile_invalid_evidence_fails_closed(evidence): + repo, warn, slot, broker, port, decisions, _alarm = _run(evidence=evidence) + + current = repo.fetch_slot(slot.slot_id) + assert current is not None + assert decisions[str(slot.slot_id)] == "qty_warning" + assert current.qty == 1452.0 + assert current.avg_price == 2100.0 + assert warn.get_warning_info(str(slot.slot_id))["case_codes"] == ["B", "C"] + assert repo.transition_log_for(slot.slot_id)[-1].actor == "test" + assert broker.balance_reads == 1 + assert broker.order_calls == 0 + assert port.calls == 1 + + +def test_auto_reconcile_pending_dispatch_metadata_fails_closed_without_evidence_read(): + repo, warn, slot, _broker, port, decisions, _alarm = _run( + mutate_slot=lambda value: replace(value, broker_order_id="pending-order") + ) + + assert decisions[str(slot.slot_id)] == "qty_warning" + assert repo.fetch_slot(slot.slot_id).qty == 1452.0 + assert warn.get_warning_info(str(slot.slot_id))["case_codes"] == ["B", "C"] + assert port.calls == 0 + + +def test_auto_reconcile_duplicate_hub_symbol_fails_closed(): + repo, warn, slot, _broker, port, decisions, _alarm = _run( + portfolio_items=[ + PortfolioItem("pf-019685", "019685", 0, 726.0), + PortfolioItem("pf-other", "019685", 0, 0.0), + ] + ) + + current = repo.fetch_slot(slot.slot_id) + assert current is not None + assert decisions[str(slot.slot_id)] == "qty_warning" + assert current.qty == 1452.0 + warning_info = warn.get_warning_info(str(slot.slot_id)) + assert warning_info is not None + assert warning_info["case_codes"] == ["B", "C"] + assert port.calls == 0 + + +def test_auto_reconcile_cas_race_preserves_intervening_state_and_records(): + class RacingStore: + def __init__(self, delegate, repo): + self.delegate = delegate + self.repo = repo + + def apply_holding_mirror(self, command): + current = self.repo.fetch_slot(command.slot_id) + self.repo.update_slot(replace(current, avg_price=9999.0)) + return self.delegate.apply_holding_mirror(command) + + repo, warn, slot, _broker, _port, decisions, _alarm = _run( + auto_store_wrapper=lambda delegate, repo: RacingStore(delegate, repo) + ) + + current = repo.fetch_slot(slot.slot_id) + assert current is not None + assert decisions[str(slot.slot_id)] == "qty_warning" + assert current.qty == 1452.0 + assert current.avg_price == 9999.0 + assert warn.get_warning_info(str(slot.slot_id))["case_codes"] == ["B", "C"] + assert repo.transition_log_for(slot.slot_id)[-1].actor == "test" + + +def test_in_memory_auto_reconcile_audit_failure_rolls_back_all_records(monkeypatch): + repo = InMemorySlotRepository() + service = SlotService(repo) + warning = InMemoryReconcileWarningStore() + slot = _holding(repo, service) + assert slot.portfolio_id is not None + assert slot.symbol is not None + assert slot.qty is not None + warning.set_warned( + str(slot.slot_id), portfolio_id=slot.portfolio_id, case_code="B", actor="reconcile" + ) + warning.set_warned( + str(slot.slot_id), portfolio_id=slot.portfolio_id, case_code="C", actor="reconcile" + ) + before_slot = repo.fetch_slot(slot.slot_id) + before_warning = warning.get_warning_info(str(slot.slot_id)) + before_log = repo.transition_log_for(slot.slot_id) + + def fail_audit(_record): + raise RuntimeError("audit insert failed") + + monkeypatch.setattr(repo, "append_transition_log", fail_audit) + with pytest.raises(RuntimeError, match="audit insert failed"): + InMemoryAutoReconcileStore(repo, warning).apply_holding_mirror( + AutoReconcileCommand( + slot_id=slot.slot_id, + instance_key=slot.instance_key, + portfolio_id=slot.portfolio_id, + symbol=slot.symbol, + expected_qty=slot.qty, + expected_avg_price=slot.avg_price, + target_qty=726.0, + target_avg_price=2172.46829187, + hub_quantity=726.0, + broker_quantity=726, + evidence_observed_at=datetime.now(tz=UTC), + ) + ) + + assert repo.fetch_slot(slot.slot_id) == before_slot + assert warning.get_warning_info(str(slot.slot_id)) == before_warning + assert repo.transition_log_for(slot.slot_id) == before_log + + +def test_auto_reconcile_emits_detected_then_one_authoritative_resolved_event(): + _repo, _warn, slot, _broker, _port, decisions, alarm = _run(notification=True) + + assert decisions[str(slot.slot_id)] == "auto_synced" + assert [code for code, _fields in alarm.calls] == [ + "reconcile_mismatch_detected", + "reconcile_mismatch_resolved", + ] + assert all("account" not in str(fields).lower() for _code, fields in alarm.calls) diff --git a/tests/application/test_snapshot_scheduler.py b/tests/application/test_snapshot_scheduler.py index 3411d6f..504df5d 100644 --- a/tests/application/test_snapshot_scheduler.py +++ b/tests/application/test_snapshot_scheduler.py @@ -60,11 +60,15 @@ def test_run_once_uses_authoritative_core_holdings_broker(monkeypatch): projected_broker = object() authoritative_broker = object() unmanaged_store = object() + auto_store = SimpleNamespace(apply_holding_mirror=lambda _command: False) captured = {} def _capture(**kwargs): captured["broker"] = kwargs["broker"] captured["unmanaged_position_store"] = kwargs.get("unmanaged_position_store") + captured["portfolio_items"] = kwargs.get("portfolio_items") + captured["cost_basis_port"] = kwargs.get("cost_basis_port") + captured["auto_reconcile_store"] = kwargs.get("auto_reconcile_store") return {} monkeypatch.setattr( @@ -79,12 +83,17 @@ def test_run_once_uses_authoritative_core_holdings_broker(monkeypatch): slot_repo=object(), slot_service=SimpleNamespace(instance_key="inst-1"), unmanaged_position_store=unmanaged_store, + cost_basis_port=read, + auto_reconcile_store=auto_store, # type: ignore[arg-type] ) sched._reconcile_slots = lambda _items: None assert sched.run_once() is True assert captured["broker"] is authoritative_broker assert captured["unmanaged_position_store"] is unmanaged_store + assert captured["portfolio_items"] == read.items + assert captured["cost_basis_port"] is read + assert captured["auto_reconcile_store"] is auto_store def test_swap_preserves_warning_for_carried_over_portfolio(): diff --git a/tests/conftest.py b/tests/conftest.py index a3a4413..2d28dc5 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -24,6 +24,7 @@ _PG_TABLES = ( "ledger_outbox", "intent_log", "slot_transition_log", + "slot_reconcile_warning", "daily_symbol_entries", "daily_slot", "reconcile_notification_state", diff --git a/tests/runtime/test_daemon_resolve_symbol.py b/tests/runtime/test_daemon_resolve_symbol.py index 4faba2e..3d5883d 100644 --- a/tests/runtime/test_daemon_resolve_symbol.py +++ b/tests/runtime/test_daemon_resolve_symbol.py @@ -83,7 +83,12 @@ def test_daemon_wires_reconciliation_broker_into_snapshot_scheduler(): / "daemon.py" ).read_text(encoding="utf-8") tree = ast.parse(src) - found = False + expected = { + "core_holdings_broker": "reconciliation_broker", + "cost_basis_port": "portfolio_read", + "auto_reconcile_store": "auto_reconcile_store", + } + found: dict[str, str] = {} for node in ast.walk(tree): if not isinstance(node, ast.Assign): continue @@ -100,17 +105,12 @@ def test_daemon_wires_reconciliation_broker_into_snapshot_scheduler(): ): continue for kw in call.keywords: - if kw.arg != "core_holdings_broker": + if kw.arg not in expected: continue if ( isinstance(kw.value, ast.Attribute) - and kw.value.attr == "reconciliation_broker" and isinstance(kw.value.value, ast.Name) and kw.value.value.id == "self" ): - found = True - break - assert found, ( - "Daemon must pass self.reconciliation_broker into " - "SnapshotScheduler(core_holdings_broker=...)" - ) + found[kw.arg] = kw.value.attr + assert found == expected diff --git a/trex_core/adapters/persistence/auto_reconcile.py b/trex_core/adapters/persistence/auto_reconcile.py new file mode 100644 index 0000000..8377f19 --- /dev/null +++ b/trex_core/adapters/persistence/auto_reconcile.py @@ -0,0 +1,175 @@ +"""Atomic persistence adapters for holding mirror auto-reconciliation.""" + +from __future__ import annotations + +import copy +import json +from dataclasses import replace +from datetime import UTC, datetime +from typing import Any + +from trex_core.adapters.persistence.postgresql import _require_psycopg +from trex_core.application.auto_reconcile import AutoReconcileCommand +from trex_core.application.portfolio import InMemoryReconcileWarningStore +from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository +from trex_core.domain.slot.models import Slot, SlotTransitionRecord +from trex_core.domain.slot.state import SlotState, slot_advisory_lock_id + +_REASON = "hub_broker_matched_core_mirror_stale" +_ACTOR = "auto_reconcile" + + +def _eligible(current: Slot | None, command: AutoReconcileCommand) -> bool: + return bool( + current is not None + and current.slot_id == command.slot_id + and current.instance_key == command.instance_key + and current.portfolio_id == command.portfolio_id + and current.symbol == command.symbol + and current.status is SlotState.HOLDING + and current.retired_at is None + and current.qty == command.expected_qty + and current.avg_price == command.expected_avg_price + and current.limit_price is None + and current.broker_order_id is None + and current.cancel_state is None + and current.cancel_requested_at is None + and current.order_submitted_at is None + and current.order_date is None + and current.cancel_attempts == 0 + ) + + +def _audit_extra(command: AutoReconcileCommand) -> dict[str, object]: + return { + "portfolio_id": command.portfolio_id, + "symbol": command.symbol, + "before_qty": command.expected_qty, + "after_qty": command.target_qty, + "before_avg_price": command.expected_avg_price, + "after_avg_price": command.target_avg_price, + "hub_quantity": command.hub_quantity, + "broker_quantity": command.broker_quantity, + } + + +class InMemoryAutoReconcileStore: + """In-process atomic-equivalent adapter used by tests and local runtimes.""" + + def __init__( + self, + slot_repo: InMemorySlotRepository, + warning_store: InMemoryReconcileWarningStore, + ) -> None: + self._slot_repo = slot_repo + self._warning_store = warning_store + + def apply_holding_mirror(self, command: AutoReconcileCommand) -> bool: + lock_id = slot_advisory_lock_id(command.instance_key) + with self._slot_repo.transactional_advisory_lock(lock_id): + with self._warning_store._lock: # same critical section as warning CRUD + current = self._slot_repo.fetch_slot(command.slot_id) + warnings = self._warning_store._warned.get(str(command.slot_id), {}) + if current is None or not _eligible(current, command) or "B" not in warnings: + return False + + previous_slot = current + previous_transitions = list(self._slot_repo._transitions) + previous_warnings = copy.deepcopy(self._warning_store._warned) + now = datetime.now(tz=UTC) + try: + self._slot_repo.update_slot( + replace( + current, + qty=command.target_qty, + avg_price=command.target_avg_price, + updated_at=now, + ) + ) + self._slot_repo.append_transition_log( + SlotTransitionRecord( + slot_id=command.slot_id, + from_state=SlotState.HOLDING, + to_state=SlotState.HOLDING, + ts=now, + reason=_REASON, + actor=_ACTOR, + extra=_audit_extra(command), + ) + ) + warnings.pop("B", None) + if not warnings: + self._warning_store._warned.pop(str(command.slot_id), None) + return True + except Exception: + self._slot_repo._slots[command.slot_id] = previous_slot + self._slot_repo._transitions[:] = previous_transitions + self._warning_store._warned = previous_warnings + raise + + +class PostgreSQLAutoReconcileStore: + """One-transaction CAS update, B-only clear, and append-only audit.""" + + def __init__(self, dsn: str, *, psycopg_module: Any = None) -> None: + self._psycopg = _require_psycopg(psycopg_module) + self._dsn = dsn + + def apply_holding_mirror(self, command: AutoReconcileCommand) -> bool: + now = datetime.now(tz=UTC) + with self._psycopg.connect(self._dsn, autocommit=False) as conn: + try: + with conn.cursor() as cur: + cur.execute( + "SELECT pg_advisory_xact_lock(%s)", + (slot_advisory_lock_id(command.instance_key),), + ) + cur.execute( + "UPDATE daily_slot SET qty=%s, avg_price=%s, updated_at=%s " + "WHERE slot_id=%s AND instance_key=%s AND portfolio_id=%s " + "AND symbol=%s AND status='holding' AND retired_at IS NULL " + "AND qty IS NOT DISTINCT FROM %s " + "AND avg_price IS NOT DISTINCT FROM %s " + "AND limit_price IS NULL AND broker_order_id IS NULL " + "AND cancel_state IS NULL AND cancel_requested_at IS NULL " + "AND order_submitted_at IS NULL AND order_date IS NULL " + "AND cancel_attempts=0", + ( + command.target_qty, + command.target_avg_price, + now, + command.slot_id, + command.instance_key, + command.portfolio_id, + command.symbol, + command.expected_qty, + command.expected_avg_price, + ), + ) + if cur.rowcount != 1: + conn.rollback() + return False + cur.execute( + "DELETE FROM slot_reconcile_warning WHERE slot_id=%s AND case_code=%s", + (str(command.slot_id), "B"), + ) + if cur.rowcount != 1: + conn.rollback() + return False + cur.execute( + "INSERT INTO slot_transition_log " + "(slot_id, from_state, to_state, ts, reason, actor, extra) " + "VALUES (%s, 'holding', 'holding', %s, %s, %s, %s)", + ( + command.slot_id, + now, + _REASON, + _ACTOR, + json.dumps(_audit_extra(command)), + ), + ) + conn.commit() + return True + except Exception: + conn.rollback() + raise diff --git a/trex_core/adapters/portfolio/hub_read.py b/trex_core/adapters/portfolio/hub_read.py index 3828ef9..12363f8 100644 --- a/trex_core/adapters/portfolio/hub_read.py +++ b/trex_core/adapters/portfolio/hub_read.py @@ -16,10 +16,12 @@ signal (REQ-028 AC-2) without callers seeing raw httpx exceptions. from __future__ import annotations import math +from datetime import UTC, datetime from typing import Any, Optional import httpx +from trex_core.application.auto_reconcile import HoldingsCostBasisEvidence from trex_core.application.portfolio import ( BrokerCredentialPayload, PortfolioHubUnavailableError, @@ -268,6 +270,56 @@ class PortfolioHubReadAdapter: return None, "invalid" return value, "ok" + def get_cost_basis_evidence(self, portfolio_id: str) -> HoldingsCostBasisEvidence: + """Return only canonical, complete evidence used by auto-reconcile.""" + url = f"{self._base_url}/v1/portfolios/{portfolio_id}/holdings/cost-basis" + try: + resp = self._get(url) + except httpx.HTTPError as exc: + raise PortfolioHubUnavailableError( + f"cost_basis network error: {exc.__class__.__name__}" + ) from exc + if not (200 <= resp.status_code < 300): + raise PortfolioHubUnavailableError( + f"cost_basis http_status_{resp.status_code}" + ) + try: + body = resp.json() + except ValueError as exc: + raise PortfolioHubUnavailableError("cost_basis invalid_json") from exc + if not isinstance(body, dict): + raise PortfolioHubUnavailableError("cost_basis invalid_payload") + + response_pid = body.get("portfolio_id") + symbol = body.get("symbol") + if str(response_pid or "") != portfolio_id or not isinstance(symbol, str) or not symbol: + raise PortfolioHubUnavailableError("cost_basis identity_mismatch") + + def _finite_number(name: str, *, positive: bool) -> float: + raw = body.get(name) + if isinstance(raw, bool): + raise PortfolioHubUnavailableError(f"cost_basis invalid_{name}") + try: + value = float(str(raw)) + except (TypeError, ValueError) as exc: + raise PortfolioHubUnavailableError(f"cost_basis invalid_{name}") from exc + if not math.isfinite(value) or (positive and value <= 0) or (not positive and value < 0): + raise PortfolioHubUnavailableError(f"cost_basis invalid_{name}") + return value + + warning = body.get("data_completeness_warning") + if warning is not None and not isinstance(warning, str): + raise PortfolioHubUnavailableError("cost_basis invalid_completeness_warning") + return HoldingsCostBasisEvidence( + portfolio_id=portfolio_id, + symbol=symbol, + current_quantity=_finite_number("current_quantity", positive=True), + cost_basis_per_unit=_finite_number("cost_basis_per_unit", positive=True), + total_cost_basis=_finite_number("total_cost_basis", positive=True), + observed_at=datetime.now(tz=UTC), + data_completeness_warning=warning or None, + ) + def get_cost_basis_with_diagnostics( self, portfolio_id: str, diff --git a/trex_core/application/auto_reconcile.py b/trex_core/application/auto_reconcile.py new file mode 100644 index 0000000..2cb761c --- /dev/null +++ b/trex_core/application/auto_reconcile.py @@ -0,0 +1,46 @@ +"""Fail-closed holding mirror auto-reconciliation contracts (issue #94).""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from typing import Protocol +from uuid import UUID + + +@dataclass(frozen=True) +class HoldingsCostBasisEvidence: + """Canonical portfolio-hub evidence captured by one successful HTTP read.""" + + portfolio_id: str + symbol: str + current_quantity: float + cost_basis_per_unit: float + total_cost_basis: float + observed_at: datetime + data_completeness_warning: str | None = None + + +@dataclass(frozen=True) +class AutoReconcileCommand: + """CAS-bound command for one atomic mirror repair.""" + + slot_id: UUID + instance_key: str + portfolio_id: str + symbol: str + expected_qty: float + expected_avg_price: float | None + target_qty: float + target_avg_price: float + hub_quantity: float + broker_quantity: int + evidence_observed_at: datetime + + +class AutoReconcileStore(Protocol): + """Atomic qty/avg/audit/B-warning persistence boundary.""" + + def apply_holding_mirror(self, command: AutoReconcileCommand) -> bool: + """Apply exactly once when every CAS/eligibility condition still holds.""" + ... diff --git a/trex_core/application/ports.py b/trex_core/application/ports.py index 4fa84d8..b71bf4a 100644 --- a/trex_core/application/ports.py +++ b/trex_core/application/ports.py @@ -8,6 +8,8 @@ from __future__ import annotations from typing import Optional, Protocol from uuid import UUID +from trex_core.application.auto_reconcile import HoldingsCostBasisEvidence + class IntentStore(Protocol): """Durable idempotency + audit store for accepted intents (dcy-002 OQ-1). @@ -64,3 +66,10 @@ class PortfolioHubCostBasisPort(Protocol): or any error occurs. The caller must handle None as a graceful-degrade signal. """ ... + + def get_cost_basis_evidence(self, portfolio_id: str) -> HoldingsCostBasisEvidence: + """Return strict canonical evidence for automatic holding reconciliation. + + Implementations must not fall back to legacy cost-basis fields. + """ + ... diff --git a/trex_core/application/reconciliation.py b/trex_core/application/reconciliation.py index e4edf8c..d599748 100644 --- a/trex_core/application/reconciliation.py +++ b/trex_core/application/reconciliation.py @@ -29,12 +29,18 @@ import hashlib import logging import math import re -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from typing import cast from trex_core.application.alarm_emitter import AlarmEmitter +from trex_core.application.auto_reconcile import ( + AutoReconcileCommand, + AutoReconcileStore, + HoldingsCostBasisEvidence, +) from trex_core.application.broker_port import BrokerPort from trex_core.application.portfolio import ( + PortfolioItem, ReconcileWarningStore, SnapshotStore, UnmanagedPositionStore, @@ -377,6 +383,113 @@ def _broker_balances_are_authoritative_for_unmanaged_resolution(broker: BrokerPo return False +_AUTO_RECONCILE_EVIDENCE_MAX_AGE = timedelta(seconds=60) +_AUTO_RECONCILE_FUTURE_TOLERANCE = timedelta(seconds=5) + + +def _try_auto_reconcile_holding( + *, + slot, + broker_quantity: int, + portfolio_items: list[PortfolioItem] | None, + cost_basis_port, + auto_reconcile_store: AutoReconcileStore | None, + observed_at: datetime, +) -> bool: + """Return True only after the atomic mirror repair commits.""" + if portfolio_items is None or cost_basis_port is None or auto_reconcile_store is None: + return False + if ( + slot.status is not SlotState.HOLDING + or slot.retired_at is not None + or not slot.portfolio_id + or not slot.symbol + or slot.qty is None + or broker_quantity <= 0 + or slot.limit_price is not None + or slot.broker_order_id is not None + or slot.cancel_state is not None + or slot.cancel_requested_at is not None + or slot.order_submitted_at is not None + or slot.order_date is not None + or slot.cancel_attempts != 0 + ): + return False + + matches = [item for item in portfolio_items if item.portfolio_id == slot.portfolio_id] + symbol_matches = [item for item in portfolio_items if item.symbol == slot.symbol] + if len(matches) != 1 or len(symbol_matches) != 1: + return False + item = matches[0] + hub_quantity = item.asset_quantity + if ( + item.symbol != slot.symbol + or not math.isfinite(hub_quantity) + or hub_quantity <= 0 + or not float(hub_quantity).is_integer() + or int(hub_quantity) != broker_quantity + ): + return False + + getter = getattr(cost_basis_port, "get_cost_basis_evidence", None) + if not callable(getter): + return False + try: + evidence = cast(HoldingsCostBasisEvidence, getter(slot.portfolio_id)) + except Exception: # noqa: BLE001 - fail closed without provider details + _log.warning( + "auto_reconcile skipped: slot_id=%s reason=cost_basis_evidence_unavailable", + slot.slot_id, + ) + return False + + age = observed_at - evidence.observed_at + expected_total = evidence.current_quantity * evidence.cost_basis_per_unit + if ( + evidence.observed_at.tzinfo is None + or age > _AUTO_RECONCILE_EVIDENCE_MAX_AGE + or age < -_AUTO_RECONCILE_FUTURE_TOLERANCE + or evidence.data_completeness_warning is not None + or evidence.portfolio_id != slot.portfolio_id + or evidence.symbol != slot.symbol + or not math.isfinite(evidence.current_quantity) + or not math.isfinite(evidence.cost_basis_per_unit) + or not math.isfinite(evidence.total_cost_basis) + or evidence.current_quantity != hub_quantity + or evidence.cost_basis_per_unit <= 0 + or evidence.total_cost_basis <= 0 + or not math.isclose( + evidence.total_cost_basis, + expected_total, + rel_tol=1e-9, + abs_tol=1e-6, + ) + ): + return False + + command = AutoReconcileCommand( + slot_id=slot.slot_id, + instance_key=slot.instance_key, + portfolio_id=slot.portfolio_id, + symbol=slot.symbol, + expected_qty=float(slot.qty), + expected_avg_price=(float(slot.avg_price) if slot.avg_price is not None else None), + target_qty=float(broker_quantity), + target_avg_price=float(evidence.cost_basis_per_unit), + hub_quantity=float(hub_quantity), + broker_quantity=broker_quantity, + evidence_observed_at=evidence.observed_at, + ) + try: + return bool(auto_reconcile_store.apply_holding_mirror(command)) + except Exception: # noqa: BLE001 - persistence failure keeps warning active + _log.warning( + "auto_reconcile failed: slot_id=%s reason=atomic_persistence_failure", + slot.slot_id, + ) + return False + + def reconcile_core_holdings( *, slot_repo, @@ -387,6 +500,9 @@ def reconcile_core_holdings( warn_store: ReconcileWarningStore | None = None, notification_store=None, unmanaged_position_store: UnmanagedPositionStore | None = None, + portfolio_items: list[PortfolioItem] | None = None, + cost_basis_port=None, + auto_reconcile_store: AutoReconcileStore | None = None, ) -> dict[str, str]: """dcy-020 / REQ-062 (BAT-029) + REQ-103 (BAT-050): surface core-slot holding vs broker divergence and set durable reconcile-warning for B case. @@ -451,6 +567,7 @@ def reconcile_core_holdings( slot_id_str, slot.portfolio_id, sym, bq, slot.qty, ) # REQ-103 / BAT-050: durable reconcile-warning (case_code='B'). + warning_persisted = False if warn_store is not None: try: warn_store.set_warned( @@ -460,6 +577,7 @@ def reconcile_core_holdings( actor="reconcile", reason=f"slot_qty={slot.qty} broker_qty={bq} symbol={sym}", ) + warning_persisted = True except Exception: # noqa: BLE001 - never let reconcile crash the loop _log.exception( "failed to set reconcile_warning for slot %s", slot.slot_id @@ -482,6 +600,38 @@ def reconcile_core_holdings( _log.exception( "reconcile_core_holdings: notify_mismatch failed for slot=%s", slot_id_str, ) + if warning_persisted and _try_auto_reconcile_holding( + slot=slot, + broker_quantity=bq, + portfolio_items=portfolio_items, + cost_basis_port=cost_basis_port, + auto_reconcile_store=auto_reconcile_store, + observed_at=snapshot_observed_at, + ): + decisions[slot_id_str] = "auto_synced" + active_b_identity_keys.discard(slot_id_str) + _log.info( + "auto_reconcile committed: slot_id=%s portfolio_id=%s symbol=%s", + slot_id_str, + slot.portfolio_id, + sym, + ) + if alarm_emitter is not None and notification_store is not None: + try: + _notify_resolved( + alarm_emitter, + notification_store, + identity_key=slot_id_str, + case_code="B", + instance_key=instance_key, + symbol=sym, + slot_id=slot_id_str, + ) + except Exception: # noqa: BLE001 - business commit is authoritative + _log.exception( + "reconcile_core_holdings: auto resolved notify failed for slot=%s", + slot_id_str, + ) else: decisions[str(slot.slot_id)] = "ok" for sym, bq in broker_qty.items(): # unmatched broker position diff --git a/trex_core/application/snapshot_scheduler.py b/trex_core/application/snapshot_scheduler.py index 547f70a..1cce410 100644 --- a/trex_core/application/snapshot_scheduler.py +++ b/trex_core/application/snapshot_scheduler.py @@ -16,6 +16,7 @@ from collections.abc import Callable from datetime import UTC, date, datetime, time, timedelta, timezone from trex_core.application.alarm_emitter import AlarmEmitter +from trex_core.application.auto_reconcile import AutoReconcileStore from trex_core.application.balance_health import BalanceHealthTracker from trex_core.application.broker_port import BrokerBalanceError, BrokerPort from trex_core.application.portfolio import ( @@ -55,6 +56,8 @@ class SnapshotScheduler: warn_store: ReconcileWarningStore | None = None, notification_store: ReconcileNotificationStore | None = None, unmanaged_position_store: UnmanagedPositionStore | None = None, + cost_basis_port=None, + auto_reconcile_store: AutoReconcileStore | None = None, credential_fetch_enabled: bool = False, kis_reinit_callback: Callable[[BrokerCredentialPayload], None] | None = None, credential_refresh: Callable[[], object] | None = None, @@ -81,6 +84,8 @@ class SnapshotScheduler: # req-0007 / ac-0024: durable notification deduplication store (Optional). self._notification_store = notification_store self._unmanaged_position_store = unmanaged_position_store + self._cost_basis_port = cost_basis_port + self._auto_reconcile_store = auto_reconcile_store # dcy-036 / REQ-113/114/115: credential fetch opt-in + break-glass detection. # credential_fetch_enabled=False (default) → env-only, no portfolio-hub fetch. self._credential_fetch_enabled: bool = credential_fetch_enabled @@ -332,7 +337,7 @@ class SnapshotScheduler: if self._slot_service is not None: try: # BAT-050 / REQ-103: pass warn_store for B-case durable warning. - self._run_core_holding_reconciliation_worker() + self._run_core_holding_reconciliation_worker(items) except Exception: # noqa: BLE001 _log.exception("core-holding divergence reconcile failed") self._last_fired_date = self._clock() @@ -353,7 +358,7 @@ class SnapshotScheduler: ), ) - def _run_core_holding_reconciliation_worker(self) -> None: + def _run_core_holding_reconciliation_worker(self, items=None) -> None: from trex_core.application.reconciliation import reconcile_core_holdings self._run_balance_worker( @@ -368,6 +373,13 @@ class SnapshotScheduler: warn_store=self._warn_store, # BAT-050 / REQ-103 notification_store=self._notification_store, # req-0007 / ac-0024 unmanaged_position_store=self._unmanaged_position_store, + portfolio_items=( + list(items) + if items is not None + else [asset.item for asset in self._store.list_assets()] + ), + cost_basis_port=self._cost_basis_port, + auto_reconcile_store=self._auto_reconcile_store, ), ) diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index 0d3f677..a6f01fe 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -34,6 +34,7 @@ from trex_core.adapters.messaging import ( from trex_core.adapters.persistence.account_snapshot import ( PostgreSQLAccountSnapshotStore, ) +from trex_core.adapters.persistence.auto_reconcile import PostgreSQLAutoReconcileStore from trex_core.adapters.persistence.ledger_skip_store import PostgreSQLLedgerSkipStore from trex_core.adapters.persistence.migrate import apply_migrations from trex_core.adapters.persistence.postgresql import ( @@ -202,6 +203,7 @@ class Daemon: self.pause_store = PostgreSQLPauseStore(config.dsn) # BAT-050 / REQ-102: durable reconcile-warning store ((slot_id, case_code) PK). self.warn_store = PostgreSQLReconcileWarningStore(config.dsn) + self.auto_reconcile_store = PostgreSQLAutoReconcileStore(config.dsn) # req-0007 / ac-0024: durable reconcile notification deduplication store. self.notification_store = PostgreSQLReconcileNotificationStore(config.dsn) self.unmanaged_position_store = PostgreSQLUnmanagedPositionStore(config.dsn) @@ -286,6 +288,8 @@ class Daemon: warn_store=self.warn_store, # BAT-050 / REQ-103: durable reconcile-warning notification_store=self.notification_store, # req-0007 / ac-0024 unmanaged_position_store=self.unmanaged_position_store, + cost_basis_port=self.portfolio_read, + auto_reconcile_store=self.auto_reconcile_store, credential_fetch_enabled=config.credential_fetch_enabled, # dcy-036 / REQ-113 credential_refresh=self.broker_provider.refresh_day_start_credential, # dcy-036 / REQ-113 AC-2 / REQ-115 AC-1: KISBroker reinit callback.