diff --git a/docs/batches/bat-029_20260610_asset-keyed-intent-and-divergence/implementation.md b/docs/batches/bat-029_20260610_asset-keyed-intent-and-divergence/implementation.md index 0264a1a..7cd89a7 100644 --- a/docs/batches/bat-029_20260610_asset-keyed-intent-and-divergence/implementation.md +++ b/docs/batches/bat-029_20260610_asset-keyed-intent-and-divergence/implementation.md @@ -1,16 +1,30 @@ # BAT-029 Implementation -- Status: not-started +- Status: completed (M2(b) 청산 브로커 정본은 BAT-δ carry-over) - Batch: BAT-029 ## 구현 요약 -- (run-batch-delivery 진입 시 작성) design.md 체크포인트(D1/D2) 확정 후 채운다. +- M1(REQ-061 slot_id 계약)·M2(a)(REQ-062 divergence reconcile)·M3(REQ-063 slots 단일 뷰) 구현. M2(b)(execution dispatch 매도 수량 브로커 정본)는 라이브 집행 변경이라 BAT-δ 로 이월. ## 변경 내역 -- (대기) +### M1 — REQ-061 intent 계약 slot_id (커밋 별도) +- BuyIntent slot_id 추가(portfolio_id Optional), `_admit_buy` 자산 중심 우선 분기, `_admit_buy_asset_centric` slot_id(canonical)/portfolio_id(legacy dual-accept) 슬롯 해소 + symbol resolve, messaging buy 파싱 slot_id 수용. + +### M2(a) — REQ-062 divergence reconcile +- `domain/slot/repository.py`·in_memory·postgresql: `list_holding_slots(instance_key)`. +- `application/reconciliation.py`: `reconcile_core_holdings(slot_repo, slot_service, broker, instance_key)` — A(코어 holding/broker 0)=슬롯 error+WARN, C(수량 불일치)=WARN(상태 불변), B(broker 보유/코어 없음)=WARN. 자동 청산 없음(issue #15). +- `application/snapshot_scheduler.py`: run_once 에서 자산 중심 모드 + slot_service 시 reconcile_core_holdings 호출(best-effort). daemon: slot_service 배선. + +### M3 — REQ-063 slots 단일 뷰 +- `application/portfolio.py` `slot_to_dict`: `universe_detached`(retired_at not null) 추가 → `trex-core slots`/envelope 자산 포지션 단일 뷰에 tombstone/soft-retire 표시. + +### M2(b) — carry-over (BAT-δ) +- execution dispatch 매도 수량 `int(slot.qty)` → 브로커 잔고 정본: **라이브 집행 루프 동작 변경**이라 cutover(BAT-δ, 자산 중심 default 전환)와 같은 게이트에서 처리. ## 테스트 -- (대기) +- 신규: test_holding_divergence(4: A/C/ok/B), test_asset_centric_admission(slot_id 매수·unknown_slot 포함 10). +- 회귀: 기존 sell(slot_id)·reconcile_positions·snapshot 무파손. ## 검증 실행 -- (대기) +- 단위 전체: 393 passed / 36 skipped / 1 deselected. ruff clean. +- PG-gated(trex_core_test): persistence 18 passed(list_holding_slots/reconcile 포함). diff --git a/docs/batches/bat-029_20260610_asset-keyed-intent-and-divergence/verification.md b/docs/batches/bat-029_20260610_asset-keyed-intent-and-divergence/verification.md index a1ee38a..4164ccf 100644 --- a/docs/batches/bat-029_20260610_asset-keyed-intent-and-divergence/verification.md +++ b/docs/batches/bat-029_20260610_asset-keyed-intent-and-divergence/verification.md @@ -1,13 +1,43 @@ # BAT-029 Verification -- Status: not-started +- Status: verified (배치 범위; M2(b) BAT-δ 이월) - Batch: BAT-029 ## Acceptance Mapping -- (draft-batch-verification / confirm-batch-verification 단계에서 작성) + +### REQ-061 (intent 계약 slot_id) +| AC | 상태 | Evidence | +|----|------|----------| +| 매수 slot_id 활성화 + symbol 슬롯 resolve | ✅ | test_activates_by_slot_id | +| 이미 활성 → already_active | ✅ | test_already_active_is_rejected | +| 보유 아닌 슬롯 매도 거부 | ✅ | 기존 sell(_admit_sell slot_not_holding) 회귀 | +| 레거시 portfolio_id dual-accept | ✅ | test_activates_asset_slot(portfolio_id) | +| intent_id 멱등 | ✅ | 기존 intent dedup 회귀 | + +### REQ-062 (수량/불일치 해소) +| AC | 상태 | Evidence | +|----|------|----------| +| A 코어 holding/broker 0 → error | ✅ | test_A_core_holding_broker_flat_goes_error | +| B 코어 flat/broker 보유 → warning | ✅ | test_B_broker_holds_unmanaged_warns | +| C 수량만 → 미러 갱신/warning(상태 불변) | ✅ | test_C_qty_mismatch_warns_no_state_change | +| 청산 수량 = 브로커 정본 | ⏳ 이월(BAT-δ) | execution dispatch qty 변경은 라이브 집행 — cutover 게이트 | + +### REQ-063 (운영자 slots 단일 뷰) +| AC | 상태 | Evidence | +|----|------|----------| +| slots 자산 포지션 단일 뷰 | ✅ | slot_to_dict(기존 envelope) | +| universe_detached(tombstone/soft-retire) 표시 | ✅ | slot_to_dict universe_detached + 회귀 | ## Evidence -- (대기) +- 단위 전체: `393 passed / 36 skipped / 1 deselected`. ruff clean. +- PG-gated(실 trex_core_test): persistence `18 passed`(list_holding_slots·reconcile·드리프트 포함). + +## Carried-Over +- **REQ-062 청산 수량 브로커 정본**: execution dispatch 매도 수량(`int(slot.qty)`) → 브로커 잔고 정본 변경은 라이브 집행 동작 → **BAT-δ**(cutover, 자산 중심 default 전환)와 같은 게이트. +- sell `slot_id` 완전 폐기·literal inactive·컬럼 drop·자산 중심 default → BAT-δ. +- 실 데몬 자산 중심 e2e(reconcile→slot_id 매수→divergence) run-the-app → BAT-δ confirm-release 전. +- 따라서 REQ-061/062/063 은 본 배치에서 **Approved 유지**(경로 동작 검증; 전체 완성·default 전환은 BAT-δ). ## Result -- Ready for Release: false (구현 전) +- Ready for Release: false (dcy-020 전체 미완 — γ 경로 도입 완료, M2(b)·default 는 BAT-δ) +- Blocking Issues: 없음(단위 393 + PG 18 green, legacy 무파손). diff --git a/tests/application/test_holding_divergence.py b/tests/application/test_holding_divergence.py new file mode 100644 index 0000000..286a2a3 --- /dev/null +++ b/tests/application/test_holding_divergence.py @@ -0,0 +1,72 @@ +"""dcy-020 / REQ-062 (BAT-029): core holding ↔ broker divergence reconcile.""" +from __future__ import annotations + +from datetime import date + +from trex_core.application.broker_port import BalanceItem +from trex_core.application.reconciliation import reconcile_core_holdings +from trex_core.domain.slot import SlotService, SlotState +from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository + +DAY = date(2026, 6, 10) +INST = "inst-1" + + +class FakeBroker: + def __init__(self, balances): + self._b = balances + def name(self): # noqa: D401 + return "fake" + def submit_order(self, req): ... + def poll_fill(self, *a, **k): ... + def cancel_order(self, *a, **k): ... + def get_balance(self): + return list(self._b) + + +def _hold(repo, svc, pid, symbol, qty): + repo.reconcile_universe(INST, DAY, {pid}) + slot = repo.find_slot_by_portfolio(INST, pid) + svc.transit_slot(slot.slot_id, slot.status, SlotState.PENDING_BUY, + reason="r", actor="a", symbol=symbol, qty=qty) + svc.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, + reason="r", actor="a", avg_price=70_000) + return repo.find_slot_by_portfolio(INST, pid) + + +def _wire(): + repo = InMemorySlotRepository() + return repo, SlotService(repo, instance_key=INST) + + +def test_A_core_holding_broker_flat_goes_error(): + repo, svc = _wire() + slot = _hold(repo, svc, "pf-a", "005930", 10) + broker = FakeBroker([]) # broker holds nothing + d = reconcile_core_holdings(slot_repo=repo, slot_service=svc, broker=broker, instance_key=INST) + assert d[str(slot.slot_id)] == "error" + assert repo.fetch_slot(slot.slot_id).status is SlotState.ERROR # quarantined + + +def test_C_qty_mismatch_warns_no_state_change(): + repo, svc = _wire() + slot = _hold(repo, svc, "pf-a", "005930", 10) + broker = FakeBroker([BalanceItem(symbol="005930", hldg_qty=7)]) # mismatch + d = reconcile_core_holdings(slot_repo=repo, slot_service=svc, broker=broker, instance_key=INST) + assert d[str(slot.slot_id)] == "qty_warning" + assert repo.fetch_slot(slot.slot_id).status is SlotState.HOLDING # unchanged + + +def test_ok_when_qty_matches(): + repo, svc = _wire() + slot = _hold(repo, svc, "pf-a", "005930", 10) + broker = FakeBroker([BalanceItem(symbol="005930", hldg_qty=10)]) + d = reconcile_core_holdings(slot_repo=repo, slot_service=svc, broker=broker, instance_key=INST) + assert d[str(slot.slot_id)] == "ok" + + +def test_B_broker_holds_unmanaged_warns(): + repo, svc = _wire() + broker = FakeBroker([BalanceItem(symbol="000660", hldg_qty=5)]) # core has none + d = reconcile_core_holdings(slot_repo=repo, slot_service=svc, broker=broker, instance_key=INST) + assert d["broker:000660"] == "unmanaged_warning" diff --git a/trex_core/adapters/persistence/postgresql.py b/trex_core/adapters/persistence/postgresql.py index e1c57f0..c643e82 100644 --- a/trex_core/adapters/persistence/postgresql.py +++ b/trex_core/adapters/persistence/postgresql.py @@ -154,6 +154,16 @@ class PostgreSQLSlotRepository: row = cur.fetchone() return int(row[0]) if row else 0 + def list_holding_slots(self, instance_key: str) -> list[Slot]: + with self._psycopg.connect(self._dsn) as conn: + with conn.cursor() as cur: + cur.execute( + f"SELECT {_SLOT_COLUMNS} FROM daily_slot " + "WHERE instance_key = %s AND status = 'holding'", + (instance_key,), + ) + return [self._row_to_slot(r) for r in cur.fetchall()] + def reconcile_universe( self, instance_key: str, trading_date: date, universe_portfolio_ids: set[str], *, allow_retire: bool = True, diff --git a/trex_core/application/portfolio.py b/trex_core/application/portfolio.py index c4ef6ca..82cc718 100644 --- a/trex_core/application/portfolio.py +++ b/trex_core/application/portfolio.py @@ -319,6 +319,9 @@ def slot_to_dict(s) -> dict: "sell_strategy_id": s.sell_strategy_id, "portfolio_id": s.portfolio_id, "forced_close_at": s.forced_close_at.isoformat() if s.forced_close_at else None, + # dcy-020 / REQ-063 (BAT-029): asset-centric single view — universe-detached + # (soft-retired/tombstone) marker so the operator sees retired assets. + "universe_detached": getattr(s, "retired_at", None) is not None, } diff --git a/trex_core/application/reconciliation.py b/trex_core/application/reconciliation.py index 6b87e60..47a1ca2 100644 --- a/trex_core/application/reconciliation.py +++ b/trex_core/application/reconciliation.py @@ -12,10 +12,62 @@ import logging from trex_core.application.broker_port import BrokerPort from trex_core.application.portfolio import SnapshotStore +from trex_core.domain.slot.state import SlotState _log = logging.getLogger(__name__) +def reconcile_core_holdings(*, slot_repo, slot_service, broker: BrokerPort, + instance_key: str) -> dict[str, str]: + """dcy-020 / REQ-062 (BAT-029): surface core-slot holding ↔ broker divergence. + + Broker balance is the reconcile ground truth. Decisions (no silent auto-fix, + no auto-liquidation — issue #15): + - A. core holding but broker flat (qty 0) -> slot to `error` + WARN (operator resolve) + - C. both hold but qty differs -> WARN (mirror is non-authoritative) + - B. broker holds, no core holding slot -> WARN (unmanaged position) + Returns {key: 'error'|'qty_warning'|'ok'|'unmanaged_warning'} for telemetry. + (Liquidation qty = broker authoritative in execution dispatch is BAT-δ.)""" + broker_qty: dict[str, int] = { + item.symbol: int(item.hldg_qty) for item in broker.get_balance() + } + decisions: dict[str, str] = {} + matched: set[str] = set() + for slot in slot_repo.list_holding_slots(instance_key): + sym = slot.symbol + matched.add(sym) + bq = broker_qty.get(sym, 0) + if bq == 0: # A — phantom holding + try: + slot_service.transit_slot( + slot.slot_id, SlotState.HOLDING, SlotState.ERROR, + reason="reconcile_mismatch:broker_flat", actor="reconcile", + ) + decisions[str(slot.slot_id)] = "error" + _log.warning( + "core holding but broker flat: slot=%s symbol=%s -> error", + slot.slot_id, sym, + ) + except Exception: # noqa: BLE001 - never let reconcile crash the loop + _log.exception("failed to error-quarantine slot %s", slot.slot_id) + elif slot.qty is not None and int(slot.qty) != bq: # C — qty mismatch + decisions[str(slot.slot_id)] = "qty_warning" + _log.warning( + "qty mismatch: slot=%s symbol=%s core=%s broker=%d (mirror non-authoritative)", + slot.slot_id, sym, slot.qty, bq, + ) + else: + decisions[str(slot.slot_id)] = "ok" + for sym, bq in broker_qty.items(): # B — unmanaged broker position + if bq > 0 and sym not in matched: + decisions[f"broker:{sym}"] = "unmanaged_warning" + _log.warning( + "broker holds %s qty=%d with no core holding slot (unmanaged position)", + sym, bq, + ) + return decisions + + def reconcile_positions(*, store: SnapshotStore, broker: BrokerPort) -> dict[str, str]: """Run one reconciliation pass. diff --git a/trex_core/application/snapshot_scheduler.py b/trex_core/application/snapshot_scheduler.py index e0a388c..949f597 100644 --- a/trex_core/application/snapshot_scheduler.py +++ b/trex_core/application/snapshot_scheduler.py @@ -38,6 +38,7 @@ class SnapshotScheduler: schedule_kst: time = time(hour=7, minute=30), clock: Optional[Callable[[], datetime]] = None, slot_repo=None, + slot_service=None, instance_key: str = "default", reconcile_enabled: bool = False, trading_date_fn: Optional[Callable[[], object]] = None, @@ -52,6 +53,7 @@ class SnapshotScheduler: # dcy-020 / REQ-058 (BAT-028): asset-centric slot reconcile after each # universe fetch (only when enabled = asset-centric mode). self._slot_repo = slot_repo + self._slot_service = slot_service self._instance_key = instance_key self._reconcile_enabled = reconcile_enabled self._trading_date_fn = trading_date_fn @@ -115,6 +117,15 @@ class SnapshotScheduler: self._reconcile_slots(items) except Exception: # noqa: BLE001 _log.exception("slot reconcile pass failed") + if self._slot_service is not None: + try: + from trex_core.application.reconciliation import reconcile_core_holdings + reconcile_core_holdings( + slot_repo=self._slot_repo, slot_service=self._slot_service, + broker=self._broker, instance_key=self._instance_key, + ) + except Exception: # noqa: BLE001 + _log.exception("core-holding divergence reconcile failed") self._last_fired_date = self._clock() return True diff --git a/trex_core/domain/slot/in_memory_repository.py b/trex_core/domain/slot/in_memory_repository.py index 12063cb..8971cb0 100644 --- a/trex_core/domain/slot/in_memory_repository.py +++ b/trex_core/domain/slot/in_memory_repository.py @@ -115,6 +115,12 @@ class InMemorySlotRepository: if s.instance_key == instance_key and s.status in self._ACTIVE ) + def list_holding_slots(self, instance_key: str) -> list[Slot]: + return [ + s for s in self._slots.values() + if s.instance_key == instance_key and s.status is SlotState.HOLDING + ] + def reconcile_universe( self, instance_key: str, trading_date: date, universe_portfolio_ids: set[str], *, allow_retire: bool = True, diff --git a/trex_core/domain/slot/repository.py b/trex_core/domain/slot/repository.py index 9008069..e67dfdf 100644 --- a/trex_core/domain/slot/repository.py +++ b/trex_core/domain/slot/repository.py @@ -75,6 +75,11 @@ class SlotRepository(Protocol): {holding, pending_buy, pending_sell} (dcy-020/REQ-059 M cap, D2).""" ... + def list_holding_slots(self, instance_key: str) -> list[Slot]: + """Return holding slots for this instance (dcy-020/REQ-062 divergence + reconcile).""" + ... + def reconcile_universe( self, instance_key: str, trading_date: date, universe_portfolio_ids: set[str], *, allow_retire: bool = True, diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index 16ddf02..4b8c8f6 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -176,8 +176,10 @@ class Daemon: store=self.snapshot_store, broker=self.broker, schedule_kst=_parse_kst_time(config.snapshot_schedule.kst_time), - # dcy-020 / REQ-058 (BAT-028): asset-centric slot reconcile when M set. + # dcy-020 / REQ-058·062 (BAT-028/029): asset-centric slot reconcile + + # core-holding divergence reconcile when M set. slot_repo=self.slot_repo, + slot_service=self.slot_service, instance_key=config.instance_key, reconcile_enabled=config.max_active_slots is not None, trading_date_fn=(lambda: datetime.now(ZoneInfo(config.trading_timezone)).date()),