Merge pull request #970 from Lumiwealth/projectx_order_management

Author: grzesir

lumibot/backtesting/backtesting_broker.py

@@ -2910,10 +2910,9 @@ def _ny_date(value):
                         if order.order_type == Order.OrderType.LIMIT
                         else f"type={order.order_type}, high={high}, low={low}, stop={getattr(order, 'stop_price', None)}"
                     )
-                    strategy.log_message(
+                    self.logger.debug(
                         f"[DIAG] Order remained open for {display_symbol} ({detail}) "
-                        f"id={order_identifier} at {self.datetime}",
-                        color="yellow",
+                        f"id={order_identifier} at {self.datetime}"
                     )
                 continue
 

lumibot/brokers/broker.py

@@ -1208,10 +1208,35 @@ def _set_initial_positions(self, strategy):
             if pos.quantity != 0.0:
                 self._filled_positions.append(pos)
 
+    def _clean_order_trackers(self, broker_order):
+        """
+        ProjectX has a race condition where an order is filled and added to new trackers at the same time and the
+        'new' queue never gets cleared. This function is used to clean the order trackers of any duplicated orders.
+        Keep orders that are completed (i.e. filled, canceled, error) and remove any duplicates from the 'new' and
+        'unprocessed' trackers.
+        """
+        if not broker_order.is_active():
+            self._new_orders.remove(broker_order.identifier, key="identifier")
+            self._unprocessed_orders.remove(broker_order.identifier, key="identifier")
+            self._partially_filled_orders.remove(broker_order.identifier, key="identifier")
+        elif broker_order in self._partially_filled_orders:
+            self._new_orders.remove(broker_order.identifier, key="identifier")
+            self._unprocessed_orders.remove(broker_order.identifier, key="identifier")
+        elif broker_order in self._new_orders:
+            self._unprocessed_orders.remove(broker_order.identifier, key="identifier")
+
     def _process_new_order(self, order):
-        # Check if this order already exists in self._new_orders based on the identifier
-        if order in self._new_orders:
-            return order
+        # Don't duplicate orders in the new orders tracker. Check if an order with the same identifier already exists
+        # in the tracked orders.
+        existing_order = self.get_tracked_order(order.identifier)
+        if existing_order:
+            # Check if this order already exists in self._new_orders based on the identifier - Do nothing
+            if existing_order in self._new_orders:
+                return existing_order
+            if existing_order not in self._unprocessed_orders:
+                return existing_order  # Exists in another tracker, return it without adding to prevent duplicates
+            else:
+                order = existing_order  # Use the existing order object from unprocessed and update status
 
         self._unprocessed_orders.remove(order.identifier, key="identifier")
         order.status = self.NEW_ORDER

lumibot/brokers/projectx.py

@@ -1000,8 +1000,9 @@ def _parse_simple_order(self, schwab_order: dict, strategy_name: str) -> List[Or
 
                 # Set the status and timestamps
                 order.status = status
-                order.created_at = entered_time
-                order.updated_at = close_time if close_time else entered_time
+                order.broker_create_date = entered_time
+                order.created_at = order.broker_create_date
+                order.broker_update_date = close_time if close_time else entered_time
 
                 order_objects.append(order)
 

lumibot/brokers/schwab.py

@@ -1235,7 +1235,7 @@ def is_canceled(self):
         bool
             True if the order has been cancelled, False otherwise.
         """
-        return self.status.lower() in ["cancelled", "canceled", "cancel", "error", "expired"]
+        return self.status.lower() in ["cancelled", "canceled", "cancel", "cancelling", "error", "expired"]
 
     def is_filled(self):
         """

lumibot/entities/order.py

@@ -1396,9 +1396,14 @@ def get_tracked_orders(self):
         self.log_message("Warning: get_tracked_orders() is deprecated, please use get_orders() instead.")
         return self.get_orders()
 
-    def get_orders(self):
+    def get_orders(self, identifiers: list[str] = None):
         """Get all the current open orders.
 
+        Parameters
+        ----------
+        identifiers : list of str
+            A list of order identifiers to filter the orders by. If None, returns all tracked orders for the strategy.
+
         Returns
         -------
         list of Order objects
@@ -1423,7 +1428,11 @@ def get_orders(self):
         >>>         self.cancel_order(order)
 
         """
-        return self.broker.get_tracked_orders(self.name)
+        all_orders = self.broker.get_tracked_orders(self.name)
+        if identifiers:
+            filtered_orders = [order for order in all_orders if order.identifier in identifiers]
+            return filtered_orders
+        return all_orders
 
     def get_tracked_assets(self):
         """Get the list of assets for positions

lumibot/strategies/strategy.py

@@ -344,7 +344,17 @@ def sync_broker(self):
             for order in orders_broker:
                 # Check against existing orders.
                 order_lumi = [ord_lumi for ord_lumi in orders_lumi if ord_lumi.identifier == order.identifier]
-                order_lumi = order_lumi[0] if len(order_lumi) > 0 else None
+                if len(order_lumi) > 1:
+                    self.strategy.logger.warning(
+                        f"Multiple orders found in lumibot with the same identifier {order.identifier}. "
+                        f"This should not happen and indicates a bug in the order tracking. This is manifesting as "
+                        f"a race condition with ProjectX where a 'new' order event is being added along with a 'fill' "
+                        f"and the 'new' queue never gets cleared causing duplicate orders to be added to lumibot. "
+                        f"Orders: {order_lumi}"
+                    )
+                    order_lumi = self.broker._clean_order_trackers(order)
+                else:
+                    order_lumi = order_lumi[0] if len(order_lumi) > 0 else None
 
                 if order_lumi:
                     # Compare the orders.
@@ -427,6 +437,9 @@ def sync_broker(self):
                             self.broker._process_partially_filled_order(order, order.avg_fill_price, order.quantity)
                         elif order.status == Order.OrderStatus.NEW:
                             self.broker._process_new_order(order)
+                        elif order.status == Order.OrderStatus.ERROR:
+                            self.broker._process_new_order(order)
+                            self.broker._process_error_order(order, order.error_message)
                     else:
                         # Add to order in lumibot.
                         self.broker._process_new_order(order)
@@ -462,7 +475,7 @@ def sync_broker(self):
                                 continue
 
                         # Check if it's a market order that might have filled instantly
-                        if order_lumi.order_type and order_lumi.order_type.lower() == "market":
+                        if order_lumi.order_type and order_lumi.order_type == Order.OrderType.MARKET:
                             self.strategy.logger.info(
                                 f"Market order {order_lumi} (id={order_lumi.identifier}) not found in broker, "
                                 f"likely filled instantly - skipping cancel"

lumibot/strategies/strategy_executor.py

@@ -622,7 +622,7 @@ def update_strategy_name(self, new_strategy_name: str):
             self.extra = extra_dict
 
 
-def _ensure_handlers_configured():
+def _ensure_handlers_configured(is_backtest=False):
     """
     Ensure that the root logger has the appropriate handlers configured.
     This is called once globally to set up consistent formatting, but we also
@@ -636,6 +636,14 @@ def _ensure_handlers_configured():
     - LUMIBOT_ERROR_CSV_PATH: Path for CSV error log file (default: logs/errors.csv)
     - BACKTESTING_QUIET_LOGS: Enable quiet logs for backtesting (true/false)
     - LUMIWEALTH_API_KEY: API key for Lumiwealth/Botspot error reporting (when set, enables automatic error reporting)
+
+    Parameters
+    ----------
+    is_backtest : bool, optional
+        Whether we are in a backtesting (default False). This can also be determined from the IS_BACKTESTING
+        environment variable, but is provided here as an optional parameter for cases
+        where we might want to explicitly control this behavior without relying on environment variables.
+
     """
     global _handlers_configured, _BACKTESTING_QUIET_LOGS_ENABLED
 
@@ -646,7 +654,7 @@ def _ensure_handlers_configured():
     except AttributeError:
         log_level = logging.INFO
 
-    is_backtesting = os.environ.get("IS_BACKTESTING", "").lower() == "true"
+    is_backtesting = is_backtest or os.environ.get("IS_BACKTESTING", "").lower() == "true"
 
     # Determine the effective file (root) log level and console level
     if is_backtesting:
@@ -929,7 +937,7 @@ def set_console_log_level(level: str):
                                      f"DEBUG, INFO, WARNING, ERROR, CRITICAL")
 
 
-def add_file_handler(file_path: str, level: str = 'INFO'):
+def add_file_handler(file_path: str, level: str = 'INFO', is_backtest: bool = False):
     """
     Add a file handler to the root logger to also log to a file.
     
@@ -939,13 +947,16 @@ def add_file_handler(file_path: str, level: str = 'INFO'):
         Path to the log file.
     level : str, optional
         Log level for the file handler. Defaults to 'INFO'.
+    is_backtest : bool, optional
+        Whether this is for backtesting (default False). This will ensure that the file handler respects
+        the backtesting quiet logs/console setting if applicable.
         
     Examples
     --------
     >>> from lumibot.tools.lumibot_logger import add_file_handler
     >>> add_file_handler('/path/to/lumibot.log', 'DEBUG')
     """
-    _ensure_handlers_configured()
+    _ensure_handlers_configured(is_backtest=is_backtest)
 
     try:
         file_level = getattr(logging, level.upper())

lumibot/tools/lumibot_logger.py

@@ -524,7 +524,32 @@ def order_cancel(self, account_id: int, order_id: int) -> dict:
         except requests.exceptions.RequestException as e:
             self.logger.error(f"Error cancelling order: {e}")
             return {"success": False, "error": str(e)}
-    
+
+    def order_modify(self, account_id: int, order_id: int, size: int = None, limit_price: float = None,
+                     stop_price: float = None, trail_price: float = None) -> dict:
+        """Modify an existing order"""
+        url = f"{self.base_url}api/order/modify"
+        payload = {
+            "accountId": account_id,
+            "orderId": order_id,
+            "size": size,
+            "limitPrice": limit_price,
+            "stopPrice": stop_price,
+            "trailPrice": trail_price,
+        }
+        # Remove None values to avoid API model binding issues
+        payload = {k: v for k, v in payload.items() if v is not None}
+        try:
+            response = requests.post(url, headers=self.headers, json=payload, timeout=10)
+            try:
+                data = response.json()
+            except ValueError:
+                data = {"success": False, "error": f"Non-JSON response {response.status_code}"}
+            return data
+        except requests.exceptions.RequestException as e:
+            self.logger.error(f"Error modifying order: {e}")
+            return {"success": False, "error": str(e)}
+
     def contract_search(self, search_text: str, live: bool = False) -> dict:
         """Search for contracts by name"""
         url = f"{self.base_url}api/contract/search"
@@ -691,7 +716,7 @@ def __init__(self, config: dict):
         self._positions_cache_time = 0
         self._orders_cache = None
         self._orders_cache_time = 0
-        self._cache_ttl = 30  # 30 seconds cache
+        self._cache_ttl = 5  # 30 seconds cache
 
     def get_accounts(self) -> List[Dict]:
         """Get list of available accounts"""
@@ -833,9 +858,20 @@ def cancel_order(self, account_id: int, order_id: str) -> bool:
 
     def order_modify(self, account_id: int, order_id: int, size: int = None, 
                     limit_price: float = None, stop_price: float = None, trail_price: float = None) -> Dict:
-        """Modify an existing order - Not supported by ProjectX API"""
-        # ProjectX doesn't support order modification - need to cancel and re-place
-        return {"success": False, "error": "Order modification not supported by ProjectX API"}
+        """Modify an existing order - definitely supported by ProjectX API"""
+        # ProjectX certainly does support order modification
+        response = self.api.order_modify(
+            account_id=account_id,
+            order_id=order_id,
+            size=size,
+            limit_price=limit_price,
+            stop_price=stop_price,
+            trail_price=trail_price
+        )
+        if response and response.get("success"):
+            return response
+        else:
+            raise Exception(f"Failed to modify order: {response}")
 
     def get_historical_data(self, contract_id: str, start_time: str, end_time: str,
                            timeframe: str = "1minute") -> pd.DataFrame:

lumibot/tools/projectx_helpers.py

@@ -7,7 +7,7 @@
 from pathlib import Path
 from typing import Optional
 
-from lumibot.tools.lumibot_logger import get_logger
+from lumibot.tools.lumibot_logger import get_logger, set_console_log_level
 
 # Overloading time.sleep to warn users against using it
 
@@ -330,6 +330,7 @@ def _set_logger(self):
                 set_log_level("ERROR")
             else:
                 set_log_level("INFO")
+                set_console_log_level("ERROR")  # Set console log level to ERROR while keeping file logs at INFO
                 # When quiet_logs=False, allow INFO logs to console (respects BACKTESTING_QUIET_LOGS)
         else:
             # Live trades should always have full logging for both console and file
@@ -347,7 +348,8 @@ def _set_logger(self):
 
         # Setting file logging if specified
         if self.logfile:
-            add_file_handler(str(self.logfile), level="DEBUG" if self.debug else "INFO")
+            add_file_handler(str(self.logfile), level="DEBUG" if self.debug else "INFO",
+                             is_backtest=self.is_backtest_broker)
 
         # Disable Interactive Brokers logs
         for log_name, log_obj in logging.Logger.manager.loggerDict.items():