fix(broker): dcy-013 KIS 외부 취소 감지 + 부분 체결 회계 + stale 워치독 (BAT-017)

KIS 계좌에서 직접 매도 주문을 취소하면 poll_fill 이 cncl_yn 을 파싱하지 않아 슬롯이 pending_sell 에 영구히 갇히고 죽은 주문을 무한 폴링하던 버그(issue #9) 정정. 거버넌스: 전부 change-req (신규 REQ 없음, dcy-013 OQ-3). - REQ-012: KIS poll_fill cncl_yn 파싱 → 외부 취소 감지. FillStatus.cancelled /is_partial_cancel 추가. 체결 0 취소→cancelled, 부분 체결 후 취소→partial+cancelled - REQ-013: 집행 _on_partial_cancel — 체결분 원장 1회 + qty 차감(매도 잔여/매수 체결분) + holding 복귀 (OQ-1 A안). 전량 취소는 기존 롤백 경로 재사용 - REQ-018: 부분 체결분 원장 1회 기록 (_record_ledger 재사용) - REQ-031: 잔여 포지션 정합 연계 (slot.qty ↔ 실 잔고, 과매도 방지) - REQ-014: pending stale 워치독 Alert-only (TREX_PENDING_STALE_THRESHOLD, WARN+카운터, 상태 미변경) — 취소 감지 누락 backstop - REQ-015: interface-contract/runtime-flows 정합 전량 취소는 holding 자동 복구(limit_price clear→재매도 시장가, dcy-012 연계). KIS cncl_yn 실 규약은 운영 검증(미지 값 보수 처리 + 워치독 backstop). 검증: 단위 294 passed/32 skipped, 인프라 게이트(PG+RabbitMQ) 326 passed, ruff clean. 신규 test_kis_cancel(httpx stub) + test_execution_cancel(programmable broker). BAT-017 release-candidate. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-08 14:11:49 +09:00
parent cc28e1fba5
commit 4746f74aae
24 changed files with 1063 additions and 6 deletions
@@ -0,0 +1,79 @@
+"""KIS poll_fill external-cancel detection (dcy-013 / REQ-012, issue #9).
+
+inquire-daily-ccld marks a cancelled order with cncl_yn="Y". These tests stub
+the HTTP layer (token pre-seeded, httpx.get faked) and assert the FillStatus
+mapping for full cancel / partial-fill-then-cancel / not-cancelled.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+from trex_core.adapters.broker import kis as kis_mod
+from trex_core.adapters.broker.kis import KISBroker
+
+ODNO = "0000123456"
+
+
+def _broker():
+    b = KISBroker(
+        "12345678-01",
+        app_key="k", app_secret="s",
+        base_url="https://openapivts.koreainvestment.com:29443",  # mock endpoint
+    )
+    b._token = "tok"
+    b._token_exp = datetime.now(tz=timezone.utc) + timedelta(hours=1)
+    return b
+
+
+class _Resp:
+    def __init__(self, payload):
+        self._payload = payload
+
+    def json(self):
+        return self._payload
+
+
+def _fake_get(payload):
+    def _get(*a, **k):
+        return _Resp(payload)
+    return _get
+
+
+def _row(**kw):
+    base = {"odno": ODNO, "ord_qty": "10", "tot_ccld_qty": "0",
+            "avg_prvs": "0", "cncl_yn": "N", "ord_dt": "20260608", "ord_gno_brno": "06010"}
+    base.update(kw)
+    return {"output1": [base]}
+
+
+def test_full_external_cancel_maps_to_cancelled(monkeypatch):
+    monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(_row(tot_ccld_qty="0", cncl_yn="Y")))
+    fs = _broker().poll_fill(f"06010:{ODNO}", side="sell", symbol="005930")
+    assert fs.state == "cancelled" and fs.is_terminal_reject and not fs.is_partial_cancel
+
+
+def test_zero_fill_not_cancelled_stays_accepted(monkeypatch):
+    monkeypatch.setattr(kis_mod.httpx, "get", _fake_get(_row(tot_ccld_qty="0", cncl_yn="N")))
+    fs = _broker().poll_fill(f"06010:{ODNO}", side="sell", symbol="005930")
+    assert fs.state == "accepted" and not fs.is_terminal_reject
+
+
+def test_partial_fill_then_cancel_flags_partial_cancel(monkeypatch):
+    monkeypatch.setattr(
+        kis_mod.httpx, "get",
+        _fake_get(_row(ord_qty="10", tot_ccld_qty="4", avg_prvs="70000", cncl_yn="Y")),
+    )
+    fs = _broker().poll_fill(f"06010:{ODNO}", side="sell", symbol="005930")
+    assert fs.state == "partially_filled"
+    assert fs.cancelled and fs.is_partial_cancel
+    assert fs.filled_qty == 4 and fs.avg_price == 70000.0
+    assert fs.execution_id == "kis-daily-ccld:20260608:06010:0000123456:4"
+
+
+def test_full_fill_unaffected_by_cancel_flag(monkeypatch):
+    monkeypatch.setattr(
+        kis_mod.httpx, "get",
+        _fake_get(_row(ord_qty="10", tot_ccld_qty="10", avg_prvs="70000", cncl_yn="N")),
+    )
+    fs = _broker().poll_fill(f"06010:{ODNO}", side="sell", symbol="005930")
+    assert fs.is_filled and fs.filled_qty == 10 and not fs.is_partial_cancel
@@ -0,0 +1,152 @@
+"""External-cancel + stale-watchdog execution tests (dcy-013 / issue #9).
+
+A programmable broker lets each poll return a chosen FillStatus so we can drive
+full external cancel (-> rollback to holding) and partial-fill-then-cancel
+(-> ledger record + qty decrement + holding remainder), plus the Alert-only
+pending stale watchdog.
+"""
+from __future__ import annotations
+
+import logging
+from dataclasses import replace
+from datetime import date, datetime, timedelta, timezone
+
+from trex_core.adapters.ledger.in_memory import InMemoryLedger
+from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore
+from trex_core.application.broker_port import FillStatus, OrderAck
+from trex_core.application.execution import ExecutionService
+from trex_core.application.intents import BuyIntent, IntentService, SellIntent
+from trex_core.application.ledger_outbox import InMemoryLedgerOutbox, LedgerWriter
+from trex_core.domain.slot import SlotService, SlotState
+from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository
+
+DAY = date(2026, 6, 4)
+
+
+class ProgrammableBroker:
+    """Broker double whose poll_fill response is chosen per call by `responder`."""
+
+    def __init__(self):
+        self._seq = 0
+        # default: full fill (used to drive a buy to holding)
+        self.responder = lambda side: FillStatus(
+            state="filled", filled_qty=10, avg_price=1000.0,
+            execution_id="exec-full", fee=1.0, tax=0.0,
+        )
+        self.polls = 0
+
+    def name(self):
+        return "stub"
+
+    def submit_order(self, req):
+        self._seq += 1
+        return OrderAck(accepted=True, broker_order_id=f"o{self._seq}")
+
+    def poll_fill(self, broker_order_id, *, side, symbol):
+        self.polls += 1
+        return self.responder(side)
+
+    def cancel_order(self, broker_order_id, *, side, symbol, qty):
+        return True
+
+    def get_balance(self):
+        return []
+
+    def get_current_price(self, symbol):
+        return None
+
+
+def _wire(broker, *, stale_sec=None):
+    repo = InMemorySlotRepository()
+    svc = SlotService(repo)
+    intents = IntentService(svc, repo, InMemoryIntentStore())
+    outbox = InMemoryLedgerOutbox()
+    ledger = InMemoryLedger()
+    execution = ExecutionService(
+        svc, repo, broker, ledger_outbox=outbox,
+        stale_pending_threshold_sec=stale_sec,
+    )
+    writer = LedgerWriter(outbox, ledger)
+    svc.preallocate_daily_slots(DAY, 5)
+    return repo, svc, intents, execution, ledger, writer
+
+
+def _buy_to_holding(intents, execution):
+    sid = intents.submit_buy(BuyIntent(
+        intent_id="b1", trading_date=DAY, symbol="005930", qty=10,
+        entry_price=1000, portfolio_id="PF-1",
+    )).slot_id
+    execution.run_once()  # buy dispatched + full fill -> holding
+    return sid
+
+
+def test_full_external_cancel_returns_slot_to_holding():
+    broker = ProgrammableBroker()
+    repo, svc, intents, execution, ledger, writer = _wire(broker)
+    sid = _buy_to_holding(intents, execution)
+    assert repo.fetch_slot(sid).status is SlotState.HOLDING
+
+    intents.submit_sell(SellIntent(intent_id="s1", slot_id=sid))
+    broker.responder = lambda side: FillStatus(state="cancelled")  # external cancel
+    r = execution.run_once()  # dispatch sell + poll(cancelled) -> rollback
+    slot = repo.fetch_slot(sid)
+    assert slot.status is SlotState.HOLDING       # recovered, not stuck pending_sell
+    assert slot.broker_order_id is None           # dispatch cleared -> re-sellable
+    assert r.rolled_back == 1
+
+
+def test_partial_fill_then_cancel_records_and_keeps_remainder():
+    broker = ProgrammableBroker()
+    repo, svc, intents, execution, ledger, writer = _wire(broker)
+    sid = _buy_to_holding(intents, execution)
+    writer.run_once()  # drain the buy ledger item
+
+    intents.submit_sell(SellIntent(intent_id="s1", slot_id=sid))
+    # 4 of 10 sold, then externally cancelled
+    broker.responder = lambda side: FillStatus(
+        state="partially_filled", filled_qty=4, avg_price=1100.0,
+        execution_id="exec-partial", fee=0.5, tax=0.0, cancelled=True,
+    )
+    r = execution.run_once()
+    slot = repo.fetch_slot(sid)
+    assert slot.status is SlotState.HOLDING        # remainder stays held
+    assert slot.qty == 6                            # 10 - 4 sold
+    assert slot.avg_price == 1000.0                 # cost basis preserved (not sell price)
+    assert slot.broker_order_id is None
+    assert r.filled == 1 and r.ledger_enqueued == 1
+
+    writer.run_once()
+    sells = [t for t in ledger.trades() if t.side == "sell"]
+    assert len(sells) == 1 and sells[0].qty == 4 and sells[0].price == 1100.0
+
+
+def test_stale_pending_watchdog_warns_without_state_change(caplog):
+    broker = ProgrammableBroker()
+    repo, svc, intents, execution, ledger, writer = _wire(broker, stale_sec=60)
+    sid = _buy_to_holding(intents, execution)
+    intents.submit_sell(SellIntent(intent_id="s1", slot_id=sid))
+    # order sits accepted (never fills)
+    broker.responder = lambda side: FillStatus(state="accepted")
+    execution.run_once()  # dispatch sell, poll -> accepted (fresh, not stale yet)
+    # backdate the dispatched slot's updated_at beyond the threshold
+    s = repo.fetch_slot(sid)
+    repo._slots[sid] = replace(s, updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=120))
+
+    with caplog.at_level(logging.WARNING):
+        r = execution.run_once()
+    assert r.stale_pending == 1
+    assert repo.fetch_slot(sid).status is SlotState.PENDING_SELL  # state unchanged
+    assert any("pending order stale" in rec.message for rec in caplog.records)
+
+
+def test_stale_watchdog_disabled_by_default():
+    broker = ProgrammableBroker()
+    repo, svc, intents, execution, ledger, writer = _wire(broker)  # no threshold
+    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()
+    s = repo.fetch_slot(sid)
+    repo._slots[sid] = replace(s, updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=9999))
+    r = execution.run_once()
+    assert r.stale_pending == 0  # disabled => never warns