Adds Support for ETF Constituent Universes (#5862)

* Adds support for ETF constituent universes

  * Adds filtering for universe data if it doesn't match the
    universe subscription type

  * Update mapping for ALL underlying Symbols if
    `Symbol.UpdateMappedSymbol(...)` is called. Required to support
    constituent ETF universes that might have mapping events

  * Delistings of composite constituent universe Symbol will result in
    removal of universe securities.

  * Added regression algorithms for ETF constituent mappings (C#/Python),
    along with data required to run locally

  * Refactor universe delistings in SubscriptionSynchronizer -
    big thank you to @Martin-Molinero :)

* Address review: update regression algorithms and add explanatory comments

* Address review: add additional checks to delisting regression algorithms

  * Adds new regression algorithm testing the addition of a universe
    without calling AddEquity() and asserts same behavior

Algorithm.Python/ETFConstituentUniverseCompositeDelistingRegressionAlgorithm.py

@@ -0,0 +1,68 @@
+# 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");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from AlgorithmImports import *
+
+### <summary>
+### Tests the delisting of the composite Symbol (ETF symbol) and the removal of
+### the universe and the symbol from the algorithm.
+### </summary>
+class ETFConstituentUniverseCompositeDelistingRegressionAlgorithm(QCAlgorithm):
+    def Initialize(self):
+        self.SetStartDate(2020, 12, 1)
+        self.SetEndDate(2021, 1, 31)
+        self.SetCash(100000)
+
+        self.universeSymbolCount = 0
+        self.universeAdded = False
+        self.universeRemoved = False
+
+        self.UniverseSettings.Resolution = Resolution.Hour
+        self.delistingDate = date(2021, 1, 21)
+
+        self.aapl = self.AddEquity("AAPL", Resolution.Hour).Symbol
+        self.gdvd = self.AddEquity("GDVD", Resolution.Hour).Symbol
+
+        self.AddUniverse(ETFConstituentsUniverse(self.gdvd, self.UniverseSettings, self.FilterETFs))
+
+    def FilterETFs(self, constituents):
+        if self.UtcTime.date() > self.delistingDate:
+            raise Exception(f"Performing constituent universe selection on {self.UtcTime.strftime('%Y-%m-%d %H:%M:%S.%f')} after composite ETF has been delisted")
+
+        constituentSymbols = [i.Symbol for i in constituents]
+        self.universeSymbolCount = len(constituentSymbols)
+
+        return constituentSymbols
+
+    def OnData(self, data):
+        if self.UtcTime.date() > self.delistingDate and any([i != self.aapl for i in data.Keys]):
+            raise Exception("Received unexpected slice in OnData(...) after universe was deselected")
+
+        if not self.Portfolio.Invested:
+            self.SetHoldings(self.aapl, 0.5)
+
+    def OnSecuritiesChanged(self, changes):
+        if len(changes.AddedSecurities) != 0 and self.UtcTime.date() > self.delistingDate:
+            raise Exception("New securities added after ETF constituents were delisted")
+
+        self.universeAdded = self.universeAdded or len(changes.AddedSecurities) >= self.universeSymbolCount
+        # Subtract 1 from universe Symbol count for AAPL, since it was manually added to the algorithm
+        self.universeRemoved = self.universeRemoved or (len(changes.RemovedSecurities) == self.universeSymbolCount - 1 and self.UtcTime.date() >= self.delistingDate and self.UtcTime.date() < self.EndDate.date())
+
+    def OnEndOfAlgorithm(self):
+        if not self.universeAdded:
+            raise Exception("ETF constituent universe was never added to the algorithm")
+        if not self.universeRemoved:
+            raise Exception("ETF constituent universe was not removed from the algorithm after delisting")
+        if len(self.ActiveSecurities) > 2:
+            raise Exception(f"Expected less than 2 securities after algorithm ended, found {len(self.Securities)}")

Algorithm.Python/ETFConstituentUniverseCompositeDelistingRegressionAlgorithmNoAddEquityETF.py

@@ -0,0 +1,68 @@
+# 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");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from AlgorithmImports import *
+
+### <summary>
+### Tests the delisting of the composite Symbol (ETF symbol) and the removal of
+### the universe and the symbol from the algorithm, without adding a subscription via AddEquity
+### </summary>
+class ETFConstituentUniverseCompositeDelistingRegressionAlgorithmNoAddEquityETF(QCAlgorithm):
+    def Initialize(self):
+        self.SetStartDate(2020, 12, 1)
+        self.SetEndDate(2021, 1, 31)
+        self.SetCash(100000)
+
+        self.universeSymbolCount = 0
+        self.universeAdded = False
+        self.universeRemoved = False
+
+        self.UniverseSettings.Resolution = Resolution.Hour
+        self.delistingDate = date(2021, 1, 21)
+
+        self.aapl = self.AddEquity("AAPL", Resolution.Hour).Symbol
+        self.gdvd = Symbol.Create("GDVD", SecurityType.Equity, Market.USA)
+
+        self.AddUniverse(ETFConstituentsUniverse(self.gdvd, self.UniverseSettings, self.FilterETFs))
+
+    def FilterETFs(self, constituents):
+        if self.UtcTime.date() > self.delistingDate:
+            raise Exception(f"Performing constituent universe selection on {self.UtcTime.strftime('%Y-%m-%d %H:%M:%S.%f')} after composite ETF has been delisted")
+
+        constituentSymbols = [i.Symbol for i in constituents]
+        self.universeSymbolCount = len(constituentSymbols)
+
+        return constituentSymbols
+
+    def OnData(self, data):
+        if self.UtcTime.date() > self.delistingDate and any([i != self.aapl for i in data.Keys]):
+            raise Exception("Received unexpected slice in OnData(...) after universe was deselected")
+
+        if not self.Portfolio.Invested:
+            self.SetHoldings(self.aapl, 0.5)
+
+    def OnSecuritiesChanged(self, changes):
+        if len(changes.AddedSecurities) != 0 and self.UtcTime.date() > self.delistingDate:
+            raise Exception("New securities added after ETF constituents were delisted")
+
+        self.universeAdded = self.universeAdded or len(changes.AddedSecurities) >= self.universeSymbolCount
+        # Subtract 1 from universe Symbol count for AAPL, since it was manually added to the algorithm
+        self.universeRemoved = self.universeRemoved or (len(changes.RemovedSecurities) == self.universeSymbolCount - 1 and self.UtcTime.date() >= self.delistingDate and self.UtcTime.date() < self.EndDate.date())
+
+    def OnEndOfAlgorithm(self):
+        if not self.universeAdded:
+            raise Exception("ETF constituent universe was never added to the algorithm")
+        if not self.universeRemoved:
+            raise Exception("ETF constituent universe was not removed from the algorithm after delisting")
+        if len(self.ActiveSecurities) > 2:
+            raise Exception(f"Expected less than 2 securities after algorithm ended, found {len(self.Securities)}")

Algorithm.Python/ETFConstituentUniverseFilterFunctionRegressionAlgorithm.py

@@ -0,0 +1,111 @@
+# 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");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from AlgorithmImports import *
+
+### <summary>
+### Tests a custom filter function when creating an ETF constituents universe for SPY
+### </summary>
+class ETFConstituentUniverseFilterFunctionRegressionAlgorithm(QCAlgorithm):
+    def Initialize(self):
+        self.SetStartDate(2020, 12, 1)
+        self.SetEndDate(2021, 1, 31)
+        self.SetCash(100000)
+
+        self.filtered = False
+        self.securitiesChanged = False
+        self.receivedData = False
+        self.etfConstituentData = {}
+        self.etfRebalanced = False
+        self.rebalanceCount = 0
+        self.rebalanceAssetCount = 0
+
+        self.UniverseSettings.Resolution = Resolution.Hour
+
+        self.spy = self.AddEquity("SPY", Resolution.Hour).Symbol
+        self.aapl = Symbol.Create("AAPL", SecurityType.Equity, Market.USA)
+
+        self.AddUniverse(ETFConstituentsUniverse(self.spy, self.UniverseSettings, self.FilterETFs))
+
+    def FilterETFs(self, constituents):
+        constituentsData = list(constituents)
+        constituentsSymbols = [i.Symbol for i in constituentsData]
+        self.etfConstituentData = {i.Symbol: i for i in constituentsData}
+
+        if len(constituentsData) == 0:
+            raise Exception(f"Constituents collection is empty on {self.UtcTime.strftime('%Y-%m-%d %H:%M:%S.%f')}")
+        if self.aapl not in constituentsSymbols:
+            raise Exception("AAPL is not int he constituents data provided to the algorithm")
+
+        aaplData = [i for i in constituentsData if i.Symbol == self.aapl][0]
+        if aaplData.Weight == 0.0:
+            raise Exception("AAPL weight is expected to be a non-zero value")
+
+        self.filtered = True
+        self.etfRebalanced = True
+
+        return constituentsSymbols
+
+    def OnData(self, data):
+        if not self.filtered and len(data.Bars) != 0 and self.aapl in data.Bars:
+            raise Exception("AAPL TradeBar data added to algorithm before constituent universe selection took place")
+        
+        if len(data.Bars) == 1 and self.spy in data.Bars:
+            return
+
+        if len(data.Bars) != 0 and self.aapl not in data.Bars:
+            raise Exception(f"Expected AAPL TradeBar data on {self.UtcTime.strftime('%Y-%m-%d %H:%M:%S.%f')}")
+
+        self.receivedData = True
+
+        if not self.etfRebalanced:
+            return
+
+        for bar in data.Bars.Values:
+            constituentData = self.etfConstituentData.get(bar.Symbol)
+            if constituentData is not None and constituentData.Weight is not None and constituentData.Weight >= 0.0001:
+                # If the weight of the constituent is less than 1%, then it will be set to 1%
+                # If the weight of the constituent exceeds more than 5%, then it will be capped to 5%
+                # Otherwise, if the weight falls in between, then we use that value.
+                boundedWeight = max(0.01, min(constituentData.Weight, 0.05))
+
+                self.SetHoldings(bar.Symbol, boundedWeight)
+
+                if self.etfRebalanced:
+                    self.rebalanceCount += 1
+
+                self.etfRebalanced = False
+                self.rebalanceAssetCount += 1
+
+
+    def OnSecuritiesChanged(self, changes):
+        if self.filtered and not self.securitiesChanged and len(changes.AddedSecurities) < 500:
+            raise Exception(f"Added SPY S&P 500 ETF to algorithm, but less than 500 equities were loaded (added {len(changes.AddedSecurities)} securities)")
+
+        self.securitiesChanged = True
+
+    def OnEndOfAlgorithm(self):
+        if self.rebalanceCount != 1:
+            raise Exception(f"Expected 1 rebalance, instead rebalanced: {self.rebalanceCount}")
+
+        if self.rebalanceAssetCount != 4:
+            raise Exception(f"Invested in {self.rebalanceAssetCount} assets (expected 4)")
+
+        if not self.filtered:
+            raise Exception("Universe selection was never triggered")
+
+        if not self.securitiesChanged:
+            raise Exception("Security changes never propagated to the algorithm")
+
+        if not self.receivedData:
+            raise Exception("Data was never loaded for the S&P 500 constituent AAPL")

Algorithm.Python/ETFConstituentUniverseMappedCompositeRegressionAlgorithm.py

@@ -0,0 +1,88 @@
+# 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");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from AlgorithmImports import *
+
+### <summary>
+### Tests the mapping of the ETF symbol that has a constituent universe attached to it and ensures
+### that data is loaded after the mapping event takes place.
+### </summary>
+class ETFConstituentUniverseFilterFunctionRegressionAlgorithm(QCAlgorithm):
+    def Initialize(self):
+        self.SetStartDate(2011, 2, 1)
+        self.SetEndDate(2011, 4, 4)
+        self.SetCash(100000)
+
+        self.filterDateConstituentSymbolCount = {}
+        self.constituentDataEncountered = {}
+        self.constituentSymbols = []
+        self.mappingEventOccurred = False
+
+        self.UniverseSettings.Resolution = Resolution.Hour
+
+        self.aapl = Symbol.Create("AAPL", SecurityType.Equity, Market.USA)
+        self.qqq = self.AddEquity("QQQ", Resolution.Daily).Symbol
+
+        self.AddUniverse(ETFConstituentsUniverse(self.qqq, self.UniverseSettings, self.FilterETFs))
+
+    def FilterETFs(self, constituents):
+        constituentSymbols = [i.Symbol for i in constituents]
+
+        if self.aapl not in constituentSymbols:
+            raise Exception("AAPL not found in QQQ constituents")
+
+        self.filterDateConstituentSymbolCount[self.UtcTime.date()] = len(constituentSymbols)
+        for symbol in constituentSymbols:
+            self.constituentSymbols.append(symbol)
+
+        self.constituentSymbols = list(set(self.constituentSymbols))
+        return constituentSymbols
+
+    def OnData(self, data):
+        if len(data.SymbolChangedEvents) != 0:
+            for symbolChanged in data.SymbolChangedEvents.Values:
+                if symbolChanged.Symbol != self.qqq:
+                    raise Exception(f"Mapped symbol is not QQQ. Instead, found: {symbolChanged.Symbol}")
+                if symbolChanged.OldSymbol != "QQQQ":
+                    raise Exception(f"Old QQQ Symbol is not QQQQ. Instead, found: {symbolChanged.OldSymbol}")
+                if symbolChanged.NewSymbol != "QQQ":
+                    raise Exception(f"New QQQ Symbol is not QQQ. Instead, found: {symbolChanged.NewSymbol}")
+
+                self.mappingEventOccurred = True
+
+        if self.qqq in data and len([i for i in data.Keys]) == 1:
+            return
+
+        if self.UtcTime.date() not in self.constituentDataEncountered:
+            self.constituentDataEncountered[self.UtcTime.date()] = False
+        
+        if len([i for i in data.Keys if i in self.constituentSymbols]) != 0:
+            self.constituentDataEncountered[self.UtcTime.date()] = True
+
+        if not self.Portfolio.Invested:
+            self.SetHoldings(self.aapl, 0.5)
+
+    def OnEndOfAlgorithm(self):
+        if len(self.filterDateConstituentSymbolCount) != 2:
+            raise Exception(f"ETF constituent filtering function was not called 2 times (actual: {len(self.filterDateConstituentSymbolCount)}")
+
+        if not self.mappingEventOccurred:
+            raise Exception("No mapping/SymbolChangedEvent occurred. Expected for QQQ to be mapped from QQQQ -> QQQ");
+
+        for constituentDate, constituentsCount in self.filterDateConstituentSymbolCount.items():
+            if constituentsCount < 25:
+                raise Exception(f"Expected 25 or more constituents in filter function on {constituentDate}, found {constituentsCount}")
+
+        for constituentDate, constituentEncountered in self.constituentDataEncountered.items():
+            if not constituentEncountered:
+                raise Exception(f"Received data in OnData(...) but it did not contain any constituent data on {constituentDate.strftime('%Y-%m-%d %H:%M:%S.%f')}")