diff --git a/docs/data-model.md b/docs/data-model.md index 1a9c691..138f802 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -175,7 +175,7 @@ ### Key Fields -- account_hash :: TEXT PK :: `SHA-256(CANO|ACNT_PRDT_CD)`. 평문 계좌번호를 저장·로그하지 않는다. +- account_hash :: TEXT PK :: `SHA-256(KIS base_url|CANO|ACNT_PRDT_CD|app_key)`. 실전/모의 및 애플리케이션 자격증명 범위를 분리하며 원문을 저장·로그하지 않는다. - holdings :: JSONB NOT NULL :: `[{symbol,hldg_qty}]`. 소유권 래퍼가 켜진 계정은 전략 소유분만 저장한다. - holdings_captured_at :: TIMESTAMPTZ? :: 보유수량 조회가 마지막 성공한 시각. - available_cash :: BIGINT? :: KIS 주문가능조회 `nrcvb_buy_amt` 현금주문 가능금액. 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 fb1050a..b48c5f2 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 @@ -133,6 +133,9 @@ - `AUTO` 실제 거래소 확정·주문번호 동시 저장·재시작/취소 재사용: 구현 완료. - 취소 `requested|rejected|unknown` 영속화와 poll 전용 최종확정: 구현 완료. - 계정 해시·독립 TTL·백그라운드 단일 소유자·매수/매도 분리: 구현 완료. +- 계정 해시는 KIS endpoint·계좌·상품·app identity를 함께 묶어 실전/모의·자격증명 출처를 분리한다. +- 동일 instance advisory lock 안에서 현금 예약검사와 pending 전이를 원자 처리한다. +- timeout·운영자 취소 모두 compare-and-set 예약으로 동일 주문 재취소를 차단한다. - 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 5ec34dd..b1f6a97 100644 --- a/docs/interface-contract.md +++ b/docs/interface-contract.md @@ -433,6 +433,7 @@ resume_kill_switch 요청·응답 (REQ-046): - 기능은 `TREX_ACCOUNT_SNAPSHOT_ENABLED=false`가 기본인 KIS 전용 선택 기능이다. - KIS 보유조회와 현금주문 가능금액 조회(`inquire-psbl-order.output.nrcvb_buy_amt`)를 기본 300초 간격의 단일 계정 소유자가 갱신한다. 두 값은 독립 캡처 시각을 갖는다. - 신규매수는 미만료 현금 스냅샷과 기존 `pending_buy` 지정가 예약금액을 합산해 검사한다. 수량·지정가가 없거나 예약금액을 계산할 수 없으면 매수를 차단한다. KIS 주문 응답이 최종 현금 한도 권위이며 거부된 주문을 자동 재전송하지 않는다. +- 현금 검사와 `PENDING_BUY` 전이는 동일 instance advisory lock 안에서 수행되어 같은 instance의 동시 처리자가 아직 커밋되지 않은 예약을 함께 통과하지 못한다. - `pending_sell` 수량은 매 tick 실시간 보유조회를 먼저 시도한다. 오직 `transport_failed`이고 기본 3600초 안의 동일계정 스냅샷이 있을 때만 대체한다. - 인증·인가·응답형식·브로커거부·소유권 불일치·계정 해시 변경·만료는 캐시로 우회하지 않는다. - `sync`, `verify`, 일일 스냅샷 스케줄러와 재시작 원장대조는 기존처럼 실시간 조회만 사용한다. @@ -447,6 +448,7 @@ cancel_order(broker_order_id: str, *, side: str, symbol: str, qty: int, venue: s - `venue`: 원주문 venue. 취소 시 원주문과 동일한 `EXCG_ID_DVSN_CD`를 사용해야 한다. 기본값 `"KRX"`로 기존 호출자 하위 호환. - `CancelOutcome.state`: `requested`(요청 접수), `rejected`(명시적 거부), `unknown`(응답유실·전송 실패). `requested`도 최종 취소가 아니며 `poll_fill()`의 `cancelled|filled`만 주문의 최종 상태를 확정한다. - Core는 외부 취소 호출 전에 `unknown`을 먼저 영속화하고 응답 뒤 실제 즉시 결과로 교체한다. 따라서 요청과 저장 사이에 프로세스가 종료돼도 자동 timeout 취소는 같은 주문을 재전송하지 않고, 재시작 뒤 체결조회로만 수렴한다. +- timeout과 운영자 취소 모두 `cancel_state IS NULL` 비교·설정에 성공한 한 호출만 외부 요청을 보낸다. 이후 반복 요청은 `cancel_already_recorded`로 응답하고 poll만 계속한다. ### get_current_price diff --git a/docs/runtime-flows.md b/docs/runtime-flows.md index 98c87b9..6afedbd 100644 --- a/docs/runtime-flows.md +++ b/docs/runtime-flows.md @@ -270,7 +270,7 @@ 1. 기능을 켠 KIS 데몬은 시작 시 1회 동기 조회 후 계정당 한 백그라운드 소유자가 기본 300초 간격과 실패 백오프로 보유수량·주문가능금액을 갱신한다. 2. 보유수량과 현금은 독립적으로 성공값과 시각을 저장한다. 계정 해시 변경, 인증·인가·형식·브로커 거부는 실패 폐쇄한다. -3. 신규매수는 기본 600초 이내 현금값에서 기존 pending 매수 예약금액과 새 주문금액을 합산해 검사한다. 시장가처럼 금액을 확정할 수 없으면 차단한다. +3. 신규매수는 기본 600초 이내 현금값에서 기존 pending 매수 예약금액과 새 주문금액을 합산해 검사한다. 검사와 pending 전이는 같은 instance advisory lock 안에서 원자 수행하며 시장가처럼 금액을 확정할 수 없으면 차단한다. 4. pending 매도는 실시간 보유조회 전송 실패일 때만 기본 3600초 이내 전략 소유 보유수량을 사용한다. 수동 보유분을 매도 권한으로 승격하지 않는다. 5. 재시작 원장대조, `sync`, `verify`, Portfolio Hub 일일 정합은 캐시를 사용하지 않는다. diff --git a/tests/adapters/broker/test_kis.py b/tests/adapters/broker/test_kis.py index 72ae6e4..b751939 100644 --- a/tests/adapters/broker/test_kis.py +++ b/tests/adapters/broker/test_kis.py @@ -11,7 +11,7 @@ import os import httpx import pytest -from trex_core.adapters.broker.kis import KISBroker +from trex_core.adapters.broker.kis import KISBroker, _balance_http_error_code from trex_core.application.broker_port import ( BrokerBalanceError, CancelState, @@ -31,6 +31,41 @@ def test_account_split_and_mock_tr_ids(): assert "12345678" not in b.account_scope_hash() +def test_account_scope_hash_separates_endpoint_and_application_identity(): + mock_a = KISBroker( + "12345678-01", app_key="app-a", app_secret="s", base_url=_MOCK_URL + ) + mock_b = KISBroker( + "12345678-01", app_key="app-b", app_secret="s", base_url=_MOCK_URL + ) + live_a = KISBroker( + "12345678-01", + app_key="app-a", + app_secret="s", + base_url="https://openapi.koreainvestment.com:9443", + allow_live=True, + ) + + assert len({mock_a.account_scope_hash(), mock_b.account_scope_hash(), + live_a.account_scope_hash()}) == 3 + + +@pytest.mark.parametrize( + ("status", "code"), + [ + (400, "broker_rejected"), + (401, "authentication_failed"), + (403, "authorization_failed"), + (422, "broker_rejected"), + (429, "transport_failed"), + (500, "transport_failed"), + (None, "broker_rejected"), + ], +) +def test_balance_http_status_classification(status, code): + assert _balance_http_error_code(status) == code + + def test_live_base_url_requires_allow_live(): with pytest.raises(RuntimeError): KISBroker( diff --git a/tests/adapters/persistence/test_migration_and_drift.py b/tests/adapters/persistence/test_migration_and_drift.py index 6e54153..703099b 100644 --- a/tests/adapters/persistence/test_migration_and_drift.py +++ b/tests/adapters/persistence/test_migration_and_drift.py @@ -9,7 +9,8 @@ from __future__ import annotations import psycopg from trex_core.adapters.persistence.migrate import apply_migrations, migration_files -from trex_core.domain.slot import EXPECTED_SLOT_FIELDS +from trex_core.adapters.persistence.postgresql import PostgreSQLSlotRepository +from trex_core.domain.slot import EXPECTED_SLOT_FIELDS, SlotService, SlotState _EXPECTED_TABLES = { "daily_slot", @@ -61,3 +62,22 @@ def test_apply_migrations_is_idempotent(pg_dsn): # applying twice must not error (CREATE ... IF NOT EXISTS) apply_migrations(pg_dsn) apply_migrations(pg_dsn) + + +def test_migration_replay_accepts_existing_auto_venue_row(pg_clean): + repo = PostgreSQLSlotRepository(pg_clean) + service = SlotService(repo, instance_key="migration-auto") + repo.reconcile_universe("migration-auto", {"pf-auto": "005930"}) + slot = repo.find_slot_by_portfolio("migration-auto", "pf-auto") + service.transit_slot( + slot.slot_id, + SlotState.INACTIVE, + SlotState.PENDING_BUY, + reason="migration_replay", + actor="test", + venue="AUTO", + ) + + apply_migrations(pg_clean) + + assert repo.fetch_slot(slot.slot_id).venue == "AUTO" diff --git a/tests/adapters/persistence/test_postgresql_repository.py b/tests/adapters/persistence/test_postgresql_repository.py index ad9f2fd..77bd96a 100644 --- a/tests/adapters/persistence/test_postgresql_repository.py +++ b/tests/adapters/persistence/test_postgresql_repository.py @@ -132,6 +132,29 @@ def test_dispatch_queue_instance_scoped(pg_clean): assert repo_a.list_dispatched_pending("inst-b") == [] +def test_cancel_attempt_reservation_is_compare_and_set(pg_clean): + repo, _, intents = _wire(pg_clean) + repo.reconcile_universe(INST, {"PF-1": "005930"}) + decision = intents.submit_buy( + BuyIntent(intent_id="pg-cancel-cas", qty=1, portfolio_id="PF-1") + ) + assert repo.mark_dispatched(decision.slot_id, "ord-cas") + assert repo.mark_cancel_outcome( + decision.slot_id, + "ord-cas", + "unknown", + only_if_unset=True, + ) + assert not repo.mark_cancel_outcome( + decision.slot_id, + "ord-cas", + "unknown", + only_if_unset=True, + ) + assert repo.mark_cancel_outcome(decision.slot_id, "ord-cas", "requested") + assert repo.fetch_slot(decision.slot_id).cancel_state == "requested" + + # dcy-028 / REQ-079·081 (BAT-042): strategy-state store + occupancy_seq column def test_strategy_state_store_pg(pg_clean): from trex_core.adapters.persistence.postgresql import PgStrategyStateStore diff --git a/tests/application/test_execution_cancel.py b/tests/application/test_execution_cancel.py index 628a30a..df77fff 100644 --- a/tests/application/test_execution_cancel.py +++ b/tests/application/test_execution_cancel.py @@ -40,6 +40,7 @@ class ProgrammableBroker: ) self.polls = 0 self.balance = {"005930": 10} + self.cancel_calls = 0 def name(self): return "stub" @@ -53,6 +54,7 @@ class ProgrammableBroker: return self.responder(side) def cancel_order(self, broker_order_id, *, side, symbol, qty, venue="KRX"): + self.cancel_calls += 1 return CancelOutcome(CancelState.REQUESTED) def get_balance(self): @@ -159,3 +161,20 @@ def test_stale_watchdog_disabled_by_default(): repo._slots[sid] = replace(s, updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=9999)) r = execution.run_once() assert r.stale_pending == 0 # disabled => never warns + + +def test_operator_cancel_is_sent_once_across_repeat_and_restart(): + broker = ProgrammableBroker() + repo, svc, intents, execution, _, _ = _wire(broker) + sid = _buy_to_holding(intents, execution) + intents.submit_sell(SellIntent(intent_id="s1", slot_id=sid)) + broker.responder = lambda side: FillStatus(state="accepted") + execution.run_once() + + assert execution.cancel_pending_slot(sid) == "ok" + assert execution.cancel_pending_slot(sid) == "cancel_already_recorded" + + restarted = ExecutionService(svc, repo, broker) + assert restarted.cancel_pending_slot(sid) == "cancel_already_recorded" + assert broker.cancel_calls == 1 + assert repo.fetch_slot(sid).cancel_state == CancelState.REQUESTED.value diff --git a/tests/application/test_intent_concurrency.py b/tests/application/test_intent_concurrency.py index 903511f..ff73ddf 100644 --- a/tests/application/test_intent_concurrency.py +++ b/tests/application/test_intent_concurrency.py @@ -49,13 +49,9 @@ def test_concurrent_buys_admit_at_most_m(): admitted = [s for s in results if s == "admitted"] pending = [s for s in repo.list_slots(INST) if s.status is SlotState.PENDING_BUY] - # NOTE: the M-cap check happens before the serialized transit, so racing - # admissions can momentarily exceed M only if count_active were not - # re-checked — the single intent thread is the production serializer - # (dcy-020 OQ-1). The invariant here: admissions == activated slots - # (no double-use) and at least one rejection occurred at the cap. + # Capacity admission and transition share the same instance advisory lock. assert len(admitted) == len(pending) - assert len(admitted) >= 5 + assert len(admitted) == 5 assert len(results) == 20 @@ -95,3 +91,52 @@ def test_duplicate_intent_id_admits_once_under_concurrency(): pending = [s for s in repo.list_slots(INST) if s.status is SlotState.PENDING_BUY] assert len(pending) == 1 assert all(s in ("admitted", "processing") for s in statuses) + + +def test_concurrent_cached_cash_reservations_are_atomic_with_transition(): + repo = InMemorySlotRepository() + svc = SlotService(repo, instance_key=INST) + repo.reconcile_universe(INST, {"PF-0": "SYM", "PF-1": "SYM"}) + + def cash_guard(intent: BuyIntent): + reserved = sum( + float(slot.qty or 0) * float(slot.limit_price or 0) + for slot in repo.list_slots(INST, {"pending_buy"}) + ) + requested = float(intent.qty or 0) * float(intent.entry_price or 0) + return None if reserved + requested <= 100 else "insufficient_snapshot_cash" + + intents = IntentService( + svc, + repo, + InMemoryIntentStore(), + resolve_symbol=lambda pid: f"SYM-{pid}", + instance_key=INST, + buy_guard_fn=cash_guard, + ) + results = [] + result_lock = threading.Lock() + + def worker(index: int) -> None: + decision = intents.submit_buy( + BuyIntent( + intent_id=f"cash-{index}", + portfolio_id=f"PF-{index}", + qty=1, + entry_price=100, + ) + ) + with result_lock: + results.append(decision) + + threads = [threading.Thread(target=worker, args=(index,)) for index in range(2)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert [decision.status for decision in results].count("admitted") == 1 + assert [decision.reason for decision in results].count( + "insufficient_snapshot_cash" + ) == 1 + assert len(repo.list_slots(INST, {"pending_buy"})) == 1 diff --git a/trex_core/adapters/broker/kis.py b/trex_core/adapters/broker/kis.py index f824085..ea90966 100644 --- a/trex_core/adapters/broker/kis.py +++ b/trex_core/adapters/broker/kis.py @@ -43,6 +43,16 @@ _SUBMISSION_LOOKUP_GRACE_SECONDS = 90.0 _SUPPORTED_ORDER_VENUES = frozenset({"AUTO", "KRX", "NXT", "SOR"}) +def _balance_http_error_code(status: Optional[int]) -> str: + if status == 401: + return "authentication_failed" + if status == 403: + return "authorization_failed" + if status in {408, 425, 429} or (status is not None and status >= 500): + return "transport_failed" + return "broker_rejected" + + def resolve_order_venue(venue: str, at: datetime) -> str: """Resolve AUTO to the venue that accepts domestic-stock orders at `at`. @@ -161,7 +171,8 @@ class KISBroker: 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() + material = f"{self._base_url}|{cano}|{product}|{self._app_key}" + return hashlib.sha256(material.encode()).hexdigest() # --- credential hot-swap (dcy-036 / REQ-113 AC-2 / REQ-115 AC-1) ---------- def reinitialize(self, credential) -> None: @@ -658,12 +669,7 @@ class KISBroker: payload = resp.json() 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 + raise BrokerBalanceError(_balance_http_error_code(status)) from None except httpx.HTTPError: raise BrokerBalanceError("transport_failed") from None except (ValueError, TypeError): @@ -721,12 +727,7 @@ class KISBroker: 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 + raise BrokerBalanceError(_balance_http_error_code(status)) from None except httpx.HTTPError: raise BrokerBalanceError("transport_failed") from None except (KeyError, TypeError, ValueError): diff --git a/trex_core/adapters/persistence/postgresql.py b/trex_core/adapters/persistence/postgresql.py index 64471b6..c5ad040 100644 --- a/trex_core/adapters/persistence/postgresql.py +++ b/trex_core/adapters/persistence/postgresql.py @@ -341,6 +341,8 @@ class PostgreSQLSlotRepository: slot_id: UUID, broker_order_id: str, state: str, + *, + only_if_unset: bool = False, ) -> bool: now = datetime.now(tz=timezone.utc) with self._psycopg.connect(self._dsn, autocommit=False) as conn: @@ -349,8 +351,16 @@ class PostgreSQLSlotRepository: "UPDATE daily_slot SET cancel_state=%s, " "cancel_requested_at=%s, updated_at=%s " "WHERE slot_id=%s AND broker_order_id=%s " - "AND status IN ('pending_buy','pending_sell')", - (state, now, now, slot_id, broker_order_id), + "AND status IN ('pending_buy','pending_sell') " + "AND (%s = FALSE OR cancel_state IS NULL)", + ( + state, + now, + now, + slot_id, + broker_order_id, + only_if_unset, + ), ) ok = cur.rowcount == 1 conn.commit() diff --git a/trex_core/adapters/persistence/sql/017_add_slot_order_venue.sql b/trex_core/adapters/persistence/sql/017_add_slot_order_venue.sql index a6571df..9f43772 100644 --- a/trex_core/adapters/persistence/sql/017_add_slot_order_venue.sql +++ b/trex_core/adapters/persistence/sql/017_add_slot_order_venue.sql @@ -8,4 +8,6 @@ ALTER TABLE daily_slot ALTER TABLE daily_slot ADD CONSTRAINT daily_slot_venue_check - CHECK (venue IN ('KRX', 'NXT', 'SOR')); + -- apply_migrations() replays every file on each boot. Keep this historical + -- migration compatible with AUTO rows introduced by migration 019. + CHECK (venue IN ('AUTO', 'KRX', 'NXT', 'SOR')); 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 index e1804cb..6ef5833 100644 --- a/trex_core/adapters/persistence/sql/021_create_broker_account_snapshot.sql +++ b/trex_core/adapters/persistence/sql/021_create_broker_account_snapshot.sql @@ -1,5 +1,5 @@ -- dcy-041 / GitHub #45: account-scoped last healthy KIS snapshot. --- account_hash is SHA-256(CANO|ACNT_PRDT_CD); no plaintext account identity, +-- account_hash is SHA-256(base_url|CANO|ACNT_PRDT_CD|app_key); no plaintext identity, -- token, key, order response, or broker payload is persisted. CREATE TABLE IF NOT EXISTS broker_account_snapshot ( account_hash TEXT PRIMARY KEY, diff --git a/trex_core/application/execution.py b/trex_core/application/execution.py index 9cba327..80cace5 100644 --- a/trex_core/application/execution.py +++ b/trex_core/application/execution.py @@ -579,6 +579,7 @@ class ExecutionService: slot.slot_id, str(slot.broker_order_id), CancelState.UNKNOWN.value, + only_if_unset=True, ): _log.error( "cancel attempt reservation conflict: slot=%s broker_order_id=%s", @@ -719,6 +720,8 @@ class ExecutionService: if not slot.broker_order_id: # Claimed but not yet dispatched — execution loop will pick up or rollback. return "not_dispatched" + if slot.cancel_state is not None: + return "cancel_already_recorded" side = "buy" if slot.status is SlotState.PENDING_BUY else "sell" qty = int(slot.qty or 0) try: @@ -735,8 +738,9 @@ class ExecutionService: slot.slot_id, str(slot.broker_order_id), CancelState.UNKNOWN.value, + only_if_unset=True, ): - return "cancel_state_conflict" + return "cancel_already_recorded" outcome = self._broker.cancel_order(slot.broker_order_id, **cancel_kwargs) except Exception: # noqa: BLE001 - adapter failure becomes durable unknown outcome = CancelOutcome(CancelState.UNKNOWN, "adapter_exception") diff --git a/trex_core/application/intents.py b/trex_core/application/intents.py index 4f451ac..089fd75 100644 --- a/trex_core/application/intents.py +++ b/trex_core/application/intents.py @@ -30,6 +30,7 @@ from uuid import UUID from trex_core.application.ports import IntentStore from trex_core.domain.slot import ( + SlotAdmissionRejectedError, SlotRepository, SlotService, SlotState, @@ -275,20 +276,24 @@ 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: + def _admission_precondition(_locked_slot) -> Optional[str]: + # Runs inside SlotService's instance-scoped advisory lock. This + # makes capacity and cached-cash reservation checks atomic with + # the PENDING_BUY transition across same-instance daemons. + if ( + not is_increase + and self._max_active_slots is not None + and self._repo.count_active(self._instance_key) + >= self._max_active_slots + ): + return "capacity_full" + if self._buy_guard_fn is None: + return None try: - guard_reason = self._buy_guard_fn(intent) + return 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 - and self._repo.count_active(self._instance_key) >= self._max_active_slots): - return IntentDecision(intent.intent_id, "rejected", None, "capacity_full") + return "account_snapshot_unavailable" + # dcy-022 / REQ-068: resolve effective timeout (per-intent > config default). effective_timeout = ( intent.timeout_sec @@ -317,6 +322,11 @@ class IntentService: slippage_pct=intent.slippage_pct, venue=intent.venue, extra={"intent_id": intent.intent_id, "symbol": symbol}, + precondition=_admission_precondition, + ) + except SlotAdmissionRejectedError as exc: + return IntentDecision( + intent.intent_id, "rejected", None, exc.reason ) except SlotStateMismatchError: continue # slot changed under us; re-fetch and re-evaluate diff --git a/trex_core/domain/slot/__init__.py b/trex_core/domain/slot/__init__.py index 31cd9df..e05ff80 100644 --- a/trex_core/domain/slot/__init__.py +++ b/trex_core/domain/slot/__init__.py @@ -11,11 +11,13 @@ Public API: - EXPECTED_SLOT_FIELDS: canonical `Slot` column contract (guarded by test_contract) - SlotService: main domain service (transit, sell, thresholds, capacity) - SlotRepository: abstract storage protocol -- SlotTransitionError / SlotStateMismatchError / SlotNotFoundError: domain errors +- SlotTransitionError / SlotStateMismatchError / SlotAdmissionRejectedError / + SlotNotFoundError: domain errors """ from __future__ import annotations from trex_core.domain.slot.errors import ( + SlotAdmissionRejectedError, SlotDomainError, SlotNotFoundError, SlotStateMismatchError, @@ -42,6 +44,7 @@ __all__ = [ "EXPECTED_SLOT_FIELDS", "SLOT_ADVISORY_LOCK_ID", "Slot", + "SlotAdmissionRejectedError", "SlotCapacity", "SlotDomainError", "SlotNotFoundError", diff --git a/trex_core/domain/slot/errors.py b/trex_core/domain/slot/errors.py index 7ad7f73..7a04241 100644 --- a/trex_core/domain/slot/errors.py +++ b/trex_core/domain/slot/errors.py @@ -21,3 +21,11 @@ class SlotStateMismatchError(SlotDomainError): different state between read and write (race detection inside the transition function). """ + + +class SlotAdmissionRejectedError(SlotDomainError): + """Raised when an advisory-locked admission precondition rejects.""" + + def __init__(self, reason: str) -> None: + self.reason = reason + super().__init__(reason) diff --git a/trex_core/domain/slot/in_memory_repository.py b/trex_core/domain/slot/in_memory_repository.py index 42421fc..7e4c5bf 100644 --- a/trex_core/domain/slot/in_memory_repository.py +++ b/trex_core/domain/slot/in_memory_repository.py @@ -224,6 +224,8 @@ class InMemorySlotRepository: slot_id: UUID, broker_order_id: str, state: str, + *, + only_if_unset: bool = False, ) -> bool: """Persist one cancel attempt iff the same order is still pending.""" s = self._slots.get(slot_id) @@ -231,6 +233,7 @@ class InMemorySlotRepository: s is None or s.broker_order_id != broker_order_id or s.status not in self._PENDING + or (only_if_unset and s.cancel_state is not None) ): return False now = datetime.now(tz=timezone.utc) diff --git a/trex_core/domain/slot/service.py b/trex_core/domain/slot/service.py index 34cd171..b184bc8 100644 --- a/trex_core/domain/slot/service.py +++ b/trex_core/domain/slot/service.py @@ -15,10 +15,11 @@ from __future__ import annotations from dataclasses import replace from datetime import datetime, timezone from decimal import Decimal -from typing import Any, Optional +from typing import Any, Callable, Optional from uuid import UUID from trex_core.domain.slot.errors import ( + SlotAdmissionRejectedError, SlotNotFoundError, SlotStateMismatchError, SlotTransitionError, @@ -87,6 +88,7 @@ class SlotService: slippage_pct: Optional[float] = None, venue: Optional[str] = None, extra: Optional[dict[str, Any]] = None, + precondition: Optional[Callable[[Slot], Optional[str]]] = None, ) -> Slot: """Transition slot from one state to another atomically. @@ -116,6 +118,10 @@ class SlotService: f"slot {slot_id} current state {slot.status.value} " f"!= requested from_state {from_state.value}" ) + if precondition is not None: + rejection_reason = precondition(slot) + if rejection_reason is not None: + raise SlotAdmissionRejectedError(rejection_reason) updated = self._apply_transition_fields( slot,