2 Commits
Author SHA1 Message Date
Hermes Agent bbdacb121c chore(release): prepare v0.8.0 2026-09-15 16:44:53 +09:00
Hermes Agent c70ac0fb24 feat: auto-reconcile stale holding mirrors
Closes #94
2026-09-14 12:55:04 +09:00
22 changed files with 1187 additions and 22 deletions
+2 -1
View File
@@ -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.<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
@@ -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)을 적용하지 않는다.
+2 -2
View File
@@ -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 변화가 없어야 한다.
+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,
"next_line_number": 10,
"next_ac_number": 32
"next_ac_number": 35
}
+4 -4
View File
@@ -3,8 +3,8 @@
운영팀은 release tag의 clean checkout에서 아래 명령으로 이미지를 빌드한다.
```bash
git checkout v0.7.0
./scripts/build_release_image.sh trex-core:0.7.0
git checkout v0.8.0
./scripts/build_release_image.sh trex-core:0.8.0
```
스크립트는 `pyproject.toml`의 canonical version과 image tag가 일치하는지 확인하고 현재
@@ -16,8 +16,8 @@ tag, label 불일치, non-root user 불일치 또는 container `--version` smoke
Registry 경로를 사용할 때도 version tag를 유지한다.
```bash
./scripts/build_release_image.sh registry.example/trex-core:0.7.0
docker push registry.example/trex-core:0.7.0
./scripts/build_release_image.sh registry.example/trex-core:0.8.0
docker push registry.example/trex-core:0.8.0
```
배포 전 `docker image inspect`에서 나온 revision과 image digest를
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "trex-core"
version = "0.7.0"
version = "0.8.0"
description = "Shared core for the trex product family; first domain: slot lifecycle (storage-agnostic pure domain)"
requires-python = ">=3.12"
dependencies = [
@@ -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)
# ---------------------------------------------------------------------------
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가 있으면 우선 사용."""
+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()
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():
+1
View File
@@ -24,6 +24,7 @@ _PG_TABLES = (
"ledger_outbox",
"intent_log",
"slot_transition_log",
"slot_reconcile_warning",
"daily_symbol_entries",
"daily_slot",
"reconcile_notification_state",
+9 -9
View File
@@ -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
@@ -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
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,
+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 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.
"""
...
+151 -1
View File
@@ -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
+14 -2
View File
@@ -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,
),
)
+4
View File
@@ -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.
Generated
+1 -1
View File
@@ -289,7 +289,7 @@ wheels = [
[[package]]
name = "trex-core"
version = "0.7.0"
version = "0.8.0"
source = { editable = "." }
dependencies = [
{ name = "httpx" },