From 13837acc6048fd3d1b96160deb70023760148a3d Mon Sep 17 00:00:00 2001 From: Justin Ko Date: Thu, 4 Jun 2026 22:57:31 +0900 Subject: [PATCH] Implement BAT-006: portfolio-hub ledger integration (REQ-017~019) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- .env.example | 9 + tests/adapters/ledger/__init__.py | 0 .../adapters/ledger/test_in_memory_ledger.py | 58 +++++ .../ledger/test_portfolio_hub_ledger.py | 71 ++++++ tests/application/test_execution_ledger.py | 179 ++++++++++++++ tests/application/test_ledger_outbox.py | 65 +++++ tests/domain/test_domain_import_sweep.py | 46 ++++ trex_core/adapters/broker/kis.py | 22 +- trex_core/adapters/broker/paper.py | 16 +- trex_core/adapters/ledger/__init__.py | 9 + trex_core/adapters/ledger/in_memory.py | 65 +++++ trex_core/adapters/ledger/portfolio_hub.py | 91 +++++++ trex_core/adapters/persistence/postgresql.py | 118 +++++++++ .../sql/002_create_ledger_outbox.sql | 33 +++ trex_core/application/broker_port.py | 25 ++ trex_core/application/execution.py | 58 ++++- trex_core/application/ledger_outbox.py | 230 ++++++++++++++++++ trex_core/application/ledger_port.py | 87 +++++++ trex_core/runtime/config.py | 23 ++ trex_core/runtime/daemon.py | 34 ++- 20 files changed, 1224 insertions(+), 15 deletions(-) create mode 100644 tests/adapters/ledger/__init__.py create mode 100644 tests/adapters/ledger/test_in_memory_ledger.py create mode 100644 tests/adapters/ledger/test_portfolio_hub_ledger.py create mode 100644 tests/application/test_execution_ledger.py create mode 100644 tests/application/test_ledger_outbox.py create mode 100644 tests/domain/test_domain_import_sweep.py create mode 100644 trex_core/adapters/ledger/__init__.py create mode 100644 trex_core/adapters/ledger/in_memory.py create mode 100644 trex_core/adapters/ledger/portfolio_hub.py create mode 100644 trex_core/adapters/persistence/sql/002_create_ledger_outbox.sql create mode 100644 trex_core/application/ledger_outbox.py create mode 100644 trex_core/application/ledger_port.py diff --git a/.env.example b/.env.example index 058f96d..3b924f6 100644 --- a/.env.example +++ b/.env.example @@ -29,3 +29,12 @@ TREX_KIS_ALLOW_LIVE=false # Optional: Redis URL for KIS token cache (restart / multi-process survival, REQ-016). # Unset => in-memory token cache only. TREX_KIS_REDIS_URL= + +# --- portfolio-hub ledger (REQ-017/018/019, BAT-006) --- +# When TREX_PORTFOLIO_HUB_BASE_URL is set, fills are recorded via the HTTP +# adapter; otherwise the daemon uses an in-memory ledger (no external writes). +TREX_PORTFOLIO_HUB_BASE_URL= +TREX_PORTFOLIO_HUB_TOKEN= +TREX_PORTFOLIO_HUB_TIMEOUT=5 +# Ledger outbox writer tick interval (seconds). +TREX_LEDGER_INTERVAL_SEC=1.0 diff --git a/tests/adapters/ledger/__init__.py b/tests/adapters/ledger/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/adapters/ledger/test_in_memory_ledger.py b/tests/adapters/ledger/test_in_memory_ledger.py new file mode 100644 index 0000000..f8e3384 --- /dev/null +++ b/tests/adapters/ledger/test_in_memory_ledger.py @@ -0,0 +1,58 @@ +"""In-memory ledger stub tests (REQ-017 AC-1).""" +from __future__ import annotations + +from datetime import datetime, timezone + +import pytest + +from trex_core.adapters.ledger.in_memory import InMemoryLedger +from trex_core.application.ledger_port import ( + LedgerConflictError, + LedgerRetryableError, + LedgerTrade, +) + + +def _trade(execution_id="paper:PAPER-000001:10") -> LedgerTrade: + return LedgerTrade( + asset_id="PF-1", order_id="PAPER-000001", execution_id=execution_id, + side="buy", qty=10, price=1000.0, fee=1.5, tax=0.0, + executed_at=datetime(2026, 6, 4, 9, 0, tzinfo=timezone.utc), + ) + + +def test_record_trade_preserves_fields_and_is_idempotent_on_execution_id(): + ledger = InMemoryLedger() + t = _trade() + ledger.record_trade(t) + ledger.record_trade(t) # same execution_id — no duplicate entry + assert ledger.count() == 1 + stored = ledger.trades()[0] + assert (stored.asset_id, stored.side, stored.qty, stored.price, + stored.fee, stored.tax, stored.executed_at) == ( + "PF-1", "buy", 10, 1000.0, 1.5, 0.0, + datetime(2026, 6, 4, 9, 0, tzinfo=timezone.utc), + ) + + +def test_retryable_failure_then_success(): + ledger = InMemoryLedger(fail_n_then_succeed=2) + t = _trade() + with pytest.raises(LedgerRetryableError): + ledger.record_trade(t) + with pytest.raises(LedgerRetryableError): + ledger.record_trade(t) + ledger.record_trade(t) # third attempt succeeds + assert ledger.count() == 1 + + +def test_conflict_surfaces_as_ledger_conflict_error(): + ledger = InMemoryLedger(conflict_executions={"paper:PAPER-000001:10"}) + with pytest.raises(LedgerConflictError): + ledger.record_trade(_trade()) + assert ledger.count() == 0 + + +def test_correlation_id_format(): + t = _trade() + assert t.correlation_id == "PF-1:PAPER-000001:paper:PAPER-000001:10" diff --git a/tests/adapters/ledger/test_portfolio_hub_ledger.py b/tests/adapters/ledger/test_portfolio_hub_ledger.py new file mode 100644 index 0000000..d97c1c8 --- /dev/null +++ b/tests/adapters/ledger/test_portfolio_hub_ledger.py @@ -0,0 +1,71 @@ +"""portfolio-hub HTTP ledger adapter tests (REQ-017 AC-2/3).""" +from __future__ import annotations + +from datetime import datetime, timezone + +import httpx +import pytest + +from trex_core.adapters.ledger.portfolio_hub import PortfolioHubLedger +from trex_core.application.ledger_port import ( + LedgerConflictError, + LedgerRetryableError, + LedgerTrade, +) + + +def _trade(asset="PF-1", execution_id="paper:PAPER-000001:10") -> LedgerTrade: + return LedgerTrade( + asset_id=asset, order_id="PAPER-000001", execution_id=execution_id, + side="buy", qty=10, price=1000.0, fee=1.5, tax=0.0, + executed_at=datetime(2026, 6, 4, 9, 0, tzinfo=timezone.utc), + ) + + +def _client(handler): + return httpx.Client(transport=httpx.MockTransport(handler)) + + +def test_post_contract_includes_execution_id_and_path_uses_asset_id(): + captured = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured["url"] = str(request.url) + captured["body"] = request.read().decode() + captured["auth"] = request.headers.get("authorization") + return httpx.Response(201, json={"ok": True}) + + adapter = PortfolioHubLedger("https://portfolio-hub.test", token="tok", + client=_client(handler)) + adapter.record_trade(_trade(asset="PF-77")) + assert captured["url"] == "https://portfolio-hub.test/v1/assets/PF-77/ledger/trades" + assert "execution_id" in captured["body"] + assert "\"side\": \"buy\"" in captured["body"] + assert captured["auth"] == "Bearer tok" + + +def test_409_surfaces_as_ledger_conflict_error(): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(409, text='{"error":"duplicate execution"}') + + adapter = PortfolioHubLedger("https://portfolio-hub.test", client=_client(handler)) + with pytest.raises(LedgerConflictError): + adapter.record_trade(_trade()) + + +def test_5xx_is_retryable(): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(503, text="upstream down") + + adapter = PortfolioHubLedger("https://portfolio-hub.test", client=_client(handler)) + with pytest.raises(LedgerRetryableError): + adapter.record_trade(_trade()) + + +def test_network_error_is_retryable(): + def handler(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("boom") + + adapter = PortfolioHubLedger("https://portfolio-hub.test", client=_client(handler)) + with pytest.raises(LedgerRetryableError): + adapter.record_trade(_trade()) diff --git a/tests/application/test_execution_ledger.py b/tests/application/test_execution_ledger.py new file mode 100644 index 0000000..bd2f803 --- /dev/null +++ b/tests/application/test_execution_ledger.py @@ -0,0 +1,179 @@ +"""Execution loop ledger-hook tests (REQ-018, REQ-019, dcy-004). + +Exercises the full buy/sell lifecycle with the in-memory paper broker and +in-memory outbox + ledger writer, ensuring the slot transition is on the +critical path while the ledger write is eventually-consistent. +""" +from __future__ import annotations + +from datetime import date + +from trex_core.adapters.broker.paper import PaperBroker +from trex_core.adapters.ledger.in_memory import InMemoryLedger +from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore +from trex_core.application.execution import ExecutionService +from trex_core.application.intents import BuyIntent, IntentService, SellIntent +from trex_core.application.ledger_outbox import ( + InMemoryLedgerOutbox, + LedgerWriter, + OutboxOutcome, + WorkItemStatus, +) +from trex_core.domain.slot import SlotService, SlotState +from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository + +DAY = date(2026, 6, 4) + + +def _wire(broker, *, ledger=None): + repo = InMemorySlotRepository() + svc = SlotService(repo) + intents = IntentService(svc, repo, InMemoryIntentStore()) + outbox = InMemoryLedgerOutbox() + execution = ExecutionService(svc, repo, broker, ledger_outbox=outbox) + writer = LedgerWriter(outbox, ledger or InMemoryLedger()) + svc.preallocate_daily_slots(DAY, 5) + return repo, svc, intents, execution, outbox, writer + + +def _buy(intents, *, iid="b1", portfolio_id="PF-1"): + return intents.submit_buy(BuyIntent( + intent_id=iid, trading_date=DAY, symbol="005930", qty=10, entry_price=1000, + portfolio_id=portfolio_id, + )) + + +def test_buy_fill_enqueues_one_ledger_item_and_writer_records_it(): + ledger = InMemoryLedger() + repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0), + ledger=ledger) + sid = _buy(intents).slot_id + r = execution.run_once() + assert r.filled == 1 and r.ledger_enqueued == 1 + assert repo.fetch_slot(sid).status is SlotState.HOLDING + stats = writer.run_once() + assert stats.completed == 1 + assert ledger.count() == 1 + t = ledger.trades()[0] + assert (t.asset_id, t.side, t.qty, t.price) == ("PF-1", "buy", 10, 1000.0) + assert t.execution_id.startswith("paper:PAPER-") + assert t.fee > 0.0 and t.tax == 0.0 + + +def test_sell_fill_enqueues_sell_ledger_item(): + ledger = InMemoryLedger() + repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0), + ledger=ledger) + sid = _buy(intents).slot_id + execution.run_once() # -> holding + buy ledger item + writer.run_once() # drain buy + intents.submit_sell(SellIntent(intent_id="s1", slot_id=sid)) + r = execution.run_once() # dispatch sell + fill -> closed + sell ledger item + assert r.ledger_enqueued == 1 + assert repo.fetch_slot(sid).status is SlotState.CLOSED + writer.run_once() + sides = [t.side for t in ledger.trades()] + assert sides == ["buy", "sell"] + + +def test_portfolio_id_absent_skips_ledger_but_transitions_slot(): + ledger = InMemoryLedger() + repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0), + ledger=ledger) + sid = _buy(intents, iid="no-pf", portfolio_id=None).slot_id + r = execution.run_once() + assert r.filled == 1 and r.ledger_enqueued == 0 and r.ledger_skipped == 1 + assert repo.fetch_slot(sid).status is SlotState.HOLDING + assert outbox.all_items() == [] + assert ledger.count() == 0 + + +def test_duplicate_execution_id_results_in_zero_duplicate_in_ledger(): + """REQ-019 AC-1: same execution_id enqueued twice = 1 ledger entry.""" + ledger = InMemoryLedger() + repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0), + ledger=ledger) + _buy(intents).slot_id + execution.run_once() + writer.run_once() + assert ledger.count() == 1 + # Re-enqueue the same trade explicitly (simulate accidental re-fill). + first = outbox.all_items()[0] + outbox.enqueue(first.trade) # idempotent on correlation_id + stats = writer.run_once() # still terminal-completed; reports duplicate + assert stats.duplicates >= 1 + assert ledger.count() == 1 # no duplicate in the ledger + + +def test_retryable_failure_then_eventually_completes(): + """REQ-019 AC-2: retry until completed; slot transition unaffected.""" + ledger = InMemoryLedger(fail_n_then_succeed=2) + repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0), + ledger=ledger) + sid = _buy(intents).slot_id + execution.run_once() + assert repo.fetch_slot(sid).status is SlotState.HOLDING # slot already transitioned + stats1 = writer.run_once() + assert stats1.retryable == 1 and stats1.completed == 0 and ledger.count() == 0 + stats2 = writer.run_once() + assert stats2.retryable == 1 and ledger.count() == 0 + stats3 = writer.run_once() + assert stats3.completed == 1 and ledger.count() == 1 + # outbox row terminal-completed + [item] = outbox.all_items() + assert item.status is WorkItemStatus.COMPLETED + assert item.last_outcome is OutboxOutcome.COMPLETED + + +def test_conflict_blocks_without_auto_retry(): + """REQ-019 AC-3: 409 conflict halts auto-retry.""" + # The execution_id paper synthesizes is paper:{oid}:{qty}; first dispatched + # order id is PAPER-000001 / qty=10 => execution_id = paper:PAPER-000001:10. + conflict_id = "paper:PAPER-000001:10" + ledger = InMemoryLedger(conflict_executions={conflict_id}) + repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0), + ledger=ledger) + _buy(intents) + execution.run_once() + stats = writer.run_once() + assert stats.conflicts == 1 + [item] = outbox.all_items() + assert item.status is WorkItemStatus.CONFLICT_BLOCKED + # second writer tick must NOT call the adapter again + stats2 = writer.run_once() + assert stats2.completed == 0 and stats2.retryable == 0 + # ledger is still empty (conflict blocked; operator resolves) + assert ledger.count() == 0 + + +def test_duplicate_completed_when_reprocessing_terminal_item(): + """REQ-019 AC-4: a completed item reports duplicate_completed if re-listed.""" + ledger = InMemoryLedger() + repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0), + ledger=ledger) + _buy(intents) + execution.run_once() + writer.run_once() + # Force a duplicate enqueue of the same trade — outbox keeps single row, + # ledger keeps single entry. + [item] = outbox.all_items() + outbox.enqueue(item.trade) + assert len(outbox.all_items()) == 1 + # Writer sees no PENDING rows now; stats are all zero (terminal). + stats = writer.run_once() + assert stats.completed == 0 and stats.duplicates == 0 and stats.retryable == 0 + assert ledger.count() == 1 + + +def test_slot_transition_proceeds_when_ledger_is_down(): + """REQ-019 AC-5: portfolio-hub outage does not block the fill transition.""" + ledger = InMemoryLedger(fail_n_then_succeed=100) + repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0), + ledger=ledger) + sid = _buy(intents).slot_id + execution.run_once() + assert repo.fetch_slot(sid).status is SlotState.HOLDING + writer.run_once() # all retries fail + assert ledger.count() == 0 + [item] = outbox.all_items() + assert item.status is WorkItemStatus.PENDING # still waiting for retry diff --git a/tests/application/test_ledger_outbox.py b/tests/application/test_ledger_outbox.py new file mode 100644 index 0000000..3386136 --- /dev/null +++ b/tests/application/test_ledger_outbox.py @@ -0,0 +1,65 @@ +"""LedgerOutbox + LedgerWriter direct unit tests (REQ-019).""" +from __future__ import annotations + +from datetime import datetime, timezone + +from trex_core.adapters.ledger.in_memory import InMemoryLedger +from trex_core.application.ledger_outbox import ( + InMemoryLedgerOutbox, + LedgerWriter, + OutboxOutcome, + WorkItemStatus, +) +from trex_core.application.ledger_port import LedgerTrade + + +def _trade(execution_id="paper:PAPER-1:10") -> LedgerTrade: + return LedgerTrade( + asset_id="PF-1", order_id="PAPER-1", execution_id=execution_id, + side="buy", qty=10, price=1000.0, + executed_at=datetime(2026, 6, 4, tzinfo=timezone.utc), + ) + + +def test_enqueue_is_idempotent_on_correlation_id(): + outbox = InMemoryLedgerOutbox() + a = outbox.enqueue(_trade()) + b = outbox.enqueue(_trade()) + assert a.correlation_id == b.correlation_id + assert len(outbox.all_items()) == 1 + + +def test_writer_run_once_drains_pending_and_marks_completed(): + outbox = InMemoryLedgerOutbox() + ledger = InMemoryLedger() + writer = LedgerWriter(outbox, ledger) + outbox.enqueue(_trade(execution_id="e-1")) + outbox.enqueue(_trade(execution_id="e-2")) + stats = writer.run_once() + assert stats.completed == 2 and ledger.count() == 2 + assert all(it.status is WorkItemStatus.COMPLETED for it in outbox.all_items()) + assert all(it.last_outcome is OutboxOutcome.COMPLETED for it in outbox.all_items()) + + +def test_writer_marks_conflict_blocked_and_does_not_retry(): + outbox = InMemoryLedgerOutbox() + ledger = InMemoryLedger(conflict_executions={"e-x"}) + writer = LedgerWriter(outbox, ledger) + outbox.enqueue(_trade(execution_id="e-x")) + writer.run_once() + [item] = outbox.all_items() + assert item.status is WorkItemStatus.CONFLICT_BLOCKED + assert item.last_outcome is OutboxOutcome.CONFLICT_BLOCKED + # No pending items remain, so a subsequent run is a no-op. + assert outbox.list_pending() == [] + + +def test_writer_retries_until_success(): + outbox = InMemoryLedgerOutbox() + ledger = InMemoryLedger(fail_n_then_succeed=1) + writer = LedgerWriter(outbox, ledger) + outbox.enqueue(_trade(execution_id="e-r")) + s1 = writer.run_once() + assert s1.retryable == 1 and ledger.count() == 0 + s2 = writer.run_once() + assert s2.completed == 1 and ledger.count() == 1 diff --git a/tests/domain/test_domain_import_sweep.py b/tests/domain/test_domain_import_sweep.py new file mode 100644 index 0000000..b41751a --- /dev/null +++ b/tests/domain/test_domain_import_sweep.py @@ -0,0 +1,46 @@ +"""Static import sweep guarding the pure-domain boundary (REQ-017 AC-4, NFR-1). + +`trex_core.domain` must NOT depend on portfolio-hub, HTTP, ledger adapters, +or any other infrastructure module. The sweep walks the source tree of +trex_core/domain and inspects every `import` statement. +""" +from __future__ import annotations + +import ast +from pathlib import Path + +_DOMAIN_ROOT = Path(__file__).resolve().parents[2] / "trex_core" / "domain" + +_FORBIDDEN_PREFIXES = ( + "httpx", + "fastapi", + "uvicorn", + "psycopg", + "redis", + "trex_core.adapters", # any adapter (broker, ledger, persistence, api) + "trex_core.runtime", # daemon / config + "trex_core.application", # ports/services live one layer up +) + + +def _iter_imports(path: Path): + tree = ast.parse(path.read_text(), filename=str(path)) + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + yield alias.name + elif isinstance(node, ast.ImportFrom): + if node.module: + yield node.module + + +def test_domain_has_no_forbidden_imports(): + offenders: list[tuple[str, str]] = [] + for py in _DOMAIN_ROOT.rglob("*.py"): + for imp in _iter_imports(py): + if any(imp == p or imp.startswith(p + ".") for p in _FORBIDDEN_PREFIXES): + offenders.append((str(py.relative_to(_DOMAIN_ROOT.parent.parent)), imp)) + assert offenders == [], ( + "trex_core/domain must remain free of adapter/runtime/application imports; " + f"found: {offenders}" + ) diff --git a/trex_core/adapters/broker/kis.py b/trex_core/adapters/broker/kis.py index 281deb8..07bdf27 100644 --- a/trex_core/adapters/broker/kis.py +++ b/trex_core/adapters/broker/kis.py @@ -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) diff --git a/trex_core/adapters/broker/paper.py b/trex_core/adapters/broker/paper.py index 5cc014d..a4ffa28 100644 --- a/trex_core/adapters/broker/paper.py +++ b/trex_core/adapters/broker/paper.py @@ -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: diff --git a/trex_core/adapters/ledger/__init__.py b/trex_core/adapters/ledger/__init__.py new file mode 100644 index 0000000..152b4e1 --- /dev/null +++ b/trex_core/adapters/ledger/__init__.py @@ -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"] diff --git a/trex_core/adapters/ledger/in_memory.py b/trex_core/adapters/ledger/in_memory.py new file mode 100644 index 0000000..6c0296e --- /dev/null +++ b/trex_core/adapters/ledger/in_memory.py @@ -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 diff --git a/trex_core/adapters/ledger/portfolio_hub.py b/trex_core/adapters/ledger/portfolio_hub.py new file mode 100644 index 0000000..775d2e1 --- /dev/null +++ b/trex_core/adapters/ledger/portfolio_hub.py @@ -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]}" + ) diff --git a/trex_core/adapters/persistence/postgresql.py b/trex_core/adapters/persistence/postgresql.py index d78dee2..27f7f85 100644 --- a/trex_core/adapters/persistence/postgresql.py +++ b/trex_core/adapters/persistence/postgresql.py @@ -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], + ) diff --git a/trex_core/adapters/persistence/sql/002_create_ledger_outbox.sql b/trex_core/adapters/persistence/sql/002_create_ledger_outbox.sql new file mode 100644 index 0000000..2b9b13d --- /dev/null +++ b/trex_core/adapters/persistence/sql/002_create_ledger_outbox.sql @@ -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); diff --git a/trex_core/application/broker_port.py b/trex_core/application/broker_port.py index d726b2a..cd57e91 100644 --- a/trex_core/application/broker_port.py +++ b/trex_core/application/broker_port.py @@ -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) diff --git a/trex_core/application/execution.py b/trex_core/application/execution.py index 31f6d50..b376de0 100644 --- a/trex_core/application/execution.py +++ b/trex_core/application/execution.py @@ -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: diff --git a/trex_core/application/ledger_outbox.py b/trex_core/application/ledger_outbox.py new file mode 100644 index 0000000..2bae31d --- /dev/null +++ b/trex_core/application/ledger_outbox.py @@ -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] diff --git a/trex_core/application/ledger_port.py b/trex_core/application/ledger_port.py new file mode 100644 index 0000000..e971207 --- /dev/null +++ b/trex_core/application/ledger_port.py @@ -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. + """ + ... diff --git a/trex_core/runtime/config.py b/trex_core/runtime/config.py index 787da67..18a088f 100644 --- a/trex_core/runtime/config.py +++ b/trex_core/runtime/config.py @@ -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")), ) diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index b19ee40..3b9bd50 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -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