"""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 import base64 import hashlib import json from dataclasses import replace from datetime import datetime, time, timedelta, timezone from decimal import Decimal, InvalidOperation from time import monotonic from typing import Callable, Optional import httpx from trex_core.application.broker_port import ( BalanceItem, BrokerBalanceError, CancelOutcome, CancelState, FillStatus, OrderAck, OrderRequest, calculate_fee, calculate_tax, ) _KST = timezone(timedelta(hours=9)) _SUBMISSION_UNKNOWN_PREFIX = "submission-unknown:" _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 _balance_http_error_code(status: Optional[int]) -> str: if status == 401: return "authentication_failed" if status == 403: return "authorization_failed" if status in {408, 425, 429} or (status is not None and status >= 500): return "transport_failed" return "broker_rejected" def resolve_order_venue(venue: str, at: datetime) -> str: """Resolve AUTO to the venue that accepts domestic-stock orders at `at`. 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]: try: return int(str(v).strip()) except (TypeError, ValueError): return None def _encode_submission_unknown(req: OrderRequest, submitted_at: datetime) -> str: payload = { "v": 1, "slot_id": req.slot_id, "side": req.side, "symbol": req.symbol, "qty": req.qty, "limit_price": req.limit_price, "venue": req.venue, "submitted_at": submitted_at.astimezone(_KST).isoformat(), } raw = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode() encoded = base64.urlsafe_b64encode(raw).decode().rstrip("=") return f"{_SUBMISSION_UNKNOWN_PREFIX}{encoded}" def _decode_submission_unknown(broker_order_id: str) -> Optional[dict]: if not broker_order_id.startswith(_SUBMISSION_UNKNOWN_PREFIX): return None encoded = broker_order_id[len(_SUBMISSION_UNKNOWN_PREFIX) :] try: raw = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4)) payload = json.loads(raw) except (ValueError, UnicodeDecodeError, json.JSONDecodeError): return None required = { "v", "slot_id", "side", "symbol", "qty", "venue", "submitted_at", } if ( not isinstance(payload, dict) or payload.get("v") != 1 or not required.issubset(payload) or payload.get("side") not in {"buy", "sell"} ): return None try: datetime.fromisoformat(str(payload["submitted_at"])) int(payload["qty"]) except (TypeError, ValueError): return None return payload 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, token_cache=None, clock: Optional[Callable[[], datetime]] = None, ) -> 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) 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: 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" def account_scope_hash(self) -> str: """Stable non-secret identity for account-scoped persisted state.""" cano, product = self._account() material = f"{self._base_url}|{cano}|{product}|{self._app_key}" return hashlib.sha256(material.encode()).hexdigest() # --- credential hot-swap (dcy-036 / REQ-113 AC-2 / REQ-115 AC-1) ---------- def reinitialize(self, credential) -> None: """Replace app_key/app_secret/account_ref with fetched day-start credential. Called once at the day-start checkpoint (07:30 KST) after a successful portfolio-hub fetch (REQ-113 AC-2 / REQ-115 AC-1). base_url / timeout / allow_live / _mock are env-sourced and remain unchanged. Thread-safety: CPython GIL makes attribute assignment atomic. This is safe because reinitialize() is called at day-start (before trading) and the "day-scoped fixed cache" principle (REQ-115) ensures no intraday re-fetch. If intraday reinitialize is ever needed a lock should be added at that time. Access-token invalidation: - L1 in-memory token is cleared so the next _ensure_token() call issues a fresh token with the new credentials. - L2 Redis cache: the new (account_ref, app_key) key will miss on first lookup, triggering a fresh token issue. The old key expires via TTL — no explicit delete needed for 1x-daily rotation. Args: credential: BrokerCredentialPayload (from trex_core.application.portfolio) providing account_ref, app_key, app_secret. """ self._account_ref = credential.account_ref self._app_key = credential.app_key self._app_secret = credential.app_secret # Invalidate L1 in-memory token so next call re-issues. self._token = None self._token_exp = datetime.min.replace(tzinfo=timezone.utc) # --- auth --- def _ensure_token(self) -> str: # L1: in-memory if self._token and datetime.now(tz=timezone.utc) < self._token_exp: return self._token # L2: shared/persistent cache (Redis) — survives restart / multi-process (REQ-016) if self._token_cache is not None: cached = self._token_cache.load(self._account_ref, self._app_key) if cached is not None: self._token, self._token_exp = cached return self._token # L3: issue from KIS 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)) if self._token_cache is not None: self._token_cache.store(self._account_ref, self._app_key, self._token, self._token_exp) 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, "custtype": "P", "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" def _tr_price(self) -> str: """tr_id for current-price inquiry (REQ-038 AC-7, CHG-20260607-04). 주식현재가 시세 uses FHKST01010100 for both live and 모의투자 — KIS does not provide separate VTS quote tr_ids (the old TTTC8397R/ VTTC8397R were rejected LIVE as "없는 서비스 코드", rt_cd=1). """ return "FHKST01010100" # --- BrokerPort --- def submit_order(self, req: OrderRequest) -> OrderAck: cano, product = self._account() 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": 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": actual_venue, } try: resp = httpx.post( f"{self._base_url}/uapi/domestic-stock/v1/trading/order-cash", headers=self._headers(self._tr_submit(resolved_req.side)), json=body, timeout=self._timeout, ) payload = resp.json() except (httpx.HTTPError, ValueError) as exc: # A transport failure after POST does not prove that KIS rejected # the order. Keep the slot dispatched under a local sentinel so # the execution loop cannot submit the same order again. return OrderAck( accepted=True, 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__}, submitted_at=submitted_at, ) if getattr(resp, "status_code", 200) >= 500: return OrderAck( accepted=True, 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}, submitted_at=submitted_at, ) 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, venue=actual_venue, raw=payload, submitted_at=submitted_at, ) return OrderAck( accepted=False, reason=payload.get("msg1") or payload.get("msg_cd"), venue=actual_venue, raw=payload, ) def _daily_fill_request( self, trading_date: str, *, symbol: str = "", order_number: str = "", ) -> tuple[str, dict[str, str]]: """Build the current KIS daily-fill contract for every order venue.""" cano, product = self._account() tr_id = "VTTC0081R" if self._mock else "TTTC0081R" return tr_id, { "CANO": cano, "ACNT_PRDT_CD": product, "INQR_STRT_DT": trading_date, "INQR_END_DT": trading_date, "SLL_BUY_DVSN_CD": "00", "INQR_DVSN": "00", "PDNO": symbol, "CCLD_DVSN": "00", "ORD_GNO_BRNO": "", "ODNO": order_number, "INQR_DVSN_3": "00", "INQR_DVSN_1": "", "CTX_AREA_FK100": "", "CTX_AREA_NK100": "", "EXCG_ID_DVSN_CD": "ALL", } 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) if order_date is None: # A legacy pending row has no authoritative date anchor. ODNO is # reusable, so querying today's rows could attach another order's # terminal evidence to this slot. Never invent an inquiry date. return FillStatus( state="unknown", raw={"reason": "order_reconciliation_metadata_unknown"}, ) now = monotonic() if now < self._fill_poll_not_before.get(broker_order_id, 0.0): return FillStatus( state="accepted", raw={"reason": "fill_poll_backoff"}, ) # Set the normal gate before I/O so concurrent execution ticks cannot # 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] inquiry_date = order_date 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", headers=self._headers(tr_id), params=params, timeout=self._timeout, ) payload = resp.json() except (httpx.HTTPError, ValueError) as exc: self._fill_poll_not_before[broker_order_id] = ( monotonic() + _FILL_POLL_TRANSPORT_BACKOFF_SECONDS ) return FillStatus(state="unknown", raw={"error": str(exc)}) if getattr(resp, "status_code", 200) >= 400 or str(payload.get("rt_cd")) != "0": self._fill_poll_not_before[broker_order_id] = ( monotonic() + _FILL_POLL_TRANSPORT_BACKOFF_SECONDS ) return FillStatus( state="unknown", raw={"reason": "fill_inquiry_rejected"}, ) self._fill_poll_not_before[broker_order_id] = monotonic() + _FILL_POLL_MIN_INTERVAL_SECONDS rows = payload.get("output1") or [] normalized_odno = odno.lstrip("0") or "0" row = next( ( r for r in rows if (str(r.get("odno") or "").strip().lstrip("0") or "0") == normalized_odno ), None, ) if row is None: visible_grace = ( submitted_at is None or (self._clock().astimezone(timezone.utc) - submitted_at.astimezone(timezone.utc)) .total_seconds() < _SUBMISSION_LOOKUP_GRACE_SECONDS ) 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, broker_order_id: str, *, side: str, symbol: str, ) -> FillStatus: metadata = _decode_submission_unknown(broker_order_id) if metadata is None: return FillStatus( state="unknown", raw={"reason": "submission_unknown_metadata_invalid"}, ) now_monotonic = monotonic() if now_monotonic < self._fill_poll_not_before.get(broker_order_id, 0.0): return FillStatus( state="accepted", raw={"reason": "submission_lookup_backoff"}, ) self._fill_poll_not_before[broker_order_id] = ( now_monotonic + _FILL_POLL_MIN_INTERVAL_SECONDS ) submitted_at = datetime.fromisoformat(str(metadata["submitted_at"])).astimezone(_KST) today = submitted_at.strftime("%Y%m%d") tr_id, params = self._daily_fill_request( today, symbol=str(metadata["symbol"]), ) 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): self._fill_poll_not_before[broker_order_id] = ( monotonic() + _FILL_POLL_TRANSPORT_BACKOFF_SECONDS ) return FillStatus( state="unknown", raw={"reason": "submission_lookup_transport"}, ) if getattr(resp, "status_code", 200) >= 400 or str(payload.get("rt_cd")) != "0": self._fill_poll_not_before[broker_order_id] = ( monotonic() + _FILL_POLL_TRANSPORT_BACKOFF_SECONDS ) return FillStatus( state="unknown", raw={"reason": "submission_lookup_rejected"}, ) matches = [ row for row in (payload.get("output1") or []) if self._matches_unknown_submission(row, metadata, submitted_at) ] if len(matches) != 1: age = (datetime.now(tz=_KST) - submitted_at).total_seconds() if not matches and age >= _SUBMISSION_LOOKUP_GRACE_SECONDS: return FillStatus( state="rejected", raw={"reason": "submission_confirmed_absent"}, ) return FillStatus( state="unknown", raw={ "reason": ( "submission_lookup_ambiguous" if matches else "submission_lookup_pending" ) }, ) row = matches[0] resolved_order_id = self._broker_order_id_from_row(row) status = self._fill_status_from_row( row, side=side, today=today, fallback_odno=resolved_order_id.split(":")[-1], ) return replace( status, raw={ **(status.raw if isinstance(status.raw, dict) else {}), "resolved_broker_order_id": resolved_order_id, }, ) @staticmethod def _broker_order_id_from_row(row: dict) -> str: odno = str(row.get("odno") or "").strip() orgno = str(row.get("ord_gno_brno") or "").strip() return f"{orgno}:{odno}" if orgno else odno @staticmethod def _matches_unknown_submission( row: dict, metadata: dict, submitted_at: datetime, ) -> bool: if str(row.get("pdno") or "").strip() != str(metadata["symbol"]): return False expected_side = "02" if metadata["side"] == "buy" else "01" if str(row.get("sll_buy_dvsn_cd") or "").strip() not in {"", expected_side}: return False if (_safe_int(row.get("ord_qty")) or 0) != int(metadata["qty"]): return False expected_price = int(metadata.get("limit_price") or 0) row_price = _safe_int(row.get("ord_unpr")) if row_price is not None and row_price != expected_price: return False order_date = str(row.get("ord_dt") or "").strip() order_time = str(row.get("ord_tmd") or "").strip()[:6] if len(order_date) != 8 or len(order_time) != 6: return False try: ordered_at = datetime.strptime(order_date + order_time, "%Y%m%d%H%M%S").replace( tzinfo=_KST ) except ValueError: return False delta = (ordered_at - submitted_at).total_seconds() return -5.0 <= delta <= 120.0 @staticmethod def _fill_status_from_row( row: dict, *, side: str, today: str, fallback_odno: str, ) -> FillStatus: ord_qty = _safe_int(row.get("ord_qty")) or 0 ccld_qty = _safe_int(row.get("tot_ccld_qty")) or 0 rejected_qty = _safe_int(row.get("rjct_qty")) or 0 avg = _safe_int(row.get("avg_prvs")) or _safe_int(row.get("ccld_prvs")) # dcy-013 / REQ-012 (issue #9): external cancel detection. KIS # inquire-daily-ccld marks a cancelled order with cncl_yn="Y". Without # this, a zero-fill cancel looked like "accepted" and the slot was stuck # in pending_sell forever. Unknown/absent values are treated as not # cancelled (conservative — the stale watchdog is the backstop, R-1). cancelled = str(row.get("cncl_yn") or "").strip().upper() == "Y" if ccld_qty <= 0: # KIS can acknowledge an order submission and later report the # whole quantity as exchange-rejected without setting cncl_yn. # Treating that row as accepted leaves the slot pending forever # and suppresses the strategy's next valid signal. if rejected_qty > 0 and (ord_qty <= 0 or rejected_qty >= ord_qty): return FillStatus(state="rejected", raw=row) if cancelled: return FillStatus(state="cancelled", raw=row) # full external cancel return FillStatus(state="accepted", raw=row) if ccld_qty >= ord_qty and ord_qty > 0: # Deterministic execution_id (dcy-004 OQ-3, orbit kis.py rule): # kis-daily-ccld:{ord_dt}:{ord_gno_brno}:{odno}:{tot_ccld_qty} ord_dt = str(row.get("ord_dt") or today) ord_gno_brno = str(row.get("ord_gno_brno") or "") row_odno = str(row.get("odno") or fallback_odno) execution_id = f"kis-daily-ccld:{ord_dt}:{ord_gno_brno}:{row_odno}:{ccld_qty}" fill_price = float(avg) if avg else 0.0 # dcy-026: inquire-daily-ccld (output1) does not expose per-fill # fee/tax fields, so use the shared policy estimate (same as paper). # A future endpoint that returns broker-authoritative fee/tax would # be parsed here first, with this as fallback (FR-3). amount = ccld_qty * fill_price fee = calculate_fee(side, amount) tax = calculate_tax(side, amount) return FillStatus( state="filled", filled_qty=ccld_qty, avg_price=fill_price or None, execution_id=execution_id, fee=fee, tax=tax, raw=row, ) # 0 < ccld_qty < ord_qty if cancelled: # Partial fill then external cancel (dcy-013 FR-2): carry a # deterministic execution_id + fee so the filled portion is recorded # once, and flag cancelled so execution returns the remainder to # holding instead of waiting for a fill that will never come. ord_dt = str(row.get("ord_dt") or today) ord_gno_brno = str(row.get("ord_gno_brno") or "") row_odno = str(row.get("odno") or fallback_odno) execution_id = f"kis-daily-ccld:{ord_dt}:{ord_gno_brno}:{row_odno}:{ccld_qty}" fill_price = float(avg) if avg else 0.0 amount = ccld_qty * fill_price fee = calculate_fee(side, amount) tax = calculate_tax(side, amount) return FillStatus( state="partially_filled", filled_qty=ccld_qty, avg_price=fill_price or None, execution_id=execution_id, fee=fee, tax=tax, cancelled=True, raw=row, ) return FillStatus( state="partially_filled", filled_qty=ccld_qty, avg_price=float(avg) if avg else None, raw=row, ) def get_balance(self) -> list[BalanceItem]: """KIS 주식잔고조회 (`TTTC8434R`/`VTTC8434R`). Aggregates `output1[*].pdno` + `hldg_qty`, drops zero-qty rows (D-2 leftover, Design Decision 7). Only a validated `rt_cd=0` response may return an empty list; token/HTTP/shape/KIS failures raise a sanitized :class:`BrokerBalanceError` so reconciliation cannot infer false zero. """ cano, product = self._account() tr_id = "VTTC8434R" if self._mock else "TTTC8434R" params = { "CANO": cano, "ACNT_PRDT_CD": product, "AFHR_FLPR_YN": "N", "OFL_YN": "", "INQR_DVSN": "02", "UNPR_DVSN": "01", "FUND_STTL_ICLD_YN": "N", "FNCG_AMT_AUTO_RDPT_YN": "N", "PRCS_DVSN": "01", "CTX_AREA_FK100": "", "CTX_AREA_NK100": "", } try: headers = self._headers(tr_id) except (httpx.HTTPError, KeyError, TypeError, ValueError): # Token issuance/header construction is distinct from a rejected # balance endpoint response and must never look like an empty book. raise BrokerBalanceError("token_issuance_failed") from None quantities: dict[str, int] = {} seen_continuations: set[tuple[str, str]] = set() continuation: Optional[tuple[str, str]] = None while True: request_headers = headers if continuation is not None: params["CTX_AREA_FK100"], params["CTX_AREA_NK100"] = continuation request_headers = {**headers, "tr_cont": "N"} try: resp = httpx.get( f"{self._base_url}/uapi/domestic-stock/v1/trading/inquire-balance", headers=request_headers, params=params, timeout=self._timeout, ) resp.raise_for_status() payload = resp.json() except httpx.HTTPStatusError as exc: status = exc.response.status_code if exc.response is not None else None raise BrokerBalanceError(_balance_http_error_code(status)) from None except httpx.HTTPError: raise BrokerBalanceError("transport_failed") from None except (ValueError, TypeError): raise BrokerBalanceError("malformed_response") from None if not isinstance(payload, dict): raise BrokerBalanceError("malformed_response") if str(payload.get("rt_cd")) != "0": raise BrokerBalanceError("broker_rejected") rows = payload.get("output1") if not isinstance(rows, list): raise BrokerBalanceError("malformed_response") for row in rows: if not isinstance(row, dict): raise BrokerBalanceError("malformed_response") qty = _safe_int(row.get("hldg_qty")) if qty is None: raise BrokerBalanceError("malformed_response") if qty <= 0: continue symbol = str(row.get("pdno") or "").strip() if not symbol: raise BrokerBalanceError("malformed_response") quantities[symbol] = quantities.get(symbol, 0) + qty response_headers = getattr(resp, "headers", None) if response_headers is None: raise BrokerBalanceError("malformed_response") tr_cont = response_headers.get("tr_cont") if not isinstance(tr_cont, str): raise BrokerBalanceError("malformed_response") tr_cont = tr_cont.strip().upper() if tr_cont in {"D", "E"}: return [ BalanceItem(symbol=symbol, hldg_qty=qty) for symbol, qty in quantities.items() ] if tr_cont not in {"F", "M"}: raise BrokerBalanceError("malformed_response") next_continuation = ( str(payload.get("ctx_area_fk100") or "").strip(), str(payload.get("ctx_area_nk100") or "").strip(), ) if not all(next_continuation) or next_continuation in seen_continuations: raise BrokerBalanceError("malformed_response") seen_continuations.add(next_continuation) continuation = next_continuation def get_available_cash(self, *, price_check_code: str = "005930") -> int: """Return KIS cash-only buy capacity (`nrcvb_buy_amt`). This is intentionally separate from ``BrokerPort``. An account compatibility service may use it to refresh per-strategy portfolio budgets without teaching the domain/Core about mixed manual holdings. """ cano, product = self._account() params = { "CANO": cano, "ACNT_PRDT_CD": product, "PDNO": str(price_check_code).zfill(6), # KIS recommends market-order mode when checking cash-only buying # capacity so the symbol's margin ratio is reflected. "ORD_UNPR": "0", "ORD_DVSN": "01", "CMA_EVLU_AMT_ICLD_YN": "Y", "OVRS_ICLD_YN": "N", } try: headers = self._headers("VTTC8908R" if self._mock else "TTTC8908R") response = httpx.get( f"{self._base_url}/uapi/domestic-stock/v1/trading/inquire-psbl-order", headers=headers, params=params, timeout=self._timeout, ) response.raise_for_status() payload = response.json() except BrokerBalanceError: raise except httpx.HTTPStatusError as exc: status = exc.response.status_code if exc.response is not None else None raise BrokerBalanceError(_balance_http_error_code(status)) from None except httpx.HTTPError: raise BrokerBalanceError("transport_failed") from None except (KeyError, TypeError, ValueError): raise BrokerBalanceError("malformed_response") from None if not isinstance(payload, dict): raise BrokerBalanceError("malformed_response") if str(payload.get("rt_cd")) != "0": raise BrokerBalanceError("broker_rejected") output = payload.get("output") if not isinstance(output, dict): raise BrokerBalanceError("malformed_response") amount = _safe_int(output.get("nrcvb_buy_amt")) if amount is None or amount < 0: raise BrokerBalanceError("malformed_response") return amount def get_current_price(self, symbol: str, market: str = "J") -> Optional[Decimal]: """Current traded price `stck_prpr` via KIS inquire-price (REQ-038, REQ-118). tr_id: `FHKST01010100` per REQ-038 AC-7 (live and mock share the same tr_id). market: KIS FID_COND_MRKT_DIV_CODE (REQ-118). Default "J" (KRX equities) preserves backward compatibility for existing safety-sell callers. Returns None on network/parse error or missing price so the safety-sell worker skips the slot this tick (no false trigger). """ params = {"FID_COND_MRKT_DIV_CODE": market, "FID_INPUT_ISCD": symbol} try: resp = httpx.get( f"{self._base_url}/uapi/domestic-stock/v1/quotations/inquire-price", headers=self._headers(self._tr_price()), params=params, timeout=self._timeout, ) payload = resp.json() except (httpx.HTTPError, ValueError): return None if str(payload.get("rt_cd")) != "0": return None out = payload.get("output") or {} raw = out.get("stck_prpr") if raw is None or str(raw).strip() == "": return None try: return Decimal(str(raw).strip()) except (InvalidOperation, ValueError): return None def cancel_order( self, broker_order_id: str, *, side: str, symbol: str, qty: int, venue: str = "KRX" ) -> CancelOutcome: """Cancel a previously submitted order (REQ-118). venue: exchange venue of the original order. Must match the EXCG_ID_DVSN_CD used at submit time. Default "KRX" preserves backward compatibility. Real-account NXT cancel (EXCG=NXT) confirmed rt_cd=0 on 2026-07-06. """ if broker_order_id.startswith(_SUBMISSION_UNKNOWN_PREFIX): return CancelOutcome( CancelState.UNKNOWN, "submission_outcome_unknown", ) cano, product = self._account() parts = broker_order_id.split(":") orgno, odno = (parts[0], parts[1]) if len(parts) == 2 else ("", parts[-1]) 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", # REQ-118: cancel must use the same EXCG_ID_DVSN_CD as the original submit. "EXCG_ID_DVSN_CD": venue, } 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, ) except httpx.HTTPError: return CancelOutcome(CancelState.UNKNOWN, "transport_failed") if getattr(resp, "status_code", 200) >= 500: return CancelOutcome(CancelState.UNKNOWN, "broker_unavailable") try: payload = resp.json() except ValueError: return CancelOutcome(CancelState.UNKNOWN, "malformed_response") rt_cd = payload.get("rt_cd") if str(rt_cd) == "0": return CancelOutcome(CancelState.REQUESTED) if rt_cd is None: if getattr(resp, "status_code", 200) >= 400: return CancelOutcome(CancelState.REJECTED, "http_rejected") return CancelOutcome(CancelState.UNKNOWN, "missing_result_code") return CancelOutcome( CancelState.REJECTED, str(payload.get("msg_cd") or "broker_rejected"), )