fix: bound repricing waits and exit after submit cycle

Author: brndnmtthws

tests/test_portfolio_manager.py

@@ -1,3 +1,4 @@
+import asyncio
 from types import SimpleNamespace
 
 import pytest
@@ -230,6 +231,147 @@ async def fake_run_post_stages(deps, _account_summary, _portfolio_positions):
         ]
         pm.get_portfolio_positions.assert_awaited_once()
 
+    @pytest.mark.asyncio
+    async def test_manage_continues_if_order_submission_wait_times_out(
+        self, mock_ib, mock_config, mocker
+    ):
+        completion_future = mocker.Mock()
+        pm = PortfolioManager(
+            mock_config,
+            mock_ib,
+            completion_future,
+            dry_run=False,
+            run_stage_order=["equity_buy_rebalance"],
+        )
+
+        pm.initialize_account = mocker.Mock()
+        pm.summarize_account = mocker.AsyncMock(return_value=({}, {}))
+        pm.get_portfolio_positions = mocker.AsyncMock(return_value={})
+        pm.orders.print_summary = mocker.Mock()
+        pm.submit_orders = mocker.Mock()
+        pm.adjust_prices = mocker.AsyncMock()
+        pm.trades = mocker.Mock()
+        pm.trades.records = mocker.Mock(return_value=[mocker.Mock()])
+
+        mocker.patch(
+            "thetagang.portfolio_manager.run_equity_rebalance_stages",
+            new=mocker.AsyncMock(),
+        )
+        pm.ibkr.wait_for_submitting_orders = mocker.AsyncMock(
+            side_effect=RuntimeError("timed out")
+        )
+
+        await pm.manage()
+
+        pm.submit_orders.assert_called_once()
+        assert pm.ibkr.wait_for_submitting_orders.await_count == 2
+        pm.adjust_prices.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_manage_allows_incomplete_working_orders(
+        self, mock_ib, mock_config, mocker
+    ):
+        completion_future = mocker.Mock()
+        pm = PortfolioManager(
+            mock_config,
+            mock_ib,
+            completion_future,
+            dry_run=False,
+            run_stage_order=["equity_buy_rebalance"],
+        )
+
+        pm.initialize_account = mocker.Mock()
+        pm.summarize_account = mocker.AsyncMock(return_value=({}, {}))
+        pm.get_portfolio_positions = mocker.AsyncMock(return_value={})
+        pm.orders.print_summary = mocker.Mock()
+        pm.submit_orders = mocker.Mock()
+        pm.adjust_prices = mocker.AsyncMock()
+
+        trade = mocker.Mock()
+        trade.contract = mocker.Mock(symbol="SPY")
+        trade.order = mocker.Mock(orderId=123)
+        trade.orderStatus = mocker.Mock(status="Submitted", filled=0.0, remaining=1.0)
+        trade.isDone.return_value = False
+
+        pm.trades = mocker.Mock()
+        pm.trades.records = mocker.Mock(return_value=[trade])
+
+        mocker.patch(
+            "thetagang.portfolio_manager.run_equity_rebalance_stages",
+            new=mocker.AsyncMock(),
+        )
+        pm.ibkr.wait_for_submitting_orders = mocker.AsyncMock(return_value=None)
+
+        await pm.manage()
+
+    @pytest.mark.asyncio
+    async def test_adjust_prices_continues_if_midpoint_market_data_missing(
+        self, portfolio_manager, mocker
+    ):
+        from thetagang.ibkr import RequiredFieldValidationError
+
+        portfolio_manager.config.runtime.orders.price_update_delay = (1, 2)
+        portfolio_manager.config.runtime.orders.minimum_credit = 0.01
+
+        trade = mocker.Mock()
+        trade.contract = mocker.Mock(symbol="SPY")
+        trade.contract.symbol = "SPY"
+        trade.order = mocker.Mock(lmtPrice=1.23, action="SELL", totalQuantity=1)
+        trade.orderId = 101
+        trade.contract.secType = "OPT"
+        trade.orderStatus = mocker.Mock(status="Submitted", filled=0.0, remaining=1.0)
+        trade.isDone.return_value = False
+
+        portfolio_manager.trades = mocker.Mock()
+        portfolio_manager.trades.records = mocker.Mock(return_value=[trade])
+        portfolio_manager.trades.is_empty = mocker.Mock(return_value=False)
+
+        portfolio_manager.config.portfolio.symbols = {
+            "SPY": mocker.Mock(adjust_price_after_delay=True)
+        }
+        portfolio_manager.ibkr.wait_for_orders_complete = mocker.AsyncMock(
+            return_value=[trade]
+        )
+        portfolio_manager.ibkr.get_ticker_for_contract = mocker.AsyncMock(
+            side_effect=RequiredFieldValidationError("market data unavailable")
+        )
+
+        await portfolio_manager.adjust_prices()
+        portfolio_manager.ibkr.get_ticker_for_contract.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_adjust_prices_continues_when_combo_bag_midpoint_times_out(
+        self, portfolio_manager, mocker
+    ):
+        portfolio_manager.config.runtime.orders.price_update_delay = (1, 2)
+        portfolio_manager.config.runtime.orders.minimum_credit = 0.01
+
+        trade = mocker.Mock()
+        trade.contract = mocker.Mock(symbol="QQQ")
+        trade.contract.symbol = "QQQ"
+        trade.contract.secType = "BAG"
+        trade.order = mocker.Mock(lmtPrice=-1.25, action="BUY", totalQuantity=1)
+        trade.orderStatus = mocker.Mock(status="Submitted", filled=0.0, remaining=1.0)
+        trade.isDone.return_value = False
+
+        portfolio_manager.trades = mocker.Mock()
+        portfolio_manager.trades.records = mocker.Mock(return_value=[trade])
+        portfolio_manager.trades.is_empty = mocker.Mock(return_value=False)
+
+        portfolio_manager.config.portfolio.symbols = {
+            "QQQ": mocker.Mock(adjust_price_after_delay=True)
+        }
+        portfolio_manager.ibkr.wait_for_orders_complete = mocker.AsyncMock(
+            return_value=[trade]
+        )
+        portfolio_manager.ibkr.get_ticker_for_contract = mocker.AsyncMock(
+            side_effect=asyncio.TimeoutError()
+        )
+
+        await portfolio_manager.adjust_prices()
+
+        portfolio_manager.ibkr.get_ticker_for_contract.assert_awaited_once()
+
     @pytest.mark.asyncio
     async def test_write_calls_respects_can_write_when_green_with_nan_close(
         self, portfolio_manager, mocker

thetagang/portfolio_manager.py

@@ -759,46 +759,49 @@ async def manage(self) -> None:
 
                 try:
                     await self.ibkr.wait_for_submitting_orders(self.trades.records())
-                except RuntimeError:
-                    log.error("Submitting orders failed. Continuing anyway..")
-                    pass
+                except RuntimeError as exc:
+                    # DAY orders can remain working at the broker after submission.
+                    # Keep running and let later status checks/logs report open orders.
+                    log.warning(f"Order submission wait timed out: {exc}")
 
                 await self.adjust_prices()
 
-                await self.ibkr.wait_for_submitting_orders(self.trades.records())
-                incomplete_trades = await self.ibkr.wait_for_orders_complete(
-                    self.trades.records()
+                try:
+                    await self.ibkr.wait_for_submitting_orders(self.trades.records())
+                except RuntimeError as exc:
+                    log.warning(f"Post-adjust order submission wait timed out: {exc}")
+                working_statuses = {"PendingSubmit", "PreSubmitted", "Submitted"}
+                incomplete_trades = [
+                    trade
+                    for trade in self.trades.records()
+                    if trade and not trade.isDone()
+                ]
+                still_working = [
+                    trade
+                    for trade in incomplete_trades
+                    if getattr(trade.orderStatus, "status", "") in working_statuses
+                ]
+                unexpected_state = [
+                    trade for trade in incomplete_trades if trade not in still_working
+                ]
+                open_orders = ", ".join(
+                    f"{trade.contract.symbol} (OrderId: {trade.order.orderId}, status={getattr(trade.orderStatus, 'status', 'UNKNOWN')})"
+                    for trade in still_working
                 )
-                if incomplete_trades:
-                    working_statuses = {"PendingSubmit", "PreSubmitted", "Submitted"}
-                    still_working = [
-                        trade
-                        for trade in incomplete_trades
-                        if getattr(trade.orderStatus, "status", "") in working_statuses
-                    ]
-                    unexpected_state = [
-                        trade
-                        for trade in incomplete_trades
-                        if trade not in still_working
-                    ]
-                    open_orders = ", ".join(
+                if open_orders:
+                    log.info(
+                        "Run completed with working submitted orders still open at broker: "
+                        f"{open_orders}"
+                    )
+                if unexpected_state:
+                    unexpected_orders = ", ".join(
                         f"{trade.contract.symbol} (OrderId: {trade.order.orderId}, status={getattr(trade.orderStatus, 'status', 'UNKNOWN')})"
-                        for trade in still_working
+                        for trade in unexpected_state
+                    )
+                    log.warning(
+                        "Run completed with non-working incomplete orders at broker: "
+                        f"{unexpected_orders}"
                     )
-                    if open_orders:
-                        log.info(
-                            "Run completed with working submitted orders still open at broker: "
-                            f"{open_orders}"
-                        )
-                    if unexpected_state:
-                        unexpected_orders = ", ".join(
-                            f"{trade.contract.symbol} (OrderId: {trade.order.orderId}, status={getattr(trade.orderStatus, 'status', 'UNKNOWN')})"
-                            for trade in unexpected_state
-                        )
-                        log.warning(
-                            "Run completed with non-working incomplete orders at broker: "
-                            f"{unexpected_orders}"
-                        )
 
             log.info("ThetaGang is done, shutting down! Cya next time. :sparkles:")
         except:
@@ -1085,10 +1088,14 @@ async def adjust_prices(self) -> None:
 
         for idx, trade in unfilled:
             try:
-                ticker = await self.ibkr.get_ticker_for_contract(
-                    trade.contract,
-                    required_fields=[TickerField.MIDPOINT],
-                    optional_fields=[TickerField.MARKET_PRICE],
+                # Bound midpoint price requests so repricing never blocks run termination.
+                ticker = await asyncio.wait_for(
+                    self.ibkr.get_ticker_for_contract(
+                        trade.contract,
+                        required_fields=[TickerField.MIDPOINT],
+                        optional_fields=[TickerField.MARKET_PRICE],
+                    ),
+                    timeout=self.config.runtime.ib_async.api_response_wait_time,
                 )
 
                 (contract, order) = (trade.contract, trade.order)
@@ -1150,10 +1157,23 @@ async def adjust_prices(self) -> None:
                     self.trades.submit_order(contract, order, idx)
 
                     log.info(f"{contract.symbol}: Order updated, order={order}")
-            except (RuntimeError, RequiredFieldValidationError):
-                log.error(
-                    f"Couldn't generate midpoint price for {trade.contract}, skipping"
+            except (
+                asyncio.TimeoutError,
+                RuntimeError,
+                RequiredFieldValidationError,
+            ) as exc:
+                log.warning(
+                    f"Couldn't generate midpoint price for {trade.contract}, skipping repricing"
                 )
+                if self.data_store:
+                    self.data_store.record_event(
+                        "order_price_adjustment_skipped",
+                        {
+                            "symbol": getattr(trade.contract, "symbol", ""),
+                            "secType": getattr(trade.contract, "secType", ""),
+                            "reason": type(exc).__name__,
+                        },
+                    )
                 continue
 
     async def get_write_threshold(