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

feat(intents): dcy-011 buy intent portfolio_id 필수 계약 + ledger skip 가시화 (BAT-015)

Browse Source 
dcy-011 전 주기 delivery. issue #6(buy 식별자 정정) + issue #11(ledger silent skip) 흡수.
거버넌스: 기존 Implemented REQ는 change-req, 신규 기능만 신규 REQ(OQ-3).

- REQ-007 (change-req): buy admission portfolio_id 필수화 + SnapshotStore symbol
  resolve + unknown_portfolio 거부 (hard cutover, OQ-1/OQ-2)
- REQ-021 (change-req): buy 메시지 payload symbol 제거
- REQ-018 (change-req): _record_ledger skip 가시화 (WARN + ledger_skipped 카운터)
- REQ-055 (신규): closed↔ledger_outbox anti-join 운영 점검 find_ledger_gap_slots

코드: intents.py(BuyIntent portfolio_id 필수 + resolve_symbol 주입 + _admit_buy
reject 분기), messaging_handlers.py, runtime/daemon.py(_resolve_symbol 배선),
execution.py(skip WARN 로그), adapters/persistence/ops_checks.py(신규 anti-join).

검증: 단위 282 passed/32 skipped, 인프라 게이트(PG+RabbitMQ) 314 passed, ruff clean.
BAT-015(standard) release-candidate. srs/index Next ID → REQ-056.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
genichin
2026-06-08 12:57:51 +09:00
co-authored by Claude Opus 4.8
parent a24ebb242e
commit aee3512383
27 changed files with 729 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
trex_core/adapters/persistence/ops_checks.py
+47
View File
@@ -0,0 +1,47 @@
"""Operational integrity checks — read-only (REQ-055 / dcy-011 FR-5, issue #11).
`find_ledger_gap_slots` detects `closed` slots that produced NO ledger_outbox
row (the silent-skip symptom): either `portfolio_id IS NULL` (skip at fill) or
no outbox row carries the slot's portfolio_id as `asset_id`. Read-only — no
schema change; `daily_slot` is durable so this is a permanent detector even
without an outbox marker row (dcy-010 OQ-4).
"""
from __future__ import annotations
from datetime import date
from typing import Any, Optional
from trex_core.adapters.persistence.postgresql import _require_psycopg
_LEDGER_GAP_SQL = """
SELECT s.slot_id, s.trading_date, s.symbol, s.portfolio_id, s.broker_order_id
FROM daily_slot s
WHERE s.status = 'closed'
AND (%(td)s::date IS NULL OR s.trading_date = %(td)s)
AND (
s.portfolio_id IS NULL
OR NOT EXISTS (
SELECT 1 FROM ledger_outbox o WHERE o.asset_id = s.portfolio_id
)
)
ORDER BY s.trading_date, s.slot_id
"""
_COLS = ("slot_id", "trading_date", "symbol", "portfolio_id", "broker_order_id")
def find_ledger_gap_slots(
dsn: str,
*,
trading_date: Optional[date] = None,
psycopg_module: Any = None,
) -> list[dict]:
"""Return closed slots with no corresponding ledger_outbox row (FR-5).
Pass `trading_date` to scope to one day; omit for all days.
"""
psycopg = _require_psycopg(psycopg_module)
with psycopg.connect(dsn) as conn:
with conn.cursor() as cur:
cur.execute(_LEDGER_GAP_SQL, {"td": trading_date})
return [dict(zip(_COLS, row)) for row in cur.fetchall()]
trex_core/application/execution.py
+16 -2
View File
@@ -18,6 +18,7 @@ single producer for ledger work items (NFR-3, single record subject).
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Callable, Optional
@@ -27,6 +28,8 @@ from trex_core.application.ledger_outbox import LedgerOutbox
from trex_core.application.ledger_port import LedgerTrade
from trex_core.domain.slot import SlotService, SlotState, SlotStateMismatchError
_log = logging.getLogger(__name__)
# REQ-021/023: event emission hook. Runtime wires a callable that publishes
# slot transition events through MessagingTransport. Application layer stays
# transport-agnostic (NFR-1). Slot type is duck-typed to keep the application
@@ -155,12 +158,23 @@ class ExecutionService:
if self._ledger_outbox is None:
return
if not slot.portfolio_id:
# REQ-018 AC-4: slot transition still proceeded; record skip for ops.
# dcy-011 / REQ-018 (issue #11): ledger skip must NOT be silent.
# Slot transition still proceeded; surface for ops (FR-4).
r.ledger_skipped += 1
_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,
)
return
if fs.execution_id is None:
# Broker contract regression — log via counter; slot already transitioned.
# Broker contract regression — slot already transitioned; surface for ops.
r.ledger_skipped += 1
_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,
)
return
price = float(fs.avg_price) if fs.avg_price is not None else 0.0
trade = LedgerTrade(
trex_core/application/intents.py
+25 -4
View File
@@ -45,15 +45,19 @@ class IntentDecision:
@dataclass(frozen=True)
class BuyIntent:
# dcy-011 / REQ-007: portfolio_id (= asset_id) is the required canonical key.
# symbol is no longer a client input — it is resolved server-side from the
# universe (SnapshotStore) at admission. It remains here as the resolved
# value carried onto the slot.
intent_id: str
trading_date: date
symbol: str
portfolio_id: str
symbol: Optional[str] = None
qty: Optional[float] = None
allocated_capital: Optional[float] = None
entry_price: Optional[float] = None
strategy_id: Optional[str] = None
sell_strategy_id: Optional[str] = None
portfolio_id: Optional[str] = None
forced_close_at: Optional[datetime] = None
stop_loss_price: Optional[Decimal] = None # REQ-035
take_profit_price: Optional[Decimal] = None # REQ-035
@@ -77,10 +81,15 @@ class IntentService:
slot_service: SlotService,
slot_repo: SlotRepository,
intent_store: IntentStore,
resolve_symbol=None,
) -> None:
self._svc = slot_service
self._repo = slot_repo
self._intents = intent_store
# dcy-011 / REQ-007: portfolio_id -> symbol resolver (e.g. SnapshotStore-backed).
# callable(portfolio_id) -> Optional[str]. When set, admission resolves symbol
# from the universe and rejects unknown_portfolio if absent.
self._resolve_symbol = resolve_symbol
# ------------------------------------------------------------
def lookup(self, intent_id: str) -> Optional[IntentDecision]:
@@ -139,6 +148,18 @@ class IntentService:
return IntentDecision(intent_id, rec.status, rec.slot_id, rec.reason)
def _admit_buy(self, intent: BuyIntent) -> IntentDecision:
# dcy-011 / REQ-007: portfolio_id is required; symbol is resolved from the
# universe (no client symbol). FR-1/FR-2/FR-3.
if not intent.portfolio_id:
return IntentDecision(intent.intent_id, "rejected", None, "portfolio_id_required")
symbol = intent.symbol
if self._resolve_symbol is not None:
symbol = self._resolve_symbol(intent.portfolio_id)
if not symbol:
return IntentDecision(intent.intent_id, "rejected", None, "unknown_portfolio")
elif not symbol:
# No resolver wired and no symbol carried — cannot determine the order target.
return IntentDecision(intent.intent_id, "rejected", None, "unknown_portfolio")
# Retry on the last-idle-slot race: another intent may claim the slot
# between our find and transit (SlotStateMismatchError).
for _ in range(_MAX_IDLE_RETRY):
@@ -152,7 +173,7 @@ class IntentService:
SlotState.PENDING_BUY,
reason="buy_intent",
actor="intent_api",
symbol=intent.symbol,
symbol=symbol,
qty=intent.qty,
allocated_capital=intent.allocated_capital,
entry_price=intent.entry_price,
@@ -162,7 +183,7 @@ class IntentService:
forced_close_at=intent.forced_close_at,
stop_loss_price=intent.stop_loss_price,
take_profit_price=intent.take_profit_price,
extra={"intent_id": intent.intent_id, "symbol": intent.symbol},
extra={"intent_id": intent.intent_id, "symbol": symbol},
)
except SlotStateMismatchError:
continue # slot was claimed concurrently; try another idle slot
trex_core/application/messaging_handlers.py
+4 -2
View File
@@ -71,16 +71,18 @@ def make_intent_rpc_handler(
intent_type = str(body.get("intent_type") or "").lower()
try:
if intent_type == "buy":
# dcy-011 / REQ-021: buy payload carries portfolio_id (required);
# symbol is NOT a client field — admission resolves it from the
# universe. portfolio_id absence is rejected at admission (FR-1).
buy = BuyIntent(
intent_id=intent_id,
trading_date=_parse_date(body.get("trading_date")),
symbol=str(body.get("symbol") or ""),
portfolio_id=_opt_str(body.get("portfolio_id")) or "",
qty=_opt_float(body.get("qty")),
allocated_capital=_opt_float(body.get("allocated_capital")),
entry_price=_opt_float(body.get("entry_price")),
strategy_id=_opt_str(body.get("strategy_id")),
sell_strategy_id=_opt_str(body.get("sell_strategy_id")),
portfolio_id=_opt_str(body.get("portfolio_id")),
forced_close_at=_opt_dt(body.get("forced_close_at")),
stop_loss_price=_opt_decimal(body.get("stop_loss_price")),
take_profit_price=_opt_decimal(body.get("take_profit_price")),
trex_core/runtime/daemon.py
+11 -1
View File
@@ -140,7 +140,10 @@ class Daemon:
self.slot_repo = PostgreSQLSlotRepository(config.dsn)
self.intent_store = PostgreSQLIntentStore(config.dsn)
self.slot_service = SlotService(self.slot_repo)
self.intent_service = IntentService(self.slot_service, self.slot_repo, self.intent_store)
self.intent_service = IntentService(
self.slot_service, self.slot_repo, self.intent_store,
resolve_symbol=self._resolve_symbol,
)
self.broker = _build_broker(config)
self.outbox = PostgreSQLLedgerOutbox(config.dsn)
self.ledger_writer = LedgerWriter(self.outbox, _build_ledger(config))
@@ -185,6 +188,13 @@ class Daemon:
config=config.safety_sell,
)
# ------------------------------------------------------------ buy symbol resolve
def _resolve_symbol(self, portfolio_id: str):
"""dcy-011 / REQ-007: resolve symbol from the universe snapshot for a
portfolio_id at buy admission (None → unknown_portfolio reject)."""
asset = self.snapshot_store.get(portfolio_id)
return asset.symbol if asset is not None else None
# ------------------------------------------------------------ event hook
def _make_event_emitter(self):
def emit(event_type: str, slot) -> None: