From 32b601cb703c7cf51195988f51e2ee23216757f0 Mon Sep 17 00:00:00 2001
From: Stefan Jansen <stefan@ml4trading.io>
Date: Sat, 17 Oct 2026 10:24:37 -0400
Subject: [PATCH 1/2] fix: stop Alpaca stream resources after provider faults

---
 src/ml4t/live/brokers/alpaca.py  |  3 --
 tests/unit/test_alpaca_broker.py | 60 ++++++++++++++++++++++++++++++++
 2 files changed, 60 insertions(+), 3 deletions(-)

diff --git a/src/ml4t/live/brokers/alpaca.py b/src/ml4t/live/brokers/alpaca.py
index b06c1cc..6266b25 100644
--- a/src/ml4t/live/brokers/alpaca.py
+++ b/src/ml4t/live/brokers/alpaca.py
@@ -201,9 +201,6 @@ async def connect(self) -> None:
 
     async def disconnect(self) -> None:
         """Disconnect from Alpaca."""
-        if not self._connected:
-            return
-
         stream_task = self._stream_task
         if self._trading_stream:
             try:
diff --git a/tests/unit/test_alpaca_broker.py b/tests/unit/test_alpaca_broker.py
index 95822a1..ee58c4b 100644
--- a/tests/unit/test_alpaca_broker.py
+++ b/tests/unit/test_alpaca_broker.py
@@ -1,5 +1,7 @@
 """Unit tests for AlpacaBroker connection and setup."""
 
+import asyncio
+import threading
 import time
 from datetime import UTC, datetime
 from types import SimpleNamespace
@@ -10,6 +12,8 @@
 from alpaca.common.enums import BaseURL
 from alpaca.trading.enums import OrderStatus as AlpacaOrderStatus
 from alpaca.trading.enums import TimeInForce
+from alpaca.trading.models import Order as AlpacaOrder
+from alpaca.trading.models import TradeUpdate as AlpacaTradeUpdate
 from alpaca.trading.requests import MarketOrderRequest
 from ml4t.backtest.types import Order, OrderSide, OrderStatus, OrderType, Position
 
@@ -1520,3 +1524,59 @@ async def test_terminal_trade_updates_remove_all_tracking(event):
 
     assert broker.pending_orders == []
     assert broker._alpaca_order_map == {}
+
+
+@patch("ml4t.live.brokers.alpaca.TradingClient")
+@patch("ml4t.live.brokers.alpaca.TradingStream")
+async def test_disconnect_stops_the_stream_after_a_provider_order_fault(
+    mock_stream_class, mock_client_class
+):
+    client = mock_client_class.return_value
+    client.get_account.return_value = MockAlpacaAccount()
+    client.get_all_positions.return_value = []
+    client.get_orders.return_value = []
+    now = datetime.now(UTC)
+    venue_order = AlpacaOrder.model_validate(
+        {
+            **vars(MockAlpacaOrder(id="00000000-0000-0000-0000-000000000001")),
+            "client_order_id": "disconnect-regression",
+            "updated_at": now,
+            "submitted_at": now,
+            "order_class": "simple",
+            "extended_hours": False,
+        }
+    )
+    client.submit_order.return_value = venue_order
+    started = threading.Event()
+    stopped = threading.Event()
+    stream = mock_stream_class.return_value
+
+    def run_stream():
+        started.set()
+        stopped.wait(5)
+
+    stream.run.side_effect = run_stream
+    stream.stop.side_effect = stopped.set
+    broker = AlpacaBroker(api_key="PKTEST", secret_key="TEST")
+    await broker.connect()
+    try:
+        assert await asyncio.to_thread(started.wait, 2)
+        await broker.submit_order_async("AAPL", 100, OrderSide.BUY)
+        callback = stream.subscribe_trade_updates.call_args.args[0]
+        invalid_fill = AlpacaOrder.model_validate(
+            venue_order.model_dump()
+            | {"status": "filled", "filled_qty": "101", "filled_avg_price": "10"}
+        )
+        await callback(AlpacaTradeUpdate(event="fill", order=invalid_fill, timestamp=now))
+        assert broker.is_connected is False
+
+        await broker.disconnect()
+
+        stream.stop.assert_called_once()
+        assert stopped.is_set()
+        await broker.disconnect()
+        stream.stop.assert_called_once()
+    finally:
+        stopped.set()
+        if broker._stream_task is not None:
+            await asyncio.wait_for(broker._stream_task, timeout=2)

From 078b7763002755ed999d5b9cef9c8f82a2eab7b8 Mon Sep 17 00:00:00 2001
From: Stefan Jansen <stefan@ml4trading.io>
Date: Sat, 17 Oct 2026 10:25:05 -0400
Subject: [PATCH 2/2] test: require faulted stream task completion before
 disconnect returns

---
 tests/unit/test_alpaca_broker.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/tests/unit/test_alpaca_broker.py b/tests/unit/test_alpaca_broker.py
index ee58c4b..54a2f0c 100644
--- a/tests/unit/test_alpaca_broker.py
+++ b/tests/unit/test_alpaca_broker.py
@@ -1570,10 +1570,12 @@ def run_stream():
         await callback(AlpacaTradeUpdate(event="fill", order=invalid_fill, timestamp=now))
         assert broker.is_connected is False
 
+        stream_task = broker._stream_task
         await broker.disconnect()
 
         stream.stop.assert_called_once()
         assert stopped.is_set()
+        assert stream_task.done()
         await broker.disconnect()
         stream.stop.assert_called_once()
     finally: