feat: auto-reconcile stale holding mirrors

Closes #94
This commit is contained in:
Hermes Agent
2026-09-14 12:55:04 +09:00
parent adf131f19c
commit c70ac0fb24
19 changed files with 1181 additions and 16 deletions
+2 -1
View File
@@ -7,7 +7,7 @@ status: active
## Criterion ## Criterion
`case_code=B`(core slot↔KIS 수량 불일치) 또는 `case_code=C`(portfolio-hub↔KIS 수량 불일치) reconcile-warning이 활성인 `holding` slot은 `trex.slots.req.<instance>` 응답에서 제외해야 한다. 동일 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.<instance>` 응답에서 제외해야 한다. 동일 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 ## Verification
@@ -23,5 +23,6 @@ status: active
- 해제 대상: HOLDING 뿐 아니라 모든 상태의 slot에 걸린 C-case 경고를 포함한다. - 해제 대상: HOLDING 뿐 아니라 모든 상태의 slot에 걸린 C-case 경고를 포함한다.
- 해제 실패: 해제 도중 예외가 발생하면 `result='clear_failed'`를 반환하고 완전 성공으로 보고하지 않아야 한다(부분 성공 = 실패). - 해제 실패: 해제 도중 예외가 발생하면 `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가 없어야 한다. - 비침습 조건: slots RPC 호출 전후로 slot 상태 전이, broker mutation, ledger write side effect가 없어야 한다.
- AC-0022 우선순위: B 또는 C reconcile warning으로 제외된 slot에는 broker-authoritative `qty` 응답 형식(AC-0022)을 적용하지 않는다. - AC-0022 우선순위: B 또는 C reconcile warning으로 제외된 slot에는 broker-authoritative `qty` 응답 형식(AC-0022)을 적용하지 않는다.
+2 -2
View File
@@ -7,11 +7,11 @@ status: active
## Criterion ## 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 ## 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로 남겨야 한다. - 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이다. - 경계 조건: 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 변화가 없어야 한다. - 보안·비변경 조건: 모든 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 변화가 없어야 한다.
+18
View File
@@ -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 회귀 테스트는 그대로 통과한다.
+18
View File
@@ -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는 발생하지 않는다.
+17
View File
@@ -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가 없다.
+1 -1
View File
@@ -1,5 +1,5 @@
{ {
"schema_version": 1, "schema_version": 1,
"next_line_number": 10, "next_line_number": 10,
"next_ac_number": 32 "next_ac_number": 35
} }
@@ -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()")
+52
View File
@@ -117,6 +117,58 @@ def test_get_portfolios_network_error_raises_unavailable():
# REQ-123: cost-basis parsing (canonical + diagnostics) # 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(): def test_get_cost_basis_with_diagnostics_prefers_canonical_when_valid():
"""REQ-123 AC-1: cost_basis_per_unit가 있으면 우선 사용.""" """REQ-123 AC-1: cost_basis_per_unit가 있으면 우선 사용."""
+326
View File
@@ -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)
@@ -60,11 +60,15 @@ def test_run_once_uses_authoritative_core_holdings_broker(monkeypatch):
projected_broker = object() projected_broker = object()
authoritative_broker = object() authoritative_broker = object()
unmanaged_store = object() unmanaged_store = object()
auto_store = SimpleNamespace(apply_holding_mirror=lambda _command: False)
captured = {} captured = {}
def _capture(**kwargs): def _capture(**kwargs):
captured["broker"] = kwargs["broker"] captured["broker"] = kwargs["broker"]
captured["unmanaged_position_store"] = kwargs.get("unmanaged_position_store") 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 {} return {}
monkeypatch.setattr( monkeypatch.setattr(
@@ -79,12 +83,17 @@ def test_run_once_uses_authoritative_core_holdings_broker(monkeypatch):
slot_repo=object(), slot_repo=object(),
slot_service=SimpleNamespace(instance_key="inst-1"), slot_service=SimpleNamespace(instance_key="inst-1"),
unmanaged_position_store=unmanaged_store, unmanaged_position_store=unmanaged_store,
cost_basis_port=read,
auto_reconcile_store=auto_store, # type: ignore[arg-type]
) )
sched._reconcile_slots = lambda _items: None sched._reconcile_slots = lambda _items: None
assert sched.run_once() is True assert sched.run_once() is True
assert captured["broker"] is authoritative_broker assert captured["broker"] is authoritative_broker
assert captured["unmanaged_position_store"] is unmanaged_store 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(): def test_swap_preserves_warning_for_carried_over_portfolio():
+1
View File
@@ -24,6 +24,7 @@ _PG_TABLES = (
"ledger_outbox", "ledger_outbox",
"intent_log", "intent_log",
"slot_transition_log", "slot_transition_log",
"slot_reconcile_warning",
"daily_symbol_entries", "daily_symbol_entries",
"daily_slot", "daily_slot",
"reconcile_notification_state", "reconcile_notification_state",
+9 -9
View File
@@ -83,7 +83,12 @@ def test_daemon_wires_reconciliation_broker_into_snapshot_scheduler():
/ "daemon.py" / "daemon.py"
).read_text(encoding="utf-8") ).read_text(encoding="utf-8")
tree = ast.parse(src) 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): for node in ast.walk(tree):
if not isinstance(node, ast.Assign): if not isinstance(node, ast.Assign):
continue continue
@@ -100,17 +105,12 @@ def test_daemon_wires_reconciliation_broker_into_snapshot_scheduler():
): ):
continue continue
for kw in call.keywords: for kw in call.keywords:
if kw.arg != "core_holdings_broker": if kw.arg not in expected:
continue continue
if ( if (
isinstance(kw.value, ast.Attribute) isinstance(kw.value, ast.Attribute)
and kw.value.attr == "reconciliation_broker"
and isinstance(kw.value.value, ast.Name) and isinstance(kw.value.value, ast.Name)
and kw.value.value.id == "self" and kw.value.value.id == "self"
): ):
found = True found[kw.arg] = kw.value.attr
break assert found == expected
assert found, (
"Daemon must pass self.reconciliation_broker into "
"SnapshotScheduler(core_holdings_broker=...)"
)
@@ -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
+52
View File
@@ -16,10 +16,12 @@ signal (REQ-028 AC-2) without callers seeing raw httpx exceptions.
from __future__ import annotations from __future__ import annotations
import math import math
from datetime import UTC, datetime
from typing import Any, Optional from typing import Any, Optional
import httpx import httpx
from trex_core.application.auto_reconcile import HoldingsCostBasisEvidence
from trex_core.application.portfolio import ( from trex_core.application.portfolio import (
BrokerCredentialPayload, BrokerCredentialPayload,
PortfolioHubUnavailableError, PortfolioHubUnavailableError,
@@ -268,6 +270,56 @@ class PortfolioHubReadAdapter:
return None, "invalid" return None, "invalid"
return value, "ok" 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( def get_cost_basis_with_diagnostics(
self, self,
portfolio_id: str, portfolio_id: str,
+46
View File
@@ -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."""
...
+9
View File
@@ -8,6 +8,8 @@ from __future__ import annotations
from typing import Optional, Protocol from typing import Optional, Protocol
from uuid import UUID from uuid import UUID
from trex_core.application.auto_reconcile import HoldingsCostBasisEvidence
class IntentStore(Protocol): class IntentStore(Protocol):
"""Durable idempotency + audit store for accepted intents (dcy-002 OQ-1). """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. 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.
"""
...
+151 -1
View File
@@ -29,12 +29,18 @@ import hashlib
import logging import logging
import math import math
import re import re
from datetime import UTC, datetime from datetime import UTC, datetime, timedelta
from typing import cast from typing import cast
from trex_core.application.alarm_emitter import AlarmEmitter 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.broker_port import BrokerPort
from trex_core.application.portfolio import ( from trex_core.application.portfolio import (
PortfolioItem,
ReconcileWarningStore, ReconcileWarningStore,
SnapshotStore, SnapshotStore,
UnmanagedPositionStore, UnmanagedPositionStore,
@@ -377,6 +383,113 @@ def _broker_balances_are_authoritative_for_unmanaged_resolution(broker: BrokerPo
return False 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( def reconcile_core_holdings(
*, *,
slot_repo, slot_repo,
@@ -387,6 +500,9 @@ def reconcile_core_holdings(
warn_store: ReconcileWarningStore | None = None, warn_store: ReconcileWarningStore | None = None,
notification_store=None, notification_store=None,
unmanaged_position_store: UnmanagedPositionStore | None = 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]: ) -> dict[str, str]:
"""dcy-020 / REQ-062 (BAT-029) + REQ-103 (BAT-050): surface core-slot holding """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. 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, slot_id_str, slot.portfolio_id, sym, bq, slot.qty,
) )
# REQ-103 / BAT-050: durable reconcile-warning (case_code='B'). # REQ-103 / BAT-050: durable reconcile-warning (case_code='B').
warning_persisted = False
if warn_store is not None: if warn_store is not None:
try: try:
warn_store.set_warned( warn_store.set_warned(
@@ -460,6 +577,7 @@ def reconcile_core_holdings(
actor="reconcile", actor="reconcile",
reason=f"slot_qty={slot.qty} broker_qty={bq} symbol={sym}", reason=f"slot_qty={slot.qty} broker_qty={bq} symbol={sym}",
) )
warning_persisted = True
except Exception: # noqa: BLE001 - never let reconcile crash the loop except Exception: # noqa: BLE001 - never let reconcile crash the loop
_log.exception( _log.exception(
"failed to set reconcile_warning for slot %s", slot.slot_id "failed to set reconcile_warning for slot %s", slot.slot_id
@@ -482,6 +600,38 @@ def reconcile_core_holdings(
_log.exception( _log.exception(
"reconcile_core_holdings: notify_mismatch failed for slot=%s", slot_id_str, "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: else:
decisions[str(slot.slot_id)] = "ok" decisions[str(slot.slot_id)] = "ok"
for sym, bq in broker_qty.items(): # unmatched broker position for sym, bq in broker_qty.items(): # unmatched broker position
+14 -2
View File
@@ -16,6 +16,7 @@ from collections.abc import Callable
from datetime import UTC, date, datetime, time, timedelta, timezone from datetime import UTC, date, datetime, time, timedelta, timezone
from trex_core.application.alarm_emitter import AlarmEmitter 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.balance_health import BalanceHealthTracker
from trex_core.application.broker_port import BrokerBalanceError, BrokerPort from trex_core.application.broker_port import BrokerBalanceError, BrokerPort
from trex_core.application.portfolio import ( from trex_core.application.portfolio import (
@@ -55,6 +56,8 @@ class SnapshotScheduler:
warn_store: ReconcileWarningStore | None = None, warn_store: ReconcileWarningStore | None = None,
notification_store: ReconcileNotificationStore | None = None, notification_store: ReconcileNotificationStore | None = None,
unmanaged_position_store: UnmanagedPositionStore | None = None, unmanaged_position_store: UnmanagedPositionStore | None = None,
cost_basis_port=None,
auto_reconcile_store: AutoReconcileStore | None = None,
credential_fetch_enabled: bool = False, credential_fetch_enabled: bool = False,
kis_reinit_callback: Callable[[BrokerCredentialPayload], None] | None = None, kis_reinit_callback: Callable[[BrokerCredentialPayload], None] | None = None,
credential_refresh: Callable[[], object] | None = None, credential_refresh: Callable[[], object] | None = None,
@@ -81,6 +84,8 @@ class SnapshotScheduler:
# req-0007 / ac-0024: durable notification deduplication store (Optional). # req-0007 / ac-0024: durable notification deduplication store (Optional).
self._notification_store = notification_store self._notification_store = notification_store
self._unmanaged_position_store = unmanaged_position_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. # dcy-036 / REQ-113/114/115: credential fetch opt-in + break-glass detection.
# credential_fetch_enabled=False (default) → env-only, no portfolio-hub fetch. # credential_fetch_enabled=False (default) → env-only, no portfolio-hub fetch.
self._credential_fetch_enabled: bool = credential_fetch_enabled self._credential_fetch_enabled: bool = credential_fetch_enabled
@@ -332,7 +337,7 @@ class SnapshotScheduler:
if self._slot_service is not None: if self._slot_service is not None:
try: try:
# BAT-050 / REQ-103: pass warn_store for B-case durable warning. # 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 except Exception: # noqa: BLE001
_log.exception("core-holding divergence reconcile failed") _log.exception("core-holding divergence reconcile failed")
self._last_fired_date = self._clock() 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 from trex_core.application.reconciliation import reconcile_core_holdings
self._run_balance_worker( self._run_balance_worker(
@@ -368,6 +373,13 @@ class SnapshotScheduler:
warn_store=self._warn_store, # BAT-050 / REQ-103 warn_store=self._warn_store, # BAT-050 / REQ-103
notification_store=self._notification_store, # req-0007 / ac-0024 notification_store=self._notification_store, # req-0007 / ac-0024
unmanaged_position_store=self._unmanaged_position_store, 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,
), ),
) )
+4
View File
@@ -34,6 +34,7 @@ from trex_core.adapters.messaging import (
from trex_core.adapters.persistence.account_snapshot import ( from trex_core.adapters.persistence.account_snapshot import (
PostgreSQLAccountSnapshotStore, 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.ledger_skip_store import PostgreSQLLedgerSkipStore
from trex_core.adapters.persistence.migrate import apply_migrations from trex_core.adapters.persistence.migrate import apply_migrations
from trex_core.adapters.persistence.postgresql import ( from trex_core.adapters.persistence.postgresql import (
@@ -202,6 +203,7 @@ class Daemon:
self.pause_store = PostgreSQLPauseStore(config.dsn) self.pause_store = PostgreSQLPauseStore(config.dsn)
# BAT-050 / REQ-102: durable reconcile-warning store ((slot_id, case_code) PK). # BAT-050 / REQ-102: durable reconcile-warning store ((slot_id, case_code) PK).
self.warn_store = PostgreSQLReconcileWarningStore(config.dsn) self.warn_store = PostgreSQLReconcileWarningStore(config.dsn)
self.auto_reconcile_store = PostgreSQLAutoReconcileStore(config.dsn)
# req-0007 / ac-0024: durable reconcile notification deduplication store. # req-0007 / ac-0024: durable reconcile notification deduplication store.
self.notification_store = PostgreSQLReconcileNotificationStore(config.dsn) self.notification_store = PostgreSQLReconcileNotificationStore(config.dsn)
self.unmanaged_position_store = PostgreSQLUnmanagedPositionStore(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 warn_store=self.warn_store, # BAT-050 / REQ-103: durable reconcile-warning
notification_store=self.notification_store, # req-0007 / ac-0024 notification_store=self.notification_store, # req-0007 / ac-0024
unmanaged_position_store=self.unmanaged_position_store, 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_fetch_enabled=config.credential_fetch_enabled, # dcy-036 / REQ-113
credential_refresh=self.broker_provider.refresh_day_start_credential, credential_refresh=self.broker_provider.refresh_day_start_credential,
# dcy-036 / REQ-113 AC-2 / REQ-115 AC-1: KISBroker reinit callback. # dcy-036 / REQ-113 AC-2 / REQ-115 AC-1: KISBroker reinit callback.