Pep8 Python Base Selection Models Update

Algorithm.Framework/Selection/ETFConstituentsUniverseSelectionModel.py

@@ -18,33 +18,33 @@ class ETFConstituentsUniverseSelectionModel(UniverseSelectionModel):
     '''Universe selection model that selects the constituents of an ETF.'''
 
     def __init__(self,
-                 etfSymbol, 
-                 universeSettings = None, 
+                 etfSymbol,
+                 universeSettings = None,
                  universeFilterFunc = None):
         '''Initializes a new instance of the ETFConstituentsUniverseSelectionModel class
         Args:
             etfSymbol: Symbol of the ETF to get constituents for
             universeSettings: Universe settings
             universeFilterFunc: Function to filter universe results'''
         if type(etfSymbol) is str:
-            symbol = SymbolCache.TryGetSymbol(etfSymbol, None)
-            if symbol[0] and symbol[1].SecurityType == SecurityType.Equity:
+            symbol = SymbolCache.try_get_symbol(etfSymbol, None)
+            if symbol[0] and symbol[1].security_type == SecurityType.EQUITY:
                 self.etf_symbol = symbol[1]
             else:
-                self.etf_symbol = Symbol.Create(etfSymbol, SecurityType.Equity, Market.USA)
+                self.etf_symbol = Symbol.create(etfSymbol, SecurityType.EQUITY, Market.USA)
         else:
             self.etf_symbol = etfSymbol
         self.universe_settings = universeSettings
         self.universe_filter_function = universeFilterFunc
 
         self.universe = None
 
-    def CreateUniverses(self, algorithm: QCAlgorithm) -> List[Universe]:
+    def create_universes(self, algorithm: QCAlgorithm) -> list[Universe]:
         '''Creates a new ETF constituents universe using this class's selection function
         Args:
             algorithm: The algorithm instance to create universes for
         Returns:
             The universe defined by this model'''
         if self.universe is None:
-            self.universe = algorithm.Universe.ETF(self.etf_symbol, self.universe_settings, self.universe_filter_function)           
+            self.universe = algorithm.universe.etf(self.etf_symbol, self.universe_settings, self.universe_filter_function)
         return [self.universe]

Algorithm.Framework/Selection/EmaCrossUniverseSelectionModel.py

@@ -30,60 +30,60 @@ def __init__(self,
             universeCount: Maximum number of members of this universe selection
             universeSettings: The settings used when adding symbols to the algorithm, specify null to use algorithm.UniverseSettings'''
         super().__init__(False, universeSettings)
-        self.fastPeriod = fastPeriod
-        self.slowPeriod = slowPeriod
-        self.universeCount = universeCount
+        self.fast_period = fastPeriod
+        self.slow_period = slowPeriod
+        self.universe_count = universeCount
         self.tolerance = 0.01
         # holds our coarse fundamental indicators by symbol
         self.averages = {}
 
-    def SelectCoarse(self, algorithm, coarse):
+    def select_coarse(self, algorithm: QCAlgorithm, fundamental: list[Fundamental]) -> list[Symbol]:
         '''Defines the coarse fundamental selection function.
         Args:
             algorithm: The algorithm instance
-            coarse: The coarse fundamental data used to perform filtering</param>
+            fundamental: The coarse fundamental data used to perform filtering</param>
         Returns:
             An enumerable of symbols passing the filter'''
         filtered = []
 
-        for cf in coarse:
-            if cf.Symbol not in self.averages:
-                self.averages[cf.Symbol] = self.SelectionData(cf.Symbol, self.fastPeriod, self.slowPeriod)
+        for cf in fundamental:
+            if cf.symbol not in self.averages:
+                self.averages[cf.symbol] = self.SelectionData(cf.symbol, self.fast_period, self.slow_period)
 
             # grab th SelectionData instance for this symbol
-            avg = self.averages.get(cf.Symbol)
+            avg = self.averages.get(cf.symbol)
 
             # Update returns true when the indicators are ready, so don't accept until they are
             # and only pick symbols who have their fastPeriod-day ema over their slowPeriod-day ema
-            if avg.Update(cf.EndTime, cf.AdjustedPrice) and avg.Fast > avg.Slow * (1 + self.tolerance):
+            if avg.Update(cf.end_time, cf.adjusted_price) and avg.fast > avg.slow * (1 + self.tolerance):
                 filtered.append(avg)
 
         # prefer symbols with a larger delta by percentage between the two averages
-        filtered = sorted(filtered, key=lambda avg: avg.ScaledDelta, reverse = True)
+        filtered = sorted(filtered, key=lambda avg: avg.scaled_delta, reverse = True)
 
         # we only need to return the symbol and return 'universeCount' symbols
-        return [x.Symbol for x in filtered[:self.universeCount]]
+        return [x.Symbol for x in filtered[:self.universe_count]]
 
     # class used to improve readability of the coarse selection function
     class SelectionData:
-        def __init__(self, symbol, fastPeriod, slowPeriod):
-            self.Symbol = symbol
-            self.FastEma = ExponentialMovingAverage(fastPeriod)
-            self.SlowEma = ExponentialMovingAverage(slowPeriod)
+        def __init__(self, symbol, fast_period, slow_period):
+            self.symbol = symbol
+            self.fast_ema = ExponentialMovingAverage(fast_period)
+            self.slow_ema = ExponentialMovingAverage(slow_period)
 
         @property
-        def Fast(self):
-            return float(self.FastEma.Current.Value)
+        def fast(self):
+            return float(self.fast_ema.current.value)
 
         @property
-        def Slow(self):
-            return float(self.SlowEma.Current.Value)
+        def slow(self):
+            return float(self.slow_ema.current.value)
 
         # computes an object score of how much large the fast is than the slow
         @property
-        def ScaledDelta(self):
-            return (self.Fast - self.Slow) / ((self.Fast + self.Slow) / 2)
+        def scaled_delta(self):
+            return (self.fast - self.slow) / ((self.fast + self.slow) / 2)
 
         # updates the EMAFast and EMASlow indicators, returning true when they're both ready
         def Update(self, time, value):
-            return self.SlowEma.Update(time, value) & self.FastEma.Update(time, value)
+            return self.slow_ema.update(time, value) & self.fast_ema.update(time, value)

Algorithm.Framework/Selection/FundamentalUniverseSelectionModel.py

@@ -23,59 +23,69 @@ def __init__(self,
         Args:
             filterFineData: [Obsolete] Fine and Coarse selection are merged
             universeSettings: The settings used when adding symbols to the algorithm, specify null to use algorithm.UniverseSettings'''
-        self.filterFineData = filterFineData
-        if self.filterFineData == None:
-            self._fundamentalData = True
+        self.filter_fine_data = filterFineData
+        if self.filter_fine_data == None:
+            self.fundamental_data = True
         else:
-            self._fundamentalData = False
+            self.fundamental_data = False
         self.market = Market.USA
-        self.universeSettings = universeSettings
+        self.universe_settings = universeSettings
 
 
-    def CreateUniverses(self, algorithm):
+    def create_universes(self, algorithm: QCAlgorithm) -> list[Universe]:
         '''Creates a new fundamental universe using this class's selection functions
         Args:
             algorithm: The algorithm instance to create universes for
         Returns:
             The universe defined by this model'''
-        if self._fundamentalData:
-            universeSettings = algorithm.UniverseSettings if self.universeSettings is None else self.universeSettings
-            universe = FundamentalUniverseFactory(self.market, universeSettings, lambda fundamental: self.Select(algorithm, fundamental))
+        if self.fundamental_data:
+            universe_settings = algorithm.universe_settings if self.universe_settings is None else self.universe_settings
+            # handle both 'Select' and 'select' for backwards compatibility
+            selection = lambda fundamental: self.select(algorithm, fundamental)
+            if hasattr(self, "Select") and callable(self.Select):
+                selection = lambda fundamental: self.Select(algorithm, fundamental)
+            universe = FundamentalUniverseFactory(self.market, universe_settings, selection)
             return [universe]
         else:
-            universe = self.CreateCoarseFundamentalUniverse(algorithm)
-            if self.filterFineData:
-                if universe.UniverseSettings.Asynchronous:
+            universe = self.create_coarse_fundamental_universe(algorithm)
+            if self.filter_fine_data:
+                if universe.universe_settings.asynchronous:
                     raise ValueError("Asynchronous universe setting is not supported for coarse & fine selections, please use the new Fundamental single pass selection")
-                universe = FineFundamentalFilteredUniverse(universe, lambda fine: self.SelectFine(algorithm, fine))
+                selection = lambda fine: self.select_fine(algorithm, fine)
+                if hasattr(self, "SelectFine") and callable(self.SelectFine):
+                    selection = lambda fine: self.SelectFine(algorithm, fine)
+                universe = FineFundamentalFilteredUniverse(universe, selection)
             return [universe]
 
 
-    def CreateCoarseFundamentalUniverse(self, algorithm):
+    def create_coarse_fundamental_universe(self, algorithm: QCAlgorithm) -> Universe:
         '''Creates the coarse fundamental universe object.
         This is provided to allow more flexibility when creating coarse universe.
         Args:
             algorithm: The algorithm instance
         Returns:
             The coarse fundamental universe'''
-        universeSettings = algorithm.UniverseSettings if self.universeSettings is None else self.universeSettings
-        return CoarseFundamentalUniverse(universeSettings, lambda coarse: self.FilteredSelectCoarse(algorithm, coarse))
+        universe_settings = algorithm.universe_settings if self.universe_settings is None else self.universe_settings
+        return CoarseFundamentalUniverse(universe_settings, lambda coarse: self.filtered_select_coarse(algorithm, coarse))
 
 
-    def FilteredSelectCoarse(self, algorithm, coarse):
+    def filtered_select_coarse(self, algorithm: QCAlgorithm, fundamental: list[Fundamental]) -> list[Symbol]:
         '''Defines the coarse fundamental selection function.
         If we're using fine fundamental selection than exclude symbols without fine data
         Args:
             algorithm: The algorithm instance
             coarse: The coarse fundamental data used to perform filtering
         Returns:
             An enumerable of symbols passing the filter'''
-        if self.filterFineData:
-            coarse = filter(lambda c: c.HasFundamentalData, coarse)
-        return self.SelectCoarse(algorithm, coarse)
+        if self.filter_fine_data:
+            fundamental = filter(lambda c: c.has_fundamental_data, fundamental)
+        if hasattr(self, "SelectCoarse") and callable(self.SelectCoarse):
+            # handle both 'select_coarse' and 'SelectCoarse' for backwards compatibility
+            return self.SelectCoarse(algorithm, fundamental)
+        return self.select_coarse(algorithm, fundamental)
 
 
-    def Select(self, algorithm, fundamental):
+    def select(self, algorithm: QCAlgorithm, fundamental: list[Fundamental]) -> list[Symbol]:
         '''Defines the fundamental selection function.
         Args:
             algorithm: The algorithm instance
@@ -85,7 +95,7 @@ def Select(self, algorithm, fundamental):
         raise NotImplementedError("Please overrride the 'Select' fundamental function")
 
 
-    def SelectCoarse(self, algorithm, coarse):
+    def select_coarse(self, algorithm: QCAlgorithm, fundamental: list[Fundamental]) -> list[Symbol]:
         '''Defines the coarse fundamental selection function.
         Args:
             algorithm: The algorithm instance
@@ -95,11 +105,11 @@ def SelectCoarse(self, algorithm, coarse):
         raise NotImplementedError("Please overrride the 'Select' fundamental function")
 
 
-    def SelectFine(self, algorithm, fine):
+    def select_fine(self, algorithm: QCAlgorithm, fundamental: list[Fundamental]) -> list[Symbol]:
         '''Defines the fine fundamental selection function.
         Args:
             algorithm: The algorithm instance
             fine: The fine fundamental data used to perform filtering
         Returns:
             An enumerable of symbols passing the filter'''
-        return [f.Symbol for f in fine]
+        return [f.symbol for f in fundamental]

Algorithm.Framework/Selection/FutureUniverseSelectionModel.py

@@ -26,36 +26,39 @@ def __init__(self,
             refreshInterval: Time interval between universe refreshes</param>
             futureChainSymbolSelector: Selects symbols from the provided future chain
             universeSettings: Universe settings define attributes of created subscriptions, such as their resolution and the minimum time in universe before they can be removed'''
-        self.nextRefreshTimeUtc = datetime.min
+        self.next_refresh_time_utc = datetime.min
 
-        self.refreshInterval = refreshInterval
-        self.futureChainSymbolSelector = futureChainSymbolSelector
-        self.universeSettings = universeSettings
+        self.refresh_interval = refreshInterval
+        self.future_chain_symbol_selector = futureChainSymbolSelector
+        self.universe_settings = universeSettings
 
-    def GetNextRefreshTimeUtc(self):
+    def get_next_refresh_time_utc(self):
         '''Gets the next time the framework should invoke the `CreateUniverses` method to refresh the set of universes.'''
-        return self.nextRefreshTimeUtc
+        return self.next_refresh_time_utc
 
-    def CreateUniverses(self, algorithm):
+    def create_universes(self, algorithm: QCAlgorithm) -> list[Universe]:
         '''Creates a new fundamental universe using this class's selection functions
         Args:
             algorithm: The algorithm instance to create universes for
         Returns:
             The universe defined by this model'''
-        self.nextRefreshTimeUtc = algorithm.UtcTime + self.refreshInterval
+        self.next_refresh_time_utc = algorithm.utc_time + self.refresh_interval
 
-        uniqueSymbols = set()
-        for futureSymbol in self.futureChainSymbolSelector(algorithm.UtcTime):
-            if futureSymbol.SecurityType != SecurityType.Future:
+        unique_symbols = set()
+        for future_symbol in self.future_chain_symbol_selector(algorithm.utc_time):
+            if future_symbol.SecurityType != SecurityType.FUTURE:
                 raise ValueError("futureChainSymbolSelector must return future symbols.")
 
             # prevent creating duplicate future chains -- one per symbol
-            if futureSymbol not in uniqueSymbols:
-                uniqueSymbols.add(futureSymbol)
-                for universe in Extensions.CreateFutureChain(algorithm, futureSymbol, self.Filter, self.universeSettings):
+            if future_symbol not in unique_symbols:
+                unique_symbols.add(future_symbol)
+                selection = self.filter
+                if hasattr(self, "Filter") and callable(self.Filter):
+                    selection = self.Filter
+                for universe in Extensions.create_future_chain(algorithm, future_symbol, selection, self.universe_settings):
                     yield universe
 
-    def Filter(self, filter):
+    def filter(self, filter):
         '''Defines the future chain universe filter'''
         # NOP
         return filter

Algorithm.Framework/Selection/OptionUniverseSelectionModel.py

@@ -25,35 +25,38 @@ def __init__(self,
             refreshInterval: Time interval between universe refreshes</param>
             optionChainSymbolSelector: Selects symbols from the provided option chain
             universeSettings: Universe settings define attributes of created subscriptions, such as their resolution and the minimum time in universe before they can be removed'''
-        self.nextRefreshTimeUtc = datetime.min
+        self.next_refresh_time_utc = datetime.min
 
-        self.refreshInterval = refreshInterval
-        self.optionChainSymbolSelector = optionChainSymbolSelector
-        self.universeSettings = universeSettings
+        self.refresh_interval = refreshInterval
+        self.option_chain_symbol_selector = optionChainSymbolSelector
+        self.universe_settings = universeSettings
 
-    def GetNextRefreshTimeUtc(self):
+    def get_next_refresh_time_utc(self):
         '''Gets the next time the framework should invoke the `CreateUniverses` method to refresh the set of universes.'''
-        return self.nextRefreshTimeUtc
+        return self.next_refresh_time_utc
 
-    def CreateUniverses(self, algorithm):
+    def create_universes(self, algorithm: QCAlgorithm) -> list[Universe]:
         '''Creates a new fundamental universe using this class's selection functions
         Args:
             algorithm: The algorithm instance to create universes for
         Returns:
             The universe defined by this model'''
-        self.nextRefreshTimeUtc = (algorithm.UtcTime + self.refreshInterval).date()
+        self.next_refresh_time_utc = (algorithm.utc_time + self.refresh_interval).date()
 
         uniqueUnderlyingSymbols = set()
-        for optionSymbol in self.optionChainSymbolSelector(algorithm.UtcTime):
-            if not Extensions.IsOption(optionSymbol.SecurityType):
+        for option_symbol in self.option_chain_symbol_selector(algorithm.utc_time):
+            if not Extensions.is_option(option_symbol.security_type):
                 raise ValueError("optionChainSymbolSelector must return option, index options, or futures options symbols.")
 
             # prevent creating duplicate option chains -- one per underlying
-            if optionSymbol.Underlying not in uniqueUnderlyingSymbols:
-                uniqueUnderlyingSymbols.add(optionSymbol.Underlying)
-                yield Extensions.CreateOptionChain(algorithm, optionSymbol, self.Filter, self.universeSettings)
-
-    def Filter(self, filter):
+            if option_symbol.underlying not in uniqueUnderlyingSymbols:
+                uniqueUnderlyingSymbols.add(option_symbol.underlying)
+                selection = self.filter
+                if hasattr(self, "Filter") and callable(self.Filter):
+                    selection = self.Filter
+                yield Extensions.create_option_chain(algorithm, option_symbol, selection, self.universe_settings)
+
+    def filter(self, filter):
         '''Defines the option chain universe filter'''
         # NOP
         return filter