Implement BAT-006: portfolio-hub ledger integration (REQ-017~019)
- REQ-017: LedgerPort contract (application/ledger_port.py) + portfolio-hub adapter + in-memory stub (adapters/ledger/), TREX_PORTFOLIO_HUB_* config - REQ-018: fill→ledger recording in execution loop; FillStatus carries execution_id (deterministic synth) + fee; paper/kis adapters updated - REQ-019: ledger outbox + retry worker (application/ledger_outbox.py), correlation_id idempotency, outcome vocabulary; 002_create_ledger_outbox.sql - Tests: in-memory/portfolio-hub ledger, execution ledger hook, outbox, domain import sweep Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.8
@@ -15,7 +15,12 @@ from typing import Optional
|
||||
|
||||
import httpx
|
||||
|
||||
from trex_core.application.broker_port import FillStatus, OrderAck, OrderRequest
|
||||
from trex_core.application.broker_port import (
|
||||
FillStatus,
|
||||
OrderAck,
|
||||
OrderRequest,
|
||||
calculate_fee,
|
||||
)
|
||||
|
||||
_KST = timezone(timedelta(hours=9))
|
||||
_MOCK_HINTS = ("openapivts", "virtual", "mock")
|
||||
@@ -175,8 +180,19 @@ class KISBroker:
|
||||
if ccld_qty <= 0:
|
||||
return FillStatus(state="accepted", raw=row)
|
||||
if ccld_qty >= ord_qty and ord_qty > 0:
|
||||
return FillStatus(state="filled", filled_qty=ccld_qty,
|
||||
avg_price=float(avg) if avg else None, raw=row)
|
||||
# Deterministic execution_id (dcy-004 OQ-3, orbit kis.py rule):
|
||||
# kis-daily-ccld:{ord_dt}:{ord_gno_brno}:{odno}:{tot_ccld_qty}
|
||||
ord_dt = str(row.get("ord_dt") or today)
|
||||
ord_gno_brno = str(row.get("ord_gno_brno") or "")
|
||||
row_odno = str(row.get("odno") or odno)
|
||||
execution_id = f"kis-daily-ccld:{ord_dt}:{ord_gno_brno}:{row_odno}:{ccld_qty}"
|
||||
fill_price = float(avg) if avg else 0.0
|
||||
fee = calculate_fee(side, ccld_qty * fill_price)
|
||||
return FillStatus(
|
||||
state="filled", filled_qty=ccld_qty,
|
||||
avg_price=fill_price or None,
|
||||
execution_id=execution_id, fee=fee, tax=0.0, raw=row,
|
||||
)
|
||||
return FillStatus(state="partially_filled", filled_qty=ccld_qty,
|
||||
avg_price=float(avg) if avg else None, raw=row)
|
||||
|
||||
|
||||
@@ -10,7 +10,12 @@ from __future__ import annotations
|
||||
import threading
|
||||
from typing import Optional
|
||||
|
||||
from trex_core.application.broker_port import FillStatus, OrderAck, OrderRequest
|
||||
from trex_core.application.broker_port import (
|
||||
FillStatus,
|
||||
OrderAck,
|
||||
OrderRequest,
|
||||
calculate_fee,
|
||||
)
|
||||
|
||||
|
||||
class _Order:
|
||||
@@ -72,7 +77,14 @@ class PaperBroker:
|
||||
step = self._qty_per_poll if self._qty_per_poll is not None else (o.qty - o.filled)
|
||||
o.filled = min(o.qty, o.filled + step)
|
||||
if o.filled >= o.qty:
|
||||
return FillStatus(state="filled", filled_qty=o.filled, avg_price=o.price)
|
||||
# Deterministic execution_id (dcy-004 OQ-3):
|
||||
# paper:{broker_order_id}:{filled_qty}
|
||||
execution_id = f"paper:{o.oid}:{o.filled}"
|
||||
fee = calculate_fee(o.side, o.filled * o.price)
|
||||
return FillStatus(
|
||||
state="filled", filled_qty=o.filled, avg_price=o.price,
|
||||
execution_id=execution_id, fee=fee, tax=0.0,
|
||||
)
|
||||
return FillStatus(state="partially_filled", filled_qty=o.filled, avg_price=o.price)
|
||||
|
||||
def cancel_order(self, broker_order_id: str, *, side: str, symbol: str, qty: int) -> bool:
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
"""Ledger adapters (REQ-017, dcy-004).
|
||||
|
||||
- InMemoryLedger: deterministic stub for tests (no network).
|
||||
- PortfolioHubLedger: HTTP adapter for the external portfolio-hub ledger.
|
||||
"""
|
||||
from trex_core.adapters.ledger.in_memory import InMemoryLedger
|
||||
from trex_core.adapters.ledger.portfolio_hub import PortfolioHubLedger
|
||||
|
||||
__all__ = ["InMemoryLedger", "PortfolioHubLedger"]
|
||||
@@ -0,0 +1,65 @@
|
||||
"""In-memory ledger — deterministic stub for tests (REQ-017 AC-1)."""
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
|
||||
from trex_core.application.ledger_port import (
|
||||
LedgerConflictError,
|
||||
LedgerRetryableError,
|
||||
LedgerTrade,
|
||||
LedgerWriteResult,
|
||||
)
|
||||
|
||||
|
||||
class InMemoryLedger:
|
||||
"""Server-side idempotent stub.
|
||||
|
||||
- records every accepted trade keyed by execution_id (mirrors orbit
|
||||
post_ledger_trade idempotency).
|
||||
- failure injection lets tests model portfolio-hub outages without HTTP:
|
||||
fail_n_then_succeed -- raise LedgerRetryableError this many times,
|
||||
then accept.
|
||||
conflict_executions -- execution_ids that raise LedgerConflictError
|
||||
(simulates portfolio-hub 409).
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
fail_n_then_succeed: int = 0,
|
||||
conflict_executions: set[str] | None = None,
|
||||
) -> None:
|
||||
self._trades: dict[str, LedgerTrade] = {}
|
||||
self._order: list[LedgerTrade] = []
|
||||
self._lock = threading.Lock()
|
||||
self._fail_remaining = fail_n_then_succeed
|
||||
self._conflicts = set(conflict_executions or set())
|
||||
|
||||
def record_trade(self, trade: LedgerTrade) -> LedgerWriteResult:
|
||||
with self._lock:
|
||||
if trade.execution_id in self._conflicts:
|
||||
raise LedgerConflictError(
|
||||
f"ledger conflict for execution_id={trade.execution_id}"
|
||||
)
|
||||
if self._fail_remaining > 0:
|
||||
self._fail_remaining -= 1
|
||||
raise LedgerRetryableError("simulated transient failure")
|
||||
if trade.execution_id in self._trades:
|
||||
# idempotent: same execution_id => no new entry, report dedup.
|
||||
return LedgerWriteResult.DUPLICATE_OK
|
||||
self._trades[trade.execution_id] = trade
|
||||
self._order.append(trade)
|
||||
return LedgerWriteResult.NEW
|
||||
|
||||
# --- introspection for tests ---
|
||||
def trades(self) -> list[LedgerTrade]:
|
||||
with self._lock:
|
||||
return list(self._order)
|
||||
|
||||
def count(self) -> int:
|
||||
with self._lock:
|
||||
return len(self._order)
|
||||
|
||||
def has_execution(self, execution_id: str) -> bool:
|
||||
with self._lock:
|
||||
return execution_id in self._trades
|
||||
@@ -0,0 +1,91 @@
|
||||
"""portfolio-hub ledger HTTP adapter (REQ-017, dcy-004).
|
||||
|
||||
Mirrors trex-orbit `PortfolioHubClient.post_ledger_trade`:
|
||||
POST /v1/assets/{asset_id}/ledger/trades
|
||||
body keys: side, qty, price, fee, tax, executed_at, execution_id, order_id
|
||||
- 2xx => recorded (or duplicate accepted idempotently)
|
||||
- 409 => LedgerConflictError (operator resolution)
|
||||
- 4xx (other)/5xx => LedgerRetryableError (outbox retries)
|
||||
- httpx error => LedgerRetryableError
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Optional
|
||||
|
||||
import httpx
|
||||
|
||||
from trex_core.application.ledger_port import (
|
||||
LedgerConflictError,
|
||||
LedgerRetryableError,
|
||||
LedgerTrade,
|
||||
LedgerWriteResult,
|
||||
)
|
||||
|
||||
|
||||
class PortfolioHubLedger:
|
||||
"""portfolio-hub ledger adapter (single endpoint, single responsibility)."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
base_url: str,
|
||||
*,
|
||||
token: Optional[str] = None,
|
||||
timeout: float = 5.0,
|
||||
client: Optional[httpx.Client] = None,
|
||||
) -> None:
|
||||
self._base_url = base_url.rstrip("/")
|
||||
self._token = token
|
||||
self._timeout = timeout
|
||||
self._client = client # injected client (e.g. MockTransport) for tests
|
||||
|
||||
def _headers(self) -> dict[str, str]:
|
||||
h = {"content-type": "application/json; charset=utf-8"}
|
||||
if self._token:
|
||||
h["authorization"] = f"Bearer {self._token}"
|
||||
return h
|
||||
|
||||
@staticmethod
|
||||
def _payload(trade: LedgerTrade) -> dict:
|
||||
return {
|
||||
"side": trade.side,
|
||||
"qty": trade.qty,
|
||||
"price": trade.price,
|
||||
"fee": trade.fee,
|
||||
"tax": trade.tax,
|
||||
"executed_at": trade.executed_at.isoformat() if trade.executed_at else None,
|
||||
"execution_id": trade.execution_id,
|
||||
"order_id": trade.order_id,
|
||||
}
|
||||
|
||||
def _post(self, url: str, json: dict) -> httpx.Response:
|
||||
if self._client is not None:
|
||||
return self._client.post(url, json=json, headers=self._headers(),
|
||||
timeout=self._timeout)
|
||||
return httpx.post(url, json=json, headers=self._headers(),
|
||||
timeout=self._timeout)
|
||||
|
||||
def record_trade(self, trade: LedgerTrade) -> LedgerWriteResult:
|
||||
url = f"{self._base_url}/v1/assets/{trade.asset_id}/ledger/trades"
|
||||
try:
|
||||
resp = self._post(url, json=self._payload(trade))
|
||||
except httpx.HTTPError as exc:
|
||||
raise LedgerRetryableError(f"portfolio-hub network error: {exc!s}") from exc
|
||||
|
||||
if 200 <= resp.status_code < 300:
|
||||
# orbit convention: response body may carry a "duplicate" flag when
|
||||
# the server has already recorded this execution_id idempotently.
|
||||
try:
|
||||
body = resp.json() if resp.content else {}
|
||||
except ValueError:
|
||||
body = {}
|
||||
if isinstance(body, dict) and body.get("duplicate") is True:
|
||||
return LedgerWriteResult.DUPLICATE_OK
|
||||
return LedgerWriteResult.NEW
|
||||
if resp.status_code == 409:
|
||||
raise LedgerConflictError(
|
||||
f"portfolio-hub 409 conflict for execution_id={trade.execution_id}: "
|
||||
f"{resp.text[:200]}"
|
||||
)
|
||||
raise LedgerRetryableError(
|
||||
f"portfolio-hub {resp.status_code} response: {resp.text[:200]}"
|
||||
)
|
||||
@@ -17,6 +17,12 @@ 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
|
||||
|
||||
@@ -300,3 +306,115 @@ class _PgIntentRecord:
|
||||
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"
|
||||
)
|
||||
|
||||
|
||||
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' ORDER BY created_at LIMIT %s",
|
||||
(limit,),
|
||||
)
|
||||
return [self._row_to_item(r) for r in cur.fetchall()]
|
||||
|
||||
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,
|
||||
) -> 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 is OutboxOutcome.DUPLICATE_COMPLETED:
|
||||
status_value = WorkItemStatus.COMPLETED.value
|
||||
attempts_delta = 0
|
||||
else: # retryable
|
||||
status_value = WorkItemStatus.PENDING.value
|
||||
attempts_delta = 1
|
||||
cur.execute(
|
||||
"UPDATE ledger_outbox SET status=%s, attempts=attempts+%s, "
|
||||
"last_outcome=%s, last_error=%s, updated_at=%s "
|
||||
"WHERE correlation_id=%s",
|
||||
(
|
||||
status_value, attempts_delta, outcome.value, error,
|
||||
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],
|
||||
)
|
||||
|
||||
@@ -0,0 +1,33 @@
|
||||
-- BAT-006 / REQ-019: ledger outbox work-item store (dcy-004 OQ-2).
|
||||
--
|
||||
-- Each row is one fill that needs to be recorded in the portfolio-hub
|
||||
-- ledger. `correlation_id` ({asset_id}:{order_id}:{execution_id}) is the
|
||||
-- local dedup key — the same fill enqueued twice must NOT produce two
|
||||
-- ledger entries. Outcomes are classified per dcy-004 FR-4:
|
||||
-- completed | duplicate_completed | retryable_failure | conflict_blocked.
|
||||
|
||||
CREATE TABLE IF NOT EXISTS ledger_outbox (
|
||||
correlation_id TEXT PRIMARY KEY,
|
||||
asset_id TEXT NOT NULL,
|
||||
order_id TEXT NOT NULL,
|
||||
execution_id TEXT NOT NULL,
|
||||
side TEXT NOT NULL CHECK (side IN ('buy','sell')),
|
||||
qty NUMERIC(18, 4) NOT NULL,
|
||||
price NUMERIC(18, 4) NOT NULL,
|
||||
fee NUMERIC(18, 4) NOT NULL DEFAULT 0,
|
||||
tax NUMERIC(18, 4) NOT NULL DEFAULT 0,
|
||||
executed_at TIMESTAMPTZ,
|
||||
status TEXT NOT NULL DEFAULT 'pending'
|
||||
CHECK (status IN ('pending','completed','conflict_blocked')),
|
||||
attempts INTEGER NOT NULL DEFAULT 0,
|
||||
last_outcome TEXT
|
||||
CHECK (last_outcome IS NULL OR last_outcome IN (
|
||||
'completed','duplicate_completed','retryable_failure','conflict_blocked'
|
||||
)),
|
||||
last_error TEXT,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT (now() AT TIME ZONE 'UTC'),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT (now() AT TIME ZONE 'UTC')
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_ledger_outbox_status_created
|
||||
ON ledger_outbox (status, created_at);
|
||||
@@ -34,6 +34,12 @@ class FillStatus:
|
||||
state: str # accepted | partially_filled | filled | rejected | cancelled | unknown
|
||||
filled_qty: int = 0
|
||||
avg_price: Optional[float] = None
|
||||
# REQ-018: deterministic execution_id + broker-computed fee/tax carried
|
||||
# all the way to the ledger writer. tax defaults to 0 (precise tax model
|
||||
# is a follow-up Discovery, dcy-004 OQ-3).
|
||||
execution_id: Optional[str] = None
|
||||
fee: float = 0.0
|
||||
tax: float = 0.0
|
||||
raw: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
@property
|
||||
@@ -62,3 +68,22 @@ class BrokerPort(Protocol):
|
||||
def cancel_order(self, broker_order_id: str, *, side: str, symbol: str, qty: int) -> bool:
|
||||
"""Best-effort cancel. Returns True if the broker accepted the cancel."""
|
||||
...
|
||||
|
||||
|
||||
# --- shared fee model (REQ-018, orbit calculate_fee pattern) -----------------
|
||||
# Conservative KR equities approximation (bps). Brokers can override, but we
|
||||
# need a single default so paper and KIS produce comparable ledger entries.
|
||||
_FEE_BPS_BUY = 1.5 # 0.015 % broker commission on buys
|
||||
_FEE_BPS_SELL = 1.5 # 0.015 % broker commission on sells (tax is tracked separately)
|
||||
|
||||
|
||||
def calculate_fee(side: str, amount: float) -> float:
|
||||
"""Estimate broker fee for a notional `amount` (qty * price).
|
||||
|
||||
Mirrors orbit `calculate_fee(side, amount)` shape — a single point of
|
||||
fee policy so all brokers and the ledger agree on the number.
|
||||
"""
|
||||
if amount <= 0:
|
||||
return 0.0
|
||||
bps = _FEE_BPS_BUY if side == "buy" else _FEE_BPS_SELL
|
||||
return round(amount * bps / 10_000.0, 4)
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
"""Order execution service (REQ-013, REQ-014, dcy-003).
|
||||
"""Order execution service (REQ-013, REQ-014, REQ-018, dcy-003, dcy-004).
|
||||
|
||||
A single background execution loop (invariant A single writer, invariant B
|
||||
single broker session) that, per tick:
|
||||
@@ -6,17 +6,25 @@ single broker session) that, per tick:
|
||||
submit the order, record broker_order_id (idempotency guard).
|
||||
poll: observe fills on dispatched pending slots and transition the slot
|
||||
(pending_buy->holding / pending_sell->closed), 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).
|
||||
reconcile() re-polls dispatched slots on restart so a crash after dispatch
|
||||
converges without duplicate orders (REQ-014).
|
||||
|
||||
All slot writes go through SlotService (advisory lock). The broker is touched
|
||||
only here, never by the domain or the HTTP API.
|
||||
only here, never by the domain or the HTTP API. The ledger outbox is the
|
||||
single producer for ledger work items (NFR-3, single record subject).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from typing import Optional
|
||||
|
||||
from trex_core.application.broker_port import BrokerPort, OrderRequest
|
||||
from trex_core.application.ledger_outbox import LedgerOutbox
|
||||
from trex_core.application.ledger_port import LedgerTrade
|
||||
from trex_core.domain.slot import SlotService, SlotState, SlotStateMismatchError
|
||||
|
||||
_MAX_DISPATCH_PER_TICK = 64
|
||||
@@ -27,13 +35,22 @@ class ExecutionResult:
|
||||
dispatched: int = 0
|
||||
filled: int = 0
|
||||
rolled_back: int = 0
|
||||
ledger_enqueued: int = 0 # REQ-019: ledger work items appended this tick
|
||||
ledger_skipped: int = 0 # REQ-018 AC-4: fills without portfolio_id
|
||||
|
||||
|
||||
class ExecutionService:
|
||||
def __init__(self, slot_service: SlotService, repo, broker: BrokerPort) -> None:
|
||||
def __init__(
|
||||
self,
|
||||
slot_service: SlotService,
|
||||
repo,
|
||||
broker: BrokerPort,
|
||||
ledger_outbox: Optional[LedgerOutbox] = None,
|
||||
) -> None:
|
||||
self._svc = slot_service
|
||||
self._repo = repo
|
||||
self._broker = broker
|
||||
self._ledger_outbox = ledger_outbox
|
||||
|
||||
# ------------------------------------------------------------
|
||||
def run_once(self) -> ExecutionResult:
|
||||
@@ -83,7 +100,7 @@ class ExecutionService:
|
||||
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:
|
||||
self._on_filled(slot, fs)
|
||||
self._on_filled(slot, fs, side, r)
|
||||
r.filled += 1
|
||||
elif fs.is_terminal_reject:
|
||||
self._rollback(slot, reason=f"order_{fs.state}")
|
||||
@@ -91,7 +108,8 @@ class ExecutionService:
|
||||
# accepted / partially_filled -> leave for next tick
|
||||
|
||||
# ------------------------------------------------------------
|
||||
def _on_filled(self, slot, fs) -> None:
|
||||
def _on_filled(self, slot, fs, side: str, r: ExecutionResult) -> None:
|
||||
broker_order_id = slot.broker_order_id # captured BEFORE clear_dispatch
|
||||
try:
|
||||
if slot.status is SlotState.PENDING_BUY:
|
||||
self._svc.transit_slot(
|
||||
@@ -106,8 +124,38 @@ class ExecutionService:
|
||||
)
|
||||
except SlotStateMismatchError:
|
||||
return # slot moved concurrently; next tick reconciles
|
||||
# REQ-018: enqueue ledger work item AFTER the slot transition is durable.
|
||||
# Slot transition is the critical path; ledger write is async (REQ-019).
|
||||
self._record_ledger(slot, fs, side, broker_order_id, r)
|
||||
self._repo.clear_dispatch(slot.slot_id) # done with this order
|
||||
|
||||
def _record_ledger(self, slot, fs, side: str, broker_order_id: str, r: ExecutionResult) -> None:
|
||||
"""Append a LedgerWorkItem for the just-filled trade (REQ-018 + REQ-019)."""
|
||||
if self._ledger_outbox is None:
|
||||
return
|
||||
if not slot.portfolio_id:
|
||||
# REQ-018 AC-4: slot transition still proceeded; record skip for ops.
|
||||
r.ledger_skipped += 1
|
||||
return
|
||||
if fs.execution_id is None:
|
||||
# Broker contract regression — log via counter; slot already transitioned.
|
||||
r.ledger_skipped += 1
|
||||
return
|
||||
price = float(fs.avg_price) if fs.avg_price is not None else 0.0
|
||||
trade = LedgerTrade(
|
||||
asset_id=str(slot.portfolio_id),
|
||||
order_id=str(broker_order_id),
|
||||
execution_id=str(fs.execution_id),
|
||||
side=side,
|
||||
qty=float(fs.filled_qty),
|
||||
price=price,
|
||||
fee=float(fs.fee),
|
||||
tax=float(fs.tax),
|
||||
executed_at=datetime.now(tz=timezone.utc),
|
||||
)
|
||||
self._ledger_outbox.enqueue(trade)
|
||||
r.ledger_enqueued += 1
|
||||
|
||||
def _rollback(self, slot, *, reason: str) -> None:
|
||||
try:
|
||||
if slot.status is SlotState.PENDING_BUY:
|
||||
|
||||
@@ -0,0 +1,230 @@
|
||||
"""Ledger outbox + writer (REQ-019, dcy-004 FR-3/FR-4).
|
||||
|
||||
The execution fill path does NOT call portfolio-hub directly. Instead it
|
||||
enqueues a durable `LedgerWorkItem` keyed by `correlation_id`
|
||||
(`{asset_id}:{order_id}:{execution_id}`). A separate `LedgerWriter` drains
|
||||
the outbox into the `LedgerPort` adapter and classifies each attempt as one
|
||||
of:
|
||||
|
||||
completed -- adapter accepted (new ledger entry).
|
||||
duplicate_completed -- the correlation_id was already done in this outbox
|
||||
(local dedup, no adapter call).
|
||||
retryable_failure -- transient adapter failure; left pending for next tick.
|
||||
conflict_blocked -- portfolio-hub 409; operator must resolve, no retry.
|
||||
|
||||
This separation (orbit `LedgerWriter` pattern) keeps slot transitions on the
|
||||
critical path while making ledger writes eventually-consistent and idempotent.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, replace
|
||||
from datetime import datetime, timezone
|
||||
from enum import Enum
|
||||
from typing import Optional, Protocol
|
||||
|
||||
from trex_core.application.ledger_port import (
|
||||
LedgerConflictError,
|
||||
LedgerPort,
|
||||
LedgerRetryableError,
|
||||
LedgerTrade,
|
||||
LedgerWriteResult,
|
||||
)
|
||||
|
||||
|
||||
class OutboxOutcome(str, Enum):
|
||||
COMPLETED = "completed"
|
||||
DUPLICATE_COMPLETED = "duplicate_completed"
|
||||
RETRYABLE_FAILURE = "retryable_failure"
|
||||
CONFLICT_BLOCKED = "conflict_blocked"
|
||||
|
||||
|
||||
class WorkItemStatus(str, Enum):
|
||||
PENDING = "pending" # awaiting next writer tick
|
||||
COMPLETED = "completed" # adapter accepted (terminal success)
|
||||
CONFLICT_BLOCKED = "conflict_blocked" # 409 — operator resolves (terminal failure)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LedgerWorkItem:
|
||||
"""One outbox row.
|
||||
|
||||
Identity is `correlation_id`; the outbox MUST upsert/dedupe on it so the
|
||||
same fill enqueued twice produces a single row.
|
||||
"""
|
||||
|
||||
correlation_id: str
|
||||
trade: LedgerTrade
|
||||
status: WorkItemStatus = WorkItemStatus.PENDING
|
||||
attempts: int = 0
|
||||
last_outcome: Optional[OutboxOutcome] = None
|
||||
last_error: Optional[str] = None
|
||||
created_at: Optional[datetime] = None
|
||||
updated_at: Optional[datetime] = None
|
||||
|
||||
|
||||
class LedgerOutbox(Protocol):
|
||||
"""Durable work-item store backing the ledger writer."""
|
||||
|
||||
def enqueue(self, trade: LedgerTrade) -> LedgerWorkItem:
|
||||
"""Enqueue a trade. Idempotent on correlation_id (returns the
|
||||
existing row unchanged if it is already present)."""
|
||||
...
|
||||
|
||||
def list_pending(self, limit: int = 32) -> list[LedgerWorkItem]:
|
||||
"""Return pending work items in FIFO order, up to `limit`."""
|
||||
...
|
||||
|
||||
def get(self, correlation_id: str) -> Optional[LedgerWorkItem]:
|
||||
...
|
||||
|
||||
def mark_outcome(
|
||||
self,
|
||||
correlation_id: str,
|
||||
outcome: OutboxOutcome,
|
||||
*,
|
||||
error: Optional[str] = None,
|
||||
) -> LedgerWorkItem:
|
||||
"""Record the outcome of a writer attempt and advance status."""
|
||||
...
|
||||
|
||||
|
||||
@dataclass
|
||||
class WriterStats:
|
||||
completed: int = 0
|
||||
duplicates: int = 0
|
||||
retryable: int = 0
|
||||
conflicts: int = 0
|
||||
|
||||
|
||||
class LedgerWriter:
|
||||
"""Drains the outbox into a `LedgerPort`. One process-wide writer.
|
||||
|
||||
Local dedup is by correlation_id: if a work item is already marked
|
||||
`completed`, the writer reports `duplicate_completed` without touching
|
||||
the adapter (REQ-019 AC-4).
|
||||
"""
|
||||
|
||||
def __init__(self, outbox: LedgerOutbox, ledger: LedgerPort) -> None:
|
||||
self._outbox = outbox
|
||||
self._ledger = ledger
|
||||
|
||||
def enqueue_fill(self, trade: LedgerTrade) -> LedgerWorkItem:
|
||||
return self._outbox.enqueue(trade)
|
||||
|
||||
def run_once(self, limit: int = 32) -> WriterStats:
|
||||
stats = WriterStats()
|
||||
for item in self._outbox.list_pending(limit):
|
||||
outcome = self._process(item)
|
||||
if outcome is OutboxOutcome.COMPLETED:
|
||||
stats.completed += 1
|
||||
elif outcome is OutboxOutcome.DUPLICATE_COMPLETED:
|
||||
stats.duplicates += 1
|
||||
elif outcome is OutboxOutcome.RETRYABLE_FAILURE:
|
||||
stats.retryable += 1
|
||||
elif outcome is OutboxOutcome.CONFLICT_BLOCKED:
|
||||
stats.conflicts += 1
|
||||
return stats
|
||||
|
||||
def _process(self, item: LedgerWorkItem) -> OutboxOutcome:
|
||||
# Local dedup: another writer pass (or restart) already completed this.
|
||||
existing = self._outbox.get(item.correlation_id)
|
||||
if existing is not None and existing.status is WorkItemStatus.COMPLETED:
|
||||
self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.DUPLICATE_COMPLETED)
|
||||
return OutboxOutcome.DUPLICATE_COMPLETED
|
||||
try:
|
||||
self._ledger.record_trade(item.trade)
|
||||
except LedgerConflictError as exc:
|
||||
self._outbox.mark_outcome(
|
||||
item.correlation_id, OutboxOutcome.CONFLICT_BLOCKED, error=str(exc)
|
||||
)
|
||||
return OutboxOutcome.CONFLICT_BLOCKED
|
||||
except LedgerRetryableError as exc:
|
||||
self._outbox.mark_outcome(
|
||||
item.correlation_id, OutboxOutcome.RETRYABLE_FAILURE, error=str(exc)
|
||||
)
|
||||
return OutboxOutcome.RETRYABLE_FAILURE
|
||||
self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.COMPLETED)
|
||||
return OutboxOutcome.COMPLETED
|
||||
|
||||
|
||||
# --- in-memory outbox (tests + slice-1 runtime) ------------------------------
|
||||
|
||||
class InMemoryLedgerOutbox:
|
||||
"""Process-local outbox. Thread-safe; idempotent on correlation_id."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
import threading
|
||||
|
||||
self._rows: dict[str, LedgerWorkItem] = {}
|
||||
self._order: list[str] = []
|
||||
self._lock = threading.Lock()
|
||||
|
||||
@staticmethod
|
||||
def _now() -> datetime:
|
||||
return datetime.now(tz=timezone.utc)
|
||||
|
||||
def enqueue(self, trade: LedgerTrade) -> LedgerWorkItem:
|
||||
with self._lock:
|
||||
cid = trade.correlation_id
|
||||
existing = self._rows.get(cid)
|
||||
if existing is not None:
|
||||
return existing
|
||||
row = LedgerWorkItem(
|
||||
correlation_id=cid,
|
||||
trade=trade,
|
||||
status=WorkItemStatus.PENDING,
|
||||
created_at=self._now(),
|
||||
updated_at=self._now(),
|
||||
)
|
||||
self._rows[cid] = row
|
||||
self._order.append(cid)
|
||||
return row
|
||||
|
||||
def list_pending(self, limit: int = 32) -> list[LedgerWorkItem]:
|
||||
with self._lock:
|
||||
out: list[LedgerWorkItem] = []
|
||||
for cid in self._order:
|
||||
row = self._rows[cid]
|
||||
if row.status is WorkItemStatus.PENDING:
|
||||
out.append(row)
|
||||
if len(out) >= limit:
|
||||
break
|
||||
return out
|
||||
|
||||
def get(self, correlation_id: str) -> Optional[LedgerWorkItem]:
|
||||
with self._lock:
|
||||
return self._rows.get(correlation_id)
|
||||
|
||||
def mark_outcome(
|
||||
self,
|
||||
correlation_id: str,
|
||||
outcome: OutboxOutcome,
|
||||
*,
|
||||
error: Optional[str] = None,
|
||||
) -> LedgerWorkItem:
|
||||
with self._lock:
|
||||
row = self._rows[correlation_id]
|
||||
if outcome is OutboxOutcome.COMPLETED:
|
||||
new_status = WorkItemStatus.COMPLETED
|
||||
elif outcome is OutboxOutcome.CONFLICT_BLOCKED:
|
||||
new_status = WorkItemStatus.CONFLICT_BLOCKED
|
||||
elif outcome is OutboxOutcome.DUPLICATE_COMPLETED:
|
||||
# already terminal; keep COMPLETED status
|
||||
new_status = WorkItemStatus.COMPLETED
|
||||
else: # retryable
|
||||
new_status = WorkItemStatus.PENDING
|
||||
updated = replace(
|
||||
row,
|
||||
status=new_status,
|
||||
attempts=row.attempts + (0 if outcome is OutboxOutcome.DUPLICATE_COMPLETED else 1),
|
||||
last_outcome=outcome,
|
||||
last_error=error,
|
||||
updated_at=self._now(),
|
||||
)
|
||||
self._rows[correlation_id] = updated
|
||||
return updated
|
||||
|
||||
# --- introspection ---
|
||||
def all_items(self) -> list[LedgerWorkItem]:
|
||||
with self._lock:
|
||||
return [self._rows[cid] for cid in self._order]
|
||||
@@ -0,0 +1,87 @@
|
||||
"""Ledger port + DTOs (REQ-017, dcy-004).
|
||||
|
||||
Application-layer contract for recording trades to an external ledger
|
||||
(portfolio-hub). Domain stays free of network/HTTP dependencies (NFR-1);
|
||||
concrete adapters live in trex_core.adapters.ledger.
|
||||
|
||||
The port models the trex-orbit `post_ledger_trade` pattern:
|
||||
POST /v1/assets/{asset_id}/ledger/trades (execution_id idempotency, 409 conflict).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from enum import Enum
|
||||
from typing import Optional, Protocol
|
||||
|
||||
|
||||
class LedgerWriteResult(str, Enum):
|
||||
"""Outcome reported by a `LedgerPort.record_trade` call.
|
||||
|
||||
NEW -- adapter recorded a fresh ledger entry for this execution_id.
|
||||
DUPLICATE_OK -- adapter accepted but recognized execution_id as already
|
||||
recorded (idempotent dedup; no new entry created).
|
||||
|
||||
Errors (LedgerConflictError, LedgerRetryableError) are raised; they are
|
||||
not part of this enum.
|
||||
"""
|
||||
NEW = "new"
|
||||
DUPLICATE_OK = "duplicate_ok"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LedgerTrade:
|
||||
"""One ledger trade record (dcy-004 FR-1 contract)."""
|
||||
|
||||
asset_id: str # == slot.portfolio_id (dcy-004 OQ-1)
|
||||
order_id: str # local order identifier (e.g. broker_order_id)
|
||||
execution_id: str # deterministic, idempotency key (dcy-004 OQ-3)
|
||||
side: str # 'buy' | 'sell'
|
||||
qty: float
|
||||
price: float
|
||||
fee: float = 0.0
|
||||
tax: float = 0.0
|
||||
executed_at: Optional[datetime] = None
|
||||
|
||||
@property
|
||||
def correlation_id(self) -> str:
|
||||
"""Local dedup key — `{asset_id}:{order_id}:{execution_id}` (REQ-019)."""
|
||||
return f"{self.asset_id}:{self.order_id}:{self.execution_id}"
|
||||
|
||||
|
||||
class LedgerError(Exception):
|
||||
"""Base class for ledger adapter failures."""
|
||||
|
||||
|
||||
class LedgerConflictError(LedgerError):
|
||||
"""portfolio-hub returned 409 conflict (duplicate execution_id collision
|
||||
that is not a safe duplicate — operator must resolve, REQ-019 AC-3)."""
|
||||
|
||||
|
||||
class LedgerRetryableError(LedgerError):
|
||||
"""Transient ledger failure (network, 5xx, timeout). The outbox retry
|
||||
loop should re-attempt this work item on the next tick (REQ-019 AC-2)."""
|
||||
|
||||
|
||||
class LedgerPort(Protocol):
|
||||
"""Ledger adapter boundary (REQ-017 AC-1/2).
|
||||
|
||||
Implementations must be idempotent on `execution_id`: calling
|
||||
`record_trade` twice with the same trade should not produce two ledger
|
||||
entries. portfolio-hub does this server-side via 409; the in-memory
|
||||
stub mimics the same contract for deterministic tests.
|
||||
"""
|
||||
|
||||
def record_trade(self, trade: LedgerTrade) -> LedgerWriteResult:
|
||||
"""Append a trade to the ledger.
|
||||
|
||||
Returns:
|
||||
LedgerWriteResult.NEW on a fresh entry,
|
||||
LedgerWriteResult.DUPLICATE_OK when the adapter recognized the
|
||||
execution_id as already recorded.
|
||||
|
||||
Raises:
|
||||
LedgerConflictError: ledger reports an unrecoverable conflict (409).
|
||||
LedgerRetryableError: transient failure; outbox should retry.
|
||||
"""
|
||||
...
|
||||
@@ -31,6 +31,15 @@ class KISConfig:
|
||||
redis_url: str = "" # REQ-016: optional token cache backend
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PortfolioHubConfig:
|
||||
"""portfolio-hub ledger adapter config (REQ-017)."""
|
||||
|
||||
base_url: str
|
||||
token: str = ""
|
||||
timeout: float = 5.0
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DaemonConfig:
|
||||
dsn: str
|
||||
@@ -40,6 +49,10 @@ class DaemonConfig:
|
||||
broker: str # 'paper' | 'kis'
|
||||
execution_interval_sec: float
|
||||
kis: Optional[KISConfig] = None
|
||||
# REQ-017/019: ledger writer config (None => in-memory stub ledger,
|
||||
# appropriate for slice-1 paper-broker integration tests).
|
||||
portfolio_hub: Optional[PortfolioHubConfig] = None
|
||||
ledger_interval_sec: float = 1.0
|
||||
|
||||
@staticmethod
|
||||
def from_env() -> "DaemonConfig":
|
||||
@@ -61,6 +74,14 @@ class DaemonConfig:
|
||||
allow_live=os.environ.get("TREX_KIS_ALLOW_LIVE", "").lower() in ("1", "true", "yes"),
|
||||
redis_url=os.environ.get("TREX_KIS_REDIS_URL", ""),
|
||||
)
|
||||
portfolio_hub: Optional[PortfolioHubConfig] = None
|
||||
pf_base = os.environ.get("TREX_PORTFOLIO_HUB_BASE_URL", "").strip()
|
||||
if pf_base:
|
||||
portfolio_hub = PortfolioHubConfig(
|
||||
base_url=pf_base,
|
||||
token=os.environ.get("TREX_PORTFOLIO_HUB_TOKEN", "").strip(),
|
||||
timeout=float(os.environ.get("TREX_PORTFOLIO_HUB_TIMEOUT", "5")),
|
||||
)
|
||||
return DaemonConfig(
|
||||
dsn=dsn,
|
||||
total_slots=int(os.environ.get("TREX_TOTAL_SLOTS", "10")),
|
||||
@@ -69,4 +90,6 @@ class DaemonConfig:
|
||||
broker=broker,
|
||||
execution_interval_sec=float(os.environ.get("TREX_EXECUTION_INTERVAL_SEC", "1.0")),
|
||||
kis=kis,
|
||||
portfolio_hub=portfolio_hub,
|
||||
ledger_interval_sec=float(os.environ.get("TREX_LEDGER_INTERVAL_SEC", "1.0")),
|
||||
)
|
||||
|
||||
@@ -16,14 +16,18 @@ from fastapi import FastAPI
|
||||
|
||||
from trex_core.adapters.api.app import create_app
|
||||
from trex_core.adapters.broker.paper import PaperBroker
|
||||
from trex_core.adapters.ledger import InMemoryLedger, PortfolioHubLedger
|
||||
from trex_core.adapters.persistence.migrate import apply_migrations
|
||||
from trex_core.adapters.persistence.postgresql import (
|
||||
PostgreSQLIntentStore,
|
||||
PostgreSQLLedgerOutbox,
|
||||
PostgreSQLSlotRepository,
|
||||
)
|
||||
from trex_core.application.broker_port import BrokerPort
|
||||
from trex_core.application.execution import ExecutionService
|
||||
from trex_core.application.intents import IntentService
|
||||
from trex_core.application.ledger_outbox import LedgerWriter
|
||||
from trex_core.application.ledger_port import LedgerPort
|
||||
from trex_core.domain.slot import SlotService
|
||||
from trex_core.runtime.config import DaemonConfig
|
||||
|
||||
@@ -46,13 +50,23 @@ def _build_broker(config: DaemonConfig) -> BrokerPort:
|
||||
return PaperBroker()
|
||||
|
||||
|
||||
def _build_ledger(config: DaemonConfig) -> LedgerPort:
|
||||
"""portfolio-hub when configured; otherwise in-memory stub (REQ-017)."""
|
||||
pf = config.portfolio_hub
|
||||
if pf is None:
|
||||
return InMemoryLedger()
|
||||
return PortfolioHubLedger(pf.base_url, token=pf.token or None, timeout=pf.timeout)
|
||||
|
||||
|
||||
def build_app(config: DaemonConfig) -> FastAPI:
|
||||
slot_repo = PostgreSQLSlotRepository(config.dsn)
|
||||
intent_store = PostgreSQLIntentStore(config.dsn)
|
||||
slot_service = SlotService(slot_repo)
|
||||
intent_service = IntentService(slot_service, slot_repo, intent_store)
|
||||
broker = _build_broker(config)
|
||||
execution = ExecutionService(slot_service, slot_repo, broker)
|
||||
outbox = PostgreSQLLedgerOutbox(config.dsn)
|
||||
execution = ExecutionService(slot_service, slot_repo, broker, ledger_outbox=outbox)
|
||||
ledger_writer = LedgerWriter(outbox, _build_ledger(config))
|
||||
|
||||
base = create_app(
|
||||
slot_service=slot_service,
|
||||
@@ -63,13 +77,20 @@ def build_app(config: DaemonConfig) -> FastAPI:
|
||||
|
||||
stop = threading.Event()
|
||||
|
||||
def _loop() -> None:
|
||||
def _execution_loop() -> None:
|
||||
while not stop.wait(config.execution_interval_sec):
|
||||
try:
|
||||
execution.run_once()
|
||||
except Exception: # noqa: BLE001 - loop must survive transient errors
|
||||
continue
|
||||
|
||||
def _ledger_loop() -> None:
|
||||
while not stop.wait(config.ledger_interval_sec):
|
||||
try:
|
||||
ledger_writer.run_once()
|
||||
except Exception: # noqa: BLE001 - loop must survive transient errors
|
||||
continue
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(_: FastAPI):
|
||||
apply_migrations(config.dsn)
|
||||
@@ -77,13 +98,16 @@ def build_app(config: DaemonConfig) -> FastAPI:
|
||||
datetime.now(tz=timezone.utc).date(), config.total_slots
|
||||
)
|
||||
execution.reconcile() # restart recovery before serving (REQ-014)
|
||||
worker = threading.Thread(target=_loop, name="trex-execution", daemon=True)
|
||||
worker.start()
|
||||
exec_worker = threading.Thread(target=_execution_loop, name="trex-execution", daemon=True)
|
||||
ledger_worker = threading.Thread(target=_ledger_loop, name="trex-ledger", daemon=True)
|
||||
exec_worker.start()
|
||||
ledger_worker.start()
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
stop.set()
|
||||
worker.join(timeout=config.execution_interval_sec * 3)
|
||||
exec_worker.join(timeout=config.execution_interval_sec * 3)
|
||||
ledger_worker.join(timeout=config.ledger_interval_sec * 3)
|
||||
|
||||
base.router.lifespan_context = lifespan
|
||||
return base
|
||||
|
||||
Reference in New Issue
Block a user