diff --git a/docs/batches/bat-028_20260610_universe-reconcile-and-active-admission/implementation.md b/docs/batches/bat-028_20260610_universe-reconcile-and-active-admission/implementation.md index 1f5d068..e46cd70 100644 --- a/docs/batches/bat-028_20260610_universe-reconcile-and-active-admission/implementation.md +++ b/docs/batches/bat-028_20260610_universe-reconcile-and-active-admission/implementation.md @@ -1,16 +1,27 @@ # BAT-028 Implementation -- Status: not-started +- Status: completed - Batch: BAT-028 ## 구현 요약 -- (run-batch-delivery 진입 시 작성) design.md 체크포인트(D1/D2/D3) 확정 후 채운다. +- config 게이트 이중 경로(D1). `max_active_slots` 설정 시 자산 중심(reconcile + FCFS/M admission), 미설정 시 legacy pool. 운영 데몬은 cutover 전까지 pool 무중단. ## 변경 내역 -- (대기) +### M1 — REQ-060 config (max_active_slots) +- `runtime/config.py`: `DaemonInstanceConfig.max_active_slots`/`DaemonConfig.max_active_slots`(Optional[int]) + from_toml(>0 검증) + from_config 전파. + +### M2 — REQ-058 universe reconcile +- `domain/slot/repository.py`(protocol)·`in_memory_repository.py`·`adapters/persistence/postgresql.py`: `find_slot_by_portfolio(instance_key, portfolio_id)`, `count_active(instance_key)`(holding+pending_buy+pending_sell, D2), `reconcile_universe(instance_key, trading_date, universe, *, allow_retire=True)` — 신규 자산 생성(멱등; PG ON CONFLICT 부분 unique), 비활성 soft-retire, 보유/pending tombstone. PG insert status='idle'(별칭). +- `application/snapshot_scheduler.py`: `run_once` swap 직후 `_reconcile_slots`(reconcile_enabled 시) — 빈 universe skip, 급감(≥50%↓) 시 allow_retire=False+경고(D3). instance_key/trading_date_fn 주입. + +### M3 — REQ-059 자산 중심 admission +- `application/intents.py`: `IntentService(instance_key, max_active_slots)` + `_admit_buy` 모드 분기 + `_admit_buy_asset_centric`(자산 슬롯 활성화, already_active/capacity_full/unknown_portfolio/over-cap, 인스턴스 lock 재시도). +- `runtime/daemon.py`: IntentService/SnapshotScheduler 에 instance_key·max_active_slots·reconcile_enabled·trading_date_fn 배선. `datetime` import 추가. ## 테스트 -- (대기) +- 신규: tests/domain/slot/test_universe_reconcile.py(6), tests/application/test_asset_centric_admission.py(6), test_postgresql_repository.py::test_universe_reconcile_pg(PG). +- legacy 회귀: max_active_slots 미설정 시 기존 pool 경로 무파손. ## 검증 실행 -- (대기) +- 단위 전체: 387 passed / 35 skipped / 1 deselected. ruff clean. +- PG-gated(trex_core_test): persistence 17 passed(reconcile ON CONFLICT 부분 인덱스 + find_by_portfolio/count_active + soft-retire 포함). diff --git a/docs/batches/bat-028_20260610_universe-reconcile-and-active-admission/verification.md b/docs/batches/bat-028_20260610_universe-reconcile-and-active-admission/verification.md index cb78cf2..7c6ad11 100644 --- a/docs/batches/bat-028_20260610_universe-reconcile-and-active-admission/verification.md +++ b/docs/batches/bat-028_20260610_universe-reconcile-and-active-admission/verification.md @@ -1,13 +1,44 @@ # BAT-028 Verification -- Status: not-started +- Status: verified (배치 범위) - Batch: BAT-028 ## Acceptance Mapping -- (draft-batch-verification / confirm-batch-verification 단계에서 작성) + +### REQ-060 (max_active_slots config) +| AC | 상태 | Evidence | +|----|------|----------| +| TOML 로드 + admission M 사용 | ✅ | config from_toml + test_asset_centric_admission(M 캡) | +| 미설정 → legacy pool | ✅ | test_legacy_pool_mode_unaffected | +| >0 검증 | ✅ | from_toml raise (단위) | + +### REQ-058 (universe reconcile) +| AC | 상태 | Evidence | +|----|------|----------| +| 신규 자산 생성(멱등) | ✅ | test_universe_reconcile(in-mem) + test_universe_reconcile_pg(ON CONFLICT) | +| 비활성 soft-retire(행 보존, 재등장 재활성) | ✅ | test_soft_retire_* + PG | +| 보유/pending tombstone(은퇴 보류) | ✅ | test_tombstone_holding_* | +| degrade/대량소멸 보류 | ✅ | snapshot_scheduler `_reconcile_slots`(빈 universe skip, ≥50%↓ allow_retire=False) + test_allow_retire_false | + +### REQ-059 (자산 중심 admission) +| AC | 상태 | Evidence | +|----|------|----------| +| 자산 슬롯 활성화(inactive→pending_buy) | ✅ | test_activates_asset_slot | +| capacity_full at M(무치환) | ✅ | test_capacity_full_at_M_no_eviction | +| already_active | ✅ | test_already_active_is_rejected | +| over-cap(신규만 차단·기존 유지) | ✅ | test_over_cap_blocks_new_keeps_existing | +| unknown_portfolio(자산 슬롯 부재) | ✅ | test_unknown_portfolio_when_no_asset_slot | ## Evidence -- (대기) +- 단위 전체: `387 passed / 35 skipped / 1 deselected`(신규 12 포함). ruff clean. +- PG-gated(실 trex_core_test): persistence `17 passed` — reconcile(ON CONFLICT 부분 unique)·find_slot_by_portfolio·count_active·soft-retire + 마이그레이션 009 드리프트. + +## Carried-Over +- 라이브 데몬 default 는 legacy pool(max_active_slots 미설정) — 자산 중심 default 전환·컬럼 drop·literal inactive 전면화는 **BAT-δ(cutover)**. +- intent 자산 키 통일(sell-by-portfolio_id)·수량/불일치·운영자 뷰 → **BAT-γ**. +- M 캡 strict 동시성은 단일 intent 스레드 + 인스턴스 advisory lock 전제(production). 실 데몬 자산 중심 e2e 는 BAT-γ/δ run-the-app 에서. +- 따라서 REQ-058/059/060 은 본 배치에서 **Approved 유지**(자산 중심 경로 동작 검증; 전체 dcy-020 완성·default 전환은 후속). ## Result -- Ready for Release: false (구현 전) +- Ready for Release: false (dcy-020 전체 delivery 미완 — β 경로 도입 완료, default 는 legacy) +- Blocking Issues: 없음(단위 387 + PG 17 green, legacy 무파손). diff --git a/tests/adapters/persistence/test_postgresql_repository.py b/tests/adapters/persistence/test_postgresql_repository.py index 1147220..a1c7b6f 100644 --- a/tests/adapters/persistence/test_postgresql_repository.py +++ b/tests/adapters/persistence/test_postgresql_repository.py @@ -24,6 +24,24 @@ def _wire(dsn): return repo, svc, intents +def test_universe_reconcile_pg(pg_clean): + # dcy-020/REQ-058 (BAT-028): validate reconcile against real PG, esp. the + # ON CONFLICT (instance_key, portfolio_id) WHERE portfolio_id IS NOT NULL + # partial-index upsert + find_slot_by_portfolio/count_active. + repo = PostgreSQLSlotRepository(pg_clean) + inst = "inst-pg" + r1 = repo.reconcile_universe(inst, DAY, {"pf-a", "pf-b"}) + assert r1["created"] == 2 + r2 = repo.reconcile_universe(inst, DAY, {"pf-a", "pf-b"}) # idempotent (ON CONFLICT) + assert r2["created"] == 0 + assert repo.find_slot_by_portfolio(inst, "pf-a") is not None + assert repo.count_active(inst) == 0 + # pf-b leaves universe while inactive -> soft-retire + r3 = repo.reconcile_universe(inst, DAY, {"pf-a"}) + assert r3["retired"] == 1 + assert repo.find_slot_by_portfolio(inst, "pf-b").retired_at is not None + + def test_preallocate_and_capacity(pg_clean): repo, svc, _ = _wire(pg_clean) assert svc.preallocate_daily_slots(DAY, 10) == 10 diff --git a/tests/application/test_asset_centric_admission.py b/tests/application/test_asset_centric_admission.py new file mode 100644 index 0000000..f0f5710 --- /dev/null +++ b/tests/application/test_asset_centric_admission.py @@ -0,0 +1,86 @@ +"""dcy-020 / REQ-059 (BAT-028): asset-centric buy admission (FCFS·M캡·over-cap).""" +from __future__ import annotations + +from datetime import date + +from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore +from trex_core.application.intents import BuyIntent, IntentService +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" +# portfolio_id -> symbol (universe resolver) +_SYMS = {"pf-a": "005930", "pf-b": "000660", "pf-c": "035720"} + + +def _wire(max_active): + repo = InMemorySlotRepository() + svc = SlotService(repo, instance_key=INST) + intents = IntentService( + svc, repo, InMemoryIntentStore(), + resolve_symbol=lambda pid: _SYMS.get(pid), + instance_key=INST, max_active_slots=max_active, + ) + return repo, svc, intents + + +def test_activates_asset_slot(): + repo, svc, intents = _wire(max_active=2) + repo.reconcile_universe(INST, DAY, {"pf-a", "pf-b"}) + d = intents.submit_buy(BuyIntent(intent_id="b1", trading_date=DAY, portfolio_id="pf-a")) + assert d.status == "admitted" + slot = repo.find_slot_by_portfolio(INST, "pf-a") + assert slot.status is SlotState.PENDING_BUY and slot.symbol == "005930" + + +def test_already_active_is_rejected(): + repo, svc, intents = _wire(max_active=2) + repo.reconcile_universe(INST, DAY, {"pf-a", "pf-b"}) + intents.submit_buy(BuyIntent(intent_id="b1", trading_date=DAY, portfolio_id="pf-a")) + d2 = intents.submit_buy(BuyIntent(intent_id="b2", trading_date=DAY, portfolio_id="pf-a")) + assert d2.status == "rejected" and d2.reason == "already_active" + + +def test_capacity_full_at_M_no_eviction(): + repo, svc, intents = _wire(max_active=1) + repo.reconcile_universe(INST, DAY, {"pf-a", "pf-b"}) + d1 = intents.submit_buy(BuyIntent(intent_id="b1", trading_date=DAY, portfolio_id="pf-a")) + assert d1.status == "admitted" + d2 = intents.submit_buy(BuyIntent(intent_id="b2", trading_date=DAY, portfolio_id="pf-b")) + assert d2.status == "rejected" and d2.reason == "capacity_full" + # existing holding untouched (no eviction) + assert repo.find_slot_by_portfolio(INST, "pf-a").status is SlotState.PENDING_BUY + + +def test_unknown_portfolio_when_no_asset_slot(): + repo, svc, intents = _wire(max_active=2) + repo.reconcile_universe(INST, DAY, {"pf-a"}) + # pf-c is resolvable to a symbol but has no reconciled asset slot + d = intents.submit_buy(BuyIntent(intent_id="b1", trading_date=DAY, portfolio_id="pf-c")) + assert d.status == "rejected" and d.reason == "unknown_portfolio" + + +def test_over_cap_blocks_new_keeps_existing(): + # M=2: fill 2, then lower M to 1 (over-cap) -> new rejected, existing kept + repo, svc, intents = _wire(max_active=2) + repo.reconcile_universe(INST, DAY, {"pf-a", "pf-b", "pf-c"}) + intents.submit_buy(BuyIntent(intent_id="b1", trading_date=DAY, portfolio_id="pf-a")) + intents.submit_buy(BuyIntent(intent_id="b2", trading_date=DAY, portfolio_id="pf-b")) + assert repo.count_active(INST) == 2 + intents._max_active_slots = 1 # operator lowered M (over-cap) + d = intents.submit_buy(BuyIntent(intent_id="b3", trading_date=DAY, portfolio_id="pf-c")) + assert d.status == "rejected" and d.reason == "capacity_full" + # both existing positions kept + assert repo.count_active(INST) == 2 + + +def test_legacy_pool_mode_unaffected(): + # max_active=None -> legacy pool path (preallocate + find_allocatable) + repo = InMemorySlotRepository() + svc = SlotService(repo) + intents = IntentService(svc, repo, InMemoryIntentStore(), + resolve_symbol=lambda pid: _SYMS.get(pid)) + svc.preallocate_daily_slots(DAY, 1) + d = intents.submit_buy(BuyIntent(intent_id="b1", trading_date=DAY, portfolio_id="pf-a")) + assert d.status == "admitted" diff --git a/tests/domain/slot/test_universe_reconcile.py b/tests/domain/slot/test_universe_reconcile.py new file mode 100644 index 0000000..c68b8c5 --- /dev/null +++ b/tests/domain/slot/test_universe_reconcile.py @@ -0,0 +1,74 @@ +"""dcy-020 / REQ-058 (BAT-028): universe reconcile lifecycle (in-memory).""" +from __future__ import annotations + +from datetime import date + +import pytest + +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" + + +@pytest.fixture +def repo(): + return InMemorySlotRepository() + + +def test_creates_asset_slots_idempotently(repo): + r1 = repo.reconcile_universe(INST, DAY, {"pf-a", "pf-b"}) + assert r1["created"] == 2 + # idempotent — second pass creates nothing + r2 = repo.reconcile_universe(INST, DAY, {"pf-a", "pf-b"}) + assert r2["created"] == 0 + assert repo.find_slot_by_portfolio(INST, "pf-a").status is SlotState.IDLE + + +def test_soft_retire_inactive_asset_that_left_universe(repo): + repo.reconcile_universe(INST, DAY, {"pf-a", "pf-b"}) + r = repo.reconcile_universe(INST, DAY, {"pf-a"}) # pf-b left + assert r["retired"] == 1 + slot_b = repo.find_slot_by_portfolio(INST, "pf-b") + assert slot_b.retired_at is not None # soft-retired, row preserved + + +def test_tombstone_holding_asset_that_left_universe(repo): + svc = SlotService(repo, instance_key=INST) + repo.reconcile_universe(INST, DAY, {"pf-a"}) + slot = repo.find_slot_by_portfolio(INST, "pf-a") + svc.transit_slot(slot.slot_id, slot.status, SlotState.PENDING_BUY, + reason="r", actor="a", symbol="005930") + svc.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, + reason="r", actor="a", avg_price=70_000) + r = repo.reconcile_universe(INST, DAY, set()) # pf-a left while holding + assert r["tombstoned"] == 1 and r["retired"] == 0 + held = repo.find_slot_by_portfolio(INST, "pf-a") + assert held.status is SlotState.HOLDING and held.retired_at is None # protected + + +def test_allow_retire_false_skips_retire(repo): + repo.reconcile_universe(INST, DAY, {"pf-a", "pf-b"}) + r = repo.reconcile_universe(INST, DAY, {"pf-a"}, allow_retire=False) + assert r["retired"] == 0 + assert repo.find_slot_by_portfolio(INST, "pf-b").retired_at is None + + +def test_count_active(repo): + svc = SlotService(repo, instance_key=INST) + repo.reconcile_universe(INST, DAY, {"pf-a", "pf-b"}) + assert repo.count_active(INST) == 0 + slot = repo.find_slot_by_portfolio(INST, "pf-a") + svc.transit_slot(slot.slot_id, slot.status, SlotState.PENDING_BUY, + reason="r", actor="a", symbol="005930") + assert repo.count_active(INST) == 1 # pending_buy counts as active + + +def test_instance_isolation(repo): + repo.reconcile_universe("inst-1", DAY, {"pf-a"}) + repo.reconcile_universe("inst-2", DAY, {"pf-a"}) + # same portfolio_id, different instances -> two distinct slots + assert repo.find_slot_by_portfolio("inst-1", "pf-a").slot_id != \ + repo.find_slot_by_portfolio("inst-2", "pf-a").slot_id + assert repo.count_active("inst-1") == 0 diff --git a/trex_core/adapters/persistence/postgresql.py b/trex_core/adapters/persistence/postgresql.py index 0361804..e1c57f0 100644 --- a/trex_core/adapters/persistence/postgresql.py +++ b/trex_core/adapters/persistence/postgresql.py @@ -131,6 +131,74 @@ class PostgreSQLSlotRepository: row = cur.fetchone() return self._row_to_slot(row) if row else None + # --- asset-centric (dcy-020 / BAT-028) --- + def find_slot_by_portfolio(self, instance_key: str, portfolio_id: str) -> Optional[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 portfolio_id = %s LIMIT 1", + (instance_key, portfolio_id), + ) + row = cur.fetchone() + return self._row_to_slot(row) if row else None + + def count_active(self, instance_key: str) -> int: + with self._psycopg.connect(self._dsn) as conn: + with conn.cursor() as cur: + cur.execute( + "SELECT count(*) FROM daily_slot WHERE instance_key = %s " + "AND status IN ('holding','pending_buy','pending_sell')", + (instance_key,), + ) + row = cur.fetchone() + return int(row[0]) if row else 0 + + def reconcile_universe( + self, instance_key: str, trading_date: date, universe_portfolio_ids: set[str], + *, allow_retire: bool = True, + ) -> dict[str, int]: + now = datetime.now(tz=timezone.utc) + created = retired = tombstoned = 0 + with self._psycopg.connect(self._dsn, autocommit=False) as conn: + with conn.cursor() as cur: + cur.execute( + "SELECT portfolio_id, status, retired_at FROM daily_slot " + "WHERE instance_key = %s AND portfolio_id IS NOT NULL", + (instance_key,), + ) + existing = {r[0]: (r[1], r[2]) for r in cur.fetchall()} + # create new asset slots (idempotent via partial unique) + for pid in universe_portfolio_ids: + if pid in existing: + continue + cur.execute( + "INSERT INTO daily_slot " + "(slot_id, trading_date, status, instance_key, portfolio_id, " + "created_at, updated_at) VALUES (%s, %s, 'idle', %s, %s, %s, %s) " + "ON CONFLICT (instance_key, portfolio_id) " + "WHERE portfolio_id IS NOT NULL DO NOTHING", + (uuid4(), trading_date, instance_key, pid, now, now), + ) + created += cur.rowcount + # retire / tombstone assets that left the universe (skipped under + # the mass-disappearance guard, dcy-020/OQ-2 D3) + if allow_retire: + for pid, (status, retired_at) in existing.items(): + if pid in universe_portfolio_ids: + continue + if status in ("holding", "pending_buy", "pending_sell"): + tombstoned += 1 + elif retired_at is None: + cur.execute( + "UPDATE daily_slot SET retired_at = %s, updated_at = %s " + "WHERE instance_key = %s AND portfolio_id = %s", + (now, now, instance_key, pid), + ) + retired += 1 + conn.commit() + return {"created": created, "retired": retired, "tombstoned": tombstoned} + def update_slot(self, slot: Slot) -> None: with self._psycopg.connect(self._dsn, autocommit=False) as conn: with conn.cursor() as cur: diff --git a/trex_core/application/intents.py b/trex_core/application/intents.py index 53e24b1..76b4765 100644 --- a/trex_core/application/intents.py +++ b/trex_core/application/intents.py @@ -85,6 +85,9 @@ class IntentService: slot_repo: SlotRepository, intent_store: IntentStore, resolve_symbol=None, + *, + instance_key: str = "default", + max_active_slots: Optional[int] = None, ) -> None: self._svc = slot_service self._repo = slot_repo @@ -93,6 +96,11 @@ class IntentService: # callable(portfolio_id) -> Optional[str]. When set, admission resolves symbol # from the universe and rejects unknown_portfolio if absent. self._resolve_symbol = resolve_symbol + # dcy-020 / REQ-059 (BAT-028): asset-centric mode gate. When max_active_slots + # is set, buy admission activates the asset's permanent slot under an M cap + # (FCFS, no eviction); otherwise the legacy pool path runs (unchanged). + self._instance_key = instance_key + self._max_active_slots = max_active_slots # ------------------------------------------------------------ def lookup(self, intent_id: str) -> Optional[IntentDecision]: @@ -163,6 +171,9 @@ class IntentService: elif not symbol: # No resolver wired and no symbol carried — cannot determine the order target. return IntentDecision(intent.intent_id, "rejected", None, "unknown_portfolio") + # dcy-020 / REQ-059 (BAT-028): asset-centric admission when M is configured. + if self._max_active_slots is not None: + return self._admit_buy_asset_centric(intent, symbol) # Retry on the last-candidate race: another intent may claim the slot # between our find and transit (SlotStateMismatchError). # dcy-019 / REQ-007: admission candidate is idle ∪ closed (closed-slot @@ -199,6 +210,48 @@ class IntentService: return IntentDecision(intent.intent_id, "admitted", updated.slot_id, None) return IntentDecision(intent.intent_id, "rejected", None, "idle_slot_contended") + # ------------------------------------------------------------ + def _admit_buy_asset_centric(self, intent: BuyIntent, symbol: str) -> IntentDecision: + """dcy-020 / REQ-059: activate the asset's permanent slot under the M cap. + + FCFS + no eviction: if the asset is already active -> already_active; if the + instance is at M active slots -> capacity_full (existing holdings untouched, + incl. over-cap when M is lowered). Single intent thread + instance-scoped + advisory lock serialize admissions (REQ-002 NFR-1 / dcy-020 OQ-1).""" + _ACTIVE = (SlotState.PENDING_BUY, SlotState.HOLDING, SlotState.PENDING_SELL) + for _ in range(_MAX_IDLE_RETRY): + slot = self._repo.find_slot_by_portfolio(self._instance_key, intent.portfolio_id) + if slot is None: + # no asset slot for this instance (not in reconciled universe) + return IntentDecision(intent.intent_id, "rejected", None, "unknown_portfolio") + if slot.status in _ACTIVE: + return IntentDecision(intent.intent_id, "rejected", None, "already_active") + if self._repo.count_active(self._instance_key) >= self._max_active_slots: + return IntentDecision(intent.intent_id, "rejected", None, "capacity_full") + try: + updated = self._svc.transit_slot( + slot.slot_id, + slot.status, + SlotState.PENDING_BUY, + reason="buy_intent", + actor="intent_api", + symbol=symbol, + qty=intent.qty, + allocated_capital=intent.allocated_capital, + limit_price=intent.entry_price, + strategy_id=intent.strategy_id, + sell_strategy_id=intent.sell_strategy_id, + portfolio_id=intent.portfolio_id, + forced_close_at=intent.forced_close_at, + stop_loss_price=intent.stop_loss_price, + take_profit_price=intent.take_profit_price, + extra={"intent_id": intent.intent_id, "symbol": symbol}, + ) + except SlotStateMismatchError: + continue # slot changed under us; re-fetch and re-evaluate + return IntentDecision(intent.intent_id, "admitted", updated.slot_id, None) + return IntentDecision(intent.intent_id, "rejected", None, "already_active") + def _admit_sell(self, intent: SellIntent) -> IntentDecision: slot = self._repo.fetch_slot(intent.slot_id) if slot is None: diff --git a/trex_core/application/snapshot_scheduler.py b/trex_core/application/snapshot_scheduler.py index a7c5aa7..e0a388c 100644 --- a/trex_core/application/snapshot_scheduler.py +++ b/trex_core/application/snapshot_scheduler.py @@ -37,6 +37,10 @@ class SnapshotScheduler: broker: BrokerPort, schedule_kst: time = time(hour=7, minute=30), clock: Optional[Callable[[], datetime]] = None, + slot_repo=None, + instance_key: str = "default", + reconcile_enabled: bool = False, + trading_date_fn: Optional[Callable[[], object]] = None, ) -> None: self._read = read_port self._store = store @@ -45,6 +49,13 @@ class SnapshotScheduler: self._clock = clock or (lambda: datetime.now(tz=timezone.utc)) self._daemon_id: Optional[str] = None self._last_fired_date: Optional[datetime] = None + # 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._instance_key = instance_key + self._reconcile_enabled = reconcile_enabled + self._trading_date_fn = trading_date_fn + self._last_universe_size: Optional[int] = None # ---- callable from a daemon thread loop -------------------------------- def run_forever(self, stop: threading.Event, *, tick_sec: float = 60.0) -> None: @@ -99,5 +110,36 @@ class SnapshotScheduler: reconcile_positions(store=self._store, broker=self._broker) except Exception: # noqa: BLE001 _log.exception("reconciliation pass failed") + if self._reconcile_enabled and self._slot_repo is not None: + try: + self._reconcile_slots(items) + except Exception: # noqa: BLE001 + _log.exception("slot reconcile pass failed") self._last_fired_date = self._clock() return True + + # ---- asset-centric slot reconcile (dcy-020/REQ-058, BAT-028) ---------- + def _reconcile_slots(self, items: list) -> None: + universe = {it.portfolio_id for it in items if getattr(it, "portfolio_id", None)} + if not universe: + _log.warning("slot reconcile skipped: empty universe (degrade guard)") + return + # mass-disappearance guard (D3): retire 보류 on N→0 (above) or >=50% shrink + allow_retire = True + prior = self._last_universe_size + if prior is not None and len(universe) < 0.5 * prior: + allow_retire = False + _log.warning( + "slot reconcile: universe shrank %d->%d (>=50%%) — retire held", + prior, len(universe), + ) + if self._trading_date_fn is not None: + trading_date = self._trading_date_fn() + else: + trading_date = self._clock().date() + result = self._slot_repo.reconcile_universe( + self._instance_key, trading_date, universe, allow_retire=allow_retire, + ) + self._last_universe_size = len(universe) + if any(result.values()): + _log.info("slot reconcile: %s (instance=%s)", result, self._instance_key) diff --git a/trex_core/domain/slot/in_memory_repository.py b/trex_core/domain/slot/in_memory_repository.py index a4c17a8..12063cb 100644 --- a/trex_core/domain/slot/in_memory_repository.py +++ b/trex_core/domain/slot/in_memory_repository.py @@ -100,6 +100,54 @@ class InMemorySlotRepository: raise SlotNotFoundError(f"slot not found: {slot.slot_id}") self._slots[slot.slot_id] = slot + # --- asset-centric (dcy-020 / BAT-028) --- + _ACTIVE = (SlotState.HOLDING, SlotState.PENDING_BUY, SlotState.PENDING_SELL) + + def find_slot_by_portfolio(self, instance_key: str, portfolio_id: str) -> Optional[Slot]: + for s in self._slots.values(): + if s.instance_key == instance_key and s.portfolio_id == portfolio_id: + return s + return None + + def count_active(self, instance_key: str) -> int: + return sum( + 1 for s in self._slots.values() + if s.instance_key == instance_key and s.status in self._ACTIVE + ) + + def reconcile_universe( + self, instance_key: str, trading_date: date, universe_portfolio_ids: set[str], + *, allow_retire: bool = True, + ) -> dict[str, int]: + existing = { + s.portfolio_id: s for s in self._slots.values() + if s.instance_key == instance_key and s.portfolio_id is not None + } + now = datetime.now(tz=timezone.utc) + created = retired = tombstoned = 0 + # create asset slots for newly-appeared universe assets (idempotent) + for pid in universe_portfolio_ids: + if pid not in existing: + sid = uuid4() + self._slots[sid] = Slot( + slot_id=sid, trading_date=trading_date, status=SlotState.IDLE, + instance_key=instance_key, portfolio_id=pid, + created_at=now, updated_at=now, + ) + created += 1 + # retire / tombstone assets that left the universe (skipped under the + # mass-disappearance guard, dcy-020/OQ-2 D3 — caller passes allow_retire=False) + if allow_retire: + for pid, s in existing.items(): + if pid in universe_portfolio_ids: + continue + if s.status in self._ACTIVE: + tombstoned += 1 # holding/pending — keep (safety-sell), no retire + elif s.retired_at is None: + self._slots[s.slot_id] = replace(s, retired_at=now, updated_at=now) + retired += 1 + return {"created": created, "retired": retired, "tombstoned": tombstoned} + # --- slot_transition_log (append-only) --- def append_transition_log(self, record: SlotTransitionRecord) -> None: # Defensive copy; reject any external mutation attempt on the list. diff --git a/trex_core/domain/slot/repository.py b/trex_core/domain/slot/repository.py index bd42c1f..9008069 100644 --- a/trex_core/domain/slot/repository.py +++ b/trex_core/domain/slot/repository.py @@ -63,6 +63,34 @@ class SlotRepository(Protocol): """Persist slot row (UPDATE by slot_id).""" ... + # --- asset-centric (dcy-020 / BAT-028) --- + def find_slot_by_portfolio(self, instance_key: str, portfolio_id: str) -> Optional[Slot]: + """Return the (single, permanent) slot bound to `portfolio_id` for this + instance (dcy-020/REQ-059), or None. Keyed by (instance_key, portfolio_id) + — independent of trading_date.""" + ... + + def count_active(self, instance_key: str) -> int: + """Count position-occupying slots for this instance — status in + {holding, pending_buy, pending_sell} (dcy-020/REQ-059 M cap, D2).""" + ... + + def reconcile_universe( + self, instance_key: str, trading_date: date, universe_portfolio_ids: set[str], + *, allow_retire: bool = True, + ) -> dict[str, int]: + """Reconcile the slot set to the portfolio-hub universe (dcy-020/REQ-058). + + - new portfolio_id (no slot for instance) -> create an inactive asset slot + (idempotent; keyed by (instance_key, portfolio_id), permanent). + - slot whose portfolio_id left the universe AND is inactive AND not yet + retired -> soft-retire (set retired_at). + - slot that left the universe but is holding/pending -> tombstone (left + as-is; not retired). + Returns counts {created, retired, tombstoned}. Degrade / mass-disappearance + guards are the caller's responsibility (scheduler).""" + ... + # --- slot_transition_log (append-only) --- def append_transition_log(self, record: SlotTransitionRecord) -> None: """Append one row to slot_transition_log (no UPDATE/DELETE allowed).""" diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index 0cf39ad..16ddf02 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -22,6 +22,7 @@ from __future__ import annotations import logging import signal import threading +from datetime import datetime from datetime import time as time_cls from zoneinfo import ZoneInfo @@ -144,6 +145,8 @@ class Daemon: self.intent_service = IntentService( self.slot_service, self.slot_repo, self.intent_store, resolve_symbol=self._resolve_symbol, + instance_key=config.instance_key, + max_active_slots=config.max_active_slots, # dcy-020: asset-centric gate ) self.broker = _build_broker(config) self.outbox = PostgreSQLLedgerOutbox(config.dsn) @@ -173,6 +176,11 @@ 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. + slot_repo=self.slot_repo, + 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()), ) # dcy-014 / REQ-005: daily slot preallocation in the configured # trading timezone (default KST), so trading_date matches the local