Deliver BAT-004: broker execution + KIS adapter + durable handoff

Implement REQ-011~015 (dcy-003): the order-execution layer on top of the
BAT-002 slot-core daemon, connecting intents to real orders.
- application: BrokerPort (submit/poll_fill/cancel) + ExecutionService
  (claim -> submit -> poll -> transition -> rollback -> reconcile)
- adapters/broker: PaperBroker (deterministic async double) and KISBroker
  (KIS Open API: token, order-cash, inquire-daily-ccld, cancel; VTS sandbox
  tr_ids; live account gated behind allow_live)
- persistence: option-A execution queue (find_pending_for_dispatch FOR
  UPDATE SKIP LOCKED, mark_dispatched idempotency guard, clear_dispatch)
- runtime: background execution loop in the daemon lifespan + startup
  reconcile; config selects paper|kis

Verification (real KIS VTS 모의투자 + PostgreSQL): 92 passed, ruff clean,
domain import sweep shows zero broker/web/db deps, buy->holding /
sell->closed / reject->idle, idempotent dispatch (exactly one order),
reconcile after crash creates no duplicate order, KIS sandbox token+submit
round-trip, and a daemon E2E through the background loop. Align the four
baseline docs. BAT-004 -> release-candidate; REQ-011~015 -> Implemented.

KIS live-account go-live, partial-fill accrual, option-B outbox, connection
pooling, and the safety floor remain deferred to follow-up Discoveries.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

trex_core/adapters/broker/__init__.py

@@ -0,0 +1 @@
+"""Broker adapters (REQ-011/012). Implement application.broker_port.BrokerPort."""

trex_core/adapters/broker/kis.py

@@ -0,0 +1,189 @@
+"""KIS (한국투자증권) Open API broker adapter (REQ-012, dcy-003).
+
+Implements BrokerPort against the KIS REST API, modeled on the trex-orbit
+reference (app/broker/kis.py). Slice 1 validates against the KIS 모의투자
+(VTS sandbox) server; real-account go-live is config-gated and out of scope.
+
+Endpoints: /oauth2/tokenP (token), /uapi/domestic-stock/v1/trading/order-cash
+(submit), /uapi/domestic-stock/v1/trading/inquire-daily-ccld (fill poll),
+/uapi/domestic-stock/v1/trading/order-rvsecncl (cancel).
+"""
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from typing import Optional
+
+import httpx
+
+from trex_core.application.broker_port import FillStatus, OrderAck, OrderRequest
+
+_KST = timezone(timedelta(hours=9))
+_MOCK_HINTS = ("openapivts", "virtual", "mock")
+
+
+def _safe_int(v: object) -> Optional[int]:
+    try:
+        return int(str(v).strip())
+    except (TypeError, ValueError):
+        return None
+
+
+class KISBroker:
+    """KIS REST broker adapter. Live account is gated behind `allow_live`."""
+
+    def __init__(
+        self,
+        account_ref: str,
+        *,
+        app_key: str,
+        app_secret: str,
+        base_url: str,
+        timeout: float = 5.0,
+        allow_live: bool = False,
+    ) -> None:
+        self._account_ref = account_ref
+        self._app_key = app_key
+        self._app_secret = app_secret
+        self._base_url = base_url.rstrip("/")
+        self._timeout = timeout
+        self._token: Optional[str] = None
+        self._token_exp: datetime = datetime.now(tz=timezone.utc)
+        is_mock = any(h in self._base_url for h in _MOCK_HINTS)
+        if not is_mock and not allow_live:
+            raise RuntimeError(
+                "KIS live account base_url requires allow_live=True (config-gated). "
+                "Use the 모의투자 (openapivts) endpoint for slice-1."
+            )
+        self._mock = is_mock
+
+    def name(self) -> str:
+        return "kis-mock" if self._mock else "kis-live"
+
+    # --- auth ---
+    def _ensure_token(self) -> str:
+        if self._token and datetime.now(tz=timezone.utc) < self._token_exp:
+            return self._token
+        resp = httpx.post(
+            f"{self._base_url}/oauth2/tokenP",
+            json={
+                "grant_type": "client_credentials",
+                "appkey": self._app_key,
+                "appsecret": self._app_secret,
+            },
+            timeout=self._timeout,
+        )
+        resp.raise_for_status()
+        data = resp.json()
+        self._token = data["access_token"]
+        ttl = int(data.get("expires_in", 3600))
+        self._token_exp = datetime.now(tz=timezone.utc) + timedelta(seconds=max(60, ttl - 300))
+        return self._token
+
+    def _headers(self, tr_id: str) -> dict[str, str]:
+        return {
+            "authorization": f"Bearer {self._ensure_token()}",
+            "appkey": self._app_key,
+            "appsecret": self._app_secret,
+            "tr_id": tr_id,
+            "content-type": "application/json; charset=utf-8",
+        }
+
+    def _account(self) -> tuple[str, str]:
+        raw = self._account_ref.strip()
+        if "-" in raw:
+            cano, product = raw.split("-", 1)
+            return cano.strip(), product.strip()
+        digits = "".join(c for c in raw if c.isdigit())
+        return digits[:8], (digits[8:] or "01")
+
+    def _tr_submit(self, side: str) -> str:
+        if self._mock:
+            return "VTTC0012U" if side == "buy" else "VTTC0011U"
+        return "TTTC0012U" if side == "buy" else "TTTC0011U"
+
+    # --- BrokerPort ---
+    def submit_order(self, req: OrderRequest) -> OrderAck:
+        cano, product = self._account()
+        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),
+        }
+        try:
+            resp = httpx.post(
+                f"{self._base_url}/uapi/domestic-stock/v1/trading/order-cash",
+                headers=self._headers(self._tr_submit(req.side)),
+                json=body,
+                timeout=self._timeout,
+            )
+            payload = resp.json()
+        except (httpx.HTTPError, ValueError) as exc:
+            return OrderAck(accepted=False, reason=f"submit_network_error:{type(exc).__name__}",
+                            raw={"error": str(exc)})
+        if str(payload.get("rt_cd")) == "0":
+            out = payload.get("output") or {}
+            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=False, reason=payload.get("msg1") or payload.get("msg_cd"),
+                        raw=payload)
+
+    def poll_fill(self, broker_order_id: str, *, side: str, symbol: str) -> FillStatus:
+        cano, product = self._account()
+        odno = broker_order_id.split(":")[-1]
+        today = datetime.now(tz=_KST).strftime("%Y%m%d")
+        tr_id = "VTTC8001R" if self._mock else "TTTC8001R"
+        params = {
+            "CANO": cano, "ACNT_PRDT_CD": product,
+            "INQR_STRT_DT": today, "INQR_END_DT": today,
+            "SLL_BUY_DVSN_CD": "00", "INQR_DVSN": "00", "PDNO": "",
+            "CCLD_DVSN": "00", "ORD_GNO_BRNO": "", "ODNO": odno,
+            "INQR_DVSN_3": "00", "INQR_DVSN_1": "",
+            "CTX_AREA_FK100": "", "CTX_AREA_NK100": "",
+        }
+        try:
+            resp = httpx.get(
+                f"{self._base_url}/uapi/domestic-stock/v1/trading/inquire-daily-ccld",
+                headers=self._headers(tr_id), params=params, timeout=self._timeout,
+            )
+            payload = resp.json()
+        except (httpx.HTTPError, ValueError) as exc:
+            return FillStatus(state="unknown", raw={"error": str(exc)})
+        rows = payload.get("output1") or []
+        row = next((r for r in rows if str(r.get("odno")) == odno), rows[0] if rows else None)
+        if row is None:
+            return FillStatus(state="accepted", raw=payload)
+        ord_qty = _safe_int(row.get("ord_qty")) or 0
+        ccld_qty = _safe_int(row.get("tot_ccld_qty")) or 0
+        avg = _safe_int(row.get("avg_prvs")) or _safe_int(row.get("ccld_prvs"))
+        if ccld_qty <= 0:
+            return FillStatus(state="accepted", raw=row)
+        if ccld_qty >= ord_qty and ord_qty > 0:
+            return FillStatus(state="filled", filled_qty=ccld_qty,
+                              avg_price=float(avg) if avg else None, raw=row)
+        return FillStatus(state="partially_filled", filled_qty=ccld_qty,
+                          avg_price=float(avg) if avg else None, raw=row)
+
+    def cancel_order(self, broker_order_id: str, *, side: str, symbol: str, qty: int) -> bool:
+        cano, product = self._account()
+        parts = broker_order_id.split(":")
+        orgno, odno = (parts[0], parts[1]) if len(parts) == 2 else ("", parts[-1])
+        body = {
+            "CANO": cano, "ACNT_PRDT_CD": product,
+            "KRX_FWDG_ORD_ORGNO": orgno, "ORGN_ODNO": odno,
+            "ORD_DVSN": "00", "RVSE_CNCL_DVSN_CD": "02",
+            "ORD_QTY": str(qty), "ORD_UNPR": "0", "QTY_ALL_ORD_YN": "Y",
+        }
+        tr_id = "VTTC0013U" if self._mock else "TTTC0013U"
+        try:
+            resp = httpx.post(
+                f"{self._base_url}/uapi/domestic-stock/v1/trading/order-rvsecncl",
+                headers=self._headers(tr_id), json=body, timeout=self._timeout,
+            )
+            return str(resp.json().get("rt_cd")) == "0"
+        except (httpx.HTTPError, ValueError):
+            return False

trex_core/adapters/broker/paper.py

@@ -0,0 +1,84 @@
+"""Paper/stub broker — deterministic async double (REQ-011, dcy-003 OQ-2).
+
+Follows the same submit -> poll contract as the real KIS adapter (no
+synchronous shortcut), but lets tests program WHEN/whether fills happen:
+immediate, after N polls, partial, or reject. This exercises the same
+execution code path that production (KIS) takes.
+"""
+from __future__ import annotations
+
+import threading
+from typing import Optional
+
+from trex_core.application.broker_port import FillStatus, OrderAck, OrderRequest
+
+
+class _Order:
+    def __init__(self, oid: str, req: OrderRequest) -> None:
+        self.oid = oid
+        self.side = req.side
+        self.symbol = req.symbol
+        self.qty = req.qty
+        self.price = float(req.limit_price) if req.limit_price is not None else 1000.0
+        self.polls = 0
+        self.filled = 0
+        self.cancelled = False
+
+
+class PaperBroker:
+    """Deterministic async paper broker.
+
+    - fill_after_polls: number of poll_fill calls before fills begin (0 => fills on first poll)
+    - qty_per_poll: filled quantity per poll once filling (None => fill remaining at once)
+    - reject_symbols: symbols whose orders are rejected on first poll
+    """
+
+    def __init__(
+        self,
+        *,
+        fill_after_polls: int = 0,
+        qty_per_poll: Optional[int] = None,
+        reject_symbols: Optional[set[str]] = None,
+    ) -> None:
+        self._fill_after = fill_after_polls
+        self._qty_per_poll = qty_per_poll
+        self._reject = reject_symbols or set()
+        self._orders: dict[str, _Order] = {}
+        self._seq = 0
+        self._lock = threading.Lock()
+
+    def name(self) -> str:
+        return "paper"
+
+    def submit_order(self, req: OrderRequest) -> OrderAck:
+        with self._lock:
+            self._seq += 1
+            oid = f"PAPER-{self._seq:06d}"
+            self._orders[oid] = _Order(oid, req)
+        return OrderAck(accepted=True, broker_order_id=oid)
+
+    def poll_fill(self, broker_order_id: str, *, side: str, symbol: str) -> FillStatus:
+        with self._lock:
+            o = self._orders.get(broker_order_id)
+            if o is None:
+                return FillStatus(state="unknown")
+            if o.cancelled:
+                return FillStatus(state="cancelled")
+            o.polls += 1
+            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)
+            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:
+                return FillStatus(state="filled", filled_qty=o.filled, avg_price=o.price)
+            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) -> bool:
+        with self._lock:
+            o = self._orders.get(broker_order_id)
+            if o is None or o.filled >= o.qty:
+                return False
+            o.cancelled = True
+            return True

trex_core/adapters/persistence/postgresql.py

@@ -170,6 +170,48 @@ class PostgreSQLSlotRepository:
                 row = cur.fetchone()
                 return int(row[0]) if row else 0
 
+    # --- order execution queue (BAT-004 / REQ-013, option A) ---
+    def find_pending_for_dispatch(self) -> Optional[Slot]:
+        with self._psycopg.connect(self._dsn) as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    f"SELECT {_SLOT_COLUMNS} FROM daily_slot "
+                    "WHERE status IN ('pending_buy','pending_sell') AND broker_order_id IS NULL "
+                    "ORDER BY updated_at LIMIT 1 FOR UPDATE SKIP LOCKED"
+                )
+                row = cur.fetchone()
+                return self._row_to_slot(row) if row else None
+
+    def mark_dispatched(self, slot_id: UUID, broker_order_id: str) -> 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 "
+                    "WHERE slot_id=%s AND broker_order_id IS NULL",
+                    (broker_order_id, datetime.now(tz=timezone.utc), slot_id),
+                )
+                ok = cur.rowcount == 1
+                conn.commit()
+                return ok
+
+    def list_dispatched_pending(self) -> list[Slot]:
+        with self._psycopg.connect(self._dsn) as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    f"SELECT {_SLOT_COLUMNS} FROM daily_slot "
+                    "WHERE status IN ('pending_buy','pending_sell') AND broker_order_id IS NOT NULL"
+                )
+                return [self._row_to_slot(r) for r in cur.fetchall()]
+
+    def clear_dispatch(self, slot_id: UUID) -> None:
+        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",
+                    (datetime.now(tz=timezone.utc), slot_id),
+                )
+                conn.commit()
+
     def count_slots_by_status(self, trading_date: date) -> dict[SlotState, int]:
         with self._psycopg.connect(self._dsn) as conn:
             with conn.cursor() as cur: