feat(slot): BAT-030 — 자산 중심 cutover 완성 (dcy-020 BAT-δ, REQ-056·057·062·064·065)

M1 literal inactive 전면화: SlotState 5-state(idle/closed 별칭 제거),
마이그레이션 010(CHECK 교체+행 rewrite), ops_checks 를 transition_log 기반 재작성.
M2 집행 매도 수량 브로커 정본(D3): dispatch 매도 qty=broker hldg_qty,
미러 fallback 없음, 브로커 0 → skip+WARN, 리스트 기반 dispatch(head-block 방지).
M3 컬럼 drop + legacy pool 제거: 마이그레이션 011(trading_date·entry_count DROP),
repo 계약 instance 스코프 개편(집행 큐 포함), preallocation 폐기(데몬 7스레드),
admission 자산 중심 단일 경로(slot_id 정본), capacity 재정의(available=M−active),
daily_symbol_entries 코드 경로 제거.
M4 cutover 도구(D1·D4): trex-core-cutover(전체 rebuild, preflight/브로커 carry-over/
--allow-unmapped/--promote-unique/dry-run) + PG 리허설 테스트 8건.
M5 baseline 마감: data-model/runtime-flows/project-structure/interface-contract
자산 중심 최종 갱신 + run-trex-core-daemon 스킬 정합.

검증: 단위 367 passed + ruff clean; PG-gated 29 passed(010/011 round-trip·드리프트
가드·cutover 리허설); RabbitMQ 21 passed; run-the-app e2e(test DB+실 MQ, M=1):
reconcile 슬롯 생성→slot_id 매수→dedup→capacity_full→fill→already_active→
매도→inactive(자산 바인딩 보존)→영구 슬롯 재활성 전 구간 green.
실 cutover 는 운영자 실행(REL 단계, D4).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-10 15:16:15 +09:00
co-authored by Claude Opus 4.8
parent cec34be5f4
commit 7d727824dc
67 changed files with 1584 additions and 1862 deletions
+17 -13
View File
@@ -1,10 +1,13 @@
"""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).
`find_ledger_gap_slots` detects sold slots that produced NO ledger_outbox row
(the silent-skip symptom). dcy-020 (BAT-030): the pool-model `closed` status no
longer exists — a sold slot returns to `inactive` and its occupancy is cleared —
so the detector keys off the append-only `slot_transition_log` (reason
`sell_filled` / `sell_partial_cancel`) instead of the slot row's status. A gap
is a sell fill whose slot has `portfolio_id IS NULL` (skip at fill) or whose
portfolio_id never appears as a ledger_outbox `asset_id`. Read-only — no schema
change; the transition log is durable so this is a permanent detector.
"""
from __future__ import annotations
@@ -14,20 +17,21 @@ 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)
SELECT DISTINCT s.slot_id, s.symbol, s.portfolio_id, s.broker_order_id, t.ts
FROM slot_transition_log t
JOIN daily_slot s ON s.slot_id = t.slot_id
WHERE t.reason IN ('sell_filled', 'sell_partial_cancel')
AND (%(td)s::date IS NULL OR (t.ts AT TIME ZONE 'UTC')::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
ORDER BY t.ts, s.slot_id
"""
_COLS = ("slot_id", "trading_date", "symbol", "portfolio_id", "broker_order_id")
_COLS = ("slot_id", "symbol", "portfolio_id", "broker_order_id", "sold_at")
def find_ledger_gap_slots(
@@ -36,9 +40,9 @@ def find_ledger_gap_slots(
trading_date: Optional[date] = None,
psycopg_module: Any = None,
) -> list[dict]:
"""Return closed slots with no corresponding ledger_outbox row (FR-5).
"""Return sell fills with no corresponding ledger_outbox row (FR-5).
Pass `trading_date` to scope to one day; omit for all days.
Pass `trading_date` to scope to one (UTC) day of sell fills; omit for all.
"""
psycopg = _require_psycopg(psycopg_module)
with psycopg.connect(dsn) as conn:
+54 -121
View File
@@ -13,7 +13,7 @@ from __future__ import annotations
import json
from contextlib import contextmanager
from datetime import date, datetime, timezone
from datetime import datetime, timezone
from typing import Any, Iterator, Optional
from uuid import UUID, uuid4
@@ -27,10 +27,10 @@ from trex_core.application.ports import IntentRecord
from trex_core.domain.slot import Slot, SlotNotFoundError, SlotState, SlotTransitionRecord
_SLOT_COLUMNS = (
"slot_id, trading_date, status, symbol, qty, avg_price, limit_price, "
"slot_id, status, symbol, qty, avg_price, limit_price, "
"allocated_capital, order_id, strategy_id, broker_order_id, portfolio_id, "
"sell_strategy_id, forced_close_at, stop_loss_price, take_profit_price, "
"created_at, updated_at, entry_count, instance_key, retired_at"
"created_at, updated_at, instance_key, retired_at"
)
@@ -65,34 +65,12 @@ class PostgreSQLSlotRepository:
conn.rollback()
raise
def pre_allocate_slots(self, trading_date: date, count: int) -> int:
with self._psycopg.connect(self._dsn, autocommit=False) as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT COUNT(*) FROM daily_slot WHERE trading_date = %s", (trading_date,)
)
row = cur.fetchone()
if row and row[0] > 0:
conn.commit()
return 0
now = datetime.now(tz=timezone.utc)
rows = [
(uuid4(), trading_date, SlotState.IDLE.value, now, now) for _ in range(count)
]
cur.executemany(
"INSERT INTO daily_slot (slot_id, trading_date, status, created_at, updated_at) "
"VALUES (%s, %s, %s, %s, %s)",
rows,
)
conn.commit()
return count
def list_slots(self, trading_date: date) -> list[Slot]:
def list_slots(self, instance_key: str) -> list[Slot]:
with self._psycopg.connect(self._dsn) as conn:
with conn.cursor() as cur:
cur.execute(
f"SELECT {_SLOT_COLUMNS} FROM daily_slot WHERE trading_date = %s",
(trading_date,),
f"SELECT {_SLOT_COLUMNS} FROM daily_slot WHERE instance_key = %s",
(instance_key,),
)
return [self._row_to_slot(r) for r in cur.fetchall()]
@@ -105,32 +83,6 @@ class PostgreSQLSlotRepository:
row = cur.fetchone()
return self._row_to_slot(row) if row else None
def find_idle_slot(self, trading_date: date) -> Optional[Slot]:
with self._psycopg.connect(self._dsn) as conn:
with conn.cursor() as cur:
cur.execute(
f"SELECT {_SLOT_COLUMNS} FROM daily_slot "
"WHERE trading_date = %s AND status = 'idle' LIMIT 1",
(trading_date,),
)
row = cur.fetchone()
return self._row_to_slot(row) if row else None
def find_allocatable_slot(self, trading_date: date) -> Optional[Slot]:
# dcy-019 / REQ-006: idle ∪ closed candidates, ordered by entry_count asc
# → updated_at asc → slot_id (deterministic, OQ-2). Concurrency on the
# last candidate handled by the caller (advisory lock + retry, NFR-1).
with self._psycopg.connect(self._dsn) as conn:
with conn.cursor() as cur:
cur.execute(
f"SELECT {_SLOT_COLUMNS} FROM daily_slot "
"WHERE trading_date = %s AND status IN ('idle','closed') "
"ORDER BY entry_count ASC, updated_at ASC, slot_id ASC LIMIT 1",
(trading_date,),
)
row = cur.fetchone()
return self._row_to_slot(row) if row else None
# --- asset-centric (dcy-020 / BAT-028) ---
def find_slot_by_portfolio(self, instance_key: str, portfolio_id: str) -> Optional[Slot]:
with self._psycopg.connect(self._dsn) as conn:
@@ -165,7 +117,7 @@ class PostgreSQLSlotRepository:
return [self._row_to_slot(r) for r in cur.fetchall()]
def reconcile_universe(
self, instance_key: str, trading_date: date, universe_portfolio_ids: set[str],
self, instance_key: str, universe_portfolio_ids: set[str],
*, allow_retire: bool = True,
) -> dict[str, int]:
now = datetime.now(tz=timezone.utc)
@@ -184,11 +136,11 @@ class PostgreSQLSlotRepository:
continue
cur.execute(
"INSERT INTO daily_slot "
"(slot_id, trading_date, status, instance_key, portfolio_id, "
"created_at, updated_at) VALUES (%s, %s, 'idle', %s, %s, %s, %s) "
"(slot_id, status, instance_key, portfolio_id, "
"created_at, updated_at) VALUES (%s, 'inactive', %s, %s, %s, %s) "
"ON CONFLICT (instance_key, portfolio_id) "
"WHERE portfolio_id IS NOT NULL DO NOTHING",
(uuid4(), trading_date, instance_key, pid, now, now),
(uuid4(), instance_key, pid, now, now),
)
created += cur.rowcount
# retire / tombstone assets that left the universe (skipped under
@@ -217,7 +169,7 @@ class PostgreSQLSlotRepository:
"limit_price=%s, allocated_capital=%s, order_id=%s, strategy_id=%s, "
"broker_order_id=%s, portfolio_id=%s, sell_strategy_id=%s, "
"forced_close_at=%s, stop_loss_price=%s, take_profit_price=%s, "
"entry_count=%s, instance_key=%s, retired_at=%s, "
"instance_key=%s, retired_at=%s, "
"updated_at=%s WHERE slot_id=%s",
(
slot.status.value, slot.symbol, slot.qty, slot.avg_price,
@@ -225,7 +177,7 @@ class PostgreSQLSlotRepository:
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.entry_count, slot.instance_key, slot.retired_at,
slot.instance_key, slot.retired_at,
slot.updated_at or datetime.now(tz=timezone.utc), slot.slot_id,
),
)
@@ -248,43 +200,23 @@ class PostgreSQLSlotRepository:
)
conn.commit()
def increment_symbol_entry(self, trading_date: date, symbol: str) -> int:
with self._psycopg.connect(self._dsn, autocommit=False) as conn:
with conn.cursor() as cur:
cur.execute(
"INSERT INTO daily_symbol_entries (trading_date, symbol, entry_count, last_entry_at) "
"VALUES (%s, %s, 1, %s) "
"ON CONFLICT (trading_date, symbol) DO UPDATE "
"SET entry_count = daily_symbol_entries.entry_count + 1, "
" last_entry_at = EXCLUDED.last_entry_at RETURNING entry_count",
(trading_date, symbol, datetime.now(tz=timezone.utc)),
)
row = cur.fetchone()
conn.commit()
return int(row[0]) if row else 0
def get_symbol_entry_count(self, trading_date: date, symbol: str) -> int:
with self._psycopg.connect(self._dsn) as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT entry_count FROM daily_symbol_entries "
"WHERE trading_date = %s AND symbol = %s",
(trading_date, symbol),
)
row = cur.fetchone()
return int(row[0]) if row else 0
# --- order execution queue (BAT-004 / REQ-013, option A) ---
def find_pending_for_dispatch(self) -> Optional[Slot]:
def list_undispatched_pending(
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
# claim) so one skipped sell (broker flat) cannot head-block the queue.
with self._psycopg.connect(self._dsn) as conn:
with conn.cursor() as cur:
cur.execute(
f"SELECT {_SLOT_COLUMNS} FROM daily_slot "
"WHERE status IN ('pending_buy','pending_sell') AND broker_order_id IS NULL "
"ORDER BY updated_at LIMIT 1 FOR UPDATE SKIP LOCKED"
"WHERE instance_key = %s "
"AND status IN ('pending_buy','pending_sell') AND broker_order_id IS NULL "
"ORDER BY updated_at LIMIT %s",
(instance_key, limit),
)
row = cur.fetchone()
return self._row_to_slot(row) if row else None
return [self._row_to_slot(r) for r in cur.fetchall()]
def mark_dispatched(self, slot_id: UUID, broker_order_id: str) -> bool:
with self._psycopg.connect(self._dsn, autocommit=False) as conn:
@@ -298,12 +230,14 @@ class PostgreSQLSlotRepository:
conn.commit()
return ok
def list_dispatched_pending(self) -> list[Slot]:
def list_dispatched_pending(self, instance_key: str = "default") -> list[Slot]:
with self._psycopg.connect(self._dsn) as conn:
with conn.cursor() as cur:
cur.execute(
f"SELECT {_SLOT_COLUMNS} FROM daily_slot "
"WHERE status IN ('pending_buy','pending_sell') AND broker_order_id IS NOT NULL"
"WHERE instance_key = %s "
"AND status IN ('pending_buy','pending_sell') AND broker_order_id IS NOT NULL",
(instance_key,),
)
return [self._row_to_slot(r) for r in cur.fetchall()]
@@ -316,12 +250,13 @@ class PostgreSQLSlotRepository:
)
conn.commit()
def count_slots_by_status(self, trading_date: date) -> dict[SlotState, int]:
def count_slots_by_status(self, instance_key: str) -> dict[SlotState, int]:
with self._psycopg.connect(self._dsn) as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT status, COUNT(*) FROM daily_slot WHERE trading_date = %s GROUP BY status",
(trading_date,),
"SELECT status, COUNT(*) FROM daily_slot WHERE instance_key = %s "
"GROUP BY status",
(instance_key,),
)
return {SlotState(r[0]): int(r[1]) for r in cur.fetchall()}
@@ -330,33 +265,31 @@ class PostgreSQLSlotRepository:
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) trading_date(1) status(2) symbol(3) qty(4)
# avg_price(5) limit_price(6) allocated_capital(7) order_id(8)
# strategy_id(9) broker_order_id(10) portfolio_id(11) sell_strategy_id(12)
# forced_close_at(13) stop_loss_price(14) take_profit_price(15)
# created_at(16) updated_at(17) entry_count(18)
# Column order: slot_id(0) status(1) symbol(2) qty(3) avg_price(4)
# limit_price(5) allocated_capital(6) order_id(7) strategy_id(8)
# broker_order_id(9) portfolio_id(10) sell_strategy_id(11)
# forced_close_at(12) stop_loss_price(13) take_profit_price(14)
# created_at(15) updated_at(16) instance_key(17) retired_at(18)
return Slot(
slot_id=row[0] if isinstance(row[0], UUID) else UUID(str(row[0])),
trading_date=row[1],
status=SlotState(row[2]),
symbol=row[3],
qty=f(row[4]),
avg_price=f(row[5]),
limit_price=f(row[6]),
allocated_capital=f(row[7]),
order_id=row[8],
strategy_id=row[9],
broker_order_id=row[10],
portfolio_id=row[11],
sell_strategy_id=row[12],
forced_close_at=row[13],
stop_loss_price=d(row[14]),
take_profit_price=d(row[15]),
created_at=row[16],
updated_at=row[17],
entry_count=int(row[18]) if row[18] is not None else 0,
instance_key=row[19] if row[19] is not None else "default",
retired_at=row[20],
status=SlotState(row[1]),
symbol=row[2],
qty=f(row[3]),
avg_price=f(row[4]),
limit_price=f(row[5]),
allocated_capital=f(row[6]),
order_id=row[7],
strategy_id=row[8],
broker_order_id=row[9],
portfolio_id=row[10],
sell_strategy_id=row[11],
forced_close_at=row[12],
stop_loss_price=d(row[13]),
take_profit_price=d(row[14]),
created_at=row[15],
updated_at=row[16],
instance_key=row[17] if row[17] is not None else "default",
retired_at=row[18],
)
@@ -27,8 +27,11 @@ CREATE TABLE IF NOT EXISTS daily_slot (
updated_at TIMESTAMPTZ NOT NULL DEFAULT (now() AT TIME ZONE 'UTC')
);
CREATE INDEX IF NOT EXISTS idx_daily_slot_trading_date_status
ON daily_slot (trading_date, status);
-- BAT-030 / dcy-020: the (trading_date, status) index is superseded by a
-- status index — trading_date is dropped at migration 011 and this file is
-- re-applied on every boot, so it must not reference the dropped column.
CREATE INDEX IF NOT EXISTS idx_daily_slot_status
ON daily_slot (status);
-- daily_symbol_entries — per-day re-entry counter (decision input).
CREATE TABLE IF NOT EXISTS daily_symbol_entries (
@@ -1,7 +1,6 @@
-- BAT-025 / dcy-019 / REQ-006: slot-scoped entry_count on daily_slot.
-- Tracks how many times a slot has entered pending_buy (lifetime, per trading
-- day). Drives the buy-admission selector's fairness ordering (idle ∪ closed,
-- least-entered first) and enables closed-slot reallocation (issue #18).
-- Distinct from daily_symbol_entries.entry_count (per-symbol cap, unchanged).
-- Additive + idempotent: existing rows default to 0 (no prior entries counted).
ALTER TABLE daily_slot ADD COLUMN IF NOT EXISTS entry_count INTEGER NOT NULL DEFAULT 0;
-- BAT-025 / dcy-019 / REQ-006: added slot-scoped entry_count to daily_slot.
-- BAT-030 / dcy-020 / REQ-056: entry_count was DROPPED at the asset-centric
-- cutover (migration 011). Because the migration set is re-applied on every
-- boot, the original ADD COLUMN here would resurrect the column each run —
-- so this file is intentionally a no-op now (kept for numbering continuity).
SELECT 1;
@@ -0,0 +1,11 @@
-- BAT-030 / dcy-020 / REQ-057: literal `inactive` 전면화 (D2).
-- The pool-model `idle` (never-bought) and `closed` (sold, flat) statuses are
-- collapsed into the asset-centric `inactive`. Rewrites any remaining rows and
-- swaps the status CHECK constraint. Idempotent: the UPDATE matches nothing on
-- re-run and the constraint is dropped/re-added by name each time.
UPDATE daily_slot SET status = 'inactive' WHERE status IN ('idle', 'closed');
ALTER TABLE daily_slot DROP CONSTRAINT IF EXISTS daily_slot_status_check;
ALTER TABLE daily_slot ADD CONSTRAINT daily_slot_status_check CHECK (
status IN ('inactive','pending_buy','holding','pending_sell','error')
);
@@ -0,0 +1,9 @@
-- BAT-030 / dcy-020 / REQ-056: drop pool-model columns (D2, 비가역).
-- The permanent asset slot is keyed by (instance_key, portfolio_id) and lives
-- across trading days — `trading_date` (day-scoped pool identity) and
-- `entry_count` (pool fairness ordering) no longer apply. Time attributes live
-- on slot_transition_log / ledger rows. DROP COLUMN also removes any dependent
-- index. Idempotent via IF EXISTS. Rollback = restore from the pre-cutover
-- backup (see BAT-030 design §4).
ALTER TABLE daily_slot DROP COLUMN IF EXISTS entry_count;
ALTER TABLE daily_slot DROP COLUMN IF EXISTS trading_date;
+40 -11
View File
@@ -4,8 +4,12 @@ A single background execution loop (invariant A single writer, invariant B
single broker session) that, per tick:
dispatch: claim pending slots with broker_order_id IS NULL (option A queue),
submit the order, record broker_order_id (idempotency guard).
dcy-020/REQ-062 (BAT-030): a SELL is dispatched with the broker
balance quantity (hldg_qty) as the single source of truth — never
the slot's mirror qty. Broker flat (qty 0) -> skip + WARN
(divergence surfaces via reconcile, REQ-062 A).
poll: observe fills on dispatched pending slots and transition the slot
(pending_buy->holding / pending_sell->closed), or roll back on reject.
(pending_buy->holding / pending_sell->inactive), or roll back on reject.
On a successful fill, enqueue a LedgerWorkItem (slot transition is
kept ON the critical path; the ledger write is eventually-consistent
via a separate writer — REQ-019).
@@ -47,6 +51,7 @@ class ExecutionResult:
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
class ExecutionService:
@@ -58,12 +63,16 @@ class ExecutionService:
ledger_outbox: Optional[LedgerOutbox] = None,
event_emitter: Optional[SlotEventEmitter] = None,
stale_pending_threshold_sec: Optional[int] = None,
instance_key: str = "default",
) -> None:
self._svc = slot_service
self._repo = repo
self._broker = broker
self._ledger_outbox = ledger_outbox
self._event_emitter = event_emitter
# dcy-020/REQ-056 (BAT-030): dispatch/poll are instance-scoped so two
# daemons sharing a DB never touch each other's orders.
self._instance_key = instance_key
# dcy-013 / REQ-014 (issue #9): Alert-only pending-order stale watchdog.
# <=0 or None disables it. When set, a dispatched-pending slot whose
# last update is older than the threshold is surfaced (WARN + counter)
@@ -93,12 +102,32 @@ class ExecutionService:
# ------------------------------------------------------------
def _dispatch(self, r: ExecutionResult) -> None:
for _ in range(_MAX_DISPATCH_PER_TICK):
slot = self._repo.find_pending_for_dispatch()
if slot is None:
return
sell_balance: Optional[dict[str, int]] = None # lazy, once per tick
for slot in self._repo.list_undispatched_pending(
self._instance_key, _MAX_DISPATCH_PER_TICK
):
side = "buy" if slot.status is SlotState.PENDING_BUY else "sell"
qty = int(slot.qty or 0)
if side == "sell":
# dcy-020 / REQ-062 (BAT-030, D3): the sell quantity is the
# BROKER balance (hldg_qty) — the slot's mirror qty is never
# trusted. Broker flat -> skip dispatch + WARN; the divergence
# surfaces via reconcile_core_holdings (REQ-062 A). No mirror
# fallback by design.
if sell_balance is None:
sell_balance = {
b.symbol: int(b.hldg_qty) for b in self._broker.get_balance()
}
qty = sell_balance.get(slot.symbol or "", 0)
if qty <= 0:
r.sell_skipped_no_balance += 1
_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,
)
continue
else:
qty = int(slot.qty or 0)
# dcy-012 / REQ-013: side-agnostic — use the slot's pending-order
# limit_price only. A sell never reuses a buy limit (issue #10):
# limit_price is cleared on entering HOLDING, so a market sell
@@ -123,7 +152,7 @@ class ExecutionService:
def _poll(self, r: ExecutionResult) -> None:
now = datetime.now(tz=timezone.utc)
for slot in self._repo.list_dispatched_pending():
for slot in self._repo.list_dispatched_pending(self._instance_key):
side = "buy" if slot.status is SlotState.PENDING_BUY else "sell"
fs = self._broker.poll_fill(slot.broker_order_id, side=side, symbol=slot.symbol or "")
if fs.is_filled:
@@ -153,9 +182,9 @@ class ExecutionService:
avg_price=fs.avg_price, qty=float(fs.filled_qty) or slot.qty,
)
event_type = "slot_filled_buy"
else: # PENDING_SELL
else: # PENDING_SELL — asset slot returns to inactive (dcy-020/REQ-057)
updated = self._svc.transit_slot(
slot.slot_id, SlotState.PENDING_SELL, SlotState.CLOSED,
slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE,
reason="sell_filled", actor="execution_loop",
)
event_type = "slot_filled_sell"
@@ -271,9 +300,9 @@ class ExecutionService:
try:
if slot.status is SlotState.PENDING_BUY:
self._svc.transit_slot(
slot.slot_id, SlotState.PENDING_BUY, SlotState.IDLE,
slot.slot_id, SlotState.PENDING_BUY, SlotState.INACTIVE,
reason=reason, actor="execution_loop",
) # idle reset clears broker_order_id in the domain
) # inactive reset clears broker_order_id in the domain
else: # PENDING_SELL rejected -> back to holding, clear order for re-dispatch
self._svc.transit_slot(
slot.slot_id, SlotState.PENDING_SELL, SlotState.HOLDING,
+29 -80
View File
@@ -1,19 +1,24 @@
"""Intent application service (REQ-007).
"""Intent application service (REQ-007; dcy-020/REQ-059·061 asset-centric).
Turns a buy/sell intent submitted over HTTP into a single-writer slot
Turns a buy/sell intent submitted over messaging into a single-writer slot
transition, with intent_id idempotency backed by a durable intent_log.
Idempotency model (dcy-002 OQ-1): `intent_store.try_insert` is the dedup
gate. On a duplicate intent_id it returns the prior decision without
re-driving any transition (NFR-3). Slot transitions go exclusively through
SlotService (advisory-lock serialized) so the single-writer invariant
holds (NFR-2). Concurrency on the last idle slot is resolved by retrying
on SlotStateMismatchError.
holds (NFR-2).
dcy-020 (BAT-030): admission is asset-centric only — the legacy pool path
(daily preallocation + idle∪closed candidate selection) was removed at the
cutover. Buy is keyed by slot_id (REQ-061 canonical; portfolio_id is the
legacy dual-accept mapping) and admits FCFS under the `max_active_slots` cap
M when configured (no cap when unset).
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import date, datetime
from datetime import datetime
from decimal import Decimal
from typing import Any, Optional
from uuid import UUID
@@ -26,7 +31,7 @@ from trex_core.domain.slot import (
SlotStateMismatchError,
)
_MAX_IDLE_RETRY = 8
_MAX_ADMIT_RETRY = 8
@dataclass(frozen=True)
@@ -45,15 +50,11 @@ 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.
# dcy-020 / REQ-061: slot_id is the canonical buy key (snapshot slots[]
# round-trip); portfolio_id is accepted as a legacy dual-accept mapping.
# symbol is not a client input — it is resolved server-side from the
# universe at admission and carried onto the slot.
intent_id: str
trading_date: date
# dcy-020 / REQ-061 (BAT-029): asset-centric contract key. slot_id is the
# canonical buy key (snapshot slots[] round-trip); portfolio_id is retained
# for the legacy pool path and as dual-accept in asset-centric mode.
portfolio_id: Optional[str] = None
slot_id: Optional[UUID] = None
symbol: Optional[str] = None
@@ -100,9 +101,8 @@ class IntentService:
# callable(portfolio_id) -> Optional[str]. When set, admission resolves symbol
# from the universe and rejects unknown_portfolio if absent.
self._resolve_symbol = resolve_symbol
# dcy-020 / REQ-059 (BAT-028): asset-centric mode gate. When max_active_slots
# is set, buy admission activates the asset's permanent slot under an M cap
# (FCFS, no eviction); otherwise the legacy pool path runs (unchanged).
# dcy-020 / REQ-059·060: M cap. None = no capacity cap (the operator is
# expected to set max_active_slots in the instance TOML — design §5).
self._instance_key = instance_key
self._max_active_slots = max_active_slots
@@ -116,14 +116,14 @@ class IntentService:
def submit_buy(self, intent: BuyIntent) -> IntentDecision:
payload = {
"trading_date": intent.trading_date.isoformat(),
"slot_id": str(intent.slot_id) if intent.slot_id else None,
"portfolio_id": intent.portfolio_id,
"symbol": intent.symbol,
"qty": intent.qty,
"allocated_capital": intent.allocated_capital,
"entry_price": intent.entry_price,
"strategy_id": intent.strategy_id,
"sell_strategy_id": intent.sell_strategy_id,
"portfolio_id": intent.portfolio_id,
}
prior = self._dedup(intent.intent_id, "buy", payload)
if prior is not None:
@@ -163,68 +163,14 @@ class IntentService:
return IntentDecision(intent_id, rec.status, rec.slot_id, rec.reason)
def _admit_buy(self, intent: BuyIntent) -> IntentDecision:
# dcy-020 / REQ-061 (BAT-029): asset-centric mode keys buy by slot_id
# (legacy portfolio_id dual-accept); the target slot + symbol are resolved
# inside the asset-centric path.
if self._max_active_slots is not None:
return self._admit_buy_asset_centric(intent)
# legacy pool path (dcy-011 / REQ-007): portfolio_id required; symbol is
# resolved server-side from the universe.
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-candidate race: another intent may claim the slot
# between our find and transit (SlotStateMismatchError).
# dcy-019 / REQ-007: admission candidate is idle ∪ closed (closed-slot
# reallocation), ordered by entry_count asc → updated_at asc → slot_id.
# The candidate may already be CLOSED, so the transit's from_state must
# follow the slot's current status rather than a hardcoded IDLE.
for _ in range(_MAX_IDLE_RETRY):
slot = self._repo.find_allocatable_slot(intent.trading_date)
if slot is None:
return IntentDecision(intent.intent_id, "rejected", None, "no_idle_slot")
try:
updated = self._svc.transit_slot(
slot.slot_id,
slot.status,
SlotState.PENDING_BUY,
reason="buy_intent",
actor="intent_api",
symbol=symbol,
qty=intent.qty,
allocated_capital=intent.allocated_capital,
# dcy-012 / REQ-007: buy entry price maps to the slot's
# side-agnostic pending limit_price (cleared on fill→HOLDING).
limit_price=intent.entry_price,
strategy_id=intent.strategy_id,
sell_strategy_id=intent.sell_strategy_id,
portfolio_id=intent.portfolio_id,
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": symbol},
)
except SlotStateMismatchError:
continue # slot was claimed concurrently; try another idle slot
return IntentDecision(intent.intent_id, "admitted", updated.slot_id, None)
return IntentDecision(intent.intent_id, "rejected", None, "idle_slot_contended")
# ------------------------------------------------------------
def _admit_buy_asset_centric(self, intent: BuyIntent) -> IntentDecision:
"""dcy-020 / REQ-059·REQ-061: activate the slot under the M cap.
"""dcy-020 / REQ-059·REQ-061: activate the asset's permanent slot.
Contract key is slot_id (REQ-061); portfolio_id is accepted as a legacy
dual-accept mapping. Symbol is resolved from the slot's portfolio_id.
FCFS + no eviction: already-active -> already_active; at M -> capacity_full
(existing holdings untouched, incl. over-cap). Single intent thread +
instance-scoped advisory lock serialize admissions (dcy-020 OQ-1)."""
FCFS + no eviction: already-active -> already_active; at M ->
capacity_full (existing holdings untouched, incl. over-cap). Single
intent thread + instance-scoped advisory lock serialize admissions
(dcy-020 OQ-1)."""
_ACTIVE = (SlotState.PENDING_BUY, SlotState.HOLDING, SlotState.PENDING_SELL)
def _resolve(sl):
@@ -233,7 +179,7 @@ class IntentService:
return self._resolve_symbol(pid)
return intent.symbol or sl.symbol
for _ in range(_MAX_IDLE_RETRY):
for _ in range(_MAX_ADMIT_RETRY):
# resolve the target slot: slot_id (canonical) or portfolio_id (legacy)
if intent.slot_id is not None:
slot = self._repo.fetch_slot(intent.slot_id)
@@ -250,7 +196,8 @@ class IntentService:
symbol = _resolve(slot)
if not symbol:
return IntentDecision(intent.intent_id, "rejected", None, "unknown_portfolio")
if self._repo.count_active(self._instance_key) >= self._max_active_slots:
if (self._max_active_slots is not None
and self._repo.count_active(self._instance_key) >= self._max_active_slots):
return IntentDecision(intent.intent_id, "rejected", None, "capacity_full")
try:
updated = self._svc.transit_slot(
@@ -262,6 +209,8 @@ class IntentService:
symbol=symbol,
qty=intent.qty,
allocated_capital=intent.allocated_capital,
# dcy-012 / REQ-007: buy entry price maps to the slot's
# side-agnostic pending limit_price (cleared on fill→HOLDING).
limit_price=intent.entry_price,
strategy_id=intent.strategy_id,
sell_strategy_id=intent.sell_strategy_id,
+5 -3
View File
@@ -13,7 +13,7 @@ from datetime import date as date_cls
from datetime import datetime, timezone
from trex_core.application.portfolio import GLOBAL_KILLSWITCH, KillSwitchStore
from trex_core.domain.slot import SlotRepository, SlotService, SlotState
from trex_core.domain.slot import SlotRepository, SlotService
_log = logging.getLogger(__name__)
_ACTOR = "kill-switch"
@@ -28,10 +28,12 @@ class KillSwitchService:
slot_repo: SlotRepository,
slot_service: SlotService,
killswitch_store: KillSwitchStore,
instance_key: str = "default",
) -> None:
self._repo = slot_repo
self._svc = slot_service
self._store = killswitch_store
self._instance_key = instance_key # dcy-020/REQ-056: instance-scoped sweep
@staticmethod
def _trading_date(now: datetime) -> date_cls:
@@ -50,8 +52,8 @@ class KillSwitchService:
scope: "portfolio" (requires portfolio_id) | "global".
"""
now = now or datetime.now(tz=timezone.utc)
day = self._trading_date(now)
holding = [s for s in self._repo.list_slots(day) if s.status is SlotState.HOLDING]
day = self._trading_date(now) # day-scoped exclusion key (REQ-044)
holding = self._repo.list_holding_slots(self._instance_key)
if scope == "portfolio":
holding = [s for s in holding if s.portfolio_id == portfolio_id]
+22 -22
View File
@@ -26,7 +26,6 @@ from trex_core.domain.slot import (
SlotNotFoundError,
SlotRepository,
SlotService,
SlotState,
)
_log = logging.getLogger(__name__)
@@ -72,12 +71,12 @@ 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 (legacy) or,
# in asset-centric mode (dcy-020/REQ-061), slot_id (canonical key).
# symbol is NOT a client field — resolved from the universe.
# dcy-020 / REQ-061: buy is keyed by slot_id (canonical);
# portfolio_id is the legacy dual-accept mapping. symbol is NOT
# a client field — resolved from the universe. trading_date was
# dropped from the contract at the cutover (BAT-030).
buy = BuyIntent(
intent_id=intent_id,
trading_date=_parse_date(body.get("trading_date")),
portfolio_id=_opt_str(body.get("portfolio_id")) or None,
slot_id=(UUID(str(body["slot_id"])) if body.get("slot_id") else None),
qty=_opt_float(body.get("qty")),
@@ -166,9 +165,10 @@ def make_snapshot_rpc_handler(
slot_repo: SlotRepository,
slot_service: SlotService,
*,
instance_key: str = "default",
snapshot_store: Optional[SnapshotStore] = None,
pause_store: Optional[PauseStore] = None,
default_total_slots: int = 10,
max_active_slots: Optional[int] = None,
watchdog_store: Optional[WatchdogStore] = None,
watchdog_timeout_sec: int = 120,
killswitch_store=None,
@@ -206,9 +206,8 @@ def make_snapshot_rpc_handler(
if str(body.get("actor") or "").lower() == "operator":
touch_watchdog = False
if watchdog_store is not None and touch_watchdog:
for s in slot_repo.list_slots(day):
if s.status is SlotState.HOLDING:
watchdog_store.update(str(s.slot_id))
for s in slot_repo.list_holding_slots(instance_key):
watchdog_store.update(str(s.slot_id))
# --- BAT-008 envelope path (snapshot store wired) ---
if snapshot_store is not None and pause_store is not None:
@@ -218,45 +217,49 @@ def make_snapshot_rpc_handler(
except ValueError:
view = View.ALL
strategy_key = _opt_str(body.get("strategy_key"))
cap = slot_service.compute_slot_capacity(day, default_total_slots)
# dcy-020 / REQ-059: available = remaining admission headroom under
# the M cap (max(0, M − active)); inactive count when no cap.
cap = slot_service.compute_slot_capacity(max_active_slots)
envelope = build_snapshot_envelope(
store=snapshot_store,
pause_store=pause_store,
slot_repo=slot_repo,
available_slots=cap.idle_count,
instance_key=instance_key,
available_slots=cap.available_count,
view=view,
strategy_key=strategy_key,
trading_date=day,
killswitch_store=killswitch_store,
)
payload = envelope.to_dict()
_attach_watchdog(payload, slot_repo, day, watchdog_store, watchdog_timeout_sec)
_attach_watchdog(payload, slot_repo, instance_key, watchdog_store, watchdog_timeout_sec)
_apply_cash_estimate(payload, ledger_outbox)
return RpcResponse(payload=payload)
# --- BAT-007 legacy stub-shape path ---
view_raw = str(body.get("view") or "all")
strategy_filter = _opt_str(body.get("strategy_id"))
slots = slot_repo.list_slots(day)
slots = slot_repo.list_slots(instance_key)
if view_raw == "open":
slots = [s for s in slots if s.status.value in ("pending_buy", "holding", "pending_sell")]
if strategy_filter is not None:
slots = [s for s in slots if s.strategy_id == strategy_filter]
cap = slot_service.compute_slot_capacity(day, max(1, len(slot_repo.list_slots(day))))
cap = slot_service.compute_slot_capacity(max_active_slots)
payload = {
"view": view_raw,
"strategy_id": strategy_filter,
"trading_date": day.isoformat(),
"slots": [_slot_to_dict(s) for s in slots],
"capacity": {
"idle_count": cap.idle_count,
"inactive_count": cap.inactive_count,
"holding_count": cap.holding_count,
"pending_count": cap.pending_count,
"active_count": cap.active_count,
"available_count": cap.available_count,
"total_slots": cap.total_slots,
"scarcity_ratio": cap.scarcity_ratio,
},
}
_attach_watchdog(payload, slot_repo, day, watchdog_store, watchdog_timeout_sec)
_attach_watchdog(payload, slot_repo, instance_key, watchdog_store, watchdog_timeout_sec)
return RpcResponse(payload=payload)
return handler
@@ -283,7 +286,7 @@ def _apply_cash_estimate(payload, ledger_outbox) -> None:
return
def _attach_watchdog(payload, slot_repo, day, watchdog_store, timeout_sec) -> None:
def _attach_watchdog(payload, slot_repo, instance_key, watchdog_store, timeout_sec) -> None:
"""Attach holding-slot watchdog state to a snapshot response (dcy-017/REQ-034).
`watchdog` = list of {slot_id, portfolio_id, symbol, expired, last_seen} for
@@ -293,9 +296,7 @@ def _attach_watchdog(payload, slot_repo, day, watchdog_store, timeout_sec) -> No
return
try:
states = []
for s in slot_repo.list_slots(day):
if s.status is not SlotState.HOLDING:
continue
for s in slot_repo.list_holding_slots(instance_key):
sid = str(s.slot_id)
last = watchdog_store.last_seen(sid)
states.append({
@@ -441,7 +442,6 @@ def slot_transition_event_payload(slot, *, event_type: str) -> dict:
"qty": float(slot.qty) if slot.qty is not None else None,
"avg_price": float(slot.avg_price) if slot.avg_price is not None else None,
"portfolio_id": slot.portfolio_id,
"trading_date": slot.trading_date.isoformat(),
"sequence_at": (slot.updated_at or datetime.now(tz=timezone.utc)).isoformat(),
}
+5 -4
View File
@@ -308,7 +308,6 @@ def slot_to_dict(s) -> dict:
"""
return {
"slot_id": str(s.slot_id),
"trading_date": s.trading_date.isoformat(),
"status": s.status.value,
"symbol": s.symbol,
"qty": float(s.qty) if s.qty is not None else None,
@@ -344,6 +343,7 @@ def build_snapshot_envelope(
store: SnapshotStore,
pause_store: PauseStore,
slot_repo,
instance_key: str = "default",
available_slots: int,
view: View = View.ALL,
strategy_key: Optional[str] = None,
@@ -352,14 +352,15 @@ def build_snapshot_envelope(
) -> SnapshotEnvelope:
"""Compose base(view) − mask(pending·warning·pause·killswitch) − strategy_key.
`slot_repo` is the SlotRepository used to read holding/pending state.
`trading_date` defaults to UTC today (matches the daemon's preallocate).
`slot_repo` is the SlotRepository used to read holding/pending state —
instance-scoped (dcy-020/REQ-056). `trading_date` (default UTC today) is
used only for the day-scoped kill-switch exclusion mask.
`killswitch_store` (REQ-046) adds the day-scoped kill-switch exclusion to
the mask: a `GLOBAL_KILLSWITCH` ("*") entry masks the whole universe.
"""
# Read core authoritative slot state once.
day = trading_date or datetime.now(tz=timezone.utc).date()
slots = slot_repo.list_slots(day)
slots = slot_repo.list_slots(instance_key)
holding_pids: set[str] = set()
pending_pids: set[str] = set()
holding_qty: dict[str, float] = {}
@@ -1,72 +0,0 @@
"""Daily slot preallocation scheduler (dcy-014 / REQ-005, issue #14).
trading_date is computed in a configurable timezone (default Asia/Seoul) so the
DB trading_date matches the local trading calendar, while all timestamps stay
UTC (NFR-1). Modeled on SnapshotScheduler (REQ-030): a tick loop fires once per
TZ-day at/after the configured trigger time, calling the idempotent
`preallocate_daily_slots` so the next day's slots are created without a daemon
restart (NFR-2/NFR-3).
"""
from __future__ import annotations
import logging
import threading
from datetime import date, datetime, time, timezone
from typing import Callable, Optional, Protocol
_log = logging.getLogger(__name__)
class _Preallocator(Protocol):
def preallocate_daily_slots(self, trading_date: date, count: int) -> int:
...
class DailyPreallocationScheduler:
"""Background-thread-friendly daily preallocation. Use `run_once()` for tests."""
def __init__(
self,
*,
slot_service: _Preallocator,
total_slots: int,
tz: timezone,
schedule_time: time = time(hour=0, minute=0),
clock: Optional[Callable[[], datetime]] = None,
) -> None:
self._svc = slot_service
self._total = total_slots
self._tz = tz # tzinfo used to derive trading_date (default Asia/Seoul)
self._schedule = schedule_time
self._clock = clock or (lambda: datetime.now(tz=timezone.utc))
self._last_fired_date: Optional[date] = None
# ---- callable from a daemon thread loop --------------------------------
def run_forever(self, stop: threading.Event, *, tick_sec: float = 60.0) -> None:
# Boot: ensure the current TZ-day's slots exist immediately (idempotent),
# then enter the daily cadence for subsequent day boundaries.
self.run_once()
while not stop.wait(tick_sec):
try:
self._maybe_tick()
except Exception: # noqa: BLE001 - loop must survive transient errors
_log.exception("preallocation scheduler tick failed")
def _maybe_tick(self) -> None:
now_tz = self._clock().astimezone(self._tz)
target = datetime.combine(now_tz.date(), self._schedule, tzinfo=self._tz)
if now_tz < target:
return # before today's trigger time
if self._last_fired_date is not None and self._last_fired_date == now_tz.date():
return # already created today's slots
self.run_once()
# ---- the actual preallocation -----------------------------------------
def run_once(self) -> date:
"""Preallocate slots for the current TZ trading_date (idempotent). Returns it."""
trading_date = self._clock().astimezone(self._tz).date()
created = self._svc.preallocate_daily_slots(trading_date, self._total)
if created:
_log.info("preallocated %d slots for trading_date=%s", created, trading_date)
self._last_fired_date = trading_date
return trading_date
+4 -7
View File
@@ -25,7 +25,7 @@ from datetime import datetime, time as time_cls, timedelta, timezone
from trex_core.application.broker_port import BrokerPort
from trex_core.application.watchdog import WatchdogStore
from trex_core.domain.slot import SlotRepository, SlotService, SlotState
from trex_core.domain.slot import SlotRepository, SlotService
from trex_core.domain.slot.models import Slot
from trex_core.runtime.config import SafetySellConfig
@@ -133,13 +133,10 @@ class SafetySellService:
))
# ---------------------------------------------------------------- helpers
@staticmethod
def _trading_date(now: datetime) -> date_cls:
return now.astimezone(timezone.utc).date()
def _holding(self, now: datetime) -> list[Slot]:
return [s for s in self._repo.list_slots(self._trading_date(now))
if s.status is SlotState.HOLDING]
# dcy-020 (BAT-030): permanent asset slots — instance-scoped holding
# sweep (the day-scoped pool listing was removed with trading_date).
return self._repo.list_holding_slots(self._instance_key)
def _watchdog_expired(self, slot: Slot) -> bool:
return self._watchdog.is_expired(str(slot.slot_id), self._cfg.watchdog_timeout_sec)
+5 -12
View File
@@ -40,8 +40,6 @@ class SnapshotScheduler:
slot_repo=None,
slot_service=None,
instance_key: str = "default",
reconcile_enabled: bool = False,
trading_date_fn: Optional[Callable[[], object]] = None,
) -> None:
self._read = read_port
self._store = store
@@ -50,13 +48,12 @@ class SnapshotScheduler:
self._clock = clock or (lambda: datetime.now(tz=timezone.utc))
self._daemon_id: Optional[str] = None
self._last_fired_date: Optional[datetime] = None
# dcy-020 / REQ-058 (BAT-028): asset-centric slot reconcile after each
# universe fetch (only when enabled = asset-centric mode).
# dcy-020 / REQ-058 (BAT-030): asset-centric slot reconcile after each
# universe fetch — always on when slot_repo is wired (the reconcile is
# what creates/retires the permanent asset slots).
self._slot_repo = slot_repo
self._slot_service = slot_service
self._instance_key = instance_key
self._reconcile_enabled = reconcile_enabled
self._trading_date_fn = trading_date_fn
self._last_universe_size: Optional[int] = None
# ---- callable from a daemon thread loop --------------------------------
@@ -112,7 +109,7 @@ class SnapshotScheduler:
reconcile_positions(store=self._store, broker=self._broker)
except Exception: # noqa: BLE001
_log.exception("reconciliation pass failed")
if self._reconcile_enabled and self._slot_repo is not None:
if self._slot_repo is not None:
try:
self._reconcile_slots(items)
except Exception: # noqa: BLE001
@@ -144,12 +141,8 @@ class SnapshotScheduler:
"slot reconcile: universe shrank %d->%d (>=50%%) — retire held",
prior, len(universe),
)
if self._trading_date_fn is not None:
trading_date = self._trading_date_fn()
else:
trading_date = self._clock().date()
result = self._slot_repo.reconcile_universe(
self._instance_key, trading_date, universe, allow_retire=allow_retire,
self._instance_key, universe, allow_retire=allow_retire,
)
self._last_universe_size = len(universe)
if any(result.values()):
+2 -4
View File
@@ -4,12 +4,12 @@ Storage-agnostic pure domain: no broker, DB driver, CLI, or config dependency.
A repository implementation is injected via the `SlotRepository` protocol.
Public API:
- SlotState: enum of 6 slot states + allowed transitions
- SlotState: enum of 5 slot states + allowed transitions
- Slot: data class for a slot row
- SlotCapacity: capacity snapshot returned by compute_slot_capacity
- SlotTransitionRecord: append-only transition record
- EXPECTED_SLOT_FIELDS: canonical `Slot` column contract (guarded by test_contract)
- SlotService: main domain service (preallocate, transit, capacity, entry count)
- SlotService: main domain service (transit, sell, thresholds, capacity)
- SlotRepository: abstract storage protocol
- SlotTransitionError / SlotStateMismatchError / SlotNotFoundError: domain errors
"""
@@ -34,7 +34,6 @@ from trex_core.domain.slot.state import (
SLOT_ADVISORY_LOCK_ID,
SlotState,
is_allowed_transition,
normalize_state,
slot_advisory_lock_id,
)
@@ -53,6 +52,5 @@ __all__ = [
"SlotTransitionError",
"SlotTransitionRecord",
"is_allowed_transition",
"normalize_state",
"slot_advisory_lock_id",
]
+28 -69
View File
@@ -1,4 +1,4 @@
"""In-memory SlotRepository — for unit tests and current trex-core runtime.
"""In-memory SlotRepository — for unit tests.
Simulates advisory lock via threading.Lock so contention tests are still
meaningful even without a real PostgreSQL instance.
@@ -8,7 +8,7 @@ from __future__ import annotations
import threading
from contextlib import contextmanager
from dataclasses import replace
from datetime import date, datetime, timezone
from datetime import datetime, timezone
from typing import Iterator, Optional
from uuid import UUID, uuid4
@@ -23,7 +23,6 @@ class InMemorySlotRepository:
def __init__(self) -> None:
self._slots: dict[UUID, Slot] = {}
self._transitions: list[SlotTransitionRecord] = []
self._entries: dict[tuple[date, str], tuple[int, datetime]] = {}
# Per-lock-id threading.Lock; mirrors pg_advisory_xact_lock semantics
# at process level. lock_time and contention can be observed.
self._lock_table: dict[int, threading.Lock] = {}
@@ -48,53 +47,16 @@ class InMemorySlotRepository:
return lock
# --- daily_slot CRUD ---
def pre_allocate_slots(self, trading_date: date, count: int) -> int:
existing = [s for s in self._slots.values() if s.trading_date == trading_date]
if existing:
return 0
now = datetime.now(tz=timezone.utc)
for _ in range(count):
slot_id = uuid4()
self._slots[slot_id] = Slot(
slot_id=slot_id,
trading_date=trading_date,
status=SlotState.IDLE,
created_at=now,
updated_at=now,
)
return count
def add_slot(self, slot: Slot) -> None:
"""Test seeding helper — insert a slot row directly."""
self._slots[slot.slot_id] = slot
def list_slots(self, trading_date: date) -> list[Slot]:
return [s for s in self._slots.values() if s.trading_date == trading_date]
def list_slots(self, instance_key: str) -> list[Slot]:
return [s for s in self._slots.values() if s.instance_key == instance_key]
def fetch_slot(self, slot_id: UUID) -> Optional[Slot]:
return self._slots.get(slot_id)
def find_idle_slot(self, trading_date: date) -> Optional[Slot]:
for s in self._slots.values():
if s.trading_date == trading_date and s.status is SlotState.IDLE:
return s
return None
def find_allocatable_slot(self, trading_date: date) -> Optional[Slot]:
# dcy-019 / REQ-006: idle ∪ closed candidates, ordered by entry_count asc
# → updated_at asc → slot_id (deterministic, OQ-2).
candidates = [
s for s in self._slots.values()
if s.trading_date == trading_date
and s.status in (SlotState.IDLE, SlotState.CLOSED)
]
if not candidates:
return None
return min(
candidates,
key=lambda s: (
s.entry_count,
s.updated_at or datetime.min.replace(tzinfo=timezone.utc),
str(s.slot_id),
),
)
def update_slot(self, slot: Slot) -> None:
if slot.slot_id not in self._slots:
raise SlotNotFoundError(f"slot not found: {slot.slot_id}")
@@ -122,7 +84,7 @@ class InMemorySlotRepository:
]
def reconcile_universe(
self, instance_key: str, trading_date: date, universe_portfolio_ids: set[str],
self, instance_key: str, universe_portfolio_ids: set[str],
*, allow_retire: bool = True,
) -> dict[str, int]:
existing = {
@@ -136,7 +98,7 @@ class InMemorySlotRepository:
if pid not in existing:
sid = uuid4()
self._slots[sid] = Slot(
slot_id=sid, trading_date=trading_date, status=SlotState.IDLE,
slot_id=sid, status=SlotState.INACTIVE,
instance_key=instance_key, portfolio_id=pid,
created_at=now, updated_at=now,
)
@@ -165,27 +127,23 @@ class InMemorySlotRepository:
def all_transition_log(self) -> list[SlotTransitionRecord]:
return list(self._transitions)
# --- daily_symbol_entries ---
def increment_symbol_entry(self, trading_date: date, symbol: str) -> int:
key = (trading_date, symbol)
count, _ = self._entries.get(key, (0, datetime.min))
new_count = count + 1
self._entries[key] = (new_count, datetime.now(tz=timezone.utc))
return new_count
def get_symbol_entry_count(self, trading_date: date, symbol: str) -> int:
count, _ = self._entries.get((trading_date, symbol), (0, datetime.min))
return count
# --- order execution queue (BAT-004 / REQ-013, option A) ---
_PENDING = (SlotState.PENDING_BUY, SlotState.PENDING_SELL)
def find_pending_for_dispatch(self) -> Optional[Slot]:
"""A pending slot not yet dispatched (broker_order_id IS NULL)."""
for s in self._slots.values():
if s.status in self._PENDING and s.broker_order_id is None:
return s
return None
def list_undispatched_pending(
self, instance_key: str = "default", limit: int = 64,
) -> list[Slot]:
"""Pending slots not yet dispatched (broker_order_id IS NULL).
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 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)
]
return out[:limit]
def mark_dispatched(self, slot_id: UUID, broker_order_id: str) -> bool:
"""Atomically set broker_order_id iff still undispatched (idempotency guard)."""
@@ -197,11 +155,12 @@ class InMemorySlotRepository:
)
return True
def list_dispatched_pending(self) -> list[Slot]:
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.status in self._PENDING and s.broker_order_id is not None
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:
@@ -213,10 +172,10 @@ class InMemorySlotRepository:
)
# --- bulk counts ---
def count_slots_by_status(self, trading_date: date) -> dict[SlotState, int]:
def count_slots_by_status(self, instance_key: str) -> dict[SlotState, int]:
result: dict[SlotState, int] = {}
for s in self._slots.values():
if s.trading_date != trading_date:
if s.instance_key != instance_key:
continue
result[s.status] = result.get(s.status, 0) + 1
return result
+24 -23
View File
@@ -1,8 +1,8 @@
"""Slot domain data classes (REQ-001, REQ-002)."""
"""Slot domain data classes (REQ-001, REQ-002; dcy-020/REQ-056 asset-centric)."""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import date, datetime
from datetime import datetime
from decimal import Decimal
from typing import Optional
from uuid import UUID
@@ -14,43 +14,37 @@ from trex_core.domain.slot.state import SlotState
class Slot:
"""Single daily_slot row.
Matches the slot column contract recorded in dcy-001 부록 A and
docs/data-model.md (trex-core stores this in-memory for now; a future
persistence layer must satisfy the same contract).
dcy-020/REQ-056 (BAT-030): the slot is a **permanent asset slot** — one row
per (instance_key, portfolio_id) universe asset, living across trading days.
Pool-model columns `trading_date` and `entry_count` were dropped at the
cutover (migration 011); time attributes live on events/ledger rows, not on
the slot. Matches docs/data-model.md.
"""
slot_id: UUID
trading_date: date
status: SlotState
symbol: Optional[str] = None
qty: Optional[float] = None
avg_price: Optional[float] = None
# dcy-012 / REQ-002: side-agnostic "current pending order limit price"
# (renamed from entry_price). Set on pending entry (buy/sell limit), cleared
# on pending exit (→HOLDING, →CLOSED) so a buy limit never leaks into a sell
# order (issue #10). None => market order at dispatch.
# on pending exit (→HOLDING, →INACTIVE) so a buy limit never leaks into a
# sell order (issue #10). None => market order at dispatch.
limit_price: Optional[float] = None
allocated_capital: Optional[float] = None
order_id: Optional[str] = None
strategy_id: Optional[str] = None
broker_order_id: Optional[str] = None # BAT-003 / REQ-011 (trex-orbit lineage)
portfolio_id: Optional[str] = None # BAT-006 / REQ-025 (trex-orbit lineage)
portfolio_id: Optional[str] = None # BAT-006 / REQ-025; asset binding (REQ-056)
sell_strategy_id: Optional[str] = None # BAT-008 / REQ-047 (trex-orbit lineage)
forced_close_at: Optional[datetime] = None # BAT-014 / REQ-066 (trex-orbit lineage)
stop_loss_price: Optional[Decimal] = None # REQ-035 / BAT-010
take_profit_price: Optional[Decimal] = None # REQ-035 / BAT-010
# dcy-019 / REQ-002 (BAT-025): slot-scoped cumulative entry count. +1 on each
# allocation transition (idle/closed → pending_buy); drives admission slot
# selection (lowest entry_count first). Distinct from the symbol-scoped
# daily_symbol_entries counter (REQ-002 AC-3) — different key/purpose.
entry_count: int = 0
created_at: Optional[datetime] = None
updated_at: Optional[datetime] = None
# dcy-020 / REQ-056 (BAT-027): asset-centric permanent slot foundation (안 B).
# instance_key scopes slots per daemon instance in a shared DB
# (UNIQUE(instance_key, portfolio_id)); advisory lock is also instance-scoped.
# retired_at marks a soft-retired / universe-detached slot (OQ-2). Both added
# additively here; entry_count/trading_date removal is deferred to cutover (BAT-δ).
# dcy-020 / REQ-056: instance_key scopes slots per daemon instance in a
# shared DB (UNIQUE(instance_key, portfolio_id)); advisory lock is also
# instance-scoped. retired_at marks a soft-retired / universe-detached slot.
instance_key: str = "default"
retired_at: Optional[datetime] = None
@@ -61,7 +55,6 @@ class Slot:
# / docs/data-model.md), not here — names/presence only.
EXPECTED_SLOT_FIELDS: frozenset[str] = frozenset({
"slot_id",
"trading_date",
"status",
"symbol",
"qty",
@@ -76,7 +69,6 @@ EXPECTED_SLOT_FIELDS: frozenset[str] = frozenset({
"forced_close_at",
"stop_loss_price", # REQ-035
"take_profit_price", # REQ-035
"entry_count", # dcy-019 / REQ-002 (BAT-025)
"created_at",
"updated_at",
"instance_key", # dcy-020 / REQ-056 (BAT-027)
@@ -86,11 +78,20 @@ EXPECTED_SLOT_FIELDS: frozenset[str] = frozenset({
@dataclass(frozen=True)
class SlotCapacity:
"""Snapshot of slot capacity returned by compute_slot_capacity (REQ-002 AC-6)."""
"""Capacity snapshot returned by compute_slot_capacity (REQ-002 AC-6;
dcy-020/REQ-059 asset-centric redefinition, BAT-030).
idle_count: int
active_count = holding + pending (position-occupying). available_count is
the remaining admission headroom: max(0, M − active) under a max_active_slots
cap M, else the inactive (admittable) slot count. total_slots = M when set,
else the instance's slot count. scarcity_ratio = 1 − available/total.
"""
inactive_count: int
holding_count: int
pending_count: int
active_count: int
available_count: int
total_slots: int
scarcity_ratio: float
+15 -42
View File
@@ -1,13 +1,12 @@
"""Abstract slot repository protocol (REQ-002).
"""Abstract slot repository protocol (REQ-002; dcy-020 asset-centric).
Concrete implementations:
- InMemorySlotRepository — unit tests / current trex-core runtime
- (future) PostgreSQL-backed repository — added in the persistence iteration
- InMemorySlotRepository — unit tests
- PostgreSQLSlotRepository — production persistence
"""
from __future__ import annotations
from contextlib import AbstractContextManager
from datetime import date
from typing import Optional, Protocol
from uuid import UUID
@@ -22,6 +21,11 @@ class SlotRepository(Protocol):
- `transactional_advisory_lock()` runs the with-block inside a single transaction
protected by an advisory lock. External I/O MUST NOT be performed inside the block.
- `append_transition_log()` is append-only (no UPDATE/DELETE).
dcy-020 (BAT-030): slots are permanent asset slots keyed by
(instance_key, portfolio_id); all listing/counting is instance-scoped.
The pool-model surface (pre_allocate_slots / find_idle_slot /
find_allocatable_slot / daily_symbol_entries) was removed at the cutover.
"""
# --- lock + transaction ---
@@ -33,32 +37,14 @@ class SlotRepository(Protocol):
...
# --- daily_slot CRUD ---
def pre_allocate_slots(self, trading_date: date, count: int) -> int:
"""Insert `count` idle slots for trading_date if none exist; return inserted count.
Idempotent — if slots for trading_date already exist, returns 0 (REQ-002 AC-4).
"""
...
def list_slots(self, trading_date: date) -> list[Slot]:
"""Return all slots for the trading day."""
def list_slots(self, instance_key: str) -> list[Slot]:
"""Return all slots owned by this instance."""
...
def fetch_slot(self, slot_id: UUID) -> Optional[Slot]:
"""Return single slot row or None."""
...
def find_idle_slot(self, trading_date: date) -> Optional[Slot]:
"""Return any one idle slot (None if exhausted)."""
...
def find_allocatable_slot(self, trading_date: date) -> Optional[Slot]:
"""Return the best buy-admission candidate among idle ∪ closed slots
(dcy-019 / REQ-006), ordered by `entry_count` asc → `updated_at` asc →
`slot_id`. None if neither idle nor closed slots remain. Concurrency on
the last candidate is handled by the caller (advisory lock + retry)."""
...
def update_slot(self, slot: Slot) -> None:
"""Persist slot row (UPDATE by slot_id)."""
...
@@ -66,8 +52,7 @@ class SlotRepository(Protocol):
# --- asset-centric (dcy-020 / BAT-028) ---
def find_slot_by_portfolio(self, instance_key: str, portfolio_id: str) -> Optional[Slot]:
"""Return the (single, permanent) slot bound to `portfolio_id` for this
instance (dcy-020/REQ-059), or None. Keyed by (instance_key, portfolio_id)
— independent of trading_date."""
instance (dcy-020/REQ-059), or None."""
...
def count_active(self, instance_key: str) -> int:
@@ -77,11 +62,11 @@ class SlotRepository(Protocol):
def list_holding_slots(self, instance_key: str) -> list[Slot]:
"""Return holding slots for this instance (dcy-020/REQ-062 divergence
reconcile)."""
reconcile; safety-sell / kill-switch sweeps)."""
...
def reconcile_universe(
self, instance_key: str, trading_date: date, universe_portfolio_ids: set[str],
self, instance_key: str, universe_portfolio_ids: set[str],
*, allow_retire: bool = True,
) -> dict[str, int]:
"""Reconcile the slot set to the portfolio-hub universe (dcy-020/REQ-058).
@@ -101,19 +86,7 @@ class SlotRepository(Protocol):
"""Append one row to slot_transition_log (no UPDATE/DELETE allowed)."""
...
# --- daily_symbol_entries ---
def increment_symbol_entry(self, trading_date: date, symbol: str) -> int:
"""Upsert (trading_date, symbol) with entry_count += 1, last_entry_at = now.
Returns the resulting entry_count (REQ-002 AC-3).
"""
...
def get_symbol_entry_count(self, trading_date: date, symbol: str) -> int:
"""Return today's entry_count for symbol, or 0 if no row."""
...
# --- bulk counts for capacity ---
def count_slots_by_status(self, trading_date: date) -> dict[SlotState, int]:
"""Return dict mapping SlotState -> count for trading_date."""
def count_slots_by_status(self, instance_key: str) -> dict[SlotState, int]:
"""Return dict mapping SlotState -> count for this instance."""
...
+44 -85
View File
@@ -1,17 +1,19 @@
"""Slot domain service — 4 main APIs (REQ-002).
"""Slot domain service (REQ-002; dcy-020 asset-centric).
- preallocate_daily_slots(trading_date, count)
- transit_slot(slot_id, from_state, to_state, reason, actor, **fields)
- get_today_entry_count(trading_date, symbol)
- compute_slot_capacity(trading_date, total_slots)
- sell(slot_id, actor, reason, limit_price=None)
- update_thresholds(slot_id, ...)
- compute_slot_capacity(max_active_slots=None)
All state-mutating APIs run inside repository.transactional_advisory_lock()
to serialize Buy/Sell decision threads + operator CLI (REQ-002 AC-2).
dcy-020 (BAT-030): the pool-model surface (daily preallocation, symbol entry
counters) was removed at the cutover; slots are permanent asset slots.
"""
from __future__ import annotations
from dataclasses import replace
from datetime import date, datetime, timezone
from datetime import datetime, timezone
from decimal import Decimal
from typing import Any, Optional
from uuid import UUID
@@ -44,21 +46,9 @@ class SlotService:
self._instance_key = instance_key
self._lock_id = slot_advisory_lock_id(instance_key)
# ------------------------------------------------------------
# preallocate_daily_slots
# ------------------------------------------------------------
def preallocate_daily_slots(self, trading_date: date, count: int) -> int:
"""Insert `count` idle slots for trading_date if none exist.
Idempotent: second+ call on the same trading_date returns 0.
Returns number of slots inserted.
REQ-002 AC-4, REQ-003 AC-5 (count must be > 0).
"""
if count <= 0:
raise ValueError(f"slot count must be > 0, got {count}")
with self._repo.transactional_advisory_lock(self._lock_id):
return self._repo.pre_allocate_slots(trading_date, count)
@property
def instance_key(self) -> str:
return self._instance_key
# ------------------------------------------------------------
# transit_slot
@@ -94,7 +84,6 @@ class SlotService:
3. verify (from_state, to_state) is in the allowed set
4. update slot row (fields per to_state semantics)
5. append transition_log row
6. on closed->idle, increment daily_symbol_entries (REQ-002 AC-3)
External I/O (broker, kis-data-hub, portfolio-hub) MUST NOT be performed
inside this function — caller does I/O outside the lock (REQ-002 AC-5).
@@ -135,14 +124,6 @@ class SlotService:
)
self._repo.update_slot(updated)
# closed -> idle recovery: bump daily_symbol_entries atomically
# (REQ-002 AC-3, REQ-001 split-by-purpose).
if from_state is SlotState.CLOSED and to_state is SlotState.IDLE:
# Use the symbol carried by the slot before reset.
symbol_for_entry = slot.symbol
if symbol_for_entry:
self._repo.increment_symbol_entry(slot.trading_date, symbol_for_entry)
self._repo.append_transition_log(
SlotTransitionRecord(
slot_id=slot_id,
@@ -177,23 +158,19 @@ class SlotService:
) -> Slot:
"""Compute the post-transition Slot row.
Special: closed -> idle resets occupancy fields (FR-2 #7).
Special: pending_buy -> idle resets occupancy (REQ-011 AC-6 rollback).
dcy-020/REQ-057 (BAT-030): entering INACTIVE deactivates the permanent
asset slot — all position/occupancy fields are cleared while the asset
identity (instance_key, portfolio_id, symbol, retired_at) is preserved.
Covers buy rollback (pending_buy→inactive), sell fill
(pending_sell→inactive) and operator error resolve (error→inactive).
dcy-012 / REQ-002: limit_price is set only while a slot is in a pending
state (the live order's limit); entering any non-pending state
(IDLE/HOLDING/CLOSED) clears it so a buy limit never leaks into the next
sell order (issue #10 root cause).
state (the live order's limit); entering any non-pending state clears it
so a buy limit never leaks into the next sell order (issue #10).
"""
if to_state is SlotState.IDLE and slot.status in {
SlotState.CLOSED,
SlotState.PENDING_BUY,
}:
# Recovery / rollback: clear occupancy + broker handle.
# transition_log retains audit trail.
if to_state is SlotState.INACTIVE:
return replace(
slot,
status=SlotState.IDLE,
symbol=None,
status=SlotState.INACTIVE,
qty=None,
avg_price=None,
limit_price=None,
@@ -201,7 +178,6 @@ class SlotService:
order_id=None,
strategy_id=None,
broker_order_id=None,
portfolio_id=None,
sell_strategy_id=None,
forced_close_at=None,
stop_loss_price=None,
@@ -209,30 +185,13 @@ class SlotService:
updated_at=datetime.now(tz=timezone.utc),
)
# dcy-019 / REQ-002 (BAT-025): closed-slot direct reallocation. A closed
# slot still carries the prior cycle's occupancy; start from a clean
# slate so a previous position/broker handle never leaks into the new
# entry. entry_count is intentionally preserved (slot-scoped cumulative).
if to_state is SlotState.PENDING_BUY and slot.status is SlotState.CLOSED:
slot = replace(
slot,
symbol=None, qty=None, avg_price=None, limit_price=None,
allocated_capital=None, order_id=None, strategy_id=None,
broker_order_id=None, portfolio_id=None, sell_strategy_id=None,
forced_close_at=None, stop_loss_price=None, take_profit_price=None,
)
# limit_price lifecycle (dcy-012 FR-1): set only on entering a pending
# state (the order's limit); cleared on entering HOLDING/CLOSED.
# state (the order's limit); cleared on entering HOLDING/ERROR.
if to_state in {SlotState.PENDING_BUY, SlotState.PENDING_SELL}:
new_limit_price = limit_price if limit_price is not None else slot.limit_price
else: # HOLDING / CLOSED — pending order limit no longer applies.
else: # HOLDING / ERROR — pending order limit no longer applies.
new_limit_price = None
# dcy-019 / REQ-002: +1 on each allocation transition (→pending_buy),
# for both idle→pending_buy and closed→pending_buy (FR-4).
new_entry_count = slot.entry_count + (1 if to_state is SlotState.PENDING_BUY else 0)
# Generic update: caller-supplied fields override; None means "keep".
return replace(
slot,
@@ -266,7 +225,6 @@ class SlotService:
take_profit_price=(
take_profit_price if take_profit_price is not None else slot.take_profit_price
),
entry_count=new_entry_count, # dcy-019 / REQ-002
updated_at=datetime.now(tz=timezone.utc),
)
@@ -353,37 +311,38 @@ class SlotService:
self._repo.update_slot(updated)
return "updated"
# ------------------------------------------------------------
# get_today_entry_count
# ------------------------------------------------------------
def get_today_entry_count(self, trading_date: date, symbol: str) -> int:
"""Return today's entry_count for a symbol (0 if no entries yet).
Used by buy decision state gate to enforce re-entry policy.
Reads daily_symbol_entries (decision input), NEVER queries
slot_transition_log (audit only). REQ-001, dcy-001 5.3.5.5.
"""
return self._repo.get_symbol_entry_count(trading_date, symbol)
# ------------------------------------------------------------
# compute_slot_capacity
# ------------------------------------------------------------
def compute_slot_capacity(self, trading_date: date, total_slots: int) -> SlotCapacity:
"""Return idle/holding/pending counts + scarcity_ratio (REQ-002 AC-6).
def compute_slot_capacity(self, max_active_slots: Optional[int] = None) -> SlotCapacity:
"""Return instance-scoped capacity counts (REQ-002 AC-6; dcy-020/REQ-059).
scarcity_ratio = 1 - idle_count / total_slots, clamped to [0.0, 1.0].
active = holding + pending (position-occupying). Under a
`max_active_slots` cap M, available = max(0, M − active) and
total_slots = M; without a cap, available = inactive slot count and
total_slots = the instance's slot count.
scarcity_ratio = 1 − available/total, clamped to [0.0, 1.0].
"""
if total_slots <= 0:
raise ValueError(f"total_slots must be > 0, got {total_slots}")
counts = self._repo.count_slots_by_status(trading_date)
idle = counts.get(SlotState.IDLE, 0)
counts = self._repo.count_slots_by_status(self._instance_key)
inactive = counts.get(SlotState.INACTIVE, 0)
holding = counts.get(SlotState.HOLDING, 0)
pending = counts.get(SlotState.PENDING_BUY, 0) + counts.get(SlotState.PENDING_SELL, 0)
scarcity = max(0.0, min(1.0, 1.0 - (idle / total_slots)))
active = holding + pending
if max_active_slots is not None and max_active_slots > 0:
total = max_active_slots
available = max(0, max_active_slots - active)
else:
total = sum(counts.values())
available = inactive
scarcity = (
max(0.0, min(1.0, 1.0 - (available / total))) if total > 0 else 0.0
)
return SlotCapacity(
idle_count=idle,
inactive_count=inactive,
holding_count=holding,
pending_count=pending,
total_slots=total_slots,
active_count=active,
available_count=available,
total_slots=total,
scarcity_ratio=scarcity,
)
+13 -35
View File
@@ -1,12 +1,10 @@
"""Slot state machine (REQ-001, REQ-002; dcy-020/REQ-057 asset-centric redefinition).
"""Slot state machine (REQ-001, REQ-002; dcy-020/REQ-057 asset-centric).
dcy-020 (안 B): the asset-centric permanent slot collapses the pool-model `idle`
(never-bought) and `closed` (sold, now flat) into a single `inactive` (no open
position) state, and removes the `closed -> idle` recovery transition. `idle` and
`closed` are retained as **backward-compatible aliases** that normalize to
`inactive` for transition checks, so existing rows/runtime stay valid until the
cutover (BAT-δ) drops the aliases. The canonical transition set is expressed in
terms of `inactive`.
The asset-centric permanent slot has five states. `inactive` means "no open
position" — it replaced the pool-model `idle` (never-bought) and `closed`
(sold, now flat) states, whose aliases were removed at the cutover (BAT-030);
migration 010 rewrites any remaining `idle`/`closed` rows to `inactive` and
swaps the status CHECK constraint accordingly.
"""
from __future__ import annotations
@@ -30,34 +28,18 @@ def slot_advisory_lock_id(instance_key: str) -> int:
class SlotState(str, Enum):
"""Slot lifecycle states. `inactive` is the asset-centric standard
(dcy-020/REQ-057); `idle`/`closed` are retained as aliases normalizing to
`inactive` (removed at cutover, BAT-δ)."""
"""Slot lifecycle states (dcy-020/REQ-057 asset-centric, 5 states)."""
INACTIVE = "inactive" # dcy-020: no open position (replaces idle/closed)
IDLE = "idle" # alias -> inactive (pool-model lineage; drop at BAT-δ)
INACTIVE = "inactive" # no open position (permanent slot at rest)
PENDING_BUY = "pending_buy"
HOLDING = "holding"
PENDING_SELL = "pending_sell"
CLOSED = "closed" # alias -> inactive (pool-model lineage; drop at BAT-δ)
ERROR = "error"
# idle/closed are pool-model aliases of the asset-centric `inactive` state.
_INACTIVE_ALIASES: frozenset[SlotState] = frozenset(
{SlotState.IDLE, SlotState.CLOSED, SlotState.INACTIVE}
)
def normalize_state(state: SlotState) -> SlotState:
"""Map pool-model aliases (idle/closed) to the canonical `inactive`."""
return SlotState.INACTIVE if state in _INACTIVE_ALIASES else state
# Canonical base transitions in terms of `inactive` (dcy-020/REQ-057).
# `closed -> idle` recovery is REMOVED (it normalizes to inactive->inactive, a
# disallowed self-loop). `inactive -> pending_buy` covers both the old
# idle->pending_buy and closed->pending_buy (reallocation) families.
# Canonical transitions (dcy-020/REQ-057). There is no inactive self-loop and
# no closed->idle recovery family — a permanent slot simply returns to
# `inactive` and may activate again via inactive->pending_buy.
_BASE_TRANSITIONS: frozenset[tuple[SlotState, SlotState]] = frozenset(
{
(SlotState.INACTIVE, SlotState.PENDING_BUY), # 1. buy decision (activate)
@@ -85,15 +67,11 @@ _FROM_ERROR_TRANSITIONS: frozenset[tuple[SlotState, SlotState]] = frozenset(
)
# Canonical allowed transitions (inactive-based). Use `is_allowed_transition`
# which normalizes aliases before lookup.
ALLOWED_TRANSITIONS: frozenset[tuple[SlotState, SlotState]] = (
_BASE_TRANSITIONS | _TO_ERROR_TRANSITIONS | _FROM_ERROR_TRANSITIONS
)
def is_allowed_transition(from_state: SlotState, to_state: SlotState) -> bool:
"""Return True iff (from_state, to_state) — after normalizing idle/closed to
inactive — is an allowed transition family. `closed->idle` and any
inactive self-loop are rejected (dcy-020/REQ-057)."""
return (normalize_state(from_state), normalize_state(to_state)) in ALLOWED_TRANSITIONS
"""Return True iff (from_state, to_state) is an allowed transition."""
return (from_state, to_state) in ALLOWED_TRANSITIONS
+4 -2
View File
@@ -135,7 +135,9 @@ def _cmd_slots(args, cfg, rpc) -> int:
if status in ("holding", "held"):
slots = [s for s in slots if s.get("status") == "holding"]
elif status == "buyable":
slots = [s for s in slots if s.get("status") == "idle"]
# dcy-020 (BAT-030): admittable = inactive, not universe-detached.
slots = [s for s in slots
if s.get("status") == "inactive" and not s.get("universe_detached")]
elif status == "pending":
slots = [s for s in slots if s.get("status") in ("pending_buy", "pending_sell")]
# dcy-017: holding-slot watchdog state (strategy-absent ownership).
@@ -203,7 +205,7 @@ def _cmd_enter(args, cfg, rpc, confirm) -> int:
return short
payload = {
"intent_type": "buy", "intent_id": str(uuid.uuid4()),
"trading_date": _today_iso(cfg), "portfolio_id": args.portfolio_id,
"portfolio_id": args.portfolio_id,
"qty": args.qty, "allocated_capital": args.capital,
"entry_price": args.entry_price, "strategy_id": args.strategy_id,
"sell_strategy_id": args.sell_strategy_id,
+8 -17
View File
@@ -139,13 +139,11 @@ class DaemonInstanceConfig:
instance_key: str
portfolio_hub_token: str = ""
# dcy-014 / REQ-048: trading_date timezone + daily preallocation trigger.
# Optional — absent keys fall back to KST / midnight defaults.
# dcy-014 / REQ-048: operator-local timezone (CLI date display etc.).
trading_timezone: str = "Asia/Seoul"
daily_preallocation_time: str = "00:00"
# dcy-020 / REQ-060 (BAT-028): max concurrent active (holding) slots, M.
# Presence is the asset-centric mode gate: set -> universe reconcile + FCFS/M
# admission; absent -> legacy pool model (preallocation + find_allocatable).
# dcy-020 / REQ-060: max concurrent active (holding) slots, M — the FCFS
# admission capacity cap. BAT-030: asset-centric is the only model; absent
# means no cap (operators are expected to set it in production).
max_active_slots: Optional[int] = None
@staticmethod
@@ -169,7 +167,6 @@ class DaemonInstanceConfig:
instance_key=str(instance_key),
portfolio_hub_token=str(data.get("portfolio_hub_token") or ""),
trading_timezone=str(data.get("trading_timezone") or "Asia/Seoul"),
daily_preallocation_time=str(data.get("daily_preallocation_time") or "00:00"),
max_active_slots=max_active,
)
@@ -177,14 +174,13 @@ class DaemonInstanceConfig:
@dataclass(frozen=True)
class DaemonConfig:
dsn: str
total_slots: int
broker: str # 'paper' | 'kis'
execution_interval_sec: float
# BAT-013 / REQ-049: instance identity for multi-instance queue naming.
# "default" keeps legacy literal queue names (backward-compat).
instance_key: str = "default"
# dcy-020 / REQ-060 (BAT-028): M (max concurrent active slots) + asset-centric
# mode gate. None -> legacy pool model.
# dcy-020 / REQ-060: M (max concurrent active slots) — FCFS admission cap.
# BAT-030: asset-centric is the only model; None = no cap.
max_active_slots: Optional[int] = None
kis: Optional[KISConfig] = None
# REQ-011: paper broker tuning (only consulted when broker == 'paper').
@@ -204,11 +200,9 @@ class DaemonConfig:
# threshold in seconds. 0 disables (default). When >0, dispatched-pending
# slots older than this are surfaced (WARN + counter) without state change.
pending_stale_threshold_sec: int = 0
# dcy-014 / REQ-005 (BAT-018): trading_date timezone (default KST) + daily
# preallocation trigger time (HH:MM, TZ-local). Operative surface is the
# per-instance TOML (REQ-048); these are the env-only defaults.
# dcy-014 / REQ-048: operator-local timezone (CLI date display etc.). The
# daily preallocation trigger was removed with the pool model (BAT-030).
trading_timezone: str = "Asia/Seoul"
daily_preallocation_time: str = "00:00"
# dcy-015 / REQ-019 (BAT-019): ledger outbox retry backoff. base*2^(n-1)
# capped at cap (+jitter); after max_attempts a retryable row dead-letters.
ledger_retry_base_sec: float = 1.0
@@ -283,7 +277,6 @@ class DaemonConfig:
)
return DaemonConfig(
dsn=dsn,
total_slots=int(os.environ.get("TREX_TOTAL_SLOTS", "10")),
broker=broker,
execution_interval_sec=float(os.environ.get("TREX_EXECUTION_INTERVAL_SEC", "1.0")),
kis=kis,
@@ -323,12 +316,10 @@ class DaemonConfig:
pf = base.portfolio_hub
if inst.portfolio_hub_token and pf is not None:
pf = replace(pf, token=inst.portfolio_hub_token)
# dcy-014 / REQ-048: trading_date TZ + daily trigger come from TOML.
return replace(
base,
instance_key=inst.instance_key,
portfolio_hub=pf,
trading_timezone=inst.trading_timezone,
daily_preallocation_time=inst.daily_preallocation_time,
max_active_slots=inst.max_active_slots,
)
+235
View File
@@ -0,0 +1,235 @@
"""Asset-centric cutover tool (dcy-020 / REQ-064, BAT-030).
Full rebuild (design D1): wipe this instance's daily_slot rows and recreate
them from the portfolio-hub universe + the broker balance, with the broker as
the single source of truth for carried-over holdings (consistent with D3).
Procedure (one transaction):
1. preflight — abort if any pending_buy/pending_sell rows exist for the
instance (in-flight orders must converge first; run outside the trading
session).
2. DELETE all of the instance's slot rows.
3. one permanent slot per universe asset, status `inactive`.
4. each broker holding (symbol, qty) is mapped symbol→portfolio_id via the
universe and its slot is set to `holding` (qty = broker quantity;
avg_price is left NULL — the balance API does not carry it — WARN).
5. a broker holding whose symbol is NOT in the universe aborts the cutover
(operator decision; `--allow-unmapped` downgrades to WARN+skip).
6. audit rows in slot_transition_log (reason `cutover_rebuild`).
7. optional `--promote-unique`: portfolio_id SET NOT NULL (only after ALL
instances sharing the DB have been cut over).
The real cutover is OPERATOR-RUN (design D4): `trex-core-cutover --config
<instance> --yes` after a rehearsal against a test DB. Rollback = restore the
pre-cutover backup + previous release (BAT-030 design §4).
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Iterable, Optional
from uuid import uuid4
_log = logging.getLogger(__name__)
class CutoverAbort(RuntimeError):
"""Preflight/mapping condition that must stop the cutover."""
@dataclass
class CutoverReport:
instance_key: str
deleted: int = 0
created: int = 0
holdings: int = 0
unmapped: list[str] = field(default_factory=list)
promoted_unique: bool = False
dry_run: bool = False
def summary(self) -> str:
return (
f"cutover[{self.instance_key}] deleted={self.deleted} "
f"created={self.created} holdings={self.holdings} "
f"unmapped={self.unmapped or '[]'} "
f"promote_unique={self.promoted_unique} dry_run={self.dry_run}"
)
def run_cutover(
dsn: str,
instance_key: str,
universe: Iterable[tuple[str, str]],
balances: Iterable[tuple[str, int]],
*,
allow_unmapped: bool = False,
promote_unique: bool = False,
dry_run: bool = False,
psycopg_module: Any = None,
) -> CutoverReport:
"""Rebuild the instance's slots from `universe` + `balances` atomically.
universe: iterable of (portfolio_id, symbol).
balances: iterable of (symbol, hldg_qty) — broker ground truth.
Raises CutoverAbort on in-flight pending rows or unmapped broker holdings
(unless allow_unmapped). dry_run rolls the transaction back at the end.
"""
if psycopg_module is None:
import psycopg # type: ignore[import-not-found]
psycopg_module = psycopg
uni = [(str(pid), str(sym)) for pid, sym in universe if pid]
by_symbol = {sym: pid for pid, sym in uni}
held = [(str(sym), int(qty)) for sym, qty in balances if int(qty) > 0]
report = CutoverReport(instance_key=instance_key, dry_run=dry_run)
# mapping check before touching the DB (step 5)
unmapped = [sym for sym, _ in held if sym not in by_symbol]
if unmapped:
if not allow_unmapped:
raise CutoverAbort(
f"broker holds symbols outside the universe: {unmapped} "
"(resolve in portfolio-hub or pass --allow-unmapped to skip them)"
)
report.unmapped = unmapped
for sym in unmapped:
_log.warning(
"cutover: broker holding %s has no universe portfolio — SKIPPED "
"(unmanaged position; operator attention)", sym,
)
now = datetime.now(tz=timezone.utc)
with psycopg_module.connect(dsn, autocommit=False) as conn:
try:
with conn.cursor() as cur:
# 1. preflight: no in-flight pending rows for this instance
cur.execute(
"SELECT count(*) FROM daily_slot WHERE instance_key = %s "
"AND status IN ('pending_buy','pending_sell')",
(instance_key,),
)
pending = int(cur.fetchone()[0])
if pending:
raise CutoverAbort(
f"{pending} in-flight pending slot(s) for instance "
f"{instance_key!r} — wait for convergence (run outside "
"the trading session) and retry"
)
# 2. wipe the instance's rows (full rebuild, D1)
cur.execute(
"DELETE FROM daily_slot WHERE instance_key = %s", (instance_key,)
)
report.deleted = cur.rowcount
# 3+4. recreate: one permanent slot per universe asset
holding_pids = {}
for sym, qty in held:
pid = by_symbol.get(sym)
if pid is not None:
holding_pids[pid] = (sym, qty)
for pid, sym in uni:
sid = uuid4()
if pid in holding_pids:
h_sym, h_qty = holding_pids[pid]
cur.execute(
"INSERT INTO daily_slot (slot_id, status, symbol, qty, "
"instance_key, portfolio_id, created_at, updated_at) "
"VALUES (%s, 'holding', %s, %s, %s, %s, %s, %s)",
(sid, h_sym, h_qty, instance_key, pid, now, now),
)
report.holdings += 1
_log.warning(
"cutover: %s carried over as holding qty=%d "
"avg_price=NULL (balance API has no cost basis)",
pid, h_qty,
)
else:
cur.execute(
"INSERT INTO daily_slot (slot_id, status, symbol, "
"instance_key, portfolio_id, created_at, updated_at) "
"VALUES (%s, 'inactive', %s, %s, %s, %s, %s)",
(sid, sym, instance_key, pid, now, now),
)
report.created += 1
# 6. audit trail
cur.execute(
"INSERT INTO slot_transition_log "
"(slot_id, from_state, to_state, ts, reason, actor) "
"VALUES (%s, 'inactive', %s, %s, 'cutover_rebuild', 'cutover')",
(sid, "holding" if pid in holding_pids else "inactive", now),
)
# 7. optional promotion (all instances cut over)
if promote_unique:
cur.execute(
"ALTER TABLE daily_slot ALTER COLUMN portfolio_id SET NOT NULL"
)
report.promoted_unique = True
if dry_run:
conn.rollback()
else:
conn.commit()
except BaseException:
conn.rollback()
raise
return report
def main(argv: Optional[list[str]] = None) -> int: # pragma: no cover - CLI shell
import argparse
from trex_core.adapters.persistence.migrate import apply_migrations
from trex_core.runtime.config import DaemonConfig, resolve_config_toml
from trex_core.runtime.daemon import _build_broker, _build_portfolio_read
logging.basicConfig(level=logging.INFO)
parser = argparse.ArgumentParser(
prog="trex-core-cutover",
description="dcy-020/REQ-064: pool→asset-centric full-rebuild cutover "
"(operator-run; rehearse on a test DB first).",
)
parser.add_argument("--config", required=True, help="instance TOML name/path")
parser.add_argument("--config-path", default=None)
parser.add_argument("--env-file", default=None)
parser.add_argument("--yes", action="store_true",
help="actually apply (otherwise dry-run + rollback)")
parser.add_argument("--allow-unmapped", action="store_true",
help="skip (WARN) broker holdings outside the universe "
"instead of aborting")
parser.add_argument("--promote-unique", action="store_true",
help="portfolio_id SET NOT NULL (only after ALL "
"instances sharing this DB are cut over)")
args = parser.parse_args(argv)
toml_path = resolve_config_toml(args.config, args.config_path)
config = DaemonConfig.from_config(toml_path, env_file=args.env_file)
apply_migrations(config.dsn)
read = _build_portfolio_read(config)
daemon_id = read.whoami()
items = read.get_portfolios(daemon_id)
universe = [(it.portfolio_id, it.symbol) for it in items]
broker = _build_broker(config)
balances = [(b.symbol, b.hldg_qty) for b in broker.get_balance()]
try:
report = run_cutover(
config.dsn, config.instance_key, universe, balances,
allow_unmapped=args.allow_unmapped,
promote_unique=args.promote_unique and args.yes,
dry_run=not args.yes,
)
except CutoverAbort as exc:
_log.error("cutover ABORTED: %s", exc)
return 2
_log.info("%s", report.summary())
if not args.yes:
_log.info("dry-run only — re-run with --yes to apply")
return 0
if __name__ == "__main__": # pragma: no cover
raise SystemExit(main())
+12 -30
View File
@@ -9,8 +9,8 @@ Threads (started in `run()`):
- control RPC consumer — `trex.control.req` (pause/resume/list + kill_switch/resume_kill_switch, REQ-032/043)
- execution loop — broker dispatch + poll + slot transitions + event emit
- ledger writer — ledger outbox drain (REQ-019)
- snapshot scheduler — daily portfolio-hub snapshot refresh (REQ-030)
- preallocation scheduler — daily slot preallocation in trading TZ (REQ-005, dcy-014)
- snapshot scheduler — daily portfolio-hub snapshot refresh + universe
slot reconcile (REQ-030, dcy-020/REQ-058)
- safety-sell worker — forced_close / stop-loss·take-profit / market-close (REQ-037/038/039)
Each consumer/publisher uses its OWN RabbitMQTransport instance because
@@ -22,9 +22,7 @@ from __future__ import annotations
import logging
import signal
import threading
from datetime import datetime
from datetime import time as time_cls
from zoneinfo import ZoneInfo
from trex_core.adapters.broker.paper import PaperBroker
from trex_core.adapters.ledger import InMemoryLedger, PortfolioHubLedger
@@ -63,7 +61,6 @@ from trex_core.application.portfolio import (
SnapshotStore,
)
from trex_core.application.safety_sell import SafetySellService
from trex_core.application.preallocation_scheduler import DailyPreallocationScheduler
from trex_core.application.snapshot_scheduler import SnapshotScheduler
from trex_core.application.watchdog import WatchdogStore
from trex_core.domain.slot import SlotService
@@ -169,6 +166,7 @@ class Daemon:
slot_repo=self.slot_repo,
slot_service=self.slot_service,
killswitch_store=self.killswitch_store,
instance_key=config.instance_key,
)
self.portfolio_read = _build_portfolio_read(config)
self.snapshot_scheduler = SnapshotScheduler(
@@ -176,22 +174,12 @@ class Daemon:
store=self.snapshot_store,
broker=self.broker,
schedule_kst=_parse_kst_time(config.snapshot_schedule.kst_time),
# dcy-020 / REQ-058·062 (BAT-028/029): asset-centric slot reconcile +
# core-holding divergence reconcile when M set.
# dcy-020 / REQ-058·062 (BAT-030): universe slot reconcile (creates/
# retires the permanent asset slots) + core-holding divergence
# reconcile — always on (asset-centric is the only model).
slot_repo=self.slot_repo,
slot_service=self.slot_service,
instance_key=config.instance_key,
reconcile_enabled=config.max_active_slots is not None,
trading_date_fn=(lambda: datetime.now(ZoneInfo(config.trading_timezone)).date()),
)
# dcy-014 / REQ-005: daily slot preallocation in the configured
# trading timezone (default KST), so trading_date matches the local
# calendar and the next day's slots appear without a restart.
self.preallocation_scheduler = DailyPreallocationScheduler(
slot_service=self.slot_service,
total_slots=config.total_slots,
tz=ZoneInfo(config.trading_timezone),
schedule_time=_parse_kst_time(config.daily_preallocation_time),
)
# Event publisher transport — one dedicated instance for the
@@ -204,6 +192,7 @@ class Daemon:
ledger_outbox=self.outbox,
event_emitter=self._make_event_emitter(),
stale_pending_threshold_sec=config.pending_stale_threshold_sec,
instance_key=config.instance_key,
)
# BAT-011: safety-sell trigger service (forced_close / SL-TP / market-close).
@@ -278,9 +267,10 @@ class Daemon:
pass
handler = make_snapshot_rpc_handler(
self.slot_repo, self.slot_service,
instance_key=self.config.instance_key,
snapshot_store=self.snapshot_store,
pause_store=self.pause_store,
default_total_slots=self.config.total_slots,
max_active_slots=self.config.max_active_slots,
watchdog_store=self.watchdog,
watchdog_timeout_sec=self.config.safety_sell.watchdog_timeout_sec,
killswitch_store=self.killswitch_store,
@@ -319,19 +309,12 @@ class Daemon:
except Exception: # noqa: BLE001
_log.exception("snapshot scheduler loop crashed")
def _preallocation_loop(self) -> None:
try:
self.preallocation_scheduler.run_forever(self.stop, tick_sec=60.0)
except Exception: # noqa: BLE001
_log.exception("preallocation scheduler loop crashed")
# ------------------------------------------------------------ lifecycle
def setup(self) -> None:
apply_migrations(self.config.dsn)
# dcy-014 / REQ-005: preallocate against the configured trading timezone
# (default KST), not the system UTC date. Boot-time run_once ensures the
# current TZ-day's slots exist immediately (idempotent).
self.preallocation_scheduler.run_once()
# dcy-020 (BAT-030): the daily pool preallocation is gone — permanent
# asset slots are created/retired by the snapshot scheduler's universe
# reconcile (boot fetch runs immediately in its thread).
self.execution.reconcile() # REQ-014 restart recovery
try:
self.event_publisher.declare_topology(self.config.instance_key)
@@ -346,7 +329,6 @@ class Daemon:
("trex-snapshot-rpc", self._snapshot_rpc_loop),
("trex-control-rpc", self._control_rpc_loop),
("trex-snapshot-scheduler", self._snapshot_scheduler_loop),
("trex-preallocation", self._preallocation_loop),
("trex-safety-sell", self._safety_sell_loop),
]
for name, target in workers: