diff --git a/config/.env.example b/config/.env.example index 93e03ad..166afe8 100644 --- a/config/.env.example +++ b/config/.env.example @@ -42,6 +42,12 @@ TREX_KIS_BROKER_ACCOUNT_REF=12345678-01 TREX_KIS_ALLOW_LIVE=false TREX_KIS_REDIS_URL= +# --- KIS account-read recovery (dcy-041; opt-in, KIS only) --- +TREX_ACCOUNT_SNAPSHOT_ENABLED=false +TREX_ACCOUNT_SNAPSHOT_BUY_MAX_AGE_SEC=600 +TREX_ACCOUNT_SNAPSHOT_SELL_MAX_AGE_SEC=3600 +TREX_ACCOUNT_SNAPSHOT_REFRESH_INTERVAL_SEC=300 + # --- portfolio-hub ledger (REQ-017/018/019, BAT-006) --- # portfolio_hub_token is per-instance and lives in the instance TOML, not here. TREX_PORTFOLIO_HUB_BASE_URL= diff --git a/docs/data-model.md b/docs/data-model.md index 0501e03..1a9c691 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -168,6 +168,26 @@ - 5xx/network/timeout → outcome `retryable_failure`, status `pending` (다음 tick 재시도). - slot 전이는 outbox 적재와 분리되어 critical path 유지. +## Entity: `broker_account_snapshot` (dcy-041 / GitHub #45) + +- Purpose: KIS 계좌조회 전송 장애 동안 신규매수와 위험축소 매도의 허용 범위를 짧게 제한하는 계정별 마지막 정상값. +- Source of Truth: 실시간 KIS 조회가 항상 우선이며 이 테이블은 임시 복구값이다. 재시작 원장대조에는 사용하지 않는다. + +### Key Fields + +- account_hash :: TEXT PK :: `SHA-256(CANO|ACNT_PRDT_CD)`. 평문 계좌번호를 저장·로그하지 않는다. +- holdings :: JSONB NOT NULL :: `[{symbol,hldg_qty}]`. 소유권 래퍼가 켜진 계정은 전략 소유분만 저장한다. +- holdings_captured_at :: TIMESTAMPTZ? :: 보유수량 조회가 마지막 성공한 시각. +- available_cash :: BIGINT? :: KIS 주문가능조회 `nrcvb_buy_amt` 현금주문 가능금액. +- cash_captured_at :: TIMESTAMPTZ? :: 주문가능금액 조회가 마지막 성공한 시각. + +### State Rules + +- 보유수량과 주문가능금액은 독립적으로 갱신한다. 한 조회의 실패가 다른 정상값과 캡처 시각을 덮지 않는다. +- 계정 해시는 저장 직전 재확인하며 바뀌면 저장하지 않는다. +- PostgreSQL advisory lock으로 계정당 백그라운드 조회 소유자는 하나만 허용한다. +- 기본 최대 나이는 매수 600초, 매도 3600초이고 매도도 실시간 보유조회가 `transport_failed`일 때만 사용한다. + ## Shared Consistency Rules - 모든 상태 변경과 전이 로그 append 는 동일 advisory-locked 트랜잭션 안에서 일어난다. advisory lock 은 인스턴스 스코프(`slot_advisory_lock_id(instance_key)`, REQ-057 NFR-1). diff --git a/docs/discovery/dcy-041_20260729_broker-uncertainty-and-adapter-retirement.md b/docs/discovery/dcy-041_20260729_broker-uncertainty-and-adapter-retirement.md index 656e178..fb1050a 100644 --- a/docs/discovery/dcy-041_20260729_broker-uncertainty-and-adapter-retirement.md +++ b/docs/discovery/dcy-041_20260729_broker-uncertainty-and-adapter-retirement.md @@ -4,7 +4,7 @@ - 문서 ID: dcy-041_20260729_broker-uncertainty-and-adapter-retirement - 이슈명: Core 브로커 불확실 상태 구분과 로컬 KIS 어댑터 축소 - 작성 시각(KST): 2026-07-29 16:59:16 KST -- 마지막 갱신 시각(KST): 2026-07-29 17:01:50 KST +- 마지막 갱신 시각(KST): 2026-07-29 17:32:04 KST - 대체됨: 없음 - 선행 Discovery: - dcy-013_20260608_kis-external-cancel-pending-sell-stuck @@ -60,7 +60,7 @@ - FR-2: 실제 제출 거래소를 슬롯과 Core 체결 전달 대기 원장에 저장하며 취소·재시작 복구에서 재사용한다. - FR-3: 취소 요청은 `requested|rejected|unknown` 결과를 반환한다. `requested`는 최종 취소가 아니며 체결조회가 `cancelled|filled`를 확정한다. - FR-4: `unknown` 취소 결과와 결과불명 주문은 자동 재주문하지 않고 기존 주문 조회로만 수렴한다. -- FR-5: 성공한 계좌조회만 계정 해시·캡처 시각·보유수량·주문가능금액과 함께 원자 저장한다. +- FR-5: 성공한 계좌조회 영역만 계정 해시·독립 캡처 시각과 함께 원자 저장한다. 보유수량과 주문가능금액 중 한쪽 실패가 다른 정상값을 덮지 않는다. - FR-6: 일시적 전송 오류에만 미만료 스냅샷을 사용한다. 계정 불일치·만료·응답 오류는 차단한다. - FR-7: 신규매수와 위험축소 매도의 스냅샷 최대 나이를 분리하며 KIS 복구 후 다음 백그라운드 조회가 정상값으로 교체한다. @@ -127,3 +127,13 @@ - Ready for REQ Drafting: true - Confirmed By: 형님 - Confirmed At (KST): 2026-07-29 17:01:50 KST + +# 12. 구현 증거 + +- `AUTO` 실제 거래소 확정·주문번호 동시 저장·재시작/취소 재사용: 구현 완료. +- 취소 `requested|rejected|unknown` 영속화와 poll 전용 최종확정: 구현 완료. +- 계정 해시·독립 TTL·백그라운드 단일 소유자·매수/매도 분리: 구현 완료. +- KIS 현금 권위 필드: `inquire-psbl-order.output.nrcvb_buy_amt`. +- 수동 혼합 보유 계정: `OwnershipAwareBroker` 투영 뒤 스냅샷을 생성하여 전략 소유 수량만 매도 허용. +- 운영 기본값: 기능 OFF, 매수 600초, 매도 3600초, 갱신 300초. +- 실전 반영: 가상 검증과 사용자 별도 승인 전까지 미적용. diff --git a/docs/interface-contract.md b/docs/interface-contract.md index b441244..5ec34dd 100644 --- a/docs/interface-contract.md +++ b/docs/interface-contract.md @@ -423,11 +423,21 @@ resume_kill_switch 요청·응답 (REQ-046): ### Balance-read failure and reconciliation boundary (BAT-057 / REQ-120/121) -- `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`다. +- `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를 바꾸지 않는다. +### Account snapshot recovery (dcy-041 / GitHub #45) + +- 기능은 `TREX_ACCOUNT_SNAPSHOT_ENABLED=false`가 기본인 KIS 전용 선택 기능이다. +- KIS 보유조회와 현금주문 가능금액 조회(`inquire-psbl-order.output.nrcvb_buy_amt`)를 기본 300초 간격의 단일 계정 소유자가 갱신한다. 두 값은 독립 캡처 시각을 갖는다. +- 신규매수는 미만료 현금 스냅샷과 기존 `pending_buy` 지정가 예약금액을 합산해 검사한다. 수량·지정가가 없거나 예약금액을 계산할 수 없으면 매수를 차단한다. KIS 주문 응답이 최종 현금 한도 권위이며 거부된 주문을 자동 재전송하지 않는다. +- `pending_sell` 수량은 매 tick 실시간 보유조회를 먼저 시도한다. 오직 `transport_failed`이고 기본 3600초 안의 동일계정 스냅샷이 있을 때만 대체한다. +- 인증·인가·응답형식·브로커거부·소유권 불일치·계정 해시 변경·만료는 캐시로 우회하지 않는다. +- `sync`, `verify`, 일일 스냅샷 스케줄러와 재시작 원장대조는 기존처럼 실시간 조회만 사용한다. +- 소유권 래퍼가 설정된 혼합계정은 `broker total = manual protected + strategy owned`를 유지하며 전략 소유분만 보유수량 스냅샷에 들어간다. + ### cancel_order ``` diff --git a/docs/runtime-flows.md b/docs/runtime-flows.md index 6ac694b..98c87b9 100644 --- a/docs/runtime-flows.md +++ b/docs/runtime-flows.md @@ -250,21 +250,29 @@ 데몬 lifecycle 에서 시작되는 단일 백그라운드 execution 루프가 주기적으로(설정 간격) 수행: 1. dispatch: `list_undispatched_pending(instance_key)`(status pending_* AND `broker_order_id IS NULL`; 인스턴스 스코프, REQ-056) → `BrokerPort.submit_order` → `mark_dispatched(broker_order_id)`(멱등 가드). 거부 시 `pending_buy→inactive` / `pending_sell→holding` 롤백. - - **매도 수량 = 브로커 잔고 정본**(dcy-020/REQ-062, D3): 매도 dispatch 는 `broker.get_balance()` 의 해당 symbol `hldg_qty` 로 수량을 정한다(슬롯 미러 `qty` fallback 없음). 브로커 0 이면 주문 미제출 + WARN + `sell_skipped_no_balance` 카운터(divergence 는 reconcile 경로로 표면화). skip 된 매도가 큐를 막지 않도록 리스트 기반 dispatch. + - **매도 수량 = 브로커 잔고 정본**(dcy-020/REQ-062, D3): 매도 dispatch 는 실시간 `broker.get_balance()` 의 해당 symbol `hldg_qty` 로 수량을 정한다(슬롯 미러 `qty` fallback 없음). 계좌 스냅샷 선택 기능이 켜진 경우에도 실시간 조회를 먼저 수행하며, `transport_failed`에 한해 동일계정·미만료 보유수량을 사용한다. 브로커 0 이면 주문 미제출 + WARN + `sell_skipped_no_balance` 카운터(divergence 는 reconcile 경로로 표면화). skip 된 매도가 큐를 막지 않도록 리스트 기반 dispatch. - 가격/주문유형 결정(dcy-012/REQ-013, issue #10): `_dispatch` 는 side 분기 없이 슬롯 `limit_price` **한 필드**만 읽는다 — 값이 있으면 지정가(`order_type="00"`, 그 가격), 없으면 시장가(`order_type="01"`). `limit_price` 는 pending 진입 시 set 되고 pending 이탈(→holding/→closed) 시 clear 되므로, 매수 지정가(과거 `entry_price`)가 매도 주문에 재사용되지 않는다. 자율 매도(safety-sell/kill-switch)는 가격 미전달 → 시장가로 즉시 체결(미체결 `pending_sell` 갇힘 방지, issue #9 연계). - **슬리피지 보호 지정가 + KRX 호가단위 스냅(dcy-022/REQ-070, dcy-030/REQ-083)**: 슬롯에 `slippage_pct` 가 있고 `limit_price` 도 있으면, raw 실효가 = `limit_price×(1±pct/100)` 를 KRX 호가단위(REQ-083 `domain/pricing.py`)로 **스냅**해 제출한다. 방향 = **상한 준수**: 매수 = `floor_to_tick(raw)`(slippage 상한 미초과), 매도 = `ceil_to_tick(raw)`(매도 하한 미초과). band<1tick 엣지에서 미체결을 막기 위해 매수 = `max(floor_to_tick(raw), ceil_to_tick(limit_price))`, 매도 = `min(ceil_to_tick(raw), floor_to_tick(limit_price))` 로 entry 유효 호가를 보장한다. `limit_price` 없으면(시장가) 스냅 무시. 미지정 슬롯은 기존 동작(회귀 없음). 주식 전용(ETF/ETN·2,000원 미만 비주식 미지원, dcy-030 OQ-1). 이전 구현은 `int(ceil/floor)` 무스냅이라 호가단위 위반으로 KIS 거부(issue #28) — CHG-20260616-01 에서 스냅·방향(체결 보장→상한 준수) 정정. 2. poll: `list_dispatched_pending(instance_key)`(broker_order_id NOT NULL) → `BrokerPort.poll_fill` → 체결 시 `pending_buy→holding`(avg_price/qty) / `pending_sell→inactive`(점유 초기화·자산 정체성 보존), 거부 시 롤백. 전이 후 `clear_dispatch`. - **체결가 백필 가드(dcy-032/REQ-095/096, BAT-048)**: `_resolve_fill_price(fs, slot)` 헬퍼가 체결가를 결정. (1) `fs.avg_price`가 양수이면 그대로 사용(happy path — WARN 없음). (2) `fs.avg_price=None` 또는 `≤0`이고 `slot.limit_price>0`이면 지정가로 백필 + `reason=avg_price_backfill` **WARN 1건** 발행(슬롯 평단·원장 price에 백필값 사용, 0 저장 방지). (3) 지정가마저 None/≤0이면 체결 처리를 **skip**(슬롯 전이·원장 기록 미수행) + `reason=fill_price_unresolvable` 강 WARN 발행 — 슬롯이 PENDING 상태로 남아 stale_pending 경보가 운영자 개입을 유도. 이 가드는 `_on_filled`·`_on_partial_cancel`·`_record_ledger` 세 경로에 공통 적용된다(`reconcile()` 경로도 동일 헬퍼를 탄다). - 외부 취소 감지(dcy-013/REQ-012/013, issue #9): 계좌에서 직접 취소된 주문을 KIS `poll_fill` 이 `cncl_yn="Y"` 로 파싱한다. **체결 0 취소** → `FillStatus(state="cancelled")` → 기존 롤백 경로(`pending_sell→holding`, `clear_dispatch`)로 자동 복구(영구 갇힘 제거). **부분 체결 후 취소**(`is_partial_cancel`) → 체결분 원장 1회 기록 + 슬롯 qty 차감(매도: 잔여 held, cost basis 유지 / 매수: 체결분만) + `holding` 복귀(OQ-1 A안, 과매도·이중기록 방지). - pending stale 워치독(dcy-013/REQ-014 FR-3, Alert-only): `accepted`/`partially_filled`(취소 아님)로 다음 tick 대기 중인 슬롯이 `TREX_PENDING_STALE_THRESHOLD`(초, 0=비활성)를 초과해 머물면 WARN 로그 + `ExecutionResult.stale_pending` 카운터로 가시화한다. **상태는 변경하지 않는다**(auto-remediation 후속) — 취소 감지 누락(R-1) 등 미지 원인의 무한 갇힘이 silent 로 남지 않게 하는 안전망. - - **timeout 자동 취소(dcy-022/REQ-069)**: `accepted`/`partially_filled` 상태 슬롯에 `timeout_sec` 가 있고 경과 시간(`now - slot.updated_at`)이 초과하면 `broker.cancel_order` 를 best-effort 로 송신하고 `ExecutionResult.timeout_cancelled` 를 증가시킨다. **슬롯 상태는 이 시점에 변경하지 않는다** — 다음 poll 에서 `is_terminal_reject`(전량 취소) → rollback / `is_partial_cancel`(부분 체결 후 취소) → 부분 보존(기존 경로 재사용). 취소-체결 race: `cancel_order` 가 이미 체결된 주문을 받으면 브로커가 거부 → 다음 poll `is_filled` → 정상 체결 경로 수렴(임의 롤백 없음, FR-4). 기존 `timeout_sec` 없는 슬롯은 stale-alert 만 실행(REQ-014 회귀 없음). + - **timeout 자동 취소(dcy-022/REQ-069, dcy-041)**: `accepted`/`partially_filled` 상태 슬롯에 `timeout_sec`가 있고 경과 시간이 초과하면 외부 호출 전에 `cancel_state=unknown`을 예약 저장한 뒤 취소를 한 번 보낸다. 즉시 결과는 `requested|rejected|unknown`으로 교체하지만 **슬롯 상태는 변경하지 않는다**. 같은 주문은 자동 재취소하지 않으며 다음 poll의 `cancelled|filled`만 최종 상태를 확정한다. 3. reconcile(재기동): 서빙 전 `list_dispatched_pending` 재폴링으로 크래시 후 복구(중복 주문 0). 미디스패치(broker_order_id NULL) 슬롯은 다음 dispatch 가 1회만 송신. - 모든 슬롯 쓰기는 SlotService(advisory lock) 단일 경로(불변식 A). 브로커 주문/취소는 BrokerPort 단일 세션만(불변식 B). - 슬롯 `limit_price` 생애주기(dcy-012/REQ-002): INACTIVE→PENDING_BUY(매수 지정가 set) → PENDING_BUY→HOLDING(clear) → HOLDING→PENDING_SELL(매도 지정가 set 또는 None=시장가) → PENDING_SELL→INACTIVE(clear). 롤백(→INACTIVE) 전이에서도 clear. 외부 취소 후 holding 복귀에서도 clear 되어 재매도는 시장가. - **실행 정책 생애주기(dcy-022/REQ-068)**: `timeout_sec`·`slippage_pct` 는 `limit_price` 와 동일한 생애주기(PENDING_* 진입 시 set, HOLDING/INACTIVE 탈출 시 clear). 슬롯의 `updated_at` 이 dispatch 시각 proxy 로 사용된다(mark_dispatched 가 갱신, stale-alert 와 동일 방법). - 어댑터: paper(결정적 async 더블) / KIS(모의투자 검증, 실계좌 게이팅). -- **venue 전달 경로(BAT-056/REQ-117/118, BAT-059/REQ-124)**: buy/sell intent `venue`(기본 KRX) → pending `Slot.venue` → `OrderRequest.venue`와 취소 venue → KIS `EXCG_ID_DVSN_CD`. 허용값은 KRX/NXT/SOR이며 invalid는 슬롯 전이 전 거부한다. INACTIVE 복귀 시 KRX로 초기화한다. +- **venue 전달 경로(BAT-056/REQ-117/118, BAT-059/REQ-124, dcy-041)**: buy/sell intent `venue`(기본 AUTO) → pending `Slot.venue` → KIS 제출 시 KST 세션별 실제 KRX/NXT 확정 → `OrderAck.venue`와 주문번호 원자 저장 → 취소·재시작·Core outbox에서 실제 거래소 재사용. 허용값은 AUTO/KRX/NXT/SOR이며 invalid는 슬롯 전이 전 거부한다. HOLDING은 다음 자율매도를 위해 AUTO, INACTIVE는 하위 호환을 위해 KRX로 초기화한다. + +### 계좌조회 단기 복구 (dcy-041 / GitHub #45) + +1. 기능을 켠 KIS 데몬은 시작 시 1회 동기 조회 후 계정당 한 백그라운드 소유자가 기본 300초 간격과 실패 백오프로 보유수량·주문가능금액을 갱신한다. +2. 보유수량과 현금은 독립적으로 성공값과 시각을 저장한다. 계정 해시 변경, 인증·인가·형식·브로커 거부는 실패 폐쇄한다. +3. 신규매수는 기본 600초 이내 현금값에서 기존 pending 매수 예약금액과 새 주문금액을 합산해 검사한다. 시장가처럼 금액을 확정할 수 없으면 차단한다. +4. pending 매도는 실시간 보유조회 전송 실패일 때만 기본 3600초 이내 전략 소유 보유수량을 사용한다. 수동 보유분을 매도 권한으로 승격하지 않는다. +5. 재시작 원장대조, `sync`, `verify`, Portfolio Hub 일일 정합은 캐시를 사용하지 않는다. ### 운영자 cancel 백스톱 (dcy-022/REQ-071) diff --git a/tests/adapters/broker/test_kis.py b/tests/adapters/broker/test_kis.py index 5c32f05..72ae6e4 100644 --- a/tests/adapters/broker/test_kis.py +++ b/tests/adapters/broker/test_kis.py @@ -27,6 +27,8 @@ def test_account_split_and_mock_tr_ids(): assert b._tr_submit("buy") == "VTTC0012U" assert b._tr_submit("sell") == "VTTC0011U" assert b.name() == "kis-mock" + assert len(b.account_scope_hash()) == 64 + assert "12345678" not in b.account_scope_hash() def test_live_base_url_requires_allow_live(): diff --git a/tests/adapters/broker/test_ownership.py b/tests/adapters/broker/test_ownership.py index 2458b05..af8e628 100644 --- a/tests/adapters/broker/test_ownership.py +++ b/tests/adapters/broker/test_ownership.py @@ -1,4 +1,5 @@ from decimal import Decimal +from datetime import datetime, timezone import pytest @@ -14,6 +15,10 @@ from trex_core.application.broker_port import ( FillStatus, OrderAck, ) +from trex_core.application.account_snapshot import ( + AccountSnapshotService, + InMemoryAccountSnapshotStore, +) class FakeBroker: @@ -36,6 +41,12 @@ class FakeBroker: def get_balance(self): return self.balance + def account_scope_hash(self): + return "c" * 64 + + def get_available_cash(self, *, price_check_code="005930"): + return 1_000_000 + def get_current_price(self, symbol, market="J"): return Decimal("1") @@ -90,3 +101,28 @@ def test_sell_cannot_reduce_below_strategy_owned_zero(tmp_path): side="sell", quantity=1, ) + + +def test_account_snapshot_keeps_manual_holdings_outside_strategy_scope(tmp_path): + ownership = StrategyOwnershipStore(tmp_path / "ownership.json") + ownership.apply_fill( + execution_id="buy-1", + symbol="005930", + side="buy", + quantity=2, + ) + broker = OwnershipAwareBroker( + FakeBroker([BalanceItem("005930", 21), BalanceItem("000660", 1)]), + ownership, + ) + snapshots = InMemoryAccountSnapshotStore() + service = AccountSnapshotService( + broker=broker, + store=snapshots, + clock=lambda: datetime(2026, 7, 29, tzinfo=timezone.utc), + ) + + assert service.refresh() + snapshot = snapshots.load("c" * 64) + assert snapshot is not None + assert snapshot.holdings == (BalanceItem("005930", 2),) diff --git a/tests/adapters/persistence/test_account_snapshot_store.py b/tests/adapters/persistence/test_account_snapshot_store.py new file mode 100644 index 0000000..e3e129b --- /dev/null +++ b/tests/adapters/persistence/test_account_snapshot_store.py @@ -0,0 +1,37 @@ +"""PostgreSQL account snapshot persistence contract.""" +from __future__ import annotations + +from datetime import datetime, timezone + +from trex_core.adapters.persistence.account_snapshot import ( + PostgreSQLAccountSnapshotStore, +) +from trex_core.application.account_snapshot import AccountSnapshot +from trex_core.application.broker_port import BalanceItem + + +def test_round_trip_keeps_independent_capture_times(pg_clean): + store = PostgreSQLAccountSnapshotStore(pg_clean) + holdings_at = datetime(2026, 7, 29, 0, 0, tzinfo=timezone.utc) + cash_at = datetime(2026, 7, 28, 23, 59, tzinfo=timezone.utc) + expected = AccountSnapshot( + account_hash="d" * 64, + holdings=(BalanceItem("005930", 2),), + holdings_captured_at=holdings_at, + available_cash=300_000, + cash_captured_at=cash_at, + ) + + store.save(expected) + + assert store.load(expected.account_hash) == expected + + +def test_refresh_guard_is_account_scoped_and_released(pg_clean): + store = PostgreSQLAccountSnapshotStore(pg_clean) + with store.refresh_guard("e" * 64) as first: + assert first is True + with store.refresh_guard("e" * 64) as second: + assert second is False + with store.refresh_guard("e" * 64) as after_release: + assert after_release is True diff --git a/tests/adapters/persistence/test_migration_and_drift.py b/tests/adapters/persistence/test_migration_and_drift.py index 51b1268..6e54153 100644 --- a/tests/adapters/persistence/test_migration_and_drift.py +++ b/tests/adapters/persistence/test_migration_and_drift.py @@ -18,6 +18,7 @@ _EXPECTED_TABLES = { "intent_log", "ledger_outbox", # BAT-006 / REQ-019 "snapshot_pause", # BAT-008 / REQ-032 + "broker_account_snapshot", # dcy-041 / GitHub #45 } diff --git a/tests/application/test_account_snapshot.py b/tests/application/test_account_snapshot.py new file mode 100644 index 0000000..7983eab --- /dev/null +++ b/tests/application/test_account_snapshot.py @@ -0,0 +1,164 @@ +"""Bounded account-read recovery (dcy-041 / GitHub #45).""" +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +import pytest + +from trex_core.application.account_snapshot import ( + AccountSnapshot, + AccountSnapshotService, + AccountSnapshotUnavailable, + InMemoryAccountSnapshotStore, +) +from trex_core.application.broker_port import BalanceItem, BrokerBalanceError + + +class FakeAccountBroker: + def __init__(self) -> None: + self.account_hash = "a" * 64 + self.balance = [BalanceItem("005930", 2)] + self.cash = 500_000 + self.balance_error: str | None = None + self.cash_error: str | None = None + + def account_scope_hash(self) -> str: + return self.account_hash + + def get_balance(self) -> list[BalanceItem]: + if self.balance_error is not None: + raise BrokerBalanceError(self.balance_error) + return list(self.balance) + + def get_available_cash(self, *, price_check_code: str = "005930") -> int: + if self.cash_error is not None: + raise BrokerBalanceError(self.cash_error) + return self.cash + + +def _wire(): + now = [datetime(2026, 7, 29, tzinfo=timezone.utc)] + broker = FakeAccountBroker() + store = InMemoryAccountSnapshotStore() + service = AccountSnapshotService( + broker=broker, + store=store, + buy_max_age_sec=600, + sell_max_age_sec=3600, + clock=lambda: now[0], + rng=lambda: 0.5, + ) + return broker, store, service, now + + +def test_refresh_persists_independent_holdings_and_cash_capture_times(): + broker, store, service, now = _wire() + assert service.refresh() is True + first = store.load(broker.account_hash) + assert first is not None + + now[0] += timedelta(seconds=120) + broker.balance = [BalanceItem("005930", 3)] + broker.cash_error = "transport_failed" + assert service.refresh() is False + + second = store.load(broker.account_hash) + assert second is not None + assert second.holdings == (BalanceItem("005930", 3),) + assert second.holdings_captured_at == now[0] + assert second.available_cash == first.available_cash + assert second.cash_captured_at == first.cash_captured_at + + +def test_sell_falls_back_only_for_transport_failure_and_within_sell_ttl(): + broker, _, service, now = _wire() + assert service.refresh() + broker.balance_error = "transport_failed" + now[0] += timedelta(seconds=601) + + assert service.holdings_for_sell() == [BalanceItem("005930", 2)] + assert service.buy_guard(qty=1, limit_price=100_000) == ( + "account_snapshot_unavailable" + ) + + now[0] += timedelta(seconds=3000) + with pytest.raises(AccountSnapshotUnavailable) as raised: + service.holdings_for_sell() + assert raised.value.code == "expired" + + +@pytest.mark.parametrize( + "error_code", + ["authentication_failed", "authorization_failed", "malformed_response", + "broker_rejected", "ownership_mismatch"], +) +def test_sell_never_falls_back_for_non_transport_failures(error_code): + broker, _, service, _ = _wire() + assert service.refresh() + broker.balance_error = error_code + + with pytest.raises(AccountSnapshotUnavailable) as raised: + service.holdings_for_sell() + assert raised.value.code == "live_read_failed" + + +def test_buy_guard_accounts_for_existing_pending_cash_and_never_renews_age(): + broker, store, service, now = _wire() + assert service.refresh() + captured = store.load(broker.account_hash).cash_captured_at + + assert service.buy_guard( + qty=2, limit_price=100_000, reserved_cash=250_000 + ) is None + assert service.buy_guard( + qty=3, limit_price=100_000, reserved_cash=250_000 + ) == "insufficient_snapshot_cash" + assert store.load(broker.account_hash).cash_captured_at == captured + + now[0] += timedelta(seconds=601) + assert service.buy_guard(qty=1, limit_price=1) == ( + "account_snapshot_unavailable" + ) + + +def test_current_account_hash_is_required_on_every_consume(): + broker, _, service, _ = _wire() + assert service.refresh() + broker.account_hash = "b" * 64 + + assert service.buy_guard(qty=1, limit_price=1) == ( + "account_snapshot_unavailable" + ) + broker.balance_error = "transport_failed" + with pytest.raises(AccountSnapshotUnavailable) as raised: + service.holdings_for_sell() + assert raised.value.code == "account_mismatch" + + +def test_restart_transport_attempt_can_use_existing_fresh_cash_snapshot(): + broker, store, service, now = _wire() + store.save( + AccountSnapshot( + account_hash=broker.account_hash, + holdings=(BalanceItem("005930", 2),), + holdings_captured_at=now[0], + available_cash=500_000, + cash_captured_at=now[0], + ) + ) + broker.balance_error = "transport_failed" + broker.cash_error = "transport_failed" + + assert service.refresh() is False + assert service.buy_guard(qty=2, limit_price=100_000) is None + + +def test_invalid_or_missing_buy_notional_fails_closed(): + _, _, service, _ = _wire() + assert service.refresh() + assert service.buy_guard(qty=None, limit_price=1) == ( + "account_snapshot_price_required" + ) + assert service.buy_guard(qty=1, limit_price=None) == ( + "account_snapshot_price_required" + ) diff --git a/tests/application/test_asset_centric_admission.py b/tests/application/test_asset_centric_admission.py index 7f62713..4907753 100644 --- a/tests/application/test_asset_centric_admission.py +++ b/tests/application/test_asset_centric_admission.py @@ -99,3 +99,52 @@ def test_no_cap_admits_without_capacity_check(): d = intents.submit_buy(BuyIntent(intent_id=f"b{i}", portfolio_id=pid)) assert d.status == "admitted" assert repo.count_active(INST) == 3 + + +def test_buy_guard_rejection_is_fail_closed_and_sell_path_is_unaffected(): + repo = InMemorySlotRepository() + svc = SlotService(repo, instance_key=INST) + repo.reconcile_universe(INST, {"pf-a": "SYM"}) + intents = IntentService( + svc, + repo, + InMemoryIntentStore(), + resolve_symbol=lambda pid: _SYMS.get(pid), + instance_key=INST, + buy_guard_fn=lambda _intent: "insufficient_snapshot_cash", + ) + + decision = intents.submit_buy( + BuyIntent( + intent_id="guarded", + portfolio_id="pf-a", + qty=1, + entry_price=1000, + ) + ) + assert decision.status == "rejected" + assert decision.reason == "insufficient_snapshot_cash" + assert repo.find_slot_by_portfolio(INST, "pf-a").status is SlotState.INACTIVE + + +def test_buy_guard_exception_rejects_without_slot_transition(): + repo = InMemorySlotRepository() + svc = SlotService(repo, instance_key=INST) + repo.reconcile_universe(INST, {"pf-a": "SYM"}) + + def broken_guard(_intent): + raise RuntimeError("unavailable") + + intents = IntentService( + svc, + repo, + InMemoryIntentStore(), + resolve_symbol=lambda pid: _SYMS.get(pid), + instance_key=INST, + buy_guard_fn=broken_guard, + ) + decision = intents.submit_buy( + BuyIntent(intent_id="broken", portfolio_id="pf-a", qty=1, entry_price=1000) + ) + assert decision.reason == "account_snapshot_unavailable" + assert repo.find_slot_by_portfolio(INST, "pf-a").status is SlotState.INACTIVE diff --git a/tests/application/test_execution.py b/tests/application/test_execution.py index 81a7e4f..6b6ba80 100644 --- a/tests/application/test_execution.py +++ b/tests/application/test_execution.py @@ -8,6 +8,7 @@ from __future__ import annotations from trex_core.adapters.broker.paper import PaperBroker from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore from trex_core.application.broker_port import ( + BalanceItem, CancelOutcome, CancelState, FillStatus, @@ -182,6 +183,37 @@ def test_full_sell_qty_is_broker_balance(): assert repo.fetch_slot(sid).status is SlotState.INACTIVE # full exit +def test_pending_sell_uses_injected_bounded_balance_reader(): + class LiveReadFailPaper(PaperBroker): + fail_live_balance = False + + def get_balance(self): + if self.fail_live_balance: + raise AssertionError("raw balance reader must not be called") + return super().get_balance() + + broker = LiveReadFailPaper(fill_after_polls=0) + repo, svc, intents, execution = _wire(broker) + sid = _buy(intents, qty=10).slot_id + execution.run_once() + svc.sell(sid, actor="t", reason="full") + broker.fail_live_balance = True + broker._fill_after = 5 + + snapshot_execution = ExecutionService( + svc, + repo, + broker, + instance_key=INST, + balance_reader=lambda: [BalanceItem("005931", 10)], + ) + snapshot_execution.run_once() + + sell_orders = [o for o in broker._orders.values() if o.side == "sell"] + assert len(sell_orders) == 1 and sell_orders[0].qty == 10 + assert repo.fetch_slot(sid).status is SlotState.PENDING_SELL + + def test_partial_sell_qty_is_min_intent_broker(): """dcy-021/REQ-062 (D4): partial sell dispatches min(intent qty, broker); remainder stays holding (broker-authoritative qty), cost basis preserved.""" diff --git a/tests/application/test_reconcile_warning_blocking.py b/tests/application/test_reconcile_warning_blocking.py index ad5ef31..169a303 100644 --- a/tests/application/test_reconcile_warning_blocking.py +++ b/tests/application/test_reconcile_warning_blocking.py @@ -3,8 +3,6 @@ from __future__ import annotations import logging -import pytest - from trex_core.application.broker_port import BalanceItem from trex_core.application.intents import BuyIntent, IntentService, SellIntent from trex_core.application.messaging import RpcRequest @@ -20,7 +18,6 @@ from trex_core.application.reconciliation import ( reconcile_core_holdings, reconcile_positions, ) -from trex_core.application.ports import IntentStore from trex_core.domain.slot import SlotService, SlotState from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository @@ -54,7 +51,8 @@ class InMemoryIntentStore: return None class R: def __init__(self, d): self.__dict__.update(d) - r = R(rec); r.intent_id = intent_id + r = R(rec) + r.intent_id = intent_id return r def finalize(self, intent_id, status, slot_id, reason): self._db[intent_id] = {"status": status, "slot_id": slot_id, "reason": reason} @@ -211,7 +209,6 @@ def test_reconcile_positions_c_case_matching_no_warn(): # REQ-103 AC-5: C case with warn_store=None => backward compat def test_reconcile_positions_no_warn_store_backward_compat(): - repo = InMemorySlotRepository() store = _store_with([ PortfolioItem(portfolio_id="pf-A", symbol="005930", cash_balance=0, asset_quantity=10), ]) @@ -458,7 +455,7 @@ def test_reconcile_core_holdings_b_case_logging(caplog): svc = SlotService(repo) warn_store = InMemoryReconcileWarningStore() - slot = _make_holding_slot(repo, svc, qty=10) + _make_holding_slot(repo, svc, qty=10) broker = FakeBroker([BalanceItem(symbol="005930", hldg_qty=5)]) with caplog.at_level(logging.WARNING, logger="trex_core.application.reconciliation"): reconcile_core_holdings( @@ -527,7 +524,7 @@ def test_short_term_instance_normal_no_warn(): svc = SlotService(repo) warn_store = InMemoryReconcileWarningStore() - slot = _make_holding_slot(repo, svc, qty=10, portfolio_id="pf-A") + _make_holding_slot(repo, svc, qty=10, portfolio_id="pf-A") store = _store_with([ PortfolioItem(portfolio_id="pf-A", symbol="005930", cash_balance=0, asset_quantity=10), ]) diff --git a/tests/application/test_reconcile_warning_store.py b/tests/application/test_reconcile_warning_store.py index 0218cf8..97f329e 100644 --- a/tests/application/test_reconcile_warning_store.py +++ b/tests/application/test_reconcile_warning_store.py @@ -5,8 +5,6 @@ PostgreSQL adapter is tested by DB-skipped integration tests. """ from __future__ import annotations -import pytest - from trex_core.application.portfolio import InMemoryReconcileWarningStore diff --git a/tests/application/test_snapshot_scheduler.py b/tests/application/test_snapshot_scheduler.py index dc23ad2..6b0c7c2 100644 --- a/tests/application/test_snapshot_scheduler.py +++ b/tests/application/test_snapshot_scheduler.py @@ -2,16 +2,13 @@ from __future__ import annotations from datetime import datetime, timezone, timedelta -from unittest.mock import MagicMock, call - -import pytest +from unittest.mock import MagicMock from trex_core.adapters.broker.paper import PaperBroker from trex_core.adapters.portfolio.stub_read import StubPortfolioRead from trex_core.application.alarm_emitter import AlarmEmitter from trex_core.application.portfolio import ( BrokerCredentialPayload, - PortfolioHubUnavailableError, PortfolioItem, SnapshotStore, TradingContext, @@ -266,7 +263,6 @@ def test_entry_allowed_fixed_for_day_no_refetch(monkeypatch): def test_successful_credential_cached_for_day(): """REQ-115 AC-3: 성공 후 그날 내 재fetch 없음 — 캐시 유지.""" fetch_count = {"n": 0} - original_fetch = StubPortfolioRead.fetch_broker_credential class CountingStub(StubPortfolioRead): def fetch_broker_credential(self): diff --git a/tests/application/test_verify.py b/tests/application/test_verify.py index 35983d4..214cd60 100644 --- a/tests/application/test_verify.py +++ b/tests/application/test_verify.py @@ -14,8 +14,6 @@ from __future__ import annotations from uuid import uuid4 -import pytest - from trex_core.application.broker_port import BalanceItem from trex_core.application.portfolio import ( InMemoryReconcileWarningStore, diff --git a/tests/conftest.py b/tests/conftest.py index 7907e06..4659215 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -15,6 +15,7 @@ import os import pytest _PG_TABLES = ( + "broker_account_snapshot", "killswitch_exclusion", "snapshot_pause", "ledger_outbox", diff --git a/tests/runtime/test_account_snapshot_config.py b/tests/runtime/test_account_snapshot_config.py new file mode 100644 index 0000000..23c1e89 --- /dev/null +++ b/tests/runtime/test_account_snapshot_config.py @@ -0,0 +1,55 @@ +"""Account snapshot configuration stays opt-in and bounded.""" +from __future__ import annotations + +import pytest + +from trex_core.runtime.config import AccountSnapshotConfig, DaemonConfig + +_KEYS = ( + "TREX_ACCOUNT_SNAPSHOT_ENABLED", + "TREX_ACCOUNT_SNAPSHOT_BUY_MAX_AGE_SEC", + "TREX_ACCOUNT_SNAPSHOT_SELL_MAX_AGE_SEC", + "TREX_ACCOUNT_SNAPSHOT_REFRESH_INTERVAL_SEC", +) + + +@pytest.fixture(autouse=True) +def _clean(monkeypatch): + for key in _KEYS: + monkeypatch.delenv(key, raising=False) + monkeypatch.setenv("TREX_POSTGRESQL_DSN", "dbname=x") + + +def test_defaults_are_disabled_and_bounded(): + config = DaemonConfig.from_env().account_snapshot + assert config == AccountSnapshotConfig() + assert config.enabled is False + + +def test_env_enables_and_tunes_account_snapshot(): + import os + + os.environ["TREX_ACCOUNT_SNAPSHOT_ENABLED"] = "true" + os.environ["TREX_ACCOUNT_SNAPSHOT_BUY_MAX_AGE_SEC"] = "300" + os.environ["TREX_ACCOUNT_SNAPSHOT_SELL_MAX_AGE_SEC"] = "1800" + os.environ["TREX_ACCOUNT_SNAPSHOT_REFRESH_INTERVAL_SEC"] = "240" + + config = DaemonConfig.from_env().account_snapshot + assert config.enabled is True + assert config.buy_max_age_sec == 300 + assert config.sell_max_age_sec == 1800 + assert config.refresh_interval_sec == 240 + + +@pytest.mark.parametrize( + "kwargs", + [ + {"buy_max_age_sec": -1}, + {"buy_max_age_sec": 601, "sell_max_age_sec": 600}, + {"sell_max_age_sec": 21_601}, + {"refresh_interval_sec": 0}, + ], +) +def test_invalid_age_or_interval_policy_is_rejected(kwargs): + with pytest.raises(ValueError): + AccountSnapshotConfig(**kwargs) diff --git a/tests/runtime/test_cli_report_wac.py b/tests/runtime/test_cli_report_wac.py index 9a3a266..fc32e05 100644 --- a/tests/runtime/test_cli_report_wac.py +++ b/tests/runtime/test_cli_report_wac.py @@ -322,7 +322,7 @@ class TestReportWacPath: return_value=mock_adapter, ), ): - rc = cli.main(["report", "--json"]) + cli.main(["report", "--json"]) j = json.loads(capsys.readouterr().out) # WAC 필드는 null (당일 net 99999로 채워지지 않음) assert j["realized_pnl_wac"] is None diff --git a/trex_core/adapters/broker/kis.py b/trex_core/adapters/broker/kis.py index 50be9f6..f824085 100644 --- a/trex_core/adapters/broker/kis.py +++ b/trex_core/adapters/broker/kis.py @@ -12,6 +12,7 @@ Endpoints: /oauth2/tokenP (token), /uapi/domestic-stock/v1/trading/order-cash from __future__ import annotations import base64 +import hashlib import json from dataclasses import replace from datetime import datetime, time, timedelta, timezone @@ -157,6 +158,11 @@ class KISBroker: def name(self) -> str: return "kis-mock" if self._mock else "kis-live" + def account_scope_hash(self) -> str: + """Stable non-secret identity for account-scoped persisted state.""" + cano, product = self._account() + return hashlib.sha256(f"{cano}|{product}".encode()).hexdigest() + # --- credential hot-swap (dcy-036 / REQ-113 AC-2 / REQ-115 AC-1) ---------- def reinitialize(self, credential) -> None: """Replace app_key/app_secret/account_ref with fetched day-start credential. @@ -711,8 +717,20 @@ class KISBroker: ) response.raise_for_status() payload = response.json() - except (httpx.HTTPError, KeyError, TypeError, ValueError): + except BrokerBalanceError: + raise + except httpx.HTTPStatusError as exc: + status = exc.response.status_code if exc.response is not None else None + code = ( + "authentication_failed" + if status == 401 + else ("authorization_failed" if status == 403 else "transport_failed") + ) + raise BrokerBalanceError(code) from None + except httpx.HTTPError: raise BrokerBalanceError("transport_failed") from None + except (KeyError, TypeError, ValueError): + raise BrokerBalanceError("malformed_response") from None if not isinstance(payload, dict): raise BrokerBalanceError("malformed_response") if str(payload.get("rt_cd")) != "0": diff --git a/trex_core/adapters/broker/ownership.py b/trex_core/adapters/broker/ownership.py index 4c99375..f174e72 100644 --- a/trex_core/adapters/broker/ownership.py +++ b/trex_core/adapters/broker/ownership.py @@ -159,5 +159,11 @@ class OwnershipAwareBroker: def get_balance(self) -> list[BalanceItem]: return self._store.project(self._delegate.get_balance()) + def account_scope_hash(self) -> str: + return self._delegate.account_scope_hash() + + def get_available_cash(self, *, price_check_code: str = "005930") -> int: + return self._delegate.get_available_cash(price_check_code=price_check_code) + def get_current_price(self, symbol: str, market: str = "J"): return self._delegate.get_current_price(symbol, market) diff --git a/trex_core/adapters/persistence/account_snapshot.py b/trex_core/adapters/persistence/account_snapshot.py new file mode 100644 index 0000000..b4212d8 --- /dev/null +++ b/trex_core/adapters/persistence/account_snapshot.py @@ -0,0 +1,104 @@ +"""PostgreSQL persistence for bounded broker account snapshots.""" +from __future__ import annotations + +import json +from contextlib import contextmanager +from datetime import datetime +from typing import Any, Optional + +from trex_core.application.account_snapshot import AccountSnapshot +from trex_core.application.broker_port import BalanceItem + + +def _require_psycopg(module: Any = None) -> Any: + if module is not None: + return module + try: + import psycopg # type: ignore[import-not-found] + except ImportError as exc: # pragma: no cover + raise RuntimeError("psycopg is required for account snapshot persistence") from exc + return psycopg + + +class PostgreSQLAccountSnapshotStore: + def __init__(self, dsn: str, *, psycopg_module: Any = None) -> None: + self._dsn = dsn + self._psycopg = _require_psycopg(psycopg_module) + + def save(self, snapshot: AccountSnapshot) -> None: + holdings = [ + {"symbol": item.symbol, "hldg_qty": item.hldg_qty} + for item in snapshot.holdings + ] + with self._psycopg.connect(self._dsn, autocommit=False) as conn: + with conn.cursor() as cur: + cur.execute( + "INSERT INTO broker_account_snapshot " + "(account_hash, holdings, holdings_captured_at, " + "available_cash, cash_captured_at) " + "VALUES (%s, %s::jsonb, %s, %s, %s) " + "ON CONFLICT (account_hash) DO UPDATE SET " + "holdings=excluded.holdings, " + "holdings_captured_at=excluded.holdings_captured_at, " + "available_cash=excluded.available_cash, " + "cash_captured_at=excluded.cash_captured_at", + ( + snapshot.account_hash, + json.dumps(holdings, separators=(",", ":"), sort_keys=True), + snapshot.holdings_captured_at, + snapshot.available_cash, + snapshot.cash_captured_at, + ), + ) + conn.commit() + + def load(self, account_hash: str) -> Optional[AccountSnapshot]: + with self._psycopg.connect(self._dsn) as conn: + with conn.cursor() as cur: + cur.execute( + "SELECT holdings, holdings_captured_at, " + "available_cash, cash_captured_at " + "FROM broker_account_snapshot WHERE account_hash=%s", + (account_hash,), + ) + row = cur.fetchone() + if row is None: + return None + raw_holdings = row[0] + if isinstance(raw_holdings, str): + raw_holdings = json.loads(raw_holdings) + holdings = tuple( + BalanceItem( + symbol=str(item["symbol"]), + hldg_qty=int(item["hldg_qty"]), + ) + for item in raw_holdings + ) + return AccountSnapshot( + account_hash=account_hash, + holdings=holdings, + holdings_captured_at=( + row[1] if isinstance(row[1], datetime) + else (datetime.fromisoformat(row[1]) if row[1] else None) + ), + available_cash=int(row[2]) if row[2] is not None else None, + cash_captured_at=( + row[3] if isinstance(row[3], datetime) + else (datetime.fromisoformat(row[3]) if row[3] else None) + ), + ) + + @contextmanager + def refresh_guard(self, account_hash: str): + """One refresh owner per account across Core processes.""" + lock_id = int(account_hash[:15], 16) + with self._psycopg.connect(self._dsn, autocommit=True) as conn: + with conn.cursor() as cur: + cur.execute("SELECT pg_try_advisory_lock(%s)", (lock_id,)) + acquired = bool(cur.fetchone()[0]) + try: + yield acquired + finally: + if acquired: + with conn.cursor() as cur: + cur.execute("SELECT pg_advisory_unlock(%s)", (lock_id,)) diff --git a/trex_core/adapters/persistence/sql/021_create_broker_account_snapshot.sql b/trex_core/adapters/persistence/sql/021_create_broker_account_snapshot.sql new file mode 100644 index 0000000..e1804cb --- /dev/null +++ b/trex_core/adapters/persistence/sql/021_create_broker_account_snapshot.sql @@ -0,0 +1,10 @@ +-- dcy-041 / GitHub #45: account-scoped last healthy KIS snapshot. +-- account_hash is SHA-256(CANO|ACNT_PRDT_CD); no plaintext account identity, +-- token, key, order response, or broker payload is persisted. +CREATE TABLE IF NOT EXISTS broker_account_snapshot ( + account_hash TEXT PRIMARY KEY, + holdings JSONB NOT NULL, + holdings_captured_at TIMESTAMPTZ, + available_cash BIGINT CHECK (available_cash IS NULL OR available_cash >= 0), + cash_captured_at TIMESTAMPTZ +); diff --git a/trex_core/application/account_snapshot.py b/trex_core/application/account_snapshot.py new file mode 100644 index 0000000..d671863 --- /dev/null +++ b/trex_core/application/account_snapshot.py @@ -0,0 +1,242 @@ +"""Bounded account snapshot recovery for transient KIS read failures. + +Holdings and cash capacity have independent capture times so a cash-read +failure never invalidates a healthy holdings refresh (and vice versa). Fresh +reconciliation remains live-only; snapshots are consumed solely by the +new-buy admission guard and the pending-sell quantity path. +""" +from __future__ import annotations + +import logging +import random +import threading +from contextlib import AbstractContextManager, contextmanager +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Callable, Optional, Protocol + +from trex_core.application.broker_port import BalanceItem, BrokerBalanceError + +_log = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class AccountSnapshot: + account_hash: str + holdings: tuple[BalanceItem, ...] + holdings_captured_at: Optional[datetime] + available_cash: Optional[int] + cash_captured_at: Optional[datetime] + + +class AccountSnapshotStore(Protocol): + def save(self, snapshot: AccountSnapshot) -> None: ... + def load(self, account_hash: str) -> Optional[AccountSnapshot]: ... + def refresh_guard(self, account_hash: str) -> AbstractContextManager[bool]: ... + + +class AccountSnapshotBroker(Protocol): + def account_scope_hash(self) -> str: ... + def get_balance(self) -> list[BalanceItem]: ... + def get_available_cash(self, *, price_check_code: str = "005930") -> int: ... + + +class AccountSnapshotUnavailable(RuntimeError): + """A sanitized fail-closed result for snapshot consumers.""" + + _CODES = frozenset({ + "not_ready", + "account_mismatch", + "expired", + "live_read_failed", + }) + + def __init__(self, code: str) -> None: + if code not in self._CODES: + raise ValueError(f"unsupported account snapshot error code: {code}") + self.code = code + super().__init__(f"account snapshot unavailable ({code})") + + +class InMemoryAccountSnapshotStore: + def __init__(self) -> None: + self._items: dict[str, AccountSnapshot] = {} + self._items_lock = threading.Lock() + self._refresh_lock = threading.Lock() + + def save(self, snapshot: AccountSnapshot) -> None: + with self._items_lock: + self._items[snapshot.account_hash] = snapshot + + def load(self, account_hash: str) -> Optional[AccountSnapshot]: + with self._items_lock: + return self._items.get(account_hash) + + @contextmanager + def refresh_guard(self, account_hash: str): + acquired = self._refresh_lock.acquire(blocking=False) + try: + yield acquired + finally: + if acquired: + self._refresh_lock.release() + + +class AccountSnapshotService: + """Refresh, age-check, and consume account-scoped broker snapshots.""" + + def __init__( + self, + *, + broker: AccountSnapshotBroker, + store: AccountSnapshotStore, + buy_max_age_sec: int = 600, + sell_max_age_sec: int = 3600, + refresh_interval_sec: float = 300.0, + clock: Optional[Callable[[], datetime]] = None, + rng: Optional[Callable[[], float]] = None, + ) -> None: + if buy_max_age_sec < 0 or sell_max_age_sec < buy_max_age_sec: + raise ValueError("invalid account snapshot age policy") + if sell_max_age_sec > 21_600: + raise ValueError("account snapshot age exceeds six-hour ceiling") + if refresh_interval_sec <= 0: + raise ValueError("account snapshot refresh interval must be positive") + self._broker = broker + self._store = store + self._buy_max_age = buy_max_age_sec + self._sell_max_age = sell_max_age_sec + self._refresh_interval = refresh_interval_sec + self._clock = clock or (lambda: datetime.now(tz=timezone.utc)) + self._rng = rng or random.random + self._refresh_lock = threading.Lock() + self._last_holdings_error: Optional[str] = "not_attempted" + self._last_cash_error: Optional[str] = "not_attempted" + + def refresh(self) -> bool: + """Refresh holdings and cash independently under one account-scoped owner.""" + account_hash = self._broker.account_scope_hash() + if not account_hash: + self._last_holdings_error = "ownership_mismatch" + self._last_cash_error = "ownership_mismatch" + return False + with self._refresh_lock, self._store.refresh_guard(account_hash) as acquired: + if not acquired: + return False + existing = self._store.load(account_hash) + holdings = existing.holdings if existing is not None else () + holdings_captured_at = ( + existing.holdings_captured_at if existing is not None else None + ) + available_cash = existing.available_cash if existing is not None else None + cash_captured_at = existing.cash_captured_at if existing is not None else None + now = self._clock() + holdings_ok = cash_ok = False + try: + holdings = tuple(self._broker.get_balance()) + holdings_captured_at = now + self._last_holdings_error = None + holdings_ok = True + except BrokerBalanceError as exc: + self._last_holdings_error = exc.code + _log.warning( + "account holdings snapshot refresh failed: code=%s", + exc.code, + ) + try: + available_cash = int(self._broker.get_available_cash()) + if available_cash < 0: + raise BrokerBalanceError("malformed_response") + cash_captured_at = now + self._last_cash_error = None + cash_ok = True + except BrokerBalanceError as exc: + self._last_cash_error = exc.code + _log.warning( + "account cash snapshot refresh failed: code=%s", + exc.code, + ) + if not account_hash or self._broker.account_scope_hash() != account_hash: + self._last_holdings_error = "ownership_mismatch" + self._last_cash_error = "ownership_mismatch" + return False + if not holdings_ok and not cash_ok: + return False + snapshot = AccountSnapshot( + account_hash=account_hash, + holdings=holdings, + holdings_captured_at=holdings_captured_at, + available_cash=available_cash, + cash_captured_at=cash_captured_at, + ) + self._store.save(snapshot) + return holdings_ok and cash_ok + + def run_forever(self, stop: threading.Event) -> None: + """Refresh continuously with bounded exponential backoff.""" + failures = 0 + while not stop.is_set(): + ok = self.refresh() + failures = 0 if ok else min(failures + 1, 4) + if ok: + wait_sec = self._refresh_interval * (0.9 + 0.2 * self._rng()) + else: + wait_sec = min(15.0 * (2 ** (failures - 1)), 300.0) + if stop.wait(wait_sec): + return + + def _load_current(self) -> AccountSnapshot: + account_hash = self._broker.account_scope_hash() + if not account_hash: + raise AccountSnapshotUnavailable("account_mismatch") + snapshot = self._store.load(account_hash) + if snapshot is None or snapshot.account_hash != account_hash: + raise AccountSnapshotUnavailable("account_mismatch") + return snapshot + + def _age(self, captured_at: Optional[datetime], max_age_sec: int) -> float: + if captured_at is None: + raise AccountSnapshotUnavailable("not_ready") + age = (self._clock() - captured_at).total_seconds() + if age < 0 or age > max_age_sec: + raise AccountSnapshotUnavailable("expired") + return age + + def holdings_for_sell(self) -> list[BalanceItem]: + """Prefer a live balance; fallback only on transient transport failure.""" + try: + return self._broker.get_balance() + except BrokerBalanceError as exc: + if exc.code != "transport_failed": + raise AccountSnapshotUnavailable("live_read_failed") from None + snapshot = self._load_current() + age = self._age(snapshot.holdings_captured_at, self._sell_max_age) + _log.warning( + "sell balance using account snapshot: age_sec=%.0f", + age, + ) + return list(snapshot.holdings) + + def buy_guard( + self, + *, + qty: Optional[float], + limit_price: Optional[float], + reserved_cash: float = 0.0, + ) -> Optional[str]: + """Return a stable rejection reason, or None when bounded admission passes.""" + if self._last_cash_error not in {None, "transport_failed"}: + return "account_snapshot_unavailable" + try: + snapshot = self._load_current() + self._age(snapshot.cash_captured_at, self._buy_max_age) + except AccountSnapshotUnavailable: + return "account_snapshot_unavailable" + if qty is None or qty <= 0 or limit_price is None or limit_price <= 0: + return "account_snapshot_price_required" + requested = float(qty) * float(limit_price) + if snapshot.available_cash is None: + return "account_snapshot_unavailable" + if requested + max(0.0, reserved_cash) > snapshot.available_cash: + return "insufficient_snapshot_cash" + return None diff --git a/trex_core/application/execution.py b/trex_core/application/execution.py index 2ea4d91..9cba327 100644 --- a/trex_core/application/execution.py +++ b/trex_core/application/execution.py @@ -30,6 +30,7 @@ from typing import Callable, Optional from trex_core.application.alarm_emitter import AlarmEmitter from trex_core.application.broker_port import ( + BalanceItem, BrokerPort, CancelOutcome, CancelState, @@ -84,6 +85,7 @@ class ExecutionService: stale_pending_threshold_sec: Optional[int] = None, instance_key: str = "default", alarm_emitter: Optional[AlarmEmitter] = None, + balance_reader: Optional[Callable[[], list[BalanceItem]]] = None, ) -> None: self._svc = slot_service self._repo = repo @@ -104,6 +106,10 @@ class ExecutionService: ) # dcy-033 / REQ-100: best-effort alarm emitter (None = no alarms sent). self._alarm_emitter = alarm_emitter + # dcy-041 / GitHub #45: pending-sell quantity may use a bounded + # account snapshot only through this injected reader. Reconciliation + # remains wired directly to the live broker. + self._sell_dispatch_balance_reader = balance_reader or broker.get_balance # ------------------------------------------------------------ def run_once(self) -> ExecutionResult: @@ -136,7 +142,7 @@ class ExecutionService: # surfaces via reconcile_core_holdings (REQ-062 A). No mirror # fallback by design. if sell_balance is None: - sell_balance = self._balance_map() + sell_balance = self._sell_dispatch_balance_map() bal = sell_balance.get(slot.symbol or "", 0) # dcy-021 / REQ-062 (D4): full sell (slot.qty is None) => broker # balance; partial sell => min(order qty, broker balance). Broker @@ -313,6 +319,13 @@ class ExecutionService: """Current broker balance as {symbol: hldg_qty} (REQ-062 ground truth).""" return {b.symbol: int(b.hldg_qty) for b in self._broker.get_balance()} + def _sell_dispatch_balance_map(self) -> dict: + """Balance used only to size a not-yet-submitted pending sell.""" + return { + b.symbol: int(b.hldg_qty) + for b in self._sell_dispatch_balance_reader() + } + def _resolve_fill_price(self, fs, slot) -> tuple[float | None, bool]: """Resolve the fill price from FillStatus, with limit_price backfill. diff --git a/trex_core/application/intents.py b/trex_core/application/intents.py index 57f6d9d..4f451ac 100644 --- a/trex_core/application/intents.py +++ b/trex_core/application/intents.py @@ -115,6 +115,7 @@ class IntentService: default_order_timeout_sec: Optional[int] = None, warn_store=None, # Optional[ReconcileWarningStore] — BAT-050 / REQ-104 entry_allowed_fn: Optional[Callable[[], bool]] = None, + buy_guard_fn: Optional[Callable[[BuyIntent], Optional[str]]] = None, ) -> None: self._svc = slot_service self._repo = slot_repo @@ -138,6 +139,10 @@ class IntentService: # When returns False, BuyIntent is rejected with reason 'day_start_load_failed'. # SellIntent is NOT gated by this (REQ-114 AC-5: 청산 경로 영향 없음). self._entry_allowed_fn: Optional[Callable[[], bool]] = entry_allowed_fn + # dcy-041 / GitHub #45: opt-in bounded cash-capacity admission guard. + # The callable returns a stable rejection reason or None. SellIntent is + # deliberately outside this gate. + self._buy_guard_fn = buy_guard_fn # ------------------------------------------------------------ def lookup(self, intent_id: str) -> Optional[IntentDecision]: @@ -270,6 +275,15 @@ class IntentService: symbol = _resolve(slot) if not symbol: return IntentDecision(intent.intent_id, "rejected", None, "unknown_portfolio") + if self._buy_guard_fn is not None: + try: + guard_reason = self._buy_guard_fn(intent) + except Exception: # noqa: BLE001 - buy guard fails closed + guard_reason = "account_snapshot_unavailable" + if guard_reason is not None: + return IntentDecision( + intent.intent_id, "rejected", None, guard_reason + ) # M cap applies only to NEW activations (initial buy). An increase on # an already-active asset does not change the active asset count. if (not is_increase and self._max_active_slots is not None diff --git a/trex_core/application/operator_report.py b/trex_core/application/operator_report.py index 3295ed9..3383bd7 100644 --- a/trex_core/application/operator_report.py +++ b/trex_core/application/operator_report.py @@ -129,7 +129,7 @@ def render_text(rep: OperatorReport) -> str: # REQ-110 WAC 실현손익 섹션 (portfolio-hub 통합) if rep.wac_unavailable: lines.append( - f"=== WAC 실현 PnL: portfolio-hub unavailable — realized PnL unavailable ===" + "=== WAC 실현 PnL: portfolio-hub unavailable — realized PnL unavailable ===" ) if rep.wac_unavailable_reason: lines.append(f" ({rep.wac_unavailable_reason})") diff --git a/trex_core/runtime/config.py b/trex_core/runtime/config.py index c9dffef..04fe027 100644 --- a/trex_core/runtime/config.py +++ b/trex_core/runtime/config.py @@ -112,6 +112,28 @@ class SnapshotScheduleConfig: tick_sec: float = 60.0 +@dataclass(frozen=True) +class AccountSnapshotConfig: + """Opt-in bounded KIS account-read recovery (dcy-041 / GitHub #45).""" + + enabled: bool = False + buy_max_age_sec: int = 600 + sell_max_age_sec: int = 3600 + refresh_interval_sec: float = 300.0 + + def __post_init__(self) -> None: + if self.buy_max_age_sec < 0: + raise ValueError("account snapshot buy_max_age_sec must be >= 0") + if self.sell_max_age_sec < self.buy_max_age_sec: + raise ValueError( + "account snapshot sell_max_age_sec must be >= buy_max_age_sec" + ) + if self.sell_max_age_sec > 21_600: + raise ValueError("account snapshot sell_max_age_sec exceeds six hours") + if self.refresh_interval_sec <= 0: + raise ValueError("account snapshot refresh_interval_sec must be > 0") + + @dataclass(frozen=True) class SafetySellConfig: """Safety-sell feature tuning (REQ-041 / BAT-010). @@ -227,6 +249,9 @@ class DaemonConfig: rabbitmq: Optional[RabbitMQConfig] = None # REQ-030 (BAT-008): daily snapshot refresh. snapshot_schedule: SnapshotScheduleConfig = SnapshotScheduleConfig() + # dcy-041 / GitHub #45: opt-in account read snapshot. Disabled keeps the + # previous live-only balance path byte-for-byte compatible. + account_snapshot: AccountSnapshotConfig = AccountSnapshotConfig() # REQ-041 (BAT-010): safety-sell feature config. safety_sell: SafetySellConfig = SafetySellConfig() # dcy-013 / REQ-014 (BAT-017): Alert-only pending-order stale watchdog @@ -301,6 +326,20 @@ class DaemonConfig: kst_time=os.environ.get("TREX_SNAPSHOT_SCHEDULE_KST", "07:30").strip() or "07:30", tick_sec=float(os.environ.get("TREX_SNAPSHOT_TICK_SEC", "60")), ) + account_snapshot = AccountSnapshotConfig( + enabled=os.environ.get( + "TREX_ACCOUNT_SNAPSHOT_ENABLED", "" + ).lower() in ("1", "true", "yes"), + buy_max_age_sec=int( + os.environ.get("TREX_ACCOUNT_SNAPSHOT_BUY_MAX_AGE_SEC", "600") + ), + sell_max_age_sec=int( + os.environ.get("TREX_ACCOUNT_SNAPSHOT_SELL_MAX_AGE_SEC", "3600") + ), + refresh_interval_sec=float( + os.environ.get("TREX_ACCOUNT_SNAPSHOT_REFRESH_INTERVAL_SEC", "300") + ), + ) _mc_time = os.environ.get("TREX_MARKET_CLOSE_LIQUIDATION_TIME", "").strip() or None safety_sell = SafetySellConfig( watchdog_timeout_sec=int(os.environ.get("TREX_WATCHDOG_TIMEOUT", "120")), @@ -329,6 +368,7 @@ class DaemonConfig: ledger_interval_sec=float(os.environ.get("TREX_LEDGER_INTERVAL_SEC", "1.0")), rabbitmq=rabbitmq, snapshot_schedule=snapshot_schedule, + account_snapshot=account_snapshot, safety_sell=safety_sell, pending_stale_threshold_sec=int( os.environ.get("TREX_PENDING_STALE_THRESHOLD", "0") diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index 612dd1a..18a4ca2 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -32,6 +32,9 @@ from trex_core.adapters.messaging import ( topology, ) from trex_core.adapters.persistence.migrate import apply_migrations +from trex_core.adapters.persistence.account_snapshot import ( + PostgreSQLAccountSnapshotStore, +) from trex_core.adapters.persistence.postgresql import ( PgStrategyStateStore, PostgreSQLIntentStore, @@ -46,9 +49,11 @@ from trex_core.adapters.portfolio import ( ) from trex_core.adapters.portfolio.killswitch_store import PostgreSQLKillSwitchStore from trex_core.application.alarm_emitter import AlarmEmitter +from trex_core.application.account_snapshot import AccountSnapshotService 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.ledger_outbox import LedgerWriter from trex_core.application.ledger_port import LedgerPort from trex_core.application.messaging import MessagingTransport @@ -199,6 +204,28 @@ 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.account_snapshot_service: AccountSnapshotService | None = None + if config.account_snapshot.enabled: + if config.broker != "kis": + raise RuntimeError( + "account snapshot recovery is supported only with broker=kis" + ) + required = ( + "account_scope_hash", + "get_balance", + "get_available_cash", + ) + if any(not callable(getattr(self.broker, name, None)) for name in required): + raise RuntimeError( + "configured KIS broker lacks account snapshot capabilities" + ) + self.account_snapshot_service = AccountSnapshotService( + broker=self.broker, # type: ignore[arg-type] + store=PostgreSQLAccountSnapshotStore(config.dsn), + buy_max_age_sec=config.account_snapshot.buy_max_age_sec, + sell_max_age_sec=config.account_snapshot.sell_max_age_sec, + refresh_interval_sec=config.account_snapshot.refresh_interval_sec, + ) # Event publisher transport — one dedicated instance for the # execution-loop thread (pika is not thread-safe). @@ -263,6 +290,11 @@ class Daemon: default_order_timeout_sec=config.default_order_timeout_sec, # dcy-022 warn_store=self.warn_store, # BAT-050 / REQ-104: reconcile-warning gate entry_allowed_fn=entry_allowed_fn, # dcy-036 / REQ-114: day-start gate + buy_guard_fn=( + self._account_snapshot_buy_guard + if self.account_snapshot_service is not None + else None + ), ) self.execution = ExecutionService( self.slot_service, self.slot_repo, self.broker, @@ -271,6 +303,11 @@ class Daemon: stale_pending_threshold_sec=config.pending_stale_threshold_sec, instance_key=config.instance_key, alarm_emitter=self.alarm_emitter, # dcy-033 / REQ-100 + balance_reader=( + self.account_snapshot_service.holdings_for_sell + if self.account_snapshot_service is not None + else None + ), ) # BAT-011: safety-sell trigger service (forced_close / SL-TP / market-close). @@ -293,6 +330,35 @@ class Daemon: asset = self.snapshot_store.get(portfolio_id) return asset.item.symbol if asset is not None else None + def _account_snapshot_buy_guard(self, intent: BuyIntent) -> str | None: + """Bound one new buy against fresh cached cash plus pending reservations.""" + service = self.account_snapshot_service + if service is None: + return None + reserved_cash = 0.0 + for slot in self.slot_repo.list_slots( + self.config.instance_key, status_filter={"pending_buy"} + ): + if ( + slot.qty is None + or slot.qty <= 0 + or slot.limit_price is None + or slot.limit_price <= 0 + ): + return "pending_buy_notional_unknown" + price = float(slot.limit_price) + if slot.slippage_pct: + price *= 1.0 + float(slot.slippage_pct) / 100.0 + reserved_cash += float(slot.qty) * price + price = intent.entry_price + if price is not None and intent.slippage_pct: + price *= 1.0 + float(intent.slippage_pct) / 100.0 + return service.buy_guard( + qty=intent.qty, + limit_price=price, + reserved_cash=reserved_cash, + ) + # ------------------------------------------------------------ event hook def _make_event_emitter(self): def emit(event_type: str, slot, routing_strategy_id: str | None = None) -> None: @@ -424,9 +490,18 @@ class Daemon: except Exception: # noqa: BLE001 _log.exception("snapshot scheduler loop crashed") + def _account_snapshot_loop(self) -> None: + service = self.account_snapshot_service + if service is not None: + service.run_forever(self.stop) + # ------------------------------------------------------------ lifecycle def setup(self) -> None: apply_migrations(self.config.dsn) + if self.account_snapshot_service is not None: + # First refresh is synchronous: buy admission never starts from a + # stale row without first attempting the current account. + self.account_snapshot_service.refresh() # dcy-020 (BAT-030): the daily pool preallocation is gone — permanent # asset slots are created/retired by the snapshot scheduler's universe # reconcile (boot fetch runs immediately in its thread). @@ -453,6 +528,8 @@ class Daemon: "safety_sell_enabled=false — trex-safety-sell worker not started. " "Autonomous safety-sell will NOT run." ) + if self.account_snapshot_service is not None: + workers.append(("trex-account-snapshot", self._account_snapshot_loop)) for name, target in workers: t = threading.Thread(target=target, name=name, daemon=True) t.start()