From 06c3f06b5df8f2ca7d8f28f46827bbee4c19a691 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Sat, 12 Sep 2026 22:12:19 +0900 Subject: [PATCH] fix: reconcile KIS correction order chains --- tests/adapters/broker/test_kis.py | 76 ++- tests/adapters/broker/test_kis_cancel.py | 203 +++++++ tests/application/test_execution_cancel.py | 52 ++ .../test_kis_submission_recovery_virtual.py | 84 ++- trex_core/adapters/broker/kis.py | 529 +++++++++++++----- trex_core/application/execution.py | 69 +++ 6 files changed, 864 insertions(+), 149 deletions(-) diff --git a/tests/adapters/broker/test_kis.py b/tests/adapters/broker/test_kis.py index 7c9d0fb..f2251bf 100644 --- a/tests/adapters/broker/test_kis.py +++ b/tests/adapters/broker/test_kis.py @@ -7,6 +7,7 @@ Unit tests run always; the sandbox integration test runs only when KIS from __future__ import annotations import os +from datetime import UTC, datetime, timedelta import httpx import pytest @@ -564,7 +565,9 @@ def test_submit_unknown_confirmed_absent_becomes_retryable_reject(monkeypatch): status = b.poll_fill(ack.broker_order_id, side="buy", symbol="005930") assert status.state == "rejected" - assert status.raw == {"reason": "submission_confirmed_absent"} + assert status.raw["reason"] == "submission_confirmed_absent" + assert status.raw["poll_performed"] is True + assert status.raw["poll_count"] == 1 def test_fill_poll_transport_failure_uses_backoff(monkeypatch): @@ -588,8 +591,12 @@ def test_fill_poll_transport_failure_uses_backoff(monkeypatch): ) assert first.state == "unknown" + assert first.raw["reason"] == "fill_inquiry_transport" + assert "timed out" not in str(first.raw) assert immediate.state == "accepted" - assert immediate.raw == {"reason": "fill_poll_backoff"} + assert immediate.raw["reason"] == "fill_poll_backoff" + assert immediate.raw["poll_performed"] is False + assert immediate.raw["poll_count"] == 1 assert len(calls) == 1 clock[0] += 31 @@ -598,3 +605,68 @@ def test_fill_poll_transport_failure_uses_backoff(monkeypatch): ) assert again.state == "unknown" assert len(calls) == 2 + + +def test_unresolved_fill_poll_uses_elapsed_bounded_backoff_and_converges(monkeypatch): + wall_now = datetime(2026, 9, 10, 12, 0, tzinfo=UTC) + b = KISBroker( + "12345678-01", + app_key="k", + app_secret="s", + base_url=_MOCK_URL, + clock=lambda: wall_now, + ) + monkeypatch.setattr(b, "_headers", lambda _tr: {}) + mono = [100.0] + monkeypatch.setattr("trex_core.adapters.broker.kis.monotonic", lambda: mono[0]) + calls = [] + terminal = [False] + + class Response: + status_code = 200 + + def json(self): + calls.append(1) + parent = { + "odno": "1234", "orgn_odno": "", "ord_qty": "10", + "tot_ccld_qty": "0", "avg_prvs": "0", "cncl_yn": "", + "rjct_qty": "0", "rmn_qty": "0", "ord_dt": "20260910", + "ord_gno_brno": "91258", "pdno": "005930", + } + child = { + **parent, "odno": "1235", "orgn_odno": "1234", + "cnc_cfrm_qty": "10" if terminal[0] else "0", + } + return {"rt_cd": "0", "output1": [parent, child]} + + monkeypatch.setattr("trex_core.adapters.broker.kis.httpx.get", lambda *a, **kw: Response()) + submitted_at = wall_now - timedelta(hours=7) + + first = b.poll_fill( + "91258:1234", side="buy", symbol="005930", + order_date="20260910", submitted_at=submitted_at, + ) + assert first.state == "unknown" + assert first.raw["poll_performed"] is True + assert first.raw["poll_count"] == 1 + assert first.raw["backoff_sec"] == 300.0 + assert first.raw["stuck_duration_sec"] == 7 * 3600 + + terminal[0] = True + mono[0] += 299 + skipped = b.poll_fill( + "91258:1234", side="buy", symbol="005930", + order_date="20260910", submitted_at=submitted_at, + ) + assert skipped.state == "accepted" + assert skipped.raw["last_result_state"] == "unknown" + assert len(calls) == 1 + + mono[0] += 1 + resolved = b.poll_fill( + "91258:1234", side="buy", symbol="005930", + order_date="20260910", submitted_at=submitted_at, + ) + assert resolved.state == "cancelled" + assert resolved.raw["poll_count"] == 2 + assert len(calls) == 2 diff --git a/tests/adapters/broker/test_kis_cancel.py b/tests/adapters/broker/test_kis_cancel.py index 8494f0a..9ff1beb 100644 --- a/tests/adapters/broker/test_kis_cancel.py +++ b/tests/adapters/broker/test_kis_cancel.py @@ -222,6 +222,209 @@ def test_correction_child_row_not_misclassified_as_cancel(monkeypatch): assert not raw.get("order_open_confirmed", False) +def test_correction_child_cancel_confirmed_quantity_maps_to_cancelled(monkeypatch): + """KIS cancel-confirm quantity is terminal evidence even when cncl_yn is blank.""" + payload = _payload_with_child_cancel(ODNO, "0000123458", ord_qty="31", symbol="053050") + parent, child = payload["output1"] + child.update(cncl_yn="", cnc_cfrm_qty="31") + monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload)) + + fs = _broker().poll_fill( + f"06010:{ODNO}", side="sell", symbol="053050", order_date="20260901" + ) + + assert fs.state == "cancelled" + assert fs.filled_qty == 0 + assert fs.raw["reason"] == "replacement_chain_terminal_cancel" + assert "resolved_broker_order_id" not in fs.raw + + +def test_live_replacement_remains_fail_closed(monkeypatch): + payload = _payload_with_child_cancel(ODNO, "0000123458") + _, child = payload["output1"] + child.update(cncl_yn="", cnc_cfrm_qty="0", rmn_qty="10", ord_qty="10") + monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload)) + + fs = _broker().poll_fill( + f"06010:{ODNO}", side="sell", symbol="225190", order_date="20260901" + ) + + assert fs.state == "unknown" + assert fs.raw["reason"] == "replacement_live_unresolved" + assert not fs.raw.get("order_open_confirmed", False) + + +def test_replacement_partial_fill_then_confirmed_cancel(monkeypatch): + payload = _payload_with_child_cancel(ODNO, "0000123458", ord_qty="10") + _, child = payload["output1"] + child.update( + cncl_yn="", + cnc_cfrm_qty="6", + tot_ccld_qty="4", + rmn_qty="0", + avg_prvs="70100", + ) + monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload)) + + fs = _broker().poll_fill( + f"06010:{ODNO}", side="sell", symbol="225190", order_date="20260901" + ) + + assert fs.state == "partially_filled" + assert fs.filled_qty == 4 + assert fs.avg_price == 70100.0 + assert fs.is_partial_cancel + assert fs.raw["reason"] == "replacement_chain_terminal_cancel" + + +def test_repeated_cumulative_fill_across_generations_is_not_ambiguous(monkeypatch): + payload = _payload_with_child_cancel(ODNO, "0000123458", ord_qty="10") + parent, child = payload["output1"] + parent.update(tot_ccld_qty="4", rmn_qty="0", avg_prvs="70100") + child.update( + cncl_yn="", + cnc_cfrm_qty="6", + tot_ccld_qty="4", + rmn_qty="0", + avg_prvs="70100", + ) + monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload)) + + fs = _broker().poll_fill( + f"06010:{ODNO}", side="sell", symbol="225190", order_date="20260901" + ) + + assert fs.state == "partially_filled" + assert fs.filled_qty == 4 + assert fs.avg_price == 70100.0 + assert fs.is_partial_cancel + + +def test_multi_generation_replacement_chain_terminal_cancel(monkeypatch): + payload = _payload_with_child_cancel(ODNO, "0000123458", ord_qty="10") + parent, replacement = payload["output1"] + replacement.update(cncl_yn="", cnc_cfrm_qty="0", rmn_qty="0") + terminal = { + **replacement, + "odno": "0000123459", + "orgn_odno": replacement["odno"], + "cnc_cfrm_qty": "10", + } + payload["output1"] = [terminal, parent, replacement] + monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload)) + + fs = _broker().poll_fill( + f"06010:{ODNO}", side="sell", symbol="225190", order_date="20260901" + ) + + assert fs.state == "cancelled" + assert fs.raw["replacement_depth"] == 2 + + +def test_replacement_chain_branching_is_ambiguous(monkeypatch): + payload = _payload_with_child_cancel(ODNO, "0000123458", ord_qty="10") + parent, child = payload["output1"] + child.update(cncl_yn="", cnc_cfrm_qty="10") + sibling = {**child, "odno": "0000123459"} + payload["output1"] = [parent, child, sibling] + monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload)) + + fs = _broker().poll_fill( + f"06010:{ODNO}", side="sell", symbol="225190", order_date="20260901" + ) + + assert fs.state == "unknown" + assert fs.raw["reason"] == "replacement_chain_ambiguous" + + +def test_zero_remaining_without_cancel_confirmation_remains_unknown(monkeypatch): + payload = _payload_with_child_cancel(ODNO, "0000123458", ord_qty="10") + _, child = payload["output1"] + child.update(cncl_yn="", cnc_cfrm_qty="0", rmn_qty="0") + monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload)) + + fs = _broker().poll_fill( + f"06010:{ODNO}", side="sell", symbol="225190", order_date="20260901" + ) + + assert fs.state == "unknown" + assert fs.raw["reason"] == "correction_child_present_unresolved" + + +def test_stable_zero_fill_replacement_chain_converges_after_grace(monkeypatch): + payload = _payload_with_child_cancel(ODNO, "0000123458", ord_qty="31", symbol="053050") + parent, child = payload["output1"] + parent.update(rmn_qty="0") + child.update(cncl_yn="", cnc_cfrm_qty="0", rmn_qty="0") + monotonic_now = [100.0] + monkeypatch.setattr(kis_mod, "monotonic", lambda: monotonic_now[0]) + monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload)) + broker = _broker() + + first = broker.poll_fill( + f"06010:{ODNO}", side="sell", symbol="053050", order_date="20260901" + ) + assert first.state == "unknown" + assert first.raw["reason"] == "correction_child_present_unresolved" + + monotonic_now[0] += 90 + confirmed = broker.poll_fill( + f"06010:{ODNO}", side="sell", symbol="053050", order_date="20260901" + ) + assert confirmed.state == "cancelled" + assert confirmed.raw["reason"] == "replacement_chain_stable_zero_terminal" + + +def test_zero_fill_terminal_grace_resets_when_chain_changes(monkeypatch): + payload = _payload_with_child_cancel(ODNO, "0000123458", ord_qty="31", symbol="053050") + parent, child = payload["output1"] + parent.update(rmn_qty="0") + child.update(cncl_yn="", cnc_cfrm_qty="0", rmn_qty="0") + monotonic_now = [100.0] + monkeypatch.setattr(kis_mod, "monotonic", lambda: monotonic_now[0]) + monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload)) + broker = _broker() + + first = broker.poll_fill( + f"06010:{ODNO}", side="sell", symbol="053050", order_date="20260901" + ) + assert first.state == "unknown" + + monotonic_now[0] += 90 + child["rmn_qty"] = "31" + changed = broker.poll_fill( + f"06010:{ODNO}", side="sell", symbol="053050", order_date="20260901" + ) + assert changed.raw["reason"] == "replacement_live_unresolved" + + monotonic_now[0] += 90 + child["rmn_qty"] = "0" + restarted_grace = broker.poll_fill( + f"06010:{ODNO}", side="sell", symbol="053050", order_date="20260901" + ) + assert restarted_grace.state == "unknown" + + monotonic_now[0] += 90 + confirmed = broker.poll_fill( + f"06010:{ODNO}", side="sell", symbol="053050", order_date="20260901" + ) + assert confirmed.state == "cancelled" + + +def test_incomplete_cancel_confirmation_remains_unknown(monkeypatch): + payload = _payload_with_child_cancel(ODNO, "0000123458", ord_qty="10") + _, child = payload["output1"] + child.update(cncl_yn="", cnc_cfrm_qty="3", rmn_qty="0") + monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload)) + + fs = _broker().poll_fill( + f"06010:{ODNO}", side="sell", symbol="225190", order_date="20260901" + ) + + assert fs.state == "unknown" + assert fs.raw["reason"] == "correction_child_present_unresolved" + + def test_partial_fill_with_child_cancel_row(monkeypatch): """Partial fill on parent + cancel child → partial_cancel.""" parent = { diff --git a/tests/application/test_execution_cancel.py b/tests/application/test_execution_cancel.py index e606db3..a88fd8c 100644 --- a/tests/application/test_execution_cancel.py +++ b/tests/application/test_execution_cancel.py @@ -104,6 +104,58 @@ def test_full_external_cancel_returns_slot_to_holding(): assert r.rolled_back == 1 +def test_unresolved_poll_observability_is_sanitized_and_deduplicated(caplog): + broker = ProgrammableBroker() + repo, _svc, intents, execution, _ledger, _writer = _wire(broker) + sid = _buy_to_holding(intents, execution) + assert sid is not None + intents.submit_sell(SellIntent(intent_id="s-poll-observe", slot_id=sid)) + samples = iter([ + ("correction_child_present_unresolved", 10), + ("correction_child_present_unresolved", 10), + ("correction_child_present_unresolved", 61), + ("correction_child_present_unresolved", 61), + ("parent_identity_mismatch", 61), + ]) + + def unresolved(side): + del side + reason, age = next(samples) + return FillStatus( + state="unknown", + raw={ + "poll_performed": True, + "poll_count": broker.polls, + "stuck_duration_sec": age, + "last_result_reason": reason, + "backoff_sec": 15.0, + "raw_payload": {"authorization": "secret-token"}, + "broker_order_id": "sensitive-order-id", + }, + ) + + broker.responder = unresolved + with caplog.at_level(logging.INFO): + for _ in range(5): + execution.run_once() + + observations = [ + record for record in caplog.records + if getattr(record, "event", None) == "order_poll" + ] + warnings = [ + record for record in caplog.records + if getattr(record, "event", None) == "order_poll_unresolved" + ] + assert len(observations) == 5 + assert len(warnings) == 3 # first, long-duration threshold, reason change + assert observations[-1].poll_count == 6 + assert observations[-1].stuck_duration_sec == 61 + assert observations[-1].result_reason == "parent_identity_mismatch" + assert "secret-token" not in caplog.text + assert "sensitive-order-id" not in caplog.text + + def test_partial_fill_then_cancel_records_and_keeps_remainder(): broker = ProgrammableBroker() repo, svc, intents, execution, ledger, writer = _wire(broker) diff --git a/tests/integration/test_kis_submission_recovery_virtual.py b/tests/integration/test_kis_submission_recovery_virtual.py index 1d316d9..0a5e545 100644 --- a/tests/integration/test_kis_submission_recovery_virtual.py +++ b/tests/integration/test_kis_submission_recovery_virtual.py @@ -12,7 +12,7 @@ from zoneinfo import ZoneInfo from trex_core.adapters.broker.kis import KISBroker from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore from trex_core.application.execution import ExecutionService -from trex_core.application.intents import BuyIntent, IntentService +from trex_core.application.intents import BuyIntent, IntentService, SellIntent from trex_core.domain.slot import SlotService, SlotState from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository @@ -30,12 +30,15 @@ class _ExchangeState: self.orders: list[dict[str, str]] = [] self.fill_orders = False self.reject_orders = False + self.next_side = "02" + self.correction_cancel_sell = False + self.balance_qty: int | None = None def submit(self, body: dict[str, str]) -> tuple[bool, dict[str, object]]: self.submit_attempts += 1 order = { "pdno": body["PDNO"], - "sll_buy_dvsn_cd": "02", + "sll_buy_dvsn_cd": self.next_side, "ord_qty": body["ORD_QTY"], "ord_unpr": body["ORD_UNPR"], "ord_dt": VIRTUAL_AT.strftime("%Y%m%d"), @@ -65,6 +68,20 @@ class _ExchangeState: order["rjct_qty"] = order["ord_qty"] order["rmn_qty"] = "0" row = dict(order) + if self.correction_cancel_sell and order["sll_buy_dvsn_cd"] == "01": + row["rmn_qty"] = "0" + rows.append(row) + rows.append( + { + **row, + "odno": f"{int(order['odno']) + 1:010d}", + "orgn_odno": order["odno"], + "cncl_yn": "", + "cnc_cfrm_qty": "0", + "rmn_qty": "0", + } + ) + continue if self.fill_orders and order["rjct_qty"] == "0": row["tot_ccld_qty"] = row["ord_qty"] row["avg_prvs"] = row["ord_unpr"] @@ -72,11 +89,13 @@ class _ExchangeState: return {"rt_cd": "0", "output1": rows} def balance(self) -> dict[str, object]: - quantity = ( - sum(int(order["ord_qty"]) for order in self.orders if order["rjct_qty"] == "0") - if self.fill_orders - else 0 - ) + quantity = self.balance_qty + if quantity is None: + quantity = ( + sum(int(order["ord_qty"]) for order in self.orders if order["rjct_qty"] == "0") + if self.fill_orders + else 0 + ) rows = [{"pdno": SYMBOL, "hldg_qty": str(quantity)}] if quantity else [] return {"rt_cd": "0", "output1": rows} @@ -265,3 +284,54 @@ def test_exchange_reject_after_broker_ack_rolls_back_and_allows_one_retry(): finally: server.shutdown() server.server_close() + + +def test_stable_correction_chain_returns_pending_sell_to_holding(monkeypatch): + state = _ExchangeState(first_attempt_reaches_exchange=True) + server = _serve(state) + mono = [100.0] + monkeypatch.setattr("trex_core.adapters.broker.kis.monotonic", lambda: mono[0]) + try: + repo, slots, intents, execution, slot_id = _wire(_broker(server)) + assert slot_id is not None + + execution.run_once() + state.fill_orders = True + execution = ExecutionService( + slots, + repo, + _broker(server), + instance_key=INSTANCE, + clock=lambda: VIRTUAL_AT, + ) + execution.reconcile() + bought = repo.fetch_slot(slot_id) + assert bought is not None + assert bought.status is SlotState.HOLDING + + state.fill_orders = False + state.balance_qty = 2 + state.next_side = "01" + state.correction_cancel_sell = True + admitted = intents.submit_sell( + SellIntent(intent_id="virtual-sell-corrected", slot_id=slot_id) + ) + assert admitted.status == "admitted" + + execution.run_once() + unresolved = repo.fetch_slot(slot_id) + assert unresolved is not None + assert unresolved.status is SlotState.PENDING_SELL + assert state.submit_attempts == 2 + + mono[0] += 90 + execution.run_once() + holding = repo.fetch_slot(slot_id) + assert holding is not None + assert holding.status is SlotState.HOLDING + assert holding.qty == 2 + assert holding.broker_order_id is None + assert state.submit_attempts == 2 + finally: + server.shutdown() + server.server_close() diff --git a/trex_core/adapters/broker/kis.py b/trex_core/adapters/broker/kis.py index 32dadcd..4c2e711 100644 --- a/trex_core/adapters/broker/kis.py +++ b/trex_core/adapters/broker/kis.py @@ -19,6 +19,7 @@ from dataclasses import replace from datetime import UTC, datetime, timedelta, timezone from decimal import Decimal, InvalidOperation from time import monotonic +from typing import Any import httpx @@ -39,6 +40,7 @@ _SUBMISSION_UNKNOWN_PREFIX = "submission-unknown:" _MOCK_HINTS = ("openapivts", "virtual", "mock") _FILL_POLL_MIN_INTERVAL_SECONDS = 5.0 _FILL_POLL_TRANSPORT_BACKOFF_SECONDS = 30.0 +_FILL_POLL_MAX_BACKOFF_SECONDS = 300.0 _SUBMISSION_LOOKUP_GRACE_SECONDS = 90.0 _SUPPORTED_ORDER_VENUES = frozenset({"KRX", "NXT", "SOR"}) @@ -150,6 +152,7 @@ class KISBroker: 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] = {} + self._fill_poll_stats: dict[str, dict[str, Any]] = {} is_mock = any(h in self._base_url for h in _MOCK_HINTS) if not is_mock and not allow_live: raise RuntimeError( @@ -358,6 +361,118 @@ class KISBroker: "EXCG_ID_DVSN_CD": "ALL", } + def _fill_poll_age_seconds(self, submitted_at: datetime | None) -> int: + if submitted_at is None: + return 0 + return max( + 0, + int( + ( + self._clock().astimezone(UTC) + - submitted_at.astimezone(UTC) + ).total_seconds() + ), + ) + + @staticmethod + def _fill_poll_backoff_seconds( + status: FillStatus, + *, + age_seconds: int, + transport_failure: bool, + ) -> float: + """Return bounded state/age-based delay for the next authoritative poll.""" + if status.state != "unknown": + delay = _FILL_POLL_MIN_INTERVAL_SECONDS + elif age_seconds < 60: + delay = _FILL_POLL_MIN_INTERVAL_SECONDS + elif age_seconds < 5 * 60: + delay = 15.0 + elif age_seconds < 60 * 60: + delay = 30.0 + elif age_seconds < 6 * 60 * 60: + delay = 120.0 + else: + delay = _FILL_POLL_MAX_BACKOFF_SECONDS + if transport_failure: + delay = max(delay, _FILL_POLL_TRANSPORT_BACKOFF_SECONDS) + return min(delay, _FILL_POLL_MAX_BACKOFF_SECONDS) + + def _record_fill_poll( + self, + broker_order_id: str, + status: FillStatus, + *, + submitted_at: datetime | None, + transport_failure: bool = False, + ) -> FillStatus: + """Attach allowlisted poll telemetry and schedule the next broker read.""" + age_seconds = self._fill_poll_age_seconds(submitted_at) + backoff_seconds = self._fill_poll_backoff_seconds( + status, + age_seconds=age_seconds, + transport_failure=transport_failure, + ) + raw = status.raw if isinstance(status.raw, dict) else {} + reason = raw.get("reason") if isinstance(raw.get("reason"), str) else status.state + previous = self._fill_poll_stats.get(broker_order_id, {}) + if reason != "correction_child_present_unresolved": + previous.pop("zero_fill_chain_fingerprint", None) + previous.pop("zero_fill_chain_first_seen", None) + poll_count = int(previous.get("poll_count", 0)) + 1 + self._fill_poll_not_before[broker_order_id] = monotonic() + backoff_seconds + self._fill_poll_stats[broker_order_id] = { + **previous, + "poll_count": poll_count, + "last_result_state": status.state, + "last_result_reason": reason, + "backoff_sec": backoff_seconds, + } + return replace( + status, + raw={ + **raw, + "poll_performed": True, + "poll_count": poll_count, + "stuck_duration_sec": age_seconds, + "last_result_state": status.state, + "last_result_reason": reason, + "backoff_sec": backoff_seconds, + }, + ) + + def _replacement_zero_fill_terminal_confirmed( + self, + broker_order_id: str, + chain: list[dict], + ) -> bool: + """Require a stable grace-period snapshot before terminal inference.""" + fingerprint = tuple( + ( + str(row.get("odno") or "").strip(), + str(row.get("orgn_odno") or "").strip(), + _safe_int(row.get("ord_qty")), + _safe_int(row.get("tot_ccld_qty")), + _safe_int(row.get("rmn_qty")), + ) + for row in chain + ) + stats = self._fill_poll_stats.setdefault(broker_order_id, {}) + now = monotonic() + if stats.get("zero_fill_chain_fingerprint") != fingerprint: + stats["zero_fill_chain_fingerprint"] = fingerprint + stats["zero_fill_chain_first_seen"] = now + return False + first_seen = float(stats.get("zero_fill_chain_first_seen", now)) + return now - first_seen >= _SUBMISSION_LOOKUP_GRACE_SECONDS + + def _clear_replacement_zero_fill_candidate(self, broker_order_id: str) -> None: + stats = self._fill_poll_stats.get(broker_order_id) + if stats is None: + return + stats.pop("zero_fill_chain_fingerprint", None) + stats.pop("zero_fill_chain_first_seen", None) + def poll_fill( self, broker_order_id: str, @@ -379,9 +494,18 @@ class KISBroker: ) now = monotonic() if now < self._fill_poll_not_before.get(broker_order_id, 0.0): + stats = self._fill_poll_stats.get(broker_order_id, {}) return FillStatus( state="accepted", - raw={"reason": "fill_poll_backoff"}, + raw={ + "reason": "fill_poll_backoff", + "poll_performed": False, + "poll_count": int(stats.get("poll_count", 0)), + "stuck_duration_sec": self._fill_poll_age_seconds(submitted_at), + "last_result_state": stats.get("last_result_state"), + "last_result_reason": stats.get("last_result_reason"), + "backoff_sec": float(stats.get("backoff_sec", 0.0)), + }, ) # Set the normal gate before I/O so concurrent execution ticks cannot # open duplicate KIS inquiries for the same order. @@ -397,20 +521,20 @@ class KISBroker: 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 + except (httpx.HTTPError, ValueError): + return self._record_fill_poll( + broker_order_id, + FillStatus(state="unknown", raw={"reason": "fill_inquiry_transport"}), + submitted_at=submitted_at, + transport_failure=True, ) - 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 self._record_fill_poll( + broker_order_id, + FillStatus(state="unknown", raw={"reason": "fill_inquiry_rejected"}), + submitted_at=submitted_at, + transport_failure=True, ) - 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" # Canonical request identity from the caller — these are authoritative @@ -454,13 +578,17 @@ class KISBroker: # No row with the exact request identity. If there are odno # matches they all failed identity validation → unknown (not grace). if odno_matches: - return FillStatus( - state="unknown", - raw={ - "reason": "parent_identity_mismatch", - "order_visible": True, - "inquiry_date": inquiry_date, - }, + return self._record_fill_poll( + broker_order_id, + FillStatus( + state="unknown", + raw={ + "reason": "parent_identity_mismatch", + "order_visible": True, + "inquiry_date": inquiry_date, + }, + ), + submitted_at=submitted_at, ) # Truly invisible — apply the submission visibility grace window. visible_grace = ( @@ -469,118 +597,210 @@ class KISBroker: .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, - }, + return self._record_fill_poll( + broker_order_id, + 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, + }, + ), + submitted_at=submitted_at, ) if len(exact_parent_matches) > 1: # More than one row claims the same exact identity — ambiguous. - return FillStatus( - state="unknown", - raw={ - "reason": "parent_identity_ambiguous", - "order_visible": True, - "inquiry_date": inquiry_date, - }, + return self._record_fill_poll( + broker_order_id, + FillStatus( + state="unknown", + raw={ + "reason": "parent_identity_ambiguous", + "order_visible": True, + "inquiry_date": inquiry_date, + }, + ), + submitted_at=submitted_at, ) row = exact_parent_matches[0] - # dcy-013 ext: KIS sometimes records an external cancel as a separate - # child row with orgn_odno= and cncl_yn="Y" rather than - # setting cncl_yn on the original row. Scan sibling rows to detect - # this case. - # - # Safety requirements: - # - Child must share the AUTHORITATIVE request identity (branch, date, - # symbol from the caller). All identity fields must be present and - # equal — absent fields are NOT treated as wildcards (fail-closed). - # - Ambiguous (> 1 matching cancel) → fail-closed "unknown". - # - Correction (정정) children (cncl_yn != "Y") that match the same - # request identity are also fail-closed → "unknown". - # - Unrelated rows whose identity fields differ from the request are - # excluded from both candidate lists. - def _child_matches_request(r: dict) -> bool: - """True when a child row's identity fields are all present and - equal to the authoritative request identity. Missing = mismatch.""" - child_brno = str(r.get("ord_gno_brno") or "").strip() - child_dt = str(r.get("ord_dt") or "").strip() - child_pdno = str(r.get("pdno") or "").strip() - if request_brno and (not child_brno or child_brno != request_brno): - return False - if not child_dt or child_dt != request_date: - return False - return bool(child_pdno) and child_pdno == request_symbol + # Follow only response rows bound to the immutable request identity. + # KIS may represent cancel/modify as a chain where each child points to + # its immediate predecessor through orgn_odno. Row order and numerical + # proximity are never evidence of parentage. + identity_rows = [r for r in rows if _row_matches_request_identity(r)] - child_cancel_candidates = [ - r - for r in rows - if (str(r.get("orgn_odno") or "").strip().lstrip("0") or "0") == normalized_odno - and str(r.get("cncl_yn") or "").strip().upper() == "Y" - and _child_matches_request(r) - ] + def _row_parent_odno_normalized(r: dict) -> str: + return str(r.get("orgn_odno") or "").strip().lstrip("0") or "0" - # Correction children: same orgn_odno + matching identity but cncl_yn != "Y". - correction_candidates = [ - r - for r in rows - if (str(r.get("orgn_odno") or "").strip().lstrip("0") or "0") == normalized_odno - and str(r.get("cncl_yn") or "").strip().upper() != "Y" - and _child_matches_request(r) - ] + chain = [row] + visited = {normalized_odno} + current_odno = normalized_odno + ambiguous_reason: str | None = None + while True: + children = [ + candidate + for candidate in identity_rows + if _row_parent_odno_normalized(candidate) == current_odno + ] + if not children: + break + if len(children) > 1: + ambiguous_reason = ( + "child_cancel_ambiguous" + if all( + str(child.get("cncl_yn") or "").strip().upper() == "Y" + for child in children + ) + else "replacement_chain_ambiguous" + ) + break + child = children[0] + child_odno = _row_odno_normalized(child) + if child_odno == "0" or child_odno in visited: + ambiguous_reason = "replacement_chain_ambiguous" + break + chain.append(child) + visited.add(child_odno) + current_odno = child_odno - if len(child_cancel_candidates) > 1: - # More than one matching cancel child — identity is ambiguous. - return FillStatus( - state="unknown", - raw={ - "reason": "child_cancel_ambiguous", - "order_visible": True, - "inquiry_date": inquiry_date, - }, + if ambiguous_reason is not None: + return self._record_fill_poll( + broker_order_id, + FillStatus( + state="unknown", + raw={ + "reason": ambiguous_reason, + "order_visible": True, + "inquiry_date": inquiry_date, + }, + ), + submitted_at=submitted_at, ) - if correction_candidates: - # A correction/replacement child is present; the original order is - # superseded but we cannot safely infer terminal status without - # explicit broker confirmation of the replacement identity. - return FillStatus( - state="unknown", - raw={ - "reason": "correction_child_present_unresolved", - "order_visible": True, - "inquiry_date": inquiry_date, - }, + terminal_row = chain[-1] + terminal_order_qty = _safe_int(terminal_row.get("ord_qty")) or 0 + terminal_filled_qty = _safe_int(terminal_row.get("tot_ccld_qty")) or 0 + terminal_cancelled_qty = _safe_int(terminal_row.get("cnc_cfrm_qty")) or 0 + terminal_cancel = ( + str(terminal_row.get("cncl_yn") or "").strip().upper() == "Y" + or ( + terminal_order_qty > 0 + and terminal_cancelled_qty > 0 + and terminal_cancelled_qty + terminal_filled_qty >= terminal_order_qty ) - - if child_cancel_candidates and str(row.get("cncl_yn") or "").strip().upper() != "Y": - child_cancel_row = child_cancel_candidates[0] - # Inject the cancel signal from the child row into the parent row so - # _fill_status_from_row can treat it as cancelled, while keeping - # the original row's fill quantities authoritative. - row = {**row, "cncl_yn": "Y", "_child_cancel_odno": str(child_cancel_row.get("odno") or "")} - - status = self._fill_status_from_row( - row, side=side, today=inquiry_date, fallback_odno=odno ) - return replace( + terminal_remaining = _safe_int(terminal_row.get("rmn_qty")) + positive_fill_rows = [ + candidate + for candidate in chain + if (_safe_int(candidate.get("tot_ccld_qty")) or 0) > 0 + ] + positive_fill_signatures = { + ( + _safe_int(candidate.get("tot_ccld_qty")), + str(candidate.get("avg_prvs") or "").strip(), + ) + for candidate in positive_fill_rows + } + repeated_cumulative_fill = ( + len(positive_fill_rows) > 1 and len(positive_fill_signatures) == 1 + ) + stable_zero_fill_candidate = ( + len(chain) > 1 + and not terminal_cancel + and not positive_fill_rows + and all((_safe_int(candidate.get("ord_qty")) or 0) > 0 for candidate in chain) + and all(_safe_int(candidate.get("rmn_qty")) == 0 for candidate in chain) + and all((_safe_int(candidate.get("rjct_qty")) or 0) == 0 for candidate in chain) + ) + stable_zero_fill_terminal = False + if stable_zero_fill_candidate: + stable_zero_fill_terminal = self._replacement_zero_fill_terminal_confirmed( + broker_order_id, + chain, + ) + else: + self._clear_replacement_zero_fill_candidate(broker_order_id) + + if len(positive_fill_rows) > 1 and not repeated_cumulative_fill: + status = FillStatus( + state="unknown", + raw={"reason": "replacement_chain_fill_ambiguous"}, + ) + elif (terminal_cancel or stable_zero_fill_terminal) and terminal_remaining in {None, 0}: + # A child cancel row often carries no fills of its own. Use the one + # unambiguous filled generation when present; otherwise preserve the + # original quantity context and inject only the terminal cancel bit. + evidence_row = positive_fill_rows[-1] if positive_fill_rows else row + status = self._fill_status_from_row( + {**evidence_row, "cncl_yn": "Y"}, + side=side, + today=inquiry_date, + fallback_odno=odno, + ) + status = replace( + status, + raw={ + **(status.raw if isinstance(status.raw, dict) else {}), + "reason": ( + "replacement_chain_terminal_cancel" + if terminal_cancel + else "replacement_chain_stable_zero_terminal" + ), + "replacement_depth": len(chain) - 1, + }, + ) + elif len(chain) > 1: + leaf_status = self._fill_status_from_row( + terminal_row, + side=side, + today=inquiry_date, + fallback_odno=odno, + ) + if leaf_status.is_filled: + status = leaf_status + else: + status = FillStatus( + state="unknown", + raw={ + "reason": ( + "replacement_live_unresolved" + if (terminal_remaining or 0) > 0 + else "correction_child_present_unresolved" + ), + "replacement_depth": len(chain) - 1, + }, + ) + else: + status = self._fill_status_from_row( + row, side=side, today=inquiry_date, fallback_odno=odno + ) + + status = replace( status, raw={ **(status.raw if isinstance(status.raw, dict) else {}), "order_visible": True, - "order_open_confirmed": status.state in {"accepted", "partially_filled"}, + "order_open_confirmed": ( + len(chain) == 1 + and status.state in {"accepted", "partially_filled"} + ), "inquiry_date": inquiry_date, }, ) + return self._record_fill_poll( + broker_order_id, + status, + submitted_at=submitted_at, + ) def _poll_unknown_submission( self, @@ -595,16 +815,33 @@ class KISBroker: state="unknown", raw={"reason": "submission_unknown_metadata_invalid"}, ) + try: + submitted_at = datetime.fromisoformat(str(metadata["submitted_at"])).astimezone( + _KST + ) + except (KeyError, TypeError, ValueError): + 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): + stats = self._fill_poll_stats.get(broker_order_id, {}) return FillStatus( state="accepted", - raw={"reason": "submission_lookup_backoff"}, + raw={ + "reason": "submission_lookup_backoff", + "poll_performed": False, + "poll_count": int(stats.get("poll_count", 0)), + "stuck_duration_sec": self._fill_poll_age_seconds(submitted_at), + "last_result_state": stats.get("last_result_state"), + "last_result_reason": stats.get("last_result_reason"), + "backoff_sec": float(stats.get("backoff_sec", 0.0)), + }, ) 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, @@ -619,20 +856,18 @@ class KISBroker: ) 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"}, + return self._record_fill_poll( + broker_order_id, + FillStatus(state="unknown", raw={"reason": "submission_lookup_transport"}), + submitted_at=submitted_at, + transport_failure=True, ) 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"}, + return self._record_fill_poll( + broker_order_id, + FillStatus(state="unknown", raw={"reason": "submission_lookup_rejected"}), + submitted_at=submitted_at, + transport_failure=True, ) matches = [ row @@ -640,19 +875,29 @@ class KISBroker: if self._matches_unknown_submission(row, metadata, submitted_at) ] if len(matches) != 1: - age = (datetime.now(tz=_KST) - submitted_at).total_seconds() + age = (self._clock().astimezone(_KST) - submitted_at).total_seconds() if not matches and age >= _SUBMISSION_LOOKUP_GRACE_SECONDS: - return FillStatus( - state="rejected", - raw={"reason": "submission_confirmed_absent"}, + return self._record_fill_poll( + broker_order_id, + FillStatus( + state="rejected", + raw={"reason": "submission_confirmed_absent"}, + ), + submitted_at=submitted_at, ) - return FillStatus( - state="unknown", - raw={ - "reason": ( - "submission_lookup_ambiguous" if matches else "submission_lookup_pending" - ) - }, + return self._record_fill_poll( + broker_order_id, + FillStatus( + state="unknown", + raw={ + "reason": ( + "submission_lookup_ambiguous" + if matches + else "submission_lookup_pending" + ) + }, + ), + submitted_at=submitted_at, ) row = matches[0] resolved_order_id = self._broker_order_id_from_row(row) @@ -662,12 +907,16 @@ class KISBroker: 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, - }, + return self._record_fill_poll( + broker_order_id, + replace( + status, + raw={ + **(status.raw if isinstance(status.raw, dict) else {}), + "resolved_broker_order_id": resolved_order_id, + }, + ), + submitted_at=submitted_at, ) @staticmethod diff --git a/trex_core/application/execution.py b/trex_core/application/execution.py index bc8d25c..dc74257 100644 --- a/trex_core/application/execution.py +++ b/trex_core/application/execution.py @@ -62,6 +62,24 @@ SlotEventEmitter = Callable[ ] # (event_type, slot, routing_strategy_id) -> None _MAX_DISPATCH_PER_TICK = 64 +_ORDER_POLL_WARNING_THRESHOLDS_SEC = (60, 5 * 60, 15 * 60, 60 * 60, 6 * 60 * 60) +_ORDER_POLL_REASON_ALLOWLIST = frozenset({ + "correction_child_present_unresolved", + "fill_inquiry_rejected", + "fill_inquiry_transport", + "order_not_visible_unresolved", + "parent_identity_ambiguous", + "parent_identity_mismatch", + "replacement_chain_ambiguous", + "replacement_chain_fill_ambiguous", + "replacement_chain_stable_zero_terminal", + "replacement_chain_terminal_cancel", + "replacement_live_unresolved", + "submission_lookup_ambiguous", + "submission_lookup_pending", + "submission_lookup_rejected", + "submission_lookup_transport", +}) @dataclass @@ -115,6 +133,7 @@ class ExecutionService: # 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._order_poll_warning_state: dict[str, tuple[str, str, int]] = {} self._clock = clock or (lambda: datetime.now(tz=UTC)) # dcy-041 / GitHub #45: pending-sell quantity may use a bounded # account snapshot only through this injected reader. Reconciliation @@ -282,6 +301,7 @@ class ExecutionService: self._emit_order_metadata_unknown_alarm(slot) continue fs = self._poll_fill(slot, side) + self._observe_order_poll(slot, side, fs) resolved_order_id = ( fs.raw.get("resolved_broker_order_id") if isinstance(fs.raw, dict) else None ) @@ -344,6 +364,55 @@ class ExecutionService: kwargs["submitted_at"] = slot.order_submitted_at return poll(slot.broker_order_id, **kwargs) + def _observe_order_poll(self, slot, side: str, fs) -> None: + """Emit sanitized poll telemetry and deduplicated unresolved warnings.""" + raw = fs.raw if isinstance(fs.raw, dict) else {} + if raw.get("poll_performed") is not True: + return + + def _metric_int(name: str) -> int: + try: + return max(0, int(raw.get(name, 0))) + except (TypeError, ValueError): + return 0 + + reason_value = raw.get("last_result_reason") + reason = ( + str(reason_value) + if reason_value in _ORDER_POLL_REASON_ALLOWLIST + else "none" + ) + poll_count = _metric_int("poll_count") + stuck_duration_sec = _metric_int("stuck_duration_sec") + try: + backoff_sec = max(0.0, float(raw.get("backoff_sec", 0.0))) + except (TypeError, ValueError): + backoff_sec = 0.0 + fields = { + "event": "order_poll", + "side": side, + "result_state": fs.state, + "result_reason": reason, + "poll_count": poll_count, + "stuck_duration_sec": stuck_duration_sec, + "backoff_sec": backoff_sec, + } + _log.info("order poll observed", extra=fields) + + slot_key = str(slot.slot_id) + if fs.state != "unknown": + self._order_poll_warning_state.pop(slot_key, None) + return + threshold_index = sum( + stuck_duration_sec >= threshold + for threshold in _ORDER_POLL_WARNING_THRESHOLDS_SEC + ) + warning_key = (fs.state, reason, threshold_index) + if self._order_poll_warning_state.get(slot_key) == warning_key: + return + self._order_poll_warning_state[slot_key] = warning_key + _log.warning("order poll unresolved", extra={**fields, "event": "order_poll_unresolved"}) + # ------------------------------------------------------------ def _balance_map(self) -> dict: """Current broker balance as {symbol: hldg_qty} (REQ-062 ground truth)."""