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..22007f51e4db 100644
--- a/Algorithm.Framework/Selection/EmaCrossUniverseSelectionModel.py
+++ b/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
+ fundamental: The coarse fundamental data used to perform filtering
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)
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);
}
}