diff --git a/docs/data-model.md b/docs/data-model.md index c41c25f..fcc5b33 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -49,7 +49,7 @@ - limit_price :: numeric? :: 현재 pending 주문 지정가(side 무관, dcy-012/REQ-002 — 과거 `entry_price`에서 rename). pending 진입(매수/매도 지정가) 시 set, pending 이탈(→holding/→closed) 시 None 으로 clear → 매수 지정가가 매도로 누수되지 않음(issue #10). None 이면 dispatch 시 시장가. - allocated_capital :: numeric? :: 배정 자본 - order_id / broker_order_id :: str? :: 주문 식별자(자유 주석 / 브로커 핸들) -- venue :: TEXT NOT NULL DEFAULT 'KRX' :: 현재 주문 거래소 구분(KRX/NXT/SOR). pending 주문·취소까지 보존하고 INACTIVE 진입 시 KRX 초기화. +- venue :: TEXT NOT NULL DEFAULT 'KRX' :: 현재 주문 거래소 구분(AUTO/KRX/NXT/SOR). intent 단계의 AUTO는 KIS 제출 시 실제 KRX/NXT로 교체되며 주문번호와 원자 저장한다. pending 주문·취소까지 보존하고 HOLDING 진입 시 다음 자율주문의 AUTO, INACTIVE 진입 시 KRX로 초기화한다. - strategy_id / sell_strategy_id :: str? :: 진입/청산 전략 식별 - portfolio_id :: str? :: 포트폴리오 식별 - forced_close_at :: datetime? :: 강제 청산 마감선 @@ -128,6 +128,7 @@ - fee :: float :: BrokerPort `calculate_fee(side, qty*price)` - tax :: float :: 기본 0(정밀 모델 후속) - executed_at :: datetime? +- venue :: str :: Core가 실제 제출에 사용한 거래소. 외부 Portfolio Hub API에는 아직 전달하지 않는다. - correlation_id (derived) :: `{asset_id}:{order_id}:{execution_id}` ### State Rules @@ -147,6 +148,7 @@ - side :: TEXT CHECK ('buy','sell') - qty / price / fee / tax :: NUMERIC(18,4) - executed_at :: TIMESTAMPTZ? +- venue :: TEXT NOT NULL DEFAULT 'KRX' :: 실제 제출 거래소. 기존 행은 KRX로 보존한다. - status :: TEXT CHECK ('pending','completed','conflict_blocked','delivery_failed') — `delivery_failed` 추가(dcy-015, 마이그레이션 007) - attempts :: INTEGER :: 재시도 누적 - last_outcome :: TEXT? CHECK ('completed','duplicate_completed','retryable_failure','conflict_blocked','permanent_failure','dead_lettered') diff --git a/docs/interface-contract.md b/docs/interface-contract.md index b4499ab..a36f566 100644 --- a/docs/interface-contract.md +++ b/docs/interface-contract.md @@ -396,7 +396,7 @@ resume_kill_switch 요청·응답 (REQ-046): ### Persistence -- PostgreSQL: `ledger_outbox(correlation_id PK, asset_id, order_id, execution_id, side, qty, price, fee, tax, executed_at, status, attempts, last_outcome, last_error, created_at, updated_at)` — `ON CONFLICT (correlation_id) DO NOTHING` 이 outbox dedup 게이트. +- PostgreSQL: `ledger_outbox(correlation_id PK, asset_id, order_id, execution_id, side, qty, price, fee, tax, executed_at, venue, status, attempts, last_outcome, last_error, created_at, updated_at)` — `ON CONFLICT (correlation_id) DO NOTHING` 이 outbox dedup 게이트. `venue`는 Core 내부 실제 제출 거래소이며 Portfolio Hub wire에는 아직 보내지 않는다. ## `OrderRequest` DTO (BAT-004 / REQ-011, BAT-056 / REQ-117) @@ -415,8 +415,9 @@ resume_kill_switch 요청·응답 (REQ-046): ### Additive Notes (REQ-117) - `venue` 필드는 BAT-056에서 additive 추가. 기존 orb/tsl 미지정 호출 → 기본값 `"KRX"` 자동 적용 → 하위 호환. -- `OrderRequest`는 AMQP wire 위를 직접 흐르지 않는다. BAT-059부터 buy/sell intent의 선택적 `venue`(기본 KRX)를 슬롯에 보존해 이 DTO까지 전달한다. 허용값 외에는 `invalid_venue`로 admission 전 거부한다. -- KIS 어댑터(REQ-118)는 `venue` → `EXCG_ID_DVSN_CD` 로 변환해 KIS HTTP 요청에 항상 명시 전송. +- `OrderRequest`는 AMQP wire 위를 직접 흐르지 않는다. buy/sell intent의 선택적 `venue` 기본은 `AUTO`이고, 명시한 KRX/NXT/SOR는 그대로 슬롯에 보존한다. 허용값 외에는 `invalid_venue`로 admission 전 거부한다. +- KIS 어댑터는 `AUTO`를 KST 세션 기준 실제 KRX/NXT로 확정한 뒤 `EXCG_ID_DVSN_CD`로 전송한다. 결정 불가능 구간은 KRX로 대체하지 않고 주문을 거부한다. +- `OrderAck.venue`의 실제 제출 거래소는 `broker_order_id`와 함께 슬롯에 원자 저장되며 재시작·취소·Core 원장 outbox에서 재사용한다. ## `BrokerPort` Protocol 메서드 (BAT-056 / REQ-118) @@ -446,6 +447,7 @@ get_current_price(symbol: str, market: str = "J") -> Optional[Decimal] ### KIS 어댑터 EXCG 명시화 (REQ-118) - `submit_order`: body에 `EXCG_ID_DVSN_CD: req.venue` 항상 명시 전송 (이전: 미전송·서버 기본값 의존 → 명시화로 견고성 향상). 실측(2026-07-06): NXT(EXCG=NXT) 주문 rt_cd=0 확인. +- `AUTO` 세션: 08:00~08:49:59 NXT, 09:00:30~15:29:59 KRX, 15:30~20:00 NXT. 그 밖의 시간은 `auto_venue_session_closed`로 fail-closed 한다. - `cancel_order`: body에 `EXCG_ID_DVSN_CD: venue` 명시 전송. 실측(2026-07-06): NXT 취소 rt_cd=0 확인. - `poll_fill`: 구 TR(`TTTC8001R`/`VTTC8001R`) 유지. NXT 체결 포함 여부 미확인(OPEN ITEM — 운영 스모크 필요). 확인 후 미포함이면 `TTTC0081R`(EXCG=ALL) 전환 별도 batch 처리. diff --git a/tests/adapters/broker/test_venue.py b/tests/adapters/broker/test_venue.py index 8dfdd61..411cb26 100644 --- a/tests/adapters/broker/test_venue.py +++ b/tests/adapters/broker/test_venue.py @@ -26,13 +26,12 @@ 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 +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 BrokerPort, OrderRequest +from trex_core.application.broker_port import OrderRequest _MOCK_URL = "https://openapivts.koreainvestment.com:29443" _LIVE_URL = "https://openapi.koreainvestment.com:9443" @@ -42,8 +41,14 @@ _LIVE_URL = "https://openapi.koreainvestment.com:9443" # Helpers # --------------------------------------------------------------------------- -def _mock_broker() -> KISBroker: - b = KISBroker("12345678-01", app_key="k", app_secret="s", base_url=_MOCK_URL) +def _mock_broker(*, clock=None) -> KISBroker: + b = KISBroker( + "12345678-01", + app_key="k", + app_secret="s", + base_url=_MOCK_URL, + clock=clock, + ) b._token = "tok" b._token_exp = datetime.now(tz=timezone.utc) + timedelta(hours=1) return b @@ -85,6 +90,84 @@ def test_order_request_venue_nxt(): assert req.venue == "NXT" +@pytest.mark.parametrize( + ("at", "expected"), + [ + (datetime(2026, 7, 29, 8, 0, tzinfo=timezone(timedelta(hours=9))), "NXT"), + (datetime(2026, 7, 29, 8, 49, 59, tzinfo=timezone(timedelta(hours=9))), "NXT"), + (datetime(2026, 7, 29, 9, 0, 30, tzinfo=timezone(timedelta(hours=9))), "KRX"), + (datetime(2026, 7, 29, 15, 29, 59, tzinfo=timezone(timedelta(hours=9))), "KRX"), + (datetime(2026, 7, 29, 15, 30, tzinfo=timezone(timedelta(hours=9))), "NXT"), + (datetime(2026, 7, 29, 20, 0, tzinfo=timezone(timedelta(hours=9))), "NXT"), + ], +) +def test_auto_venue_resolves_by_kst_session(at, expected): + assert resolve_order_venue("AUTO", at) == expected + + +@pytest.mark.parametrize( + "at", + [ + datetime(2026, 7, 29, 7, 59, 59, tzinfo=timezone(timedelta(hours=9))), + datetime(2026, 7, 29, 8, 50, tzinfo=timezone(timedelta(hours=9))), + datetime(2026, 7, 29, 9, 0, 29, tzinfo=timezone(timedelta(hours=9))), + datetime(2026, 7, 29, 20, 0, 1, tzinfo=timezone(timedelta(hours=9))), + ], +) +def test_auto_venue_fails_closed_outside_supported_sessions(at): + with pytest.raises(ValueError, match="auto_venue_session_closed"): + resolve_order_venue("AUTO", at) + + +def test_submit_auto_uses_nxt_and_reports_actual_venue(monkeypatch): + captured = {} + at = datetime(2026, 7, 29, 8, 1, tzinfo=timezone(timedelta(hours=9))) + + def fake_post(url, headers=None, json=None, timeout=None): + captured["body"] = json + return _FakeResp( + {"rt_cd": "0", "output": {"ODNO": "0002", "KRX_FWDG_ORD_ORGNO": "06010"}} + ) + + broker = _mock_broker(clock=lambda: at) + monkeypatch.setattr(kis_mod.httpx, "post", fake_post) + ack = broker.submit_order( + OrderRequest( + slot_id="s-auto", + side="buy", + symbol="005930", + qty=1, + venue="AUTO", + ) + ) + + assert ack.accepted is True + assert ack.venue == "NXT" + assert captured["body"]["EXCG_ID_DVSN_CD"] == "NXT" + + +def test_submit_auto_outside_session_rejects_without_http(monkeypatch): + at = datetime(2026, 7, 29, 8, 55, tzinfo=timezone(timedelta(hours=9))) + broker = _mock_broker(clock=lambda: at) + + def unexpected_post(*args, **kwargs): + raise AssertionError("AUTO session rejection must happen before HTTP") + + monkeypatch.setattr(kis_mod.httpx, "post", unexpected_post) + ack = broker.submit_order( + OrderRequest( + slot_id="s-auto-closed", + side="buy", + symbol="005930", + qty=1, + venue="AUTO", + ) + ) + + assert ack.accepted is False + assert ack.reason == "auto_venue_session_closed" + + # --------------------------------------------------------------------------- # REQ-117 AC-4: 기존 위치 인수 순서 불변 # --------------------------------------------------------------------------- diff --git a/tests/adapters/persistence/test_postgresql_ledger_outbox.py b/tests/adapters/persistence/test_postgresql_ledger_outbox.py index 4836d08..f2bff2e 100644 --- a/tests/adapters/persistence/test_postgresql_ledger_outbox.py +++ b/tests/adapters/persistence/test_postgresql_ledger_outbox.py @@ -20,7 +20,7 @@ def _trade(execution_id="paper:1:10") -> LedgerTrade: return LedgerTrade( asset_id="PF-1", order_id="ORD-1", execution_id=execution_id, side="buy", qty=10, price=1000.0, fee=1.5, tax=0.0, - executed_at=datetime(2026, 6, 4, tzinfo=timezone.utc), + executed_at=datetime(2026, 6, 4, tzinfo=timezone.utc), venue="NXT", ) @@ -32,6 +32,7 @@ def test_enqueue_persists_and_is_idempotent(pg_clean): pending = outbox.list_pending() assert len(pending) == 1 assert pending[0].status is WorkItemStatus.PENDING + assert pending[0].trade.venue == "NXT" def test_writer_completes_persisted_item(pg_clean): diff --git a/tests/application/test_execution_ledger.py b/tests/application/test_execution_ledger.py index 6f4d904..476aff4 100644 --- a/tests/application/test_execution_ledger.py +++ b/tests/application/test_execution_ledger.py @@ -59,6 +59,7 @@ def test_buy_fill_enqueues_one_ledger_item_and_writer_records_it(): assert (t.asset_id, t.side, t.qty, t.price) == ("PF-1", "buy", 10, 1000.0) assert t.execution_id.startswith("paper:PAPER-") assert t.fee > 0.0 and t.tax == 0.0 + assert t.venue == "AUTO" def test_sell_fill_enqueues_sell_ledger_item(): diff --git a/tests/application/test_execution_policy.py b/tests/application/test_execution_policy.py index 4c26f86..adcbe0b 100644 --- a/tests/application/test_execution_policy.py +++ b/tests/application/test_execution_policy.py @@ -43,8 +43,8 @@ class _RecordingBroker: def poll_fill(self, broker_order_id, *, side, symbol): return self.responder(side) - def cancel_order(self, broker_order_id, *, side, symbol, qty): - self.cancel_calls.append((broker_order_id, side, symbol, qty)) + def cancel_order(self, broker_order_id, *, side, symbol, qty, venue="KRX"): + self.cancel_calls.append((broker_order_id, side, symbol, qty, venue)) if self.cancel_raises: raise RuntimeError("broker error") return True diff --git a/tests/application/test_venue_intent_flow.py b/tests/application/test_venue_intent_flow.py index d70fed6..19a50d2 100644 --- a/tests/application/test_venue_intent_flow.py +++ b/tests/application/test_venue_intent_flow.py @@ -1,10 +1,12 @@ """Venue propagation from the strategy wire contract to broker execution.""" 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.execution import ExecutionService -from trex_core.application.intents import IntentService +from trex_core.application.intents import BuyIntent, IntentService 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 @@ -12,13 +14,18 @@ from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository class RecordingBroker: - def __init__(self) -> None: + def __init__(self, *, actual_venue: str | None = None) -> None: self.orders: list[OrderRequest] = [] self.cancel_venues: list[str] = [] + self.actual_venue = actual_venue def submit_order(self, req: OrderRequest) -> OrderAck: self.orders.append(req) - return OrderAck(accepted=True, broker_order_id=f"order-{len(self.orders)}") + return OrderAck( + accepted=True, + broker_order_id=f"order-{len(self.orders)}", + venue=self.actual_venue or req.venue, + ) def poll_fill(self, broker_order_id: str, *, side: str, symbol: str) -> FillStatus: return FillStatus(state="accepted") @@ -104,6 +111,41 @@ def test_invalid_venue_is_rejected_before_slot_transition() -> None: assert repo.list_slots("default")[0].status is SlotState.INACTIVE +def test_omitted_wire_venue_is_auto_then_actual_venue_is_persisted() -> None: + repo = InMemorySlotRepository() + service = SlotService(repo) + intents = IntentService( + service, + repo, + InMemoryIntentStore(), + resolve_symbol=lambda portfolio_id: "005930", + ) + repo.reconcile_universe("default", {"PF-1": "005930"}) + broker = RecordingBroker(actual_venue="NXT") + execution = ExecutionService(service, repo, broker) + handler = make_intent_rpc_handler(intents) + + response = handler( + RpcRequest( + payload={ + "intent_type": "buy", + "intent_id": "auto-buy", + "portfolio_id": "PF-1", + "qty": 1, + }, + correlation_id="auto-buy", + ) + ) + + slot_id = UUID(response.payload["slot_id"]) + pending = repo.fetch_slot(slot_id) + assert pending is not None and pending.venue == "AUTO" + execution.run_once() + dispatched = repo.fetch_slot(slot_id) + assert broker.orders[0].venue == "AUTO" + assert dispatched is not None and dispatched.venue == "NXT" + + def test_inactive_transition_resets_venue_to_krx() -> None: repo, intents, _, _ = _wire() handler = make_intent_rpc_handler(intents) @@ -131,6 +173,30 @@ def test_inactive_transition_resets_venue_to_krx() -> None: assert reset.venue == "KRX" +def test_holding_transition_resets_next_order_venue_to_auto() -> None: + repo, intents, _, _ = _wire() + decision = intents.submit_buy( + BuyIntent( + intent_id="nxt-buy-holding", + portfolio_id="PF-1", + qty=1, + venue="NXT", + ) + ) + slot = repo.fetch_slot(decision.slot_id) + service = SlotService(repo) + holding = service.transit_slot( + slot.slot_id, + SlotState.PENDING_BUY, + SlotState.HOLDING, + reason="fill", + actor="test", + qty=1, + avg_price=1000, + ) + assert holding.venue == "AUTO" + + def test_operator_cancel_uses_original_nxt_venue() -> None: repo, intents, execution, broker = _wire() handler = make_intent_rpc_handler(intents) diff --git a/trex_core/adapters/broker/kis.py b/trex_core/adapters/broker/kis.py index e251184..f78bce3 100644 --- a/trex_core/adapters/broker/kis.py +++ b/trex_core/adapters/broker/kis.py @@ -14,10 +14,10 @@ from __future__ import annotations import base64 import json from dataclasses import replace -from datetime import datetime, timedelta, timezone +from datetime import datetime, time, timedelta, timezone from decimal import Decimal, InvalidOperation from time import monotonic -from typing import Optional +from typing import Callable, Optional import httpx @@ -37,6 +37,30 @@ _MOCK_HINTS = ("openapivts", "virtual", "mock") _FILL_POLL_MIN_INTERVAL_SECONDS = 5.0 _FILL_POLL_TRANSPORT_BACKOFF_SECONDS = 30.0 _SUBMISSION_LOOKUP_GRACE_SECONDS = 90.0 +_SUPPORTED_ORDER_VENUES = frozenset({"AUTO", "KRX", "NXT", "SOR"}) + + +def resolve_order_venue(venue: str, at: datetime) -> str: + """Resolve AUTO to the venue that accepts domestic-stock orders at `at`. + + The gaps are intentional fail-closed intervals. They prevent an AUTO order + from silently falling back to KRX while the opening venue is indeterminate. + Explicit KRX/NXT/SOR values are preserved. + """ + normalized = str(venue).strip().upper() + if normalized not in _SUPPORTED_ORDER_VENUES: + raise ValueError("unsupported_order_venue") + if normalized != "AUTO": + return normalized + + current = at.astimezone(_KST).time().replace(tzinfo=None) + if time(8, 0) <= current < time(8, 50): + return "NXT" + if time(9, 0, 30) <= current < time(15, 30): + return "KRX" + if time(15, 30) <= current <= time(20, 0): + return "NXT" + raise ValueError("auto_venue_session_closed") def _safe_int(v: object) -> Optional[int]: @@ -108,6 +132,7 @@ class KISBroker: timeout: float = 5.0, allow_live: bool = False, token_cache=None, + clock: Optional[Callable[[], datetime]] = None, ) -> None: self._account_ref = account_ref self._app_key = app_key @@ -117,6 +142,7 @@ class KISBroker: self._token: Optional[str] = None self._token_exp: datetime = datetime.now(tz=timezone.utc) self._token_cache = token_cache # optional TokenCache (L2, REQ-016) + self._clock = clock or (lambda: datetime.now(tz=_KST)) self._fill_poll_not_before: dict[str, float] = {} is_mock = any(h in self._base_url for h in _MOCK_HINTS) if not is_mock and not allow_live: @@ -226,23 +252,28 @@ class KISBroker: # --- BrokerPort --- def submit_order(self, req: OrderRequest) -> OrderAck: cano, product = self._account() - submitted_at = datetime.now(tz=_KST) + submitted_at = self._clock().astimezone(_KST) + try: + actual_venue = resolve_order_venue(req.venue, submitted_at) + except ValueError as exc: + return OrderAck(accepted=False, reason=str(exc)) + resolved_req = replace(req, venue=actual_venue) body = { "CANO": cano, "ACNT_PRDT_CD": product, - "PDNO": req.symbol, - "ORD_DVSN": req.order_type, - "ORD_QTY": str(req.qty), - "ORD_UNPR": str(req.limit_price or 0), + "PDNO": resolved_req.symbol, + "ORD_DVSN": resolved_req.order_type, + "ORD_QTY": str(resolved_req.qty), + "ORD_UNPR": str(resolved_req.limit_price or 0), # REQ-118: always send EXCG_ID_DVSN_CD from req.venue (default "KRX"). # Previously omitted (server defaulted to KRX); explicit is more robust. # Real-account NXT order confirmed rt_cd=0 on 2026-07-06. - "EXCG_ID_DVSN_CD": req.venue, + "EXCG_ID_DVSN_CD": actual_venue, } try: resp = httpx.post( f"{self._base_url}/uapi/domestic-stock/v1/trading/order-cash", - headers=self._headers(self._tr_submit(req.side)), + headers=self._headers(self._tr_submit(resolved_req.side)), json=body, timeout=self._timeout, ) @@ -253,15 +284,17 @@ class KISBroker: # the execution loop cannot submit the same order again. return OrderAck( accepted=True, - broker_order_id=_encode_submission_unknown(req, submitted_at), + broker_order_id=_encode_submission_unknown(resolved_req, submitted_at), reason=f"submit_outcome_unknown:{type(exc).__name__}", + venue=actual_venue, raw={"error_class": type(exc).__name__}, ) if getattr(resp, "status_code", 200) >= 500: return OrderAck( accepted=True, - broker_order_id=_encode_submission_unknown(req, submitted_at), + broker_order_id=_encode_submission_unknown(resolved_req, submitted_at), reason=f"submit_outcome_unknown:http_{resp.status_code}", + venue=actual_venue, raw={"http_status": resp.status_code}, ) if str(payload.get("rt_cd")) == "0": @@ -269,9 +302,17 @@ class KISBroker: odno = out.get("ODNO") or out.get("odno") orgno = out.get("KRX_FWDG_ORD_ORGNO") or out.get("ord_gno_brno") or "" bid = f"{orgno}:{odno}" if orgno else str(odno) - return OrderAck(accepted=True, broker_order_id=bid, raw=payload) + return OrderAck( + accepted=True, + broker_order_id=bid, + venue=actual_venue, + raw=payload, + ) return OrderAck( - accepted=False, reason=payload.get("msg1") or payload.get("msg_cd"), raw=payload + accepted=False, + reason=payload.get("msg1") or payload.get("msg_cd"), + venue=actual_venue, + raw=payload, ) def poll_fill(self, broker_order_id: str, *, side: str, symbol: str) -> FillStatus: diff --git a/trex_core/adapters/broker/paper.py b/trex_core/adapters/broker/paper.py index 5afb701..017409b 100644 --- a/trex_core/adapters/broker/paper.py +++ b/trex_core/adapters/broker/paper.py @@ -76,7 +76,7 @@ class PaperBroker: seq = self._seq oid = f"PAPER-{seq:06d}" self._orders[oid] = _Order(oid, req) - return OrderAck(accepted=True, broker_order_id=oid) + return OrderAck(accepted=True, broker_order_id=oid, venue=req.venue) def poll_fill(self, broker_order_id: str, *, side: str, symbol: str) -> FillStatus: with self._lock: diff --git a/trex_core/adapters/persistence/postgresql.py b/trex_core/adapters/persistence/postgresql.py index 1016bc2..36da514 100644 --- a/trex_core/adapters/persistence/postgresql.py +++ b/trex_core/adapters/persistence/postgresql.py @@ -285,13 +285,25 @@ class PostgreSQLSlotRepository: ) return [self._row_to_slot(r) for r in cur.fetchall()] - def mark_dispatched(self, slot_id: UUID, broker_order_id: str) -> bool: + def mark_dispatched( + self, + slot_id: UUID, + broker_order_id: str, + *, + venue: Optional[str] = None, + ) -> bool: 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, updated_at=%s " + "UPDATE daily_slot SET broker_order_id=%s, " + "venue=COALESCE(%s, venue), updated_at=%s " "WHERE slot_id=%s AND broker_order_id IS NULL", - (broker_order_id, datetime.now(tz=timezone.utc), slot_id), + ( + broker_order_id, + venue, + datetime.now(tz=timezone.utc), + slot_id, + ), ) ok = cur.rowcount == 1 conn.commit() @@ -456,7 +468,7 @@ class _PgIntentRecord: _OUTBOX_COLUMNS = ( "correlation_id, asset_id, order_id, execution_id, side, qty, price, " "fee, tax, executed_at, status, attempts, last_outcome, last_error, " - "created_at, updated_at, next_retry_at" + "created_at, updated_at, next_retry_at, venue" ) @@ -473,8 +485,8 @@ class PostgreSQLLedgerOutbox: cur.execute( "INSERT INTO ledger_outbox " "(correlation_id, asset_id, order_id, execution_id, side, " - " qty, price, fee, tax, executed_at) " - "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) " + " qty, price, fee, tax, executed_at, venue) " + "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) " "ON CONFLICT (correlation_id) DO NOTHING", ( trade.correlation_id, @@ -487,6 +499,7 @@ class PostgreSQLLedgerOutbox: trade.fee, trade.tax, trade.executed_at, + trade.venue, ), ) conn.commit() @@ -590,6 +603,7 @@ class PostgreSQLLedgerOutbox: fee=f(row[7]), tax=f(row[8]), executed_at=row[9], + venue=row[17], ) return LedgerWorkItem( correlation_id=row[0], diff --git a/trex_core/adapters/persistence/sql/019_preserve_actual_order_venue.sql b/trex_core/adapters/persistence/sql/019_preserve_actual_order_venue.sql new file mode 100644 index 0000000..54b4610 --- /dev/null +++ b/trex_core/adapters/persistence/sql/019_preserve_actual_order_venue.sql @@ -0,0 +1,20 @@ +-- dcy-041 / GitHub #44: AUTO may exist while an intent is pending, but the +-- actual submitted KRX/NXT value replaces it atomically with broker_order_id. +ALTER TABLE daily_slot + DROP CONSTRAINT IF EXISTS daily_slot_venue_check; + +ALTER TABLE daily_slot + ADD CONSTRAINT daily_slot_venue_check + CHECK (venue IN ('AUTO', 'KRX', 'NXT', 'SOR')); + +-- Core keeps the actual venue in its durable ledger outbox. Portfolio Hub's +-- HTTP schema is intentionally unchanged in this migration. +ALTER TABLE ledger_outbox + ADD COLUMN IF NOT EXISTS venue TEXT NOT NULL DEFAULT 'KRX'; + +ALTER TABLE ledger_outbox + DROP CONSTRAINT IF EXISTS ledger_outbox_venue_check; + +ALTER TABLE ledger_outbox + ADD CONSTRAINT ledger_outbox_venue_check + CHECK (venue IN ('AUTO', 'KRX', 'NXT', 'SOR')); diff --git a/trex_core/application/broker_port.py b/trex_core/application/broker_port.py index 7da3c1f..e9bbc73 100644 --- a/trex_core/application/broker_port.py +++ b/trex_core/application/broker_port.py @@ -45,6 +45,10 @@ class OrderAck: accepted: bool broker_order_id: Optional[str] = None reason: Optional[str] = None + # Actual venue used for submission. KIS resolves AUTO before the HTTP call; + # execution persists this value with the broker order id so restart/cancel + # never has to infer the original venue. + venue: Optional[str] = None raw: dict[str, Any] = field(default_factory=dict) diff --git a/trex_core/application/execution.py b/trex_core/application/execution.py index 0987cee..2a8d874 100644 --- a/trex_core/application/execution.py +++ b/trex_core/application/execution.py @@ -201,7 +201,11 @@ class ExecutionService: ) ) if ack.accepted and ack.broker_order_id: - self._repo.mark_dispatched(slot.slot_id, ack.broker_order_id) + self._repo.mark_dispatched( + slot.slot_id, + ack.broker_order_id, + venue=ack.venue, + ) r.dispatched += 1 if ack.reason and ack.reason.startswith("submit_outcome_unknown:"): _log.warning( @@ -643,6 +647,7 @@ class ExecutionService: fee=float(fs.fee), tax=float(fs.tax), executed_at=datetime.now(tz=timezone.utc), + venue=slot.venue, ) self._ledger_outbox.enqueue(trade) r.ledger_enqueued += 1 diff --git a/trex_core/application/intents.py b/trex_core/application/intents.py index 50d9b99..57f6d9d 100644 --- a/trex_core/application/intents.py +++ b/trex_core/application/intents.py @@ -75,7 +75,9 @@ class BuyIntent: # slippage_pct: effective limit = entry_price*(1+pct/100) buy / *(1-pct/100) sell. timeout_sec: Optional[int] = None slippage_pct: Optional[float] = None - venue: str = "KRX" + # Strategy wire default. AUTO is resolved at the broker boundary so + # strategies remain session-agnostic while explicit KRX/NXT/SOR is kept. + venue: str = "AUTO" extra: dict[str, Any] = field(default_factory=dict) @@ -94,7 +96,7 @@ class SellIntent: # dcy-022 / REQ-068: execution policy (optional, backward-compat). timeout_sec: Optional[int] = None slippage_pct: Optional[float] = None - venue: str = "KRX" + venue: str = "AUTO" extra: dict[str, Any] = field(default_factory=dict) @@ -211,7 +213,7 @@ class IntentService: dcy-036 / REQ-114: entry gate — if entry_allowed_fn returns False, BuyIntent is rejected before any slot lookup.""" - if intent.venue not in {"KRX", "NXT", "SOR"}: + if intent.venue not in {"AUTO", "KRX", "NXT", "SOR"}: return IntentDecision(intent.intent_id, "rejected", None, "invalid_venue") # dcy-036 / REQ-114: day-start entry gate (fail-closed). @@ -308,7 +310,7 @@ class IntentService: return IntentDecision(intent.intent_id, "rejected", None, "already_active") def _admit_sell(self, intent: SellIntent) -> IntentDecision: - if intent.venue not in {"KRX", "NXT", "SOR"}: + if intent.venue not in {"AUTO", "KRX", "NXT", "SOR"}: return IntentDecision(intent.intent_id, "rejected", None, "invalid_venue") slot = self._repo.fetch_slot(intent.slot_id) diff --git a/trex_core/application/ledger_port.py b/trex_core/application/ledger_port.py index 56dcf8e..ee37273 100644 --- a/trex_core/application/ledger_port.py +++ b/trex_core/application/ledger_port.py @@ -57,6 +57,9 @@ class LedgerTrade: fee: float = 0.0 tax: float = 0.0 executed_at: Optional[datetime] = None + # Actual order venue persisted in the Core outbox. Portfolio Hub remains + # unchanged until its API contract explicitly accepts this additive field. + venue: str = "KRX" @property def correlation_id(self) -> str: diff --git a/trex_core/application/messaging_handlers.py b/trex_core/application/messaging_handlers.py index 2a4f3e6..d0d7a8a 100644 --- a/trex_core/application/messaging_handlers.py +++ b/trex_core/application/messaging_handlers.py @@ -117,7 +117,7 @@ def make_intent_rpc_handler( # dcy-022 / REQ-068: execution policy (optional, backward-compat). timeout_sec=_opt_int(body.get("timeout_sec")), slippage_pct=_opt_float(body.get("slippage_pct")), - venue=_opt_str(body.get("venue")) or "KRX", + venue=_opt_str(body.get("venue")) or "AUTO", ) d = intent_service.submit_buy(buy) elif intent_type == "sell": @@ -133,7 +133,7 @@ def make_intent_rpc_handler( # dcy-022 / REQ-068: execution policy (optional, backward-compat). timeout_sec=_opt_int(body.get("timeout_sec")), slippage_pct=_opt_float(body.get("slippage_pct")), - venue=_opt_str(body.get("venue")) or "KRX", + venue=_opt_str(body.get("venue")) or "AUTO", ) d = intent_service.submit_sell(sell) else: @@ -627,4 +627,3 @@ def _parse_date(v) -> date_cls: def _today() -> date_cls: return datetime.now(tz=timezone.utc).date() - diff --git a/trex_core/domain/slot/in_memory_repository.py b/trex_core/domain/slot/in_memory_repository.py index 494d5fa..f4420f8 100644 --- a/trex_core/domain/slot/in_memory_repository.py +++ b/trex_core/domain/slot/in_memory_repository.py @@ -185,13 +185,22 @@ class InMemorySlotRepository: ] return out[:limit] - def mark_dispatched(self, slot_id: UUID, broker_order_id: str) -> bool: - """Atomically set broker_order_id iff still undispatched (idempotency guard).""" + def mark_dispatched( + self, + slot_id: UUID, + broker_order_id: str, + *, + venue: Optional[str] = None, + ) -> bool: + """Atomically persist dispatch identity and the actual broker venue.""" s = self._slots.get(slot_id) if s is None or s.broker_order_id is not None: return False self._slots[slot_id] = replace( - s, broker_order_id=broker_order_id, updated_at=datetime.now(tz=timezone.utc) + s, + broker_order_id=broker_order_id, + venue=venue if venue is not None else s.venue, + updated_at=datetime.now(tz=timezone.utc), ) return True diff --git a/trex_core/domain/slot/service.py b/trex_core/domain/slot/service.py index 80ca97d..6767d14 100644 --- a/trex_core/domain/slot/service.py +++ b/trex_core/domain/slot/service.py @@ -215,10 +215,15 @@ class SlotService: new_limit_price = limit_price if limit_price is not None else slot.limit_price new_timeout_sec = timeout_sec if timeout_sec is not None else slot.timeout_sec new_slippage_pct = slippage_pct if slippage_pct is not None else slot.slippage_pct + new_venue = venue if venue is not None else slot.venue else: # HOLDING / ERROR — pending order policy no longer applies. new_limit_price = None new_timeout_sec = None new_slippage_pct = None + # A completed order's venue must not leak into a later autonomous + # sell. The next order starts session-aware unless its intent + # explicitly chooses KRX/NXT/SOR. + new_venue = "AUTO" if to_state is SlotState.HOLDING else slot.venue # occupancy_seq (dcy-028/REQ-081): bump when a slot begins a new # occupancy (inactive→pending_buy). Stable across the rest of the @@ -271,7 +276,7 @@ class SlotService: timeout_sec=new_timeout_sec, slippage_pct=new_slippage_pct, occupancy_seq=new_occupancy_seq, - venue=venue if venue is not None else slot.venue, + venue=new_venue, updated_at=datetime.now(tz=timezone.utc), )