fix: converge broker failures without retry

This commit is contained in:
Hermes Agent
2026-08-19 11:29:26 +09:00
parent be98965f2c
commit 15aeae6930
15 changed files with 181 additions and 211 deletions
+1 -1
View File
@@ -49,7 +49,7 @@
- limit_price :: numeric? :: 현재 pending 주문 지정가(side 무관, dcy-012/REQ-002 — 과거 `entry_price`에서 rename). pending 진입(매수/매도 지정가) 시 set, pending 이탈(→holding/→closed) 시 None 으로 clear → 매수 지정가가 매도로 누수되지 않음(issue #10). None 이면 dispatch 시 시장가. - limit_price :: numeric? :: 현재 pending 주문 지정가(side 무관, dcy-012/REQ-002 — 과거 `entry_price`에서 rename). pending 진입(매수/매도 지정가) 시 set, pending 이탈(→holding/→closed) 시 None 으로 clear → 매수 지정가가 매도로 누수되지 않음(issue #10). None 이면 dispatch 시 시장가.
- allocated_capital :: numeric? :: 배정 자본 - allocated_capital :: numeric? :: 배정 자본
- order_id / broker_order_id :: str? :: 주문 식별자(자유 주석 / 브로커 핸들) - order_id / broker_order_id :: str? :: 주문 식별자(자유 주석 / 브로커 핸들)
- venue :: TEXT NOT NULL DEFAULT 'KRX' :: 현재 주문 거래소 구분(AUTO/KRX/NXT/SOR). intent 단계의 AUTO는 KIS 제출 시 실제 KRX/NXT로 교체되며 주문번호와 원자 저장한다. pending 주문·취소까지 보존하고 HOLDING 진입 시 다음 자율주문의 AUTO, INACTIVE 진입 시 KRX로 초기화한다. - venue :: TEXT NOT NULL DEFAULT 'KRX' :: 현재 주문 거래소 구분(KRX/NXT/SOR). venue 누락 intent는 SOR로 저장하고, 명시 venue는 보존한다. pending 주문·취소까지 보존하며 HOLDING 진입 시 다음 주문의 SOR, INACTIVE 진입 시 KRX로 초기화한다.
- cancel_state :: TEXT? :: 취소 요청의 즉시 결과(requested/rejected/unknown). 최종 취소 상태가 아니다. requested는 재시도하지 않고, rejected/unknown만 authoritative open-row 확인과 bounded backoff 뒤 재시도할 수 있다. - cancel_state :: TEXT? :: 취소 요청의 즉시 결과(requested/rejected/unknown). 최종 취소 상태가 아니다. requested는 재시도하지 않고, rejected/unknown만 authoritative open-row 확인과 bounded backoff 뒤 재시도할 수 있다.
- cancel_requested_at :: TIMESTAMPTZ? :: 마지막 취소 시도 예약 시각. 주문 종료·dispatch clear 때 함께 제거한다. - cancel_requested_at :: TIMESTAMPTZ? :: 마지막 취소 시도 예약 시각. 주문 종료·dispatch clear 때 함께 제거한다.
- order_submitted_at :: TIMESTAMPTZ? :: timeout deadline의 immutable 주문 제출 기준점. poll/cancel bookkeeping 및 daemon 재시작에서 보존하고 terminal dispatch clear 때 제거한다. - order_submitted_at :: TIMESTAMPTZ? :: timeout deadline의 immutable 주문 제출 기준점. poll/cancel bookkeeping 및 daemon 재시작에서 보존하고 terminal dispatch clear 때 제거한다.
+4 -4
View File
@@ -64,7 +64,7 @@
| 패턴 | 표면 | 요청 속성 | 응답 | | 패턴 | 표면 | 요청 속성 | 응답 |
|---|---|---|---| |---|---|---|---|
| 슬롯 RPC | `trex.slots.req` | `correlation_id`, `reply_to=amq.rabbitmq.reply-to` | reply-to 큐로 `{slots[], degraded, reason, message}` (REQ-073/074, BAT-035) | | 슬롯 RPC | `trex.slots.req` | `correlation_id`, `reply_to=amq.rabbitmq.reply-to` | reply-to 큐로 `{slots[], degraded, reason, message}` (REQ-073/074, BAT-035) |
| intent RPC | `trex.intent.req` | `correlation_id` = intent 멱등키, `reply_to=amq.rabbitmq.reply-to`, `delivery_mode=2` | reply-to 큐로 `{intent_id, status:admitted|rejected, slot_id?, reason?}`. **rejected sell** 응답은 additive `retryable: bool` 포함(REQ-007 AC-7/issue #27 — core 소유 reason→retryable, 미인식 기본 true, `bad_request:*` false; admitted/processing/buy 는 미포함). 체결/종료는 응답 아닌 이벤트로 | | intent RPC | `trex.intent.req` | `correlation_id` = intent 멱등키, `reply_to=amq.rabbitmq.reply-to`, `delivery_mode=2` | reply-to 큐로 `{intent_id, status:admitted|rejected, slot_id?, reason?}`. retryability·broker attempt·KIS failure 정보는 strategy 응답에 포함하지 않는다. 체결/종료는 응답 아닌 이벤트로 |
| update_thresholds | `trex.intent.req` | `action="update_thresholds"`, `reply_to` | reply-to 큐로 `{status:updated|ignored, slot_id, reason?}` (REQ-040) | | update_thresholds | `trex.intent.req` | `action="update_thresholds"`, `reply_to` | reply-to 큐로 `{status:updated|ignored, slot_id, reason?}` (REQ-040) |
| 이벤트 pub/sub | `trex.events` (topic) | routing key `{event_type}.{strategy_id}.{symbol}` | 전략 큐가 자기 `strategy_id` 패턴으로 바인딩 | | 이벤트 pub/sub | `trex.events` (topic) | routing key `{event_type}.{strategy_id}.{symbol}` | 전략 큐가 자기 `strategy_id` 패턴으로 바인딩 |
@@ -415,8 +415,8 @@ resume_kill_switch 요청·응답 (REQ-046):
### Additive Notes (REQ-117) ### Additive Notes (REQ-117)
- `venue` 필드는 BAT-056에서 additive 추가. 기존 orb/tsl 미지정 호출 → 기본값 `"KRX"` 자동 적용 → 하위 호환. - `venue` 필드는 BAT-056에서 additive 추가. 기존 orb/tsl 미지정 호출 → 기본값 `"KRX"` 자동 적용 → 하위 호환.
- `OrderRequest`는 AMQP wire 위를 직접 흐르지 않는다. buy/sell intent의 선택적 `venue` 기본은 `AUTO`이고, 명시한 KRX/NXT/SOR는 그대로 슬롯에 보존한다. 허용값 외에는 `invalid_venue`로 admission 전 거부한다. - `OrderRequest`는 AMQP wire 위를 직접 흐르지 않는다. buy/sell intent의 선택적 `venue` 누락은 `SOR`이고, 명시한 KRX/NXT/SOR는 그대로 슬롯에 보존한다. `AUTO`와 그 밖의 값은 `invalid_venue`로 admission 전 거부한다.
- KIS 어댑터는 `AUTO`를 KST 세션 기준 실제 KRX/NXT로 확정한 뒤 `EXCG_ID_DVSN_CD`로 전송한다. 결정 불가능 구간은 KRX로 대체하지 않고 주문을 거부한다. - KIS 어댑터는 KRX/NXT/SOR를 시간과 무관하게 보존해 `EXCG_ID_DVSN_CD`로 전송한다.
- `OrderAck.venue`의 실제 제출 거래소는 `broker_order_id`와 함께 슬롯에 원자 저장되며 재시작·취소·Core 원장 outbox에서 재사용한다. - `OrderAck.venue`의 실제 제출 거래소는 `broker_order_id`와 함께 슬롯에 원자 저장되며 재시작·취소·Core 원장 outbox에서 재사용한다.
## `BrokerPort` Protocol 메서드 (BAT-056 / REQ-118) ## `BrokerPort` Protocol 메서드 (BAT-056 / REQ-118)
@@ -461,7 +461,7 @@ get_current_price(symbol: str, market: str = "J") -> Optional[Decimal]
### KIS 어댑터 EXCG 명시화 (REQ-118) ### KIS 어댑터 EXCG 명시화 (REQ-118)
- `submit_order`: body에 `EXCG_ID_DVSN_CD: req.venue` 항상 명시 전송 (이전: 미전송·서버 기본값 의존 → 명시화로 견고성 향상). 실측(2026-07-06): NXT(EXCG=NXT) 주문 rt_cd=0 확인. - `submit_order`: body에 `EXCG_ID_DVSN_CD: req.venue` 항상 명시 전송 (이전: 미전송·서버 기본값 의존 → 명시화로 견고성 향상). 실측(2026-07-06): NXT(EXCG=NXT) 주문 rt_cd=0 확인.
- `AUTO` 세션: 08:00~08:49:59 NXT, 09:00:30~15:29:59 KRX, 15:30~20:00 NXT. 그 밖의 시간은 `auto_venue_session_closed`로 fail-closed 한다. - venue 누락은 `SOR`로 제출하며, 명시 KRX/NXT/SOR는 그대로 제출한다. core는 wall-clock으로 venue를 선택하지 않는다.
- `cancel_order`: body에 `EXCG_ID_DVSN_CD: venue` 명시 전송. 실측(2026-07-06): NXT 취소 rt_cd=0 확인. - `cancel_order`: body에 `EXCG_ID_DVSN_CD: venue` 명시 전송. 실측(2026-07-06): NXT 취소 rt_cd=0 확인.
- `poll_fill`: KIS 신형 일별주문체결조회 실전 `TTTC0081R`, 모의 - `poll_fill`: KIS 신형 일별주문체결조회 실전 `TTTC0081R`, 모의
`VTTC0081R`을 사용하고 `EXCG_ID_DVSN_CD=ALL`을 전달한다. 일반 주문번호 `VTTC0081R`을 사용하고 `EXCG_ID_DVSN_CD=ALL`을 전달한다. 일반 주문번호
+1 -1
View File
@@ -264,7 +264,7 @@
- 슬롯 `limit_price` 생애주기(dcy-012/REQ-002): INACTIVE→PENDING_BUY(매수 지정가 set) → PENDING_BUY→HOLDING(clear) → HOLDING→PENDING_SELL(매도 지정가 set 또는 None=시장가) → PENDING_SELL→INACTIVE(clear). 롤백(→INACTIVE) 전이에서도 clear. 외부 취소 후 holding 복귀에서도 clear 되어 재매도는 시장가. - 슬롯 `limit_price` 생애주기(dcy-012/REQ-002): INACTIVE→PENDING_BUY(매수 지정가 set) → PENDING_BUY→HOLDING(clear) → HOLDING→PENDING_SELL(매도 지정가 set 또는 None=시장가) → PENDING_SELL→INACTIVE(clear). 롤백(→INACTIVE) 전이에서도 clear. 외부 취소 후 holding 복귀에서도 clear 되어 재매도는 시장가.
- **실행 정책 생애주기(dcy-022/REQ-068)**: `timeout_sec`·`slippage_pct` 는 `limit_price` 와 동일한 생애주기(PENDING_* 진입 시 set, HOLDING/INACTIVE 탈출 시 clear). 슬롯의 `updated_at` 이 dispatch 시각 proxy 로 사용된다(mark_dispatched 가 갱신, stale-alert 와 동일 방법). - **실행 정책 생애주기(dcy-022/REQ-068)**: `timeout_sec`·`slippage_pct` 는 `limit_price` 와 동일한 생애주기(PENDING_* 진입 시 set, HOLDING/INACTIVE 탈출 시 clear). 슬롯의 `updated_at` 이 dispatch 시각 proxy 로 사용된다(mark_dispatched 가 갱신, stale-alert 와 동일 방법).
- 어댑터: paper(결정적 async 더블) / KIS(모의투자 검증, 실계좌 게이팅). - 어댑터: paper(결정적 async 더블) / KIS(모의투자 검증, 실계좌 게이팅).
- **venue 전달 경로(BAT-056/REQ-117/118, BAT-059/REQ-124, dcy-041)**: buy/sell intent `venue`(기본 AUTO) → pending `Slot.venue` → KIS 제출 시 KST 세션별 실제 KRX/NXT 확정 → `OrderAck.venue`와 주문번호 원자 저장 → 취소·재시작·Core outbox에서 실제 거래소 재사용. 허용값은 AUTO/KRX/NXT/SOR이며 invalid는 슬롯 전이 전 거부한다. HOLDING은 다음 자율매도를 위해 AUTO, INACTIVE는 하위 호환을 위해 KRX로 초기화한다. - **venue 전달 경로(BAT-056/REQ-117/118, BAT-059/REQ-124, dcy-041, GitHub #61)**: buy/sell intent `venue`(누락 시 SOR) → pending `Slot.venue` → KIS에 명시 KRX/NXT/SOR 제출 → `OrderAck.venue`와 주문번호 원자 저장 → 취소·재시작·Core outbox에서 실제 거래소 재사용. AUTO와 invalid 값은 슬롯 전이 전 거부한다. HOLDING은 다음 주문을 위해 SOR, INACTIVE는 하위 호환을 위해 KRX로 초기화한다.
### 계좌조회 단기 복구 (dcy-041 / GitHub #45) ### 계좌조회 단기 복구 (dcy-041 / GitHub #45)
+16 -48
View File
@@ -91,38 +91,28 @@ def test_order_request_venue_nxt():
assert req.venue == "NXT" assert req.venue == "NXT"
@pytest.mark.parametrize( @pytest.mark.parametrize("venue", ["KRX", "NXT", "SOR"])
("at", "expected"),
[
(datetime(2026, 7, 29, 8, 0, tzinfo=timezone(timedelta(hours=9))), "NXT"),
(datetime(2026, 7, 29, 8, 49, 59, tzinfo=timezone(timedelta(hours=9))), "NXT"),
(datetime(2026, 7, 29, 9, 0, 30, tzinfo=timezone(timedelta(hours=9))), "KRX"),
(datetime(2026, 7, 29, 15, 29, 59, tzinfo=timezone(timedelta(hours=9))), "KRX"),
(datetime(2026, 7, 29, 15, 30, tzinfo=timezone(timedelta(hours=9))), "NXT"),
(datetime(2026, 7, 29, 20, 0, tzinfo=timezone(timedelta(hours=9))), "NXT"),
],
)
def test_auto_venue_resolves_by_kst_session(at, expected):
assert resolve_order_venue("AUTO", at) == expected
@pytest.mark.parametrize( @pytest.mark.parametrize(
"at", "at",
[ [
datetime(2026, 7, 29, 7, 59, 59, tzinfo=timezone(timedelta(hours=9))), datetime(2026, 7, 29, 8, 0, tzinfo=timezone(timedelta(hours=9))),
datetime(2026, 7, 29, 8, 50, tzinfo=timezone(timedelta(hours=9))), datetime(2026, 7, 29, 8, 55, tzinfo=timezone(timedelta(hours=9))),
datetime(2026, 7, 29, 9, 0, 29, tzinfo=timezone(timedelta(hours=9))),
datetime(2026, 7, 29, 20, 0, 1, tzinfo=timezone(timedelta(hours=9))), datetime(2026, 7, 29, 20, 0, 1, tzinfo=timezone(timedelta(hours=9))),
], ],
) )
def test_auto_venue_fails_closed_outside_supported_sessions(at): def test_explicit_venue_is_preserved_independent_of_time(venue, at):
with pytest.raises(ValueError, match="auto_venue_session_closed"): assert resolve_order_venue(venue, at) == venue
def test_auto_venue_is_rejected():
at = datetime(2026, 7, 29, 8, 1, tzinfo=timezone(timedelta(hours=9)))
with pytest.raises(ValueError, match="unsupported_order_venue"):
resolve_order_venue("AUTO", at) resolve_order_venue("AUTO", at)
def test_submit_auto_uses_nxt_and_reports_actual_venue(monkeypatch): def test_submit_sor_preserves_sor(monkeypatch):
captured = {} captured = {}
at = datetime(2026, 7, 29, 8, 1, tzinfo=timezone(timedelta(hours=9))) at = datetime(2026, 7, 29, 8, 55, tzinfo=timezone(timedelta(hours=9)))
def fake_post(url, headers=None, json=None, timeout=None): def fake_post(url, headers=None, json=None, timeout=None):
captured["body"] = json captured["body"] = json
@@ -134,39 +124,17 @@ def test_submit_auto_uses_nxt_and_reports_actual_venue(monkeypatch):
monkeypatch.setattr(kis_mod.httpx, "post", fake_post) monkeypatch.setattr(kis_mod.httpx, "post", fake_post)
ack = broker.submit_order( ack = broker.submit_order(
OrderRequest( OrderRequest(
slot_id="s-auto", slot_id="s-sor",
side="buy", side="buy",
symbol="005930", symbol="005930",
qty=1, qty=1,
venue="AUTO", venue="SOR",
) )
) )
assert ack.accepted is True assert ack.accepted is True
assert ack.venue == "NXT" assert ack.venue == "SOR"
assert captured["body"]["EXCG_ID_DVSN_CD"] == "NXT" assert captured["body"]["EXCG_ID_DVSN_CD"] == "SOR"
def test_submit_auto_outside_session_rejects_without_http(monkeypatch):
at = datetime(2026, 7, 29, 8, 55, tzinfo=timezone(timedelta(hours=9)))
broker = _mock_broker(clock=lambda: at)
def unexpected_post(*args, **kwargs):
raise AssertionError("AUTO session rejection must happen before HTTP")
monkeypatch.setattr(kis_mod.httpx, "post", unexpected_post)
ack = broker.submit_order(
OrderRequest(
slot_id="s-auto-closed",
side="buy",
symbol="005930",
qty=1,
venue="AUTO",
)
)
assert ack.accepted is False
assert ack.reason == "auto_venue_session_closed"
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -7,10 +7,11 @@ iteration: EXPECTED_SLOT_FIELDS (code contract) vs daily_slot actual columns
from __future__ import annotations from __future__ import annotations
import psycopg import psycopg
import pytest
from trex_core.adapters.persistence.migrate import apply_migrations, migration_files from trex_core.adapters.persistence.migrate import apply_migrations, migration_files
from trex_core.adapters.persistence.postgresql import PostgreSQLSlotRepository from trex_core.adapters.persistence.postgresql import PostgreSQLSlotRepository
from trex_core.domain.slot import EXPECTED_SLOT_FIELDS, SlotService, SlotState from trex_core.domain.slot import EXPECTED_SLOT_FIELDS
_EXPECTED_TABLES = { _EXPECTED_TABLES = {
"daily_slot", "daily_slot",
@@ -64,20 +65,29 @@ def test_apply_migrations_is_idempotent(pg_dsn):
apply_migrations(pg_dsn) apply_migrations(pg_dsn)
def test_migration_replay_accepts_existing_auto_venue_row(pg_clean): def test_migration_replay_converts_existing_auto_venue_row_to_sor(pg_clean):
repo = PostgreSQLSlotRepository(pg_clean) repo = PostgreSQLSlotRepository(pg_clean)
service = SlotService(repo, instance_key="migration-auto")
repo.reconcile_universe("migration-auto", {"pf-auto": "005930"}) repo.reconcile_universe("migration-auto", {"pf-auto": "005930"})
slot = repo.find_slot_by_portfolio("migration-auto", "pf-auto") slot = repo.find_slot_by_portfolio("migration-auto", "pf-auto")
service.transit_slot( assert slot is not None
slot.slot_id,
SlotState.INACTIVE, # Simulate a row written before migration 023 tightened the constraint.
SlotState.PENDING_BUY, with psycopg.connect(pg_clean) as conn:
reason="migration_replay", with conn.cursor() as cur:
actor="test", cur.execute("ALTER TABLE daily_slot DROP CONSTRAINT daily_slot_venue_check")
venue="AUTO", cur.execute(
) "UPDATE daily_slot SET venue = 'AUTO' WHERE slot_id = %s",
(slot.slot_id,),
)
conn.commit()
apply_migrations(pg_clean) apply_migrations(pg_clean)
assert repo.fetch_slot(slot.slot_id).venue == "AUTO" assert repo.fetch_slot(slot.slot_id).venue == "SOR"
with psycopg.connect(pg_clean) as conn:
with conn.cursor() as cur:
with pytest.raises(psycopg.errors.CheckViolation):
cur.execute(
"UPDATE daily_slot SET venue = 'AUTO' WHERE slot_id = %s",
(slot.slot_id,),
)
+1 -1
View File
@@ -59,7 +59,7 @@ def test_buy_fill_enqueues_one_ledger_item_and_writer_records_it():
assert (t.asset_id, t.side, t.qty, t.price) == ("PF-1", "buy", 10, 1000.0) assert (t.asset_id, t.side, t.qty, t.price) == ("PF-1", "buy", 10, 1000.0)
assert t.execution_id.startswith("paper:PAPER-") assert t.execution_id.startswith("paper:PAPER-")
assert t.fee > 0.0 and t.tax == 0.0 assert t.fee > 0.0 and t.tax == 0.0
assert t.venue == "AUTO" assert t.venue == "SOR"
def test_sell_fill_enqueues_sell_ledger_item(): def test_sell_fill_enqueues_sell_ledger_item():
@@ -253,6 +253,33 @@ class TestOrderResultLogging:
assert all(r.levelno == logging.ERROR for r in rejects) assert all(r.levelno == logging.ERROR for r in rejects)
assert "호가단위" in rejects[0].getMessage() assert "호가단위" in rejects[0].getMessage()
def test_sell_reject_returns_to_holding_without_new_attempt(self):
broker = _RecordingBroker()
repo, _, intents, execution = _wire(broker)
slot_id = _buy_to_pending(intents)
assert slot_id is not None
execution.run_once()
holding = repo.fetch_slot(slot_id)
assert holding is not None and holding.status is SlotState.HOLDING
intents.submit_sell(SellIntent(intent_id="s1", slot_id=slot_id))
broker.submit_order = lambda req: (
broker.orders.append(req)
or OrderAck(accepted=False, reason="broker_rejected", venue=req.venue)
)
result = execution.run_once()
recovered = repo.fetch_slot(slot_id)
assert recovered is not None
assert recovered.status is SlotState.HOLDING
assert recovered.qty == 10
assert recovered.broker_order_id is None
assert result.rolled_back == 1
assert len([order for order in broker.orders if order.side == "sell"]) == 1
execution.run_once()
assert len([order for order in broker.orders if order.side == "sell"]) == 1
def test_normal_fill_emits_no_error(self, caplog): def test_normal_fill_emits_no_error(self, caplog):
broker = _RecordingBroker() broker = _RecordingBroker()
repo, svc, intents, execution = _wire(broker) repo, svc, intents, execution = _wire(broker)
+66 -74
View File
@@ -1,18 +1,12 @@
"""rejected SellIntent `retryable` flag (issue #27 / REQ-007 CHG-20260615-01). """Strategy RPC responses never expose broker retryability (GitHub #60)."""
Additive: only on intent_type="sell" + status="rejected". Maps reason→retryable
via the core-owned table (unrecognized → True, safety default; bad_request → False).
admitted / processing / buy responses must NOT carry the field.
"""
from __future__ import annotations from __future__ import annotations
from uuid import uuid4
from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore
from trex_core.application.intents import IntentService from trex_core.application.intents import IntentService
from trex_core.application.messaging import RpcRequest from trex_core.application.messaging import RpcRequest
from trex_core.application.messaging_handlers import ( from trex_core.application.messaging_handlers import make_intent_rpc_handler
_sell_reject_retryable,
make_intent_rpc_handler,
)
from trex_core.domain.slot import SlotService, SlotState from trex_core.domain.slot import SlotService, SlotState
from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository
@@ -22,9 +16,14 @@ INST = "default"
def _wire(): def _wire():
repo = InMemorySlotRepository() repo = InMemorySlotRepository()
svc = SlotService(repo, instance_key=INST) svc = SlotService(repo, instance_key=INST)
intents = IntentService(svc, repo, InMemoryIntentStore(), intents = IntentService(
resolve_symbol=lambda pid: "005930", svc,
instance_key=INST, max_active_slots=5) repo,
InMemoryIntentStore(),
resolve_symbol=lambda pid: "005930",
instance_key=INST,
max_active_slots=5,
)
repo.reconcile_universe(INST, {"PF-1": "SYM"}) repo.reconcile_universe(INST, {"PF-1": "SYM"})
handler = make_intent_rpc_handler(intents) handler = make_intent_rpc_handler(intents)
slot = repo.list_slots(INST)[0] slot = repo.list_slots(INST)[0]
@@ -32,72 +31,65 @@ def _wire():
def _sell(handler, slot_id, intent_id="s1"): def _sell(handler, slot_id, intent_id="s1"):
return handler(RpcRequest(payload={ return handler(
"intent_type": "sell", "intent_id": intent_id, "slot_id": str(slot_id), RpcRequest(
}, correlation_id=intent_id)) payload={
"intent_type": "sell",
"intent_id": intent_id,
"slot_id": str(slot_id),
},
correlation_id=intent_id,
)
)
class TestRetryableMapping: def test_rejected_sell_does_not_expose_retryability():
def test_each_reason_maps(self): _, _, _, handler, slot = _wire()
assert _sell_reject_retryable("slot_state_changed") is True response = _sell(handler, slot.slot_id)
assert _sell_reject_retryable("slot_not_holding:pending_buy") is True assert response.payload["status"] == "rejected"
assert _sell_reject_retryable("duplicate_in_flight") is True assert response.payload["reason"] == "slot_not_holding:inactive"
assert _sell_reject_retryable("slot_not_found") is False assert "retryable" not in response.payload
assert _sell_reject_retryable("slot_not_holding:inactive") is False
assert _sell_reject_retryable("slot_not_holding:pending_sell") is False
assert _sell_reject_retryable("slot_not_holding:error") is False
def test_unrecognized_defaults_true(self):
assert _sell_reject_retryable("some_future_reason") is True
assert _sell_reject_retryable(None) is True
def test_bad_request_is_false(self):
assert _sell_reject_retryable("bad_request:KeyError('slot_id')") is False
class TestResponseField: def test_unknown_slot_sell_does_not_expose_retryability():
def test_rejected_sell_inactive_slot_has_retryable_false(self): _, _, _, handler, _ = _wire()
# slot is inactive (never bought) → slot_not_holding:inactive → terminal response = _sell(handler, uuid4())
repo, svc, intents, handler, slot = _wire() assert response.payload["status"] == "rejected"
resp = _sell(handler, slot.slot_id) assert response.payload["reason"] == "slot_not_found"
assert resp.payload["status"] == "rejected" assert "retryable" not in response.payload
assert resp.payload["reason"] == "slot_not_holding:inactive"
assert resp.payload["retryable"] is False
def test_rejected_sell_unknown_slot_false(self):
from uuid import uuid4
_, _, _, handler, _ = _wire()
resp = _sell(handler, uuid4())
assert resp.payload["status"] == "rejected"
assert resp.payload["reason"] == "slot_not_found"
assert resp.payload["retryable"] is False
def test_bad_request_sell_carries_retryable_false(self): def test_malformed_sell_does_not_expose_retryability():
_, _, _, handler, _ = _wire() _, _, _, handler, _ = _wire()
resp = handler(RpcRequest(payload={ response = handler(
"intent_type": "sell", "intent_id": "x", # missing slot_id RpcRequest(
}, correlation_id="x")) payload={"intent_type": "sell", "intent_id": "x"},
assert resp.payload["status"] == "rejected" correlation_id="x",
assert resp.payload["reason"].startswith("bad_request") )
assert resp.payload["retryable"] is False )
assert response.payload["status"] == "rejected"
assert response.payload["reason"].startswith("bad_request")
assert "retryable" not in response.payload
def test_admitted_sell_has_no_retryable(self):
repo, svc, intents, handler, slot = _wire()
# bring slot to holding so the sell is admitted
svc.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY,
reason="buy", actor="t")
svc.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING,
reason="filled", actor="t", qty=10, avg_price=100)
resp = _sell(handler, slot.slot_id)
assert resp.payload["status"] == "admitted"
assert "retryable" not in resp.payload
def test_buy_rejection_has_no_retryable(self): def test_admitted_sell_does_not_expose_retryability():
# buy on a non-existent slot → rejected, but buy responses omit retryable _, svc, _, handler, slot = _wire()
from uuid import uuid4 svc.transit_slot(
_, _, _, handler, _ = _wire() slot.slot_id,
resp = handler(RpcRequest(payload={ SlotState.INACTIVE,
"intent_type": "buy", "intent_id": "b1", "slot_id": str(uuid4()), SlotState.PENDING_BUY,
}, correlation_id="b1")) reason="buy",
assert resp.payload["status"] == "rejected" actor="test",
assert "retryable" not in resp.payload )
svc.transit_slot(
slot.slot_id,
SlotState.PENDING_BUY,
SlotState.HOLDING,
reason="filled",
actor="test",
qty=10,
avg_price=100,
)
response = _sell(handler, slot.slot_id)
assert response.payload["status"] == "admitted"
assert "retryable" not in response.payload
+10 -7
View File
@@ -3,6 +3,8 @@ from __future__ import annotations
from uuid import UUID from uuid import UUID
import pytest
from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore
from trex_core.application.broker_port import ( from trex_core.application.broker_port import (
BalanceItem, BalanceItem,
@@ -96,7 +98,8 @@ def test_nxt_buy_wire_value_reaches_broker_order() -> None:
assert broker.orders[0].venue == "NXT" assert broker.orders[0].venue == "NXT"
def test_invalid_venue_is_rejected_before_slot_transition() -> None: @pytest.mark.parametrize("invalid_venue", ["AUTO", "UNKNOWN"])
def test_invalid_venue_is_rejected_before_slot_transition(invalid_venue: str) -> None:
repo, intents, _, _ = _wire() repo, intents, _, _ = _wire()
handler = make_intent_rpc_handler(intents) handler = make_intent_rpc_handler(intents)
@@ -107,7 +110,7 @@ def test_invalid_venue_is_rejected_before_slot_transition() -> None:
"intent_id": "bad-venue", "intent_id": "bad-venue",
"portfolio_id": "PF-1", "portfolio_id": "PF-1",
"qty": 1, "qty": 1,
"venue": "UNKNOWN", "venue": invalid_venue,
}, },
correlation_id="bad-venue", correlation_id="bad-venue",
) )
@@ -118,7 +121,7 @@ def test_invalid_venue_is_rejected_before_slot_transition() -> None:
assert repo.list_slots("default")[0].status is SlotState.INACTIVE assert repo.list_slots("default")[0].status is SlotState.INACTIVE
def test_omitted_wire_venue_is_auto_then_actual_venue_is_persisted() -> None: def test_omitted_wire_venue_is_sor_then_actual_venue_is_persisted() -> None:
repo = InMemorySlotRepository() repo = InMemorySlotRepository()
service = SlotService(repo) service = SlotService(repo)
intents = IntentService( intents = IntentService(
@@ -146,10 +149,10 @@ def test_omitted_wire_venue_is_auto_then_actual_venue_is_persisted() -> None:
slot_id = UUID(response.payload["slot_id"]) slot_id = UUID(response.payload["slot_id"])
pending = repo.fetch_slot(slot_id) pending = repo.fetch_slot(slot_id)
assert pending is not None and pending.venue == "AUTO" assert pending is not None and pending.venue == "SOR"
execution.run_once() execution.run_once()
dispatched = repo.fetch_slot(slot_id) dispatched = repo.fetch_slot(slot_id)
assert broker.orders[0].venue == "AUTO" assert broker.orders[0].venue == "SOR"
assert dispatched is not None and dispatched.venue == "NXT" assert dispatched is not None and dispatched.venue == "NXT"
@@ -180,7 +183,7 @@ def test_inactive_transition_resets_venue_to_krx() -> None:
assert reset.venue == "KRX" assert reset.venue == "KRX"
def test_holding_transition_resets_next_order_venue_to_auto() -> None: def test_holding_transition_resets_next_order_venue_to_sor() -> None:
repo, intents, _, _ = _wire() repo, intents, _, _ = _wire()
decision = intents.submit_buy( decision = intents.submit_buy(
BuyIntent( BuyIntent(
@@ -201,7 +204,7 @@ def test_holding_transition_resets_next_order_venue_to_auto() -> None:
qty=1, qty=1,
avg_price=1000, avg_price=1000,
) )
assert holding.venue == "AUTO" assert holding.venue == "SOR"
def test_operator_cancel_uses_original_nxt_venue() -> None: def test_operator_cancel_uses_original_nxt_venue() -> None:
+6 -17
View File
@@ -15,7 +15,7 @@ import base64
import hashlib import hashlib
import json import json
from dataclasses import replace from dataclasses import replace
from datetime import datetime, time, timedelta, timezone from datetime import datetime, timedelta, timezone
from decimal import Decimal, InvalidOperation from decimal import Decimal, InvalidOperation
from time import monotonic from time import monotonic
from typing import Callable, Optional from typing import Callable, Optional
@@ -40,7 +40,7 @@ _MOCK_HINTS = ("openapivts", "virtual", "mock")
_FILL_POLL_MIN_INTERVAL_SECONDS = 5.0 _FILL_POLL_MIN_INTERVAL_SECONDS = 5.0
_FILL_POLL_TRANSPORT_BACKOFF_SECONDS = 30.0 _FILL_POLL_TRANSPORT_BACKOFF_SECONDS = 30.0
_SUBMISSION_LOOKUP_GRACE_SECONDS = 90.0 _SUBMISSION_LOOKUP_GRACE_SECONDS = 90.0
_SUPPORTED_ORDER_VENUES = frozenset({"AUTO", "KRX", "NXT", "SOR"}) _SUPPORTED_ORDER_VENUES = frozenset({"KRX", "NXT", "SOR"})
def _balance_http_error_code(status: Optional[int]) -> str: def _balance_http_error_code(status: Optional[int]) -> str:
@@ -54,26 +54,15 @@ def _balance_http_error_code(status: Optional[int]) -> str:
def resolve_order_venue(venue: str, at: datetime) -> str: def resolve_order_venue(venue: str, at: datetime) -> str:
"""Resolve AUTO to the venue that accepts domestic-stock orders at `at`. """Validate and preserve an explicit KRX/NXT/SOR order venue.
The gaps are intentional fail-closed intervals. They prevent an AUTO order ``at`` remains in the signature for compatibility with existing callers;
from silently falling back to KRX while the opening venue is indeterminate. venue selection is deliberately independent of wall-clock time.
Explicit KRX/NXT/SOR values are preserved.
""" """
normalized = str(venue).strip().upper() normalized = str(venue).strip().upper()
if normalized not in _SUPPORTED_ORDER_VENUES: if normalized not in _SUPPORTED_ORDER_VENUES:
raise ValueError("unsupported_order_venue") raise ValueError("unsupported_order_venue")
if normalized != "AUTO": return normalized
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]: def _safe_int(v: object) -> Optional[int]:
@@ -0,0 +1,16 @@
-- GitHub #61: remove Core-owned wall-clock AUTO routing.
-- Existing durable AUTO values are normalized before the stricter constraints.
UPDATE daily_slot SET venue = 'SOR' WHERE venue = 'AUTO';
UPDATE ledger_outbox SET venue = 'SOR' WHERE venue = 'AUTO';
ALTER TABLE daily_slot
DROP CONSTRAINT IF EXISTS daily_slot_venue_check;
ALTER TABLE daily_slot
ADD CONSTRAINT daily_slot_venue_check
CHECK (venue IN ('KRX', 'NXT', 'SOR'));
ALTER TABLE ledger_outbox
DROP CONSTRAINT IF EXISTS ledger_outbox_venue_check;
ALTER TABLE ledger_outbox
ADD CONSTRAINT ledger_outbox_venue_check
CHECK (venue IN ('KRX', 'NXT', 'SOR'));
+2 -3
View File
@@ -71,9 +71,8 @@ class OrderAck:
accepted: bool accepted: bool
broker_order_id: Optional[str] = None broker_order_id: Optional[str] = None
reason: Optional[str] = None reason: Optional[str] = None
# Actual venue used for submission. KIS resolves AUTO before the HTTP call; # Actual venue used for submission. Execution persists this value with the
# execution persists this value with the broker order id so restart/cancel # broker order id so restart/cancel never has to infer the original venue.
# never has to infer the original venue.
venue: Optional[str] = None venue: Optional[str] = None
raw: dict[str, Any] = field(default_factory=dict) raw: dict[str, Any] = field(default_factory=dict)
submitted_at: Optional[datetime] = None submitted_at: Optional[datetime] = None
+5 -6
View File
@@ -77,9 +77,8 @@ class BuyIntent:
# slippage_pct: effective limit = entry_price*(1+pct/100) buy / *(1-pct/100) sell. # slippage_pct: effective limit = entry_price*(1+pct/100) buy / *(1-pct/100) sell.
timeout_sec: Optional[int] = None timeout_sec: Optional[int] = None
slippage_pct: Optional[float] = None slippage_pct: Optional[float] = None
# Strategy wire default. AUTO is resolved at the broker boundary so # Missing strategy venue uses KIS SOR; explicit KRX/NXT/SOR is preserved.
# strategies remain session-agnostic while explicit KRX/NXT/SOR is kept. venue: str = "SOR"
venue: str = "AUTO"
extra: dict[str, Any] = field(default_factory=dict) extra: dict[str, Any] = field(default_factory=dict)
@@ -98,7 +97,7 @@ class SellIntent:
# dcy-022 / REQ-068: execution policy (optional, backward-compat). # dcy-022 / REQ-068: execution policy (optional, backward-compat).
timeout_sec: Optional[int] = None timeout_sec: Optional[int] = None
slippage_pct: Optional[float] = None slippage_pct: Optional[float] = None
venue: str = "AUTO" venue: str = "SOR"
extra: dict[str, Any] = field(default_factory=dict) extra: dict[str, Any] = field(default_factory=dict)
@@ -229,7 +228,7 @@ class IntentService:
timeout_error = self._timeout_error(intent.timeout_sec) timeout_error = self._timeout_error(intent.timeout_sec)
if timeout_error is not None: if timeout_error is not None:
return IntentDecision(intent.intent_id, "rejected", None, timeout_error) return IntentDecision(intent.intent_id, "rejected", None, timeout_error)
if intent.venue not in {"AUTO", "KRX", "NXT", "SOR"}: if intent.venue not in {"KRX", "NXT", "SOR"}:
return IntentDecision(intent.intent_id, "rejected", None, "invalid_venue") return IntentDecision(intent.intent_id, "rejected", None, "invalid_venue")
# dcy-036 / REQ-114: day-start entry gate (fail-closed). # dcy-036 / REQ-114: day-start entry gate (fail-closed).
@@ -347,7 +346,7 @@ class IntentService:
timeout_error = self._timeout_error(intent.timeout_sec) timeout_error = self._timeout_error(intent.timeout_sec)
if timeout_error is not None: if timeout_error is not None:
return IntentDecision(intent.intent_id, "rejected", None, timeout_error) return IntentDecision(intent.intent_id, "rejected", None, timeout_error)
if intent.venue not in {"AUTO", "KRX", "NXT", "SOR"}: if intent.venue not in {"KRX", "NXT", "SOR"}:
return IntentDecision(intent.intent_id, "rejected", None, "invalid_venue") return IntentDecision(intent.intent_id, "rejected", None, "invalid_venue")
slot = self._repo.fetch_slot(intent.slot_id) slot = self._repo.fetch_slot(intent.slot_id)
+2 -34
View File
@@ -30,32 +30,6 @@ from trex_core.domain.slot import (
_log = logging.getLogger(__name__) _log = logging.getLogger(__name__)
# ---- sell-reject retryability (issue #27 / REQ-007 CHG-20260615-01) ---------
# core-owned reason -> retryable mapping for the rejected SellIntent response.
# Maintained with the reject vocabulary (intents.py _admit_sell/_dedup) so a new
# reason gets a correct flag without the daemon touching a table. Unrecognized
# reasons default to retryable=True: a stranded stop-loss (held losing position)
# is worse than bounded, observable hammering on a terminal decision.
_SELL_REJECT_RETRYABLE: dict[str, bool] = {
"slot_state_changed": True,
"slot_not_holding:pending_buy": True,
"duplicate_in_flight": True,
"slot_not_found": False,
"slot_not_holding:inactive": False,
"slot_not_holding:pending_sell": False,
"slot_not_holding:error": False,
}
def _sell_reject_retryable(reason: Optional[str]) -> bool:
"""Map a rejected-sell reason to retryable (issue #27). Default True (safety)."""
if reason is None:
return True
# malformed payload: client defect — retrying the same bytes is pointless.
if reason.startswith("bad_request"):
return False
return _SELL_REJECT_RETRYABLE.get(reason, True)
# ---- intent RPC handler ----------------------------------------------------- # ---- intent RPC handler -----------------------------------------------------
@@ -117,7 +91,7 @@ def make_intent_rpc_handler(
# dcy-022 / REQ-068: execution policy (optional, backward-compat). # dcy-022 / REQ-068: execution policy (optional, backward-compat).
timeout_sec=_opt_int(body.get("timeout_sec")), timeout_sec=_opt_int(body.get("timeout_sec")),
slippage_pct=_opt_float(body.get("slippage_pct")), slippage_pct=_opt_float(body.get("slippage_pct")),
venue=_opt_str(body.get("venue")) or "AUTO", venue=_opt_str(body.get("venue")) or "SOR",
) )
d = intent_service.submit_buy(buy) d = intent_service.submit_buy(buy)
elif intent_type == "sell": elif intent_type == "sell":
@@ -133,7 +107,7 @@ def make_intent_rpc_handler(
# dcy-022 / REQ-068: execution policy (optional, backward-compat). # dcy-022 / REQ-068: execution policy (optional, backward-compat).
timeout_sec=_opt_int(body.get("timeout_sec")), timeout_sec=_opt_int(body.get("timeout_sec")),
slippage_pct=_opt_float(body.get("slippage_pct")), slippage_pct=_opt_float(body.get("slippage_pct")),
venue=_opt_str(body.get("venue")) or "AUTO", venue=_opt_str(body.get("venue")) or "SOR",
) )
d = intent_service.submit_sell(sell) d = intent_service.submit_sell(sell)
else: else:
@@ -146,9 +120,6 @@ def make_intent_rpc_handler(
"intent_id": intent_id, "status": "rejected", "intent_id": intent_id, "status": "rejected",
"slot_id": None, "reason": f"bad_request:{exc}", "slot_id": None, "reason": f"bad_request:{exc}",
} }
# issue #27: sell-side malformed payload carries retryable=False.
if intent_type == "sell":
payload["retryable"] = _sell_reject_retryable(payload["reason"])
return RpcResponse(payload=payload) return RpcResponse(payload=payload)
payload = { payload = {
"intent_id": d.intent_id, "intent_id": d.intent_id,
@@ -156,9 +127,6 @@ def make_intent_rpc_handler(
"slot_id": (str(d.slot_id) if d.slot_id else None), "slot_id": (str(d.slot_id) if d.slot_id else None),
"reason": d.reason, "reason": d.reason,
} }
# issue #27 / REQ-007 AC-7: additive retryable on rejected sell only.
if intent_type == "sell" and d.status == "rejected":
payload["retryable"] = _sell_reject_retryable(d.reason)
return RpcResponse(payload=payload) return RpcResponse(payload=payload)
return handler return handler
+2 -3
View File
@@ -232,9 +232,8 @@ class SlotService:
new_timeout_sec = None new_timeout_sec = None
new_slippage_pct = None new_slippage_pct = None
# A completed order's venue must not leak into a later autonomous # A completed order's venue must not leak into a later autonomous
# sell. The next order starts session-aware unless its intent # sell. Missing venue defaults to KIS SOR for the next order.
# explicitly chooses KRX/NXT/SOR. new_venue = "SOR" if to_state is SlotState.HOLDING else slot.venue
new_venue = "AUTO" if to_state is SlotState.HOLDING else slot.venue
# occupancy_seq (dcy-028/REQ-081): bump when a slot begins a new # occupancy_seq (dcy-028/REQ-081): bump when a slot begins a new
# occupancy (inactive→pending_buy). Stable across the rest of the # occupancy (inactive→pending_buy). Stable across the rest of the