From 591db7adb77304e66a0dd8ecaef52340da222866 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 13 Sep 2026 02:21:37 +0000 Subject: [PATCH] Fix unmanaged reconciliation review blockers Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- .proofline/criteria/ac-0029.md | 18 ++++ .proofline/identities.json | 4 +- .proofline/lines/line-0010/dcy-0010.md | 26 +++++ .proofline/lines/line-0010/line-0010.md | 3 + .proofline/lines/line-0010/req-0010.md | 40 +++++++ docs/interface-contract.md | 7 +- docs/operations.md | 1 + docs/runtime-flows.md | 3 +- .../persistence/test_migration_and_drift.py | 102 ++++++++++-------- .../test_unmanaged_position_store.py | 12 ++- tests/application/test_snapshot_scheduler.py | 45 ++++++-- .../test_unmanaged_position_lifecycle.py | 39 ++++++- tests/runtime/test_broker_provider.py | 21 ++++ tests/runtime/test_cli_operator.py | 88 ++++++++++++++- tests/runtime/test_daemon_resolve_symbol.py | 41 +++++++ trex_core/adapters/broker/kis.py | 7 +- trex_core/adapters/broker/ownership.py | 4 + trex_core/adapters/broker/paper.py | 3 + ...> 030_create_unmanaged_position_state.sql} | 0 .../portfolio/unmanaged_position_store.py | 2 + trex_core/application/portfolio.py | 5 - trex_core/application/reconciliation.py | 2 +- trex_core/application/snapshot_scheduler.py | 4 +- trex_core/runtime/broker_provider.py | 16 ++- trex_core/runtime/cli.py | 60 ++++++++++- trex_core/runtime/daemon.py | 18 ++-- 26 files changed, 484 insertions(+), 87 deletions(-) create mode 100644 .proofline/criteria/ac-0029.md create mode 100644 .proofline/lines/line-0010/dcy-0010.md create mode 100644 .proofline/lines/line-0010/line-0010.md create mode 100644 .proofline/lines/line-0010/req-0010.md rename trex_core/adapters/persistence/sql/{028_create_unmanaged_position_state.sql => 030_create_unmanaged_position_state.sql} (100%) diff --git a/.proofline/criteria/ac-0029.md b/.proofline/criteria/ac-0029.md new file mode 100644 index 0000000..af4cfa9 --- /dev/null +++ b/.proofline/criteria/ac-0029.md @@ -0,0 +1,18 @@ +--- +id: "ac-0029" +status: active +--- + +# Durable unmanaged-position lifecycle, authority boundary, and operator audit + +## Criterion + +`trex-core`는 broker-only unmanaged holding을 `(instance_key, symbol)` keyed durable lifecycle state로 추적해야 한다. State는 `quantity`, `first_seen_at`, `last_seen_at`, `quantity_changed_at`, `resolved_at`, lifecycle state와 latest non-secret operator disposition/evidence를 제공해야 한다. `reconcile_core_holdings()`는 authoritative full-balance snapshot에서만 unresolved active rows의 absence를 `resolved`로 전이할 수 있으며, marker가 없거나 projected/partial balance인 broker reader는 resolve authority가 없다. 동일 active lifecycle에서 같은 수량 반복 관측은 새 lifecycle alarm/audit를 만들지 않고 `last_seen_at`만 갱신해야 한다. 운영자 disposition은 explicit evidence-backed command로만 기록되고 audit trail에 append되어야 하며, current state 조회와 audit 조회 모두 inspectable해야 한다. 이 기능은 broker 주문, slot/ledger mutation, warning 자동 삭제 또는 manual-holding cause inference를 수행해서는 안 된다. + +## Verification + +- Identity/lifecycle: 같은 symbol을 서로 다른 `instance_key`에서 독립 row로 저장하고, first detect → same-qty repeat → quantity change → authoritative absence resolved → re-detect reopen 순서를 InMemory/PostgreSQL parity tests로 검증한다. +- Authority boundary: missing marker, `False`, exception-throwing marker, ownership/projected broker, broker read failure 각각에서 active unmanaged row가 resolved 되지 않음을 검증한다. Explicit authoritative broker만 absence resolution을 수행해야 한다. +- Runtime wiring: production broker provider/daemon/snapshot scheduler tests에서 주문용 projected broker와 unmanaged reconciliation용 authoritative full-balance reader가 분리 주입되는지 고정한다. +- Operator audit: CLI/parser/store tests에서 `unmanaged list`, evidence-required `unmanaged record`, symbol별 `unmanaged audit`를 검증하고 audit actor가 caller-supplied arbitrary string이 아니라 trusted local identity에서 결정됨을 확인한다. +- Safety boundary: any unmanaged lifecycle path에서 broker order submit/cancel, slot create/mutate, synthetic ledger write, warning auto-clear가 발생하지 않음을 existing regression surface와 focused tests로 유지한다. diff --git a/.proofline/identities.json b/.proofline/identities.json index 34f4cf3..2c78cba 100644 --- a/.proofline/identities.json +++ b/.proofline/identities.json @@ -1,5 +1,5 @@ { "schema_version": 1, - "next_line_number": 10, - "next_ac_number": 29 + "next_line_number": 11, + "next_ac_number": 30 } diff --git a/.proofline/lines/line-0010/dcy-0010.md b/.proofline/lines/line-0010/dcy-0010.md new file mode 100644 index 0000000..861bcdb --- /dev/null +++ b/.proofline/lines/line-0010/dcy-0010.md @@ -0,0 +1,26 @@ +--- +id: "dcy-0010" +status: confirmed +--- + +# Durable unmanaged-position lifecycle + +## Problem + +현재 `reconcile_core_holdings()`는 broker 계정에만 존재하고 대상 `instance_key`의 managed slot·ledger에 없는 종목을 `unmanaged_position` warning/log로만 표면화한다. 이 상태는 매 주기 반복 관측되어도 최초 감지, 수량 변경, 해제, 운영자 triage 여부를 durable하게 구분하지 못한다. 또한 ownership projection이 켜진 KIS 경로에서는 전략용 projected balance가 manual/타 instance 보유를 제거하므로, reconciliation이 실제 broker-only unmanaged position을 아예 보지 못할 수 있다. + +## Evidence + +- GitHub issue [#85](https://github.com/fourblocks2020/trex-core/issues/85)는 unmanaged identity를 `(instance_key, symbol)`로 격리하고, 최초 감지·수량 변경·해제와 마지막 broker 관측 시각을 durable하게 보존하며, 같은 수량 반복 관측의 로그/알림 폭주를 억제할 것을 요구한다. +- Issue 본문은 운영자가 broker evidence와 함께 managed inclusion, approved exception, investigating disposition을 기록할 수 있어야 하고, broker 조회 실패 또는 partial/projected balance 부재를 quantity 0이나 resolved로 해석해서는 안 된다고 명시한다. +- 현재 ownership wrapper `get_balance()`는 strategy ownership projection만 반환하므로 unmanaged/manual holdings를 숨긴다. 따라서 production wiring은 주문용 projected broker와 별도로 authoritative full-balance reader를 reconciliation에 주입해야 한다. +- 현행 문서 `docs/interface-contract.md`와 `docs/runtime-flows.md`는 broker-only 분류가 persistence나 DB schema를 바꾸지 않는다고 적고 있어, 새 durable lifecycle contract를 소유하는 승인된 AC가 필요하다. + +## Scope + +- 이 Line은 broker-only `unmanaged_position`을 `(instance_key, symbol)` keyed durable lifecycle state로 저장하고, `quantity`, `first_seen_at`, `last_seen_at`, `quantity_changed_at`, `resolved_at`, lifecycle state 및 non-secret operator evidence를 조회 가능하게 하는 contract를 다룬다. +- Reconciliation은 authoritative full-balance snapshot에서만 부재 해제를 수행한다. Marker가 없거나 projected/partial balance인 reader는 detect/update에는 사용할 수 있어도 resolve authority를 갖지 않는다. +- Ownership-aware trading broker는 그대로 주문/전략 조회에 사용하되, unmanaged reconciliation 입력은 별도의 authoritative balance reader를 사용한다. +- 동일 active lifecycle에서 반복 같은 수량 관측은 `last_seen_at`만 갱신하고 새 detect/change alarm·audit를 만들지 않는다. Resolved 후 재탐지되면 새 active lifecycle로 reopen되며 이전 operator disposition 현재값은 carry-over 하지 않는다. +- 운영자 disposition은 explicit command와 evidence reference를 요구하고 audit trail에 append된다. CLI는 current state와 audit trail을 둘 다 조회 가능해야 한다. +- 이 Line은 broker 주문, slot 생성, synthetic ledger backfill, warning 자동 삭제 또는 manual holdings의 원인 추정을 도입하지 않는다. diff --git a/.proofline/lines/line-0010/line-0010.md b/.proofline/lines/line-0010/line-0010.md new file mode 100644 index 0000000..70331b6 --- /dev/null +++ b/.proofline/lines/line-0010/line-0010.md @@ -0,0 +1,3 @@ +--- +id: "line-0010" +--- diff --git a/.proofline/lines/line-0010/req-0010.md b/.proofline/lines/line-0010/req-0010.md new file mode 100644 index 0000000..71a8c19 --- /dev/null +++ b/.proofline/lines/line-0010/req-0010.md @@ -0,0 +1,40 @@ +--- +id: "req-0010" +status: approved +discovery: "dcy-0010" +criteria: + create: + - "ac-0029" + update: [] + retire: [] + satisfy: [] +--- + +# Durable unmanaged-position lifecycle + +## Objective + +`trex-core`는 broker 계정에 존재하지만 대상 `instance_key`의 managed slot·ledger가 소유하지 않는 종목을 stateless warning이 아니라 durable unmanaged-position lifecycle로 추적해야 한다. Lifecycle은 authoritative broker balance를 근거로 `(instance_key, symbol)`별 최초 감지, 수량 변경, 마지막 관측, 해제, 운영자 disposition과 audit를 구분해야 하며, 반복 같은 수량 관측의 로그/알림 폭주를 억제해야 한다. + +## Scope + +- Unmanaged identity는 `(instance_key, symbol)`이며 다른 instance/account의 같은 symbol과 섞이지 않는다. +- Persistent state는 최소 `symbol`, `quantity`, `first_seen_at`, `last_seen_at`, `quantity_changed_at`, `resolved_at`, lifecycle state와 latest non-secret operator evidence/disposition을 제공한다. +- `reconcile_core_holdings()`는 authoritative broker snapshot에서 broker-only holding을 처음 보면 `detected`, 수량이 바뀌면 `quantity_changed`, 이전 active row가 authoritative snapshot에서 사라지면 `resolved`로 기록한다. 같은 수량 반복 관측은 state의 `last_seen_at`만 갱신한다. +- Broker 조회 실패, missing authoritative marker, projected ownership balance, filtered/partial reader 또는 self-report failure는 resolved authority가 아니다. 이런 입력 부재를 quantity 0이나 resolved로 해석하지 않는다. +- Runtime wiring은 주문/전략용 broker projection과 unmanaged reconciliation용 authoritative full-balance reader를 분리한다. +- 운영자 disposition은 explicit command와 `evidence_ref`를 요구하고 audit trail에 append한다. Caller-supplied arbitrary actor string은 현재 계약에 포함하지 않고 trusted local CLI identity만 기록한다. +- Resolved row가 다시 나타나면 `first_seen_at`은 유지하되 새 active lifecycle로 reopen한다. Audit history는 남기고 current operator disposition/evidence는 reset한다. +- Query surface는 current state list와 symbol별 audit trail을 모두 제공한다. + +## Constraints + +- 이 기능은 slot 생성, slot 상태 변경, warning 자동 삭제, synthetic BUY ledger, broker order submit/cancel 또는 manual holdings 원인 추정을 수행하지 않는다. +- `managed_non_holding_divergence`와 기존 reconcile warning lifecycle·clear authority는 유지한다. +- Normative Design Documents: 없음. 이 REQ와 AC-0029가 normative specification을 소유한다. + +## Non-Goals + +- Broker statement ingestion, manual holding 원인 판정, 자동 편입/자동 정정 +- AMQP/portfolio-hub external wire 변경 +- Ledger remediation, slot lifecycle 재설계, ownership model 변경 diff --git a/docs/interface-contract.md b/docs/interface-contract.md index 05b00df..c44325b 100644 --- a/docs/interface-contract.md +++ b/docs/interface-contract.md @@ -437,8 +437,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 `, `trex-core unmanaged record --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) diff --git a/docs/operations.md b/docs/operations.md index 7cdcd2d..c27010e 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -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 `로 lifecycle/operator audit trail을 확인한 뒤 `trex-core unmanaged record --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) diff --git a/docs/runtime-flows.md b/docs/runtime-flows.md index 68b97cd..42c6da2 100644 --- a/docs/runtime-flows.md +++ b/docs/runtime-flows.md @@ -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, req-0010/ac-0029): 브로커 잔고 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) diff --git a/tests/adapters/persistence/test_migration_and_drift.py b/tests/adapters/persistence/test_migration_and_drift.py index 3a35cd3..20f3f1d 100644 --- a/tests/adapters/persistence/test_migration_and_drift.py +++ b/tests/adapters/persistence/test_migration_and_drift.py @@ -6,9 +6,10 @@ iteration: EXPECTED_SLOT_FIELDS (code contract) vs daily_slot actual columns """ from __future__ import annotations +from datetime import UTC, datetime + import psycopg import pytest -from datetime import datetime, timezone from trex_core.adapters.persistence.migrate import apply_migrations, migration_files from trex_core.adapters.persistence.postgresql import ( @@ -29,28 +30,28 @@ _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", # req-0010 / ac-0029 + "unmanaged_position_audit", # req-0010 / ac-0029 } def _columns(dsn: str, table: str) -> set[str]: - with psycopg.connect(dsn) as conn: - with conn.cursor() as cur: - cur.execute( - "SELECT column_name FROM information_schema.columns " - "WHERE table_schema = 'public' AND table_name = %s", - (table,), - ) - return {r[0] for r in cur.fetchall()} + with psycopg.connect(dsn) as conn, conn.cursor() as cur: + cur.execute( + "SELECT column_name FROM information_schema.columns " + "WHERE table_schema = 'public' AND table_name = %s", + (table,), + ) + return {r[0] for r in cur.fetchall()} def test_migrations_create_expected_tables(pg_clean): assert migration_files(), "no migration files found" - with psycopg.connect(pg_clean) as conn: - with conn.cursor() as cur: - cur.execute( - "SELECT table_name FROM information_schema.tables WHERE table_schema='public'" - ) - tables = {r[0] for r in cur.fetchall()} + with psycopg.connect(pg_clean) as conn, conn.cursor() as cur: + cur.execute( + "SELECT table_name FROM information_schema.tables WHERE table_schema='public'" + ) + tables = {r[0] for r in cur.fetchall()} assert _EXPECTED_TABLES <= tables @@ -61,20 +62,19 @@ def test_daily_slot_columns_match_expected_contract(pg_clean): def test_strategy_state_schema_matches_instance_map_contract(pg_clean): - with psycopg.connect(pg_clean) as conn: - with conn.cursor() as cur: - cur.execute( - "SELECT column_name, data_type, udt_name " - "FROM information_schema.columns " - "WHERE table_schema = 'public' AND table_name = 'slot_strategy_state'" - ) - columns = {row[0]: (row[1], row[2]) for row in cur.fetchall()} - cur.execute( - "SELECT a.attname FROM pg_index i " - "JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey) " - "WHERE i.indrelid = 'slot_strategy_state'::regclass AND i.indisprimary" - ) - primary_key = {row[0] for row in cur.fetchall()} + with psycopg.connect(pg_clean) as conn, conn.cursor() as cur: + cur.execute( + "SELECT column_name, data_type, udt_name " + "FROM information_schema.columns " + "WHERE table_schema = 'public' AND table_name = 'slot_strategy_state'" + ) + columns = {row[0]: (row[1], row[2]) for row in cur.fetchall()} + cur.execute( + "SELECT a.attname FROM pg_index i " + "JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey) " + "WHERE i.indrelid = 'slot_strategy_state'::regclass AND i.indisprimary" + ) + primary_key = {row[0] for row in cur.fetchall()} assert columns == {"slot_id": ("uuid", "uuid"), "data": ("jsonb", "jsonb")} assert primary_key == {"slot_id"} @@ -92,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"}) @@ -111,13 +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: - with 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): @@ -129,7 +139,7 @@ def test_migration_024_separates_legacy_sor_and_is_replay_safe(pg_clean): side="buy", qty=1, price=1000, - executed_at=datetime(2026, 8, 19, tzinfo=timezone.utc), + executed_at=datetime(2026, 8, 19, tzinfo=UTC), venue="NXT", order_venue="KRX", ) @@ -140,7 +150,7 @@ def test_migration_024_separates_legacy_sor_and_is_replay_safe(pg_clean): side="buy", qty=1, price=1000, - executed_at=datetime(2026, 8, 19, tzinfo=timezone.utc), + executed_at=datetime(2026, 8, 19, tzinfo=UTC), venue="KRX", order_venue="SOR", ) @@ -172,10 +182,12 @@ 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: - with 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,), + ) diff --git a/tests/adapters/portfolio/test_unmanaged_position_store.py b/tests/adapters/portfolio/test_unmanaged_position_store.py index 4639bf2..c03a2a8 100644 --- a/tests/adapters/portfolio/test_unmanaged_position_store.py +++ b/tests/adapters/portfolio/test_unmanaged_position_store.py @@ -15,7 +15,7 @@ def pg_unmanaged_store(pg_clean): return PostgreSQLUnmanagedPositionStore(pg_clean) -class TestMigration028: +class TestMigration030: def test_tables_exist_after_migration(self, pg_clean): import psycopg @@ -111,6 +111,14 @@ class TestCRUDRoundTrip: 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(), @@ -126,3 +134,5 @@ class TestCRUDRoundTrip: 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 diff --git a/tests/application/test_snapshot_scheduler.py b/tests/application/test_snapshot_scheduler.py index 6b0c7c2..8b3faba 100644 --- a/tests/application/test_snapshot_scheduler.py +++ b/tests/application/test_snapshot_scheduler.py @@ -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) diff --git a/tests/application/test_unmanaged_position_lifecycle.py b/tests/application/test_unmanaged_position_lifecycle.py index b906e06..ed94ed4 100644 --- a/tests/application/test_unmanaged_position_lifecycle.py +++ b/tests/application/test_unmanaged_position_lifecycle.py @@ -35,6 +35,11 @@ class RaisingBroker(FakeBroker): 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 @@ -116,6 +121,13 @@ def test_in_memory_store_preserves_first_seen_when_resolved_symbol_reappears(): 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) @@ -123,6 +135,8 @@ def test_in_memory_store_preserves_first_seen_when_resolved_symbol_reappears(): 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_reconcile_core_holdings_suppresses_repeated_unmanaged_alarm_spam(): @@ -134,7 +148,7 @@ def test_reconcile_core_holdings_suppresses_repeated_unmanaged_alarm_spam(): decisions = reconcile_core_holdings( slot_repo=repo, slot_service=svc, - broker=FakeBroker({"008110": 3}), + broker=AuthoritativeBroker({"008110": 3}), instance_key=INSTANCE, alarm_emitter=emitter, unmanaged_position_store=store, @@ -166,7 +180,7 @@ def test_reconcile_core_holdings_tracks_change_and_resolution(): reconcile_core_holdings( slot_repo=repo, slot_service=svc, - broker=FakeBroker({"008110": 5}), + broker=AuthoritativeBroker({"008110": 5}), instance_key=INSTANCE, alarm_emitter=emitter, unmanaged_position_store=store, @@ -174,7 +188,7 @@ def test_reconcile_core_holdings_tracks_change_and_resolution(): reconcile_core_holdings( slot_repo=repo, slot_service=svc, - broker=FakeBroker({}), + broker=AuthoritativeBroker({}), instance_key=INSTANCE, alarm_emitter=emitter, unmanaged_position_store=store, @@ -211,6 +225,25 @@ def test_broker_failure_does_not_resolve_unmanaged_state(): 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() diff --git a/tests/runtime/test_broker_provider.py b/tests/runtime/test_broker_provider.py index 2472ec1..2cd654f 100644 --- a/tests/runtime/test_broker_provider.py +++ b/tests/runtime/test_broker_provider.py @@ -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") diff --git a/tests/runtime/test_cli_operator.py b/tests/runtime/test_cli_operator.py index 91d3775..643b4ab 100644 --- a/tests/runtime/test_cli_operator.py +++ b/tests/runtime/test_cli_operator.py @@ -523,12 +523,15 @@ def test_verify_no_db_no_live_flag_needed(capsys, monkeypatch): # unmanaged parser/tests # --------------------------------------------------------------------------- -def test_unmanaged_parser_accepts_list_and_record(): +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", @@ -617,7 +620,6 @@ def test_unmanaged_record_requires_existing_state(monkeypatch, capsys): action="approved_exception", evidence_ref="kis-balance:2026-09-11", note="confirmed", - actor="operator", instance="inst-1", config=None, config_path=None, @@ -634,6 +636,8 @@ def test_unmanaged_record_requires_existing_state(monkeypatch, capsys): 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 @@ -671,7 +675,6 @@ def test_unmanaged_record_persists_operator_action(monkeypatch, capsys): action="managed_inclusion", evidence_ref="kis-balance:2026-09-11", note="manual inclusion planned", - actor="operator", instance="inst-1", config=None, config_path=None, @@ -687,9 +690,12 @@ def test_unmanaged_record_persists_operator_action(monkeypatch, capsys): 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 @@ -724,7 +730,6 @@ def test_unmanaged_record_text_output(monkeypatch, capsys): action="approved_exception", evidence_ref="kis-balance:2026-09-11", note="confirmed", - actor="operator", instance="inst-1", config=None, config_path=None, @@ -742,6 +747,8 @@ def test_unmanaged_record_text_output(monkeypatch, capsys): def test_unmanaged_record_main_dispatch(monkeypatch, capsys): + monkeypatch.setattr(cli, "_operator_actor", lambda: "cli:test") + class _Store: def __init__(self, *_args, **_kwargs): pass @@ -788,6 +795,79 @@ def test_unmanaged_record_main_dispatch(monkeypatch, capsys): 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 # --------------------------------------------------------------------------- diff --git a/tests/runtime/test_daemon_resolve_symbol.py b/tests/runtime/test_daemon_resolve_symbol.py index 140d417..4faba2e 100644 --- a/tests/runtime/test_daemon_resolve_symbol.py +++ b/tests/runtime/test_daemon_resolve_symbol.py @@ -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=...)" + ) diff --git a/trex_core/adapters/broker/kis.py b/trex_core/adapters/broker/kis.py index 4c2e711..c480cc2 100644 --- a/trex_core/adapters/broker/kis.py +++ b/trex_core/adapters/broker/kis.py @@ -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`). diff --git a/trex_core/adapters/broker/ownership.py b/trex_core/adapters/broker/ownership.py index ab3cc0d..8ede722 100644 --- a/trex_core/adapters/broker/ownership.py +++ b/trex_core/adapters/broker/ownership.py @@ -17,6 +17,7 @@ from threading import Lock from trex_core.application.broker_port import ( BalanceItem, BrokerBalanceError, + BrokerPort, CancelOutcome, FillStatus, OrderAck, @@ -174,6 +175,9 @@ 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 diff --git a/trex_core/adapters/broker/paper.py b/trex_core/adapters/broker/paper.py index bcf9683..d7ea95a 100644 --- a/trex_core/adapters/broker/paper.py +++ b/trex_core/adapters/broker/paper.py @@ -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).""" diff --git a/trex_core/adapters/persistence/sql/028_create_unmanaged_position_state.sql b/trex_core/adapters/persistence/sql/030_create_unmanaged_position_state.sql similarity index 100% rename from trex_core/adapters/persistence/sql/028_create_unmanaged_position_state.sql rename to trex_core/adapters/persistence/sql/030_create_unmanaged_position_state.sql diff --git a/trex_core/adapters/portfolio/unmanaged_position_store.py b/trex_core/adapters/portfolio/unmanaged_position_store.py index 0d15adf..291842b 100644 --- a/trex_core/adapters/portfolio/unmanaged_position_store.py +++ b/trex_core/adapters/portfolio/unmanaged_position_store.py @@ -59,6 +59,8 @@ class PostgreSQLUnmanagedPositionStore: "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 " f"RETURNING {_STATE_COLUMNS}", diff --git a/trex_core/application/portfolio.py b/trex_core/application/portfolio.py index ced6b87..3346e7b 100644 --- a/trex_core/application/portfolio.py +++ b/trex_core/application/portfolio.py @@ -615,11 +615,6 @@ class InMemoryUnmanagedPositionStore: existing = self._states.get(key) previous_quantity = existing.get("quantity") if existing is not None else None preserved = {} if existing is None else { - "operator_action": existing.get("operator_action"), - "evidence_ref": existing.get("evidence_ref"), - "operator_note": existing.get("operator_note"), - "operator_actor": existing.get("operator_actor"), - "operator_updated_at": existing.get("operator_updated_at"), "created_at": existing.get("created_at", now), } if existing is None or existing.get("lifecycle_state") == "resolved": diff --git a/trex_core/application/reconciliation.py b/trex_core/application/reconciliation.py index 29be749..9aabe84 100644 --- a/trex_core/application/reconciliation.py +++ b/trex_core/application/reconciliation.py @@ -373,7 +373,7 @@ def _broker_balances_are_authoritative_for_unmanaged_resolution(broker: BrokerPo return False if isinstance(marker, bool): return marker - return True + return False def reconcile_core_holdings( diff --git a/trex_core/application/snapshot_scheduler.py b/trex_core/application/snapshot_scheduler.py index 611a477..da9aef8 100644 --- a/trex_core/application/snapshot_scheduler.py +++ b/trex_core/application/snapshot_scheduler.py @@ -44,6 +44,7 @@ class SnapshotScheduler: read_port: PortfolioHubReadPort, store: SnapshotStore, broker: BrokerPort, + core_holdings_broker: BrokerPort | None = None, schedule_kst: time = time(hour=7, minute=30), clock: Callable[[], datetime] | None = None, slot_repo=None, @@ -60,6 +61,7 @@ class SnapshotScheduler: 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=UTC)) self._daemon_id: str | None = None @@ -337,7 +339,7 @@ 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 diff --git a/trex_core/runtime/broker_provider.py b/trex_core/runtime/broker_provider.py index 51a8ccc..1aa3677 100644 --- a/trex_core/runtime/broker_provider.py +++ b/trex_core/runtime/broker_provider.py @@ -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(): diff --git a/trex_core/runtime/cli.py b/trex_core/runtime/cli.py index bcbe6bc..7caef82 100644 --- a/trex_core/runtime/cli.py +++ b/trex_core/runtime/cli.py @@ -12,6 +12,7 @@ Subcommands: 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 @@ -38,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 @@ -78,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) # --------------------------------------------------------------------------- # @@ -659,6 +669,13 @@ def _build_parser() -> argparse.ArgumentParser: 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", @@ -670,7 +687,6 @@ def _build_parser() -> argparse.ArgumentParser: ) p_unmanaged_record.add_argument("--evidence-ref", required=True) p_unmanaged_record.add_argument("--note", default="") - p_unmanaged_record.add_argument("--actor", default="operator") _add_common(p_unmanaged_record) p_ledger = sub.add_parser( @@ -1040,6 +1056,44 @@ def _cmd_unmanaged_list(args, cfg) -> int: 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, @@ -1051,7 +1105,7 @@ def _cmd_unmanaged_record(args, cfg) -> int: symbol=args.symbol, action=args.action, evidence_ref=args.evidence_ref, - actor=args.actor, + actor=_operator_actor(), note=args.note, ) if state is None: @@ -1536,6 +1590,8 @@ def main(argv: list[str] | None = None, *, rpc=None, confirm=None) -> int: 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) diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index e5f117e..3532427 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -213,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": @@ -271,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 @@ -425,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: @@ -433,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, @@ -448,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, @@ -472,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, @@ -486,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, @@ -501,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: @@ -566,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