diff --git a/docs/data-model.md b/docs/data-model.md index 138f802..3819559 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -50,8 +50,11 @@ - 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 때 함께 제거한다. +- cancel_state :: TEXT? :: 취소 요청의 즉시 결과(requested/rejected/unknown). 최종 취소 상태가 아니다. requested는 재시도하지 않고, rejected/unknown만 authoritative open-row 확인과 bounded backoff 뒤 재시도할 수 있다. +- cancel_requested_at :: TIMESTAMPTZ? :: 마지막 취소 시도 예약 시각. 주문 종료·dispatch clear 때 함께 제거한다. +- order_submitted_at :: TIMESTAMPTZ? :: timeout deadline의 immutable 주문 제출 기준점. poll/cancel bookkeeping 및 daemon 재시작에서 보존하고 terminal dispatch clear 때 제거한다. +- order_date :: TEXT? :: 원주문 KST 날짜(`YYYYMMDD`). 날짜 rollover 뒤 KIS 일별주문체결조회가 현재 날짜가 아닌 원주문 날짜를 사용하도록 보존한다. +- cancel_attempts :: INTEGER NOT NULL DEFAULT 0 :: 원자적으로 예약된 취소 시도 횟수. 새 dispatch에서 0, terminal dispatch clear에서 0으로 초기화한다. - 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 ebaa721..1df6911 100644 --- a/docs/interface-contract.md +++ b/docs/interface-contract.md @@ -447,8 +447,8 @@ 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만 계속한다. +- Core는 외부 취소 호출 전에 `unknown`과 시도 횟수를 원자적으로 영속화하고 응답 뒤 실제 즉시 결과로 교체한다. 따라서 요청과 저장 사이에 프로세스가 종료돼도 재시작 시 먼저 체결조회를 수행한다. +- `requested`는 재시도하지 않는다. `rejected|unknown`은 후속 `poll_fill()`이 원주문을 open으로 확인하고 bounded backoff·최대 시도 조건을 만족한 경우에만 재시도한다. row 미가시성·missing·accepted만으로 blind retry하지 않는다. 운영자 반복 요청은 기존 `cancel_already_recorded` 경계를 유지한다. ### get_current_price @@ -466,6 +466,10 @@ get_current_price(symbol: str, market: str = "J") -> Optional[Decimal] - `poll_fill`: KIS 신형 일별주문체결조회 실전 `TTTC0081R`, 모의 `VTTC0081R`을 사용하고 `EXCG_ID_DVSN_CD=ALL`을 전달한다. 일반 주문번호 조회와 결과불명 주문 탐색이 같은 계약으로 KRX·NXT 체결을 확인한다. + 일반 주문은 영속화된 `order_date`(`YYYYMMDD`)를 조회일로 사용하며, + row가 없으면 bounded 제출 직후 visibility grace 이후 `unknown`으로 + fail-closed 한다. `order_submitted_at`/`order_date`를 알 수 없는 legacy + pending row는 mutable `updated_at`로 추정하지 않고 CRITICAL alarm과 pending을 유지한다. ## `FillStatus` 확장 (BAT-006 / REQ-018) diff --git a/tests/adapters/broker/test_ownership.py b/tests/adapters/broker/test_ownership.py index 8e91bfd..063917f 100644 --- a/tests/adapters/broker/test_ownership.py +++ b/tests/adapters/broker/test_ownership.py @@ -25,6 +25,7 @@ class FakeBroker: def __init__(self, balance, fill=None): self.balance = balance self.fill = fill or FillStatus(state="accepted") + self.poll_metadata = None def name(self): return "fake" @@ -32,7 +33,16 @@ class FakeBroker: def submit_order(self, req): return OrderAck(accepted=True, broker_order_id="1") - def poll_fill(self, broker_order_id, *, side, symbol): + def poll_fill( + self, + broker_order_id, + *, + side, + symbol, + order_date=None, + submitted_at=None, + ): + self.poll_metadata = (order_date, submitted_at) return self.fill def cancel_order(self, broker_order_id, *, side, symbol, qty, venue="KRX"): @@ -77,6 +87,23 @@ def test_confirmed_fill_is_persisted_once_and_manual_remainder_stays_hidden(tmp_ assert StrategyOwnershipStore(store.path).snapshot()[0] == {"005930": 2} +def test_poll_preserves_original_order_metadata_through_ownership_decorator(tmp_path): + store = StrategyOwnershipStore(tmp_path / "ownership.json") + delegate = FakeBroker([]) + broker = OwnershipAwareBroker(delegate, store) + submitted_at = datetime(2026, 8, 14, 6, 20, tzinfo=timezone.utc) + + broker.poll_fill( + "1", + side="buy", + symbol="005930", + order_date="20260814", + submitted_at=submitted_at, + ) + + assert delegate.poll_metadata == ("20260814", submitted_at) + + def test_strategy_quantity_above_raw_broker_total_fails_closed(tmp_path): store = StrategyOwnershipStore(tmp_path / "ownership.json") store.apply_fill( diff --git a/tests/adapters/persistence/test_postgresql_slot_repository_parity.py b/tests/adapters/persistence/test_postgresql_slot_repository_parity.py index 48afe98..e7525b0 100644 --- a/tests/adapters/persistence/test_postgresql_slot_repository_parity.py +++ b/tests/adapters/persistence/test_postgresql_slot_repository_parity.py @@ -36,7 +36,7 @@ _SLOT_COLUMNS = ( "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, " - "cancel_state, cancel_requested_at" + "cancel_state, cancel_requested_at, order_submitted_at, order_date, cancel_attempts" ) _MANAGED_SLOT_SQL = ( f"SELECT {_SLOT_COLUMNS} FROM daily_slot " @@ -149,6 +149,9 @@ def _row(slot: Slot) -> tuple: slot.venue, slot.cancel_state, slot.cancel_requested_at, + slot.order_submitted_at, + slot.order_date, + slot.cancel_attempts, ) diff --git a/tests/application/test_execution_policy.py b/tests/application/test_execution_policy.py index ed0fe3d..17e3a04 100644 --- a/tests/application/test_execution_policy.py +++ b/tests/application/test_execution_policy.py @@ -290,7 +290,8 @@ class TestTimeoutCancel: # age the slot past timeout s = repo.fetch_slot(sid) repo._slots[sid] = replace( - s, updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60), + s, + order_submitted_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60), ) r = execution.run_once() assert r.timeout_cancelled == 1 @@ -310,13 +311,13 @@ class TestTimeoutCancel: assert r.timeout_cancelled == 0 assert len(broker.cancel_calls) == 0 - def test_no_timeout_sec_no_cancel(self): + def test_updated_at_does_not_move_hard_fallback_deadline(self): broker = _RecordingBroker() repo, svc, intents, execution = _wire(broker) sid = _buy_to_pending(intents, timeout_sec=None) broker.responder = lambda side: FillStatus(state="accepted") execution.run_once() # dispatch - # age the slot + # Mutable row bookkeeping is not the dispatch/deadline clock. s = repo.fetch_slot(sid) repo._slots[sid] = replace( s, updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=9999), @@ -335,7 +336,8 @@ class TestTimeoutCancel: execution.run_once() # dispatch s = repo.fetch_slot(sid) repo._slots[sid] = replace( - s, updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60), + s, + order_submitted_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60), ) with caplog.at_level(logging.WARNING): r = execution.run_once() @@ -357,7 +359,8 @@ class TestTimeoutCancel: execution.run_once() s = repo.fetch_slot(sid) repo._slots[sid] = replace( - s, updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60), + s, + order_submitted_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60), ) execution.run_once() # slot still PENDING_BUY — cancel is best-effort; rollback on next poll @@ -376,7 +379,7 @@ class TestTimeoutCancel: slot = repo.fetch_slot(sid) repo._slots[sid] = replace( slot, - updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60), + order_submitted_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60), ) result = execution.run_once() @@ -395,7 +398,8 @@ class TestTimeoutCancel: # age the slot → cancel sent s = repo.fetch_slot(sid) repo._slots[sid] = replace( - s, updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60), + s, + order_submitted_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60), ) r1 = execution.run_once() assert r1.timeout_cancelled == 1 diff --git a/tests/application/test_execution_policy_handler.py b/tests/application/test_execution_policy_handler.py index b9e948c..43a812a 100644 --- a/tests/application/test_execution_policy_handler.py +++ b/tests/application/test_execution_policy_handler.py @@ -75,14 +75,14 @@ class TestIntentPayloadParsing: assert slot.timeout_sec == 60 assert slot.slippage_pct == 1.0 - def test_buy_policy_fields_absent_means_none(self): - """Backward-compat: no policy fields in payload → None on slot.""" + def test_buy_policy_fields_absent_uses_hard_fallback(self): + """No policy field still receives the hard 3,600-second fallback.""" repo, _, intents = _wire() handler = make_intent_rpc_handler(intents) resp = handler(_buy_req()) sid = UUID(resp.payload["slot_id"]) slot = repo.fetch_slot(sid) - assert slot.timeout_sec is None + assert slot.timeout_sec == 3600 assert slot.slippage_pct is None def test_sell_timeout_sec_parsed(self): @@ -166,13 +166,13 @@ class TestDefaultOrderTimeoutSec: slot = repo.fetch_slot(d.slot_id) assert slot.timeout_sec == 30 - def test_no_config_default_no_intent_timeout_means_none(self): + def test_no_config_default_no_intent_timeout_uses_hard_fallback(self): repo, _, intents = self._wire_with_default(default_timeout=None) d = intents.submit_buy(BuyIntent( intent_id="b1", portfolio_id="PF-1", qty=10, )) slot = repo.fetch_slot(d.slot_id) - assert slot.timeout_sec is None + assert slot.timeout_sec == 3600 # ============================================================================ diff --git a/tests/application/test_order_timeout_reconciliation.py b/tests/application/test_order_timeout_reconciliation.py new file mode 100644 index 0000000..24a3ee9 --- /dev/null +++ b/tests/application/test_order_timeout_reconciliation.py @@ -0,0 +1,342 @@ +"""Synthetic coverage for GitHub #58 timeout/cancel reconciliation.""" +from __future__ import annotations + +from dataclasses import replace +from datetime import datetime, timedelta, timezone +from pathlib import Path +from uuid import UUID + +import pytest + +from trex_core.adapters.broker import kis as kis_mod +from trex_core.adapters.broker.kis import KISBroker +from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore +from trex_core.application.broker_port import CancelOutcome, CancelState, FillStatus +from trex_core.application.execution import ExecutionService +from trex_core.application.intents import BuyIntent, IntentService, SellIntent +from trex_core.application.messaging import RpcRequest +from trex_core.application.messaging_handlers import make_intent_rpc_handler +from trex_core.domain.slot import SlotService, SlotState +from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository +from trex_core.runtime.config import DaemonInstanceConfig +from tests.application.test_execution_policy import _RecordingBroker + +_KST = timezone(timedelta(hours=9)) + + +def _wire(*, default_timeout=None): + repo = InMemorySlotRepository() + service = SlotService(repo) + intents = IntentService( + service, + repo, + InMemoryIntentStore(), + resolve_symbol=lambda _pid: "005930", + default_order_timeout_sec=default_timeout, + ) + repo.reconcile_universe("default", {"PF-1": "005930"}) + return repo, service, intents + + +@pytest.mark.parametrize("value", [0, -1, 3601, True, 1.5, "30"]) +def test_direct_strategy_timeout_outside_integer_bounds_is_rejected(value): + repo, _, intents = _wire() + decision = intents.submit_buy( + BuyIntent(intent_id=f"b-{value!r}", portfolio_id="PF-1", qty=1, timeout_sec=value) + ) + assert decision.status == "rejected" + assert decision.reason.startswith("bad_request:timeout_sec_") + assert repo.list_undispatched_pending() == [] + + +@pytest.mark.parametrize("value", [0, -1, 3601, True, 1.5, "30"]) +def test_direct_sell_timeout_outside_integer_bounds_is_rejected(value): + repo, _, intents = _wire() + decision = intents.submit_sell( + SellIntent(intent_id=f"s-{value!r}", slot_id=UUID(int=1), timeout_sec=value) + ) + assert decision.status == "rejected" + assert decision.reason.startswith("bad_request:timeout_sec_") + assert repo.list_undispatched_pending() == [] + + +@pytest.mark.parametrize("value", [0, -1, 3601, 1.5, True, "30", "not-an-int"]) +@pytest.mark.parametrize("intent_type", ["buy", "sell"]) +def test_wire_strategy_timeout_bad_request_before_admission(value, intent_type): + repo, _, intents = _wire() + payload = { + "intent_type": intent_type, + "intent_id": "bad-timeout", + "timeout_sec": value, + } + if intent_type == "buy": + payload.update({"portfolio_id": "PF-1", "qty": 1}) + else: + payload["slot_id"] = str(UUID(int=1)) + response = make_intent_rpc_handler(intents)( + RpcRequest( + correlation_id="bad-timeout", + payload=payload, + ) + ) + assert response.payload["status"] == "rejected" + assert response.payload["reason"].startswith("bad_request:") + assert repo.list_undispatched_pending() == [] + + +@pytest.mark.parametrize("value", [1, 3600]) +def test_strategy_timeout_inclusive_bounds_are_admitted(value): + repo, _, intents = _wire() + decision = intents.submit_buy( + BuyIntent(intent_id=f"b-{value}", portfolio_id="PF-1", qty=1, timeout_sec=value) + ) + assert decision.admitted + assert repo.fetch_slot(decision.slot_id).timeout_sec == value + + +def test_buy_and_sell_omission_use_instance_override_or_hard_fallback(): + repo, service, intents = _wire(default_timeout=120) + buy = intents.submit_buy(BuyIntent(intent_id="b1", portfolio_id="PF-1", qty=1)) + assert repo.fetch_slot(buy.slot_id).timeout_sec == 120 + service.transit_slot( + buy.slot_id, + SlotState.PENDING_BUY, + SlotState.HOLDING, + reason="fixture", + actor="test", + qty=1, + avg_price=1000, + ) + sell = intents.submit_sell(SellIntent(intent_id="s1", slot_id=buy.slot_id)) + assert sell.admitted + assert repo.fetch_slot(buy.slot_id).timeout_sec == 120 + + fallback_repo, _, fallback = _wire(default_timeout=None) + fallback_buy = fallback.submit_buy( + BuyIntent(intent_id="fallback", portfolio_id="PF-1", qty=1) + ) + assert fallback_repo.fetch_slot(fallback_buy.slot_id).timeout_sec == 3600 + + +@pytest.mark.parametrize("value", [0, -1, 3601, 1.5, True, "30"]) +def test_instance_timeout_outside_toml_integer_contract_fails_startup(tmp_path, value): + path = tmp_path / "instance.toml" + literal = str(value).lower() if isinstance(value, bool) else repr(value) + path.write_text(f'instance_key = "x"\ndefault_order_timeout_sec = {literal}\n') + with pytest.raises(RuntimeError, match="default_order_timeout_sec"): + DaemonInstanceConfig.from_toml(str(path)) + + +def test_instance_timeout_omission_and_bounds(tmp_path): + path = tmp_path / "instance.toml" + path.write_text('instance_key = "x"\n') + assert DaemonInstanceConfig.from_toml(str(path)).default_order_timeout_sec == 3600 + for value in (1, 3600): + path.write_text(f'instance_key = "x"\ndefault_order_timeout_sec = {value}\n') + assert DaemonInstanceConfig.from_toml(str(path)).default_order_timeout_sec == value + + +def test_migration_does_not_invent_legacy_submission_metadata(): + migration = ( + Path(__file__).parents[2] + / "trex_core/adapters/persistence/sql/022_persist_order_timeout_anchor.sql" + ).read_text(encoding="utf-8") + + assert "ADD COLUMN IF NOT EXISTS order_submitted_at" in migration + assert "ADD COLUMN IF NOT EXISTS order_date" in migration + assert "UPDATE daily_slot" not in migration + assert "COALESCE(updated_at, created_at)" not in migration + + +def test_dispatch_anchor_and_original_date_survive_bookkeeping_and_restart(): + broker = _RecordingBroker() + repo, service, intents = _wire() + execution = ExecutionService(service, repo, broker) + decision = intents.submit_buy( + BuyIntent(intent_id="b1", portfolio_id="PF-1", qty=1, timeout_sec=60) + ) + broker.responder = lambda _side: FillStatus(state="accepted") + execution.run_once() + dispatched = repo.fetch_slot(decision.slot_id) + submitted_at = dispatched.order_submitted_at + order_date = dispatched.order_date + assert submitted_at is not None and order_date == submitted_at.astimezone(_KST).strftime("%Y%m%d") + + assert repo.replace_dispatch(dispatched.slot_id, dispatched.broker_order_id, "resolved-1") + resolved = repo.fetch_slot(dispatched.slot_id) + assert resolved.order_submitted_at == submitted_at + assert resolved.order_date == order_date + + repo.mark_cancel_outcome(dispatched.slot_id, "resolved-1", "unknown") + persisted = repo.fetch_slot(dispatched.slot_id) + assert persisted.updated_at != dispatched.updated_at + assert persisted.order_submitted_at == submitted_at + assert persisted.order_date == order_date + assert ExecutionService(service, repo, broker).reconcile().filled == 0 + restarted = repo.fetch_slot(dispatched.slot_id) + assert restarted.order_submitted_at == submitted_at + assert restarted.order_date == order_date + + service.transit_slot( + dispatched.slot_id, + SlotState.PENDING_BUY, + SlotState.HOLDING, + reason="terminal-fixture", + actor="test", + qty=1, + avg_price=1000, + ) + repo.clear_dispatch(dispatched.slot_id) + terminal = repo.fetch_slot(dispatched.slot_id) + assert terminal.order_submitted_at is None + assert terminal.order_date is None + assert terminal.cancel_attempts == 0 + + +class _AlarmRecorder: + def __init__(self): + self.calls = [] + + def publish_alarm(self, code, title, *, fields=None, severity="critical"): + self.calls.append((code, severity, fields)) + + +def test_legacy_pending_without_authoritative_metadata_is_fail_closed_and_alarm_bounded(): + broker = _RecordingBroker() + repo, service, intents = _wire() + alarms = _AlarmRecorder() + execution = ExecutionService(service, repo, broker, alarm_emitter=alarms) + decision = intents.submit_buy( + BuyIntent(intent_id="legacy", portfolio_id="PF-1", qty=1, timeout_sec=1) + ) + broker.responder = lambda _side: FillStatus(state="accepted") + execution.run_once() + slot = repo.fetch_slot(decision.slot_id) + repo._slots[slot.slot_id] = replace( + slot, + order_submitted_at=None, + order_date=None, + updated_at=datetime.now(tz=timezone.utc) - timedelta(days=2), + ) + + execution.run_once() + execution.run_once() + + persisted = repo.fetch_slot(slot.slot_id) + assert persisted.status is SlotState.PENDING_BUY + assert broker.cancel_calls == [] + metadata_alarms = [ + call for call in alarms.calls if call[0] == "order_reconciliation_metadata_unknown" + ] + assert len(metadata_alarms) == 1 + assert set(metadata_alarms[0][2]) == {"slot_id", "symbol"} + + +def test_cancel_rejected_is_fail_closed_alarm_and_authoritative_bounded_retry(): + broker = _RecordingBroker() + broker.cancel_outcome = CancelOutcome(CancelState.REJECTED, "closed_session") + repo, service, intents = _wire() + alarms = _AlarmRecorder() + execution = ExecutionService(service, repo, broker, alarm_emitter=alarms) + decision = intents.submit_buy( + BuyIntent(intent_id="b1", portfolio_id="PF-1", qty=1, timeout_sec=1) + ) + broker.responder = lambda _side: FillStatus(state="accepted") + execution.run_once() + slot = repo.fetch_slot(decision.slot_id) + repo._slots[slot.slot_id] = replace( + slot, + order_submitted_at=datetime.now(tz=timezone.utc) - timedelta(minutes=1), + ) + execution.run_once() + rejected = repo.fetch_slot(slot.slot_id) + assert rejected.status is SlotState.PENDING_BUY + assert rejected.cancel_state == "rejected" + assert len(broker.cancel_calls) == 1 + assert any(code == "order_cancel_unresolved" and severity == "critical" for code, severity, _ in alarms.calls) + + # No visible broker row => no blind retry, even after restart/backoff. + repo._slots[slot.slot_id] = replace( + rejected, + cancel_requested_at=datetime.now(tz=timezone.utc) - timedelta(minutes=10), + ) + ExecutionService(service, repo, broker, alarm_emitter=alarms).run_once() + assert len(broker.cancel_calls) == 1 + + # An authoritative open row permits one backoff-gated retry. + broker.responder = lambda _side: FillStatus( + state="accepted", raw={"order_open_confirmed": True} + ) + broker.cancel_outcome = CancelOutcome(CancelState.REQUESTED) + ExecutionService(service, repo, broker, alarm_emitter=alarms).run_once() + assert len(broker.cancel_calls) == 2 + retried = repo.fetch_slot(slot.slot_id) + assert retried.cancel_state == "requested" + assert retried.cancel_attempts == 2 + ExecutionService(service, repo, broker).run_once() + assert len(broker.cancel_calls) == 2 + + +class _Resp: + status_code = 200 + + def __init__(self, payload): + self._payload = payload + + def json(self): + return self._payload + + +def _kis(clock): + broker = KISBroker( + "12345678-01", + app_key="k", + app_secret="s", + base_url="https://openapivts.koreainvestment.com:29443", + clock=clock, + ) + broker._token = "token" + broker._token_exp = datetime.now(tz=timezone.utc) + timedelta(hours=1) + return broker + + +def test_kis_rollover_uses_original_date_and_row_visibility_is_nonterminal(monkeypatch): + now = datetime(2026, 8, 15, 9, 0, tzinfo=_KST) + submitted = datetime(2026, 8, 14, 15, 20, tzinfo=_KST) + seen = {} + + def fake_get(*_args, **kwargs): + seen.update(kwargs["params"]) + return _Resp({"rt_cd": "0", "output1": []}) + + monkeypatch.setattr(kis_mod.httpx, "get", fake_get) + status = _kis(lambda: now).poll_fill( + "06010:0000123456", + side="buy", + symbol="005930", + order_date="20260814", + submitted_at=submitted, + ) + assert seen["INQR_STRT_DT"] == seen["INQR_END_DT"] == "20260814" + assert status.state == "unknown" + assert status.raw["reason"] == "order_not_visible_unresolved" + assert status.raw["order_visible"] is False + + +def test_kis_recent_missing_row_stays_in_bounded_visibility_grace(monkeypatch): + now = datetime(2026, 8, 14, 9, 0, tzinfo=_KST) + monkeypatch.setattr( + kis_mod.httpx, + "get", + lambda *_args, **_kwargs: _Resp({"rt_cd": "0", "output1": []}), + ) + status = _kis(lambda: now).poll_fill( + "06010:0000123457", + side="sell", + symbol="005930", + order_date="20260814", + submitted_at=now - timedelta(seconds=30), + ) + assert status.state == "accepted" + assert status.raw["reason"] == "order_visibility_grace" + assert status.raw["order_visible"] is False diff --git a/tests/integration/test_kis_submission_recovery_virtual.py b/tests/integration/test_kis_submission_recovery_virtual.py index 67cb87d..1d316d9 100644 --- a/tests/integration/test_kis_submission_recovery_virtual.py +++ b/tests/integration/test_kis_submission_recovery_virtual.py @@ -144,7 +144,13 @@ def _wire(broker: KISBroker): instance_key=INSTANCE, ) repo.reconcile_universe(INSTANCE, {PORTFOLIO: SYMBOL}) - execution = ExecutionService(slots, repo, broker, instance_key=INSTANCE) + execution = ExecutionService( + slots, + repo, + broker, + instance_key=INSTANCE, + clock=lambda: VIRTUAL_AT, + ) decision = intents.submit_buy( BuyIntent( intent_id="virtual-buy-1", @@ -170,7 +176,13 @@ def test_timeout_after_exchange_accept_resolves_without_duplicate_and_survives_r assert len(state.orders) == 1 state.fill_orders = True - restarted = ExecutionService(slots, repo, _broker(server), instance_key=INSTANCE) + restarted = ExecutionService( + slots, + repo, + _broker(server), + instance_key=INSTANCE, + clock=lambda: VIRTUAL_AT, + ) restarted.reconcile() holding = repo.fetch_slot(slot_id) diff --git a/trex_core/adapters/broker/kis.py b/trex_core/adapters/broker/kis.py index 80054af..a2a1d25 100644 --- a/trex_core/adapters/broker/kis.py +++ b/trex_core/adapters/broker/kis.py @@ -307,6 +307,7 @@ class KISBroker: reason=f"submit_outcome_unknown:{type(exc).__name__}", venue=actual_venue, raw={"error_class": type(exc).__name__}, + submitted_at=submitted_at, ) if getattr(resp, "status_code", 200) >= 500: return OrderAck( @@ -315,6 +316,7 @@ class KISBroker: reason=f"submit_outcome_unknown:http_{resp.status_code}", venue=actual_venue, raw={"http_status": resp.status_code}, + submitted_at=submitted_at, ) if str(payload.get("rt_cd")) == "0": out = payload.get("output") or {} @@ -326,6 +328,7 @@ class KISBroker: broker_order_id=bid, venue=actual_venue, raw=payload, + submitted_at=submitted_at, ) return OrderAck( accepted=False, @@ -362,7 +365,15 @@ class KISBroker: "EXCG_ID_DVSN_CD": "ALL", } - def poll_fill(self, broker_order_id: str, *, side: str, symbol: str) -> FillStatus: + def poll_fill( + self, + broker_order_id: str, + *, + side: str, + symbol: str, + order_date: Optional[str] = None, + submitted_at: Optional[datetime] = None, + ) -> FillStatus: if broker_order_id.startswith(_SUBMISSION_UNKNOWN_PREFIX): return self._poll_unknown_submission(broker_order_id, side=side, symbol=symbol) now = monotonic() @@ -375,8 +386,8 @@ class KISBroker: # open duplicate KIS inquiries for the same order. self._fill_poll_not_before[broker_order_id] = now + _FILL_POLL_MIN_INTERVAL_SECONDS odno = broker_order_id.split(":")[-1] - today = datetime.now(tz=_KST).strftime("%Y%m%d") - tr_id, params = self._daily_fill_request(today, order_number=odno) + inquiry_date = order_date or self._clock().astimezone(_KST).strftime("%Y%m%d") + tr_id, params = self._daily_fill_request(inquiry_date, order_number=odno) try: resp = httpx.get( f"{self._base_url}/uapi/domestic-stock/v1/trading/inquire-daily-ccld", @@ -410,11 +421,36 @@ class KISBroker: None, ) if row is None: - return FillStatus( - state="accepted", - raw={"reason": "order_not_visible_in_fill_inquiry"}, + visible_grace = ( + submitted_at is None + or (self._clock().astimezone(timezone.utc) - submitted_at.astimezone(timezone.utc)) + .total_seconds() + < _SUBMISSION_LOOKUP_GRACE_SECONDS ) - return self._fill_status_from_row(row, side=side, today=today, fallback_odno=odno) + return FillStatus( + state="accepted" if visible_grace else "unknown", + raw={ + "reason": ( + "order_visibility_grace" + if visible_grace + else "order_not_visible_unresolved" + ), + "order_visible": False, + "inquiry_date": inquiry_date, + }, + ) + status = self._fill_status_from_row( + row, side=side, today=inquiry_date, fallback_odno=odno + ) + return replace( + status, + raw={ + **(status.raw if isinstance(status.raw, dict) else {}), + "order_visible": True, + "order_open_confirmed": status.state in {"accepted", "partially_filled"}, + "inquiry_date": inquiry_date, + }, + ) def _poll_unknown_submission( self, diff --git a/trex_core/adapters/broker/ownership.py b/trex_core/adapters/broker/ownership.py index f174e72..63a3ffa 100644 --- a/trex_core/adapters/broker/ownership.py +++ b/trex_core/adapters/broker/ownership.py @@ -10,8 +10,10 @@ from __future__ import annotations import json import os +from datetime import datetime from pathlib import Path from threading import Lock +from typing import Optional from trex_core.application.broker_port import ( BalanceItem, @@ -126,8 +128,22 @@ class OwnershipAwareBroker: def submit_order(self, req: OrderRequest) -> OrderAck: return self._delegate.submit_order(req) - def poll_fill(self, broker_order_id: str, *, side: str, symbol: str) -> FillStatus: - result = self._delegate.poll_fill(broker_order_id, side=side, symbol=symbol) + def poll_fill( + self, + broker_order_id: str, + *, + side: str, + symbol: str, + order_date: Optional[str] = None, + submitted_at: Optional[datetime] = None, + ) -> FillStatus: + result = self._delegate.poll_fill( + broker_order_id, + side=side, + symbol=symbol, + order_date=order_date, + submitted_at=submitted_at, + ) if result.execution_id and result.filled_qty > 0 and ( result.is_filled or result.is_partial_cancel ): diff --git a/trex_core/adapters/broker/paper.py b/trex_core/adapters/broker/paper.py index e95ffc6..8b3582c 100644 --- a/trex_core/adapters/broker/paper.py +++ b/trex_core/adapters/broker/paper.py @@ -9,6 +9,7 @@ from __future__ import annotations import threading from decimal import Decimal +from datetime import datetime from typing import Optional from trex_core.adapters.broker.seq_store import SeqStore @@ -80,7 +81,15 @@ class PaperBroker: self._orders[oid] = _Order(oid, req) return OrderAck(accepted=True, broker_order_id=oid, venue=req.venue) - def poll_fill(self, broker_order_id: str, *, side: str, symbol: str) -> FillStatus: + def poll_fill( + self, + broker_order_id: str, + *, + side: str, + symbol: str, + order_date: Optional[str] = None, + submitted_at: Optional[datetime] = None, + ) -> FillStatus: with self._lock: o = self._orders.get(broker_order_id) if o is None: @@ -91,7 +100,12 @@ class PaperBroker: if symbol in self._reject: return FillStatus(state="rejected", raw={"reason": "paper_reject"}) if o.polls <= self._fill_after: - return FillStatus(state="accepted", filled_qty=o.filled, avg_price=o.price) + return FillStatus( + state="accepted", + filled_qty=o.filled, + avg_price=o.price, + raw={"order_open_confirmed": True}, + ) step = self._qty_per_poll if self._qty_per_poll is not None else (o.qty - o.filled) o.filled = min(o.qty, o.filled + step) if o.filled >= o.qty: @@ -105,7 +119,12 @@ class PaperBroker: state="filled", filled_qty=o.filled, avg_price=o.price, execution_id=execution_id, fee=fee, tax=tax, ) - return FillStatus(state="partially_filled", filled_qty=o.filled, avg_price=o.price) + return FillStatus( + state="partially_filled", + filled_qty=o.filled, + avg_price=o.price, + raw={"order_open_confirmed": True}, + ) def cancel_order( self, diff --git a/trex_core/adapters/persistence/postgresql.py b/trex_core/adapters/persistence/postgresql.py index c5ad040..07c0a31 100644 --- a/trex_core/adapters/persistence/postgresql.py +++ b/trex_core/adapters/persistence/postgresql.py @@ -14,7 +14,7 @@ from __future__ import annotations import json from contextlib import contextmanager -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from typing import Any, Iterator, Optional from uuid import UUID, uuid4 @@ -38,9 +38,11 @@ _SLOT_COLUMNS = ( "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, " - "cancel_state, cancel_requested_at" + "cancel_state, cancel_requested_at, order_submitted_at, order_date, cancel_attempts" ) +_KST = timezone(timedelta(hours=9)) + def _require_psycopg(module: Any = None) -> Any: if module is not None: @@ -218,7 +220,8 @@ class PostgreSQLSlotRepository: "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, cancel_state=%s, " - "cancel_requested_at=%s, updated_at=%s WHERE slot_id=%s", + "cancel_requested_at=%s, order_submitted_at=%s, order_date=%s, " + "cancel_attempts=%s, updated_at=%s WHERE slot_id=%s", ( slot.status.value, slot.symbol, @@ -241,6 +244,9 @@ class PostgreSQLSlotRepository: slot.venue, slot.cancel_state, slot.cancel_requested_at, + slot.order_submitted_at, + slot.order_date, + slot.cancel_attempts, slot.updated_at or datetime.now(tz=timezone.utc), slot.slot_id, ), @@ -295,16 +301,22 @@ class PostgreSQLSlotRepository: broker_order_id: str, *, venue: Optional[str] = None, + submitted_at: Optional[datetime] = None, ) -> bool: + submitted_at = submitted_at or datetime.now(tz=timezone.utc) + order_date = submitted_at.astimezone(_KST).strftime("%Y%m%d") with self._psycopg.connect(self._dsn, autocommit=False) as conn: with conn.cursor() as cur: cur.execute( "UPDATE daily_slot SET broker_order_id=%s, " - "venue=COALESCE(%s, venue), updated_at=%s " + "venue=COALESCE(%s, venue), order_submitted_at=%s, order_date=%s, " + "cancel_attempts=0, updated_at=%s " "WHERE slot_id=%s AND broker_order_id IS NULL", ( broker_order_id, venue, + submitted_at, + order_date, datetime.now(tz=timezone.utc), slot_id, ), @@ -366,6 +378,31 @@ class PostgreSQLSlotRepository: conn.commit() return ok + def reserve_cancel_attempt( + self, + slot_id: UUID, + broker_order_id: str, + *, + retry_before: Optional[datetime] = None, + ) -> bool: + """Atomically reserve first cancel or an authoritative backoff-gated retry.""" + 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='unknown', cancel_requested_at=%s, " + "cancel_attempts=cancel_attempts+1, updated_at=%s " + "WHERE slot_id=%s AND broker_order_id=%s " + "AND status IN ('pending_buy','pending_sell') " + "AND (cancel_state IS NULL OR " + "(%s IS NOT NULL AND cancel_state IN ('rejected','unknown') " + "AND cancel_attempts < 4 AND cancel_requested_at <= %s))", + (now, now, slot_id, broker_order_id, retry_before, retry_before), + ) + 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: @@ -382,7 +419,8 @@ class PostgreSQLSlotRepository: with conn.cursor() as cur: cur.execute( "UPDATE daily_slot SET broker_order_id=NULL, cancel_state=NULL, " - "cancel_requested_at=NULL, updated_at=%s WHERE slot_id=%s", + "cancel_requested_at=NULL, order_submitted_at=NULL, order_date=NULL, " + "cancel_attempts=0, updated_at=%s WHERE slot_id=%s", (datetime.now(tz=timezone.utc), slot_id), ) conn.commit() @@ -410,7 +448,8 @@ class PostgreSQLSlotRepository: # 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), - # cancel_state(22), cancel_requested_at(23) + # cancel_state(22), cancel_requested_at(23), order_submitted_at(24), + # order_date(25), cancel_attempts(26) return Slot( slot_id=row[0] if isinstance(row[0], UUID) else UUID(str(row[0])), status=SlotState(row[1]), @@ -436,6 +475,9 @@ class PostgreSQLSlotRepository: 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, + order_submitted_at=row[24] if len(row) > 24 else None, + order_date=str(row[25]) if len(row) > 25 and row[25] is not None else None, + cancel_attempts=int(row[26]) if len(row) > 26 and row[26] is not None else 0, ) diff --git a/trex_core/adapters/persistence/sql/022_persist_order_timeout_anchor.sql b/trex_core/adapters/persistence/sql/022_persist_order_timeout_anchor.sql new file mode 100644 index 0000000..7b90f06 --- /dev/null +++ b/trex_core/adapters/persistence/sql/022_persist_order_timeout_anchor.sql @@ -0,0 +1,27 @@ +-- GitHub #58: persist the effective timeout's immutable dispatch anchor and +-- original KST order date separately from mutable row bookkeeping. Existing +-- dispatched rows deliberately remain NULL: updated_at/created_at are mutable +-- bookkeeping and cannot prove the original submission timestamp or order day. +-- Runtime keeps those legacy rows fail-closed and raises a sanitized alarm. +ALTER TABLE daily_slot + ADD COLUMN IF NOT EXISTS order_submitted_at TIMESTAMPTZ; + +ALTER TABLE daily_slot + ADD COLUMN IF NOT EXISTS order_date TEXT; + +ALTER TABLE daily_slot + ADD COLUMN IF NOT EXISTS cancel_attempts INTEGER NOT NULL DEFAULT 0; + +ALTER TABLE daily_slot + DROP CONSTRAINT IF EXISTS daily_slot_order_date_check; + +ALTER TABLE daily_slot + ADD CONSTRAINT daily_slot_order_date_check + CHECK (order_date IS NULL OR order_date ~ '^[0-9]{8}$'); + +ALTER TABLE daily_slot + DROP CONSTRAINT IF EXISTS daily_slot_cancel_attempts_check; + +ALTER TABLE daily_slot + ADD CONSTRAINT daily_slot_cancel_attempts_check + CHECK (cancel_attempts >= 0); diff --git a/trex_core/application/broker_port.py b/trex_core/application/broker_port.py index 7828198..afb16a0 100644 --- a/trex_core/application/broker_port.py +++ b/trex_core/application/broker_port.py @@ -9,6 +9,7 @@ from __future__ import annotations import os from dataclasses import dataclass, field +from datetime import datetime from decimal import Decimal from enum import Enum from typing import Any, Optional, Protocol @@ -75,6 +76,7 @@ class OrderAck: # never has to infer the original venue. venue: Optional[str] = None raw: dict[str, Any] = field(default_factory=dict) + submitted_at: Optional[datetime] = None @dataclass(frozen=True) @@ -135,7 +137,15 @@ class BrokerPort(Protocol): """Submit an order. Returns accepted + broker_order_id (no synchronous fill).""" ... - def poll_fill(self, broker_order_id: str, *, side: str, symbol: str) -> FillStatus: + def poll_fill( + self, + broker_order_id: str, + *, + side: str, + symbol: str, + order_date: Optional[str] = None, + submitted_at: Optional[datetime] = None, + ) -> FillStatus: """Observe current fill status for a previously accepted order.""" ... diff --git a/trex_core/application/execution.py b/trex_core/application/execution.py index 80cace5..b7f8f87 100644 --- a/trex_core/application/execution.py +++ b/trex_core/application/execution.py @@ -23,9 +23,10 @@ single producer for ledger work items (NFR-3, single record subject). from __future__ import annotations +import inspect import logging from dataclasses import dataclass, replace -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from typing import Callable, Optional from trex_core.application.alarm_emitter import AlarmEmitter @@ -86,6 +87,7 @@ class ExecutionService: instance_key: str = "default", alarm_emitter: Optional[AlarmEmitter] = None, balance_reader: Optional[Callable[[], list[BalanceItem]]] = None, + clock: Optional[Callable[[], datetime]] = None, ) -> None: self._svc = slot_service self._repo = repo @@ -106,6 +108,11 @@ class ExecutionService: ) # dcy-033 / REQ-100: best-effort alarm emitter (None = no alarms sent). self._alarm_emitter = alarm_emitter + # One emission per durable unresolved state/attempt in this process. + # A restart intentionally emits once again, covering a crash between a + # durable cancel reservation and best-effort alarm publication. + self._order_alarm_keys: set[tuple[str, str, int]] = set() + self._clock = clock or (lambda: datetime.now(tz=timezone.utc)) # dcy-041 / GitHub #45: pending-sell quantity may use a bounded # account snapshot only through this injected reader. Reconciliation # remains wired directly to the live broker. @@ -216,6 +223,7 @@ class ExecutionService: slot.slot_id, ack.broker_order_id, venue=ack.venue, + submitted_at=ack.submitted_at, ) r.dispatched += 1 if ack.reason and ack.reason.startswith("submit_outcome_unknown:"): @@ -258,11 +266,11 @@ class ExecutionService: r.rolled_back += 1 def _poll(self, r: ExecutionResult) -> None: - now = datetime.now(tz=timezone.utc) + now = self._clock().astimezone(timezone.utc) for slot in self._repo.list_dispatched_pending(self._instance_key): side = "buy" if slot.status is SlotState.PENDING_BUY else "sell" set_correlation_id(str(slot.slot_id)) # REQ-084: same key as submit - fs = self._broker.poll_fill(slot.broker_order_id, side=side, symbol=slot.symbol or "") + fs = self._poll_fill(slot, side) resolved_order_id = ( fs.raw.get("resolved_broker_order_id") if isinstance(fs.raw, dict) else None ) @@ -312,7 +320,18 @@ class ExecutionService: # tick, but surface if it has been stuck too long (FR-3), and # send a cancel if the slot-level timeout is exceeded (REQ-069). self._check_stale_pending(slot, now, r) - self._try_timeout_cancel(slot, now, side, r) + self._try_timeout_cancel(slot, now, side, fs, r) + + def _poll_fill(self, slot, side: str): + """Use immutable order metadata when the adapter supports the new seam.""" + poll = self._broker.poll_fill + parameters = inspect.signature(poll).parameters + kwargs = {"side": side, "symbol": slot.symbol or ""} + if "order_date" in parameters: + kwargs["order_date"] = slot.order_date + if "submitted_at" in parameters: + kwargs["submitted_at"] = slot.order_submitted_at + return poll(slot.broker_order_id, **kwargs) # ------------------------------------------------------------ def _balance_map(self) -> dict: @@ -553,39 +572,40 @@ class ExecutionService: self._stale_threshold_sec, ) - def _try_timeout_cancel(self, slot, now: datetime, side: str, r: ExecutionResult) -> None: - """Send cancel_order when the slot-level timeout is exceeded (dcy-022/REQ-069). - - Best-effort: a failure only logs a warning — the slot is NOT changed here. - Actual rollback happens on the next poll tick when the broker confirms the - cancel (is_terminal_reject) or partial fill then cancel (is_partial_cancel). - This design is safe across the cancel-fill race: if the cancel races with a - fill, the next poll sees is_filled and follows the normal fill path (FR-4). - """ + def _try_timeout_cancel(self, slot, now: datetime, side: str, fs, r: ExecutionResult) -> None: + """Begin or safely retry cancellation after the immutable deadline.""" if str(slot.broker_order_id or "").startswith("submission-unknown:"): return - if slot.cancel_state is not None: + if slot.timeout_sec is None: return - if slot.timeout_sec is None or slot.updated_at is None: + if slot.order_submitted_at is None or slot.order_date is None: + self._emit_order_metadata_unknown_alarm(slot) return - age = (now - slot.updated_at).total_seconds() + age = (now - slot.order_submitted_at).total_seconds() if age < slot.timeout_sec: return + + is_retry = slot.cancel_state in { + CancelState.REJECTED.value, + CancelState.UNKNOWN.value, + } + if slot.cancel_state is not None and not is_retry: + return + if is_retry: + self._emit_unresolved_cancel_alarm(slot) + if not (isinstance(fs.raw, dict) and fs.raw.get("order_open_confirmed") is True): + return + backoff_sec = min(300, 5 * (2 ** max(0, slot.cancel_attempts - 1))) + retry_before = now - timedelta(seconds=backoff_sec) + else: + retry_before = None + 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( + if not self._repo.reserve_cancel_attempt( slot.slot_id, str(slot.broker_order_id), - CancelState.UNKNOWN.value, - only_if_unset=True, + retry_before=retry_before, ): - _log.error( - "cancel attempt reservation conflict: slot=%s broker_order_id=%s", - slot.slot_id, - slot.broker_order_id, - ) return try: cancel_kwargs = { @@ -610,8 +630,7 @@ class ExecutionService: ) if not persisted: _log.error( - "cancel outcome persistence conflict: slot=%s broker_order_id=%s " - "outcome=%s", + "cancel outcome persistence conflict: slot=%s broker_order_id=%s outcome=%s", slot.slot_id, slot.broker_order_id, outcome.state.value, @@ -619,6 +638,14 @@ class ExecutionService: return if outcome.is_requested: r.timeout_cancelled += 1 + elif outcome.state in {CancelState.REJECTED, CancelState.UNKNOWN}: + self._emit_unresolved_cancel_alarm( + replace( + slot, + cancel_state=outcome.state.value, + cancel_attempts=slot.cancel_attempts + 1, + ) + ) _log.warning( "timeout cancel result: slot=%s status=%s broker_order_id=%s " "age_sec=%.0f timeout_sec=%d outcome=%s", @@ -630,6 +657,52 @@ class ExecutionService: outcome.state.value, ) + def _emit_unresolved_cancel_alarm(self, slot) -> None: + key = (str(slot.slot_id), str(slot.cancel_state), slot.cancel_attempts) + if key in self._order_alarm_keys: + return + self._order_alarm_keys.add(key) + _log.critical( + "cancel unresolved: slot=%s status=%s broker_order_id=%s outcome=%s " + "(fail-closed pending; authoritative polling continues)", + slot.slot_id, + slot.status.value, + slot.broker_order_id, + slot.cancel_state, + ) + if self._alarm_emitter is not None: + self._alarm_emitter.publish_alarm( + "order_cancel_unresolved", + f"주문 취소 미확정: 슬롯 pending 유지 (slot={slot.slot_id})", + fields={ + "slot_id": str(slot.slot_id), + "symbol": slot.symbol, + "cancel_state": slot.cancel_state, + "order_date": slot.order_date, + }, + ) + + def _emit_order_metadata_unknown_alarm(self, slot) -> None: + key = (str(slot.slot_id), "order_metadata_unknown", slot.cancel_attempts) + if key in self._order_alarm_keys: + return + self._order_alarm_keys.add(key) + _log.critical( + "order reconciliation metadata unknown: slot=%s status=%s " + "(fail-closed pending; no timeout cancel retry)", + slot.slot_id, + slot.status.value, + ) + if self._alarm_emitter is not None: + self._alarm_emitter.publish_alarm( + "order_reconciliation_metadata_unknown", + f"주문 원시각/원주문일 미확인: 슬롯 pending 유지 (slot={slot.slot_id})", + fields={ + "slot_id": str(slot.slot_id), + "symbol": slot.symbol, + }, + ) + def _emit_event(self, event_type: str, slot, routing_strategy_id: Optional[str] = None) -> None: if self._event_emitter is None: return @@ -734,11 +807,9 @@ class ExecutionService: cancel_kwargs["venue"] = slot.venue # Durable unknown reservation closes the crash window between the # outbound request and saving its immediate result. - if not self._repo.mark_cancel_outcome( + if not self._repo.reserve_cancel_attempt( slot.slot_id, str(slot.broker_order_id), - CancelState.UNKNOWN.value, - only_if_unset=True, ): return "cancel_already_recorded" outcome = self._broker.cancel_order(slot.broker_order_id, **cancel_kwargs) diff --git a/trex_core/application/intents.py b/trex_core/application/intents.py index 089fd75..d0f05cb 100644 --- a/trex_core/application/intents.py +++ b/trex_core/application/intents.py @@ -38,6 +38,7 @@ from trex_core.domain.slot import ( ) _MAX_ADMIT_RETRY = 8 +_HARD_ORDER_TIMEOUT_SEC = 3600 @dataclass(frozen=True) @@ -129,9 +130,15 @@ class IntentService: # expected to set max_active_slots in the instance TOML — design §5). self._instance_key = instance_key self._max_active_slots = max_active_slots - # dcy-022 / REQ-068: default timeout when intent omits timeout_sec. - # None = OFF = indefinite wait (legacy behavior). Per-intent value overrides. - self._default_order_timeout_sec = default_order_timeout_sec + # Omission never disables timeout. Runtime validates configured values; + # this seam remains fail-safe for direct application construction. + self._default_order_timeout_sec = ( + default_order_timeout_sec + if isinstance(default_order_timeout_sec, int) + and not isinstance(default_order_timeout_sec, bool) + and 1 <= default_order_timeout_sec <= _HARD_ORDER_TIMEOUT_SEC + else _HARD_ORDER_TIMEOUT_SEC + ) # BAT-050 / REQ-104: reconcile-warning gate store (Optional). # When None, gate is skipped (backward compat, NFR-4). self._warn_store = warn_store @@ -219,6 +226,9 @@ class IntentService: dcy-036 / REQ-114: entry gate — if entry_allowed_fn returns False, BuyIntent is rejected before any slot lookup.""" + timeout_error = self._timeout_error(intent.timeout_sec) + if timeout_error is not None: + return IntentDecision(intent.intent_id, "rejected", None, timeout_error) if intent.venue not in {"AUTO", "KRX", "NXT", "SOR"}: return IntentDecision(intent.intent_id, "rejected", None, "invalid_venue") @@ -334,6 +344,9 @@ class IntentService: return IntentDecision(intent.intent_id, "rejected", None, "already_active") def _admit_sell(self, intent: SellIntent) -> IntentDecision: + timeout_error = self._timeout_error(intent.timeout_sec) + if timeout_error is not None: + return IntentDecision(intent.intent_id, "rejected", None, timeout_error) if intent.venue not in {"AUTO", "KRX", "NXT", "SOR"}: return IntentDecision(intent.intent_id, "rejected", None, "invalid_venue") @@ -385,3 +398,13 @@ class IntentService: except SlotStateMismatchError: return IntentDecision(intent.intent_id, "rejected", None, "slot_state_changed") return IntentDecision(intent.intent_id, "admitted", updated.slot_id, None) + + @staticmethod + def _timeout_error(timeout_sec: object) -> Optional[str]: + if timeout_sec is None: + return None + if not isinstance(timeout_sec, int) or isinstance(timeout_sec, bool): + return "bad_request:timeout_sec_must_be_integer" + if not 1 <= timeout_sec <= _HARD_ORDER_TIMEOUT_SEC: + return "bad_request:timeout_sec_out_of_range" + return None diff --git a/trex_core/application/messaging_handlers.py b/trex_core/application/messaging_handlers.py index d0d7a8a..525bc78 100644 --- a/trex_core/application/messaging_handlers.py +++ b/trex_core/application/messaging_handlers.py @@ -608,7 +608,9 @@ def _opt_float(v) -> Optional[float]: def _opt_int(v) -> Optional[int]: if v is None: return None - return int(v) + if not isinstance(v, int) or isinstance(v, bool): + raise ValueError("integer required") + return v def _opt_dt(v) -> Optional[datetime]: diff --git a/trex_core/domain/slot/in_memory_repository.py b/trex_core/domain/slot/in_memory_repository.py index 7e4c5bf..dcf2b19 100644 --- a/trex_core/domain/slot/in_memory_repository.py +++ b/trex_core/domain/slot/in_memory_repository.py @@ -9,7 +9,7 @@ from __future__ import annotations import threading from contextlib import contextmanager from dataclasses import replace -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from typing import Iterator, Optional from uuid import UUID, uuid4 @@ -163,6 +163,7 @@ class InMemorySlotRepository: # --- order execution queue (BAT-004 / REQ-013, option A) --- _PENDING = (SlotState.PENDING_BUY, SlotState.PENDING_SELL) + _KST = timezone(timedelta(hours=9)) def list_undispatched_pending( self, @@ -191,15 +192,20 @@ class InMemorySlotRepository: broker_order_id: str, *, venue: Optional[str] = None, + submitted_at: Optional[datetime] = None, ) -> bool: - """Atomically persist dispatch identity and the actual broker venue.""" + """Atomically persist dispatch identity and its immutable clock.""" s = self._slots.get(slot_id) if s is None or s.broker_order_id is not None: return False + submitted_at = submitted_at or datetime.now(tz=timezone.utc) self._slots[slot_id] = replace( s, broker_order_id=broker_order_id, venue=venue if venue is not None else s.venue, + order_submitted_at=submitted_at, + order_date=submitted_at.astimezone(self._KST).strftime("%Y%m%d"), + cancel_attempts=0, updated_at=datetime.now(tz=timezone.utc), ) return True @@ -245,6 +251,39 @@ class InMemorySlotRepository: ) return True + def reserve_cancel_attempt( + self, + slot_id: UUID, + broker_order_id: str, + *, + retry_before: Optional[datetime] = None, + ) -> bool: + s = self._slots.get(slot_id) + retry_allowed = ( + retry_before is not None + and s is not None + and s.cancel_state in {"rejected", "unknown"} + and s.cancel_attempts < 4 + and s.cancel_requested_at is not None + and s.cancel_requested_at <= retry_before + ) + if ( + s is None + or s.broker_order_id != broker_order_id + or s.status not in self._PENDING + or (s.cancel_state is not None and not retry_allowed) + ): + return False + now = datetime.now(tz=timezone.utc) + self._slots[slot_id] = replace( + s, + cancel_state="unknown", + cancel_requested_at=now, + cancel_attempts=s.cancel_attempts + 1, + 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 [ @@ -266,6 +305,9 @@ class InMemorySlotRepository: broker_order_id=None, cancel_state=None, cancel_requested_at=None, + order_submitted_at=None, + order_date=None, + cancel_attempts=0, updated_at=datetime.now(tz=timezone.utc), ) diff --git a/trex_core/domain/slot/models.py b/trex_core/domain/slot/models.py index c1a7543..ffdb158 100644 --- a/trex_core/domain/slot/models.py +++ b/trex_core/domain/slot/models.py @@ -64,6 +64,11 @@ class Slot: # across ticks and restarts. cancel_state: Optional[str] = None cancel_requested_at: Optional[datetime] = None + # Immutable dispatch clock and original KST order date. Unlike updated_at, + # these survive polling/cancel bookkeeping and anchor deadline/recovery. + order_submitted_at: Optional[datetime] = None + order_date: Optional[str] = None + cancel_attempts: int = 0 # Canonical column contract for `Slot` (dcy-001 OQ-2 / FR-5). @@ -95,6 +100,9 @@ EXPECTED_SLOT_FIELDS: frozenset[str] = frozenset({ "venue", "cancel_state", "cancel_requested_at", + "order_submitted_at", + "order_date", + "cancel_attempts", }) diff --git a/trex_core/domain/slot/service.py b/trex_core/domain/slot/service.py index b184bc8..db3f892 100644 --- a/trex_core/domain/slot/service.py +++ b/trex_core/domain/slot/service.py @@ -213,6 +213,9 @@ class SlotService: venue="KRX", cancel_state=None, cancel_requested_at=None, + order_submitted_at=None, + order_date=None, + cancel_attempts=0, updated_at=datetime.now(tz=timezone.utc), ) @@ -287,6 +290,21 @@ class SlotService: venue=new_venue, cancel_state=None, cancel_requested_at=None, + order_submitted_at=( + slot.order_submitted_at + if to_state in {SlotState.PENDING_BUY, SlotState.PENDING_SELL} + else None + ), + order_date=( + slot.order_date + if to_state in {SlotState.PENDING_BUY, SlotState.PENDING_SELL} + else None + ), + cancel_attempts=( + slot.cancel_attempts + if to_state in {SlotState.PENDING_BUY, SlotState.PENDING_SELL} + else 0 + ), updated_at=datetime.now(tz=timezone.utc), ) diff --git a/trex_core/runtime/config.py b/trex_core/runtime/config.py index 04fe027..7d0ea43 100644 --- a/trex_core/runtime/config.py +++ b/trex_core/runtime/config.py @@ -170,9 +170,8 @@ class DaemonInstanceConfig: # admission capacity cap. BAT-030: asset-centric is the only model; absent # means no cap (operators are expected to set it in production). max_active_slots: Optional[int] = None - # dcy-022 / REQ-068: default order timeout in seconds (opt-in per instance). - # None = OFF = indefinite wait (legacy behavior). Per-intent timeout_sec overrides. - default_order_timeout_sec: Optional[int] = None + # Per-instance override; omission uses the hard 3,600-second fallback. + default_order_timeout_sec: int = 3600 # dcy-024 / REQ-048 CHG-20260611-01: disable autonomous safety-sell worker. # Absent = True (safe default). False → trex-safety-sell thread not started. safety_sell_enabled: bool = True @@ -202,11 +201,17 @@ class DaemonInstanceConfig: raise RuntimeError( f"instance TOML 'max_active_slots' must be > 0, got {raw_max}: {path}" ) - raw_timeout = data.get("default_order_timeout_sec") - default_timeout = int(raw_timeout) if raw_timeout is not None else None - if default_timeout is not None and default_timeout <= 0: + raw_timeout = data.get("default_order_timeout_sec", 3600) + if not isinstance(raw_timeout, int) or isinstance(raw_timeout, bool): raise RuntimeError( - f"instance TOML 'default_order_timeout_sec' must be > 0, got {raw_timeout}: {path}" + f"instance TOML 'default_order_timeout_sec' must be an integer in 1..3600, " + f"got {raw_timeout}: {path}" + ) + default_timeout = raw_timeout + if not 1 <= default_timeout <= 3600: + raise RuntimeError( + f"instance TOML 'default_order_timeout_sec' must be in 1..3600, " + f"got {raw_timeout}: {path}" ) raw_wd = data.get("watchdog_timeout_sec") watchdog_timeout = int(raw_wd) if raw_wd is not None else None @@ -258,9 +263,8 @@ class DaemonConfig: # threshold in seconds. 0 disables (default). When >0, dispatched-pending # slots older than this are surfaced (WARN + counter) without state change. pending_stale_threshold_sec: int = 0 - # dcy-022 / REQ-068: default order timeout (seconds). None = OFF = indefinite. - # Per-intent timeout_sec overrides. Passed to IntentService at wiring time. - default_order_timeout_sec: Optional[int] = None + # Effective fallback passed to IntentService; never disabled by omission. + default_order_timeout_sec: int = 3600 # dcy-014 / REQ-048: operator-local timezone (CLI date display etc.). The # daily preallocation trigger was removed with the pool model (BAT-030). trading_timezone: str = "Asia/Seoul" @@ -417,4 +421,5 @@ class DaemonConfig: safety_sell_enabled=inst.safety_sell_enabled, safety_sell=safety_sell, credential_fetch_enabled=inst.credential_fetch_enabled, + default_order_timeout_sec=inst.default_order_timeout_sec, )