From f0ac909958ee91c1e68cbcdd21c026f590fea153 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 1 Sep 2026 03:33:38 +0000 Subject: [PATCH] fix(execution): cancel SQL nullable type error, KIS child cancel detection, worker exception logging Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- tests/adapters/broker/test_kis_cancel.py | 111 ++++++++++++++++++ tests/application/test_execution_cancel.py | 46 ++++++++ ...daemon_execution_loop_exception_logging.py | 53 +++++++++ trex_core/adapters/broker/kis.py | 19 +++ trex_core/adapters/persistence/postgresql.py | 39 ++++-- trex_core/runtime/daemon.py | 2 +- 6 files changed, 258 insertions(+), 12 deletions(-) create mode 100644 tests/runtime/test_daemon_execution_loop_exception_logging.py diff --git a/tests/adapters/broker/test_kis_cancel.py b/tests/adapters/broker/test_kis_cancel.py index b18d074..124e447 100644 --- a/tests/adapters/broker/test_kis_cancel.py +++ b/tests/adapters/broker/test_kis_cancel.py @@ -130,3 +130,114 @@ def test_full_fill_unaffected_by_cancel_flag(monkeypatch): f"06010:{ODNO}", side="sell", symbol="005930", order_date="20260608" ) assert fs.is_filled and fs.filled_qty == 10 and not fs.is_partial_cancel + + +# ── Regression: KIS child cancel row (orgn_odno) ───────────────────────────── + +def _payload_with_child_cancel(parent_odno: str, child_odno: str, ord_qty: str = "89"): + """Simulate KIS returning original row (cncl_yn='') + separate cancel child.""" + parent = { + "odno": parent_odno, + "orgn_odno": "", + "ord_qty": ord_qty, + "tot_ccld_qty": "0", + "avg_prvs": "0", + "cncl_yn": "", + "rjct_qty": "0", + "rmn_qty": "0", + "ord_dt": "20260901", + "ord_gno_brno": "06010", + } + child = { + "odno": child_odno, + "orgn_odno": parent_odno, + "ord_qty": ord_qty, + "tot_ccld_qty": "0", + "avg_prvs": "0", + "cncl_yn": "Y", + "rjct_qty": "0", + "rmn_qty": "0", + "ord_dt": "20260901", + "ord_gno_brno": "06010", + } + return {"rt_cd": "0", "output1": [parent, child]} + + +def test_child_cancel_row_orgn_odno_maps_to_cancelled(monkeypatch): + """KIS child row with orgn_odno= + cncl_yn=Y → full cancel.""" + payload = _payload_with_child_cancel(ODNO, "0000123457") + monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(payload)) + fs = _broker().poll_fill( + f"06010:{ODNO}", side="buy", symbol="225190", order_date="20260901" + ) + assert fs.state == "cancelled", f"expected cancelled, got {fs.state}" + assert fs.is_terminal_reject + + +def test_correction_child_row_not_misclassified_as_cancel(monkeypatch): + """Correction child (cncl_yn='N') must not be treated as a cancel.""" + parent = { + "odno": ODNO, + "orgn_odno": "", + "ord_qty": "10", + "tot_ccld_qty": "0", + "avg_prvs": "0", + "cncl_yn": "", + "rjct_qty": "0", + "ord_dt": "20260901", + "ord_gno_brno": "06010", + } + correction_child = { + "odno": "0000123458", + "orgn_odno": ODNO, + "ord_qty": "10", + "tot_ccld_qty": "0", + "avg_prvs": "0", + "cncl_yn": "N", # correction, not cancel + "rjct_qty": "0", + "ord_dt": "20260901", + "ord_gno_brno": "06010", + } + monkeypatch.setattr( + kis_mod.httpx, "get", _fake_get({"rt_cd": "0", "output1": [parent, correction_child]}) + ) + fs = _broker().poll_fill( + f"06010:{ODNO}", side="buy", symbol="225190", order_date="20260901" + ) + # Correction child must not infer cancellation; original has no fill evidence + assert fs.state == "accepted", f"expected accepted, got {fs.state}" + + +def test_partial_fill_with_child_cancel_row(monkeypatch): + """Partial fill on parent + cancel child → partial_cancel.""" + parent = { + "odno": ODNO, + "orgn_odno": "", + "ord_qty": "10", + "tot_ccld_qty": "4", + "avg_prvs": "70000", + "cncl_yn": "", + "rjct_qty": "0", + "ord_dt": "20260901", + "ord_gno_brno": "06010", + } + child = { + "odno": "0000123459", + "orgn_odno": ODNO, + "ord_qty": "6", + "tot_ccld_qty": "0", + "avg_prvs": "0", + "cncl_yn": "Y", + "rjct_qty": "0", + "ord_dt": "20260901", + "ord_gno_brno": "06010", + } + monkeypatch.setattr( + kis_mod.httpx, "get", _fake_get({"rt_cd": "0", "output1": [parent, child]}) + ) + fs = _broker().poll_fill( + f"06010:{ODNO}", side="sell", symbol="225190", order_date="20260901" + ) + assert fs.state == "partially_filled" + assert fs.cancelled and fs.is_partial_cancel + assert fs.filled_qty == 4 diff --git a/tests/application/test_execution_cancel.py b/tests/application/test_execution_cancel.py index 1fae52f..1b7a7f2 100644 --- a/tests/application/test_execution_cancel.py +++ b/tests/application/test_execution_cancel.py @@ -179,3 +179,49 @@ def test_operator_cancel_is_sent_once_across_repeat_and_restart(): assert restarted.cancel_pending_slot(sid) == "cancel_already_recorded" assert broker.cancel_calls == 1 assert repo.fetch_slot(sid).cancel_state == CancelState.REQUESTED.value + + +# ── Regression: first-cancel reservation (retry_before=None) ───────────────── + +def test_first_cancel_reservation_with_no_retry_before_succeeds(): + """First cancel reservation with retry_before=None must reserve exactly once (fix for + PostgreSQL IndeterminateDatatype; the in-memory repo mirrors the corrected SQL path).""" + broker = ProgrammableBroker() + repo, svc, intents, execution, _, _ = _wire(broker) + sid = _buy_to_holding(intents, execution) + intents.submit_sell(SellIntent(intent_id="s1", slot_id=sid)) + broker.responder = lambda side: FillStatus(state="accepted") + execution.run_once() # dispatch + + # First operator cancel — retry_before=None path + result = execution.cancel_pending_slot(sid) + assert result == "ok", f"expected ok, got {result}" + assert broker.cancel_calls == 1 + # Second call must be idempotent (cancel_state already set) + result2 = execution.cancel_pending_slot(sid) + assert result2 == "cancel_already_recorded" + assert broker.cancel_calls == 1 # no second broker call + + +def test_timeout_cancel_first_attempt_with_retry_before_none(): + """Timeout cancel's first attempt passes retry_before=None; must call broker once.""" + broker = ProgrammableBroker() + repo, svc, intents, execution, _, _ = _wire(broker) + sid = _buy_to_holding(intents, execution) + intents.submit_sell(SellIntent(intent_id="s1", slot_id=sid)) + broker.responder = lambda side: FillStatus( + state="accepted", + raw={"order_open_confirmed": True}, + ) + execution.run_once() # dispatch + + # Simulate timeout: set order_submitted_at well in the past + s = repo.fetch_slot(sid) + repo._slots[sid] = replace( + s, + order_submitted_at=datetime.now(tz=timezone.utc) - timedelta(seconds=120), + order_date="20260901", + timeout_sec=30, + ) + execution.run_once() # should trigger timeout cancel + assert broker.cancel_calls == 1 diff --git a/tests/runtime/test_daemon_execution_loop_exception_logging.py b/tests/runtime/test_daemon_execution_loop_exception_logging.py new file mode 100644 index 0000000..e20545b --- /dev/null +++ b/tests/runtime/test_daemon_execution_loop_exception_logging.py @@ -0,0 +1,53 @@ +"""Regression: execution loop exceptions must be logged, not silently swallowed. + +dcy-031 / issue #81: exceptions from execution.run_once() were silently ignored +via bare `continue`, making it impossible to observe repository or adapter errors. +""" +from __future__ import annotations + +import logging +import threading +import types + +import pytest + + +def test_execution_loop_logs_exception(caplog): + """When execution.run_once() raises, the error is logged and the loop continues.""" + import trex_core.runtime.daemon as daemon_mod + + call_count = 0 + stop_event = threading.Event() + + class _FakeExecution: + def run_once(self): + nonlocal call_count + call_count += 1 + if call_count == 1: + raise RuntimeError("simulated repo error: IndeterminateDatatype") + stop_event.set() + + class _FakePublisher: + def keepalive_tick(self): + pass + + class _FakeConfig: + execution_interval_sec = 0.001 + + # Build a minimal stub Daemon without invoking its real __init__, so we can + # inject only the attributes _execution_loop reads. + daemon = object.__new__(daemon_mod.Daemon) + daemon.stop = stop_event + daemon.config = _FakeConfig() + daemon.event_publisher = _FakePublisher() + daemon.execution = _FakeExecution() + + with caplog.at_level(logging.ERROR, logger="trex_core.runtime.daemon"): + t = threading.Thread(target=daemon._execution_loop, daemon=True) + t.start() + t.join(timeout=2.0) + + assert not t.is_alive(), "execution loop did not exit cleanly" + assert any( + "execution loop error" in r.message for r in caplog.records + ), f"expected 'execution loop error' log, got: {[r.message for r in caplog.records]}" diff --git a/trex_core/adapters/broker/kis.py b/trex_core/adapters/broker/kis.py index abb6982..7b8fd31 100644 --- a/trex_core/adapters/broker/kis.py +++ b/trex_core/adapters/broker/kis.py @@ -440,6 +440,25 @@ class KISBroker: "inquiry_date": inquiry_date, }, ) + # 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. Correction (정정) children will NOT have cncl_yn="Y" so + # they are not misclassified as cancels here. + child_cancel_row = next( + ( + 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" + ), + None, + ) + if child_cancel_row is not None and str(row.get("cncl_yn") or "").strip().upper() != "Y": + # 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 ) diff --git a/trex_core/adapters/persistence/postgresql.py b/trex_core/adapters/persistence/postgresql.py index cbe5154..6a52b3d 100644 --- a/trex_core/adapters/persistence/postgresql.py +++ b/trex_core/adapters/persistence/postgresql.py @@ -385,20 +385,37 @@ class PostgreSQLSlotRepository: *, retry_before: Optional[datetime] = None, ) -> bool: - """Atomically reserve first cancel or an authoritative backoff-gated retry.""" + """Atomically reserve first cancel or an authoritative backoff-gated retry. + + Two separate SQL paths avoid nullable-placeholder type-inference errors + (PostgreSQL IndeterminateDatatype when binding None to %s IS NOT NULL). + """ now = datetime.now(tz=timezone.utc) with self._psycopg.connect(self._dsn, autocommit=False) as conn: with conn.cursor() as cur: - cur.execute( - "UPDATE daily_slot SET cancel_state='unknown', cancel_requested_at=%s, " - "cancel_attempts=cancel_attempts+1, updated_at=%s " - "WHERE slot_id=%s AND broker_order_id=%s " - "AND status IN ('pending_buy','pending_sell') " - "AND (cancel_state IS NULL OR " - "(%s IS NOT NULL AND cancel_state IN ('rejected','unknown') " - "AND cancel_attempts < 4 AND cancel_requested_at <= %s))", - (now, now, slot_id, broker_order_id, retry_before, retry_before), - ) + if retry_before is None: + # First reservation: only match when no cancel has been + # attempted yet (cancel_state IS NULL). + cur.execute( + "UPDATE daily_slot SET cancel_state='unknown', cancel_requested_at=%s, " + "cancel_attempts=cancel_attempts+1, updated_at=%s " + "WHERE slot_id=%s AND broker_order_id=%s " + "AND status IN ('pending_buy','pending_sell') " + "AND cancel_state IS NULL", + (now, now, slot_id, broker_order_id), + ) + else: + # Retry: backoff-gated, only match retryable states. + cur.execute( + "UPDATE daily_slot SET cancel_state='unknown', cancel_requested_at=%s, " + "cancel_attempts=cancel_attempts+1, updated_at=%s " + "WHERE slot_id=%s AND broker_order_id=%s " + "AND status IN ('pending_buy','pending_sell') " + "AND cancel_state IN ('rejected','unknown') " + "AND cancel_attempts < 4 " + "AND cancel_requested_at <= %s", + (now, now, slot_id, broker_order_id, retry_before), + ) ok = cur.rowcount == 1 conn.commit() return ok diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index 9e8e61e..12f0132 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -409,7 +409,7 @@ class Daemon: try: self.execution.run_once() except Exception: # noqa: BLE001 - loop must survive transient errors - continue + _log.exception("execution loop error — worker continues") def _ledger_loop(self) -> None: while not self.stop.wait(self.config.ledger_interval_sec):