Fix bug/syntax in python examples

Algorithm.Python/AutoRegressiveIntegratedMovingAverageRegressionAlgorithm.py

@@ -19,25 +19,26 @@
 # is sufficiently large (which would be due to the inclusion of the MA(1) term).
 # </summary>
 class AutoRegressiveIntegratedMovingAverageRegressionAlgorithm(QCAlgorithm):
-    def initialize(self):
+    def initialize(self) -> None:
         '''Initialise the data and resolution required, as well as the cash and start-end dates for your algorithm. All algorithms must initialized.'''
         self.set_start_date(2013, 1, 7)
         self.set_end_date(2013, 12, 11)
         self.settings.automatic_indicator_warm_up = True
         self.add_equity("SPY", Resolution.DAILY)
         self._arima = self.arima("SPY", 1, 1, 1, 50)
         self._ar = self.arima("SPY", 1, 1, 0, 50)
+        self._last = None
 
-    def on_data(self, data):
+    def on_data(self, data: Slice) -> None:
         '''OnData event is the primary entry point for your algorithm. Each new data point will be pumped in here.
 
         Arguments:
             data: Slice object keyed by symbol containing the stock data
         '''
         if self._arima.is_ready:
             if abs(self._arima.current.value - self._ar.current.value) > 1:
-                if self._arima.current.value > self.last:
+                if self._arima.current.value > self._last:
                     self.market_order("SPY", 1)
                 else:
                     self.market_order("SPY", -1)
-            self.last = self._arima.current.value
+            self._last = self._arima.current.value

Algorithm.Python/BasicTemplateOptionEquityStrategyAlgorithm.py

@@ -24,24 +24,25 @@
 class BasicTemplateOptionEquityStrategyAlgorithm(QCAlgorithm):
     underlying_ticker = "GOOG"
 
-    def initialize(self):
+    def initialize(self) -> None:
         self.set_start_date(2015, 12, 24)
         self.set_end_date(2015, 12, 24)
 
         equity = self.add_equity(self.underlying_ticker)
         option = self.add_option(self.underlying_ticker)
-        self.option_symbol = option.symbol
+        self._option_symbol = option.symbol
 
         # set our strike/expiry filter for this option chain
         option.set_filter(lambda u: (u.strikes(-2, +2)
                                      # Expiration method accepts TimeSpan objects or integer for days.
                                      # The following statements yield the same filtering criteria
                                      .expiration(0, 180)))
 
-    def on_data(self, slice):
-        if self.portfolio.invested or not self.is_market_open(self.option_symbol): return
+    def on_data(self, slice: Slice) -> None:
+        if self.portfolio.invested or not self.is_market_open(self._option_symbol):
+            return
 
-        chain = slice.option_chains.get_value(self.option_symbol)
+        chain = slice.option_chains.get_value(self._option_symbol)
         if chain is None:
             return
 
@@ -57,9 +58,9 @@ def on_data(self, slice):
         middle_strike = call_contracts[1].strike
         higher_strike = call_contracts[2].strike
 
-        option_strategy = OptionStrategies.call_butterfly(self.option_symbol, higher_strike, middle_strike, lower_strike, expiry)
+        option_strategy = OptionStrategies.call_butterfly(self._option_symbol, higher_strike, middle_strike, lower_strike, expiry)
 
         self.order(option_strategy, 10)
 
-    def on_order_event(self, order_event):
+    def on_order_event(self, order_event: OrderEvent) -> None:
         self.log(str(order_event))

Algorithm.Python/Benchmarks/IndicatorRibbonBenchmark.py

@@ -19,25 +19,25 @@ class IndicatorRibbonBenchmark(QCAlgorithm):
     def initialize(self):
         self.set_start_date(2010, 1, 1)  #Set Start Date
         self.set_end_date(2018, 1, 1)    #Set End Date
-        self.spy = self.add_equity("SPY", Resolution.MINUTE).symbol
+        self._spy = self.add_equity("SPY", Resolution.MINUTE).symbol
         count = 50
         offset = 5
         period = 15
-        self.ribbon = []
+        self._ribbon = []
         # define our sma as the base of the ribbon
-        self.sma = SimpleMovingAverage(period)
+        self._sma = SimpleMovingAverage(period)
 
         for x in range(count):
             # define our offset to the zero sma, these various offsets will create our 'displaced' ribbon
             delay = Delay(offset*(x+1))
             # define an indicator that takes the output of the sma and pipes it into our delay indicator
-            delayed_sma = IndicatorExtensions.of(delay, self.sma)
+            delayed_sma = IndicatorExtensions.of(delay, self._sma)
             # register our new 'delayed_sma' for automatic updates on a daily resolution
-            self.register_indicator(self.spy, delayed_sma, Resolution.DAILY)
-            self.ribbon.append(delayed_sma)
+            self.register_indicator(self._spy, delayed_sma, Resolution.DAILY)
+            self._ribbon.append(delayed_sma)
 
     def on_data(self, data):
         # wait for our entire ribbon to be ready
-        if not all(x.is_ready for x in self.ribbon): return
-        for x in self.ribbon:
+        if not all(x.is_ready for x in self._ribbon): return
+        for x in self._ribbon:
             value = x.current.value

Algorithm.Python/BybitCustomDataCryptoRegressionAlgorithm.py

@@ -11,15 +11,16 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from datetime import time
+import os
 from AlgorithmImports import *
-import datetime
 
 ### <summary>
 ### Algorithm demonstrating and ensuring that Bybit crypto brokerage model works as expected with custom data types
 ### </summary>
 class BybitCustomDataCryptoRegressionAlgorithm(QCAlgorithm):
 
-    def initialize(self):
+    def initialize(self) -> None:
         self.set_start_date(2022, 12, 13)
         self.set_end_date(2022, 12, 13)
 
@@ -29,42 +30,42 @@ def initialize(self):
         self.set_brokerage_model(BrokerageName.BYBIT, AccountType.CASH)
 
         symbol = self.add_crypto("BTCUSDT").symbol
-        self.btc_usdt = self.add_data(CustomCryptoData, symbol, Resolution.MINUTE).symbol
+        self._btc_usdt = self.add_data(CustomCryptoData, symbol, Resolution.MINUTE).symbol
 
         # create two moving averages
-        self.fast = self.ema(self.btc_usdt, 30, Resolution.MINUTE)
-        self.slow = self.ema(self.btc_usdt, 60, Resolution.MINUTE)
+        self._fast = self.ema(self._btc_usdt, 30, Resolution.MINUTE)
+        self._slow = self.ema(self._btc_usdt, 60, Resolution.MINUTE)
 
-    def on_data(self, data):
-        if not self.slow.is_ready:
+    def on_data(self, data: Slice) -> None:
+        if not self._slow.is_ready:
             return
 
-        if self.fast.current.value > self.slow.current.value:
+        if self._fast.current.value > self._slow.current.value:
             if self.transactions.orders_count == 0:
-                self.buy(self.btc_usdt, 1)
+                self.buy(self._btc_usdt, 1)
         else:
             if self.transactions.orders_count == 1:
-                self.liquidate(self.btc_usdt)
+                self.liquidate(self._btc_usdt)
 
-    def on_order_event(self, order_event):
+    def on_order_event(self, order_event: OrderEvent) -> None:
         self.debug(f"{self.time} {order_event}")
 
 class CustomCryptoData(PythonData):
-    def get_source(self, config, date, is_live_mode):
+    def get_source(self, config: SubscriptionDataConfig, date: datetime, is_live_mode: bool) -> SubscriptionDataSource:
         tick_type_string = Extensions.tick_type_to_lower(config.tick_type)
         formatted_date = date.strftime("%Y%m%d")
-        source = os.path.join(Globals.DataFolder, "crypto", "bybit", "minute",
+        source = os.path.join(Globals.data_folder, "crypto", "bybit", "minute",
                               config.symbol.value.lower(), f"{formatted_date}_{tick_type_string}.zip")
 
         return SubscriptionDataSource(source, SubscriptionTransportMedium.LOCAL_FILE, FileFormat.CSV)
 
-    def reader(self, config, line, date, is_live_mode):
+    def reader(self, config: SubscriptionDataConfig, line: str, date: datetime, is_live_mode: bool) -> BaseData:
         csv = line.split(',')
 
         data = CustomCryptoData()
         data.symbol = config.symbol
 
-        data_datetime = datetime.datetime.combine(date.date(), datetime.time()) + timedelta(milliseconds=int(csv[0]))
+        data_datetime = datetime.combine(date.date(), time()) + timedelta(milliseconds=int(csv[0]))
         data.time = Extensions.convert_to(data_datetime, config.data_time_zone, config.exchange_time_zone)
         data.end_time = data.time + timedelta(minutes=1)
 

Algorithm.Python/CallbackCommandRegressionAlgorithm.py

@@ -22,11 +22,12 @@ class VoidCommand():
     parameters = {}
     targettime = None
 
-    def run(self, algo: QCAlgorithm) -> bool | None:
+    def run(self, algo: IAlgorithm) -> bool:
         if not self.targettime or self.targettime != algo.time:
-            return
+            return False
         tag = self.parameters["tag"]
         algo.order(self.target[0], self.get_quantity(), tag=tag)
+        return True
 
     def get_quantity(self):
         return self.quantity
@@ -36,7 +37,7 @@ class BoolCommand(Command):
     array_test = []
     result = False
 
-    def run(self, algo: QCAlgorithm) -> bool | None:
+    def run(self, algo: QCAlgorithm) -> bool:
         trade_ibm = self.my_custom_method()
         if trade_ibm:
             algo.debug(f"BoolCommand.run: {str(self)}")
@@ -96,13 +97,13 @@ def initialize(self):
 
         # We need to create a project on QuantConnect to test the broadcast_command method
         # and use the project_id in the broadcast_command call
-        self.project_id = 21805137;
+        self.project_id = 21805137
 
         # All live deployments receive the broadcasts below
-        broadcast_result = self.broadcast_command(potential_command);
-        broadcast_result2 = self.broadcast_command({ "symbol": "SPY", "parameters": { "quantity": 10 } });
+        broadcast_result = self.broadcast_command(potential_command)
+        broadcast_result2 = self.broadcast_command({ "symbol": "SPY", "parameters": { "quantity": 10 } })
 
-    def on_command(self, data):
+    def on_command(self, data: object) -> bool:
         self.debug(f"on_command: {str(data)}")
         self.buy(data.symbol, data.parameters["quantity"])
         return True # False, None

Algorithm.Python/ClassicRangeConsolidatorAlgorithm.py

@@ -19,10 +19,10 @@
 ### Example algorithm of how to use ClassicRangeConsolidator
 ### </summary>
 class ClassicRangeConsolidatorAlgorithm(RangeConsolidatorAlgorithm):
-    def create_range_consolidator(self):
+    def create_range_consolidator(self) -> ClassicRangeConsolidator:
         return ClassicRangeConsolidator(self.get_range())
 
-    def on_data_consolidated(self, sender, range_bar):
+    def on_data_consolidated(self, sender: object, range_bar: RangeBar):
         super().on_data_consolidated(sender, range_bar)
 
         if range_bar.volume == 0:

Algorithm.Python/ClassicRenkoConsolidatorAlgorithm.py

@@ -23,8 +23,7 @@
 class ClassicRenkoConsolidatorAlgorithm(QCAlgorithm):
     '''Demonstration of how to initialize and use the RenkoConsolidator'''
 
-    def initialize(self):
-
+    def initialize(self) -> None:
         self.set_start_date(2012, 1, 1)
         self.set_end_date(2013, 1, 1)
 
@@ -48,11 +47,11 @@ def initialize(self):
 
 
     # We're doing our analysis in the on_renko_bar method, but the framework verifies that this method exists, so we define it.
-    def on_data(self, data):
+    def on_data(self, data: Slice) -> None:
         pass
 
 
-    def handle_renko_close(self, sender, data):
+    def handle_renko_close(self, sender: object, data: RenkoBar) -> None:
         '''This function is called by our renko_close consolidator defined in Initialize()
         Args:
             data: The new renko bar produced by the consolidator'''
@@ -62,7 +61,7 @@ def handle_renko_close(self, sender, data):
         self.log(f"CLOSE - {data.time} - {data.open} {data.close}")
 
 
-    def handle_renko7_bar(self, sender, data):
+    def handle_renko7_bar(self, sender: object, data: RenkoBar) -> None:
         '''This function is called by our renko7bar consolidator defined in Initialize()
         Args:
             data: The new renko bar produced by the consolidator'''

Algorithm.Python/CoarseFineOptionUniverseChainRegressionAlgorithm.py

@@ -19,7 +19,7 @@
 ### </summary>
 class CoarseFineOptionUniverseChainRegressionAlgorithm(QCAlgorithm):
 
-    def initialize(self):
+    def initialize(self) -> None:
         '''Initialise the data and resolution required, as well as the cash and start-end dates for your algorithm. All algorithms must initialized.'''
 
         self.set_start_date(2014,6,4)
@@ -41,27 +41,27 @@ def initialize(self):
 
         self.add_universe_options(universe, self.option_filter_function)
 
-    def option_filter_function(self, universe):
+    def option_filter_function(self, universe: OptionFilterUniverse) -> OptionFilterUniverse:
         universe.include_weeklys().front_month()
 
-        contracts: list[OptionUniverse] = list()
+        contracts = list()
         for contract in universe:
             if len(contracts) == 5:
                 break
             contracts.append(contract)
         return universe.contracts(contracts)
 
-    def coarse_selection_function(self, coarse):
+    def coarse_selection_function(self, coarse: list[CoarseFundamental]) -> list[Symbol]:
         if self.time <= datetime(2014,6,5):
             return [ self._twx ]
         return [ self._aapl ]
 
-    def fine_selection_function(self, fine):
+    def fine_selection_function(self, fine: list[FineFundamental]) -> list[Symbol]:
         if self.time <= datetime(2014,6,5):
             return [ self._twx ]
         return [ self._aapl ]
 
-    def on_data(self, data):
+    def on_data(self, data: Slice) -> None:
         if self._changes == None or any(security.price == 0 for security in self._changes.added_securities):
             return
 
@@ -83,12 +83,12 @@ def on_data(self, data):
         self._changes = None
 
     # this event fires whenever we have changes to our universe
-    def on_securities_changed(self, changes):
+    def on_securities_changed(self, changes: SecurityChanges) -> None:
         if self._changes == None:
             self._changes = changes
             return
         self._changes = self._changes + changes
 
-    def on_end_of_algorithm(self):
+    def on_end_of_algorithm(self) -> None:
         if self._option_count == 0:
             raise ValueError("Option universe chain did not add any option!")