genichinandClaude Opus 4.8
78a743a2ef
- REQ-062 divergence: list_holding_slots(protocol+in_memory+PG) + reconcile_core_holdings(A=코어 holding/broker0→error, C=수량 불일치→warning, B=broker 보유/코어 없음→warning; 자동 청산 없음 issue #15). snapshot_scheduler 자산 중심 모드에서 best-effort 호출, daemon slot_service 배선. - REQ-063: slot_to_dict 에 universe_detached(retired_at) 추가 — slots 단일 뷰 tombstone 표시. - 신규 테스트: test_holding_divergence(4). M2(b) 청산 수량 브로커 정본(execution dispatch)은 라이브 집행 변경 → BAT-δ carry-over. 검증: 단위 393 passed / 36 skipped, PG-gated 18 passed, ruff clean. legacy 무파손. REQ-061/062/063 Approved 유지(γ 경로 도입 검증; default 전환·M2(b) 는 BAT-δ). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
555 lines
25 KiB
Python
555 lines
25 KiB
Python
"""PostgreSQL persistence adapters (REQ-006).
|
||
|
||
- PostgreSQLSlotRepository: implements the domain SlotRepository protocol
|
||
(single-writer, pg_advisory_xact_lock serialization).
|
||
- PostgreSQLIntentStore: implements the application IntentStore port
|
||
(intent_log dedup gate + audit).
|
||
|
||
psycopg (v3) is imported lazily so the domain/test paths that use the
|
||
in-memory adapters don't require it. External I/O lives here, never in the
|
||
domain layer (NFR-1).
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from contextlib import contextmanager
|
||
from datetime import date, datetime, timezone
|
||
from typing import Any, Iterator, Optional
|
||
from uuid import UUID, uuid4
|
||
|
||
from trex_core.application.ledger_outbox import (
|
||
LedgerWorkItem,
|
||
OutboxOutcome,
|
||
WorkItemStatus,
|
||
)
|
||
from trex_core.application.ledger_port import LedgerTrade
|
||
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, "
|
||
"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"
|
||
)
|
||
|
||
|
||
def _require_psycopg(module: Any = None) -> Any:
|
||
if module is not None:
|
||
return module
|
||
try:
|
||
import psycopg # type: ignore[import-not-found]
|
||
except ImportError as exc: # pragma: no cover - environment guard
|
||
raise RuntimeError(
|
||
"psycopg is required for PostgreSQL adapters. Install 'trex-core[postgres]'."
|
||
) from exc
|
||
return psycopg
|
||
|
||
|
||
class PostgreSQLSlotRepository:
|
||
"""psycopg-backed SlotRepository. Connection-per-call (orbit-proven)."""
|
||
|
||
def __init__(self, dsn: str, *, psycopg_module: Any = None) -> None:
|
||
self._psycopg = _require_psycopg(psycopg_module)
|
||
self._dsn = dsn
|
||
|
||
@contextmanager
|
||
def transactional_advisory_lock(self, lock_id: int) -> Iterator[None]:
|
||
with self._psycopg.connect(self._dsn, autocommit=False) as conn:
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute("SELECT pg_advisory_xact_lock(%s)", (lock_id,))
|
||
yield
|
||
conn.commit()
|
||
except BaseException:
|
||
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]:
|
||
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,),
|
||
)
|
||
return [self._row_to_slot(r) for r in cur.fetchall()]
|
||
|
||
def fetch_slot(self, slot_id: UUID) -> 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 slot_id = %s", (slot_id,)
|
||
)
|
||
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:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
f"SELECT {_SLOT_COLUMNS} FROM daily_slot "
|
||
"WHERE instance_key = %s AND portfolio_id = %s LIMIT 1",
|
||
(instance_key, portfolio_id),
|
||
)
|
||
row = cur.fetchone()
|
||
return self._row_to_slot(row) if row else None
|
||
|
||
def count_active(self, instance_key: str) -> int:
|
||
with self._psycopg.connect(self._dsn) as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT count(*) FROM daily_slot WHERE instance_key = %s "
|
||
"AND status IN ('holding','pending_buy','pending_sell')",
|
||
(instance_key,),
|
||
)
|
||
row = cur.fetchone()
|
||
return int(row[0]) if row else 0
|
||
|
||
def list_holding_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 instance_key = %s AND status = 'holding'",
|
||
(instance_key,),
|
||
)
|
||
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],
|
||
*, allow_retire: bool = True,
|
||
) -> dict[str, int]:
|
||
now = datetime.now(tz=timezone.utc)
|
||
created = retired = tombstoned = 0
|
||
with self._psycopg.connect(self._dsn, autocommit=False) as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT portfolio_id, status, retired_at FROM daily_slot "
|
||
"WHERE instance_key = %s AND portfolio_id IS NOT NULL",
|
||
(instance_key,),
|
||
)
|
||
existing = {r[0]: (r[1], r[2]) for r in cur.fetchall()}
|
||
# create new asset slots (idempotent via partial unique)
|
||
for pid in universe_portfolio_ids:
|
||
if pid in existing:
|
||
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) "
|
||
"ON CONFLICT (instance_key, portfolio_id) "
|
||
"WHERE portfolio_id IS NOT NULL DO NOTHING",
|
||
(uuid4(), trading_date, instance_key, pid, now, now),
|
||
)
|
||
created += cur.rowcount
|
||
# retire / tombstone assets that left the universe (skipped under
|
||
# the mass-disappearance guard, dcy-020/OQ-2 D3)
|
||
if allow_retire:
|
||
for pid, (status, retired_at) in existing.items():
|
||
if pid in universe_portfolio_ids:
|
||
continue
|
||
if status in ("holding", "pending_buy", "pending_sell"):
|
||
tombstoned += 1
|
||
elif retired_at is None:
|
||
cur.execute(
|
||
"UPDATE daily_slot SET retired_at = %s, updated_at = %s "
|
||
"WHERE instance_key = %s AND portfolio_id = %s",
|
||
(now, now, instance_key, pid),
|
||
)
|
||
retired += 1
|
||
conn.commit()
|
||
return {"created": created, "retired": retired, "tombstoned": tombstoned}
|
||
|
||
def update_slot(self, slot: Slot) -> None:
|
||
with self._psycopg.connect(self._dsn, autocommit=False) as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"UPDATE daily_slot SET status=%s, symbol=%s, qty=%s, avg_price=%s, "
|
||
"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, "
|
||
"updated_at=%s WHERE slot_id=%s",
|
||
(
|
||
slot.status.value, slot.symbol, slot.qty, slot.avg_price,
|
||
slot.limit_price, slot.allocated_capital, slot.order_id,
|
||
slot.strategy_id, slot.broker_order_id, slot.portfolio_id,
|
||
slot.sell_strategy_id, slot.forced_close_at,
|
||
slot.stop_loss_price, slot.take_profit_price,
|
||
slot.entry_count, slot.instance_key, slot.retired_at,
|
||
slot.updated_at or datetime.now(tz=timezone.utc), slot.slot_id,
|
||
),
|
||
)
|
||
if cur.rowcount == 0:
|
||
conn.rollback()
|
||
raise SlotNotFoundError(f"slot not found: {slot.slot_id}")
|
||
conn.commit()
|
||
|
||
def append_transition_log(self, record: SlotTransitionRecord) -> None:
|
||
with self._psycopg.connect(self._dsn, autocommit=False) as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"INSERT INTO slot_transition_log "
|
||
"(slot_id, from_state, to_state, ts, reason, actor, extra) "
|
||
"VALUES (%s, %s, %s, %s, %s, %s, %s)",
|
||
(
|
||
record.slot_id, record.from_state.value, record.to_state.value,
|
||
record.ts, record.reason, record.actor, json.dumps(record.extra),
|
||
),
|
||
)
|
||
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]:
|
||
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"
|
||
)
|
||
row = cur.fetchone()
|
||
return self._row_to_slot(row) if row else None
|
||
|
||
def mark_dispatched(self, slot_id: UUID, broker_order_id: str) -> bool:
|
||
with self._psycopg.connect(self._dsn, autocommit=False) as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"UPDATE daily_slot SET broker_order_id=%s, updated_at=%s "
|
||
"WHERE slot_id=%s AND broker_order_id IS NULL",
|
||
(broker_order_id, datetime.now(tz=timezone.utc), slot_id),
|
||
)
|
||
ok = cur.rowcount == 1
|
||
conn.commit()
|
||
return ok
|
||
|
||
def list_dispatched_pending(self) -> 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"
|
||
)
|
||
return [self._row_to_slot(r) for r in cur.fetchall()]
|
||
|
||
def clear_dispatch(self, slot_id: UUID) -> None:
|
||
with self._psycopg.connect(self._dsn, autocommit=False) as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"UPDATE daily_slot SET broker_order_id=NULL, updated_at=%s WHERE slot_id=%s",
|
||
(datetime.now(tz=timezone.utc), slot_id),
|
||
)
|
||
conn.commit()
|
||
|
||
def count_slots_by_status(self, trading_date: date) -> 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,),
|
||
)
|
||
return {SlotState(r[0]): int(r[1]) for r in cur.fetchall()}
|
||
|
||
@staticmethod
|
||
def _row_to_slot(row: tuple) -> Slot:
|
||
from decimal import Decimal
|
||
f = lambda v: float(v) if v is not None else None # noqa: E731
|
||
d = lambda v: Decimal(str(v)) if v is not None else None # noqa: E731
|
||
# Column order: slot_id(0) 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)
|
||
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],
|
||
)
|
||
|
||
|
||
class PostgreSQLIntentStore:
|
||
"""intent_log-backed IntentStore. intent_id UNIQUE is the dedup gate."""
|
||
|
||
def __init__(self, dsn: str, *, psycopg_module: Any = None) -> None:
|
||
self._psycopg = _require_psycopg(psycopg_module)
|
||
self._dsn = dsn
|
||
|
||
def try_insert(self, intent_id: str, intent_type: str, payload: dict) -> bool:
|
||
with self._psycopg.connect(self._dsn, autocommit=False) as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"INSERT INTO intent_log (intent_id, intent_type, payload, status) "
|
||
"VALUES (%s, %s, %s, 'processing') ON CONFLICT (intent_id) DO NOTHING",
|
||
(intent_id, intent_type, json.dumps(payload)),
|
||
)
|
||
inserted = cur.rowcount == 1
|
||
conn.commit()
|
||
return inserted
|
||
|
||
def get(self, intent_id: str) -> Optional[IntentRecord]:
|
||
with self._psycopg.connect(self._dsn) as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT intent_id, intent_type, status, slot_id, reason "
|
||
"FROM intent_log WHERE intent_id = %s",
|
||
(intent_id,),
|
||
)
|
||
row = cur.fetchone()
|
||
if row is None:
|
||
return None
|
||
return _PgIntentRecord(
|
||
intent_id=row[0], intent_type=row[1], status=row[2],
|
||
slot_id=(row[3] if isinstance(row[3], UUID) or row[3] is None else UUID(str(row[3]))),
|
||
reason=row[4],
|
||
)
|
||
|
||
def finalize(
|
||
self, intent_id: str, status: str, slot_id: Optional[UUID], reason: Optional[str]
|
||
) -> None:
|
||
with self._psycopg.connect(self._dsn, autocommit=False) as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"UPDATE intent_log SET status=%s, slot_id=%s, reason=%s, updated_at=%s "
|
||
"WHERE intent_id=%s",
|
||
(status, slot_id, reason, datetime.now(tz=timezone.utc), intent_id),
|
||
)
|
||
conn.commit()
|
||
|
||
|
||
class _PgIntentRecord:
|
||
def __init__(self, intent_id, intent_type, status, slot_id, reason):
|
||
self.intent_id = intent_id
|
||
self.intent_type = intent_type
|
||
self.status = status
|
||
self.slot_id = slot_id
|
||
self.reason = reason
|
||
|
||
|
||
# --- ledger outbox (BAT-006 / REQ-019) ---------------------------------------
|
||
|
||
_OUTBOX_COLUMNS = (
|
||
"correlation_id, asset_id, order_id, execution_id, side, qty, price, "
|
||
"fee, tax, executed_at, status, attempts, last_outcome, last_error, "
|
||
"created_at, updated_at, next_retry_at"
|
||
)
|
||
|
||
|
||
class PostgreSQLLedgerOutbox:
|
||
"""ledger_outbox-backed LedgerOutbox. correlation_id PK is the dedup gate."""
|
||
|
||
def __init__(self, dsn: str, *, psycopg_module: Any = None) -> None:
|
||
self._psycopg = _require_psycopg(psycopg_module)
|
||
self._dsn = dsn
|
||
|
||
def enqueue(self, trade: LedgerTrade) -> LedgerWorkItem:
|
||
with self._psycopg.connect(self._dsn, autocommit=False) as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"INSERT INTO ledger_outbox "
|
||
"(correlation_id, asset_id, order_id, execution_id, side, "
|
||
" qty, price, fee, tax, executed_at) "
|
||
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) "
|
||
"ON CONFLICT (correlation_id) DO NOTHING",
|
||
(
|
||
trade.correlation_id, trade.asset_id, trade.order_id,
|
||
trade.execution_id, trade.side,
|
||
trade.qty, trade.price, trade.fee, trade.tax,
|
||
trade.executed_at,
|
||
),
|
||
)
|
||
conn.commit()
|
||
existing = self.get(trade.correlation_id)
|
||
assert existing is not None
|
||
return existing
|
||
|
||
def list_pending(self, limit: int = 32) -> list[LedgerWorkItem]:
|
||
with self._psycopg.connect(self._dsn) as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
f"SELECT {_OUTBOX_COLUMNS} FROM ledger_outbox "
|
||
"WHERE status = 'pending' "
|
||
"AND (next_retry_at IS NULL OR next_retry_at <= now()) "
|
||
"ORDER BY created_at LIMIT %s",
|
||
(limit,),
|
||
)
|
||
return [self._row_to_item(r) for r in cur.fetchall()]
|
||
|
||
def pending_cash_delta(self) -> dict[str, float]:
|
||
# dcy-018 / REQ-029 (OQ-4): estimated cash delta of all pending rows by
|
||
# asset_id. SUM in SQL: sell → +(qty*price - fee - tax), buy → negative.
|
||
with self._psycopg.connect(self._dsn) as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT asset_id, "
|
||
"SUM(CASE WHEN side='sell' "
|
||
" THEN (qty*price - fee - tax) "
|
||
" ELSE -(qty*price + fee + tax) END) "
|
||
"FROM ledger_outbox WHERE status='pending' GROUP BY asset_id"
|
||
)
|
||
return {r[0]: float(r[1]) for r in cur.fetchall() if r[1] is not None}
|
||
|
||
def get(self, correlation_id: str) -> Optional[LedgerWorkItem]:
|
||
with self._psycopg.connect(self._dsn) as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
f"SELECT {_OUTBOX_COLUMNS} FROM ledger_outbox WHERE correlation_id = %s",
|
||
(correlation_id,),
|
||
)
|
||
row = cur.fetchone()
|
||
return self._row_to_item(row) if row else None
|
||
|
||
def mark_outcome(
|
||
self,
|
||
correlation_id: str,
|
||
outcome: OutboxOutcome,
|
||
*,
|
||
error: Optional[str] = None,
|
||
next_retry_at: Optional[datetime] = None,
|
||
) -> LedgerWorkItem:
|
||
with self._psycopg.connect(self._dsn, autocommit=False) as conn:
|
||
with conn.cursor() as cur:
|
||
if outcome is OutboxOutcome.COMPLETED:
|
||
status_value = WorkItemStatus.COMPLETED.value
|
||
attempts_delta = 1
|
||
elif outcome is OutboxOutcome.CONFLICT_BLOCKED:
|
||
status_value = WorkItemStatus.CONFLICT_BLOCKED.value
|
||
attempts_delta = 1
|
||
elif outcome in (OutboxOutcome.PERMANENT_FAILURE, OutboxOutcome.DEAD_LETTERED):
|
||
# dcy-015 / REQ-019: terminal delivery_failed (reason in last_outcome).
|
||
status_value = WorkItemStatus.DELIVERY_FAILED.value
|
||
attempts_delta = 1
|
||
elif outcome is OutboxOutcome.DUPLICATE_COMPLETED:
|
||
status_value = WorkItemStatus.COMPLETED.value
|
||
attempts_delta = 0
|
||
else: # retryable
|
||
status_value = WorkItemStatus.PENDING.value
|
||
attempts_delta = 1
|
||
# next_retry_at only persists for a retryable backoff; cleared otherwise.
|
||
nra = next_retry_at if outcome is OutboxOutcome.RETRYABLE_FAILURE else None
|
||
cur.execute(
|
||
"UPDATE ledger_outbox SET status=%s, attempts=attempts+%s, "
|
||
"last_outcome=%s, last_error=%s, next_retry_at=%s, updated_at=%s "
|
||
"WHERE correlation_id=%s",
|
||
(
|
||
status_value, attempts_delta, outcome.value, error, nra,
|
||
datetime.now(tz=timezone.utc), correlation_id,
|
||
),
|
||
)
|
||
conn.commit()
|
||
row = self.get(correlation_id)
|
||
assert row is not None
|
||
return row
|
||
|
||
@staticmethod
|
||
def _row_to_item(row: tuple) -> LedgerWorkItem:
|
||
f = lambda v: float(v) if v is not None else 0.0 # noqa: E731
|
||
trade = LedgerTrade(
|
||
asset_id=row[1], order_id=row[2], execution_id=row[3], side=row[4],
|
||
qty=f(row[5]), price=f(row[6]), fee=f(row[7]), tax=f(row[8]),
|
||
executed_at=row[9],
|
||
)
|
||
return LedgerWorkItem(
|
||
correlation_id=row[0],
|
||
trade=trade,
|
||
status=WorkItemStatus(row[10]),
|
||
attempts=int(row[11]) if row[11] is not None else 0,
|
||
last_outcome=OutboxOutcome(row[12]) if row[12] else None,
|
||
last_error=row[13],
|
||
created_at=row[14],
|
||
updated_at=row[15],
|
||
next_retry_at=row[16],
|
||
)
|