trex/trex-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: reconcile ambiguous KIS order submissions

Browse Source
This commit is contained in:
jordan
2026-07-28 11:17:06 +09:00
parent b72bf9f394
commit 24c42d1c92
11 changed files with 1420 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files
trex_core/adapters/broker/kis.py
+338 -37
View File
@@ -8,10 +8,15 @@ Endpoints: /oauth2/tokenP (token), /uapi/domestic-stock/v1/trading/order-cash
(submit), /uapi/domestic-stock/v1/trading/inquire-daily-ccld (fill poll),
/uapi/domestic-stock/v1/trading/order-rvsecncl (cancel).
"""
from __future__ import annotations
import base64
import json
from dataclasses import replace
from datetime import datetime, timedelta, timezone
from decimal import Decimal, InvalidOperation
from time import monotonic
from typing import Optional
import httpx
@@ -27,7 +32,11 @@ from trex_core.application.broker_port import (
)
_KST = timezone(timedelta(hours=9))
_SUBMISSION_UNKNOWN_PREFIX = "submission-unknown:"
_MOCK_HINTS = ("openapivts", "virtual", "mock")
_FILL_POLL_MIN_INTERVAL_SECONDS = 5.0
_FILL_POLL_TRANSPORT_BACKOFF_SECONDS = 30.0
_SUBMISSION_LOOKUP_GRACE_SECONDS = 90.0
def _safe_int(v: object) -> Optional[int]:
@@ -37,6 +46,55 @@ def _safe_int(v: object) -> Optional[int]:
return None
def _encode_submission_unknown(req: OrderRequest, submitted_at: datetime) -> str:
payload = {
"v": 1,
"slot_id": req.slot_id,
"side": req.side,
"symbol": req.symbol,
"qty": req.qty,
"limit_price": req.limit_price,
"venue": req.venue,
"submitted_at": submitted_at.astimezone(_KST).isoformat(),
}
raw = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode()
encoded = base64.urlsafe_b64encode(raw).decode().rstrip("=")
return f"{_SUBMISSION_UNKNOWN_PREFIX}{encoded}"
def _decode_submission_unknown(broker_order_id: str) -> Optional[dict]:
if not broker_order_id.startswith(_SUBMISSION_UNKNOWN_PREFIX):
return None
encoded = broker_order_id[len(_SUBMISSION_UNKNOWN_PREFIX) :]
try:
raw = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4))
payload = json.loads(raw)
except (ValueError, UnicodeDecodeError, json.JSONDecodeError):
return None
required = {
"v",
"slot_id",
"side",
"symbol",
"qty",
"venue",
"submitted_at",
}
if (
not isinstance(payload, dict)
or payload.get("v") != 1
or not required.issubset(payload)
or payload.get("side") not in {"buy", "sell"}
):
return None
try:
datetime.fromisoformat(str(payload["submitted_at"]))
int(payload["qty"])
except (TypeError, ValueError):
return None
return payload
class KISBroker:
"""KIS REST broker adapter. Live account is gated behind `allow_live`."""
@@ -59,6 +117,7 @@ class KISBroker:
self._token: Optional[str] = None
self._token_exp: datetime = datetime.now(tz=timezone.utc)
self._token_cache = token_cache # optional TokenCache (L2, REQ-016)
self._fill_poll_not_before: dict[str, float] = {}
is_mock = any(h in self._base_url for h in _MOCK_HINTS)
if not is_mock and not allow_live:
raise RuntimeError(
@@ -167,6 +226,7 @@ class KISBroker:
# --- BrokerPort ---
def submit_order(self, req: OrderRequest) -> OrderAck:
cano, product = self._account()
submitted_at = datetime.now(tz=_KST)
body = {
"CANO": cano,
"ACNT_PRDT_CD": product,
@@ -188,18 +248,44 @@ class KISBroker:
)
payload = resp.json()
except (httpx.HTTPError, ValueError) as exc:
return OrderAck(accepted=False, reason=f"submit_network_error:{type(exc).__name__}",
raw={"error": str(exc)})
# A transport failure after POST does not prove that KIS rejected
# the order. Keep the slot dispatched under a local sentinel so
# the execution loop cannot submit the same order again.
return OrderAck(
accepted=True,
broker_order_id=_encode_submission_unknown(req, submitted_at),
reason=f"submit_outcome_unknown:{type(exc).__name__}",
raw={"error_class": type(exc).__name__},
)
if getattr(resp, "status_code", 200) >= 500:
return OrderAck(
accepted=True,
broker_order_id=_encode_submission_unknown(req, submitted_at),
reason=f"submit_outcome_unknown:http_{resp.status_code}",
raw={"http_status": resp.status_code},
)
if str(payload.get("rt_cd")) == "0":
out = payload.get("output") or {}
odno = out.get("ODNO") or out.get("odno")
orgno = out.get("KRX_FWDG_ORD_ORGNO") or out.get("ord_gno_brno") or ""
bid = f"{orgno}:{odno}" if orgno else str(odno)
return OrderAck(accepted=True, broker_order_id=bid, raw=payload)
return OrderAck(accepted=False, reason=payload.get("msg1") or payload.get("msg_cd"),
raw=payload)
return OrderAck(
accepted=False, reason=payload.get("msg1") or payload.get("msg_cd"), raw=payload
)
def poll_fill(self, broker_order_id: str, *, side: str, symbol: str) -> FillStatus:
if broker_order_id.startswith(_SUBMISSION_UNKNOWN_PREFIX):
return self._poll_unknown_submission(broker_order_id, side=side, symbol=symbol)
now = monotonic()
if now < self._fill_poll_not_before.get(broker_order_id, 0.0):
return FillStatus(
state="accepted",
raw={"reason": "fill_poll_backoff"},
)
# Set the normal gate before I/O so concurrent execution ticks cannot
# open duplicate KIS inquiries for the same order.
self._fill_poll_not_before[broker_order_id] = now + _FILL_POLL_MIN_INTERVAL_SECONDS
cano, product = self._account()
odno = broker_order_id.split(":")[-1]
today = datetime.now(tz=_KST).strftime("%Y%m%d")
@@ -208,25 +294,206 @@ class KISBroker:
# 향후 NXT 체결 미포함 실측 확인 시 TTTC0081R(EXCG=ALL) 전환 별도 batch 처리.
tr_id = "VTTC8001R" if self._mock else "TTTC8001R"
params = {
"CANO": cano, "ACNT_PRDT_CD": product,
"INQR_STRT_DT": today, "INQR_END_DT": today,
"SLL_BUY_DVSN_CD": "00", "INQR_DVSN": "00", "PDNO": "",
"CCLD_DVSN": "00", "ORD_GNO_BRNO": "", "ODNO": odno,
"INQR_DVSN_3": "00", "INQR_DVSN_1": "",
"CTX_AREA_FK100": "", "CTX_AREA_NK100": "",
"CANO": cano,
"ACNT_PRDT_CD": product,
"INQR_STRT_DT": today,
"INQR_END_DT": today,
"SLL_BUY_DVSN_CD": "00",
"INQR_DVSN": "00",
"PDNO": "",
"CCLD_DVSN": "00",
"ORD_GNO_BRNO": "",
"ODNO": odno,
"INQR_DVSN_3": "00",
"INQR_DVSN_1": "",
"CTX_AREA_FK100": "",
"CTX_AREA_NK100": "",
}
try:
resp = httpx.get(
f"{self._base_url}/uapi/domestic-stock/v1/trading/inquire-daily-ccld",
headers=self._headers(tr_id), params=params, timeout=self._timeout,
headers=self._headers(tr_id),
params=params,
timeout=self._timeout,
)
payload = resp.json()
except (httpx.HTTPError, ValueError) as exc:
self._fill_poll_not_before[broker_order_id] = (
monotonic() + _FILL_POLL_TRANSPORT_BACKOFF_SECONDS
)
return FillStatus(state="unknown", raw={"error": str(exc)})
if getattr(resp, "status_code", 200) >= 400 or str(payload.get("rt_cd")) != "0":
self._fill_poll_not_before[broker_order_id] = (
monotonic() + _FILL_POLL_TRANSPORT_BACKOFF_SECONDS
)
return FillStatus(
state="unknown",
raw={"reason": "fill_inquiry_rejected"},
)
self._fill_poll_not_before[broker_order_id] = monotonic() + _FILL_POLL_MIN_INTERVAL_SECONDS
rows = payload.get("output1") or []
row = next((r for r in rows if str(r.get("odno")) == odno), rows[0] if rows else None)
normalized_odno = odno.lstrip("0") or "0"
row = next(
(
r
for r in rows
if (str(r.get("odno") or "").strip().lstrip("0") or "0") == normalized_odno
),
None,
)
if row is None:
return FillStatus(state="accepted", raw=payload)
return FillStatus(
state="accepted",
raw={"reason": "order_not_visible_in_fill_inquiry"},
)
return self._fill_status_from_row(row, side=side, today=today, fallback_odno=odno)
def _poll_unknown_submission(
self,
broker_order_id: str,
*,
side: str,
symbol: str,
) -> FillStatus:
metadata = _decode_submission_unknown(broker_order_id)
if metadata is None:
return FillStatus(
state="unknown",
raw={"reason": "submission_unknown_metadata_invalid"},
)
now_monotonic = monotonic()
if now_monotonic < self._fill_poll_not_before.get(broker_order_id, 0.0):
return FillStatus(
state="accepted",
raw={"reason": "submission_lookup_backoff"},
)
self._fill_poll_not_before[broker_order_id] = (
now_monotonic + _FILL_POLL_MIN_INTERVAL_SECONDS
)
cano, product = self._account()
submitted_at = datetime.fromisoformat(str(metadata["submitted_at"])).astimezone(_KST)
today = submitted_at.strftime("%Y%m%d")
tr_id = "VTTC8001R" if self._mock else "TTTC8001R"
params = {
"CANO": cano,
"ACNT_PRDT_CD": product,
"INQR_STRT_DT": today,
"INQR_END_DT": today,
"SLL_BUY_DVSN_CD": "00",
"INQR_DVSN": "00",
"PDNO": str(metadata["symbol"]),
"CCLD_DVSN": "00",
"ORD_GNO_BRNO": "",
"ODNO": "",
"INQR_DVSN_3": "00",
"INQR_DVSN_1": "",
"CTX_AREA_FK100": "",
"CTX_AREA_NK100": "",
}
try:
resp = httpx.get(
f"{self._base_url}/uapi/domestic-stock/v1/trading/inquire-daily-ccld",
headers=self._headers(tr_id),
params=params,
timeout=self._timeout,
)
payload = resp.json()
except (httpx.HTTPError, ValueError):
self._fill_poll_not_before[broker_order_id] = (
monotonic() + _FILL_POLL_TRANSPORT_BACKOFF_SECONDS
)
return FillStatus(
state="unknown",
raw={"reason": "submission_lookup_transport"},
)
if getattr(resp, "status_code", 200) >= 400 or str(payload.get("rt_cd")) != "0":
self._fill_poll_not_before[broker_order_id] = (
monotonic() + _FILL_POLL_TRANSPORT_BACKOFF_SECONDS
)
return FillStatus(
state="unknown",
raw={"reason": "submission_lookup_rejected"},
)
matches = [
row
for row in (payload.get("output1") or [])
if self._matches_unknown_submission(row, metadata, submitted_at)
]
if len(matches) != 1:
age = (datetime.now(tz=_KST) - submitted_at).total_seconds()
if not matches and age >= _SUBMISSION_LOOKUP_GRACE_SECONDS:
return FillStatus(
state="rejected",
raw={"reason": "submission_confirmed_absent"},
)
return FillStatus(
state="unknown",
raw={
"reason": (
"submission_lookup_ambiguous" if matches else "submission_lookup_pending"
)
},
)
row = matches[0]
resolved_order_id = self._broker_order_id_from_row(row)
status = self._fill_status_from_row(
row,
side=side,
today=today,
fallback_odno=resolved_order_id.split(":")[-1],
)
return replace(
status,
raw={
**(status.raw if isinstance(status.raw, dict) else {}),
"resolved_broker_order_id": resolved_order_id,
},
)
@staticmethod
def _broker_order_id_from_row(row: dict) -> str:
odno = str(row.get("odno") or "").strip()
orgno = str(row.get("ord_gno_brno") or "").strip()
return f"{orgno}:{odno}" if orgno else odno
@staticmethod
def _matches_unknown_submission(
row: dict,
metadata: dict,
submitted_at: datetime,
) -> bool:
if str(row.get("pdno") or "").strip() != str(metadata["symbol"]):
return False
expected_side = "02" if metadata["side"] == "buy" else "01"
if str(row.get("sll_buy_dvsn_cd") or "").strip() not in {"", expected_side}:
return False
if (_safe_int(row.get("ord_qty")) or 0) != int(metadata["qty"]):
return False
expected_price = int(metadata.get("limit_price") or 0)
row_price = _safe_int(row.get("ord_unpr"))
if row_price is not None and row_price != expected_price:
return False
order_date = str(row.get("ord_dt") or "").strip()
order_time = str(row.get("ord_tmd") or "").strip()[:6]
if len(order_date) != 8 or len(order_time) != 6:
return False
try:
ordered_at = datetime.strptime(order_date + order_time, "%Y%m%d%H%M%S").replace(
tzinfo=_KST
)
except ValueError:
return False
delta = (ordered_at - submitted_at).total_seconds()
return -5.0 <= delta <= 120.0
@staticmethod
def _fill_status_from_row(
row: dict,
*,
side: str,
today: str,
fallback_odno: str,
) -> FillStatus:
ord_qty = _safe_int(row.get("ord_qty")) or 0
ccld_qty = _safe_int(row.get("tot_ccld_qty")) or 0
avg = _safe_int(row.get("avg_prvs")) or _safe_int(row.get("ccld_prvs"))
@@ -245,7 +512,7 @@ class KISBroker:
# kis-daily-ccld:{ord_dt}:{ord_gno_brno}:{odno}:{tot_ccld_qty}
ord_dt = str(row.get("ord_dt") or today)
ord_gno_brno = str(row.get("ord_gno_brno") or "")
row_odno = str(row.get("odno") or odno)
row_odno = str(row.get("odno") or fallback_odno)
execution_id = f"kis-daily-ccld:{ord_dt}:{ord_gno_brno}:{row_odno}:{ccld_qty}"
fill_price = float(avg) if avg else 0.0
# dcy-026: inquire-daily-ccld (output1) does not expose per-fill
@@ -256,9 +523,13 @@ class KISBroker:
fee = calculate_fee(side, amount)
tax = calculate_tax(side, amount)
return FillStatus(
state="filled", filled_qty=ccld_qty,
state="filled",
filled_qty=ccld_qty,
avg_price=fill_price or None,
execution_id=execution_id, fee=fee, tax=tax, raw=row,
execution_id=execution_id,
fee=fee,
tax=tax,
raw=row,
)
# 0 < ccld_qty < ord_qty
if cancelled:
@@ -268,20 +539,28 @@ class KISBroker:
# holding instead of waiting for a fill that will never come.
ord_dt = str(row.get("ord_dt") or today)
ord_gno_brno = str(row.get("ord_gno_brno") or "")
row_odno = str(row.get("odno") or odno)
row_odno = str(row.get("odno") or fallback_odno)
execution_id = f"kis-daily-ccld:{ord_dt}:{ord_gno_brno}:{row_odno}:{ccld_qty}"
fill_price = float(avg) if avg else 0.0
amount = ccld_qty * fill_price
fee = calculate_fee(side, amount)
tax = calculate_tax(side, amount)
return FillStatus(
state="partially_filled", filled_qty=ccld_qty,
state="partially_filled",
filled_qty=ccld_qty,
avg_price=fill_price or None,
execution_id=execution_id, fee=fee, tax=tax,
cancelled=True, raw=row,
execution_id=execution_id,
fee=fee,
tax=tax,
cancelled=True,
raw=row,
)
return FillStatus(state="partially_filled", filled_qty=ccld_qty,
avg_price=float(avg) if avg else None, raw=row)
return FillStatus(
state="partially_filled",
filled_qty=ccld_qty,
avg_price=float(avg) if avg else None,
raw=row,
)
def get_balance(self) -> list[BalanceItem]:
"""KIS 주식잔고조회 (`TTTC8434R`/`VTTC8434R`).
@@ -294,12 +573,17 @@ class KISBroker:
cano, product = self._account()
tr_id = "VTTC8434R" if self._mock else "TTTC8434R"
params = {
"CANO": cano, "ACNT_PRDT_CD": product,
"AFHR_FLPR_YN": "N", "OFL_YN": "",
"INQR_DVSN": "02", "UNPR_DVSN": "01",
"FUND_STTL_ICLD_YN": "N", "FNCG_AMT_AUTO_RDPT_YN": "N",
"CANO": cano,
"ACNT_PRDT_CD": product,
"AFHR_FLPR_YN": "N",
"OFL_YN": "",
"INQR_DVSN": "02",
"UNPR_DVSN": "01",
"FUND_STTL_ICLD_YN": "N",
"FNCG_AMT_AUTO_RDPT_YN": "N",
"PRCS_DVSN": "01",
"CTX_AREA_FK100": "", "CTX_AREA_NK100": "",
"CTX_AREA_FK100": "",
"CTX_AREA_NK100": "",
}
try:
headers = self._headers(tr_id)
@@ -310,14 +594,18 @@ class KISBroker:
try:
resp = httpx.get(
f"{self._base_url}/uapi/domestic-stock/v1/trading/inquire-balance",
headers=headers, params=params, timeout=self._timeout,
headers=headers,
params=params,
timeout=self._timeout,
)
resp.raise_for_status()
payload = resp.json()
except httpx.HTTPStatusError as exc:
status = exc.response.status_code if exc.response is not None else None
code = "authentication_failed" if status == 401 else (
"authorization_failed" if status == 403 else "transport_failed"
code = (
"authentication_failed"
if status == 401
else ("authorization_failed" if status == 403 else "transport_failed")
)
raise BrokerBalanceError(code) from None
except httpx.HTTPError:
@@ -400,7 +688,9 @@ class KISBroker:
try:
resp = httpx.get(
f"{self._base_url}/uapi/domestic-stock/v1/quotations/inquire-price",
headers=self._headers(self._tr_price()), params=params, timeout=self._timeout,
headers=self._headers(self._tr_price()),
params=params,
timeout=self._timeout,
)
payload = resp.json()
except (httpx.HTTPError, ValueError):
@@ -416,21 +706,30 @@ class KISBroker:
except (InvalidOperation, ValueError):
return None
def cancel_order(self, broker_order_id: str, *, side: str, symbol: str, qty: int, venue: str = "KRX") -> bool:
def cancel_order(
self, broker_order_id: str, *, side: str, symbol: str, qty: int, venue: str = "KRX"
) -> bool:
"""Cancel a previously submitted order (REQ-118).
venue: exchange venue of the original order. Must match the EXCG_ID_DVSN_CD
used at submit time. Default "KRX" preserves backward compatibility.
Real-account NXT cancel (EXCG=NXT) confirmed rt_cd=0 on 2026-07-06.
"""
if broker_order_id.startswith(_SUBMISSION_UNKNOWN_PREFIX):
return False
cano, product = self._account()
parts = broker_order_id.split(":")
orgno, odno = (parts[0], parts[1]) if len(parts) == 2 else ("", parts[-1])
body = {
"CANO": cano, "ACNT_PRDT_CD": product,
"KRX_FWDG_ORD_ORGNO": orgno, "ORGN_ODNO": odno,
"ORD_DVSN": "00", "RVSE_CNCL_DVSN_CD": "02",
"ORD_QTY": str(qty), "ORD_UNPR": "0", "QTY_ALL_ORD_YN": "Y",
"CANO": cano,
"ACNT_PRDT_CD": product,
"KRX_FWDG_ORD_ORGNO": orgno,
"ORGN_ODNO": odno,
"ORD_DVSN": "00",
"RVSE_CNCL_DVSN_CD": "02",
"ORD_QTY": str(qty),
"ORD_UNPR": "0",
"QTY_ALL_ORD_YN": "Y",
# REQ-118: cancel must use the same EXCG_ID_DVSN_CD as the original submit.
"EXCG_ID_DVSN_CD": venue,
}
@@ -438,7 +737,9 @@ class KISBroker:
try:
resp = httpx.post(
f"{self._base_url}/uapi/domestic-stock/v1/trading/order-rvsecncl",
headers=self._headers(tr_id), json=body, timeout=self._timeout,
headers=self._headers(tr_id),
json=body,
timeout=self._timeout,
)
return str(resp.json().get("rt_cd")) == "0"
except (httpx.HTTPError, ValueError):
trex_core/adapters/persistence/postgresql.py
+98 -30
View File
@@ -9,6 +9,7 @@ psycopg (v3) is imported lazily so the domain/test paths that use the
in-memory adapters don't require it. External I/O lives here, never in the
domain layer (NFR-1).
"""
from __future__ import annotations
import json
@@ -71,9 +72,7 @@ class PostgreSQLSlotRepository:
conn.rollback()
raise
def list_slots(
self, instance_key: str, status_filter: Optional[set[str]] = None
) -> list[Slot]:
def list_slots(self, instance_key: str, status_filter: Optional[set[str]] = None) -> list[Slot]:
with self._psycopg.connect(self._dsn) as conn:
with conn.cursor() as cur:
if status_filter:
@@ -147,8 +146,11 @@ class PostgreSQLSlotRepository:
return self._row_to_slot(row) if row else None
def reconcile_universe(
self, instance_key: str, universe: dict[str, str],
*, allow_retire: bool = True,
self,
instance_key: str,
universe: dict[str, str],
*,
allow_retire: bool = True,
) -> dict[str, int]:
now = datetime.now(tz=timezone.utc)
created = retired = tombstoned = 0
@@ -216,14 +218,26 @@ class PostgreSQLSlotRepository:
"instance_key=%s, retired_at=%s, timeout_sec=%s, slippage_pct=%s, "
"occupancy_seq=%s, updated_at=%s WHERE slot_id=%s",
(
slot.status.value, slot.symbol, slot.qty, slot.avg_price,
slot.limit_price, slot.order_id,
slot.strategy_id, slot.broker_order_id, slot.portfolio_id,
slot.sell_strategy_id, slot.forced_close_at,
slot.stop_loss_price, slot.take_profit_price,
slot.instance_key, slot.retired_at,
slot.timeout_sec, slot.slippage_pct, slot.occupancy_seq,
slot.updated_at or datetime.now(tz=timezone.utc), slot.slot_id,
slot.status.value,
slot.symbol,
slot.qty,
slot.avg_price,
slot.limit_price,
slot.order_id,
slot.strategy_id,
slot.broker_order_id,
slot.portfolio_id,
slot.sell_strategy_id,
slot.forced_close_at,
slot.stop_loss_price,
slot.take_profit_price,
slot.instance_key,
slot.retired_at,
slot.timeout_sec,
slot.slippage_pct,
slot.occupancy_seq,
slot.updated_at or datetime.now(tz=timezone.utc),
slot.slot_id,
),
)
if cur.rowcount == 0:
@@ -239,15 +253,22 @@ class PostgreSQLSlotRepository:
"(slot_id, from_state, to_state, ts, reason, actor, extra) "
"VALUES (%s, %s, %s, %s, %s, %s, %s)",
(
record.slot_id, record.from_state.value, record.to_state.value,
record.ts, record.reason, record.actor, json.dumps(record.extra),
record.slot_id,
record.from_state.value,
record.to_state.value,
record.ts,
record.reason,
record.actor,
json.dumps(record.extra),
),
)
conn.commit()
# --- order execution queue (BAT-004 / REQ-013, option A) ---
def list_undispatched_pending(
self, instance_key: str = "default", limit: int = 64,
self,
instance_key: str = "default",
limit: int = 64,
) -> list[Slot]:
# dcy-020/REQ-056 (BAT-030): instance-scoped so two daemons sharing a
# DB never dispatch each other's orders. List-based (vs single-row
@@ -275,6 +296,29 @@ class PostgreSQLSlotRepository:
conn.commit()
return ok
def replace_dispatch(
self,
slot_id: UUID,
expected_broker_order_id: str,
broker_order_id: str,
) -> bool:
"""Atomically persist a broker order ID resolved after submit timeout."""
with self._psycopg.connect(self._dsn, autocommit=False) as conn:
with conn.cursor() as cur:
cur.execute(
"UPDATE daily_slot SET broker_order_id=%s, updated_at=%s "
"WHERE slot_id=%s AND broker_order_id=%s",
(
broker_order_id,
datetime.now(tz=timezone.utc),
slot_id,
expected_broker_order_id,
),
)
ok = cur.rowcount == 1
conn.commit()
return ok
def list_dispatched_pending(self, instance_key: str = "default") -> list[Slot]:
with self._psycopg.connect(self._dsn) as conn:
with conn.cursor() as cur:
@@ -308,6 +352,7 @@ class PostgreSQLSlotRepository:
@staticmethod
def _row_to_slot(row: tuple) -> Slot:
from decimal import Decimal
f = lambda v: float(v) if v is not None else None # noqa: E731
d = lambda v: Decimal(str(v)) if v is not None else None # noqa: E731
# Column order: slot_id(0) status(1) symbol(2) qty(3) avg_price(4)
@@ -373,8 +418,12 @@ class PostgreSQLIntentStore:
if row is None:
return None
return _PgIntentRecord(
intent_id=row[0], intent_type=row[1], status=row[2],
slot_id=(row[3] if isinstance(row[3], UUID) or row[3] is None else UUID(str(row[3]))),
intent_id=row[0],
intent_type=row[1],
status=row[2],
slot_id=(
row[3] if isinstance(row[3], UUID) or row[3] is None else UUID(str(row[3]))
),
reason=row[4],
)
@@ -426,9 +475,15 @@ class PostgreSQLLedgerOutbox:
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) "
"ON CONFLICT (correlation_id) DO NOTHING",
(
trade.correlation_id, trade.asset_id, trade.order_id,
trade.execution_id, trade.side,
trade.qty, trade.price, trade.fee, trade.tax,
trade.correlation_id,
trade.asset_id,
trade.order_id,
trade.execution_id,
trade.side,
trade.qty,
trade.price,
trade.fee,
trade.tax,
trade.executed_at,
),
)
@@ -506,8 +561,13 @@ class PostgreSQLLedgerOutbox:
"last_outcome=%s, last_error=%s, next_retry_at=%s, updated_at=%s "
"WHERE correlation_id=%s",
(
status_value, attempts_delta, outcome.value, error, nra,
datetime.now(tz=timezone.utc), correlation_id,
status_value,
attempts_delta,
outcome.value,
error,
nra,
datetime.now(tz=timezone.utc),
correlation_id,
),
)
conn.commit()
@@ -519,8 +579,14 @@ class PostgreSQLLedgerOutbox:
def _row_to_item(row: tuple) -> LedgerWorkItem:
f = lambda v: float(v) if v is not None else 0.0 # noqa: E731
trade = LedgerTrade(
asset_id=row[1], order_id=row[2], execution_id=row[3], side=row[4],
qty=f(row[5]), price=f(row[6]), fee=f(row[7]), tax=f(row[8]),
asset_id=row[1],
order_id=row[2],
execution_id=row[3],
side=row[4],
qty=f(row[5]),
price=f(row[6]),
fee=f(row[7]),
tax=f(row[8]),
executed_at=row[9],
)
return LedgerWorkItem(
@@ -601,8 +667,12 @@ class PgStrategyStateStore:
"version = EXCLUDED.version, data = EXCLUDED.data, "
"updated_at = EXCLUDED.updated_at",
(
slot_id, strategy_id, current_occupancy_seq,
result.version, data, datetime.now(tz=timezone.utc),
slot_id,
strategy_id,
current_occupancy_seq,
result.version,
data,
datetime.now(tz=timezone.utc),
),
)
conn.commit()
@@ -641,7 +711,5 @@ class PgStrategyStateStore:
def delete_all(self, slot_id: UUID) -> None:
with self._psycopg.connect(self._dsn, autocommit=False) as conn:
with conn.cursor() as cur:
cur.execute(
"DELETE FROM slot_strategy_state WHERE slot_id = %s", (slot_id,)
)
cur.execute("DELETE FROM slot_strategy_state WHERE slot_id = %s", (slot_id,))
conn.commit()
trex_core/application/execution.py
+182 -62
View File
@@ -20,10 +20,11 @@ All slot writes go through SlotService (advisory lock). The broker is touched
only here, never by the domain or the HTTP API. The ledger outbox is the
single producer for ledger work items (NFR-3, single record subject).
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from dataclasses import dataclass, replace
from datetime import datetime, timezone
from typing import Callable, Optional
@@ -48,7 +49,9 @@ _log = logging.getLogger(__name__)
# _apply_transition_fields). The emitter MUST prefer this hint over
# `slot.strategy_id` so that slot_filled_sell events reach the correct
# per-strategy queue instead of being NO_ROUTEd to the DLQ.
SlotEventEmitter = Callable[[str, object, Optional[str]], None] # (event_type, slot, routing_strategy_id) -> None
SlotEventEmitter = Callable[
[str, object, Optional[str]], None
] # (event_type, slot, routing_strategy_id) -> None
_MAX_DISPATCH_PER_TICK = 64
@@ -58,11 +61,11 @@ class ExecutionResult:
dispatched: int = 0
filled: int = 0
rolled_back: int = 0
ledger_enqueued: int = 0 # REQ-019: ledger work items appended this tick
ledger_skipped: int = 0 # REQ-018 AC-4: fills without portfolio_id
stale_pending: int = 0 # dcy-013 / REQ-014 FR-3: pending orders over threshold
ledger_enqueued: int = 0 # REQ-019: ledger work items appended this tick
ledger_skipped: int = 0 # REQ-018 AC-4: fills without portfolio_id
stale_pending: int = 0 # dcy-013 / REQ-014 FR-3: pending orders over threshold
sell_skipped_no_balance: int = 0 # dcy-020 / REQ-062: broker flat -> sell skipped
timeout_cancelled: int = 0 # dcy-022 / REQ-069: cancel_order sent on timeout
timeout_cancelled: int = 0 # dcy-022 / REQ-069: cancel_order sent on timeout
class ExecutionService:
@@ -139,7 +142,8 @@ class ExecutionService:
_log.warning(
"sell dispatch skipped: slot=%s symbol=%s broker balance is 0 "
"(divergence — no order submitted; operator/reconcile attention)",
slot.slot_id, slot.symbol,
slot.slot_id,
slot.symbol,
)
# dcy-033 / REQ-100: CRITICAL alarm — PENDING_SELL may stick.
if self._alarm_emitter is not None:
@@ -175,7 +179,12 @@ class ExecutionService:
_log.debug(
"slippage applied: slot=%s side=%s base_limit=%s pct=%.4f "
"raw=%.2f effective=%s (tick-snapped)",
slot.slot_id, side, slot.limit_price, slot.slippage_pct, raw, limit,
slot.slot_id,
side,
slot.limit_price,
slot.slippage_pct,
raw,
limit,
)
# REQ-084/085: correlation id (intent→order→fill join key = slot_id)
# so submit/fill/reject logs for one slot share a key.
@@ -193,20 +202,41 @@ class ExecutionService:
if ack.accepted and ack.broker_order_id:
self._repo.mark_dispatched(slot.slot_id, ack.broker_order_id)
r.dispatched += 1
# REQ-085: order-result logging — tailable submit line.
_log.info(
"order submitted event=order_submit side=%s symbol=%s slot_id=%s "
"qty=%s limit_price=%s order_type=%s",
side, slot.symbol, slot.slot_id, qty, limit,
"00" if limit is not None else "01",
)
if ack.reason and ack.reason.startswith("submit_outcome_unknown:"):
_log.warning(
"order submission outcome unknown event=order_submit_unknown "
"side=%s symbol=%s slot_id=%s qty=%s limit_price=%s "
"automatic_retry=blocked reason=%s",
side,
slot.symbol,
slot.slot_id,
qty,
limit,
ack.reason,
)
else:
# REQ-085: order-result logging — tailable submit line.
_log.info(
"order submitted event=order_submit side=%s symbol=%s slot_id=%s "
"qty=%s limit_price=%s order_type=%s",
side,
slot.symbol,
slot.slot_id,
qty,
limit,
"00" if limit is not None else "01",
)
else:
# REQ-085/087: reject is visible (event key) and ERROR level
# (no more silent rollback — issue #28 surfaced too late).
_log.error(
"order rejected event=order_reject side=%s symbol=%s slot_id=%s "
"qty=%s reason=%s",
side, slot.symbol, slot.slot_id, qty, ack.reason or "order_rejected",
side,
slot.symbol,
slot.slot_id,
qty,
ack.reason or "order_rejected",
)
self._rollback(slot, reason=ack.reason or "order_rejected")
r.rolled_back += 1
@@ -217,6 +247,30 @@ class ExecutionService:
side = "buy" if slot.status is SlotState.PENDING_BUY else "sell"
set_correlation_id(str(slot.slot_id)) # REQ-084: same key as submit
fs = self._broker.poll_fill(slot.broker_order_id, side=side, symbol=slot.symbol or "")
resolved_order_id = (
fs.raw.get("resolved_broker_order_id") if isinstance(fs.raw, dict) else None
)
if resolved_order_id and resolved_order_id != slot.broker_order_id:
if not self._repo.replace_dispatch(
slot.slot_id,
str(slot.broker_order_id),
str(resolved_order_id),
):
_log.error(
"resolved broker order id persistence failed "
"event=order_submit_resolution_conflict slot_id=%s symbol=%s",
slot.slot_id,
slot.symbol,
)
continue
slot = replace(slot, broker_order_id=str(resolved_order_id))
_log.info(
"order submission resolved event=order_submit_resolved "
"side=%s symbol=%s slot_id=%s",
side,
slot.symbol,
slot.slot_id,
)
if fs.is_filled:
self._on_filled(slot, fs, side, r)
r.filled += 1
@@ -230,7 +284,10 @@ class ExecutionService:
_log.error(
"order rejected event=order_reject side=%s symbol=%s slot_id=%s "
"reason=order_%s",
side, slot.symbol, slot.slot_id, fs.state,
side,
slot.symbol,
slot.slot_id,
fs.state,
)
self._rollback(slot, reason=f"order_{fs.state}")
r.rolled_back += 1
@@ -270,8 +327,11 @@ class ExecutionService:
_log.warning(
"fill price backfill: slot=%s symbol=%s broker_order_id=%s "
"avg_price=%s backfill_price=%s reason=avg_price_backfill",
slot.slot_id, slot.symbol, slot.broker_order_id,
avg, limit,
slot.slot_id,
slot.symbol,
slot.broker_order_id,
avg,
limit,
)
return float(limit), True
# limit_price also absent/≤0: skip + strong WARN (REQ-095 AC-6 / REQ-096 AC-2)
@@ -280,16 +340,22 @@ class ExecutionService:
"fill price unresolvable: slot=%s symbol=%s broker_order_id=%s "
"avg_price=%s limit_price=%s reason=fill_price_unresolvable "
"(slot stays PENDING — operator attention required)",
slot.slot_id, slot.symbol, slot.broker_order_id,
avg, limit,
slot.slot_id,
slot.symbol,
slot.broker_order_id,
avg,
limit,
)
# dcy-033 / REQ-100: CRITICAL alarm — slot stuck in PENDING.
if self._alarm_emitter is not None:
self._alarm_emitter.publish_alarm(
"fill_price_unresolvable",
f"체결가 미해결: 슬롯 PENDING 고착 위험 (slot={slot.slot_id} symbol={slot.symbol})",
fields={"slot_id": str(slot.slot_id), "symbol": slot.symbol,
"broker_order_id": slot.broker_order_id},
fields={
"slot_id": str(slot.slot_id),
"symbol": slot.symbol,
"broker_order_id": slot.broker_order_id,
},
)
return None, False
@@ -318,12 +384,17 @@ class ExecutionService:
old_avg = float(slot.avg_price) if slot.avg_price else 0.0
new_avg = (
(old_qty * old_avg + filled * fill_price) / new_qty
if new_qty > 0 else fill_price
if new_qty > 0
else fill_price
)
updated = self._svc.transit_slot(
slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING,
reason="buy_filled", actor="execution_loop",
avg_price=new_avg, qty=new_qty,
slot.slot_id,
SlotState.PENDING_BUY,
SlotState.HOLDING,
reason="buy_filled",
actor="execution_loop",
avg_price=new_avg,
qty=new_qty,
)
event_type = "slot_filled_buy"
else: # PENDING_SELL — dcy-021/REQ-066: remainder>0 stays holding,
@@ -331,14 +402,20 @@ class ExecutionService:
total = self._balance_map().get(slot.symbol or "", 0)
if total > 0:
updated = self._svc.transit_slot(
slot.slot_id, SlotState.PENDING_SELL, SlotState.HOLDING,
reason="sell_filled_partial", actor="execution_loop",
slot.slot_id,
SlotState.PENDING_SELL,
SlotState.HOLDING,
reason="sell_filled_partial",
actor="execution_loop",
qty=float(total), # avg_price (cost basis) preserved
)
else:
updated = self._svc.transit_slot(
slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE,
reason="sell_filled", actor="execution_loop",
slot.slot_id,
SlotState.PENDING_SELL,
SlotState.INACTIVE,
reason="sell_filled",
actor="execution_loop",
)
event_type = "slot_filled_sell"
except SlotStateMismatchError:
@@ -347,9 +424,13 @@ class ExecutionService:
_log.info(
"order filled event=order_fill side=%s symbol=%s slot_id=%s "
"fill_price=%s filled_qty=%s fee=%s tax=%s",
side, slot.symbol, slot.slot_id,
fs.avg_price, fs.filled_qty,
getattr(fs, "fee", None), getattr(fs, "tax", None),
side,
slot.symbol,
slot.slot_id,
fs.avg_price,
fs.filled_qty,
getattr(fs, "fee", None),
getattr(fs, "tax", None),
)
# REQ-018: enqueue ledger work item AFTER the slot transition is durable.
# Slot transition is the critical path; ledger write is async (REQ-019).
@@ -389,26 +470,37 @@ class ExecutionService:
old_avg = float(slot.avg_price) if slot.avg_price else 0.0
new_avg = (
(old_qty * old_avg + filled * fill_price) / new_qty
if new_qty > 0 else fill_price
if new_qty > 0
else fill_price
)
updated = self._svc.transit_slot(
slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING,
reason="buy_partial_cancel", actor="execution_loop",
avg_price=new_avg, qty=new_qty,
slot.slot_id,
SlotState.PENDING_BUY,
SlotState.HOLDING,
reason="buy_partial_cancel",
actor="execution_loop",
avg_price=new_avg,
qty=new_qty,
)
event_type = "slot_filled_buy"
else: # PENDING_SELL: remainder = broker balance, cost basis unchanged.
total = self._balance_map().get(slot.symbol or "", 0)
if total > 0:
updated = self._svc.transit_slot(
slot.slot_id, SlotState.PENDING_SELL, SlotState.HOLDING,
reason="sell_partial_cancel", actor="execution_loop",
slot.slot_id,
SlotState.PENDING_SELL,
SlotState.HOLDING,
reason="sell_partial_cancel",
actor="execution_loop",
qty=float(total),
)
else:
updated = self._svc.transit_slot(
slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE,
reason="sell_partial_cancel", actor="execution_loop",
slot.slot_id,
SlotState.PENDING_SELL,
SlotState.INACTIVE,
reason="sell_partial_cancel",
actor="execution_loop",
)
event_type = "slot_filled_sell"
except SlotStateMismatchError:
@@ -431,13 +523,14 @@ class ExecutionService:
_log.warning(
"pending order stale: slot=%s status=%s broker_order_id=%s "
"age_sec=%.0f threshold_sec=%d (no state change — ops attention)",
slot.slot_id, slot.status.value, slot.broker_order_id,
age, self._stale_threshold_sec,
slot.slot_id,
slot.status.value,
slot.broker_order_id,
age,
self._stale_threshold_sec,
)
def _try_timeout_cancel(
self, slot, now: datetime, side: str, r: ExecutionResult
) -> None:
def _try_timeout_cancel(self, slot, now: datetime, side: str, r: ExecutionResult) -> None:
"""Send cancel_order when the slot-level timeout is exceeded (dcy-022/REQ-069).
Best-effort: a failure only logs a warning — the slot is NOT changed here.
@@ -446,6 +539,8 @@ class ExecutionService:
This design is safe across the cancel-fill race: if the cancel races with a
fill, the next poll sees is_filled and follows the normal fill path (FR-4).
"""
if str(slot.broker_order_id or "").startswith("submission-unknown:"):
return
if slot.timeout_sec is None or slot.updated_at is None:
return
age = (now - slot.updated_at).total_seconds()
@@ -463,14 +558,19 @@ class ExecutionService:
_log.warning(
"timeout cancel sent: slot=%s status=%s broker_order_id=%s "
"age_sec=%.0f timeout_sec=%d broker_accepted=%s",
slot.slot_id, slot.status.value, slot.broker_order_id,
age, slot.timeout_sec, accepted,
slot.slot_id,
slot.status.value,
slot.broker_order_id,
age,
slot.timeout_sec,
accepted,
)
except Exception: # noqa: BLE001 - best-effort; do not block the loop
_log.warning(
"timeout cancel failed (exception): slot=%s broker_order_id=%s "
"(will retry next tick if still pending)",
slot.slot_id, slot.broker_order_id,
slot.slot_id,
slot.broker_order_id,
)
def _emit_event(self, event_type: str, slot, routing_strategy_id: Optional[str] = None) -> None:
@@ -507,7 +607,9 @@ class ExecutionService:
_log.warning(
"ledger skip: slot=%s symbol=%s side=%s reason=missing_portfolio_id "
"(trade NOT recorded to portfolio-hub)",
slot.slot_id, slot.symbol, side,
slot.slot_id,
slot.symbol,
side,
)
return
if fs.execution_id is None:
@@ -516,13 +618,17 @@ class ExecutionService:
_log.warning(
"ledger skip: slot=%s portfolio_id=%s side=%s reason=missing_execution_id "
"(trade NOT recorded to portfolio-hub)",
slot.slot_id, slot.portfolio_id, side,
slot.slot_id,
slot.portfolio_id,
side,
)
return
# REQ-095 KD-3: use pre-resolved price when available; fall back to
# fs.avg_price for any direct callers that do not pass resolved_price.
price = resolved_price if resolved_price is not None else (
float(fs.avg_price) if fs.avg_price is not None else 0.0
price = (
resolved_price
if resolved_price is not None
else (float(fs.avg_price) if fs.avg_price is not None else 0.0)
)
trade = LedgerTrade(
asset_id=str(slot.portfolio_id),
@@ -546,6 +652,7 @@ class ExecutionService:
next poll handles the rollback via the normal is_terminal_reject path.
"""
from uuid import UUID
sid = slot_id if isinstance(slot_id, UUID) else UUID(str(slot_id))
slot = self._repo.fetch_slot(sid)
if slot is None:
@@ -567,12 +674,15 @@ class ExecutionService:
except Exception: # noqa: BLE001 - best-effort
_log.warning(
"operator cancel failed: slot=%s broker_order_id=%s",
slot.slot_id, slot.broker_order_id,
slot.slot_id,
slot.broker_order_id,
)
return "cancel_failed"
_log.warning(
"operator cancel sent: slot=%s status=%s broker_order_id=%s",
slot.slot_id, slot.status.value, slot.broker_order_id,
slot.slot_id,
slot.status.value,
slot.broker_order_id,
)
return "ok"
@@ -584,19 +694,29 @@ class ExecutionService:
# dcy-021: increase rejected — the prior holding is intact
# (broker still holds), so return to holding, not inactive.
self._svc.transit_slot(
slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING,
reason=reason, actor="execution_loop", qty=float(total),
slot.slot_id,
SlotState.PENDING_BUY,
SlotState.HOLDING,
reason=reason,
actor="execution_loop",
qty=float(total),
)
self._repo.clear_dispatch(slot.slot_id)
else: # initial buy rejected -> inactive (clears broker_order_id)
self._svc.transit_slot(
slot.slot_id, SlotState.PENDING_BUY, SlotState.INACTIVE,
reason=reason, actor="execution_loop",
slot.slot_id,
SlotState.PENDING_BUY,
SlotState.INACTIVE,
reason=reason,
actor="execution_loop",
)
else: # PENDING_SELL rejected -> back to holding (broker authoritative qty)
self._svc.transit_slot(
slot.slot_id, SlotState.PENDING_SELL, SlotState.HOLDING,
reason=reason, actor="execution_loop",
slot.slot_id,
SlotState.PENDING_SELL,
SlotState.HOLDING,
reason=reason,
actor="execution_loop",
qty=float(total) if total > 0 else None,
)
self._repo.clear_dispatch(slot.slot_id)
trex_core/domain/slot/in_memory_repository.py
+63 -28
View File
@@ -3,6 +3,7 @@
Simulates advisory lock via threading.Lock so contention tests are still
meaningful even without a real PostgreSQL instance.
"""
from __future__ import annotations
import threading
@@ -51,11 +52,10 @@ class InMemorySlotRepository:
"""Test seeding helper — insert a slot row directly."""
self._slots[slot.slot_id] = slot
def list_slots(
self, instance_key: str, status_filter: Optional[set[str]] = None
) -> list[Slot]:
def list_slots(self, instance_key: str, status_filter: Optional[set[str]] = None) -> list[Slot]:
return [
s for s in self._slots.values()
s
for s in self._slots.values()
if s.instance_key == instance_key
and (status_filter is None or s.status.value in status_filter)
]
@@ -79,52 +79,62 @@ class InMemorySlotRepository:
def count_active(self, instance_key: str) -> int:
return sum(
1 for s in self._slots.values()
1
for s in self._slots.values()
if s.instance_key == instance_key and s.status in self._ACTIVE
)
def list_holding_slots(self, instance_key: str) -> list[Slot]:
return [
s for s in self._slots.values()
s
for s in self._slots.values()
if s.instance_key == instance_key and s.status is SlotState.HOLDING
]
def find_managed_slot_by_symbol(self, instance_key: str, symbol: str) -> Optional[Slot]:
candidates = [
s for s in self._slots.values()
if s.instance_key == instance_key and s.symbol == symbol
s for s in self._slots.values() if s.instance_key == instance_key and s.symbol == symbol
]
# PostgreSQL parity: updated_at DESC NULLS LAST, slot_id ASC.
candidates.sort(key=lambda s: (
s.updated_at is None,
-s.updated_at.timestamp() if s.updated_at is not None else 0,
str(s.slot_id),
))
candidates.sort(
key=lambda s: (
s.updated_at is None,
-s.updated_at.timestamp() if s.updated_at is not None else 0,
str(s.slot_id),
)
)
return candidates[0] if candidates else None
def reconcile_universe(
self, instance_key: str, universe: dict[str, str],
*, allow_retire: bool = True,
self,
instance_key: str,
universe: dict[str, str],
*,
allow_retire: bool = True,
) -> dict[str, int]:
existing = {
s.portfolio_id: s for s in self._slots.values()
s.portfolio_id: s
for s in self._slots.values()
if s.instance_key == instance_key and s.portfolio_id is not None
}
now = datetime.now(tz=timezone.utc)
created = retired = tombstoned = 0
# backfill symbol on slots that were created before this field was populated
for s in self._slots.values():
if (s.instance_key == instance_key and s.portfolio_id in universe
and s.symbol is None):
if s.instance_key == instance_key and s.portfolio_id in universe and s.symbol is None:
self._slots[s.slot_id] = replace(s, symbol=universe[s.portfolio_id], updated_at=now)
# create asset slots for newly-appeared universe assets (idempotent)
for pid, symbol in universe.items():
if pid not in existing:
sid = uuid4()
self._slots[sid] = Slot(
slot_id=sid, status=SlotState.INACTIVE,
instance_key=instance_key, portfolio_id=pid, symbol=symbol,
created_at=now, updated_at=now,
slot_id=sid,
status=SlotState.INACTIVE,
instance_key=instance_key,
portfolio_id=pid,
symbol=symbol,
created_at=now,
updated_at=now,
)
created += 1
# retire / tombstone assets that left the universe (skipped under the
@@ -155,7 +165,9 @@ class InMemorySlotRepository:
_PENDING = (SlotState.PENDING_BUY, SlotState.PENDING_SELL)
def list_undispatched_pending(
self, instance_key: str = "default", limit: int = 64,
self,
instance_key: str = "default",
limit: int = 64,
) -> list[Slot]:
"""Pending slots not yet dispatched (broker_order_id IS NULL).
@@ -163,9 +175,13 @@ class InMemorySlotRepository:
never dispatch each other's orders. List-based (vs single-row claim) so
one skipped sell (broker flat) cannot head-block the dispatch queue."""
out = [
s for s in self._slots.values()
if (s.instance_key == instance_key and s.status in self._PENDING
and s.broker_order_id is None)
s
for s in self._slots.values()
if (
s.instance_key == instance_key
and s.status in self._PENDING
and s.broker_order_id is None
)
]
return out[:limit]
@@ -179,12 +195,31 @@ class InMemorySlotRepository:
)
return True
def replace_dispatch(
self,
slot_id: UUID,
expected_broker_order_id: str,
broker_order_id: str,
) -> bool:
"""Replace a result-unknown sentinel with the broker-confirmed order ID."""
s = self._slots.get(slot_id)
if s is None or s.broker_order_id != expected_broker_order_id:
return False
self._slots[slot_id] = replace(
s, broker_order_id=broker_order_id, updated_at=datetime.now(tz=timezone.utc)
)
return True
def list_dispatched_pending(self, instance_key: str = "default") -> list[Slot]:
"""Pending slots already dispatched (broker_order_id NOT NULL) — poll targets."""
return [
s for s in self._slots.values()
if (s.instance_key == instance_key and s.status in self._PENDING
and s.broker_order_id is not None)
s
for s in self._slots.values()
if (
s.instance_key == instance_key
and s.status in self._PENDING
and s.broker_order_id is not None
)
]
def clear_dispatch(self, slot_id: UUID) -> None: