Merge pull request #91 from fourblocks2020/copilot/feat-reconcile-unmanaged-position-durable-lifecycl
feat(reconcile): durably track unmanaged position lifecycle by instance and symbol
This commit is contained in:
@@ -0,0 +1,22 @@
|
||||
---
|
||||
id: "ac-0030"
|
||||
status: active
|
||||
---
|
||||
|
||||
# Unmanaged position durable lifecycle, authority fencing, and operator audit
|
||||
|
||||
## Criterion
|
||||
|
||||
`trex-core`는 broker 계정에 존재하지만 대상 instance의 managed slot/ledger가 소유하지 않는 종목을 `(instance_key, symbol)` 단위의 durable unmanaged-position lifecycle로 추적해야 한다. Lifecycle state는 instance 간에 섞이지 않아야 하며, 최소 `detected`, `quantity_changed`, `resolved`, `reopened` 전이와 `first_seen_at`, `last_seen_at`, `quantity_changed_at`, `resolved_at` 보존을 제공해야 한다. 같은 수량의 반복 authoritative 관측은 `last_seen_at`만 갱신하고 lifecycle audit/alarm을 다시 만들지 않아야 한다.
|
||||
|
||||
absence resolution은 explicit authoritative full-balance snapshot에서만 허용되며 projected/partial/non-authoritative broker view 또는 broker read failure는 resolve authority가 아니다. observe와 resolve는 reconcile pass 전체에 전달된 동일 snapshot timestamp 또는 monotonic generation으로 fenced 되어야 하며 stale observe/stale resolve는 더 최신 state나 audit를 reopen, overwrite 또는 resolved 처리하지 못해야 한다. 최종 state와 append-only audit는 항상 가장 최신 authoritative snapshot이 결정해야 한다.
|
||||
|
||||
운영자 disposition은 explicit evidence reference가 있는 명령으로만 기록되고 inspectable append-only audit trail에 남아야 한다. 이 기능은 broker order submit/cancel, slot 생성·상태 변경, synthetic ledger mutation 또는 warning 자동 삭제를 수행해서는 안 된다.
|
||||
|
||||
## Verification
|
||||
|
||||
- identity/lifecycle: 같은 symbol을 서로 다른 `instance_key`에서 독립 state로 저장하고 first detect, quantity change, resolved, resolved 뒤 reopened, first/last-seen 보존을 InMemory/PostgreSQL parity로 검증한다.
|
||||
- authority boundary: missing marker, `False`, ownership/projected broker, partial reader, broker read failure에서는 absence resolution이 발생하지 않고 explicit authoritative full-balance snapshot에서만 resolved가 기록됨을 검증한다.
|
||||
- temporal fencing: 더 오래된 absent snapshot이 더 최신 observe 뒤늦게 commit 되어도 resolved 되지 않고, 더 오래된 observe가 더 최신 resolved/observe를 reopen 또는 overwrite하지 않으며 final state/audit가 최신 authoritative snapshot 순서를 따른다는 PostgreSQL concurrency/interleaving 회귀를 검증한다.
|
||||
- suppression/audit: 같은 수량 반복 관측은 duplicate lifecycle alarm/audit를 만들지 않고 `last_seen_at`만 갱신한다. operator `record`는 `evidence_ref`를 요구하고 `unmanaged audit`로 append-only history를 조회할 수 있어야 한다.
|
||||
- safety boundary: unmanaged lifecycle path가 broker order, slot mutation, ledger write 또는 warning auto-clear를 수행하지 않음을 existing regression surface와 focused tests로 유지한다.
|
||||
@@ -1,5 +1,5 @@
|
||||
{
|
||||
"schema_version": 1,
|
||||
"next_line_number": 10,
|
||||
"next_ac_number": 30
|
||||
"next_ac_number": 31
|
||||
}
|
||||
|
||||
@@ -447,8 +447,11 @@ resume_kill_switch 요청·응답 (REQ-046):
|
||||
|
||||
- `get_balance()`는 검증된 `list[BalanceItem]`만 반환한다. KIS의 token/auth/authorization/transport/response/rejection 실패는 빈 목록으로 축소되지 않고 `BrokerBalanceError`를 raise한다. 안정 code allow-list는 `token_issuance_failed`, `authentication_failed`, `authorization_failed`, `transport_failed`, `malformed_response`, `broker_rejected`, `ownership_mismatch`다.
|
||||
- `sync`와 `verify` CLI는 첫 balance read를 완료한 뒤 cache한 결과를 해당 invocation에 전달한다. 이 read 또는 broker credential resolution이 실패하면 exit code `1`이며, `--json` stderr는 `{ "error": {"code": "…", "message": "broker balance unavailable"} }` 형태의 안전한 값만 낸다. upstream payload, credential, account 값은 계약상 출력하지 않는다.
|
||||
- `reconcile_core_holdings()`의 broker-only 결과는 `unmanaged_position`(대상 instance의 managed slot 없음) 또는 `managed_non_holding_divergence`(동일 instance의 non-holding managed slot 존재)다. `find_managed_slot_by_symbol(instance_key, symbol)` 조회는 instance-scoped이며, 이 두 분류 자체는 slot/warning write나 자동 주문·청산·slot 생성을 하지 않는다.
|
||||
- 이 변경은 AMQP 또는 portfolio-hub external wire contract와 DB schema를 바꾸지 않는다.
|
||||
- `reconcile_core_holdings()`의 broker-only 결과는 `unmanaged_position`(대상 instance의 managed slot 없음) 또는 `managed_non_holding_divergence`(동일 instance의 non-holding managed slot 존재)다. `find_managed_slot_by_symbol(instance_key, symbol)` 조회는 instance-scoped이며, 두 분류 모두 slot 생성, 자동 주문·청산을 하지 않는다.
|
||||
- `managed_non_holding_divergence`는 기존 reconcile warning lifecycle을 유지한다. 반면 `unmanaged_position`은 `(instance_key, symbol)` keyed durable state로 추적되며 `quantity`, `first_seen_at`, `last_seen_at`, `quantity_changed_at`, `resolved_at`, lifecycle state와 latest operator evidence/disposition을 저장한다.
|
||||
- `unmanaged_position`의 absence resolution은 explicit authoritative full-balance snapshot reader에서만 허용된다. Ownership/projected broker 또는 marker가 없는 balance reader는 detect/update 입력일 수 있어도 resolved authority는 아니다.
|
||||
- Operator surface는 `trex-core unmanaged list`, `trex-core unmanaged audit <symbol>`, `trex-core unmanaged record <symbol> <action> --evidence-ref ...`이며 audit actor는 CLI의 trusted local identity에서 결정된다. 임의 caller-supplied actor string은 current contract가 아니다.
|
||||
- 이 변경은 AMQP 또는 portfolio-hub external wire contract는 바꾸지 않지만, internal PostgreSQL persistence에는 unmanaged-position state/audit schema를 추가한다.
|
||||
|
||||
### Account snapshot recovery (dcy-041 / GitHub #45)
|
||||
|
||||
|
||||
@@ -66,6 +66,7 @@ pkill -TERM -f trex-core-daemon
|
||||
- `sync`와 `verify`는 broker balance를 먼저 한 번 읽어 해당 실행 안에서 재사용한다. balance/credential read 실패는 안전한 error code와 함께 exit `1`로 종료하며, 빈 broker balance를 추정하지 않는다. 성공한 빈 목록만 flat balance로 취급한다.
|
||||
- source evidence에는 mode(`fetched`/`env`/`break_glass`)와 비밀 없는 label만 기록한다. fetched mode는 KIS + `credential_fetch_enabled=true`에서만 사용하고, break-glass는 명시적 운영자 환경 설정으로 fetch를 건너뛴다. Paper와 env KIS 경로는 fetch하지 않는다.
|
||||
- reconciliation에서 `unmanaged_position` 또는 `managed_non_holding_divergence`가 보이면 자동 보정·자동 주문·자동 청산이 없다. broker, ledger, slot evidence를 교차 확인하고 warning/slot 상태를 보존한 채 triage한다.
|
||||
- `unmanaged_position` triage는 `trex-core unmanaged list`로 현재 lifecycle state를, `trex-core unmanaged audit <symbol>`로 lifecycle/operator audit trail을 확인한 뒤 `trex-core unmanaged record <symbol> <action> --evidence-ref ...`로만 disposition을 남긴다.
|
||||
- remediation은 **dry-run → broker/ledger/slot 교차 확인 → 사장님 명시 승인 → 단일 대상 `sync --live --yes` → 재-reconcile** 순서의 별도 release gate다. 이 문서는 실제 target 선택, live 실행 또는 승인을 부여하지 않는다.
|
||||
|
||||
### `report` — 일일 운영 리포트 (REQ-088 / dcy-029, REQ-110/111 / dcy-035)
|
||||
|
||||
@@ -99,13 +99,14 @@
|
||||
1. 신규 자산(portfolio_id 에 슬롯 없음) → `inactive` 영구 슬롯 생성(멱등 — `ON CONFLICT (instance_key, portfolio_id)` partial unique).
|
||||
2. universe 이탈 + inactive → soft-retire(`retired_at` set; 행 보존, 운영 뷰 `universe_detached`). 이탈 + holding/pending → tombstone(그대로 유지 — safety-sell 경로 보존).
|
||||
3. 가드: universe 빈 응답이면 reconcile skip(degrade 가드); 직전 대비 ≥50% 축소면 retire 보류(`allow_retire=False`, 대량 소실 가드 — dcy-020 OQ-2 D3).
|
||||
4. 이어서 `reconcile_core_holdings`(REQ-062/121): 브로커 잔고 ground truth 로 core holding 검증 — core holding/브로커 0 → slot `error` 격리+WARN, 수량 불일치 → WARN(상태 불변), 브로커 보유인데 대상 instance에 관리 slot이 없으면 `unmanaged_position`, 동일 instance에 `inactive`/`error` 등 non-holding 관리 slot이 있으면 `managed_non_holding_divergence`를 WARN으로 남긴다. 두 broker-only 분류는 상태·warning persistence·주문·청산·slot 생성의 자동 수정을 하지 않는다.
|
||||
4. 이어서 `reconcile_core_holdings`(REQ-062/121): 브로커 잔고 ground truth 로 core holding 검증 — core holding/브로커 0 → slot `error` 격리+WARN, 수량 불일치 → WARN(상태 불변), 브로커 보유인데 대상 instance에 관리 slot이 없으면 `unmanaged_position`, 동일 instance에 `inactive`/`error` 등 non-holding 관리 slot이 있으면 `managed_non_holding_divergence`를 WARN으로 남긴다. `managed_non_holding_divergence`는 기존 warning 경로를 유지하고, `unmanaged_position`은 `(instance_key, symbol)` durable lifecycle state/audit에 기록된다. 같은 수량 반복 관측은 `last_seen_at`만 갱신하고 새 detect/change alarm을 만들지 않는다. 어떤 broker-only 분류도 주문·청산·slot 생성·warning 자동 삭제를 하지 않는다.
|
||||
|
||||
### Broker provider and operator balance-read barrier (BAT-057 / REQ-119/120)
|
||||
|
||||
1. daemon과 `sync`/`verify` CLI는 같은 broker provider를 통해 브로커를 얻는다. `broker=kis` 및 `credential_fetch_enabled=true`일 때만 portfolio-hub의 유효한 인증정보를 fetched source로 사용한다. daemon의 day-start refresh는 이미 같은 KST 날짜에 얻은 source가 있으면 이를 재사용한다.
|
||||
2. 명시적 `TREX_KIS_BROKER_BREAK_GLASS`는 fetched source를 건너뛰고 env KIS 설정을 사용하는 별도 source다. Paper와 `credential_fetch_enabled=false` KIS 경로는 env/config source이며 credential fetch를 하지 않는다.
|
||||
3. `sync`/`verify`는 실행당 잔고를 한 번 먼저 검증해 cache한 뒤 그 값만 사용한다. KIS 잔고 읽기 실패는 빈 잔고로 해석하거나 후속 DB/slot write로 진행하지 않고 command boundary에서 안전하게 실패한다. 성공한 빈 잔고(`[]`)만 검증된 flat balance다.
|
||||
4. Ownership projection이 켜진 mixed-account KIS 경로에서는 주문/전략 조회용 broker가 strategy-owned quantity만 보이도록 balance를 투영할 수 있다. 이때 unmanaged reconciliation은 provider가 별도로 넘기는 authoritative full-balance reader를 사용한다. Missing authority marker, projected/partial reader, self-report failure 또는 balance read failure는 absence resolution authority가 아니며 active unmanaged row를 `resolved`로 바꾸지 않는다.
|
||||
|
||||
## Flow: `포지션 정합 검증` (BAT-008 / REQ-031)
|
||||
|
||||
|
||||
@@ -30,6 +30,8 @@ _EXPECTED_TABLES = {
|
||||
"ledger_skip", # GitHub #68 skipped-ledger durable audit
|
||||
"reconcile_notification_state", # req-0007 / ac-0024
|
||||
"slot_strategy_state", # req-0009 / ac-0027
|
||||
"unmanaged_position_state", # GitHub #85 unmanaged lifecycle state
|
||||
"unmanaged_position_audit", # GitHub #85 unmanaged lifecycle audit
|
||||
}
|
||||
|
||||
|
||||
@@ -90,6 +92,14 @@ def test_apply_migrations_is_idempotent(pg_dsn):
|
||||
apply_migrations(pg_dsn)
|
||||
|
||||
|
||||
def test_migration_sequence_includes_canonical_unmanaged_position_number():
|
||||
names = [path.name for path in migration_files()]
|
||||
assert "030_create_unmanaged_position_state.sql" in names
|
||||
assert "028_create_unmanaged_position_state.sql" not in names
|
||||
prefixes = [name.split("_", 1)[0] for name in names]
|
||||
assert len(prefixes) == len(set(prefixes))
|
||||
|
||||
|
||||
def test_migration_replay_converts_existing_auto_venue_row_to_sor(pg_clean):
|
||||
repo = PostgreSQLSlotRepository(pg_clean)
|
||||
repo.reconcile_universe("migration-auto", {"pf-auto": "005930"})
|
||||
@@ -109,12 +119,15 @@ def test_migration_replay_converts_existing_auto_venue_row_to_sor(pg_clean):
|
||||
apply_migrations(pg_clean)
|
||||
|
||||
assert repo.fetch_slot(slot.slot_id).venue == "SOR"
|
||||
with psycopg.connect(pg_clean) as conn, conn.cursor() as cur:
|
||||
with pytest.raises(psycopg.errors.CheckViolation):
|
||||
cur.execute(
|
||||
"UPDATE daily_slot SET venue = 'AUTO' WHERE slot_id = %s",
|
||||
(slot.slot_id,),
|
||||
)
|
||||
with (
|
||||
psycopg.connect(pg_clean) as conn,
|
||||
conn.cursor() as cur,
|
||||
pytest.raises(psycopg.errors.CheckViolation),
|
||||
):
|
||||
cur.execute(
|
||||
"UPDATE daily_slot SET venue = 'AUTO' WHERE slot_id = %s",
|
||||
(slot.slot_id,),
|
||||
)
|
||||
|
||||
|
||||
def test_migration_024_separates_legacy_sor_and_is_replay_safe(pg_clean):
|
||||
@@ -169,12 +182,15 @@ def test_migration_024_separates_legacy_sor_and_is_replay_safe(pg_clean):
|
||||
assert legacy_row.trade.venue is None
|
||||
assert legacy_row.trade.order_venue == "SOR"
|
||||
|
||||
with psycopg.connect(pg_clean) as conn, conn.cursor() as cur:
|
||||
with pytest.raises(psycopg.errors.CheckViolation):
|
||||
cur.execute(
|
||||
"UPDATE ledger_outbox SET venue='SOR' WHERE correlation_id=%s",
|
||||
(normal.correlation_id,),
|
||||
)
|
||||
with (
|
||||
psycopg.connect(pg_clean) as conn,
|
||||
conn.cursor() as cur,
|
||||
pytest.raises(psycopg.errors.CheckViolation),
|
||||
):
|
||||
cur.execute(
|
||||
"UPDATE ledger_outbox SET venue='SOR' WHERE correlation_id=%s",
|
||||
(normal.correlation_id,),
|
||||
)
|
||||
|
||||
|
||||
def test_migration_029_fail_closes_unowned_legacy_outbox_rows(pg_clean):
|
||||
|
||||
@@ -0,0 +1,268 @@
|
||||
"""PostgreSQLUnmanagedPositionStore integration tests."""
|
||||
from __future__ import annotations
|
||||
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def pg_unmanaged_store(pg_clean):
|
||||
from trex_core.adapters.portfolio.unmanaged_position_store import (
|
||||
PostgreSQLUnmanagedPositionStore,
|
||||
)
|
||||
|
||||
return PostgreSQLUnmanagedPositionStore(pg_clean)
|
||||
|
||||
|
||||
class TestMigration030:
|
||||
def test_tables_exist_after_migration(self, pg_clean):
|
||||
import psycopg
|
||||
|
||||
with psycopg.connect(pg_clean) as conn, conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"SELECT table_name FROM information_schema.tables "
|
||||
"WHERE table_schema = 'public' "
|
||||
"AND table_name IN ('unmanaged_position_state', 'unmanaged_position_audit')"
|
||||
)
|
||||
assert {row[0] for row in cur.fetchall()} == {
|
||||
"unmanaged_position_state",
|
||||
"unmanaged_position_audit",
|
||||
}
|
||||
|
||||
|
||||
class TestCRUDRoundTrip:
|
||||
def test_observe_change_resolve_and_audit(self, pg_unmanaged_store):
|
||||
t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC)
|
||||
t2 = t1 + timedelta(minutes=1)
|
||||
t3 = t2 + timedelta(minutes=1)
|
||||
|
||||
first = pg_unmanaged_store.observe(
|
||||
instance_key="inst-A",
|
||||
symbol="008110",
|
||||
quantity=3,
|
||||
observed_at=t1,
|
||||
)
|
||||
repeat = pg_unmanaged_store.observe(
|
||||
instance_key="inst-A",
|
||||
symbol="008110",
|
||||
quantity=3,
|
||||
observed_at=t2,
|
||||
)
|
||||
changed = pg_unmanaged_store.observe(
|
||||
instance_key="inst-A",
|
||||
symbol="008110",
|
||||
quantity=5,
|
||||
observed_at=t3,
|
||||
)
|
||||
recorded = pg_unmanaged_store.record_operator_action(
|
||||
instance_key="inst-A",
|
||||
symbol="008110",
|
||||
action="approved_exception",
|
||||
evidence_ref="kis:2026-09-11",
|
||||
actor="cli",
|
||||
note="broker-confirmed",
|
||||
)
|
||||
resolved = pg_unmanaged_store.resolve_absent(
|
||||
instance_key="inst-A",
|
||||
observed_symbols=set(),
|
||||
observed_at=t3 + timedelta(minutes=1),
|
||||
)
|
||||
|
||||
assert first["transition"] == "detected"
|
||||
assert repeat["transition"] == "unchanged"
|
||||
assert changed["transition"] == "quantity_changed"
|
||||
assert recorded is not None
|
||||
assert len(resolved) == 1
|
||||
state = pg_unmanaged_store.get_state(instance_key="inst-A", symbol="008110")
|
||||
assert state is not None
|
||||
assert state["quantity"] == 5
|
||||
assert state["first_seen_at"] == t1
|
||||
assert state["last_seen_at"] == t3
|
||||
assert state["quantity_changed_at"] == t3
|
||||
assert state["lifecycle_state"] == "resolved"
|
||||
assert state["evidence_ref"] == "kis:2026-09-11"
|
||||
audit = pg_unmanaged_store.list_audit(instance_key="inst-A", symbol="008110")
|
||||
assert [row["event_type"] for row in audit] == [
|
||||
"detected",
|
||||
"quantity_changed",
|
||||
"operator_action",
|
||||
"resolved",
|
||||
]
|
||||
|
||||
def test_same_symbol_isolated_by_instance(self, pg_unmanaged_store):
|
||||
pg_unmanaged_store.observe(instance_key="inst-A", symbol="008110", quantity=3)
|
||||
pg_unmanaged_store.observe(instance_key="inst-B", symbol="008110", quantity=7)
|
||||
pg_unmanaged_store.resolve_absent(instance_key="inst-A", observed_symbols=set())
|
||||
|
||||
rows_a = pg_unmanaged_store.list_states(instance_key="inst-A", lifecycle_state="all")
|
||||
rows_b = pg_unmanaged_store.list_states(instance_key="inst-B", lifecycle_state="all")
|
||||
assert len(rows_a) == 1 and rows_a[0]["lifecycle_state"] == "resolved"
|
||||
assert len(rows_b) == 1 and rows_b[0]["quantity"] == 7
|
||||
|
||||
def test_reactivated_state_preserves_original_first_seen(self, pg_unmanaged_store):
|
||||
t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC)
|
||||
t2 = t1 + timedelta(minutes=1)
|
||||
t3 = t2 + timedelta(minutes=1)
|
||||
|
||||
pg_unmanaged_store.observe(
|
||||
instance_key="inst-A",
|
||||
symbol="008110",
|
||||
quantity=3,
|
||||
observed_at=t1,
|
||||
)
|
||||
pg_unmanaged_store.record_operator_action(
|
||||
instance_key="inst-A",
|
||||
symbol="008110",
|
||||
action="investigating",
|
||||
evidence_ref="kis:2026-09-07",
|
||||
actor="cli:test",
|
||||
note="before resolve",
|
||||
)
|
||||
pg_unmanaged_store.resolve_absent(
|
||||
instance_key="inst-A",
|
||||
observed_symbols=set(),
|
||||
observed_at=t2,
|
||||
)
|
||||
reopened = pg_unmanaged_store.observe(
|
||||
instance_key="inst-A",
|
||||
symbol="008110",
|
||||
quantity=5,
|
||||
observed_at=t3,
|
||||
)
|
||||
|
||||
assert reopened["transition"] == "detected"
|
||||
assert reopened["state"]["first_seen_at"] == t1
|
||||
assert reopened["state"]["last_seen_at"] == t3
|
||||
assert reopened["state"]["operator_action"] is None
|
||||
assert reopened["state"]["evidence_ref"] is None
|
||||
|
||||
def test_concurrent_first_observe_keeps_single_detected_audit(self, pg_clean):
|
||||
from trex_core.adapters.portfolio.unmanaged_position_store import (
|
||||
PostgreSQLUnmanagedPositionStore,
|
||||
)
|
||||
|
||||
def _observe():
|
||||
return PostgreSQLUnmanagedPositionStore(pg_clean).observe(
|
||||
instance_key="inst-A",
|
||||
symbol="008110",
|
||||
quantity=3,
|
||||
)
|
||||
|
||||
with ThreadPoolExecutor(max_workers=2) as pool:
|
||||
results = list(pool.map(lambda _n: _observe(), range(2)))
|
||||
|
||||
assert sorted(result["transition"] for result in results) == ["detected", "unchanged"]
|
||||
store = PostgreSQLUnmanagedPositionStore(pg_clean)
|
||||
state = store.get_state(instance_key="inst-A", symbol="008110")
|
||||
assert state is not None
|
||||
assert state["lifecycle_state"] == "active"
|
||||
audit = store.list_audit(instance_key="inst-A", symbol="008110")
|
||||
assert [row["event_type"] for row in audit] == ["detected"]
|
||||
|
||||
def test_older_absent_snapshot_applied_after_newer_observation_is_ignored(self, pg_unmanaged_store):
|
||||
t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC)
|
||||
t2 = t1 + timedelta(minutes=1)
|
||||
t3 = t2 + timedelta(minutes=1)
|
||||
|
||||
pg_unmanaged_store.observe(
|
||||
instance_key="inst-A",
|
||||
symbol="008110",
|
||||
quantity=3,
|
||||
observed_at=t1,
|
||||
)
|
||||
observed = pg_unmanaged_store.observe(
|
||||
instance_key="inst-A",
|
||||
symbol="008110",
|
||||
quantity=5,
|
||||
observed_at=t3,
|
||||
)
|
||||
resolved = pg_unmanaged_store.resolve_absent(
|
||||
instance_key="inst-A",
|
||||
observed_symbols=set(),
|
||||
observed_at=t2,
|
||||
)
|
||||
|
||||
assert observed["transition"] == "quantity_changed"
|
||||
assert resolved == []
|
||||
state = pg_unmanaged_store.get_state(instance_key="inst-A", symbol="008110")
|
||||
assert state is not None
|
||||
assert state["lifecycle_state"] == "active"
|
||||
assert state["quantity"] == 5
|
||||
assert [row["event_type"] for row in pg_unmanaged_store.list_audit(
|
||||
instance_key="inst-A",
|
||||
symbol="008110",
|
||||
)] == ["detected", "quantity_changed"]
|
||||
|
||||
def test_older_observation_applied_after_newer_resolution_is_ignored(self, pg_unmanaged_store):
|
||||
t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC)
|
||||
t2 = t1 + timedelta(minutes=1)
|
||||
t3 = t2 + timedelta(minutes=1)
|
||||
|
||||
pg_unmanaged_store.observe(
|
||||
instance_key="inst-A",
|
||||
symbol="008110",
|
||||
quantity=3,
|
||||
observed_at=t1,
|
||||
)
|
||||
pg_unmanaged_store.resolve_absent(
|
||||
instance_key="inst-A",
|
||||
observed_symbols=set(),
|
||||
observed_at=t3,
|
||||
)
|
||||
stale = pg_unmanaged_store.observe(
|
||||
instance_key="inst-A",
|
||||
symbol="008110",
|
||||
quantity=5,
|
||||
observed_at=t2,
|
||||
)
|
||||
|
||||
assert stale["transition"] == "stale"
|
||||
state = pg_unmanaged_store.get_state(instance_key="inst-A", symbol="008110")
|
||||
assert state is not None
|
||||
assert state["lifecycle_state"] == "resolved"
|
||||
assert state["quantity"] == 3
|
||||
assert [row["event_type"] for row in pg_unmanaged_store.list_audit(
|
||||
instance_key="inst-A",
|
||||
symbol="008110",
|
||||
)] == ["detected", "resolved"]
|
||||
|
||||
def test_concurrent_observe_and_resolve_respects_newest_snapshot_timestamp(self, pg_clean):
|
||||
from trex_core.adapters.portfolio.unmanaged_position_store import (
|
||||
PostgreSQLUnmanagedPositionStore,
|
||||
)
|
||||
|
||||
store = PostgreSQLUnmanagedPositionStore(pg_clean)
|
||||
t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC)
|
||||
t2 = t1 + timedelta(minutes=1)
|
||||
t3 = t2 + timedelta(minutes=1)
|
||||
store.observe(instance_key="inst-A", symbol="008110", quantity=3, observed_at=t1)
|
||||
|
||||
with ThreadPoolExecutor(max_workers=2) as pool:
|
||||
observe_future = pool.submit(
|
||||
PostgreSQLUnmanagedPositionStore(pg_clean).observe,
|
||||
instance_key="inst-A",
|
||||
symbol="008110",
|
||||
quantity=5,
|
||||
observed_at=t3,
|
||||
)
|
||||
resolve_future = pool.submit(
|
||||
PostgreSQLUnmanagedPositionStore(pg_clean).resolve_absent,
|
||||
instance_key="inst-A",
|
||||
observed_symbols=set(),
|
||||
observed_at=t2,
|
||||
)
|
||||
observe_result = observe_future.result()
|
||||
resolve_result = resolve_future.result()
|
||||
|
||||
state = store.get_state(instance_key="inst-A", symbol="008110")
|
||||
assert state is not None
|
||||
assert observe_result["transition"] == "quantity_changed"
|
||||
assert resolve_result == []
|
||||
assert state["lifecycle_state"] == "active"
|
||||
assert state["quantity"] == 5
|
||||
assert [row["event_type"] for row in store.list_audit(
|
||||
instance_key="inst-A",
|
||||
symbol="008110",
|
||||
)] == ["detected", "quantity_changed"]
|
||||
@@ -1,7 +1,8 @@
|
||||
"""Snapshot scheduler tests (REQ-030 AC-1; REQ-114 AC-1~9; REQ-115 AC-1~6)."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone, timedelta
|
||||
from datetime import UTC, datetime, timedelta, timezone
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from trex_core.adapters.broker.paper import PaperBroker
|
||||
@@ -44,10 +45,42 @@ def test_run_once_failure_sets_degraded_signal():
|
||||
store = SnapshotStore()
|
||||
sched = SnapshotScheduler(read_port=BadRead(), store=store, broker=PaperBroker())
|
||||
assert sched.run_once() is False
|
||||
reason, msg = store.degraded_state()
|
||||
reason, _msg = store.degraded_state()
|
||||
assert reason is not None and reason.value == "portfolio_hub_unavailable"
|
||||
|
||||
|
||||
def test_run_once_uses_authoritative_core_holdings_broker(monkeypatch):
|
||||
read = StubPortfolioRead(
|
||||
daemon_id="d-1",
|
||||
items=[PortfolioItem(portfolio_id="pf-X", symbol="X", cash_balance=100, asset_quantity=0)],
|
||||
)
|
||||
store = SnapshotStore()
|
||||
projected_broker = object()
|
||||
authoritative_broker = object()
|
||||
captured = {}
|
||||
|
||||
def _capture(**kwargs):
|
||||
captured["broker"] = kwargs["broker"]
|
||||
return {}
|
||||
|
||||
monkeypatch.setattr(
|
||||
"trex_core.application.reconciliation.reconcile_core_holdings",
|
||||
_capture,
|
||||
)
|
||||
sched = SnapshotScheduler(
|
||||
read_port=read,
|
||||
store=store,
|
||||
broker=projected_broker,
|
||||
core_holdings_broker=authoritative_broker,
|
||||
slot_repo=object(),
|
||||
slot_service=SimpleNamespace(instance_key="inst-1"),
|
||||
)
|
||||
sched._reconcile_slots = lambda _items: None
|
||||
|
||||
assert sched.run_once() is True
|
||||
assert captured["broker"] is authoritative_broker
|
||||
|
||||
|
||||
def test_swap_preserves_warning_for_carried_over_portfolio():
|
||||
store = SnapshotStore()
|
||||
store.swap([PortfolioItem(portfolio_id="pf-X", symbol="X",
|
||||
@@ -99,7 +132,7 @@ def _fixed_clock(kst_hour=8, kst_minute=0):
|
||||
"""Return a clock callable fixed at a given KST time today."""
|
||||
now_kst = datetime.now(tz=_KST).replace(hour=kst_hour, minute=kst_minute,
|
||||
second=0, microsecond=0)
|
||||
return lambda: now_kst.astimezone(timezone.utc)
|
||||
return lambda: now_kst.astimezone(UTC)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -616,8 +649,8 @@ def test_kis_broker_reinitialize_preserves_base_url_and_flags():
|
||||
def test_build_kis_reinit_callback_returns_none_for_paper():
|
||||
"""REQ-113 AC-2: broker=paper이면 callback=None."""
|
||||
from trex_core.adapters.broker.paper import PaperBroker as PB
|
||||
from trex_core.runtime.daemon import _build_kis_reinit_callback
|
||||
from trex_core.runtime.config import DaemonConfig
|
||||
from trex_core.runtime.daemon import _build_kis_reinit_callback
|
||||
|
||||
# minimal config: paper broker, credential_fetch_enabled=True
|
||||
cfg = DaemonConfig(dsn="x", broker="paper", execution_interval_sec=1.0,
|
||||
@@ -629,8 +662,8 @@ def test_build_kis_reinit_callback_returns_none_for_paper():
|
||||
def test_build_kis_reinit_callback_returns_none_when_disabled():
|
||||
"""REQ-113 AC-1: credential_fetch_enabled=False이면 callback=None."""
|
||||
from trex_core.adapters.broker.paper import PaperBroker as PB
|
||||
from trex_core.runtime.daemon import _build_kis_reinit_callback
|
||||
from trex_core.runtime.config import DaemonConfig
|
||||
from trex_core.runtime.daemon import _build_kis_reinit_callback
|
||||
|
||||
cfg = DaemonConfig(dsn="x", broker="kis", execution_interval_sec=1.0,
|
||||
credential_fetch_enabled=False)
|
||||
@@ -642,8 +675,8 @@ def test_build_kis_reinit_callback_returns_callable_for_kis():
|
||||
"""REQ-113 AC-2 / REQ-115 AC-1: broker=kis + fetch enabled → callable 반환 + 실제 reinitialize 호출됨."""
|
||||
from trex_core.adapters.broker.kis import KISBroker
|
||||
from trex_core.application.portfolio import BrokerCredentialPayload
|
||||
from trex_core.runtime.daemon import _build_kis_reinit_callback
|
||||
from trex_core.runtime.config import DaemonConfig
|
||||
from trex_core.runtime.daemon import _build_kis_reinit_callback
|
||||
|
||||
cfg = DaemonConfig(dsn="x", broker="kis", execution_interval_sec=1.0,
|
||||
credential_fetch_enabled=True)
|
||||
|
||||
@@ -0,0 +1,307 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
import pytest
|
||||
|
||||
from trex_core.adapters.messaging.in_memory import InMemoryTransport
|
||||
from trex_core.application.alarm_emitter import AlarmEmitter
|
||||
from trex_core.application.broker_port import BalanceItem, BrokerBalanceError
|
||||
from trex_core.application.portfolio import InMemoryUnmanagedPositionStore
|
||||
from trex_core.application.reconciliation import reconcile_core_holdings
|
||||
from trex_core.domain.slot import SlotService
|
||||
from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository
|
||||
|
||||
INSTANCE = "inst-1"
|
||||
|
||||
|
||||
class FakeBroker:
|
||||
def __init__(self, balances):
|
||||
self._balances = balances
|
||||
|
||||
def name(self):
|
||||
return "fake"
|
||||
|
||||
def submit_order(self, req): ...
|
||||
def poll_fill(self, *args, **kwargs): ...
|
||||
def cancel_order(self, *args, **kwargs): ...
|
||||
|
||||
def get_balance(self):
|
||||
return [BalanceItem(symbol=symbol, hldg_qty=qty) for symbol, qty in self._balances.items()]
|
||||
|
||||
|
||||
class RaisingBroker(FakeBroker):
|
||||
def get_balance(self):
|
||||
raise BrokerBalanceError("transport_failed")
|
||||
|
||||
|
||||
class AuthoritativeBroker(FakeBroker):
|
||||
def exposes_full_balance_snapshot(self) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
class FilteredBroker(FakeBroker):
|
||||
def exposes_full_balance_snapshot(self) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
def _wire():
|
||||
repo = InMemorySlotRepository()
|
||||
return repo, SlotService(repo, instance_key=INSTANCE)
|
||||
|
||||
|
||||
def _make_emitter():
|
||||
transport = InMemoryTransport()
|
||||
return AlarmEmitter(transport), transport
|
||||
|
||||
|
||||
def test_in_memory_store_tracks_lifecycle_and_operator_audit():
|
||||
store = InMemoryUnmanagedPositionStore()
|
||||
t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC)
|
||||
t2 = t1 + timedelta(minutes=1)
|
||||
t3 = t2 + timedelta(minutes=1)
|
||||
|
||||
first = store.observe(instance_key=INSTANCE, symbol="008110", quantity=3, observed_at=t1)
|
||||
repeat = store.observe(instance_key=INSTANCE, symbol="008110", quantity=3, observed_at=t2)
|
||||
changed = store.observe(instance_key=INSTANCE, symbol="008110", quantity=4, observed_at=t3)
|
||||
recorded = store.record_operator_action(
|
||||
instance_key=INSTANCE,
|
||||
symbol="008110",
|
||||
action="investigating",
|
||||
evidence_ref="kis-balance:2026-09-07",
|
||||
note="confirmed at broker",
|
||||
)
|
||||
resolved = store.resolve_absent(
|
||||
instance_key=INSTANCE,
|
||||
observed_symbols=set(),
|
||||
observed_at=t3 + timedelta(minutes=1),
|
||||
)
|
||||
|
||||
assert first["transition"] == "detected"
|
||||
assert repeat["transition"] == "unchanged"
|
||||
assert changed["transition"] == "quantity_changed"
|
||||
assert recorded is not None
|
||||
assert recorded["evidence_ref"] == "kis-balance:2026-09-07"
|
||||
assert len(resolved) == 1
|
||||
state = store.get_state(instance_key=INSTANCE, symbol="008110")
|
||||
assert state is not None
|
||||
assert state["quantity"] == 4
|
||||
assert state["first_seen_at"] == t1
|
||||
assert state["last_seen_at"] == t3
|
||||
assert state["quantity_changed_at"] == t3
|
||||
assert state["lifecycle_state"] == "resolved"
|
||||
assert state["resolved_at"] == t3 + timedelta(minutes=1)
|
||||
audit = store.list_audit(instance_key=INSTANCE, symbol="008110")
|
||||
assert [row["event_type"] for row in audit] == [
|
||||
"detected",
|
||||
"quantity_changed",
|
||||
"operator_action",
|
||||
"resolved",
|
||||
]
|
||||
assert all("recorded_at" in row for row in audit)
|
||||
|
||||
|
||||
def test_in_memory_store_isolates_same_symbol_by_instance():
|
||||
store = InMemoryUnmanagedPositionStore()
|
||||
|
||||
store.observe(instance_key="inst-A", symbol="008110", quantity=3)
|
||||
store.observe(instance_key="inst-B", symbol="008110", quantity=7)
|
||||
store.resolve_absent(instance_key="inst-A", observed_symbols=set())
|
||||
|
||||
state_a = store.get_state(instance_key="inst-A", symbol="008110")
|
||||
state_b = store.get_state(instance_key="inst-B", symbol="008110")
|
||||
assert state_a is not None and state_a["lifecycle_state"] == "resolved"
|
||||
assert state_b is not None and state_b["lifecycle_state"] == "active"
|
||||
assert state_b["quantity"] == 7
|
||||
|
||||
|
||||
def test_in_memory_store_preserves_first_seen_when_resolved_symbol_reappears():
|
||||
store = InMemoryUnmanagedPositionStore()
|
||||
t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC)
|
||||
t2 = t1 + timedelta(minutes=1)
|
||||
t3 = t2 + timedelta(minutes=1)
|
||||
|
||||
store.observe(instance_key=INSTANCE, symbol="008110", quantity=3, observed_at=t1)
|
||||
store.record_operator_action(
|
||||
instance_key=INSTANCE,
|
||||
symbol="008110",
|
||||
action="investigating",
|
||||
evidence_ref="kis-balance:2026-09-07",
|
||||
note="before resolve",
|
||||
)
|
||||
store.resolve_absent(instance_key=INSTANCE, observed_symbols=set(), observed_at=t2)
|
||||
reopened = store.observe(instance_key=INSTANCE, symbol="008110", quantity=5, observed_at=t3)
|
||||
|
||||
assert reopened["transition"] == "detected"
|
||||
assert reopened["state"]["first_seen_at"] == t1
|
||||
assert reopened["state"]["last_seen_at"] == t3
|
||||
assert reopened["state"]["quantity_changed_at"] == t3
|
||||
assert reopened["state"]["operator_action"] is None
|
||||
assert reopened["state"]["evidence_ref"] is None
|
||||
|
||||
|
||||
def test_in_memory_store_ignores_stale_resolve_after_newer_observation():
|
||||
store = InMemoryUnmanagedPositionStore()
|
||||
t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC)
|
||||
t2 = t1 + timedelta(minutes=1)
|
||||
t3 = t2 + timedelta(minutes=1)
|
||||
|
||||
store.observe(instance_key=INSTANCE, symbol="008110", quantity=3, observed_at=t1)
|
||||
store.observe(instance_key=INSTANCE, symbol="008110", quantity=5, observed_at=t3)
|
||||
resolved = store.resolve_absent(
|
||||
instance_key=INSTANCE,
|
||||
observed_symbols=set(),
|
||||
observed_at=t2,
|
||||
)
|
||||
|
||||
assert resolved == []
|
||||
state = store.get_state(instance_key=INSTANCE, symbol="008110")
|
||||
assert state is not None
|
||||
assert state["lifecycle_state"] == "active"
|
||||
assert state["quantity"] == 5
|
||||
assert store.list_audit(instance_key=INSTANCE, symbol="008110")[-1]["event_type"] == "quantity_changed"
|
||||
|
||||
|
||||
def test_in_memory_store_ignores_stale_observation_after_newer_resolution():
|
||||
store = InMemoryUnmanagedPositionStore()
|
||||
t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC)
|
||||
t2 = t1 + timedelta(minutes=1)
|
||||
t3 = t2 + timedelta(minutes=1)
|
||||
|
||||
store.observe(instance_key=INSTANCE, symbol="008110", quantity=3, observed_at=t1)
|
||||
store.resolve_absent(instance_key=INSTANCE, observed_symbols=set(), observed_at=t3)
|
||||
stale = store.observe(instance_key=INSTANCE, symbol="008110", quantity=5, observed_at=t2)
|
||||
|
||||
assert stale["transition"] == "stale"
|
||||
state = store.get_state(instance_key=INSTANCE, symbol="008110")
|
||||
assert state is not None
|
||||
assert state["lifecycle_state"] == "resolved"
|
||||
assert state["quantity"] == 3
|
||||
assert [row["event_type"] for row in store.list_audit(instance_key=INSTANCE, symbol="008110")] == [
|
||||
"detected",
|
||||
"resolved",
|
||||
]
|
||||
|
||||
|
||||
def test_reconcile_core_holdings_suppresses_repeated_unmanaged_alarm_spam():
|
||||
repo, svc = _wire()
|
||||
emitter, transport = _make_emitter()
|
||||
store = InMemoryUnmanagedPositionStore()
|
||||
|
||||
for _ in range(3):
|
||||
decisions = reconcile_core_holdings(
|
||||
slot_repo=repo,
|
||||
slot_service=svc,
|
||||
broker=AuthoritativeBroker({"008110": 3}),
|
||||
instance_key=INSTANCE,
|
||||
alarm_emitter=emitter,
|
||||
unmanaged_position_store=store,
|
||||
)
|
||||
assert decisions == {"broker:008110": "unmanaged_position"}
|
||||
|
||||
state = store.get_state(instance_key=INSTANCE, symbol="008110")
|
||||
assert state is not None
|
||||
assert state["quantity"] == 3
|
||||
codes = [payload["code"] for _, payload in transport.alarm_events()]
|
||||
assert codes == ["unmanaged_position_detected"]
|
||||
audit = store.list_audit(instance_key=INSTANCE, symbol="008110")
|
||||
assert [row["event_type"] for row in audit] == ["detected"]
|
||||
|
||||
|
||||
def test_reconcile_core_holdings_tracks_change_and_resolution():
|
||||
repo, svc = _wire()
|
||||
emitter, transport = _make_emitter()
|
||||
store = InMemoryUnmanagedPositionStore()
|
||||
|
||||
reconcile_core_holdings(
|
||||
slot_repo=repo,
|
||||
slot_service=svc,
|
||||
broker=FakeBroker({"008110": 3}),
|
||||
instance_key=INSTANCE,
|
||||
alarm_emitter=emitter,
|
||||
unmanaged_position_store=store,
|
||||
)
|
||||
reconcile_core_holdings(
|
||||
slot_repo=repo,
|
||||
slot_service=svc,
|
||||
broker=AuthoritativeBroker({"008110": 5}),
|
||||
instance_key=INSTANCE,
|
||||
alarm_emitter=emitter,
|
||||
unmanaged_position_store=store,
|
||||
)
|
||||
reconcile_core_holdings(
|
||||
slot_repo=repo,
|
||||
slot_service=svc,
|
||||
broker=AuthoritativeBroker({}),
|
||||
instance_key=INSTANCE,
|
||||
alarm_emitter=emitter,
|
||||
unmanaged_position_store=store,
|
||||
)
|
||||
|
||||
state = store.get_state(instance_key=INSTANCE, symbol="008110")
|
||||
assert state is not None
|
||||
assert state["quantity"] == 5
|
||||
assert state["lifecycle_state"] == "resolved"
|
||||
codes = [payload["code"] for _, payload in transport.alarm_events()]
|
||||
assert codes == [
|
||||
"unmanaged_position_detected",
|
||||
"unmanaged_position_changed",
|
||||
"unmanaged_position_resolved",
|
||||
]
|
||||
|
||||
|
||||
def test_broker_failure_does_not_resolve_unmanaged_state():
|
||||
repo, svc = _wire()
|
||||
store = InMemoryUnmanagedPositionStore()
|
||||
store.observe(instance_key=INSTANCE, symbol="008110", quantity=3)
|
||||
|
||||
with pytest.raises(BrokerBalanceError):
|
||||
reconcile_core_holdings(
|
||||
slot_repo=repo,
|
||||
slot_service=svc,
|
||||
broker=RaisingBroker({}),
|
||||
instance_key=INSTANCE,
|
||||
unmanaged_position_store=store,
|
||||
)
|
||||
|
||||
state = store.get_state(instance_key=INSTANCE, symbol="008110")
|
||||
assert state is not None
|
||||
assert state["lifecycle_state"] == "active"
|
||||
|
||||
|
||||
def test_missing_authoritative_marker_does_not_resolve_unmanaged_state():
|
||||
repo, svc = _wire()
|
||||
store = InMemoryUnmanagedPositionStore()
|
||||
store.observe(instance_key=INSTANCE, symbol="008110", quantity=3)
|
||||
|
||||
decisions = reconcile_core_holdings(
|
||||
slot_repo=repo,
|
||||
slot_service=svc,
|
||||
broker=FakeBroker({}),
|
||||
instance_key=INSTANCE,
|
||||
unmanaged_position_store=store,
|
||||
)
|
||||
|
||||
assert decisions == {}
|
||||
state = store.get_state(instance_key=INSTANCE, symbol="008110")
|
||||
assert state is not None
|
||||
assert state["lifecycle_state"] == "active"
|
||||
|
||||
|
||||
def test_non_authoritative_broker_view_does_not_resolve_unmanaged_state():
|
||||
repo, svc = _wire()
|
||||
store = InMemoryUnmanagedPositionStore()
|
||||
store.observe(instance_key=INSTANCE, symbol="008110", quantity=3)
|
||||
|
||||
decisions = reconcile_core_holdings(
|
||||
slot_repo=repo,
|
||||
slot_service=svc,
|
||||
broker=FilteredBroker({}),
|
||||
instance_key=INSTANCE,
|
||||
unmanaged_position_store=store,
|
||||
)
|
||||
|
||||
assert decisions == {}
|
||||
state = store.get_state(instance_key=INSTANCE, symbol="008110")
|
||||
assert state is not None
|
||||
assert state["lifecycle_state"] == "active"
|
||||
+4
-3
@@ -28,6 +28,8 @@ _PG_TABLES = (
|
||||
"daily_slot",
|
||||
"reconcile_notification_state",
|
||||
"slot_strategy_state",
|
||||
"unmanaged_position_audit",
|
||||
"unmanaged_position_state",
|
||||
)
|
||||
|
||||
|
||||
@@ -94,7 +96,6 @@ def pg_clean(pg_dsn: str):
|
||||
from trex_core.adapters.persistence.migrate import apply_migrations
|
||||
|
||||
apply_migrations(pg_dsn)
|
||||
with psycopg.connect(pg_dsn, autocommit=True) as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(f"TRUNCATE {', '.join(_PG_TABLES)}")
|
||||
with psycopg.connect(pg_dsn, autocommit=True) as conn, conn.cursor() as cur:
|
||||
cur.execute(f"TRUNCATE {', '.join(_PG_TABLES)}")
|
||||
return pg_dsn
|
||||
|
||||
@@ -23,6 +23,15 @@ class _Broker:
|
||||
def reinitialize(self, credential): self.credentials.append(credential)
|
||||
|
||||
|
||||
class _ProjectedBroker(_Broker):
|
||||
def __init__(self, authoritative):
|
||||
super().__init__()
|
||||
self.authoritative = authoritative
|
||||
|
||||
def authoritative_balance_reader(self):
|
||||
return self.authoritative
|
||||
|
||||
|
||||
class _Read:
|
||||
def __init__(self, ctx=None, error=None): self.ctx, self.error, self.calls = ctx, error, 0
|
||||
def fetch_broker_credential(self):
|
||||
@@ -72,6 +81,18 @@ def test_enabled_paper_never_fetches_kis_credentials(monkeypatch):
|
||||
assert read.calls == 0
|
||||
|
||||
|
||||
def test_reconciliation_broker_uses_authoritative_balance_reader(monkeypatch):
|
||||
authoritative = _Broker()
|
||||
projected = _ProjectedBroker(authoritative)
|
||||
monkeypatch.setattr("trex_core.runtime.broker_provider._make_env_broker", lambda cfg: projected)
|
||||
provider = ConfigBrokerProvider(_cfg(broker="paper"), read_port=_Read(_ctx()))
|
||||
|
||||
broker, source = provider.get_broker_for_reconciliation()
|
||||
|
||||
assert broker is authoritative
|
||||
assert source.mode == "env"
|
||||
|
||||
|
||||
def test_break_glass_is_distinct_and_skips_fetch(monkeypatch):
|
||||
broker = _Broker()
|
||||
monkeypatch.setenv("TREX_KIS_BROKER_BREAK_GLASS", "true")
|
||||
|
||||
@@ -383,10 +383,13 @@ def test_sync_requires_yes_when_not_dry_run(capsys, monkeypatch):
|
||||
"""sync without --yes (and not dry-run) should abort."""
|
||||
def _fake_sync(args, cfg, confirm):
|
||||
# Simulate the confirmation prompt returning False (no --yes, interactive aborts)
|
||||
if not getattr(args, "dry_run", False) and not getattr(args, "yes", False):
|
||||
if not confirm("Sync?"):
|
||||
print("aborted")
|
||||
return 1
|
||||
if (
|
||||
not getattr(args, "dry_run", False)
|
||||
and not getattr(args, "yes", False)
|
||||
and not confirm("Sync?")
|
||||
):
|
||||
print("aborted")
|
||||
return 1
|
||||
return 0
|
||||
monkeypatch.setattr(cli, "_cmd_sync", _fake_sync)
|
||||
rc = cli.main(["sync", "--portfolio-id", "PF-1"], confirm=lambda _: False)
|
||||
@@ -398,7 +401,7 @@ def test_sync_live_requires_live_flag(capsys, monkeypatch):
|
||||
monkeypatch.setattr(cli, "_broker_mode", lambda cfg: ("kis", True))
|
||||
|
||||
def _fake_sync(args, cfg, confirm):
|
||||
broker, is_live = cli._broker_mode(cfg)
|
||||
_broker, is_live = cli._broker_mode(cfg)
|
||||
if is_live and not getattr(args, "live", False):
|
||||
cli._fail("live account requires --live (with --yes) to confirm a real sync", args.json)
|
||||
return 2
|
||||
@@ -415,7 +418,7 @@ def test_sync_fail_closed_skipped_error_exits_nonzero_without_writes(
|
||||
"""REQ-123: a fail-closed skipped live sync is a CLI failure."""
|
||||
class _Provider:
|
||||
def get_broker_for_read(self):
|
||||
return SimpleNamespace(get_balance=lambda: []), None
|
||||
return SimpleNamespace(get_balance=list), None
|
||||
|
||||
class _WarningStore:
|
||||
def clear_warned(self, *_args, **_kwargs):
|
||||
@@ -452,7 +455,7 @@ def test_sync_fail_closed_skipped_error_exits_nonzero_without_writes(
|
||||
}
|
||||
|
||||
psycopg = types.ModuleType("psycopg")
|
||||
setattr(psycopg, "connect", lambda *_args, **_kwargs: _Connection())
|
||||
psycopg.connect = lambda *_args, **_kwargs: _Connection()
|
||||
monkeypatch.setitem(sys.modules, "psycopg", psycopg)
|
||||
monkeypatch.setattr("trex_core.runtime.broker_provider.build_broker_provider", lambda _cfg: _Provider())
|
||||
monkeypatch.setattr("trex_core.adapters.portfolio.reconcile_warning_store.PostgreSQLReconcileWarningStore", lambda *_args: _WarningStore())
|
||||
@@ -516,6 +519,355 @@ def test_verify_no_db_no_live_flag_needed(capsys, monkeypatch):
|
||||
assert rc == 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# unmanaged parser/tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_unmanaged_parser_accepts_list_audit_and_record():
|
||||
parser = cli._build_parser()
|
||||
args = parser.parse_args(["unmanaged", "list", "--status", "all", "--symbol", "008110"])
|
||||
assert args.cmd == "unmanaged"
|
||||
assert args.unmanaged_cmd == "list"
|
||||
assert args.status == "all"
|
||||
args = parser.parse_args(["unmanaged", "audit", "008110"])
|
||||
assert args.unmanaged_cmd == "audit"
|
||||
assert args.symbol == "008110"
|
||||
args = parser.parse_args(
|
||||
[
|
||||
"unmanaged",
|
||||
"record",
|
||||
"008110",
|
||||
"investigating",
|
||||
"--evidence-ref",
|
||||
"kis-balance:2026-09-11",
|
||||
]
|
||||
)
|
||||
assert args.unmanaged_cmd == "record"
|
||||
assert args.action == "investigating"
|
||||
|
||||
|
||||
def test_unmanaged_list_outputs_durable_state(monkeypatch, capsys):
|
||||
class _Store:
|
||||
def __init__(self, *_args, **_kwargs):
|
||||
pass
|
||||
|
||||
def list_states(self, **kwargs):
|
||||
assert kwargs["instance_key"] == "inst-1"
|
||||
assert kwargs["lifecycle_state"] == "all"
|
||||
return [
|
||||
{
|
||||
"instance_key": "inst-1",
|
||||
"symbol": "008110",
|
||||
"quantity": 3,
|
||||
"lifecycle_state": "active",
|
||||
"first_seen_at": "2026-09-07T00:00:00+00:00",
|
||||
"last_seen_at": "2026-09-11T00:00:00+00:00",
|
||||
"quantity_changed_at": "2026-09-07T00:00:00+00:00",
|
||||
"resolved_at": None,
|
||||
"operator_action": "investigating",
|
||||
"evidence_ref": "kis-balance:2026-09-11",
|
||||
"operator_note": "confirmed",
|
||||
"operator_actor": "operator",
|
||||
"operator_updated_at": "2026-09-11T00:00:00+00:00",
|
||||
"created_at": "2026-09-07T00:00:00+00:00",
|
||||
"updated_at": "2026-09-11T00:00:00+00:00",
|
||||
}
|
||||
]
|
||||
|
||||
monkeypatch.setattr(
|
||||
"trex_core.adapters.portfolio.unmanaged_position_store.PostgreSQLUnmanagedPositionStore",
|
||||
_Store,
|
||||
)
|
||||
args = SimpleNamespace(
|
||||
cmd="unmanaged",
|
||||
unmanaged_cmd="list",
|
||||
status="all",
|
||||
symbol=None,
|
||||
instance="inst-1",
|
||||
config=None,
|
||||
config_path=None,
|
||||
env_file=None,
|
||||
json=True,
|
||||
timeout=10.0,
|
||||
)
|
||||
|
||||
rc = cli._cmd_unmanaged_list(args, SimpleNamespace(dsn="postgresql://x/y"))
|
||||
|
||||
assert rc == 0
|
||||
out = json.loads(capsys.readouterr().out)
|
||||
assert out["count"] == 1
|
||||
assert out["instance_key"] == "inst-1"
|
||||
assert out["states"][0]["symbol"] == "008110"
|
||||
assert out["states"][0]["evidence_ref"] == "kis-balance:2026-09-11"
|
||||
|
||||
|
||||
def test_unmanaged_record_requires_existing_state(monkeypatch, capsys):
|
||||
class _Store:
|
||||
def __init__(self, *_args, **_kwargs):
|
||||
pass
|
||||
|
||||
def record_operator_action(self, **_kwargs):
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(
|
||||
"trex_core.adapters.portfolio.unmanaged_position_store.PostgreSQLUnmanagedPositionStore",
|
||||
_Store,
|
||||
)
|
||||
args = SimpleNamespace(
|
||||
cmd="unmanaged",
|
||||
unmanaged_cmd="record",
|
||||
symbol="008110",
|
||||
action="approved_exception",
|
||||
evidence_ref="kis-balance:2026-09-11",
|
||||
note="confirmed",
|
||||
instance="inst-1",
|
||||
config=None,
|
||||
config_path=None,
|
||||
env_file=None,
|
||||
json=True,
|
||||
timeout=10.0,
|
||||
)
|
||||
|
||||
rc = cli._cmd_unmanaged_record(args, SimpleNamespace(dsn="postgresql://x/y"))
|
||||
|
||||
assert rc == 1
|
||||
err = json.loads(capsys.readouterr().err)
|
||||
assert "not found" in err["error"]
|
||||
|
||||
|
||||
def test_unmanaged_record_persists_operator_action(monkeypatch, capsys):
|
||||
monkeypatch.setattr(cli, "_operator_actor", lambda: "cli:test")
|
||||
|
||||
class _Store:
|
||||
def __init__(self, *_args, **_kwargs):
|
||||
pass
|
||||
|
||||
def record_operator_action(self, **kwargs):
|
||||
assert kwargs["instance_key"] == "inst-1"
|
||||
assert kwargs["symbol"] == "008110"
|
||||
assert kwargs["action"] == "managed_inclusion"
|
||||
return {
|
||||
"instance_key": "inst-1",
|
||||
"symbol": "008110",
|
||||
"quantity": 3,
|
||||
"lifecycle_state": "active",
|
||||
"first_seen_at": "2026-09-07T00:00:00+00:00",
|
||||
"last_seen_at": "2026-09-11T00:00:00+00:00",
|
||||
"quantity_changed_at": "2026-09-07T00:00:00+00:00",
|
||||
"resolved_at": None,
|
||||
"operator_action": "managed_inclusion",
|
||||
"evidence_ref": kwargs["evidence_ref"],
|
||||
"operator_note": kwargs["note"],
|
||||
"operator_actor": kwargs["actor"],
|
||||
"operator_updated_at": "2026-09-11T00:00:00+00:00",
|
||||
"created_at": "2026-09-07T00:00:00+00:00",
|
||||
"updated_at": "2026-09-11T00:00:00+00:00",
|
||||
}
|
||||
|
||||
monkeypatch.setattr(
|
||||
"trex_core.adapters.portfolio.unmanaged_position_store.PostgreSQLUnmanagedPositionStore",
|
||||
_Store,
|
||||
)
|
||||
args = SimpleNamespace(
|
||||
cmd="unmanaged",
|
||||
unmanaged_cmd="record",
|
||||
symbol="008110",
|
||||
action="managed_inclusion",
|
||||
evidence_ref="kis-balance:2026-09-11",
|
||||
note="manual inclusion planned",
|
||||
instance="inst-1",
|
||||
config=None,
|
||||
config_path=None,
|
||||
env_file=None,
|
||||
json=True,
|
||||
timeout=10.0,
|
||||
)
|
||||
|
||||
rc = cli._cmd_unmanaged_record(args, SimpleNamespace(dsn="postgresql://x/y"))
|
||||
|
||||
assert rc == 0
|
||||
out = json.loads(capsys.readouterr().out)
|
||||
assert out["result"] == "recorded"
|
||||
assert out["state"]["operator_action"] == "managed_inclusion"
|
||||
assert out["state"]["evidence_ref"] == "kis-balance:2026-09-11"
|
||||
assert out["state"]["operator_actor"] == "cli:test"
|
||||
|
||||
|
||||
def test_unmanaged_record_text_output(monkeypatch, capsys):
|
||||
monkeypatch.setattr(cli, "_operator_actor", lambda: "cli:test")
|
||||
|
||||
class _Store:
|
||||
def __init__(self, *_args, **_kwargs):
|
||||
pass
|
||||
|
||||
def record_operator_action(self, **kwargs):
|
||||
return {
|
||||
"instance_key": "inst-1",
|
||||
"symbol": kwargs["symbol"],
|
||||
"quantity": 3,
|
||||
"lifecycle_state": "active",
|
||||
"first_seen_at": "2026-09-07T00:00:00+00:00",
|
||||
"last_seen_at": "2026-09-11T00:00:00+00:00",
|
||||
"quantity_changed_at": "2026-09-07T00:00:00+00:00",
|
||||
"resolved_at": None,
|
||||
"operator_action": kwargs["action"],
|
||||
"evidence_ref": kwargs["evidence_ref"],
|
||||
"operator_note": kwargs["note"],
|
||||
"operator_actor": kwargs["actor"],
|
||||
"operator_updated_at": "2026-09-11T00:00:00+00:00",
|
||||
"created_at": "2026-09-07T00:00:00+00:00",
|
||||
"updated_at": "2026-09-11T00:00:00+00:00",
|
||||
}
|
||||
|
||||
monkeypatch.setattr(
|
||||
"trex_core.adapters.portfolio.unmanaged_position_store.PostgreSQLUnmanagedPositionStore",
|
||||
_Store,
|
||||
)
|
||||
args = SimpleNamespace(
|
||||
cmd="unmanaged",
|
||||
unmanaged_cmd="record",
|
||||
symbol="008110",
|
||||
action="approved_exception",
|
||||
evidence_ref="kis-balance:2026-09-11",
|
||||
note="confirmed",
|
||||
instance="inst-1",
|
||||
config=None,
|
||||
config_path=None,
|
||||
env_file=None,
|
||||
json=False,
|
||||
timeout=10.0,
|
||||
)
|
||||
|
||||
rc = cli._cmd_unmanaged_record(args, SimpleNamespace(dsn="postgresql://x/y"))
|
||||
|
||||
assert rc == 0
|
||||
out = capsys.readouterr().out
|
||||
assert "recorded unmanaged position instance=inst-1 symbol=008110" in out
|
||||
assert "evidence_ref=kis-balance:2026-09-11" in out
|
||||
|
||||
|
||||
def test_unmanaged_record_main_dispatch(monkeypatch, capsys):
|
||||
monkeypatch.setattr(cli, "_operator_actor", lambda: "cli:test")
|
||||
|
||||
class _Store:
|
||||
def __init__(self, *_args, **_kwargs):
|
||||
pass
|
||||
|
||||
def record_operator_action(self, **kwargs):
|
||||
return {
|
||||
"instance_key": "default",
|
||||
"symbol": kwargs["symbol"],
|
||||
"quantity": 3,
|
||||
"lifecycle_state": "active",
|
||||
"first_seen_at": "2026-09-07T00:00:00+00:00",
|
||||
"last_seen_at": "2026-09-11T00:00:00+00:00",
|
||||
"quantity_changed_at": "2026-09-07T00:00:00+00:00",
|
||||
"resolved_at": None,
|
||||
"operator_action": kwargs["action"],
|
||||
"evidence_ref": kwargs["evidence_ref"],
|
||||
"operator_note": kwargs["note"],
|
||||
"operator_actor": kwargs["actor"],
|
||||
"operator_updated_at": "2026-09-11T00:00:00+00:00",
|
||||
"created_at": "2026-09-07T00:00:00+00:00",
|
||||
"updated_at": "2026-09-11T00:00:00+00:00",
|
||||
}
|
||||
|
||||
monkeypatch.setattr(
|
||||
"trex_core.adapters.portfolio.unmanaged_position_store.PostgreSQLUnmanagedPositionStore",
|
||||
_Store,
|
||||
)
|
||||
|
||||
rc = cli.main(
|
||||
[
|
||||
"unmanaged",
|
||||
"record",
|
||||
"008110",
|
||||
"investigating",
|
||||
"--evidence-ref",
|
||||
"kis-balance:2026-09-11",
|
||||
"--json",
|
||||
]
|
||||
)
|
||||
|
||||
assert rc == 0
|
||||
out = json.loads(capsys.readouterr().out)
|
||||
assert out["result"] == "recorded"
|
||||
assert out["state"]["symbol"] == "008110"
|
||||
|
||||
|
||||
def test_unmanaged_audit_outputs_lifecycle_and_operator_history(monkeypatch, capsys):
|
||||
class _Store:
|
||||
def __init__(self, *_args, **_kwargs):
|
||||
pass
|
||||
|
||||
def get_state(self, **kwargs):
|
||||
assert kwargs == {"instance_key": "inst-1", "symbol": "008110"}
|
||||
return {
|
||||
"instance_key": "inst-1",
|
||||
"symbol": "008110",
|
||||
"quantity": 3,
|
||||
"lifecycle_state": "active",
|
||||
"first_seen_at": "2026-09-07T00:00:00+00:00",
|
||||
"last_seen_at": "2026-09-11T00:00:00+00:00",
|
||||
"quantity_changed_at": "2026-09-07T00:00:00+00:00",
|
||||
"resolved_at": None,
|
||||
"operator_action": "investigating",
|
||||
"evidence_ref": "kis-balance:2026-09-11",
|
||||
"operator_note": "confirmed",
|
||||
"operator_actor": "cli:test",
|
||||
"operator_updated_at": "2026-09-11T00:00:00+00:00",
|
||||
"created_at": "2026-09-07T00:00:00+00:00",
|
||||
"updated_at": "2026-09-11T00:00:00+00:00",
|
||||
}
|
||||
|
||||
def list_audit(self, **kwargs):
|
||||
assert kwargs == {"instance_key": "inst-1", "symbol": "008110"}
|
||||
return [
|
||||
{
|
||||
"event_type": "detected",
|
||||
"recorded_at": "2026-09-07T00:00:00+00:00",
|
||||
"quantity": 3,
|
||||
"action": None,
|
||||
"actor": "reconcile",
|
||||
"evidence_ref": None,
|
||||
"note": "",
|
||||
},
|
||||
{
|
||||
"event_type": "operator_action",
|
||||
"recorded_at": "2026-09-11T00:00:00+00:00",
|
||||
"quantity": 3,
|
||||
"action": "investigating",
|
||||
"actor": "cli:test",
|
||||
"evidence_ref": "kis-balance:2026-09-11",
|
||||
"note": "confirmed",
|
||||
},
|
||||
]
|
||||
|
||||
monkeypatch.setattr(
|
||||
"trex_core.adapters.portfolio.unmanaged_position_store.PostgreSQLUnmanagedPositionStore",
|
||||
_Store,
|
||||
)
|
||||
args = SimpleNamespace(
|
||||
cmd="unmanaged",
|
||||
unmanaged_cmd="audit",
|
||||
symbol="008110",
|
||||
instance="inst-1",
|
||||
config=None,
|
||||
config_path=None,
|
||||
env_file=None,
|
||||
json=True,
|
||||
timeout=10.0,
|
||||
)
|
||||
|
||||
rc = cli._cmd_unmanaged_audit(args, SimpleNamespace(dsn="postgresql://x/y"))
|
||||
|
||||
assert rc == 0
|
||||
out = json.loads(capsys.readouterr().out)
|
||||
assert out["symbol"] == "008110"
|
||||
assert out["state"]["operator_actor"] == "cli:test"
|
||||
assert [row["event_type"] for row in out["audit"]] == ["detected", "operator_action"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# REL-010 regression: _cmd_verify broker import must not raise ImportError
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -538,10 +890,10 @@ def test_verify_cmd_broker_import_does_not_raise(monkeypatch):
|
||||
|
||||
# Patch at module level — _cmd_verify uses local `from ... import` so we
|
||||
# must patch the source module, not the cli namespace.
|
||||
import trex_core.adapters.persistence.postgresql as pg_mod
|
||||
import trex_core.adapters.portfolio.reconcile_warning_store as rws_mod
|
||||
from trex_core.application.portfolio import InMemoryReconcileWarningStore
|
||||
from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository
|
||||
import trex_core.adapters.portfolio.reconcile_warning_store as rws_mod
|
||||
import trex_core.adapters.persistence.postgresql as pg_mod
|
||||
|
||||
monkeypatch.setattr(rws_mod, "PostgreSQLReconcileWarningStore",
|
||||
lambda dsn: InMemoryReconcileWarningStore())
|
||||
|
||||
@@ -73,3 +73,44 @@ def test_daemon_wires_shared_alarm_emitter_into_intent_service():
|
||||
found = True
|
||||
break
|
||||
assert found, "Daemon must pass self.alarm_emitter into IntentService(alarm_emitter=...)"
|
||||
|
||||
|
||||
def test_daemon_wires_reconciliation_broker_into_snapshot_scheduler():
|
||||
src = (
|
||||
Path(__file__).resolve().parents[2]
|
||||
/ "trex_core"
|
||||
/ "runtime"
|
||||
/ "daemon.py"
|
||||
).read_text(encoding="utf-8")
|
||||
tree = ast.parse(src)
|
||||
found = False
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, ast.Assign):
|
||||
continue
|
||||
if not any(
|
||||
isinstance(t, ast.Attribute) and t.attr == "snapshot_scheduler"
|
||||
for t in node.targets
|
||||
):
|
||||
continue
|
||||
call = node.value
|
||||
if not (
|
||||
isinstance(call, ast.Call)
|
||||
and isinstance(call.func, ast.Name)
|
||||
and call.func.id == "SnapshotScheduler"
|
||||
):
|
||||
continue
|
||||
for kw in call.keywords:
|
||||
if kw.arg != "core_holdings_broker":
|
||||
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=...)"
|
||||
)
|
||||
|
||||
@@ -382,9 +382,7 @@ class KISBroker:
|
||||
transport_failure: bool,
|
||||
) -> float:
|
||||
"""Return bounded state/age-based delay for the next authoritative poll."""
|
||||
if status.state != "unknown":
|
||||
delay = _FILL_POLL_MIN_INTERVAL_SECONDS
|
||||
elif age_seconds < 60:
|
||||
if status.state != "unknown" or age_seconds < 60:
|
||||
delay = _FILL_POLL_MIN_INTERVAL_SECONDS
|
||||
elif age_seconds < 5 * 60:
|
||||
delay = 15.0
|
||||
@@ -1284,6 +1282,9 @@ class KISBroker:
|
||||
seen_continuations.add(next_continuation)
|
||||
continuation = next_continuation
|
||||
|
||||
def exposes_full_balance_snapshot(self) -> bool:
|
||||
return True
|
||||
|
||||
def get_available_cash(self, *, price_check_code: str = "005930") -> int:
|
||||
"""Return KIS cash-only buy capacity (`nrcvb_buy_amt`).
|
||||
|
||||
|
||||
@@ -13,11 +13,11 @@ import os
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from threading import Lock
|
||||
from typing import Optional
|
||||
|
||||
from trex_core.application.broker_port import (
|
||||
BalanceItem,
|
||||
BrokerBalanceError,
|
||||
BrokerPort,
|
||||
CancelOutcome,
|
||||
FillStatus,
|
||||
OrderAck,
|
||||
@@ -134,8 +134,8 @@ class OwnershipAwareBroker:
|
||||
*,
|
||||
side: str,
|
||||
symbol: str,
|
||||
order_date: Optional[str] = None,
|
||||
submitted_at: Optional[datetime] = None,
|
||||
order_date: str | None = None,
|
||||
submitted_at: datetime | None = None,
|
||||
) -> FillStatus:
|
||||
result = self._delegate.poll_fill(
|
||||
broker_order_id,
|
||||
@@ -175,6 +175,13 @@ class OwnershipAwareBroker:
|
||||
def get_balance(self) -> list[BalanceItem]:
|
||||
return self._store.project(self._delegate.get_balance())
|
||||
|
||||
def authoritative_balance_reader(self) -> BrokerPort:
|
||||
return self._delegate
|
||||
|
||||
def exposes_full_balance_snapshot(self) -> bool:
|
||||
"""False: projected balances omit operator-owned/unmanaged broker positions."""
|
||||
return False
|
||||
|
||||
def account_scope_hash(self) -> str:
|
||||
return self._delegate.account_scope_hash()
|
||||
|
||||
|
||||
@@ -166,6 +166,9 @@ class PaperBroker:
|
||||
return [BalanceItem(symbol=sym, hldg_qty=qty)
|
||||
for sym, qty in balances.items() if qty != 0]
|
||||
|
||||
def exposes_full_balance_snapshot(self) -> bool:
|
||||
return True
|
||||
|
||||
# --- REQ-038: programmable current price (safety-sell scenario double) ---
|
||||
def set_current_price(self, symbol: str, price: Decimal) -> None:
|
||||
"""Program the current price returned by get_current_price(symbol)."""
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
CREATE TABLE IF NOT EXISTS unmanaged_position_state (
|
||||
instance_key TEXT NOT NULL,
|
||||
symbol TEXT NOT NULL,
|
||||
quantity INTEGER NOT NULL,
|
||||
lifecycle_state TEXT NOT NULL,
|
||||
first_seen_at TIMESTAMPTZ NOT NULL,
|
||||
last_seen_at TIMESTAMPTZ NOT NULL,
|
||||
quantity_changed_at TIMESTAMPTZ NOT NULL,
|
||||
resolved_at TIMESTAMPTZ NULL,
|
||||
operator_action TEXT NULL,
|
||||
evidence_ref TEXT NULL,
|
||||
operator_note TEXT NOT NULL DEFAULT '',
|
||||
operator_actor TEXT NULL,
|
||||
operator_updated_at TIMESTAMPTZ NULL,
|
||||
created_at TIMESTAMPTZ NOT NULL,
|
||||
updated_at TIMESTAMPTZ NOT NULL,
|
||||
PRIMARY KEY (instance_key, symbol)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_unmanaged_position_state_lifecycle
|
||||
ON unmanaged_position_state (instance_key, lifecycle_state, updated_at DESC);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS unmanaged_position_audit (
|
||||
audit_id BIGSERIAL PRIMARY KEY,
|
||||
instance_key TEXT NOT NULL,
|
||||
symbol TEXT NOT NULL,
|
||||
event_type TEXT NOT NULL,
|
||||
action TEXT NULL,
|
||||
quantity INTEGER NOT NULL,
|
||||
lifecycle_state TEXT NOT NULL,
|
||||
actor TEXT NOT NULL,
|
||||
evidence_ref TEXT NULL,
|
||||
note TEXT NOT NULL DEFAULT '',
|
||||
created_at TIMESTAMPTZ NOT NULL
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_unmanaged_position_audit_lookup
|
||||
ON unmanaged_position_audit (instance_key, symbol, audit_id);
|
||||
@@ -3,10 +3,12 @@ from trex_core.adapters.portfolio.hub_read import PortfolioHubReadAdapter
|
||||
from trex_core.adapters.portfolio.pause_store import PostgreSQLPauseStore
|
||||
from trex_core.adapters.portfolio.reconcile_warning_store import PostgreSQLReconcileWarningStore
|
||||
from trex_core.adapters.portfolio.stub_read import StubPortfolioRead
|
||||
from trex_core.adapters.portfolio.unmanaged_position_store import PostgreSQLUnmanagedPositionStore
|
||||
|
||||
__all__ = [
|
||||
"PortfolioHubReadAdapter",
|
||||
"PostgreSQLPauseStore",
|
||||
"PostgreSQLReconcileWarningStore",
|
||||
"PostgreSQLUnmanagedPositionStore",
|
||||
"StubPortfolioRead",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,379 @@
|
||||
"""PostgreSQL-backed unmanaged-position lifecycle store."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
|
||||
from trex_core.adapters.persistence.postgresql import _require_psycopg
|
||||
|
||||
_STATE_COLUMNS = (
|
||||
"instance_key, symbol, quantity, lifecycle_state, "
|
||||
"first_seen_at, last_seen_at, quantity_changed_at, resolved_at, "
|
||||
"operator_action, evidence_ref, operator_note, operator_actor, "
|
||||
"operator_updated_at, created_at, updated_at"
|
||||
)
|
||||
|
||||
|
||||
class PostgreSQLUnmanagedPositionStore:
|
||||
"""CRUD + audit for unmanaged positions keyed by (instance_key, symbol)."""
|
||||
|
||||
def __init__(self, dsn: str, *, psycopg_module: Any = None) -> None:
|
||||
self._psycopg = _require_psycopg(psycopg_module)
|
||||
self._dsn = dsn
|
||||
|
||||
@staticmethod
|
||||
def _latest_snapshot_at(state: dict) -> datetime:
|
||||
resolved_at = state.get("resolved_at")
|
||||
last_seen_at = state.get("last_seen_at")
|
||||
if isinstance(resolved_at, datetime) and (
|
||||
not isinstance(last_seen_at, datetime) or resolved_at >= last_seen_at
|
||||
):
|
||||
return resolved_at
|
||||
if isinstance(last_seen_at, datetime):
|
||||
return last_seen_at
|
||||
return state.get("updated_at") or state.get("created_at") or datetime.min.replace(tzinfo=UTC)
|
||||
|
||||
def observe(
|
||||
self,
|
||||
*,
|
||||
instance_key: str,
|
||||
symbol: str,
|
||||
quantity: int,
|
||||
observed_at: datetime | None = None,
|
||||
actor: str = "reconcile",
|
||||
) -> dict:
|
||||
now = observed_at or datetime.now(tz=UTC)
|
||||
for _attempt in range(2):
|
||||
try:
|
||||
with self._psycopg.connect(self._dsn, autocommit=False) as conn, conn.cursor() as cur:
|
||||
cur.execute(
|
||||
f"SELECT {_STATE_COLUMNS} FROM unmanaged_position_state "
|
||||
"WHERE instance_key = %s AND symbol = %s "
|
||||
"FOR UPDATE",
|
||||
(instance_key, symbol),
|
||||
)
|
||||
row = cur.fetchone()
|
||||
previous = self._row_to_state(row) if row is not None else None
|
||||
previous_quantity = previous.get("quantity") if previous is not None else None
|
||||
if previous is not None and now < self._latest_snapshot_at(previous):
|
||||
conn.rollback()
|
||||
return {
|
||||
"transition": "stale",
|
||||
"previous_quantity": previous_quantity,
|
||||
"state": previous,
|
||||
}
|
||||
if previous is None:
|
||||
transition = "detected"
|
||||
cur.execute(
|
||||
"INSERT INTO unmanaged_position_state "
|
||||
f"({_STATE_COLUMNS}) "
|
||||
"VALUES (%s, %s, %s, 'active', %s, %s, %s, NULL, "
|
||||
" NULL, NULL, '', NULL, NULL, %s, %s) "
|
||||
f"RETURNING {_STATE_COLUMNS}",
|
||||
(instance_key, symbol, quantity, now, now, now, now, now),
|
||||
)
|
||||
elif previous["lifecycle_state"] == "resolved":
|
||||
if now <= self._latest_snapshot_at(previous):
|
||||
conn.rollback()
|
||||
return {
|
||||
"transition": "stale",
|
||||
"previous_quantity": previous_quantity,
|
||||
"state": previous,
|
||||
}
|
||||
transition = "detected"
|
||||
cur.execute(
|
||||
"UPDATE unmanaged_position_state "
|
||||
"SET quantity = %s, lifecycle_state = 'active', "
|
||||
" last_seen_at = %s, quantity_changed_at = %s, resolved_at = NULL, "
|
||||
" operator_action = NULL, evidence_ref = NULL, operator_note = '', "
|
||||
" operator_actor = NULL, operator_updated_at = NULL, "
|
||||
" updated_at = %s "
|
||||
"WHERE instance_key = %s AND symbol = %s "
|
||||
" AND lifecycle_state = 'resolved' AND resolved_at < %s "
|
||||
f"RETURNING {_STATE_COLUMNS}",
|
||||
(quantity, now, now, now, instance_key, symbol, now),
|
||||
)
|
||||
elif int(previous["quantity"]) != int(quantity):
|
||||
if now <= previous["last_seen_at"]:
|
||||
conn.rollback()
|
||||
return {
|
||||
"transition": "stale",
|
||||
"previous_quantity": previous_quantity,
|
||||
"state": previous,
|
||||
}
|
||||
transition = "quantity_changed"
|
||||
cur.execute(
|
||||
"UPDATE unmanaged_position_state "
|
||||
"SET quantity = %s, last_seen_at = %s, quantity_changed_at = %s, "
|
||||
" resolved_at = NULL, lifecycle_state = 'active', updated_at = %s "
|
||||
"WHERE instance_key = %s AND symbol = %s "
|
||||
" AND lifecycle_state = 'active' AND last_seen_at < %s "
|
||||
f"RETURNING {_STATE_COLUMNS}",
|
||||
(quantity, now, now, now, instance_key, symbol, now),
|
||||
)
|
||||
else:
|
||||
transition = "unchanged"
|
||||
if now > previous["last_seen_at"]:
|
||||
cur.execute(
|
||||
"UPDATE unmanaged_position_state "
|
||||
"SET last_seen_at = %s, updated_at = %s "
|
||||
"WHERE instance_key = %s AND symbol = %s "
|
||||
" AND lifecycle_state = 'active' AND quantity = %s AND last_seen_at < %s "
|
||||
f"RETURNING {_STATE_COLUMNS}",
|
||||
(now, now, instance_key, symbol, quantity, now),
|
||||
)
|
||||
fetched = cur.fetchone()
|
||||
state = previous if fetched is None else self._row_to_state(fetched)
|
||||
else:
|
||||
conn.rollback()
|
||||
return {
|
||||
"transition": "unchanged",
|
||||
"previous_quantity": previous_quantity,
|
||||
"state": previous,
|
||||
}
|
||||
if transition != "unchanged":
|
||||
fetched = cur.fetchone()
|
||||
if fetched is None:
|
||||
conn.rollback()
|
||||
return {
|
||||
"transition": "stale",
|
||||
"previous_quantity": previous_quantity,
|
||||
"state": previous,
|
||||
}
|
||||
state = self._row_to_state(fetched)
|
||||
if transition != "unchanged":
|
||||
note = (
|
||||
f"previous_quantity={previous_quantity}"
|
||||
if transition == "quantity_changed" and previous_quantity is not None
|
||||
else ""
|
||||
)
|
||||
self._insert_audit(
|
||||
cur,
|
||||
instance_key=instance_key,
|
||||
symbol=symbol,
|
||||
event_type=transition,
|
||||
action=None,
|
||||
quantity=quantity,
|
||||
lifecycle_state=state["lifecycle_state"],
|
||||
actor=actor,
|
||||
evidence_ref=state.get("evidence_ref"),
|
||||
note=note,
|
||||
created_at=now,
|
||||
)
|
||||
conn.commit()
|
||||
return {
|
||||
"transition": transition,
|
||||
"previous_quantity": previous_quantity,
|
||||
"state": state,
|
||||
}
|
||||
except self._psycopg.errors.UniqueViolation:
|
||||
continue
|
||||
raise RuntimeError(
|
||||
"failed to persist unmanaged_position_state due to repeated concurrent insert race"
|
||||
)
|
||||
|
||||
def resolve_absent(
|
||||
self,
|
||||
*,
|
||||
instance_key: str,
|
||||
observed_symbols: set[str],
|
||||
observed_at: datetime | None = None,
|
||||
actor: str = "reconcile",
|
||||
) -> list[dict]:
|
||||
now = observed_at or datetime.now(tz=UTC)
|
||||
with self._psycopg.connect(self._dsn, autocommit=False) as conn, conn.cursor() as cur:
|
||||
cur.execute(
|
||||
f"SELECT {_STATE_COLUMNS} FROM unmanaged_position_state "
|
||||
"WHERE instance_key = %s AND lifecycle_state = 'active' "
|
||||
"ORDER BY symbol "
|
||||
"FOR UPDATE",
|
||||
(instance_key,),
|
||||
)
|
||||
active = [self._row_to_state(row) for row in cur.fetchall()]
|
||||
resolved: list[dict] = []
|
||||
for state in active:
|
||||
if state["symbol"] in observed_symbols:
|
||||
continue
|
||||
if now <= state["last_seen_at"]:
|
||||
continue
|
||||
cur.execute(
|
||||
"UPDATE unmanaged_position_state "
|
||||
"SET lifecycle_state = 'resolved', resolved_at = %s, updated_at = %s "
|
||||
"WHERE instance_key = %s AND symbol = %s "
|
||||
" AND lifecycle_state = 'active' AND last_seen_at < %s "
|
||||
f"RETURNING {_STATE_COLUMNS}",
|
||||
(now, now, instance_key, state["symbol"], now),
|
||||
)
|
||||
row = cur.fetchone()
|
||||
if row is None:
|
||||
continue
|
||||
resolved_state = self._row_to_state(row)
|
||||
resolved.append(resolved_state)
|
||||
self._insert_audit(
|
||||
cur,
|
||||
instance_key=instance_key,
|
||||
symbol=state["symbol"],
|
||||
event_type="resolved",
|
||||
action=None,
|
||||
quantity=state["quantity"],
|
||||
lifecycle_state="resolved",
|
||||
actor=actor,
|
||||
evidence_ref=resolved_state.get("evidence_ref"),
|
||||
note="",
|
||||
created_at=now,
|
||||
)
|
||||
conn.commit()
|
||||
return resolved
|
||||
|
||||
def get_state(self, *, instance_key: str, symbol: str) -> dict | None:
|
||||
with self._psycopg.connect(self._dsn) as conn, conn.cursor() as cur:
|
||||
cur.execute(
|
||||
f"SELECT {_STATE_COLUMNS} FROM unmanaged_position_state "
|
||||
"WHERE instance_key = %s AND symbol = %s "
|
||||
"LIMIT 1",
|
||||
(instance_key, symbol),
|
||||
)
|
||||
row = cur.fetchone()
|
||||
return None if row is None else self._row_to_state(row)
|
||||
|
||||
def list_states(
|
||||
self,
|
||||
*,
|
||||
instance_key: str,
|
||||
lifecycle_state: str = "active",
|
||||
symbol: str | None = None,
|
||||
) -> list[dict]:
|
||||
clauses = ["instance_key = %s"]
|
||||
params: list[object] = [instance_key]
|
||||
if lifecycle_state != "all":
|
||||
clauses.append("lifecycle_state = %s")
|
||||
params.append(lifecycle_state)
|
||||
if symbol is not None:
|
||||
clauses.append("symbol = %s")
|
||||
params.append(symbol)
|
||||
with self._psycopg.connect(self._dsn) as conn, conn.cursor() as cur:
|
||||
cur.execute(
|
||||
f"SELECT {_STATE_COLUMNS} FROM unmanaged_position_state "
|
||||
f"WHERE {' AND '.join(clauses)} ORDER BY symbol",
|
||||
tuple(params),
|
||||
)
|
||||
return [self._row_to_state(row) for row in cur.fetchall()]
|
||||
|
||||
def record_operator_action(
|
||||
self,
|
||||
*,
|
||||
instance_key: str,
|
||||
symbol: str,
|
||||
action: str,
|
||||
evidence_ref: str,
|
||||
actor: str = "operator",
|
||||
note: str = "",
|
||||
) -> dict | None:
|
||||
now = datetime.now(tz=UTC)
|
||||
with self._psycopg.connect(self._dsn, autocommit=False) as conn, conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"UPDATE unmanaged_position_state "
|
||||
"SET operator_action = %s, evidence_ref = %s, operator_note = %s, "
|
||||
" operator_actor = %s, operator_updated_at = %s, updated_at = %s "
|
||||
"WHERE instance_key = %s AND symbol = %s "
|
||||
f"RETURNING {_STATE_COLUMNS}",
|
||||
(action, evidence_ref, note, actor, now, now, instance_key, symbol),
|
||||
)
|
||||
row = cur.fetchone()
|
||||
if row is None:
|
||||
conn.rollback()
|
||||
return None
|
||||
state = self._row_to_state(row)
|
||||
self._insert_audit(
|
||||
cur,
|
||||
instance_key=instance_key,
|
||||
symbol=symbol,
|
||||
event_type="operator_action",
|
||||
action=action,
|
||||
quantity=state["quantity"],
|
||||
lifecycle_state=state["lifecycle_state"],
|
||||
actor=actor,
|
||||
evidence_ref=evidence_ref,
|
||||
note=note,
|
||||
created_at=now,
|
||||
)
|
||||
conn.commit()
|
||||
return state
|
||||
|
||||
def list_audit(self, *, instance_key: str, symbol: str) -> list[dict]:
|
||||
with self._psycopg.connect(self._dsn) as conn, conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"SELECT event_type, action, quantity, lifecycle_state, actor, "
|
||||
" evidence_ref, note, created_at "
|
||||
"FROM unmanaged_position_audit "
|
||||
"WHERE instance_key = %s AND symbol = %s "
|
||||
"ORDER BY audit_id",
|
||||
(instance_key, symbol),
|
||||
)
|
||||
return [
|
||||
{
|
||||
"event_type": row[0],
|
||||
"action": row[1],
|
||||
"quantity": row[2],
|
||||
"lifecycle_state": row[3],
|
||||
"actor": row[4],
|
||||
"evidence_ref": row[5],
|
||||
"note": row[6],
|
||||
"recorded_at": row[7],
|
||||
}
|
||||
for row in cur.fetchall()
|
||||
]
|
||||
|
||||
def _insert_audit(
|
||||
self,
|
||||
cur,
|
||||
*,
|
||||
instance_key: str,
|
||||
symbol: str,
|
||||
event_type: str,
|
||||
action: str | None,
|
||||
quantity: int,
|
||||
lifecycle_state: str,
|
||||
actor: str,
|
||||
evidence_ref: str | None,
|
||||
note: str,
|
||||
created_at: datetime,
|
||||
) -> None:
|
||||
cur.execute(
|
||||
"INSERT INTO unmanaged_position_audit "
|
||||
"(instance_key, symbol, event_type, action, quantity, lifecycle_state, "
|
||||
" actor, evidence_ref, note, created_at) "
|
||||
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
|
||||
(
|
||||
instance_key,
|
||||
symbol,
|
||||
event_type,
|
||||
action,
|
||||
quantity,
|
||||
lifecycle_state,
|
||||
actor,
|
||||
evidence_ref,
|
||||
note,
|
||||
created_at,
|
||||
),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _row_to_state(row) -> dict:
|
||||
return {
|
||||
"instance_key": row[0],
|
||||
"symbol": row[1],
|
||||
"quantity": row[2],
|
||||
"lifecycle_state": row[3],
|
||||
"first_seen_at": row[4],
|
||||
"last_seen_at": row[5],
|
||||
"quantity_changed_at": row[6],
|
||||
"resolved_at": row[7],
|
||||
"operator_action": row[8],
|
||||
"evidence_ref": row[9],
|
||||
"operator_note": row[10],
|
||||
"operator_actor": row[11],
|
||||
"operator_updated_at": row[12],
|
||||
"created_at": row[13],
|
||||
"updated_at": row[14],
|
||||
}
|
||||
@@ -562,6 +562,291 @@ class InMemoryReconcileNotificationStore:
|
||||
]
|
||||
|
||||
|
||||
class UnmanagedPositionStore(Protocol):
|
||||
"""Durable unmanaged-position lifecycle store keyed by (instance_key, symbol)."""
|
||||
|
||||
def observe(
|
||||
self,
|
||||
*,
|
||||
instance_key: str,
|
||||
symbol: str,
|
||||
quantity: int,
|
||||
observed_at: datetime | None = None,
|
||||
actor: str = "reconcile",
|
||||
) -> dict:
|
||||
"""Return {"transition", "previous_quantity", "state"} for one observation."""
|
||||
...
|
||||
|
||||
def resolve_absent(
|
||||
self,
|
||||
*,
|
||||
instance_key: str,
|
||||
observed_symbols: set[str],
|
||||
observed_at: datetime | None = None,
|
||||
actor: str = "reconcile",
|
||||
) -> list[dict]:
|
||||
"""Mark active positions absent from a successful broker read as resolved."""
|
||||
...
|
||||
|
||||
def get_state(self, *, instance_key: str, symbol: str) -> dict | None:
|
||||
...
|
||||
|
||||
def list_states(
|
||||
self,
|
||||
*,
|
||||
instance_key: str,
|
||||
lifecycle_state: str = "active",
|
||||
symbol: str | None = None,
|
||||
) -> list[dict]:
|
||||
...
|
||||
|
||||
def record_operator_action(
|
||||
self,
|
||||
*,
|
||||
instance_key: str,
|
||||
symbol: str,
|
||||
action: str,
|
||||
evidence_ref: str,
|
||||
actor: str = "operator",
|
||||
note: str = "",
|
||||
) -> dict | None:
|
||||
"""Persist an explicit operator disposition and append an audit event."""
|
||||
...
|
||||
|
||||
def list_audit(self, *, instance_key: str, symbol: str) -> list[dict]:
|
||||
...
|
||||
|
||||
|
||||
class InMemoryUnmanagedPositionStore:
|
||||
"""Process-local unmanaged-position lifecycle store for tests + CI."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._lock = threading.Lock()
|
||||
self._states: dict[tuple[str, str], dict] = {}
|
||||
self._audit: list[dict] = []
|
||||
|
||||
@staticmethod
|
||||
def _latest_snapshot_at(state: dict) -> datetime:
|
||||
resolved_at = state.get("resolved_at")
|
||||
last_seen_at = state.get("last_seen_at")
|
||||
if isinstance(resolved_at, datetime) and (
|
||||
not isinstance(last_seen_at, datetime) or resolved_at >= last_seen_at
|
||||
):
|
||||
return resolved_at
|
||||
if isinstance(last_seen_at, datetime):
|
||||
return last_seen_at
|
||||
return state.get("updated_at") or state.get("created_at") or datetime.min.replace(tzinfo=UTC)
|
||||
|
||||
def observe(
|
||||
self,
|
||||
*,
|
||||
instance_key: str,
|
||||
symbol: str,
|
||||
quantity: int,
|
||||
observed_at: datetime | None = None,
|
||||
actor: str = "reconcile",
|
||||
) -> dict:
|
||||
now = observed_at or datetime.now(tz=UTC)
|
||||
key = (instance_key, symbol)
|
||||
with self._lock:
|
||||
existing = self._states.get(key)
|
||||
previous_quantity = existing.get("quantity") if existing is not None else None
|
||||
preserved = {} if existing is None else {
|
||||
"created_at": existing.get("created_at", now),
|
||||
}
|
||||
if existing is not None:
|
||||
latest_snapshot_at = self._latest_snapshot_at(existing)
|
||||
if now < latest_snapshot_at:
|
||||
return {
|
||||
"transition": "stale",
|
||||
"previous_quantity": previous_quantity,
|
||||
"state": dict(existing),
|
||||
}
|
||||
if existing is None or existing.get("lifecycle_state") == "resolved":
|
||||
if existing is not None and now == self._latest_snapshot_at(existing):
|
||||
return {
|
||||
"transition": "stale",
|
||||
"previous_quantity": previous_quantity,
|
||||
"state": dict(existing),
|
||||
}
|
||||
transition = "detected"
|
||||
state = {
|
||||
"instance_key": instance_key,
|
||||
"symbol": symbol,
|
||||
"quantity": quantity,
|
||||
"lifecycle_state": "active",
|
||||
"first_seen_at": (
|
||||
existing.get("first_seen_at", now) if existing is not None else now
|
||||
),
|
||||
"last_seen_at": now,
|
||||
"quantity_changed_at": now,
|
||||
"resolved_at": None,
|
||||
"operator_action": None,
|
||||
"evidence_ref": None,
|
||||
"operator_note": "",
|
||||
"operator_actor": None,
|
||||
"operator_updated_at": None,
|
||||
"recorded_at": now,
|
||||
**preserved,
|
||||
"updated_at": now,
|
||||
}
|
||||
self._states[key] = state
|
||||
elif int(existing["quantity"]) != int(quantity):
|
||||
if now <= existing["last_seen_at"]:
|
||||
return {
|
||||
"transition": "stale",
|
||||
"previous_quantity": previous_quantity,
|
||||
"state": dict(existing),
|
||||
}
|
||||
transition = "quantity_changed"
|
||||
existing["quantity"] = quantity
|
||||
existing["last_seen_at"] = now
|
||||
existing["quantity_changed_at"] = now
|
||||
existing["resolved_at"] = None
|
||||
existing["lifecycle_state"] = "active"
|
||||
existing["updated_at"] = now
|
||||
state = existing
|
||||
else:
|
||||
transition = "unchanged"
|
||||
if now > existing["last_seen_at"]:
|
||||
existing["last_seen_at"] = now
|
||||
existing["updated_at"] = now
|
||||
state = existing
|
||||
if transition != "unchanged":
|
||||
self._audit.append(
|
||||
{
|
||||
"instance_key": instance_key,
|
||||
"symbol": symbol,
|
||||
"event_type": transition,
|
||||
"action": None,
|
||||
"quantity": quantity,
|
||||
"lifecycle_state": state["lifecycle_state"],
|
||||
"actor": actor,
|
||||
"evidence_ref": state.get("evidence_ref"),
|
||||
"note": (
|
||||
f"previous_quantity={previous_quantity}"
|
||||
if transition == "quantity_changed" and previous_quantity is not None
|
||||
else ""
|
||||
),
|
||||
"recorded_at": now,
|
||||
}
|
||||
)
|
||||
return {
|
||||
"transition": transition,
|
||||
"previous_quantity": previous_quantity,
|
||||
"state": dict(state),
|
||||
}
|
||||
|
||||
def resolve_absent(
|
||||
self,
|
||||
*,
|
||||
instance_key: str,
|
||||
observed_symbols: set[str],
|
||||
observed_at: datetime | None = None,
|
||||
actor: str = "reconcile",
|
||||
) -> list[dict]:
|
||||
now = observed_at or datetime.now(tz=UTC)
|
||||
with self._lock:
|
||||
resolved: list[dict] = []
|
||||
for key, state in sorted(self._states.items()):
|
||||
if key[0] != instance_key:
|
||||
continue
|
||||
if state.get("lifecycle_state") != "active":
|
||||
continue
|
||||
if state["symbol"] in observed_symbols:
|
||||
continue
|
||||
if now <= state["last_seen_at"]:
|
||||
continue
|
||||
state["lifecycle_state"] = "resolved"
|
||||
state["resolved_at"] = now
|
||||
state["updated_at"] = now
|
||||
resolved.append(dict(state))
|
||||
self._audit.append(
|
||||
{
|
||||
"instance_key": state["instance_key"],
|
||||
"symbol": state["symbol"],
|
||||
"event_type": "resolved",
|
||||
"action": None,
|
||||
"quantity": state["quantity"],
|
||||
"lifecycle_state": "resolved",
|
||||
"actor": actor,
|
||||
"evidence_ref": state.get("evidence_ref"),
|
||||
"note": "",
|
||||
"recorded_at": now,
|
||||
}
|
||||
)
|
||||
return resolved
|
||||
|
||||
def get_state(self, *, instance_key: str, symbol: str) -> dict | None:
|
||||
with self._lock:
|
||||
state = self._states.get((instance_key, symbol))
|
||||
return dict(state) if state is not None else None
|
||||
|
||||
def list_states(
|
||||
self,
|
||||
*,
|
||||
instance_key: str,
|
||||
lifecycle_state: str = "active",
|
||||
symbol: str | None = None,
|
||||
) -> list[dict]:
|
||||
with self._lock:
|
||||
rows = []
|
||||
for state in self._states.values():
|
||||
if state["instance_key"] != instance_key:
|
||||
continue
|
||||
if lifecycle_state != "all" and state["lifecycle_state"] != lifecycle_state:
|
||||
continue
|
||||
if symbol is not None and state["symbol"] != symbol:
|
||||
continue
|
||||
rows.append(dict(state))
|
||||
return sorted(rows, key=lambda item: item["symbol"])
|
||||
|
||||
def record_operator_action(
|
||||
self,
|
||||
*,
|
||||
instance_key: str,
|
||||
symbol: str,
|
||||
action: str,
|
||||
evidence_ref: str,
|
||||
actor: str = "operator",
|
||||
note: str = "",
|
||||
) -> dict | None:
|
||||
now = datetime.now(tz=UTC)
|
||||
with self._lock:
|
||||
state = self._states.get((instance_key, symbol))
|
||||
if state is None:
|
||||
return None
|
||||
state["operator_action"] = action
|
||||
state["evidence_ref"] = evidence_ref
|
||||
state["operator_note"] = note
|
||||
state["operator_actor"] = actor
|
||||
state["operator_updated_at"] = now
|
||||
state["updated_at"] = now
|
||||
self._audit.append(
|
||||
{
|
||||
"instance_key": instance_key,
|
||||
"symbol": symbol,
|
||||
"event_type": "operator_action",
|
||||
"action": action,
|
||||
"quantity": state["quantity"],
|
||||
"lifecycle_state": state["lifecycle_state"],
|
||||
"actor": actor,
|
||||
"evidence_ref": evidence_ref,
|
||||
"note": note,
|
||||
"recorded_at": now,
|
||||
}
|
||||
)
|
||||
return dict(state)
|
||||
|
||||
def list_audit(self, *, instance_key: str, symbol: str) -> list[dict]:
|
||||
with self._lock:
|
||||
return [
|
||||
dict(row)
|
||||
for row in self._audit
|
||||
if row["instance_key"] == instance_key and row["symbol"] == symbol
|
||||
]
|
||||
|
||||
|
||||
def slot_to_dict(
|
||||
s,
|
||||
*,
|
||||
|
||||
@@ -29,11 +29,16 @@ import hashlib
|
||||
import logging
|
||||
import math
|
||||
import re
|
||||
from datetime import UTC, datetime
|
||||
from typing import cast
|
||||
|
||||
from trex_core.application.alarm_emitter import AlarmEmitter
|
||||
from trex_core.application.broker_port import BrokerPort
|
||||
from trex_core.application.portfolio import ReconcileWarningStore, SnapshotStore
|
||||
from trex_core.application.portfolio import (
|
||||
ReconcileWarningStore,
|
||||
SnapshotStore,
|
||||
UnmanagedPositionStore,
|
||||
)
|
||||
from trex_core.domain.slot.state import SlotState
|
||||
|
||||
_log = logging.getLogger(__name__)
|
||||
@@ -341,6 +346,37 @@ def _retry_pending_resolved(
|
||||
)
|
||||
|
||||
|
||||
def _notify_unmanaged_position_transition(
|
||||
alarm_emitter: AlarmEmitter,
|
||||
*,
|
||||
event_type: str,
|
||||
instance_key: str,
|
||||
symbol: str,
|
||||
quantity: int,
|
||||
) -> None:
|
||||
alarm_emitter.publish_alarm(
|
||||
f"unmanaged_position_{event_type}",
|
||||
f"unmanaged position {event_type}: symbol={symbol}",
|
||||
fields={
|
||||
"instance_key": instance_key,
|
||||
"symbol": symbol,
|
||||
"broker_quantity": quantity,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def _broker_balances_are_authoritative_for_unmanaged_resolution(broker: BrokerPort) -> bool:
|
||||
marker = getattr(broker, "exposes_full_balance_snapshot", None)
|
||||
if callable(marker):
|
||||
try:
|
||||
return bool(marker())
|
||||
except Exception: # noqa: BLE001 - fail closed on broker self-report failure
|
||||
return False
|
||||
if isinstance(marker, bool):
|
||||
return marker
|
||||
return False
|
||||
|
||||
|
||||
def reconcile_core_holdings(
|
||||
*,
|
||||
slot_repo,
|
||||
@@ -350,6 +386,7 @@ def reconcile_core_holdings(
|
||||
alarm_emitter: AlarmEmitter | None = None,
|
||||
warn_store: ReconcileWarningStore | None = None,
|
||||
notification_store=None,
|
||||
unmanaged_position_store: UnmanagedPositionStore | 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.
|
||||
@@ -371,12 +408,14 @@ def reconcile_core_holdings(
|
||||
Returns {key: 'error'|'qty_warning'|'ok'|'unmanaged_position'|
|
||||
'managed_non_holding_divergence'} for telemetry.
|
||||
"""
|
||||
snapshot_observed_at = datetime.now(tz=UTC)
|
||||
broker_qty: dict[str, int] = {
|
||||
item.symbol: int(item.hldg_qty) for item in broker.get_balance()
|
||||
}
|
||||
decisions: dict[str, str] = {}
|
||||
matched: set[str] = set()
|
||||
active_b_identity_keys: set[str] = set() # req-0007: track active B mismatches for retry
|
||||
active_unmanaged_symbols: set[str] = set()
|
||||
for slot in slot_repo.list_holding_slots(instance_key):
|
||||
sym = slot.symbol
|
||||
matched.add(sym)
|
||||
@@ -458,10 +497,88 @@ def reconcile_core_holdings(
|
||||
)
|
||||
else:
|
||||
decisions[f"broker:{sym}"] = "unmanaged_position"
|
||||
_log.warning(
|
||||
"unmanaged position: symbol=%s qty=%d instance_key=%s",
|
||||
sym, bq, instance_key,
|
||||
active_unmanaged_symbols.add(sym)
|
||||
if unmanaged_position_store is None:
|
||||
_log.warning(
|
||||
"unmanaged position: symbol=%s qty=%d instance_key=%s",
|
||||
sym, bq, instance_key,
|
||||
)
|
||||
else:
|
||||
try:
|
||||
observed = unmanaged_position_store.observe(
|
||||
instance_key=instance_key,
|
||||
symbol=sym,
|
||||
quantity=bq,
|
||||
observed_at=snapshot_observed_at,
|
||||
)
|
||||
except Exception: # noqa: BLE001
|
||||
_log.exception(
|
||||
"failed to persist unmanaged_position for symbol=%s instance=%s",
|
||||
sym,
|
||||
instance_key,
|
||||
)
|
||||
_log.warning(
|
||||
"unmanaged position: symbol=%s qty=%d instance_key=%s",
|
||||
sym, bq, instance_key,
|
||||
)
|
||||
else:
|
||||
transition = observed["transition"]
|
||||
if transition == "detected":
|
||||
_log.warning(
|
||||
"unmanaged position detected: symbol=%s qty=%d instance_key=%s",
|
||||
sym, bq, instance_key,
|
||||
)
|
||||
if alarm_emitter is not None:
|
||||
_notify_unmanaged_position_transition(
|
||||
alarm_emitter,
|
||||
event_type="detected",
|
||||
instance_key=instance_key,
|
||||
symbol=sym,
|
||||
quantity=bq,
|
||||
)
|
||||
elif transition == "quantity_changed":
|
||||
_log.warning(
|
||||
"unmanaged position quantity changed: symbol=%s "
|
||||
"previous_qty=%s qty=%d instance_key=%s",
|
||||
sym, observed.get("previous_quantity"), bq, instance_key,
|
||||
)
|
||||
if alarm_emitter is not None:
|
||||
_notify_unmanaged_position_transition(
|
||||
alarm_emitter,
|
||||
event_type="changed",
|
||||
instance_key=instance_key,
|
||||
symbol=sym,
|
||||
quantity=bq,
|
||||
)
|
||||
if (
|
||||
unmanaged_position_store is not None
|
||||
and _broker_balances_are_authoritative_for_unmanaged_resolution(broker)
|
||||
):
|
||||
try:
|
||||
for state in unmanaged_position_store.resolve_absent(
|
||||
instance_key=instance_key,
|
||||
observed_symbols=active_unmanaged_symbols,
|
||||
observed_at=snapshot_observed_at,
|
||||
):
|
||||
_log.info(
|
||||
"unmanaged position resolved: symbol=%s qty=%d instance_key=%s",
|
||||
state["symbol"], state["quantity"], instance_key,
|
||||
)
|
||||
if alarm_emitter is not None:
|
||||
_notify_unmanaged_position_transition(
|
||||
alarm_emitter,
|
||||
event_type="resolved",
|
||||
instance_key=instance_key,
|
||||
symbol=state["symbol"],
|
||||
quantity=state["quantity"],
|
||||
)
|
||||
except Exception: # noqa: BLE001
|
||||
_log.exception("reconcile_core_holdings: unmanaged_position resolve failed")
|
||||
elif unmanaged_position_store is not None:
|
||||
_log.debug(
|
||||
"reconcile_core_holdings: unmanaged_position resolve skipped "
|
||||
"(non-authoritative broker balance view)"
|
||||
)
|
||||
# req-0007 / ac-0024: retry pending_resolved for B-case identities no longer mismatched.
|
||||
if alarm_emitter is not None and notification_store is not None:
|
||||
try:
|
||||
|
||||
@@ -12,8 +12,8 @@ from __future__ import annotations
|
||||
import logging
|
||||
import os
|
||||
import threading
|
||||
from datetime import date, datetime, time, timedelta, timezone
|
||||
from typing import Callable, Optional
|
||||
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.broker_port import BrokerPort
|
||||
@@ -26,6 +26,7 @@ from trex_core.application.portfolio import (
|
||||
ReconcileWarningStore,
|
||||
SnapshotStore,
|
||||
TradingContext,
|
||||
UnmanagedPositionStore,
|
||||
)
|
||||
from trex_core.application.reconciliation import reconcile_positions
|
||||
|
||||
@@ -43,45 +44,51 @@ class SnapshotScheduler:
|
||||
read_port: PortfolioHubReadPort,
|
||||
store: SnapshotStore,
|
||||
broker: BrokerPort,
|
||||
core_holdings_broker: BrokerPort | None = None,
|
||||
schedule_kst: time = time(hour=7, minute=30),
|
||||
clock: Optional[Callable[[], datetime]] = None,
|
||||
clock: Callable[[], datetime] | None = None,
|
||||
slot_repo=None,
|
||||
slot_service=None,
|
||||
instance_key: str = "default",
|
||||
alarm_emitter: Optional[AlarmEmitter] = None,
|
||||
warn_store: Optional[ReconcileWarningStore] = None,
|
||||
notification_store: Optional[ReconcileNotificationStore] = None,
|
||||
alarm_emitter: AlarmEmitter | None = None,
|
||||
warn_store: ReconcileWarningStore | None = None,
|
||||
notification_store: ReconcileNotificationStore | None = None,
|
||||
unmanaged_position_store: UnmanagedPositionStore | None = None,
|
||||
credential_fetch_enabled: bool = False,
|
||||
kis_reinit_callback: Optional[Callable[[BrokerCredentialPayload], None]] = None,
|
||||
credential_refresh: Optional[Callable[[], object]] = None,
|
||||
kis_reinit_callback: Callable[[BrokerCredentialPayload], None] | None = None,
|
||||
credential_refresh: Callable[[], object] | None = None,
|
||||
) -> None:
|
||||
self._read = read_port
|
||||
self._store = store
|
||||
self._broker = broker
|
||||
self._core_holdings_broker = core_holdings_broker or broker
|
||||
self._schedule_kst = schedule_kst
|
||||
self._clock = clock or (lambda: datetime.now(tz=timezone.utc))
|
||||
self._daemon_id: Optional[str] = None
|
||||
self._last_fired_date: Optional[datetime] = None
|
||||
self._clock = clock or (lambda: datetime.now(tz=UTC))
|
||||
self._daemon_id: str | None = None
|
||||
self._last_fired_date: datetime | None = None
|
||||
# dcy-020 / REQ-058 (BAT-030): asset-centric slot reconcile after each
|
||||
# universe fetch — always on when slot_repo is wired (the reconcile is
|
||||
# what creates/retires the permanent asset slots).
|
||||
self._slot_repo = slot_repo
|
||||
self._slot_service = slot_service
|
||||
self._instance_key = instance_key
|
||||
self._last_universe_size: Optional[int] = None
|
||||
self._last_universe_size: int | None = None
|
||||
# dcy-033 / REQ-100: best-effort alarm emitter.
|
||||
self._alarm_emitter = alarm_emitter
|
||||
# BAT-050 / REQ-103: durable reconcile-warning store (Optional).
|
||||
self._warn_store = warn_store
|
||||
# req-0007 / ac-0024: durable notification deduplication store (Optional).
|
||||
self._notification_store = notification_store
|
||||
self._unmanaged_position_store = unmanaged_position_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
|
||||
# dcy-036 / REQ-113 AC-2 / REQ-115 AC-1: optional callback to reinitialize
|
||||
# KISBroker with the fetched day-start credential. None = no reinit (paper
|
||||
# broker or credential_fetch_enabled=False). Set by daemon.py assembly.
|
||||
self._kis_reinit_callback: Optional[Callable[[BrokerCredentialPayload], None]] = kis_reinit_callback
|
||||
self._kis_reinit_callback: Callable[[BrokerCredentialPayload], None] | None = (
|
||||
kis_reinit_callback
|
||||
)
|
||||
# REQ-119: provider-owned refresh supersedes the legacy raw fetch hook.
|
||||
self._credential_refresh = credential_refresh
|
||||
# REQ-113: detect break-glass at init time (env var set by operator).
|
||||
@@ -90,8 +97,8 @@ class SnapshotScheduler:
|
||||
in ("1", "true", "yes")
|
||||
)
|
||||
# REQ-115: day-scoped credential cache. None until first successful fetch.
|
||||
self._cached_credential: Optional[BrokerCredentialPayload] = None
|
||||
self._credential_date: Optional[date] = None
|
||||
self._cached_credential: BrokerCredentialPayload | None = None
|
||||
self._credential_date: date | None = None
|
||||
# REQ-114: entry gate. True = new entries allowed (default).
|
||||
# credential_fetch_enabled=False keeps this always True (no gate).
|
||||
self._entry_allowed: bool = True
|
||||
@@ -263,7 +270,7 @@ class SnapshotScheduler:
|
||||
try:
|
||||
self._credential_refresh()
|
||||
self._entry_allowed = True
|
||||
except Exception as exc: # provider exposes only sanitized failure
|
||||
except Exception as exc: # noqa: BLE001 - provider exposes only sanitized failure
|
||||
_log.error("DAY_START_LOAD_FAILED: broker credential refresh failed: %s (instance=%s)",
|
||||
exc.__class__.__name__, self._instance_key)
|
||||
self._entry_allowed = False
|
||||
@@ -332,10 +339,11 @@ class SnapshotScheduler:
|
||||
# BAT-050 / REQ-103: pass warn_store for B-case durable warning.
|
||||
reconcile_core_holdings(
|
||||
slot_repo=self._slot_repo, slot_service=self._slot_service,
|
||||
broker=self._broker, instance_key=self._instance_key,
|
||||
broker=self._core_holdings_broker, instance_key=self._instance_key,
|
||||
alarm_emitter=self._alarm_emitter, # dcy-033 / REQ-100
|
||||
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,
|
||||
)
|
||||
except Exception: # noqa: BLE001
|
||||
_log.exception("core-holding divergence reconcile failed")
|
||||
|
||||
@@ -3,7 +3,7 @@ from __future__ import annotations
|
||||
|
||||
import os
|
||||
from dataclasses import dataclass
|
||||
from datetime import date, datetime, timedelta, timezone
|
||||
from datetime import UTC, date, datetime, timedelta, timezone
|
||||
from typing import Literal, Protocol
|
||||
|
||||
from trex_core.adapters.broker.paper import PaperBroker
|
||||
@@ -33,6 +33,7 @@ class BrokerCredentialUnavailable(RuntimeError):
|
||||
|
||||
class BrokerProvider(Protocol):
|
||||
def get_broker_for_read(self) -> tuple[BrokerPort, BrokerSource]: ...
|
||||
def get_broker_for_reconciliation(self) -> tuple[BrokerPort, BrokerSource]: ...
|
||||
def refresh_day_start_credential(self) -> BrokerSource: ...
|
||||
|
||||
|
||||
@@ -80,7 +81,7 @@ class ConfigBrokerProvider:
|
||||
def __init__(self, config: DaemonConfig, *, read_port=None, clock=None) -> None:
|
||||
self._config = config
|
||||
self._read = read_port
|
||||
self._clock = clock or (lambda: datetime.now(tz=timezone.utc))
|
||||
self._clock = clock or (lambda: datetime.now(tz=UTC))
|
||||
self._broker: BrokerPort | None = None
|
||||
self._source: BrokerSource | None = None
|
||||
self._credential_date: date | None = None
|
||||
@@ -101,7 +102,7 @@ class ConfigBrokerProvider:
|
||||
raise BrokerCredentialUnavailable("portfolio_hub_unavailable") from None
|
||||
except BrokerCredentialUnavailable:
|
||||
raise
|
||||
except Exception:
|
||||
except Exception: # noqa: BLE001 - sanitize arbitrary read adapter failures
|
||||
raise BrokerCredentialUnavailable("portfolio_hub_unavailable") from None
|
||||
if ctx is None or getattr(ctx, "broker", None) is None:
|
||||
raise BrokerCredentialUnavailable("broker_absent")
|
||||
@@ -121,7 +122,7 @@ class ConfigBrokerProvider:
|
||||
self._broker = broker
|
||||
else:
|
||||
self._broker.reinitialize(credential) # type: ignore[attr-defined]
|
||||
except Exception:
|
||||
except Exception: # noqa: BLE001 - sanitize broker construction/reload failures
|
||||
raise BrokerCredentialUnavailable("broker_initialization_failed") from None
|
||||
self._credential_date = self._clock().astimezone(_KST).date()
|
||||
# Version is designated non-secret by the portfolio contract; no account/key material.
|
||||
@@ -147,6 +148,13 @@ class ConfigBrokerProvider:
|
||||
self._source = source
|
||||
return self._broker, source # type: ignore[return-value]
|
||||
|
||||
def get_broker_for_reconciliation(self) -> tuple[BrokerPort, BrokerSource]:
|
||||
broker, source = self.get_broker_for_read()
|
||||
authoritative_reader = getattr(broker, "authoritative_balance_reader", None)
|
||||
if callable(authoritative_reader):
|
||||
return authoritative_reader(), source
|
||||
return broker, source
|
||||
|
||||
def refresh_day_start_credential(self) -> BrokerSource:
|
||||
# Paper/env/break-glass paths deliberately do no portfolio-hub call.
|
||||
if self._config.broker != "kis" or not self._config.credential_fetch_enabled or _break_glass_enabled():
|
||||
|
||||
+166
-6
@@ -11,6 +11,9 @@ Subcommands:
|
||||
cancel <slot_id> — cancel a pending order (control RPC; safeguarded)
|
||||
sync — sync slot qty to broker ground truth (direct-DB; REQ-105)
|
||||
verify — re-check C-case reconcile-warning (direct-DB; REQ-106)
|
||||
unmanaged list — query durable unmanaged-position lifecycle state
|
||||
unmanaged audit — inspect lifecycle/operator audit trail for one symbol
|
||||
unmanaged record — record explicit operator disposition + evidence reference
|
||||
ledger backfill — evidence-backed historical ledger recovery
|
||||
ledger skips list — query durable skipped-ledger records
|
||||
ledger skips resolve — explicitly resolve a skipped-ledger record
|
||||
@@ -36,6 +39,7 @@ LedgerWriter path, and resolves matching durable skipped-ledger records.
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import getpass
|
||||
import json as _json
|
||||
import logging
|
||||
import re
|
||||
@@ -76,6 +80,14 @@ def _resolve_instance(args) -> str:
|
||||
return topology.DEFAULT_INSTANCE_KEY
|
||||
|
||||
|
||||
def _operator_actor() -> str:
|
||||
try:
|
||||
username = getpass.getuser().strip()
|
||||
except Exception: # noqa: BLE001
|
||||
username = ""
|
||||
return f"cli:{username or 'unknown'}"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# default transport-backed RPC (overridable in tests)
|
||||
# --------------------------------------------------------------------------- #
|
||||
@@ -640,6 +652,43 @@ def _build_parser() -> argparse.ArgumentParser:
|
||||
)
|
||||
_add_common(p_verify)
|
||||
|
||||
p_unmanaged = sub.add_parser(
|
||||
"unmanaged",
|
||||
help="List or annotate durable unmanaged broker positions (direct-DB)",
|
||||
)
|
||||
p_unmanaged_sub = p_unmanaged.add_subparsers(dest="unmanaged_cmd", required=True)
|
||||
p_unmanaged_list = p_unmanaged_sub.add_parser(
|
||||
"list",
|
||||
help="List unmanaged broker positions with lifecycle state",
|
||||
)
|
||||
p_unmanaged_list.add_argument(
|
||||
"--status",
|
||||
choices=("active", "resolved", "all"),
|
||||
default="active",
|
||||
)
|
||||
p_unmanaged_list.add_argument("--symbol", default=None)
|
||||
_add_common(p_unmanaged_list)
|
||||
|
||||
p_unmanaged_audit = p_unmanaged_sub.add_parser(
|
||||
"audit",
|
||||
help="Show lifecycle/operator audit trail for one unmanaged position",
|
||||
)
|
||||
p_unmanaged_audit.add_argument("symbol")
|
||||
_add_common(p_unmanaged_audit)
|
||||
|
||||
p_unmanaged_record = p_unmanaged_sub.add_parser(
|
||||
"record",
|
||||
help="Record an explicit operator disposition for an unmanaged position",
|
||||
)
|
||||
p_unmanaged_record.add_argument("symbol")
|
||||
p_unmanaged_record.add_argument(
|
||||
"action",
|
||||
choices=("investigating", "approved_exception", "managed_inclusion"),
|
||||
)
|
||||
p_unmanaged_record.add_argument("--evidence-ref", required=True)
|
||||
p_unmanaged_record.add_argument("--note", default="")
|
||||
_add_common(p_unmanaged_record)
|
||||
|
||||
p_ledger = sub.add_parser(
|
||||
"ledger",
|
||||
help="Ledger operator tools (direct-DB enqueue + immediate writer path)",
|
||||
@@ -704,12 +753,12 @@ def _build_parser() -> argparse.ArgumentParser:
|
||||
def _cmd_sync(args, cfg, confirm) -> int:
|
||||
"""REQ-105 (dcy-034, BAT-051): sync slot qty to broker ground truth."""
|
||||
|
||||
from trex_core.adapters.portfolio.reconcile_warning_store import (
|
||||
PostgreSQLReconcileWarningStore,
|
||||
)
|
||||
from trex_core.adapters.portfolio.reconcile_notification_store import (
|
||||
PostgreSQLReconcileNotificationStore,
|
||||
)
|
||||
from trex_core.adapters.portfolio.reconcile_warning_store import (
|
||||
PostgreSQLReconcileWarningStore,
|
||||
)
|
||||
from trex_core.application.reconciliation import sync_slot_to_broker
|
||||
|
||||
if not args.portfolio_id and not args.all:
|
||||
@@ -858,12 +907,12 @@ def _cmd_sync(args, cfg, confirm) -> int:
|
||||
def _cmd_verify(args, cfg) -> int:
|
||||
"""REQ-106 (dcy-034, BAT-051): re-check C-case reconcile-warning (read-centric)."""
|
||||
from trex_core.adapters.persistence.postgresql import PostgreSQLSlotRepository
|
||||
from trex_core.adapters.portfolio.reconcile_warning_store import (
|
||||
PostgreSQLReconcileWarningStore,
|
||||
)
|
||||
from trex_core.adapters.portfolio.reconcile_notification_store import (
|
||||
PostgreSQLReconcileNotificationStore,
|
||||
)
|
||||
from trex_core.adapters.portfolio.reconcile_warning_store import (
|
||||
PostgreSQLReconcileWarningStore,
|
||||
)
|
||||
from trex_core.application.portfolio import SnapshotStore
|
||||
from trex_core.application.reconciliation import verify_slot_reconcile
|
||||
|
||||
@@ -979,6 +1028,111 @@ def _cmd_verify(args, cfg) -> int:
|
||||
return exit_code
|
||||
|
||||
|
||||
def _cmd_unmanaged_list(args, cfg) -> int:
|
||||
from trex_core.adapters.portfolio.unmanaged_position_store import (
|
||||
PostgreSQLUnmanagedPositionStore,
|
||||
)
|
||||
|
||||
instance = _resolve_instance(args)
|
||||
states = PostgreSQLUnmanagedPositionStore(cfg.dsn).list_states(
|
||||
instance_key=instance,
|
||||
lifecycle_state=args.status,
|
||||
symbol=args.symbol,
|
||||
)
|
||||
payload = {
|
||||
"instance_key": instance,
|
||||
"status": args.status,
|
||||
"count": len(states),
|
||||
"states": states,
|
||||
}
|
||||
if args.json:
|
||||
_emit(payload, as_json=True)
|
||||
else:
|
||||
print(
|
||||
f"instance={instance} status={args.status} count={len(states)} "
|
||||
f"symbol={args.symbol or '*'}"
|
||||
)
|
||||
for state in states:
|
||||
print(
|
||||
f" {state['symbol']}\tqty={state['quantity']}\t"
|
||||
f"lifecycle={state['lifecycle_state']}\t"
|
||||
f"first_seen={state['first_seen_at']}\tlast_seen={state['last_seen_at']}\t"
|
||||
f"resolved_at={state['resolved_at']}\t"
|
||||
f"operator_action={state.get('operator_action') or '-'}\t"
|
||||
f"evidence_ref={state.get('evidence_ref') or '-'}"
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_unmanaged_audit(args, cfg) -> int:
|
||||
from trex_core.adapters.portfolio.unmanaged_position_store import (
|
||||
PostgreSQLUnmanagedPositionStore,
|
||||
)
|
||||
|
||||
instance = _resolve_instance(args)
|
||||
store = PostgreSQLUnmanagedPositionStore(cfg.dsn)
|
||||
state = store.get_state(instance_key=instance, symbol=args.symbol)
|
||||
if state is None:
|
||||
_fail(
|
||||
f"unmanaged position not found: instance={instance} symbol={args.symbol}",
|
||||
args.json,
|
||||
)
|
||||
return 1
|
||||
audit = store.list_audit(instance_key=instance, symbol=args.symbol)
|
||||
payload = {
|
||||
"instance_key": instance,
|
||||
"symbol": args.symbol,
|
||||
"state": state,
|
||||
"audit": audit,
|
||||
}
|
||||
if args.json:
|
||||
_emit(payload, as_json=True)
|
||||
else:
|
||||
print(
|
||||
f"instance={instance} symbol={args.symbol} lifecycle={state['lifecycle_state']} "
|
||||
f"entries={len(audit)}"
|
||||
)
|
||||
for row in audit:
|
||||
print(
|
||||
f" {row['event_type']}\tat={row['recorded_at']}\t"
|
||||
f"quantity={row.get('quantity')}\taction={row.get('action') or '-'}\t"
|
||||
f"actor={row.get('actor') or '-'}\tevidence_ref={row.get('evidence_ref') or '-'}\t"
|
||||
f"note={row.get('note') or '-'}"
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_unmanaged_record(args, cfg) -> int:
|
||||
from trex_core.adapters.portfolio.unmanaged_position_store import (
|
||||
PostgreSQLUnmanagedPositionStore,
|
||||
)
|
||||
|
||||
instance = _resolve_instance(args)
|
||||
state = PostgreSQLUnmanagedPositionStore(cfg.dsn).record_operator_action(
|
||||
instance_key=instance,
|
||||
symbol=args.symbol,
|
||||
action=args.action,
|
||||
evidence_ref=args.evidence_ref,
|
||||
actor=_operator_actor(),
|
||||
note=args.note,
|
||||
)
|
||||
if state is None:
|
||||
_fail(
|
||||
f"unmanaged position not found: instance={instance} symbol={args.symbol}",
|
||||
args.json,
|
||||
)
|
||||
return 1
|
||||
payload = {"result": "recorded", "state": state}
|
||||
if args.json:
|
||||
_emit(payload, as_json=True)
|
||||
else:
|
||||
print(
|
||||
f"recorded unmanaged position instance={instance} symbol={args.symbol} "
|
||||
f"action={args.action} evidence_ref={args.evidence_ref}"
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_report(args, cfg) -> int:
|
||||
"""REQ-088/110/111: holdings / today's fills / WAC realized PnL / pending — direct-DB."""
|
||||
import psycopg
|
||||
@@ -1442,6 +1596,12 @@ def main(argv: list[str] | None = None, *, rpc=None, confirm=None) -> int:
|
||||
# REQ-106 (BAT-051): verify — direct-DB, C-case re-check (read-centric).
|
||||
if args.cmd == "verify":
|
||||
return _cmd_verify(args, cfg)
|
||||
if args.cmd == "unmanaged" and getattr(args, "unmanaged_cmd", None) == "list":
|
||||
return _cmd_unmanaged_list(args, cfg)
|
||||
if args.cmd == "unmanaged" and getattr(args, "unmanaged_cmd", None) == "audit":
|
||||
return _cmd_unmanaged_audit(args, cfg)
|
||||
if args.cmd == "unmanaged" and getattr(args, "unmanaged_cmd", None) == "record":
|
||||
return _cmd_unmanaged_record(args, cfg)
|
||||
|
||||
if args.cmd == "ledger" and getattr(args, "ledger_cmd", None) == "backfill":
|
||||
return _cmd_ledger_backfill(args, cfg, confirm)
|
||||
|
||||
+20
-12
@@ -31,11 +31,11 @@ from trex_core.adapters.messaging import (
|
||||
RabbitMQTransport,
|
||||
topology,
|
||||
)
|
||||
from trex_core.adapters.persistence.migrate import apply_migrations
|
||||
from trex_core.adapters.persistence.account_snapshot import (
|
||||
PostgreSQLAccountSnapshotStore,
|
||||
)
|
||||
from trex_core.adapters.persistence.ledger_skip_store import PostgreSQLLedgerSkipStore
|
||||
from trex_core.adapters.persistence.migrate import apply_migrations
|
||||
from trex_core.adapters.persistence.postgresql import (
|
||||
PgStrategyStateStore,
|
||||
PostgreSQLIntentStore,
|
||||
@@ -52,16 +52,18 @@ from trex_core.adapters.portfolio.killswitch_store import PostgreSQLKillSwitchSt
|
||||
from trex_core.adapters.portfolio.reconcile_notification_store import (
|
||||
PostgreSQLReconcileNotificationStore,
|
||||
)
|
||||
from trex_core.application.alarm_emitter import AlarmEmitter
|
||||
from trex_core.adapters.portfolio.unmanaged_position_store import (
|
||||
PostgreSQLUnmanagedPositionStore,
|
||||
)
|
||||
from trex_core.application.account_snapshot import AccountSnapshotService
|
||||
from trex_core.application.alarm_emitter import AlarmEmitter
|
||||
from trex_core.application.broker_port import BrokerPort
|
||||
from trex_core.application.execution import ExecutionService
|
||||
from trex_core.application.intents import IntentService
|
||||
from trex_core.application.intents import BuyIntent
|
||||
from trex_core.application.intents import BuyIntent, IntentService
|
||||
from trex_core.application.killswitch import KillSwitchService
|
||||
from trex_core.application.ledger_outbox import LedgerWriter
|
||||
from trex_core.application.ledger_port import LedgerPort
|
||||
from trex_core.application.messaging import MessagingTransport
|
||||
from trex_core.application.killswitch import KillSwitchService
|
||||
from trex_core.application.messaging_handlers import (
|
||||
make_control_rpc_handler,
|
||||
make_intent_rpc_handler,
|
||||
@@ -202,6 +204,7 @@ class Daemon:
|
||||
self.warn_store = PostgreSQLReconcileWarningStore(config.dsn)
|
||||
# req-0007 / ac-0024: durable reconcile notification deduplication store.
|
||||
self.notification_store = PostgreSQLReconcileNotificationStore(config.dsn)
|
||||
self.unmanaged_position_store = PostgreSQLUnmanagedPositionStore(config.dsn)
|
||||
# BAT-012: day-scoped kill-switch exclusion store + service.
|
||||
self.killswitch_store = PostgreSQLKillSwitchStore(config.dsn)
|
||||
self.portfolio_read = _build_portfolio_read(config)
|
||||
@@ -210,6 +213,9 @@ class Daemon:
|
||||
from trex_core.runtime.broker_provider import build_broker_provider
|
||||
self.broker_provider = build_broker_provider(config, read_port=self.portfolio_read)
|
||||
self.broker, self.broker_source = self.broker_provider.get_broker_for_read()
|
||||
self.reconciliation_broker, self.reconciliation_broker_source = (
|
||||
self.broker_provider.get_broker_for_reconciliation()
|
||||
)
|
||||
self.account_snapshot_service: AccountSnapshotService | None = None
|
||||
if config.account_snapshot.enabled:
|
||||
if config.broker != "kis":
|
||||
@@ -268,6 +274,7 @@ class Daemon:
|
||||
read_port=self.portfolio_read,
|
||||
store=self.snapshot_store,
|
||||
broker=self.broker,
|
||||
core_holdings_broker=self.reconciliation_broker,
|
||||
schedule_kst=_parse_kst_time(config.snapshot_schedule.kst_time),
|
||||
# dcy-020 / REQ-058·062 (BAT-030): universe slot reconcile (creates/
|
||||
# retires the permanent asset slots) + core-holding divergence
|
||||
@@ -278,6 +285,7 @@ class Daemon:
|
||||
alarm_emitter=self.alarm_emitter, # dcy-033 / REQ-100
|
||||
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,
|
||||
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.
|
||||
@@ -421,7 +429,7 @@ class Daemon:
|
||||
while not self.stop.wait(self.config.ledger_interval_sec):
|
||||
try:
|
||||
self.ledger_writer.run_once()
|
||||
except Exception: # noqa: BLE001
|
||||
except Exception: # noqa: BLE001,S112
|
||||
continue
|
||||
|
||||
def _intent_rpc_loop(self) -> None:
|
||||
@@ -429,7 +437,7 @@ class Daemon:
|
||||
self._transports.append(transport)
|
||||
try:
|
||||
transport.declare_topology(self.config.instance_key)
|
||||
except Exception: # noqa: BLE001
|
||||
except Exception: # noqa: BLE001,S110
|
||||
pass
|
||||
handler = make_intent_rpc_handler(
|
||||
self.intent_service,
|
||||
@@ -444,7 +452,7 @@ class Daemon:
|
||||
self._transports.append(transport)
|
||||
try:
|
||||
transport.declare_topology(self.config.instance_key)
|
||||
except Exception: # noqa: BLE001
|
||||
except Exception: # noqa: BLE001,S110
|
||||
pass
|
||||
handler = make_slots_rpc_handler(
|
||||
self.slot_repo,
|
||||
@@ -468,7 +476,7 @@ class Daemon:
|
||||
self._transports.append(transport)
|
||||
try:
|
||||
transport.declare_topology(self.config.instance_key)
|
||||
except Exception: # noqa: BLE001
|
||||
except Exception: # noqa: BLE001,S110
|
||||
pass
|
||||
handler = make_strategy_state_set_handler(
|
||||
self.slot_repo, self.strategy_state_store,
|
||||
@@ -482,7 +490,7 @@ class Daemon:
|
||||
self._transports.append(transport)
|
||||
try:
|
||||
transport.declare_topology(self.config.instance_key)
|
||||
except Exception: # noqa: BLE001
|
||||
except Exception: # noqa: BLE001,S110
|
||||
pass
|
||||
handler = make_control_rpc_handler(
|
||||
self.pause_store,
|
||||
@@ -497,7 +505,7 @@ class Daemon:
|
||||
while not self.stop.wait(self.safety_sell.base_tick_sec):
|
||||
try:
|
||||
self.safety_sell.run_once()
|
||||
except Exception: # noqa: BLE001 - loop must survive transient errors
|
||||
except Exception: # noqa: BLE001,S112 - loop must survive transient errors
|
||||
continue
|
||||
|
||||
def _snapshot_scheduler_loop(self) -> None:
|
||||
@@ -562,7 +570,7 @@ class Daemon:
|
||||
for tr in self._transports:
|
||||
try:
|
||||
tr.close()
|
||||
except Exception: # noqa: BLE001
|
||||
except Exception: # noqa: BLE001,S110
|
||||
pass
|
||||
|
||||
def run(self) -> None: # pragma: no cover - process entrypoint
|
||||
|
||||
Reference in New Issue
Block a user