From 5c2015e2bd6db240dc47f135e7320f3bca072315 Mon Sep 17 00:00:00 2001 From: Martin Molinero Date: Mon, 15 Apr 2024 12:41:25 -0300 Subject: [PATCH] Pep8 Python Selection Models Update - Update python selection models to pep 8, keeping backwards compatibility support --- .../ETFConstituentsUniverseSelectionModel.py | 14 ++--- .../EmaCrossUniverseSelectionModel.py | 30 +++++----- .../FundamentalUniverseSelectionModel.py | 58 ++++++++++-------- .../Selection/FutureUniverseSelectionModel.py | 33 +++++----- .../Selection/OptionUniverseSelectionModel.py | 33 +++++----- .../Selection/QC500UniverseSelectionModel.py | 60 +++++++++---------- .../Selection/ManualUniverseSelectionModel.py | 42 ++++++------- Algorithm/Selection/UniverseSelectionModel.py | 10 +++- .../QC500UniverseSelectionModelTests.cs | 4 +- 9 files changed, 152 insertions(+), 132 deletions(-) diff --git a/Algorithm.Framework/Selection/ETFConstituentsUniverseSelectionModel.py b/Algorithm.Framework/Selection/ETFConstituentsUniverseSelectionModel.py index be0fea0695d1..b3e0f9ca1405 100644 --- a/Algorithm.Framework/Selection/ETFConstituentsUniverseSelectionModel.py +++ b/Algorithm.Framework/Selection/ETFConstituentsUniverseSelectionModel.py @@ -18,8 +18,8 @@ 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: @@ -27,11 +27,11 @@ def __init__(self, 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 @@ -39,12 +39,12 @@ def __init__(self, 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] diff --git a/Algorithm.Framework/Selection/EmaCrossUniverseSelectionModel.py b/Algorithm.Framework/Selection/EmaCrossUniverseSelectionModel.py index 2c2696e8e0fa..a2e5b8a5c350 100644 --- a/Algorithm.Framework/Selection/EmaCrossUniverseSelectionModel.py +++ b/Algorithm.Framework/Selection/EmaCrossUniverseSelectionModel.py @@ -30,14 +30,14 @@ 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 @@ -46,12 +46,12 @@ def SelectCoarse(self, algorithm, coarse): 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 @@ -62,22 +62,22 @@ def SelectCoarse(self, algorithm, coarse): filtered = sorted(filtered, key=lambda avg: avg.ScaledDelta, 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) + self.symbol = symbol + self.fast_ema = ExponentialMovingAverage(fastPeriod) + self.slow_ema = ExponentialMovingAverage(slowPeriod) @property def Fast(self): - return float(self.FastEma.Current.Value) + return float(self.fast_ema.current.value) @property def Slow(self): - return float(self.SlowEma.Current.Value) + return float(self.slow_ema.current.value) # computes an object score of how much large the fast is than the slow @property @@ -86,4 +86,4 @@ def ScaledDelta(self): # 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) diff --git a/Algorithm.Framework/Selection/FundamentalUniverseSelectionModel.py b/Algorithm.Framework/Selection/FundamentalUniverseSelectionModel.py index 74720061fb4a..4452e2562022 100644 --- a/Algorithm.Framework/Selection/FundamentalUniverseSelectionModel.py +++ b/Algorithm.Framework/Selection/FundamentalUniverseSelectionModel.py @@ -23,46 +23,53 @@ 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: @@ -70,12 +77,15 @@ def FilteredSelectCoarse(self, algorithm, coarse): 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] diff --git a/Algorithm.Framework/Selection/FutureUniverseSelectionModel.py b/Algorithm.Framework/Selection/FutureUniverseSelectionModel.py index 1094d54646f8..ab019474f3d3 100644 --- a/Algorithm.Framework/Selection/FutureUniverseSelectionModel.py +++ b/Algorithm.Framework/Selection/FutureUniverseSelectionModel.py @@ -26,36 +26,39 @@ def __init__(self, refreshInterval: Time interval between universe refreshes 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 diff --git a/Algorithm.Framework/Selection/OptionUniverseSelectionModel.py b/Algorithm.Framework/Selection/OptionUniverseSelectionModel.py index ab4d9a0df0f9..2adeaa3ff9f9 100644 --- a/Algorithm.Framework/Selection/OptionUniverseSelectionModel.py +++ b/Algorithm.Framework/Selection/OptionUniverseSelectionModel.py @@ -25,35 +25,38 @@ def __init__(self, refreshInterval: Time interval between universe refreshes 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 diff --git a/Algorithm.Framework/Selection/QC500UniverseSelectionModel.py b/Algorithm.Framework/Selection/QC500UniverseSelectionModel.py index 33fc0a616d1b..5f9383c56f55 100644 --- a/Algorithm.Framework/Selection/QC500UniverseSelectionModel.py +++ b/Algorithm.Framework/Selection/QC500UniverseSelectionModel.py @@ -1,4 +1,4 @@ -# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. +# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. # Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation. # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -23,63 +23,63 @@ class QC500UniverseSelectionModel(FundamentalUniverseSelectionModel): def __init__(self, filterFineData = True, universeSettings = None): '''Initializes a new default instance of the QC500UniverseSelectionModel''' super().__init__(filterFineData, universeSettings) - self.numberOfSymbolsCoarse = 1000 - self.numberOfSymbolsFine = 500 - self.dollarVolumeBySymbol = {} - self.lastMonth = -1 + self.number_of_symbols_coarse = 1000 + self.number_of_symbols_fine = 500 + self.dollar_volume_by_symbol = {} + self.last_month = -1 - def SelectCoarse(self, algorithm, coarse): + def select_coarse(self, algorithm: QCAlgorithm, fundamental: list[Fundamental]): '''Performs coarse selection for the QC500 constituents. The stocks must have fundamental data The stock must have positive previous-day close price The stock must have positive volume on the previous trading day''' - if algorithm.Time.month == self.lastMonth: - return Universe.Unchanged + if algorithm.time.month == self.last_month: + return Universe.UNCHANGED - sortedByDollarVolume = sorted([x for x in coarse if x.HasFundamentalData and x.Volume > 0 and x.Price > 0], - key = lambda x: x.DollarVolume, reverse=True)[:self.numberOfSymbolsCoarse] + sorted_by_dollar_volume = sorted([x for x in fundamental if x.has_fundamental_data and x.volume > 0 and x.price > 0], + key = lambda x: x.dollar_volume, reverse=True)[:self.number_of_symbols_coarse] - self.dollarVolumeBySymbol = {x.Symbol:x.DollarVolume for x in sortedByDollarVolume} + self.dollar_volume_by_symbol = {x.Symbol:x.dollar_volume for x in sorted_by_dollar_volume} # If no security has met the QC500 criteria, the universe is unchanged. # A new selection will be attempted on the next trading day as self.lastMonth is not updated - if len(self.dollarVolumeBySymbol) == 0: - return Universe.Unchanged + if len(self.dollar_volume_by_symbol) == 0: + return Universe.UNCHANGED # return the symbol objects our sorted collection - return list(self.dollarVolumeBySymbol.keys()) + return list(self.dollar_volume_by_symbol.keys()) - def SelectFine(self, algorithm, fine): + def select_fine(self, algorithm: QCAlgorithm, fundamental: list[Fundamental]): '''Performs fine selection for the QC500 constituents The company's headquarter must in the U.S. The stock must be traded on either the NYSE or NASDAQ At least half a year since its initial public offering The stock's market cap must be greater than 500 million''' - sortedBySector = sorted([x for x in fine if x.CompanyReference.CountryId == "USA" - and x.CompanyReference.PrimaryExchangeID in ["NYS","NAS"] - and (algorithm.Time - x.SecurityReference.IPODate).days > 180 - and x.MarketCap > 5e8], - key = lambda x: x.CompanyReference.IndustryTemplateCode) + sorted_by_sector = sorted([x for x in fundamental if x.company_reference.country_id == "USA" + and x.company_reference.primary_exchange_id in ["NYS","NAS"] + and (algorithm.time - x.security_reference.ipo_date).days > 180 + and x.market_cap > 5e8], + key = lambda x: x.company_reference.industry_template_code) - count = len(sortedBySector) + count = len(sorted_by_sector) # If no security has met the QC500 criteria, the universe is unchanged. # A new selection will be attempted on the next trading day as self.lastMonth is not updated if count == 0: - return Universe.Unchanged + return Universe.UNCHANGED # Update self.lastMonth after all QC500 criteria checks passed - self.lastMonth = algorithm.Time.month + self.last_month = algorithm.time.month - percent = self.numberOfSymbolsFine / count - sortedByDollarVolume = [] + percent = self.number_of_symbols_fine / count + sorted_by_dollar_volume = [] # select stocks with top dollar volume in every single sector - for code, g in groupby(sortedBySector, lambda x: x.CompanyReference.IndustryTemplateCode): - y = sorted(g, key = lambda x: self.dollarVolumeBySymbol[x.Symbol], reverse = True) + for code, g in groupby(sorted_by_sector, lambda x: x.company_reference.industry_template_code): + y = sorted(g, key = lambda x: self.dollar_volume_by_symbol[x.Symbol], reverse = True) c = ceil(len(y) * percent) - sortedByDollarVolume.extend(y[:c]) + sorted_by_dollar_volume.extend(y[:c]) - sortedByDollarVolume = sorted(sortedByDollarVolume, key = lambda x: self.dollarVolumeBySymbol[x.Symbol], reverse=True) - return [x.Symbol for x in sortedByDollarVolume[:self.numberOfSymbolsFine]] + sorted_by_dollar_volume = sorted(sorted_by_dollar_volume, key = lambda x: self.dollar_volume_by_symbol[x.Symbol], reverse=True) + return [x.Symbol for x in sorted_by_dollar_volume[:self.number_of_symbols_fine]] diff --git a/Algorithm/Selection/ManualUniverseSelectionModel.py b/Algorithm/Selection/ManualUniverseSelectionModel.py index b535e50e331e..d4889e74b00e 100644 --- a/Algorithm/Selection/ManualUniverseSelectionModel.py +++ b/Algorithm/Selection/ManualUniverseSelectionModel.py @@ -20,48 +20,48 @@ class ManualUniverseSelectionModel(UniverseSelectionModel): '''Provides an implementation of IUniverseSelectionModel that simply subscribes to the specified set of symbols''' - def __init__(self, symbols = list(), universeSettings = None): - self.MarketHours = MarketHoursDatabase.FromDataFolder() + def __init__(self, symbols = list(), universe_settings = None): + self.marketHours = MarketHoursDatabase.from_data_folder() self.symbols = symbols - self.universeSettings = universeSettings + self.universe_settings = universe_settings for symbol in symbols: - SymbolCache.Set(symbol.Value, symbol) + SymbolCache.set(symbol.Value, symbol) - def CreateUniverses(self, algorithm): + def create_universes(self, algorithm: QCAlgorithm) -> list[Universe]: '''Creates the universes for this algorithm. Called once after IAlgorithm.Initialize Args: algorithm: The algorithm instance to create universes for Returns: The universes to be used by the algorithm''' - universeSettings = self.universeSettings \ - if self.universeSettings is not None else algorithm.UniverseSettings + universe_settings = self.universe_settings \ + if self.universe_settings is not None else algorithm.universe_settings - resolution = universeSettings.Resolution - type = typeof(Tick) if resolution == Resolution.Tick else typeof(TradeBar) + resolution = universe_settings.resolution + type = typeof(Tick) if resolution == Resolution.TICK else typeof(TradeBar) universes = list() # universe per security type/market - self.symbols = sorted(self.symbols, key=lambda s: (s.ID.Market, s.SecurityType)) - for key, grp in groupby(self.symbols, lambda s: (s.ID.Market, s.SecurityType)): + self.symbols = sorted(self.symbols, key=lambda s: (s.id.market, s.security_type)) + for key, grp in groupby(self.symbols, lambda s: (s.id.market, s.security_type)): market = key[0] - securityType = key[1] - securityTypeString = Extensions.GetEnumString(securityType, SecurityType) - universeSymbol = Symbol.Create(f"manual-universe-selection-model-{securityTypeString}-{market}", securityType, market) + security_type = key[1] + security_type_str = Extensions.get_enum_string(security_type, SecurityType) + universe_symbol = Symbol.create(f"manual-universe-selection-model-{security_type_str}-{market}", security_type, market) - if securityType == SecurityType.Base: + if security_type == SecurityType.BASE: # add an entry for this custom universe symbol -- we don't really know the time zone for sure, # but we set it to TimeZones.NewYork in AddData, also, since this is a manual universe, the time # zone doesn't actually matter since this universe specifically doesn't do anything with data. - symbolString = MarketHoursDatabase.GetDatabaseSymbolKey(universeSymbol) - alwaysOpen = SecurityExchangeHours.AlwaysOpen(TimeZones.NewYork) - entry = self.MarketHours.SetEntry(market, symbolString, securityType, alwaysOpen, TimeZones.NewYork) + symbol_string = MarketHoursDatabase.get_database_symbol_key(universe_symbol) + always_open = SecurityExchangeHours.always_open(TimeZones.NEW_YORK) + entry = self.marketHours.set_entry(market, symbol_string, security_type, always_open, TimeZones.NEW_YORK) else: - entry = self.MarketHours.GetEntry(market, None, securityType) + entry = self.marketHours.get_entry(market, None, security_type) - config = SubscriptionDataConfig(type, universeSymbol, resolution, entry.DataTimeZone, entry.ExchangeHours.TimeZone, False, False, True) - universes.append( ManualUniverse(config, universeSettings, list(grp))) + config = SubscriptionDataConfig(type, universe_symbol, resolution, entry.data_time_zone, entry.exchange_hours.time_zone, False, False, True) + universes.append( ManualUniverse(config, universe_settings, list(grp))) return universes diff --git a/Algorithm/Selection/UniverseSelectionModel.py b/Algorithm/Selection/UniverseSelectionModel.py index 2eabf8d3b7af..5c5cb01733a4 100644 --- a/Algorithm/Selection/UniverseSelectionModel.py +++ b/Algorithm/Selection/UniverseSelectionModel.py @@ -1,4 +1,4 @@ -# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. +# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. # Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation. # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -16,14 +16,18 @@ class UniverseSelectionModel: '''Provides a base class for universe selection models.''' - def GetNextRefreshTimeUtc(self): + def get_next_refresh_time_utc(self) -> datetime: '''Gets the next time the framework should invoke the `CreateUniverses` method to refresh the set of universes.''' + if hasattr(self, "GetNextRefreshTimeUtc") and callable(self.GetNextRefreshTimeUtc): + return self.GetNextRefreshTimeUtc() return datetime.max - def CreateUniverses(self, algorithm): + def create_universes(self, algorithm: QCAlgorithm) -> list[Universe]: '''Creates the universes for this algorithm. Called once after Args: algorithm: The algorithm instance to create universes for Returns: The universes to be used by the algorithm''' + if hasattr(self, "CreateUniverses") and callable(self.CreateUniverses): + return self.CreateUniverses(algorithm) raise NotImplementedError("Types deriving from 'UniverseSelectionModel' must implement the 'def CreateUniverses(QCAlgorithm) method.") diff --git a/Tests/Algorithm/Framework/Selection/QC500UniverseSelectionModelTests.cs b/Tests/Algorithm/Framework/Selection/QC500UniverseSelectionModelTests.cs index 9259775fc053..76d5877e5762 100644 --- a/Tests/Algorithm/Framework/Selection/QC500UniverseSelectionModelTests.cs +++ b/Tests/Algorithm/Framework/Selection/QC500UniverseSelectionModelTests.cs @@ -200,8 +200,8 @@ private void GetUniverseSelectionModel( { var name = "QC500UniverseSelectionModel"; dynamic model = Py.Import(name).GetAttr(name).Invoke(); - SelectCoarse = ConvertToUniverseSelectionSymbolDelegate>(model.SelectCoarse); - SelectFine = ConvertToUniverseSelectionSymbolDelegate>(model.SelectFine); + SelectCoarse = ConvertToUniverseSelectionSymbolDelegate>(model.select_coarse); + SelectFine = ConvertToUniverseSelectionSymbolDelegate>(model.select_fine); } }