"""Tests for the trade reconciliation background task (Phase 4). A Kevin bracket entry places three orders at Alpaca: the entry (parent) plus a stop-loss and a take-profit leg. When a leg fills automatically at Alpaca, the close never passes through our executor — so locally there is no closing Trade and no booked P&L. The reconcile task fetches each open entry's order via ``broker.get_order(broker_order_id, nested=True)``, detects a filled SL/TP leg, and books the close (idempotently). It also syncs non-terminal local statuses and logs drift it cannot auto-resolve. """ from __future__ import annotations import asyncio import uuid from datetime import datetime, timezone from unittest.mock import AsyncMock, MagicMock, patch import pytest from services.api_gateway.config import ApiGatewayConfig from services.api_gateway.tasks.trade_reconcile import ( reconcile_once, trade_reconcile_loop, ) from shared.constants.kevin import KEVIN_STRATEGY_UUID from shared.models.trading import Trade, TradeSide, TradeStatus from shared.schemas.trading import ( BrokerOrder, OrderResult, OrderSide, OrderStatus, ) # --------------------------------------------------------------------------- # Builders # --------------------------------------------------------------------------- def _entry_trade( *, ticker: str = "NVDA", qty: float = 10.0, price: float = 100.0, broker_order_id: str = "parent-1", status: TradeStatus = TradeStatus.FILLED, signal_id: uuid.UUID | None = None, ) -> Trade: return Trade( id=uuid.uuid4(), ticker=ticker, side=TradeSide.BUY, qty=qty, price=price, status=status, strategy_id=KEVIN_STRATEGY_UUID, signal_id=signal_id or uuid.uuid4(), broker_order_id=broker_order_id, pnl=None, ) def _leg( *, order_id: str, status: OrderStatus, filled_price: float | None, qty: float = 10.0, ticker: str = "NVDA", ) -> OrderResult: return OrderResult( order_id=order_id, ticker=ticker, side=OrderSide.SELL, qty=qty, filled_price=filled_price, status=status, timestamp=datetime.now(timezone.utc), ) def _bracket( *, parent_id: str = "parent-1", parent_status: OrderStatus = OrderStatus.FILLED, ticker: str = "NVDA", qty: float = 10.0, entry_price: float = 100.0, legs: list[OrderResult] | None = None, ) -> BrokerOrder: return BrokerOrder( order_id=parent_id, ticker=ticker, side=OrderSide.BUY, qty=qty, filled_price=entry_price, status=parent_status, timestamp=datetime.now(timezone.utc), legs=legs or [], ) class _FakeSession: """Async session double driven by intent. ``open_trades`` is what the ``select(Trade).where(...open entries...)`` query returns. ``existing_close_ids`` is the set of broker_order_ids already booked as closing trades (drives the idempotency dedup query, which is modelled as ``scalar_one_or_none``). New rows land in ``added``. """ def __init__( self, open_trades: list[Trade], existing_close_ids: set[str] | None = None, ) -> None: self._open_trades = open_trades self._existing_close_ids = existing_close_ids or set() self.added: list[Trade] = [] self.committed = False async def __aenter__(self) -> "_FakeSession": return self async def __aexit__(self, *exc) -> bool: return False def add(self, obj: Trade) -> None: self.added.append(obj) # A freshly-booked close becomes visible to subsequent dedup lookups # within the same reconcile pass. if obj.broker_order_id: self._existing_close_ids.add(obj.broker_order_id) async def commit(self) -> None: self.committed = True async def execute(self, stmt): # noqa: ANN001 text = str(stmt).lower() # Dedup lookup is the only query with an equality on broker_order_id; # the open-entries scan uses ``broker_order_id IS NOT NULL``. if "broker_order_id =" in text: params = stmt.compile().params target = next( (v for v in params.values() if isinstance(v, str)), None, ) found = ( _entry_trade(broker_order_id=target) if target in self._existing_close_ids else None ) return _Result(scalar=found) # Open-entries scan return _Result(rows=list(self._open_trades)) class _Result: def __init__(self, rows=None, scalar=None) -> None: # noqa: ANN001 self._rows = rows or [] self._scalar = scalar def scalars(self) -> "_Result": return self def all(self) -> list: return self._rows def scalar_one_or_none(self): # noqa: ANN001 return self._scalar def _factory(session: _FakeSession) -> MagicMock: factory = MagicMock() factory.return_value = session return factory def _config(**overrides) -> ApiGatewayConfig: defaults = dict( jwt_secret_key="test-secret-for-reconcile", database_url="sqlite+aiosqlite:///:memory:", redis_url="redis://localhost:6379/0", alpaca_api_key="test-key", alpaca_secret_key="test-secret", paper_trading=True, snapshot_interval_seconds=1, ) defaults.update(overrides) return ApiGatewayConfig(**defaults) # --------------------------------------------------------------------------- # Booking auto-closes # --------------------------------------------------------------------------- class TestReconcileBooksClose: async def test_books_close_on_take_profit_fill(self) -> None: """TP leg FILLED @ 120 vs entry 100, qty 10 -> closing SELL, pnl=200.""" entry = _entry_trade(broker_order_id="parent-1", price=100.0, qty=10.0) tp = _leg(order_id="tp-1", status=OrderStatus.FILLED, filled_price=120.0) sl = _leg(order_id="sl-1", status=OrderStatus.PENDING, filled_price=None) broker = AsyncMock() broker.get_order = AsyncMock( return_value=_bracket(parent_id="parent-1", legs=[tp, sl]) ) session = _FakeSession([entry]) await reconcile_once(broker, _factory(session)) assert len(session.added) == 1 close = session.added[0] assert close.side == TradeSide.SELL assert close.ticker == "NVDA" assert close.qty == 10.0 assert close.price == 120.0 assert close.status == TradeStatus.FILLED assert close.pnl == pytest.approx(200.0) assert close.strategy_id == KEVIN_STRATEGY_UUID assert close.signal_id == entry.signal_id # Dedup key is the filled leg's order id assert close.broker_order_id == "tp-1" broker.get_order.assert_awaited_once_with("parent-1", nested=True) async def test_books_close_on_stop_loss_fill(self) -> None: """SL leg FILLED @ 92 vs entry 100, qty 10 -> closing SELL, pnl=-80.""" entry = _entry_trade(broker_order_id="parent-1", price=100.0, qty=10.0) tp = _leg(order_id="tp-1", status=OrderStatus.PENDING, filled_price=None) sl = _leg(order_id="sl-1", status=OrderStatus.FILLED, filled_price=92.0) broker = AsyncMock() broker.get_order = AsyncMock( return_value=_bracket(parent_id="parent-1", legs=[tp, sl]) ) session = _FakeSession([entry]) await reconcile_once(broker, _factory(session)) assert len(session.added) == 1 close = session.added[0] assert close.price == 92.0 assert close.pnl == pytest.approx(-80.0) assert close.broker_order_id == "sl-1" async def test_noop_when_legs_unfilled(self) -> None: """Both legs still open -> nothing booked.""" entry = _entry_trade(broker_order_id="parent-1") tp = _leg(order_id="tp-1", status=OrderStatus.PENDING, filled_price=None) sl = _leg(order_id="sl-1", status=OrderStatus.PENDING, filled_price=None) broker = AsyncMock() broker.get_order = AsyncMock( return_value=_bracket(parent_id="parent-1", legs=[tp, sl]) ) session = _FakeSession([entry]) await reconcile_once(broker, _factory(session)) assert session.added == [] async def test_booked_close_notifies_slack(self) -> None: """A reconcile-booked close must post to Slack with realized P&L.""" entry = _entry_trade(broker_order_id="parent-1", price=100.0, qty=10.0) tp = _leg(order_id="tp-1", status=OrderStatus.PENDING, filled_price=None) sl = _leg(order_id="sl-1", status=OrderStatus.FILLED, filled_price=92.0) broker = AsyncMock() broker.get_order = AsyncMock( return_value=_bracket(parent_id="parent-1", legs=[tp, sl]) ) session = _FakeSession([entry]) notifier = AsyncMock() await reconcile_once(broker, _factory(session), notifier=notifier) notifier.notify_close.assert_awaited_once() kwargs = notifier.notify_close.await_args.kwargs assert kwargs["ticker"] == "NVDA" assert kwargs["qty"] == 10.0 assert kwargs["price"] == 92.0 assert kwargs["pnl"] == pytest.approx(-80.0) assert kwargs["strategy_id"] == KEVIN_STRATEGY_UUID async def test_no_notification_when_nothing_booked(self) -> None: entry = _entry_trade(broker_order_id="parent-1") tp = _leg(order_id="tp-1", status=OrderStatus.PENDING, filled_price=None) sl = _leg(order_id="sl-1", status=OrderStatus.PENDING, filled_price=None) broker = AsyncMock() broker.get_order = AsyncMock( return_value=_bracket(parent_id="parent-1", legs=[tp, sl]) ) session = _FakeSession([entry]) notifier = AsyncMock() await reconcile_once(broker, _factory(session), notifier=notifier) notifier.notify_close.assert_not_awaited() async def test_notifier_failure_does_not_block_booking(self) -> None: """Slack is an observer — its failure must not lose the Trade row.""" entry = _entry_trade(broker_order_id="parent-1", price=100.0, qty=10.0) sl = _leg(order_id="sl-1", status=OrderStatus.FILLED, filled_price=92.0) broker = AsyncMock() broker.get_order = AsyncMock( return_value=_bracket(parent_id="parent-1", legs=[sl]) ) session = _FakeSession([entry]) notifier = AsyncMock() notifier.notify_close = AsyncMock(side_effect=RuntimeError("slack down")) await reconcile_once(broker, _factory(session), notifier=notifier) assert len(session.added) == 1 assert session.committed # --------------------------------------------------------------------------- # Idempotency # --------------------------------------------------------------------------- class TestReconcileIdempotent: async def test_does_not_double_book_filled_leg(self) -> None: """If a closing trade already carries the filled leg's order id, a second pass books nothing.""" entry = _entry_trade(broker_order_id="parent-1", price=100.0, qty=10.0) tp = _leg(order_id="tp-1", status=OrderStatus.FILLED, filled_price=120.0) broker = AsyncMock() broker.get_order = AsyncMock( return_value=_bracket(parent_id="parent-1", legs=[tp]) ) # tp-1 already booked session = _FakeSession([entry], existing_close_ids={"tp-1"}) await reconcile_once(broker, _factory(session)) assert session.added == [] async def test_running_twice_books_once(self) -> None: """Two back-to-back reconcile passes over the same fixtures book the close exactly once.""" entry = _entry_trade(broker_order_id="parent-1", price=100.0, qty=10.0) tp = _leg(order_id="tp-1", status=OrderStatus.FILLED, filled_price=120.0) broker = AsyncMock() broker.get_order = AsyncMock( return_value=_bracket(parent_id="parent-1", legs=[tp]) ) session = _FakeSession([entry]) await reconcile_once(broker, _factory(session)) await reconcile_once(broker, _factory(session)) assert len(session.added) == 1 # --------------------------------------------------------------------------- # Status sync # --------------------------------------------------------------------------- class TestReconcileStatusSync: async def test_pending_entry_promoted_to_filled(self) -> None: """A local PENDING entry whose Alpaca parent is now FILLED is updated in place (and not booked as a close).""" entry = _entry_trade( broker_order_id="parent-1", status=TradeStatus.PENDING, price=0.0 ) broker = AsyncMock() broker.get_order = AsyncMock( return_value=_bracket( parent_id="parent-1", parent_status=OrderStatus.FILLED, entry_price=101.5, legs=[], ) ) session = _FakeSession([entry]) await reconcile_once(broker, _factory(session)) assert entry.status == TradeStatus.FILLED assert session.added == [] async def test_pending_entry_promoted_to_rejected(self) -> None: entry = _entry_trade( broker_order_id="parent-1", status=TradeStatus.PENDING, price=0.0 ) broker = AsyncMock() broker.get_order = AsyncMock( return_value=_bracket( parent_id="parent-1", parent_status=OrderStatus.REJECTED, legs=[], ) ) session = _FakeSession([entry]) await reconcile_once(broker, _factory(session)) assert entry.status == TradeStatus.REJECTED # --------------------------------------------------------------------------- # Drift handling # --------------------------------------------------------------------------- class TestReconcileDrift: async def test_missing_alpaca_order_logs_warning_no_raise(self, caplog) -> None: # noqa: ANN001 """broker.get_order returns None (order gone) -> warn, do not raise, do not book.""" entry = _entry_trade(broker_order_id="parent-gone") broker = AsyncMock() broker.get_order = AsyncMock(return_value=None) session = _FakeSession([entry]) with caplog.at_level("WARNING"): await reconcile_once(broker, _factory(session)) assert session.added == [] assert any("parent-gone" in r.message or "missing" in r.message.lower() for r in caplog.records) async def test_bad_row_does_not_abort_others(self) -> None: """One trade whose get_order raises must not stop a sibling trade's close from being booked.""" bad = _entry_trade(ticker="BAD", broker_order_id="bad-1") good = _entry_trade(ticker="NVDA", broker_order_id="good-1", price=100.0, qty=10.0) async def get_order(order_id, *, nested=True): # noqa: ANN001 if order_id == "bad-1": raise RuntimeError("alpaca blew up on this one") tp = _leg(order_id="tp-good", status=OrderStatus.FILLED, filled_price=130.0) return _bracket(parent_id="good-1", legs=[tp]) broker = AsyncMock() broker.get_order = AsyncMock(side_effect=get_order) session = _FakeSession([bad, good]) await reconcile_once(broker, _factory(session)) # The good one still books assert len(session.added) == 1 assert session.added[0].ticker == "NVDA" assert session.added[0].broker_order_id == "tp-good" # --------------------------------------------------------------------------- # Loop wiring # --------------------------------------------------------------------------- class TestReconcileLoop: async def test_no_credentials_returns_immediately(self) -> None: cfg = _config(alpaca_api_key="", alpaca_secret_key="") await asyncio.wait_for( trade_reconcile_loop(cfg, MagicMock()), timeout=2.0 ) async def test_error_does_not_crash_loop(self) -> None: cfg = _config() calls = 0 async def boom(broker, sf, notifier=None): # noqa: ANN001 nonlocal calls calls += 1 if calls == 1: raise ConnectionError("down") with ( patch("services.api_gateway.tasks.trade_reconcile.AlpacaBroker") as MB, patch( "services.api_gateway.tasks.trade_reconcile.reconcile_once", side_effect=boom, ), patch( "services.api_gateway.tasks.trade_reconcile.is_market_open", return_value=True, ), ): MB.return_value = AsyncMock() task = asyncio.create_task(trade_reconcile_loop(cfg, MagicMock())) await asyncio.sleep(2.5) task.cancel() try: await task except asyncio.CancelledError: pass assert calls >= 2