diff --git a/docs/data-model.md b/docs/data-model.md index fcc5b33..0501e03 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -50,6 +50,8 @@ - allocated_capital :: numeric? :: 배정 자본 - order_id / broker_order_id :: str? :: 주문 식별자(자유 주석 / 브로커 핸들) - venue :: TEXT NOT NULL DEFAULT 'KRX' :: 현재 주문 거래소 구분(AUTO/KRX/NXT/SOR). intent 단계의 AUTO는 KIS 제출 시 실제 KRX/NXT로 교체되며 주문번호와 원자 저장한다. pending 주문·취소까지 보존하고 HOLDING 진입 시 다음 자율주문의 AUTO, INACTIVE 진입 시 KRX로 초기화한다. +- cancel_state :: TEXT? :: 취소 요청의 즉시 결과(requested/rejected/unknown). 최종 취소 상태가 아니며 같은 주문의 자동 재취소를 막는다. +- cancel_requested_at :: TIMESTAMPTZ? :: 취소 요청 결과를 저장한 시각. 주문 종료·dispatch clear 때 함께 제거한다. - strategy_id / sell_strategy_id :: str? :: 진입/청산 전략 식별 - portfolio_id :: str? :: 포트폴리오 식별 - forced_close_at :: datetime? :: 강제 청산 마감선 diff --git a/docs/interface-contract.md b/docs/interface-contract.md index a36f566..b441244 100644 --- a/docs/interface-contract.md +++ b/docs/interface-contract.md @@ -431,10 +431,12 @@ resume_kill_switch 요청·응답 (REQ-046): ### cancel_order ``` -cancel_order(broker_order_id: str, *, side: str, symbol: str, qty: int, venue: str = "KRX") -> bool +cancel_order(broker_order_id: str, *, side: str, symbol: str, qty: int, venue: str = "KRX") -> CancelOutcome ``` - `venue`: 원주문 venue. 취소 시 원주문과 동일한 `EXCG_ID_DVSN_CD`를 사용해야 한다. 기본값 `"KRX"`로 기존 호출자 하위 호환. +- `CancelOutcome.state`: `requested`(요청 접수), `rejected`(명시적 거부), `unknown`(응답유실·전송 실패). `requested`도 최종 취소가 아니며 `poll_fill()`의 `cancelled|filled`만 주문의 최종 상태를 확정한다. +- Core는 외부 취소 호출 전에 `unknown`을 먼저 영속화하고 응답 뒤 실제 즉시 결과로 교체한다. 따라서 요청과 저장 사이에 프로세스가 종료돼도 자동 timeout 취소는 같은 주문을 재전송하지 않고, 재시작 뒤 체결조회로만 수렴한다. ### get_current_price diff --git a/tests/adapters/broker/test_kis.py b/tests/adapters/broker/test_kis.py index 57d23f9..5c32f05 100644 --- a/tests/adapters/broker/test_kis.py +++ b/tests/adapters/broker/test_kis.py @@ -12,7 +12,11 @@ import httpx import pytest from trex_core.adapters.broker.kis import KISBroker -from trex_core.application.broker_port import BrokerBalanceError, OrderRequest +from trex_core.application.broker_port import ( + BrokerBalanceError, + CancelState, + OrderRequest, +) _MOCK_URL = "https://openapivts.koreainvestment.com:29443" @@ -179,15 +183,13 @@ def test_submit_transport_failure_blocks_automatic_duplicate(monkeypatch): assert ack.accepted is True assert ack.broker_order_id.startswith("submission-unknown:") assert ack.reason == "submit_outcome_unknown:ReadTimeout" - assert ( - b.cancel_order( - ack.broker_order_id, - side="buy", - symbol="005930", - qty=4, - ) - is False + cancel = b.cancel_order( + ack.broker_order_id, + side="buy", + symbol="005930", + qty=4, ) + assert cancel.state is CancelState.UNKNOWN def test_submit_gateway_5xx_blocks_automatic_duplicate(monkeypatch): diff --git a/tests/adapters/broker/test_ownership.py b/tests/adapters/broker/test_ownership.py index 7042bc7..2458b05 100644 --- a/tests/adapters/broker/test_ownership.py +++ b/tests/adapters/broker/test_ownership.py @@ -9,6 +9,8 @@ from trex_core.adapters.broker.ownership import ( from trex_core.application.broker_port import ( BalanceItem, BrokerBalanceError, + CancelOutcome, + CancelState, FillStatus, OrderAck, ) @@ -29,7 +31,7 @@ class FakeBroker: return self.fill def cancel_order(self, broker_order_id, *, side, symbol, qty, venue="KRX"): - return True + return CancelOutcome(CancelState.REQUESTED) def get_balance(self): return self.balance diff --git a/tests/adapters/broker/test_venue.py b/tests/adapters/broker/test_venue.py index 411cb26..c5d6701 100644 --- a/tests/adapters/broker/test_venue.py +++ b/tests/adapters/broker/test_venue.py @@ -26,12 +26,13 @@ from __future__ import annotations from datetime import datetime, timedelta, timezone from decimal import Decimal +import httpx import pytest from trex_core.adapters.broker import kis as kis_mod from trex_core.adapters.broker.kis import KISBroker, resolve_order_venue from trex_core.adapters.broker.paper import PaperBroker -from trex_core.application.broker_port import OrderRequest +from trex_core.application.broker_port import CancelState, OrderRequest _MOCK_URL = "https://openapivts.koreainvestment.com:29443" _LIVE_URL = "https://openapi.koreainvestment.com:9443" @@ -218,7 +219,7 @@ def test_paper_cancel_venue_ignored(): ack = b.submit_order(req) # NXT venue로 cancel — paper는 venue 무관하게 취소 성공 result = b.cancel_order(ack.broker_order_id, side="buy", symbol="005930", qty=5, venue="NXT") - assert result is True + assert result.state is CancelState.REQUESTED def test_paper_cancel_default_venue(): @@ -227,7 +228,7 @@ def test_paper_cancel_default_venue(): req = OrderRequest(slot_id="s1", side="buy", symbol="005930", qty=3) ack = b.submit_order(req) result = b.cancel_order(ack.broker_order_id, side="buy", symbol="005930", qty=3) - assert result is True + assert result.state is CancelState.REQUESTED # --------------------------------------------------------------------------- @@ -301,6 +302,40 @@ def test_cancel_order_krx_includes_excg(monkeypatch): assert captured["body"]["EXCG_ID_DVSN_CD"] == "KRX" +def test_cancel_explicit_broker_rejection_is_distinct(monkeypatch): + def fake_post(url, headers=None, json=None, timeout=None): + return _FakeResp({"rt_cd": "1", "msg_cd": "ORDER_NOT_CANCELLABLE"}) + + b = _mock_broker() + monkeypatch.setattr(kis_mod.httpx, "post", fake_post) + outcome = b.cancel_order( + "06010:0000123456", + side="buy", + symbol="005930", + qty=1, + ) + + assert outcome.state is CancelState.REJECTED + assert outcome.reason == "ORDER_NOT_CANCELLABLE" + + +def test_cancel_transport_loss_is_unknown(monkeypatch): + def fake_post(url, headers=None, json=None, timeout=None): + raise httpx.ReadTimeout("timed out") + + b = _mock_broker() + monkeypatch.setattr(kis_mod.httpx, "post", fake_post) + outcome = b.cancel_order( + "06010:0000123456", + side="buy", + symbol="005930", + qty=1, + ) + + assert outcome.state is CancelState.UNKNOWN + assert outcome.reason == "transport_failed" + + # --------------------------------------------------------------------------- # REQ-118 AC-5: get_current_price 기본 market=J # --------------------------------------------------------------------------- @@ -422,4 +457,4 @@ def test_paper_cancel_backward_compat_no_venue(): ack = b.submit_order(req) # 기존 스타일 호출 (venue 없음) result = b.cancel_order(ack.broker_order_id, side="sell", symbol="005930", qty=2) - assert result is True + assert result.state is CancelState.REQUESTED diff --git a/tests/adapters/persistence/test_postgresql_slot_repository_parity.py b/tests/adapters/persistence/test_postgresql_slot_repository_parity.py index df812cf..48afe98 100644 --- a/tests/adapters/persistence/test_postgresql_slot_repository_parity.py +++ b/tests/adapters/persistence/test_postgresql_slot_repository_parity.py @@ -35,7 +35,8 @@ _SLOT_COLUMNS = ( "order_id, strategy_id, broker_order_id, portfolio_id, " "sell_strategy_id, forced_close_at, stop_loss_price, take_profit_price, " "created_at, updated_at, instance_key, retired_at, " - "timeout_sec, slippage_pct, occupancy_seq, venue" + "timeout_sec, slippage_pct, occupancy_seq, venue, " + "cancel_state, cancel_requested_at" ) _MANAGED_SLOT_SQL = ( f"SELECT {_SLOT_COLUMNS} FROM daily_slot " @@ -146,6 +147,8 @@ def _row(slot: Slot) -> tuple: Decimal(str(slot.slippage_pct)), slot.occupancy_seq, slot.venue, + slot.cancel_state, + slot.cancel_requested_at, ) diff --git a/tests/application/test_execution.py b/tests/application/test_execution.py index 01fd853..81a7e4f 100644 --- a/tests/application/test_execution.py +++ b/tests/application/test_execution.py @@ -7,7 +7,12 @@ 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 FillStatus, OrderAck +from trex_core.application.broker_port import ( + CancelOutcome, + CancelState, + FillStatus, + OrderAck, +) from trex_core.application.execution import ExecutionService from trex_core.application.intents import BuyIntent, IntentService, SellIntent from trex_core.domain.slot import SlotService, SlotState @@ -78,7 +83,10 @@ def test_unknown_submit_outcome_stays_dispatched_without_retry(): def cancel_order(self, broker_order_id, *, side, symbol, qty, venue="KRX"): self.cancel_count += 1 - return False + return CancelOutcome( + CancelState.UNKNOWN, + "submission_outcome_unknown", + ) broker = UnknownOutcomeBroker() repo, svc, intents, execution = _wire(broker) diff --git a/tests/application/test_execution_cancel.py b/tests/application/test_execution_cancel.py index 8baa58a..628a30a 100644 --- a/tests/application/test_execution_cancel.py +++ b/tests/application/test_execution_cancel.py @@ -13,7 +13,13 @@ from datetime import datetime, timedelta, timezone from trex_core.adapters.ledger.in_memory import InMemoryLedger from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore -from trex_core.application.broker_port import BalanceItem, FillStatus, OrderAck +from trex_core.application.broker_port import ( + BalanceItem, + CancelOutcome, + CancelState, + FillStatus, + OrderAck, +) from trex_core.application.execution import ExecutionService from trex_core.application.intents import BuyIntent, IntentService, SellIntent from trex_core.application.ledger_outbox import InMemoryLedgerOutbox, LedgerWriter @@ -46,8 +52,8 @@ class ProgrammableBroker: self.polls += 1 return self.responder(side) - def cancel_order(self, broker_order_id, *, side, symbol, qty): - return True + def cancel_order(self, broker_order_id, *, side, symbol, qty, venue="KRX"): + return CancelOutcome(CancelState.REQUESTED) def get_balance(self): # M2 (dcy-020/REQ-062): sell dispatch reads the broker balance. diff --git a/tests/application/test_execution_fill_price_backfill.py b/tests/application/test_execution_fill_price_backfill.py index 62a6b7e..6503e5a 100644 --- a/tests/application/test_execution_fill_price_backfill.py +++ b/tests/application/test_execution_fill_price_backfill.py @@ -19,6 +19,8 @@ from trex_core.adapters.ledger.in_memory import InMemoryLedger from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore from trex_core.application.broker_port import ( BalanceItem, + CancelOutcome, + CancelState, FillStatus, OrderAck, OrderRequest, @@ -83,8 +85,16 @@ class StubBroker: return FillStatus(state="unknown") return o.fill_status - def cancel_order(self, broker_order_id: str, *, side: str, symbol: str, qty: int) -> bool: - return False + def cancel_order( + self, + broker_order_id: str, + *, + side: str, + symbol: str, + qty: int, + venue: str = "KRX", + ) -> CancelOutcome: + return CancelOutcome(CancelState.REJECTED, "order_not_cancellable") def get_balance(self) -> list[BalanceItem]: return [ diff --git a/tests/application/test_execution_policy.py b/tests/application/test_execution_policy.py index adcbe0b..ed0fe3d 100644 --- a/tests/application/test_execution_policy.py +++ b/tests/application/test_execution_policy.py @@ -11,7 +11,14 @@ from datetime import datetime, timedelta, timezone from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore -from trex_core.application.broker_port import BalanceItem, FillStatus, OrderAck, OrderRequest +from trex_core.application.broker_port import ( + BalanceItem, + CancelOutcome, + CancelState, + FillStatus, + OrderAck, + OrderRequest, +) from trex_core.application.execution import ExecutionService from trex_core.application.intents import BuyIntent, IntentService, SellIntent from trex_core.domain.slot import SlotService, SlotState @@ -26,6 +33,7 @@ class _RecordingBroker: self.orders: list[OrderRequest] = [] self.cancel_calls: list = [] self.cancel_raises = False + self.cancel_outcome = CancelOutcome(CancelState.REQUESTED) self.responder = lambda side: FillStatus( state="filled", filled_qty=10, avg_price=1000.0, execution_id="exec-1", fee=0.0, tax=0.0, @@ -47,7 +55,7 @@ class _RecordingBroker: self.cancel_calls.append((broker_order_id, side, symbol, qty, venue)) if self.cancel_raises: raise RuntimeError("broker error") - return True + return self.cancel_outcome def get_balance(self): return [BalanceItem(symbol=s, hldg_qty=q) for s, q in self.balance.items() if q > 0] @@ -287,6 +295,9 @@ class TestTimeoutCancel: r = execution.run_once() assert r.timeout_cancelled == 1 assert len(broker.cancel_calls) == 1 + assert repo.fetch_slot(sid).cancel_state == "requested" + execution.run_once() + assert len(broker.cancel_calls) == 1 def test_timeout_not_exceeded_no_cancel(self): broker = _RecordingBroker() @@ -330,7 +341,12 @@ class TestTimeoutCancel: r = execution.run_once() # timeout_cancelled NOT incremented when the cancel call raises assert r.timeout_cancelled == 0 - assert repo.fetch_slot(sid).status is SlotState.PENDING_BUY # state unchanged + slot = repo.fetch_slot(sid) + assert slot.status is SlotState.PENDING_BUY + assert slot.cancel_state == "unknown" + execution.run_once() + ExecutionService(svc, repo, broker).run_once() + assert len(broker.cancel_calls) == 1 def test_timeout_cancel_does_not_change_slot_state(self): """Timeout cancel is best-effort: slot stays PENDING_BUY after cancel_order.""" @@ -347,6 +363,28 @@ class TestTimeoutCancel: # slot still PENDING_BUY — cancel is best-effort; rollback on next poll assert repo.fetch_slot(sid).status is SlotState.PENDING_BUY + def test_timeout_cancel_rejection_is_persisted_without_retry(self): + broker = _RecordingBroker() + broker.cancel_outcome = CancelOutcome( + CancelState.REJECTED, + "broker_rejected", + ) + repo, svc, intents, execution = _wire(broker) + sid = _buy_to_pending(intents, timeout_sec=1) + broker.responder = lambda side: FillStatus(state="accepted") + execution.run_once() + slot = repo.fetch_slot(sid) + repo._slots[sid] = replace( + slot, + updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60), + ) + + result = execution.run_once() + assert result.timeout_cancelled == 0 + assert repo.fetch_slot(sid).cancel_state == "rejected" + ExecutionService(svc, repo, broker).run_once() + assert len(broker.cancel_calls) == 1 + def test_timeout_cancel_then_fill_race_handled(self): """Cancel sent, but broker fills the order — is_filled path wins on next poll.""" broker = _RecordingBroker() @@ -368,7 +406,10 @@ class TestTimeoutCancel: ) r2 = execution.run_once() assert r2.filled == 1 - assert repo.fetch_slot(sid).status is SlotState.HOLDING + slot = repo.fetch_slot(sid) + assert slot.status is SlotState.HOLDING + assert slot.cancel_state is None + assert slot.cancel_requested_at is None # ============================================================================ @@ -418,7 +459,7 @@ class TestCancelPendingSlot: result = execution.cancel_pending_slot(sid) assert result == "not_dispatched" - def test_cancel_fail_returns_cancel_failed(self): + def test_cancel_exception_returns_unknown(self): broker = _RecordingBroker() broker.cancel_raises = True repo, svc, intents, execution = _wire(broker) @@ -427,7 +468,22 @@ class TestCancelPendingSlot: execution.run_once() # dispatch result = execution.cancel_pending_slot(sid) - assert result == "cancel_failed" + assert result == "cancel_unknown" + assert repo.fetch_slot(sid).cancel_state == "unknown" + + def test_cancel_rejected_is_distinct_and_persisted(self): + broker = _RecordingBroker() + broker.cancel_outcome = CancelOutcome( + CancelState.REJECTED, + "broker_rejected", + ) + repo, svc, intents, execution = _wire(broker) + sid = _buy_to_pending(intents) + broker.responder = lambda side: FillStatus(state="accepted") + execution.run_once() + + assert execution.cancel_pending_slot(sid) == "cancel_rejected" + assert repo.fetch_slot(sid).cancel_state == "rejected" def test_cancel_does_not_change_slot_state(self): """Operator cancel sends cancel_order; slot stays PENDING_BUY (execution loop transitions).""" diff --git a/tests/application/test_venue_intent_flow.py b/tests/application/test_venue_intent_flow.py index 19a50d2..aee5e22 100644 --- a/tests/application/test_venue_intent_flow.py +++ b/tests/application/test_venue_intent_flow.py @@ -4,7 +4,14 @@ from __future__ import annotations from uuid import UUID from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore -from trex_core.application.broker_port import BalanceItem, FillStatus, OrderAck, OrderRequest +from trex_core.application.broker_port import ( + BalanceItem, + CancelOutcome, + CancelState, + FillStatus, + OrderAck, + OrderRequest, +) from trex_core.application.execution import ExecutionService from trex_core.application.intents import BuyIntent, IntentService from trex_core.application.messaging import RpcRequest @@ -38,9 +45,9 @@ class RecordingBroker: symbol: str, qty: int, venue: str = "KRX", - ) -> bool: + ) -> CancelOutcome: self.cancel_venues.append(venue) - return True + return CancelOutcome(CancelState.REQUESTED) def get_balance(self) -> list[BalanceItem]: return [] diff --git a/trex_core/adapters/broker/kis.py b/trex_core/adapters/broker/kis.py index f78bce3..50be9f6 100644 --- a/trex_core/adapters/broker/kis.py +++ b/trex_core/adapters/broker/kis.py @@ -24,6 +24,8 @@ import httpx from trex_core.application.broker_port import ( BalanceItem, BrokerBalanceError, + CancelOutcome, + CancelState, FillStatus, OrderAck, OrderRequest, @@ -756,7 +758,7 @@ class KISBroker: def cancel_order( self, broker_order_id: str, *, side: str, symbol: str, qty: int, venue: str = "KRX" - ) -> bool: + ) -> CancelOutcome: """Cancel a previously submitted order (REQ-118). venue: exchange venue of the original order. Must match the EXCG_ID_DVSN_CD @@ -764,7 +766,10 @@ class KISBroker: Real-account NXT cancel (EXCG=NXT) confirmed rt_cd=0 on 2026-07-06. """ if broker_order_id.startswith(_SUBMISSION_UNKNOWN_PREFIX): - return False + return CancelOutcome( + CancelState.UNKNOWN, + "submission_outcome_unknown", + ) cano, product = self._account() parts = broker_order_id.split(":") orgno, odno = (parts[0], parts[1]) if len(parts) == 2 else ("", parts[-1]) @@ -789,6 +794,22 @@ class KISBroker: json=body, timeout=self._timeout, ) - return str(resp.json().get("rt_cd")) == "0" - except (httpx.HTTPError, ValueError): - return False + except httpx.HTTPError: + return CancelOutcome(CancelState.UNKNOWN, "transport_failed") + if getattr(resp, "status_code", 200) >= 500: + return CancelOutcome(CancelState.UNKNOWN, "broker_unavailable") + try: + payload = resp.json() + except ValueError: + return CancelOutcome(CancelState.UNKNOWN, "malformed_response") + rt_cd = payload.get("rt_cd") + if str(rt_cd) == "0": + return CancelOutcome(CancelState.REQUESTED) + if rt_cd is None: + if getattr(resp, "status_code", 200) >= 400: + return CancelOutcome(CancelState.REJECTED, "http_rejected") + return CancelOutcome(CancelState.UNKNOWN, "missing_result_code") + return CancelOutcome( + CancelState.REJECTED, + str(payload.get("msg_cd") or "broker_rejected"), + ) diff --git a/trex_core/adapters/broker/ownership.py b/trex_core/adapters/broker/ownership.py index cd57e28..4c99375 100644 --- a/trex_core/adapters/broker/ownership.py +++ b/trex_core/adapters/broker/ownership.py @@ -16,6 +16,7 @@ from threading import Lock from trex_core.application.broker_port import ( BalanceItem, BrokerBalanceError, + CancelOutcome, FillStatus, OrderAck, OrderRequest, @@ -146,7 +147,7 @@ class OwnershipAwareBroker: symbol: str, qty: int, venue: str = "KRX", - ) -> bool: + ) -> CancelOutcome: return self._delegate.cancel_order( broker_order_id, side=side, diff --git a/trex_core/adapters/broker/paper.py b/trex_core/adapters/broker/paper.py index 017409b..e95ffc6 100644 --- a/trex_core/adapters/broker/paper.py +++ b/trex_core/adapters/broker/paper.py @@ -14,6 +14,8 @@ from typing import Optional from trex_core.adapters.broker.seq_store import SeqStore from trex_core.application.broker_port import ( BalanceItem, + CancelOutcome, + CancelState, FillStatus, OrderAck, OrderRequest, @@ -105,16 +107,24 @@ class PaperBroker: ) return FillStatus(state="partially_filled", filled_qty=o.filled, avg_price=o.price) - def cancel_order(self, broker_order_id: str, *, side: str, symbol: str, qty: int, venue: str = "KRX") -> bool: + def cancel_order( + self, + broker_order_id: str, + *, + side: str, + symbol: str, + qty: int, + venue: str = "KRX", + ) -> CancelOutcome: """Cancel a paper order. venue is accepted for BrokerPort protocol conformance (REQ-118) but ignored — paper has no exchange concept. Existing behavior preserved. """ with self._lock: o = self._orders.get(broker_order_id) if o is None or o.filled >= o.qty: - return False + return CancelOutcome(CancelState.REJECTED, "order_not_cancellable") o.cancelled = True - return True + return CancelOutcome(CancelState.REQUESTED) def get_balance(self) -> list[BalanceItem]: """Deterministic per-symbol fill aggregate (REQ-031 AC-3). diff --git a/trex_core/adapters/persistence/postgresql.py b/trex_core/adapters/persistence/postgresql.py index 36da514..64471b6 100644 --- a/trex_core/adapters/persistence/postgresql.py +++ b/trex_core/adapters/persistence/postgresql.py @@ -37,7 +37,8 @@ _SLOT_COLUMNS = ( "order_id, strategy_id, broker_order_id, portfolio_id, " "sell_strategy_id, forced_close_at, stop_loss_price, take_profit_price, " "created_at, updated_at, instance_key, retired_at, " - "timeout_sec, slippage_pct, occupancy_seq, venue" + "timeout_sec, slippage_pct, occupancy_seq, venue, " + "cancel_state, cancel_requested_at" ) @@ -216,7 +217,8 @@ class PostgreSQLSlotRepository: "broker_order_id=%s, portfolio_id=%s, sell_strategy_id=%s, " "forced_close_at=%s, stop_loss_price=%s, take_profit_price=%s, " "instance_key=%s, retired_at=%s, timeout_sec=%s, slippage_pct=%s, " - "occupancy_seq=%s, venue=%s, updated_at=%s WHERE slot_id=%s", + "occupancy_seq=%s, venue=%s, cancel_state=%s, " + "cancel_requested_at=%s, updated_at=%s WHERE slot_id=%s", ( slot.status.value, slot.symbol, @@ -237,6 +239,8 @@ class PostgreSQLSlotRepository: slot.slippage_pct, slot.occupancy_seq, slot.venue, + slot.cancel_state, + slot.cancel_requested_at, slot.updated_at or datetime.now(tz=timezone.utc), slot.slot_id, ), @@ -332,6 +336,26 @@ class PostgreSQLSlotRepository: conn.commit() return ok + def mark_cancel_outcome( + self, + slot_id: UUID, + broker_order_id: str, + state: str, + ) -> bool: + now = datetime.now(tz=timezone.utc) + with self._psycopg.connect(self._dsn, autocommit=False) as conn: + with conn.cursor() as cur: + cur.execute( + "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), + ) + ok = cur.rowcount == 1 + conn.commit() + return ok + def list_dispatched_pending(self, instance_key: str = "default") -> list[Slot]: with self._psycopg.connect(self._dsn) as conn: with conn.cursor() as cur: @@ -347,7 +371,8 @@ class PostgreSQLSlotRepository: with self._psycopg.connect(self._dsn, autocommit=False) as conn: with conn.cursor() as cur: cur.execute( - "UPDATE daily_slot SET broker_order_id=NULL, updated_at=%s WHERE slot_id=%s", + "UPDATE daily_slot SET broker_order_id=NULL, cancel_state=NULL, " + "cancel_requested_at=NULL, updated_at=%s WHERE slot_id=%s", (datetime.now(tz=timezone.utc), slot_id), ) conn.commit() @@ -374,7 +399,8 @@ class PostgreSQLSlotRepository: # forced_close_at(11) stop_loss_price(12) take_profit_price(13) # created_at(14) updated_at(15) instance_key(16) retired_at(17) # timeout_sec(18) slippage_pct(19) [dcy-022/REQ-068] - # occupancy_seq(20) [dcy-028/REQ-081], venue(21) + # occupancy_seq(20) [dcy-028/REQ-081], venue(21), + # cancel_state(22), cancel_requested_at(23) return Slot( slot_id=row[0] if isinstance(row[0], UUID) else UUID(str(row[0])), status=SlotState(row[1]), @@ -398,6 +424,8 @@ class PostgreSQLSlotRepository: slippage_pct=f(row[19]), occupancy_seq=int(row[20]) if len(row) > 20 and row[20] is not None else 0, venue=str(row[21]) if len(row) > 21 and row[21] is not None else "KRX", + cancel_state=str(row[22]) if len(row) > 22 and row[22] is not None else None, + cancel_requested_at=row[23] if len(row) > 23 else None, ) diff --git a/trex_core/adapters/persistence/sql/020_persist_cancel_request_outcome.sql b/trex_core/adapters/persistence/sql/020_persist_cancel_request_outcome.sql new file mode 100644 index 0000000..b1fe153 --- /dev/null +++ b/trex_core/adapters/persistence/sql/020_persist_cancel_request_outcome.sql @@ -0,0 +1,18 @@ +-- dcy-041 / GitHub #44: cancellation request acceptance is distinct from +-- final order cancellation. Persist the immediate result to prevent blind +-- automatic retries across execution ticks and daemon restarts. +ALTER TABLE daily_slot + ADD COLUMN IF NOT EXISTS cancel_state TEXT; + +ALTER TABLE daily_slot + ADD COLUMN IF NOT EXISTS cancel_requested_at TIMESTAMPTZ; + +ALTER TABLE daily_slot + DROP CONSTRAINT IF EXISTS daily_slot_cancel_state_check; + +ALTER TABLE daily_slot + ADD CONSTRAINT daily_slot_cancel_state_check + CHECK ( + cancel_state IS NULL + OR cancel_state IN ('requested', 'rejected', 'unknown') + ); diff --git a/trex_core/application/broker_port.py b/trex_core/application/broker_port.py index e9bbc73..7828198 100644 --- a/trex_core/application/broker_port.py +++ b/trex_core/application/broker_port.py @@ -10,6 +10,7 @@ from __future__ import annotations import os from dataclasses import dataclass, field from decimal import Decimal +from enum import Enum from typing import Any, Optional, Protocol @@ -29,6 +30,30 @@ class BrokerBalanceError(RuntimeError): super().__init__(f"broker balance unavailable ({code})") +class CancelState(str, Enum): + """Immediate result of a cancel request, not the final order state.""" + + REQUESTED = "requested" + REJECTED = "rejected" + UNKNOWN = "unknown" + + +@dataclass(frozen=True) +class CancelOutcome: + """Sanitized cancel request outcome. + + REQUESTED only means that the broker accepted the cancellation request. + Final cancelled/filled state is established exclusively by poll_fill(). + """ + + state: CancelState + reason: Optional[str] = None + + @property + def is_requested(self) -> bool: + return self.state is CancelState.REQUESTED + + @dataclass(frozen=True) class OrderRequest: slot_id: str # originating slot (audit/correlation) @@ -114,12 +139,23 @@ class BrokerPort(Protocol): """Observe current fill status for a previously accepted order.""" ... - def cancel_order(self, broker_order_id: str, *, side: str, symbol: str, qty: int, venue: str = "KRX") -> bool: - """Best-effort cancel. Returns True if the broker accepted the cancel. + def cancel_order( + self, + broker_order_id: str, + *, + side: str, + symbol: str, + qty: int, + venue: str = "KRX", + ) -> CancelOutcome: + """Request cancellation and classify the immediate broker outcome. venue: exchange venue of the original order (REQ-118). Cancel must use the same EXCG_ID_DVSN_CD as the original submit. Default "KRX" preserves backward compatibility for existing callers (orb/tsl). + + REQUESTED is not a final cancellation. poll_fill() remains the only + authority for cancelled/filled terminal state. """ ... diff --git a/trex_core/application/execution.py b/trex_core/application/execution.py index 2a8d874..2ea4d91 100644 --- a/trex_core/application/execution.py +++ b/trex_core/application/execution.py @@ -29,7 +29,12 @@ from datetime import datetime, timezone from typing import Callable, Optional from trex_core.application.alarm_emitter import AlarmEmitter -from trex_core.application.broker_port import BrokerPort, OrderRequest +from trex_core.application.broker_port import ( + BrokerPort, + CancelOutcome, + CancelState, + OrderRequest, +) from trex_core.application.ledger_outbox import LedgerOutbox from trex_core.application.ledger_port import LedgerTrade from trex_core.domain.pricing import ceil_to_tick, floor_to_tick @@ -546,12 +551,28 @@ class ExecutionService: """ if str(slot.broker_order_id or "").startswith("submission-unknown:"): return + if slot.cancel_state is not None: + return if slot.timeout_sec is None or slot.updated_at is None: return age = (now - slot.updated_at).total_seconds() if age < slot.timeout_sec: return qty = int(slot.qty or 0) + # Write UNKNOWN before external I/O. If the process dies after KIS + # receives the request but before the response is persisted, restart + # polls the order instead of issuing a blind second cancellation. + if not self._repo.mark_cancel_outcome( + slot.slot_id, + str(slot.broker_order_id), + CancelState.UNKNOWN.value, + ): + _log.error( + "cancel attempt reservation conflict: slot=%s broker_order_id=%s", + slot.slot_id, + slot.broker_order_id, + ) + return try: cancel_kwargs = { "side": side, @@ -560,25 +581,40 @@ class ExecutionService: } if slot.venue != "KRX": cancel_kwargs["venue"] = slot.venue - accepted = self._broker.cancel_order(slot.broker_order_id, **cancel_kwargs) + 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") + _log.warning( + "timeout cancel outcome unknown: slot=%s broker_order_id=%s", + slot.slot_id, + slot.broker_order_id, + ) + persisted = self._repo.mark_cancel_outcome( + slot.slot_id, + str(slot.broker_order_id), + outcome.state.value, + ) + if not persisted: + _log.error( + "cancel outcome persistence conflict: slot=%s broker_order_id=%s " + "outcome=%s", + slot.slot_id, + slot.broker_order_id, + outcome.state.value, + ) + return + if outcome.is_requested: r.timeout_cancelled += 1 - _log.warning( - "timeout cancel sent: slot=%s status=%s broker_order_id=%s " - "age_sec=%.0f timeout_sec=%d broker_accepted=%s", - slot.slot_id, - slot.status.value, - slot.broker_order_id, - age, - slot.timeout_sec, - accepted, - ) - except Exception: # noqa: BLE001 - best-effort; do not block the loop - _log.warning( - "timeout cancel failed (exception): slot=%s broker_order_id=%s " - "(will retry next tick if still pending)", - slot.slot_id, - slot.broker_order_id, - ) + _log.warning( + "timeout cancel result: slot=%s status=%s broker_order_id=%s " + "age_sec=%.0f timeout_sec=%d outcome=%s", + slot.slot_id, + slot.status.value, + slot.broker_order_id, + age, + slot.timeout_sec, + outcome.state.value, + ) def _emit_event(self, event_type: str, slot, routing_strategy_id: Optional[str] = None) -> None: if self._event_emitter is None: @@ -655,9 +691,9 @@ class ExecutionService: def cancel_pending_slot(self, slot_id) -> str: """Operator backstop: send cancel_order for a pending slot (dcy-022/REQ-071). - Returns "ok" (cancel sent), "not_pending" (slot is HOLDING/INACTIVE), - or "slot_not_found". Does NOT transition the slot — the execution loop's - next poll handles the rollback via the normal is_terminal_reject path. + Returns "ok" (request accepted), "cancel_rejected", "cancel_unknown", + "not_pending", "not_dispatched", or "slot_not_found". Does NOT + transition the slot — fill polling alone confirms cancelled/filled. """ from uuid import UUID @@ -680,21 +716,40 @@ class ExecutionService: } if slot.venue != "KRX": cancel_kwargs["venue"] = slot.venue - self._broker.cancel_order(slot.broker_order_id, **cancel_kwargs) - except Exception: # noqa: BLE001 - best-effort + # Durable unknown reservation closes the crash window between the + # outbound request and saving its immediate result. + if not self._repo.mark_cancel_outcome( + slot.slot_id, + str(slot.broker_order_id), + CancelState.UNKNOWN.value, + ): + return "cancel_state_conflict" + 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") _log.warning( - "operator cancel failed: slot=%s broker_order_id=%s", + "operator cancel outcome unknown: slot=%s broker_order_id=%s", slot.slot_id, slot.broker_order_id, ) - return "cancel_failed" + if not self._repo.mark_cancel_outcome( + slot.slot_id, + str(slot.broker_order_id), + outcome.state.value, + ): + return "cancel_state_conflict" _log.warning( - "operator cancel sent: slot=%s status=%s broker_order_id=%s", + "operator cancel result: slot=%s status=%s broker_order_id=%s outcome=%s", slot.slot_id, slot.status.value, slot.broker_order_id, + outcome.state.value, ) - return "ok" + if outcome.state is CancelState.REQUESTED: + return "ok" + if outcome.state is CancelState.REJECTED: + return "cancel_rejected" + return "cancel_unknown" def _rollback(self, slot, *, reason: str) -> None: total = self._balance_map().get(slot.symbol or "", 0) diff --git a/trex_core/domain/slot/in_memory_repository.py b/trex_core/domain/slot/in_memory_repository.py index f4420f8..42421fc 100644 --- a/trex_core/domain/slot/in_memory_repository.py +++ b/trex_core/domain/slot/in_memory_repository.py @@ -219,6 +219,29 @@ class InMemorySlotRepository: ) return True + def mark_cancel_outcome( + self, + slot_id: UUID, + broker_order_id: str, + state: str, + ) -> bool: + """Persist one cancel attempt iff the same order is still pending.""" + s = self._slots.get(slot_id) + if ( + s is None + or s.broker_order_id != broker_order_id + or s.status not in self._PENDING + ): + return False + now = datetime.now(tz=timezone.utc) + self._slots[slot_id] = replace( + s, + cancel_state=state, + cancel_requested_at=now, + updated_at=now, + ) + return True + def list_dispatched_pending(self, instance_key: str = "default") -> list[Slot]: """Pending slots already dispatched (broker_order_id NOT NULL) — poll targets.""" return [ @@ -236,7 +259,11 @@ class InMemorySlotRepository: s = self._slots.get(slot_id) if s is not None: self._slots[slot_id] = replace( - s, broker_order_id=None, updated_at=datetime.now(tz=timezone.utc) + s, + broker_order_id=None, + cancel_state=None, + cancel_requested_at=None, + updated_at=datetime.now(tz=timezone.utc), ) # --- bulk counts --- diff --git a/trex_core/domain/slot/models.py b/trex_core/domain/slot/models.py index 1b584e2..c1a7543 100644 --- a/trex_core/domain/slot/models.py +++ b/trex_core/domain/slot/models.py @@ -59,6 +59,11 @@ class Slot: # Order venue carried from the admitted intent through dispatch/cancel. # It is reset to KRX when the permanent asset slot returns to INACTIVE. venue: str = "KRX" + # Immediate cancel-request result. The slot remains pending until fill + # polling confirms cancelled/filled. Persisted to prevent blind retries + # across ticks and restarts. + cancel_state: Optional[str] = None + cancel_requested_at: Optional[datetime] = None # Canonical column contract for `Slot` (dcy-001 OQ-2 / FR-5). @@ -88,6 +93,8 @@ EXPECTED_SLOT_FIELDS: frozenset[str] = frozenset({ "slippage_pct", # dcy-022 / REQ-068 (BAT-032) "occupancy_seq", # dcy-028 / REQ-081 (BAT-042) "venue", + "cancel_state", + "cancel_requested_at", }) diff --git a/trex_core/domain/slot/service.py b/trex_core/domain/slot/service.py index 6767d14..34cd171 100644 --- a/trex_core/domain/slot/service.py +++ b/trex_core/domain/slot/service.py @@ -205,6 +205,8 @@ class SlotService: timeout_sec=None, slippage_pct=None, venue="KRX", + cancel_state=None, + cancel_requested_at=None, updated_at=datetime.now(tz=timezone.utc), ) @@ -277,6 +279,8 @@ class SlotService: slippage_pct=new_slippage_pct, occupancy_seq=new_occupancy_seq, venue=new_venue, + cancel_state=None, + cancel_requested_at=None, updated_at=datetime.now(tz=timezone.utc), )