Merge pull request #681 from Lumiwealth/dev

update master

.gitignore

@@ -25,6 +25,7 @@ test_bot.py
 lumi_tradier
 lumiwealth_tradier
 ThetaTerminal.jar
+pytest.ini
 
 # Pypi deployment
 build

README.md

@@ -40,8 +40,11 @@ To run this example strategy, click on the `Deploy to Render` button below to de
 
 If you want to contribute to Lumibot, you can check how to get started below. We are always looking for contributors to help us out!
 
+Here's a video to help you get started with contributing to Lumibot: [Watch The Video](https://youtu.be/Huz6VxqafZs)
+
 **Steps to contribute:**
 
+0. Watch the video: [Watch The Video](https://youtu.be/Huz6VxqafZs)
 1. Clone the repository to your local machine
 2. Create a new branch for your feature
 3. Run `pip install -r requirements_dev.txt` to install the developer dependencies

lumibot/backtesting/backtesting_broker.py

@@ -4,7 +4,7 @@
 from decimal import Decimal
 from functools import wraps
 
-import pandas as pd
+import pytz
 
 from lumibot.brokers import Broker
 from lumibot.data_sources import DataSourceBacktesting
@@ -88,13 +88,19 @@ def get_historical_account_value(self):
     def _update_datetime(self, update_dt, cash=None, portfolio_value=None):
         """Works with either timedelta or datetime input
         and updates the datetime of the broker"""
+        tz = self.datetime.tzinfo
+        is_pytz = isinstance(tz, (pytz.tzinfo.StaticTzInfo, pytz.tzinfo.DstTzInfo))
 
         if isinstance(update_dt, timedelta):
             new_datetime = self.datetime + update_dt
         elif isinstance(update_dt, int) or isinstance(update_dt, float):
             new_datetime = self.datetime + timedelta(seconds=update_dt)
         else:
             new_datetime = update_dt
+
+        # This is needed to handle Daylight Savings Time changes
+        new_datetime = tz.normalize(new_datetime) if is_pytz else new_datetime
+
         self.data_source._update_datetime(new_datetime, cash=cash, portfolio_value=portfolio_value)
         if self.option_source:
             self.option_source._update_datetime(new_datetime, cash=cash, portfolio_value=portfolio_value)
@@ -386,6 +392,8 @@ def _process_cash_settlement(self, order, price, quantity):
 
     def submit_order(self, order):
         """Submit an order for an asset"""
+        self._conform_order(order)
+
         # NOTE: This code is to address Tradier API requirements, they want is as "to_open" or "to_close" instead of just "buy" or "sell"
         # If the order has a "buy_to_open" or "buy_to_close" side, then we should change it to "buy"
         if order.is_buy_order():

lumibot/brokers/alpaca.py

@@ -14,6 +14,7 @@
 
 from lumibot.data_sources import AlpacaData
 from lumibot.entities import Asset, Order, Position
+from lumibot.tools.helpers import has_more_than_n_decimal_places
 
 from .broker import Broker
 
@@ -446,6 +447,7 @@ def _submit_order(self, order):
             order.set_identifier(response.id)
             order.status = response.status
             order.update_raw(response)
+            self._unprocessed_orders.append(order)
 
         except Exception as e:
             order.set_error(e)
@@ -467,6 +469,32 @@ def _submit_order(self, order):
 
         return order
 
+    def _conform_order(self, order):
+        """Conform an order to Alpaca's requirements
+        See: https://docs.alpaca.markets/docs/orders-at-alpaca
+        """
+        if order.asset.asset_type == "stock" and order.type == "limit":
+            """
+            The minimum price variance exists for limit orders.
+            Orders received in excess of the minimum price variance will be rejected.
+            Limit price >=$1.00: Max Decimals = 2
+            Limit price <$1.00: Max Decimals = 4
+            """
+            orig_price = order.limit_price
+            conformed = False
+            if order.limit_price >= 1.0 and has_more_than_n_decimal_places(order.limit_price, 2):
+                    order.limit_price = round(order.limit_price, 2)
+                    conformed = True
+            elif order.limit_price < 1.0 and has_more_than_n_decimal_places(order.limit_price, 4):
+                order.limit_price = round(order.limit_price, 4)
+                conformed = True
+
+            if conformed:
+                logging.warning(
+                    f"Order {order} was changed to conform to Alpaca's requirements. "
+                    f"The limit price was changed from {orig_price} to {order.limit_price}."
+                )
+
     def cancel_order(self, order):
         """Cancel an order
 
@@ -520,14 +548,13 @@ def _run_stream(self):
         """
 
         async def _trade_update(trade_update):
-            self._orders_queue.join()
             try:
                 logged_order = trade_update.order
                 type_event = trade_update.event
                 identifier = logged_order.id
                 stored_order = self.get_tracked_order(identifier)
                 if stored_order is None:
-                    logging.info(f"Untracked order {identifier} was logged by broker {self.name}")
+                    logging.debug(f"Untracked order {identifier} was logged by broker {self.name}")
                     return False
 
                 price = trade_update.price

lumibot/brokers/broker.py

@@ -13,9 +13,9 @@
 from dateutil import tz
 from termcolor import colored
 
-from lumibot.data_sources import DataSource
-from lumibot.entities import Asset, Order, Position
-from lumibot.trading_builtins import SafeList
+from ..data_sources import DataSource
+from ..entities import Asset, Order, Position
+from ..trading_builtins import SafeList
 
 
 class CustomLoggerAdapter(logging.LoggerAdapter):
@@ -287,7 +287,7 @@ def sync_positions(self, strategy):
             else:
                 # Add to positions in lumibot, position does not exist
                 # in lumibot.
-                if position.quantity != 0:
+                if position.quantity != 0.0:
                     self._filled_positions.append(position)
 
         # Now iterate through lumibot positions.
@@ -566,28 +566,14 @@ def _wait_for_orders(self):
 
             self._orders_queue.task_done()
 
-    def _submit_orders(self, orders) -> list[Order]:
-        with ThreadPoolExecutor(
-            max_workers=self.max_workers,
-            thread_name_prefix=f"{self.name}_submitting_orders",
-        ) as executor:
-            tasks = []
-            for order in orders:
-                tasks.append(executor.submit(self._submit_order, order))
-
-            result = []
-            for task in as_completed(tasks):
-                result.append(task.result())
-
-        return result
-
     # =========Internal functions==============
 
     def _set_initial_positions(self, strategy):
         """Set initial positions"""
         positions = self._pull_positions(strategy)
         for pos in positions:
-            self._filled_positions.append(pos)
+            if pos.quantity != 0.0:
+                self._filled_positions.append(pos)
 
     def _process_new_order(self, order):
         # Check if this order already exists in self._new_orders based on the identifier
@@ -957,12 +943,30 @@ def _pull_all_orders(self, strategy_name, strategy_object):
         return result
 
     def submit_order(self, order):
-        """Submit an order for an asset"""
+        """Conform an order for an asset to broker constraints and submit it."""
+        self._conform_order(order)
         self._submit_order(order)
 
+    def _conform_order(self, order):
+        """Conform an order to broker constraints. Derived brokers should implement this method."""
+        pass
+
     def submit_orders(self, orders, **kwargs):
         """Submit orders"""
-        self._submit_orders(orders, **kwargs)
+        if hasattr(self, '_submit_orders'):
+            self._submit_orders(orders, **kwargs)
+        else:
+            with ThreadPoolExecutor(
+                max_workers=self.max_workers,
+                thread_name_prefix=f"{self.name}_submitting_orders",
+            ) as executor:
+                tasks = []
+                for order in orders:
+                    tasks.append(executor.submit(self._submit_order, order))
+
+                result = []
+                for task in as_completed(tasks):
+                    result.append(task.result())
 
     def wait_for_order_registration(self, order):
         """Wait for the order to be registered by the broker"""
@@ -1200,21 +1204,21 @@ def _process_trade_event(self, stored_order, type_event, price=None, filled_quan
             except ValueError:
                 raise error
 
-        if type_event == self.NEW_ORDER:
+        if Order.is_equivalent_status(type_event, self.NEW_ORDER):
             stored_order = self._process_new_order(stored_order)
             self._on_new_order(stored_order)
-        elif type_event == self.CANCELED_ORDER:
+        elif Order.is_equivalent_status(type_event, self.CANCELED_ORDER):
             # Do not cancel or re-cancel already completed orders
             if stored_order.is_active():
                 stored_order = self._process_canceled_order(stored_order)
                 self._on_canceled_order(stored_order)
-        elif type_event == self.PARTIALLY_FILLED_ORDER:
+        elif Order.is_equivalent_status(type_event, self.PARTIALLY_FILLED_ORDER):
             stored_order, position = self._process_partially_filled_order(stored_order, price, filled_quantity)
             self._on_partially_filled_order(position, stored_order, price, filled_quantity, multiplier)
-        elif type_event == self.FILLED_ORDER:
+        elif Order.is_equivalent_status(type_event, self.FILLED_ORDER):
             position = self._process_filled_order(stored_order, price, filled_quantity)
             self._on_filled_order(position, stored_order, price, filled_quantity, multiplier)
-        elif type_event == self.CASH_SETTLED:
+        elif Order.is_equivalent_status(type_event, self.CASH_SETTLED):
             self._process_cash_settlement(stored_order, price, filled_quantity)
             stored_order.type = self.CASH_SETTLED
         else:

lumibot/brokers/ccxt.py

@@ -299,7 +299,7 @@ def _parse_broker_order(self, response, strategy_name, strategy_object=None):
             response["side"],
             limit_price=response["price"],
             stop_price=response["stopPrice"],
-            time_in_force=response["timeInForce"].lower(),
+            time_in_force=response["timeInForce"].lower() if response["timeInForce"] else None,
             quote=Asset(
                 symbol=pair[1],
                 asset_type="crypto",
@@ -324,7 +324,10 @@ def _pull_broker_order(self, identifier):
     def _pull_broker_closed_orders(self):
         params = {}
 
-        if self.is_margin_enabled():
+        if self.api.id == "kraken":  # Check if the exchange is Kraken
+            logging.info("Detected Kraken exchange. Not sending params for closed orders.")
+            params = None  # Ensure no parameters are sent
+        elif self.is_margin_enabled():
             params["tradeType"] = "MARGIN_TRADE"
 
         closed_orders = self.api.fetch_closed_orders(params)

lumibot/brokers/interactive_brokers.py

@@ -2,7 +2,7 @@
 import logging
 import random
 import time
-from collections import defaultdict, deque
+from collections import deque
 from decimal import Decimal
 from threading import Thread
 import math
@@ -316,7 +316,7 @@ def _flatten_order(self, orders): # implement for stop loss.
         """Not used for Interactive Brokers. Just returns the orders."""
         return orders
 
-    def submit_orders(self, orders, is_multileg=False, duration="day", price=None, **kwargs):
+    def _submit_orders(self, orders, is_multileg=False, duration="day", price=None, **kwargs):
         if is_multileg:
             multileg_order = OrderLum(orders[0].strategy)
             multileg_order.order_class = OrderLum.OrderClass.MULTILEG
@@ -331,8 +331,10 @@ def submit_orders(self, orders, is_multileg=False, duration="day", price=None, *
 
             # Submit the multileg order.
             self._orders_queue.put(multileg_order)
+            return multileg_order
         else:
             self._orders_queue.put(orders)
+            return orders
 
     def _submit_order(self, order):
         """Submit an order for an asset"""