From 523d2d848ddf8e745156dd6fd1541b24113142c7 Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Fri, 17 Jul 2026 10:53:45 -0400 Subject: [PATCH 1/8] Fix universe subscription exception when option universe is removed and re-added in the same time step --- ...rseRemovedAndReAddedRegressionAlgorithm.py | 60 +++++++++ Engine/DataFeeds/DataManager.cs | 34 ++++- Tests/Engine/DataFeeds/DataManagerTests.cs | 118 ++++++++++++++++++ 3 files changed, 207 insertions(+), 5 deletions(-) create mode 100644 Algorithm.Python/OptionUniverseRemovedAndReAddedRegressionAlgorithm.py diff --git a/Algorithm.Python/OptionUniverseRemovedAndReAddedRegressionAlgorithm.py b/Algorithm.Python/OptionUniverseRemovedAndReAddedRegressionAlgorithm.py new file mode 100644 index 000000000000..dbf2b32151e4 --- /dev/null +++ b/Algorithm.Python/OptionUniverseRemovedAndReAddedRegressionAlgorithm.py @@ -0,0 +1,60 @@ +# 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 * + +### +### Regression algorithm reproducing GH issue #7682: removing an option universe and re-adding +### one for the same underlying within the same time step must not throw, and the re-added +### universe must keep providing option chain data. +### +class OptionUniverseRemovedAndReAddedRegressionAlgorithm(QCAlgorithm): + + def initialize(self): + self.set_start_date(2014, 6, 6) + self.set_end_date(2014, 6, 9) + self.set_cash(100000) + + self._canonical = self._add_aapl_option() + self._readded = False + self._chains_before = 0 + self._chains_after = 0 + + def _add_aapl_option(self) -> Symbol: + option = self.add_option("AAPL", Resolution.MINUTE) + option.set_filter(-2, 2, 0, 180) + return option.symbol + + def on_data(self, data): + chain = data.option_chains.get(self._canonical) + has_chain = chain is not None and len(chain) > 0 + + if not self._readded: + if has_chain: + self._chains_before += 1 + + if self.time.hour >= 10: + # Remove the option universe and re-add one for the same underlying in the same time step + self.remove_security(self._canonical) + self._canonical = self._add_aapl_option() + self._readded = True + elif has_chain: + self._chains_after += 1 + + def on_end_of_algorithm(self): + if not self._readded: + raise Exception("The option universe was never removed and re-added") + if self._chains_before == 0: + raise Exception("Expected option chain data before the universe was removed") + if self._chains_after == 0: + raise Exception("Expected option chain data after the universe was re-added") diff --git a/Engine/DataFeeds/DataManager.cs b/Engine/DataFeeds/DataManager.cs index b634111f4287..0914dd170b11 100644 --- a/Engine/DataFeeds/DataManager.cs +++ b/Engine/DataFeeds/DataManager.cs @@ -292,12 +292,36 @@ public bool AddSubscription(SubscriptionRequest request) { if (!subscription.EndOfStream) { - // duplicate subscription request - subscription.AddSubscriptionRequest(request); - // only result true if the existing subscription is internal, we actually added something from the users perspective - return subscription.Configuration.IsInternalFeed; + if (request.IsUniverseSubscription + && subscription.IsUniverseSelectionSubscription + && subscription.Universes.All(universe => universe.DisposeRequested)) + { + // GH issue 7682: a universe was removed and a new one with the same configuration was added in the + // same time step. The old subscription removal is performed asynchronously by the synchronizer, + // so we force it out now and carry on creating a new subscription for the new universe + RemoveSubscriptionInternal(request.Configuration, universe: null, forceSubscriptionRemoval: true); + + // the removal above unregistered the configuration, which the new subscription requires + lock (_subscriptionManagerSubscriptions) + { + if (_subscriptionManagerSubscriptions.TryAdd(request.Configuration, request.Configuration)) + { + _subscriptionDataConfigsEnumerator = null; + } + } + } + else + { + // duplicate subscription request + subscription.AddSubscriptionRequest(request); + // only result true if the existing subscription is internal, we actually added something from the users perspective + return subscription.Configuration.IsInternalFeed; + } + } + else + { + DataFeedSubscriptions.TryRemove(request.Configuration, out _); } - DataFeedSubscriptions.TryRemove(request.Configuration, out _); } if (request.Configuration.DataNormalizationMode == DataNormalizationMode.ScaledRaw) diff --git a/Tests/Engine/DataFeeds/DataManagerTests.cs b/Tests/Engine/DataFeeds/DataManagerTests.cs index ee8755b87045..86b049f91f29 100644 --- a/Tests/Engine/DataFeeds/DataManagerTests.cs +++ b/Tests/Engine/DataFeeds/DataManagerTests.cs @@ -221,6 +221,124 @@ public void ConfigurationForAddedSubscriptionIsAlwaysPresent() dataManager.RemoveAllSubscriptions(); } + [Test] + // reproduces GH issue 7682 + public void ReplacesSubscriptionOfUniverseRemovedAndReAddedInSameTimeStep() + { + var dataPermissionManager = new DataPermissionManager(); + var dataFeed = new TestDataFeed(); + var dataManager = new DataManager(dataFeed, + new UniverseSelection(_algorithm, + _securityService, + dataPermissionManager, + TestGlobals.DataProvider), + _algorithm, + _algorithm.TimeKeeper, + MarketHoursDatabase.AlwaysOpen, + false, + new RegisteredSecurityDataTypesProvider(), + dataPermissionManager); + + var config = new SubscriptionDataConfig(typeof(TradeBar), + Symbols.SPY, + Resolution.Daily, + TimeZones.NewYork, + TimeZones.NewYork, + false, + false, + false); + var universeSettings = new UniverseSettings(Resolution.Daily, 1, false, false, TimeSpan.FromDays(365)); + var security = new Equity( + config.Symbol, + SecurityExchangeHours.AlwaysOpen(TimeZones.NewYork), + new Cash(Currencies.USD, 1, 1), + SymbolProperties.GetDefault(Currencies.USD), + new IdentityCurrencyConverter(Currencies.USD), + new RegisteredSecurityDataTypesProvider(), + new SecurityCache()); + + using var oldUniverse = new TestUniverse(config, universeSettings); + var oldRequest = new SubscriptionRequest(true, + oldUniverse, + security, + config, + new DateTime(2019, 1, 1), + new DateTime(2019, 1, 2)); + using var oldEnumerator = new EnqueueableEnumerator(); + dataFeed.Subscription = new Subscription(oldRequest, oldEnumerator, null); + Assert.IsTrue(dataManager.AddSubscription(oldRequest)); + + // the universe is removed: it gets disposed synchronously but its subscription removal + // is deferred to the next synchronizer loop + oldUniverse.Dispose(); + + // a new universe with an equal configuration is added in the same time step, + // before the old subscription has been removed + var newConfig = new SubscriptionDataConfig(config); + using var newUniverse = new TestUniverse(newConfig, universeSettings); + var newRequest = new SubscriptionRequest(true, + newUniverse, + security, + newConfig, + new DateTime(2019, 1, 1), + new DateTime(2019, 1, 2)); + using var newEnumerator = new EnqueueableEnumerator(); + var newSubscription = new Subscription(newRequest, newEnumerator, null); + dataFeed.Subscription = newSubscription; + + Assert.IsTrue(dataManager.AddSubscription(newRequest)); + + // the stale subscription was replaced by one belonging to the new universe + Assert.IsTrue(dataManager.DataFeedSubscriptions.TryGetValue(newConfig, out var currentSubscription)); + Assert.AreSame(newSubscription, currentSubscription); + CollectionAssert.AreEqual(new[] { newUniverse }, currentSubscription.Universes); + // the configuration is still registered for the new universe + Assert.AreEqual(1, dataManager.GetSubscriptionDataConfigs(config.Symbol).Count); + + dataManager.RemoveAllSubscriptions(); + } + + [Test] + // reproduces GH issue 7682 + public void OptionUniverseCanBeRemovedAndReAddedInSameTimeStep() + { + AlgorithmRunner.RunLocalBacktest("OptionUniverseRemovedAndReAddedRegressionAlgorithm", + new Dictionary + { + {"Total Orders", "0"}, + {"Average Win", "0%"}, + {"Average Loss", "0%"}, + {"Compounding Annual Return", "0%"}, + {"Drawdown", "0%"}, + {"Expectancy", "0"}, + {"Start Equity", "100000"}, + {"End Equity", "100000"}, + {"Net Profit", "0%"}, + {"Sharpe Ratio", "0"}, + {"Sortino Ratio", "0"}, + {"Probabilistic Sharpe Ratio", "0%"}, + {"Loss Rate", "0%"}, + {"Win Rate", "0%"}, + {"Profit-Loss Ratio", "0"}, + {"Alpha", "0"}, + {"Beta", "0"}, + {"Annual Standard Deviation", "0"}, + {"Annual Variance", "0"}, + {"Information Ratio", "-9.486"}, + {"Tracking Error", "0.008"}, + {"Treynor Ratio", "0"}, + {"Total Fees", "$0.00"}, + {"Estimated Strategy Capacity", "$0"}, + {"Lowest Capacity Asset", ""}, + {"Portfolio Turnover", "0%"}, + {"Drawdown Recovery", "0"}, + {"OrderListHash", "d41d8cd98f00b204e9800998ecf8427e"} + }, + Language.Python, + AlgorithmStatus.Completed, + algorithmLocation: "../../../Algorithm.Python/OptionUniverseRemovedAndReAddedRegressionAlgorithm.py"); + } + [Test] public void ScaledRawNormalizationModeIsNotAllowed() { From c1a85220cae58337ca6c96eff97d3692e58cd32e Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Fri, 17 Jul 2026 10:56:56 -0400 Subject: [PATCH 2/8] Remove incorrect issue reference from comments --- .../OptionUniverseRemovedAndReAddedRegressionAlgorithm.py | 6 +++--- Engine/DataFeeds/DataManager.cs | 4 ++-- Tests/Engine/DataFeeds/DataManagerTests.cs | 2 -- 3 files changed, 5 insertions(+), 7 deletions(-) diff --git a/Algorithm.Python/OptionUniverseRemovedAndReAddedRegressionAlgorithm.py b/Algorithm.Python/OptionUniverseRemovedAndReAddedRegressionAlgorithm.py index dbf2b32151e4..ea66adb8e14d 100644 --- a/Algorithm.Python/OptionUniverseRemovedAndReAddedRegressionAlgorithm.py +++ b/Algorithm.Python/OptionUniverseRemovedAndReAddedRegressionAlgorithm.py @@ -14,9 +14,9 @@ from AlgorithmImports import * ### -### Regression algorithm reproducing GH issue #7682: removing an option universe and re-adding -### one for the same underlying within the same time step must not throw, and the re-added -### universe must keep providing option chain data. +### Regression algorithm asserting that removing an option universe and re-adding one for the +### same underlying within the same time step does not throw, and that the re-added universe +### keeps providing option chain data. ### class OptionUniverseRemovedAndReAddedRegressionAlgorithm(QCAlgorithm): diff --git a/Engine/DataFeeds/DataManager.cs b/Engine/DataFeeds/DataManager.cs index 0914dd170b11..d86f55b0060e 100644 --- a/Engine/DataFeeds/DataManager.cs +++ b/Engine/DataFeeds/DataManager.cs @@ -296,8 +296,8 @@ public bool AddSubscription(SubscriptionRequest request) && subscription.IsUniverseSelectionSubscription && subscription.Universes.All(universe => universe.DisposeRequested)) { - // GH issue 7682: a universe was removed and a new one with the same configuration was added in the - // same time step. The old subscription removal is performed asynchronously by the synchronizer, + // a universe was removed and a new one with the same configuration was added in the same + // time step. The old subscription removal is performed asynchronously by the synchronizer, // so we force it out now and carry on creating a new subscription for the new universe RemoveSubscriptionInternal(request.Configuration, universe: null, forceSubscriptionRemoval: true); diff --git a/Tests/Engine/DataFeeds/DataManagerTests.cs b/Tests/Engine/DataFeeds/DataManagerTests.cs index 86b049f91f29..3cc4620815ef 100644 --- a/Tests/Engine/DataFeeds/DataManagerTests.cs +++ b/Tests/Engine/DataFeeds/DataManagerTests.cs @@ -222,7 +222,6 @@ public void ConfigurationForAddedSubscriptionIsAlwaysPresent() } [Test] - // reproduces GH issue 7682 public void ReplacesSubscriptionOfUniverseRemovedAndReAddedInSameTimeStep() { var dataPermissionManager = new DataPermissionManager(); @@ -299,7 +298,6 @@ public void ReplacesSubscriptionOfUniverseRemovedAndReAddedInSameTimeStep() } [Test] - // reproduces GH issue 7682 public void OptionUniverseCanBeRemovedAndReAddedInSameTimeStep() { AlgorithmRunner.RunLocalBacktest("OptionUniverseRemovedAndReAddedRegressionAlgorithm", From 07e256427fd74a0e173ab886883b161630817e99 Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Fri, 17 Jul 2026 11:14:23 -0400 Subject: [PATCH 3/8] Use RegressionTestException in regression algorithm --- .../OptionUniverseRemovedAndReAddedRegressionAlgorithm.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/Algorithm.Python/OptionUniverseRemovedAndReAddedRegressionAlgorithm.py b/Algorithm.Python/OptionUniverseRemovedAndReAddedRegressionAlgorithm.py index ea66adb8e14d..19fc01c67517 100644 --- a/Algorithm.Python/OptionUniverseRemovedAndReAddedRegressionAlgorithm.py +++ b/Algorithm.Python/OptionUniverseRemovedAndReAddedRegressionAlgorithm.py @@ -53,8 +53,8 @@ def on_data(self, data): def on_end_of_algorithm(self): if not self._readded: - raise Exception("The option universe was never removed and re-added") + raise RegressionTestException("The option universe was never removed and re-added") if self._chains_before == 0: - raise Exception("Expected option chain data before the universe was removed") + raise RegressionTestException("Expected option chain data before the universe was removed") if self._chains_after == 0: - raise Exception("Expected option chain data after the universe was re-added") + raise RegressionTestException("Expected option chain data after the universe was re-added") From 434d503289878098f3adb34a7099d4929e7852db Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Fri, 17 Jul 2026 15:43:07 -0400 Subject: [PATCH 4/8] Clean up stale universe members when universe is removed and re-added in the same time step --- ...ReAddedMemberCleanupRegressionAlgorithm.py | 89 +++++++++++++++++++ Engine/DataFeeds/DataManager.cs | 8 ++ Engine/DataFeeds/SubscriptionSynchronizer.cs | 16 ++++ Engine/DataFeeds/UniverseSelection.cs | 22 +++++ Tests/Engine/DataFeeds/DataManagerTests.cs | 40 +++++++++ 5 files changed, 175 insertions(+) create mode 100644 Algorithm.Python/OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm.py diff --git a/Algorithm.Python/OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm.py b/Algorithm.Python/OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm.py new file mode 100644 index 000000000000..7ce51be18433 --- /dev/null +++ b/Algorithm.Python/OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm.py @@ -0,0 +1,89 @@ +# 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 * + +### +### Regression algorithm asserting that when an option universe is removed and one for the same +### underlying is re-added in the same time step, the previously selected contracts that are not +### re-selected by the new universe are properly cleaned up: their subscriptions are removed and +### removed security changes are emitted for them. +### +class OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm(QCAlgorithm): + + def initialize(self): + self.set_start_date(2014, 6, 6) + self.set_end_date(2014, 6, 9) + self.set_cash(100000) + + # Wide filter: several strikes and expirations will be selected + self._canonical = self._add_aapl_option(lambda u: u.strikes(-2, 2).expiration(0, 180)) + self._readded = False + self._checked = False + self._old_members = [] + self._removed_symbols = set() + + def _add_aapl_option(self, filter_func) -> Symbol: + option = self.add_option("AAPL", Resolution.MINUTE) + option.set_filter(filter_func) + return option.symbol + + def on_securities_changed(self, changes): + for security in changes.removed_securities: + self._removed_symbols.add(security.symbol) + + def on_data(self, data): + if not self._readded: + if self.time.hour < 10: + return + universe = self.universe_manager[self._canonical] + if universe.members.count == 0: + return + + self._old_members = [kvp.key for kvp in universe.members if not kvp.key.is_canonical()] + + # Remove the universe and re-add it with a narrower filter in the same time step: + # most of the previously selected contracts will not be re-selected by the new universe + self.remove_security(self._canonical) + self._canonical = self._add_aapl_option(lambda u: u.strikes(0, 0).expiration(0, 30)) + self._readded = True + elif not self._checked and (self.time.hour > 10 or self.time.minute >= 30): + self._checked = True + self._assert_old_members_cleaned_up() + + def _assert_old_members_cleaned_up(self): + universe = self.universe_manager[self._canonical] + current_members = set(kvp.key for kvp in universe.members) + subscribed = set(config.symbol for config in self.subscription_manager.subscriptions) + + not_reselected = [s for s in self._old_members if s not in current_members] + if len(not_reselected) == 0: + raise RegressionTestException("Expected some previously selected contracts to not be re-selected") + + still_subscribed = [s for s in not_reselected if s in subscribed] + if len(still_subscribed) > 0: + raise RegressionTestException( + f"Expected the subscriptions of the {len(not_reselected)} deselected contracts to be removed, " + f"but {len(still_subscribed)} are still subscribed, e.g. {[str(s) for s in still_subscribed[:3]]}") + + missing_removed_events = [s for s in not_reselected if s not in self._removed_symbols] + if len(missing_removed_events) > 0: + raise RegressionTestException( + f"Expected removed security changes for the {len(not_reselected)} deselected contracts, " + f"but {len(missing_removed_events)} were not notified, e.g. {[str(s) for s in missing_removed_events[:3]]}") + + def on_end_of_algorithm(self): + if not self._readded: + raise RegressionTestException("The option universe was never removed and re-added") + if not self._checked: + raise RegressionTestException("The clean up assertions were never performed") diff --git a/Engine/DataFeeds/DataManager.cs b/Engine/DataFeeds/DataManager.cs index d86f55b0060e..a280ccfd3426 100644 --- a/Engine/DataFeeds/DataManager.cs +++ b/Engine/DataFeeds/DataManager.cs @@ -299,6 +299,7 @@ public bool AddSubscription(SubscriptionRequest request) // a universe was removed and a new one with the same configuration was added in the same // time step. The old subscription removal is performed asynchronously by the synchronizer, // so we force it out now and carry on creating a new subscription for the new universe + var staleUniverses = subscription.Universes.ToList(); RemoveSubscriptionInternal(request.Configuration, universe: null, forceSubscriptionRemoval: true); // the removal above unregistered the configuration, which the new subscription requires @@ -309,6 +310,13 @@ public bool AddSubscription(SubscriptionRequest request) _subscriptionDataConfigsEnumerator = null; } } + + // the synchronizer will not see the removed subscription, so we schedule the final + // selection it would have triggered for the stale universes to deselect their members + foreach (var staleUniverse in staleUniverses) + { + UniverseSelection.ScheduleFinalSelection(staleUniverse); + } } else { diff --git a/Engine/DataFeeds/SubscriptionSynchronizer.cs b/Engine/DataFeeds/SubscriptionSynchronizer.cs index 06c062270208..4ceee2045c4a 100644 --- a/Engine/DataFeeds/SubscriptionSynchronizer.cs +++ b/Engine/DataFeeds/SubscriptionSynchronizer.cs @@ -222,6 +222,22 @@ public IEnumerable Sync(IEnumerable subscriptions, } _perfTrackingTool.Stop(PerformanceTarget.Subscriptions); + // process any disposed universe whose subscription was removed before we could trigger + // its last selection above, e.g. a universe removed and re-added in the same time step, + // so that it deselects all the members it added + Universe pendingDisposedUniverse; + while (_universeSelection.TryGetPendingFinalSelection(out pendingDisposedUniverse)) + { + if (universeData == null) + { + universeData = new Dictionary(); + } + if (!universeData.ContainsKey(pendingDisposedUniverse)) + { + universeData[pendingDisposedUniverse] = new BaseDataCollection(frontierUtc, pendingDisposedUniverse.Configuration.Symbol); + } + } + if (universeData != null && universeData.Count > 0) { // if we are going to perform universe selection we emit an empty diff --git a/Engine/DataFeeds/UniverseSelection.cs b/Engine/DataFeeds/UniverseSelection.cs index e5512ba46122..98152337c500 100644 --- a/Engine/DataFeeds/UniverseSelection.cs +++ b/Engine/DataFeeds/UniverseSelection.cs @@ -14,6 +14,7 @@ */ using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; using QuantConnect.Benchmarks; @@ -42,6 +43,7 @@ public class UniverseSelection private bool _initializedSecurityBenchmark; private bool _anyDoesNotHaveFundamentalDataWarningLogged; private readonly SecurityChangesConstructor _securityChangesConstructor; + private readonly ConcurrentQueue _pendingDisposedUniverseSelections = new(); /// /// Initializes a new instance of the class @@ -92,6 +94,26 @@ public void SetDataManager(IDataFeedSubscriptionManager dataManager) }; } + /// + /// Schedules a final selection for a disposed universe whose subscription was removed + /// before the synchronizer could trigger it, so that it deselects all the members it added + /// + /// The disposed universe requiring a final selection + public void ScheduleFinalSelection(Universe universe) + { + _pendingDisposedUniverseSelections.Enqueue(universe); + } + + /// + /// Gets the next disposed universe with a pending scheduled final selection, if any + /// + /// The next disposed universe with a pending final selection + /// True if there was a pending disposed universe final selection + public bool TryGetPendingFinalSelection(out Universe universe) + { + return _pendingDisposedUniverseSelections.TryDequeue(out universe); + } + /// /// Applies universe selection the the data feed and algorithm /// diff --git a/Tests/Engine/DataFeeds/DataManagerTests.cs b/Tests/Engine/DataFeeds/DataManagerTests.cs index 3cc4620815ef..b1fba1c4061e 100644 --- a/Tests/Engine/DataFeeds/DataManagerTests.cs +++ b/Tests/Engine/DataFeeds/DataManagerTests.cs @@ -337,6 +337,46 @@ public void OptionUniverseCanBeRemovedAndReAddedInSameTimeStep() algorithmLocation: "../../../Algorithm.Python/OptionUniverseRemovedAndReAddedRegressionAlgorithm.py"); } + [Test] + public void UniverseMembersAreCleanedUpWhenUniverseIsRemovedAndReAddedInSameTimeStep() + { + AlgorithmRunner.RunLocalBacktest("OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm", + new Dictionary + { + {"Total Orders", "0"}, + {"Average Win", "0%"}, + {"Average Loss", "0%"}, + {"Compounding Annual Return", "0%"}, + {"Drawdown", "0%"}, + {"Expectancy", "0"}, + {"Start Equity", "100000"}, + {"End Equity", "100000"}, + {"Net Profit", "0%"}, + {"Sharpe Ratio", "0"}, + {"Sortino Ratio", "0"}, + {"Probabilistic Sharpe Ratio", "0%"}, + {"Loss Rate", "0%"}, + {"Win Rate", "0%"}, + {"Profit-Loss Ratio", "0"}, + {"Alpha", "0"}, + {"Beta", "0"}, + {"Annual Standard Deviation", "0"}, + {"Annual Variance", "0"}, + {"Information Ratio", "-9.486"}, + {"Tracking Error", "0.008"}, + {"Treynor Ratio", "0"}, + {"Total Fees", "$0.00"}, + {"Estimated Strategy Capacity", "$0"}, + {"Lowest Capacity Asset", ""}, + {"Portfolio Turnover", "0%"}, + {"Drawdown Recovery", "0"}, + {"OrderListHash", "d41d8cd98f00b204e9800998ecf8427e"} + }, + Language.Python, + AlgorithmStatus.Completed, + algorithmLocation: "../../../Algorithm.Python/OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm.py"); + } + [Test] public void ScaledRawNormalizationModeIsNotAllowed() { From 8033bcdcd84985a677f719efe3b8763306b51938 Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Fri, 17 Jul 2026 16:03:22 -0400 Subject: [PATCH 5/8] Convert regression algorithms to C# and centralize universe selection data handling --- ...ReAddedMemberCleanupRegressionAlgorithm.cs | 192 ++++++++++++++++++ ...rseRemovedAndReAddedRegressionAlgorithm.cs | 152 ++++++++++++++ ...ReAddedMemberCleanupRegressionAlgorithm.py | 89 -------- ...rseRemovedAndReAddedRegressionAlgorithm.py | 60 ------ Engine/DataFeeds/DataManager.cs | 7 +- Engine/DataFeeds/SubscriptionSynchronizer.cs | 63 +++--- Tests/Engine/DataFeeds/DataManagerTests.cs | 80 -------- 7 files changed, 380 insertions(+), 263 deletions(-) create mode 100644 Algorithm.CSharp/OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm.cs create mode 100644 Algorithm.CSharp/OptionUniverseRemovedAndReAddedRegressionAlgorithm.cs delete mode 100644 Algorithm.Python/OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm.py delete mode 100644 Algorithm.Python/OptionUniverseRemovedAndReAddedRegressionAlgorithm.py diff --git a/Algorithm.CSharp/OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm.cs b/Algorithm.CSharp/OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm.cs new file mode 100644 index 000000000000..f385de09d78c --- /dev/null +++ b/Algorithm.CSharp/OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm.cs @@ -0,0 +1,192 @@ +/* + * 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. +*/ + +using System.Collections.Generic; +using System.Linq; +using QuantConnect.Data; +using QuantConnect.Data.UniverseSelection; +using QuantConnect.Interfaces; + +namespace QuantConnect.Algorithm.CSharp +{ + /// + /// Regression algorithm asserting that when an option universe is removed and one for the same + /// underlying is re-added in the same time step, the previously selected contracts that are not + /// re-selected by the new universe are properly cleaned up: their subscriptions are removed and + /// removed security changes are emitted for them. + /// + public class OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm : QCAlgorithm, IRegressionAlgorithmDefinition + { + private Symbol _canonical; + private bool _readded; + private bool _checked; + private List _oldMembers; + private HashSet _removedSymbols; + + public override void Initialize() + { + SetStartDate(2014, 6, 6); + SetEndDate(2014, 6, 9); + SetCash(100000); + + _removedSymbols = new HashSet(); + + // Wide filter: several strikes and expirations will be selected + var option = AddOption("AAPL", Resolution.Minute); + option.SetFilter(-2, 2, 0, 180); + _canonical = option.Symbol; + } + + public override void OnSecuritiesChanged(SecurityChanges changes) + { + foreach (var security in changes.RemovedSecurities) + { + _removedSymbols.Add(security.Symbol); + } + } + + public override void OnData(Slice slice) + { + if (!_readded) + { + if (Time.Hour < 10) + { + return; + } + + var universe = UniverseManager[_canonical]; + if (universe.Members.Count == 0) + { + return; + } + + _oldMembers = universe.Members.Keys.Where(symbol => !symbol.IsCanonical()).ToList(); + + // Remove the universe and re-add it with a narrower filter in the same time step: + // most of the previously selected contracts will not be re-selected by the new universe + RemoveSecurity(_canonical); + var option = AddOption("AAPL", Resolution.Minute); + option.SetFilter(universeFilter => universeFilter.Strikes(0, 0).Expiration(0, 30)); + _canonical = option.Symbol; + _readded = true; + } + else if (!_checked && (Time.Hour > 10 || Time.Minute >= 30)) + { + _checked = true; + AssertOldMembersCleanedUp(); + } + } + + private void AssertOldMembersCleanedUp() + { + var currentMembers = UniverseManager[_canonical].Members.Keys.ToHashSet(); + var subscribed = SubscriptionManager.Subscriptions.Select(config => config.Symbol).ToHashSet(); + + var notReselected = _oldMembers.Where(symbol => !currentMembers.Contains(symbol)).ToList(); + if (notReselected.Count == 0) + { + throw new RegressionTestException("Expected some previously selected contracts to not be re-selected"); + } + + var stillSubscribed = notReselected.Where(subscribed.Contains).ToList(); + if (stillSubscribed.Count > 0) + { + throw new RegressionTestException( + $"Expected the subscriptions of the {notReselected.Count} deselected contracts to be removed, " + + $"but {stillSubscribed.Count} are still subscribed, e.g. {string.Join(", ", stillSubscribed.Take(3))}"); + } + + var missingRemovedEvents = notReselected.Where(symbol => !_removedSymbols.Contains(symbol)).ToList(); + if (missingRemovedEvents.Count > 0) + { + throw new RegressionTestException( + $"Expected removed security changes for the {notReselected.Count} deselected contracts, " + + $"but {missingRemovedEvents.Count} were not notified, e.g. {string.Join(", ", missingRemovedEvents.Take(3))}"); + } + } + + public override void OnEndOfAlgorithm() + { + if (!_readded) + { + throw new RegressionTestException("The option universe was never removed and re-added"); + } + if (!_checked) + { + throw new RegressionTestException("The clean up assertions were never performed"); + } + } + + /// + /// This is used by the regression test system to indicate if the open source Lean repository has the required data to run this algorithm. + /// + public bool CanRunLocally { get; } = true; + + /// + /// This is used by the regression test system to indicate which languages this algorithm is written in. + /// + public List Languages { get; } = new() { Language.CSharp }; + + /// + /// Data Points count of all timeslices of algorithm + /// + public long DataPoints => 34112; + + /// + /// Data Points count of the algorithm history + /// + public int AlgorithmHistoryDataPoints => 0; + + /// + /// Final status of the algorithm + /// + public AlgorithmStatus AlgorithmStatus => AlgorithmStatus.Completed; + + /// + /// This is used by the regression test system to indicate what the expected statistics are from running the algorithm + /// + public Dictionary ExpectedStatistics => new Dictionary + { + {"Total Orders", "0"}, + {"Average Win", "0%"}, + {"Average Loss", "0%"}, + {"Compounding Annual Return", "0%"}, + {"Drawdown", "0%"}, + {"Expectancy", "0"}, + {"Start Equity", "100000"}, + {"End Equity", "100000"}, + {"Net Profit", "0%"}, + {"Sharpe Ratio", "0"}, + {"Sortino Ratio", "0"}, + {"Probabilistic Sharpe Ratio", "0%"}, + {"Loss Rate", "0%"}, + {"Win Rate", "0%"}, + {"Profit-Loss Ratio", "0"}, + {"Alpha", "0"}, + {"Beta", "0"}, + {"Annual Standard Deviation", "0"}, + {"Annual Variance", "0"}, + {"Information Ratio", "-9.486"}, + {"Tracking Error", "0.008"}, + {"Treynor Ratio", "0"}, + {"Total Fees", "$0.00"}, + {"Estimated Strategy Capacity", "$0"}, + {"Lowest Capacity Asset", ""}, + {"Portfolio Turnover", "0%"}, + {"Drawdown Recovery", "0"}, + {"OrderListHash", "d41d8cd98f00b204e9800998ecf8427e"} + }; + } +} diff --git a/Algorithm.CSharp/OptionUniverseRemovedAndReAddedRegressionAlgorithm.cs b/Algorithm.CSharp/OptionUniverseRemovedAndReAddedRegressionAlgorithm.cs new file mode 100644 index 000000000000..338963c60aa5 --- /dev/null +++ b/Algorithm.CSharp/OptionUniverseRemovedAndReAddedRegressionAlgorithm.cs @@ -0,0 +1,152 @@ +/* + * 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. +*/ + +using System.Collections.Generic; +using System.Linq; +using QuantConnect.Data; +using QuantConnect.Interfaces; + +namespace QuantConnect.Algorithm.CSharp +{ + /// + /// Regression algorithm asserting that removing an option universe and re-adding one for the + /// same underlying within the same time step does not throw, and that the re-added universe + /// keeps providing option chain data. + /// + public class OptionUniverseRemovedAndReAddedRegressionAlgorithm : QCAlgorithm, IRegressionAlgorithmDefinition + { + private Symbol _canonical; + private bool _readded; + private int _chainsBefore; + private int _chainsAfter; + + public override void Initialize() + { + SetStartDate(2014, 6, 6); + SetEndDate(2014, 6, 9); + SetCash(100000); + + _canonical = AddAaplOption(); + } + + private Symbol AddAaplOption() + { + var option = AddOption("AAPL", Resolution.Minute); + option.SetFilter(-2, 2, 0, 180); + return option.Symbol; + } + + public override void OnData(Slice slice) + { + var hasChain = slice.OptionChains.TryGetValue(_canonical, out var chain) && chain.Any(); + + if (!_readded) + { + if (hasChain) + { + _chainsBefore++; + } + + if (Time.Hour >= 10) + { + // Remove the option universe and re-add one for the same underlying in the same time step + RemoveSecurity(_canonical); + _canonical = AddAaplOption(); + _readded = true; + } + } + else if (hasChain) + { + _chainsAfter++; + } + } + + public override void OnEndOfAlgorithm() + { + if (!_readded) + { + throw new RegressionTestException("The option universe was never removed and re-added"); + } + if (_chainsBefore == 0) + { + throw new RegressionTestException("Expected option chain data before the universe was removed"); + } + if (_chainsAfter == 0) + { + throw new RegressionTestException("Expected option chain data after the universe was re-added"); + } + } + + /// + /// This is used by the regression test system to indicate if the open source Lean repository has the required data to run this algorithm. + /// + public bool CanRunLocally { get; } = true; + + /// + /// This is used by the regression test system to indicate which languages this algorithm is written in. + /// + public List Languages { get; } = new() { Language.CSharp }; + + /// + /// Data Points count of all timeslices of algorithm + /// + public long DataPoints => 108539; + + /// + /// Data Points count of the algorithm history + /// + public int AlgorithmHistoryDataPoints => 0; + + /// + /// Final status of the algorithm + /// + public AlgorithmStatus AlgorithmStatus => AlgorithmStatus.Completed; + + /// + /// This is used by the regression test system to indicate what the expected statistics are from running the algorithm + /// + public Dictionary ExpectedStatistics => new Dictionary + { + {"Total Orders", "0"}, + {"Average Win", "0%"}, + {"Average Loss", "0%"}, + {"Compounding Annual Return", "0%"}, + {"Drawdown", "0%"}, + {"Expectancy", "0"}, + {"Start Equity", "100000"}, + {"End Equity", "100000"}, + {"Net Profit", "0%"}, + {"Sharpe Ratio", "0"}, + {"Sortino Ratio", "0"}, + {"Probabilistic Sharpe Ratio", "0%"}, + {"Loss Rate", "0%"}, + {"Win Rate", "0%"}, + {"Profit-Loss Ratio", "0"}, + {"Alpha", "0"}, + {"Beta", "0"}, + {"Annual Standard Deviation", "0"}, + {"Annual Variance", "0"}, + {"Information Ratio", "-9.486"}, + {"Tracking Error", "0.008"}, + {"Treynor Ratio", "0"}, + {"Total Fees", "$0.00"}, + {"Estimated Strategy Capacity", "$0"}, + {"Lowest Capacity Asset", ""}, + {"Portfolio Turnover", "0%"}, + {"Drawdown Recovery", "0"}, + {"OrderListHash", "d41d8cd98f00b204e9800998ecf8427e"} + }; + } +} diff --git a/Algorithm.Python/OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm.py b/Algorithm.Python/OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm.py deleted file mode 100644 index 7ce51be18433..000000000000 --- a/Algorithm.Python/OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm.py +++ /dev/null @@ -1,89 +0,0 @@ -# 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 * - -### -### Regression algorithm asserting that when an option universe is removed and one for the same -### underlying is re-added in the same time step, the previously selected contracts that are not -### re-selected by the new universe are properly cleaned up: their subscriptions are removed and -### removed security changes are emitted for them. -### -class OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm(QCAlgorithm): - - def initialize(self): - self.set_start_date(2014, 6, 6) - self.set_end_date(2014, 6, 9) - self.set_cash(100000) - - # Wide filter: several strikes and expirations will be selected - self._canonical = self._add_aapl_option(lambda u: u.strikes(-2, 2).expiration(0, 180)) - self._readded = False - self._checked = False - self._old_members = [] - self._removed_symbols = set() - - def _add_aapl_option(self, filter_func) -> Symbol: - option = self.add_option("AAPL", Resolution.MINUTE) - option.set_filter(filter_func) - return option.symbol - - def on_securities_changed(self, changes): - for security in changes.removed_securities: - self._removed_symbols.add(security.symbol) - - def on_data(self, data): - if not self._readded: - if self.time.hour < 10: - return - universe = self.universe_manager[self._canonical] - if universe.members.count == 0: - return - - self._old_members = [kvp.key for kvp in universe.members if not kvp.key.is_canonical()] - - # Remove the universe and re-add it with a narrower filter in the same time step: - # most of the previously selected contracts will not be re-selected by the new universe - self.remove_security(self._canonical) - self._canonical = self._add_aapl_option(lambda u: u.strikes(0, 0).expiration(0, 30)) - self._readded = True - elif not self._checked and (self.time.hour > 10 or self.time.minute >= 30): - self._checked = True - self._assert_old_members_cleaned_up() - - def _assert_old_members_cleaned_up(self): - universe = self.universe_manager[self._canonical] - current_members = set(kvp.key for kvp in universe.members) - subscribed = set(config.symbol for config in self.subscription_manager.subscriptions) - - not_reselected = [s for s in self._old_members if s not in current_members] - if len(not_reselected) == 0: - raise RegressionTestException("Expected some previously selected contracts to not be re-selected") - - still_subscribed = [s for s in not_reselected if s in subscribed] - if len(still_subscribed) > 0: - raise RegressionTestException( - f"Expected the subscriptions of the {len(not_reselected)} deselected contracts to be removed, " - f"but {len(still_subscribed)} are still subscribed, e.g. {[str(s) for s in still_subscribed[:3]]}") - - missing_removed_events = [s for s in not_reselected if s not in self._removed_symbols] - if len(missing_removed_events) > 0: - raise RegressionTestException( - f"Expected removed security changes for the {len(not_reselected)} deselected contracts, " - f"but {len(missing_removed_events)} were not notified, e.g. {[str(s) for s in missing_removed_events[:3]]}") - - def on_end_of_algorithm(self): - if not self._readded: - raise RegressionTestException("The option universe was never removed and re-added") - if not self._checked: - raise RegressionTestException("The clean up assertions were never performed") diff --git a/Algorithm.Python/OptionUniverseRemovedAndReAddedRegressionAlgorithm.py b/Algorithm.Python/OptionUniverseRemovedAndReAddedRegressionAlgorithm.py deleted file mode 100644 index 19fc01c67517..000000000000 --- a/Algorithm.Python/OptionUniverseRemovedAndReAddedRegressionAlgorithm.py +++ /dev/null @@ -1,60 +0,0 @@ -# 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 * - -### -### Regression algorithm asserting that removing an option universe and re-adding one for the -### same underlying within the same time step does not throw, and that the re-added universe -### keeps providing option chain data. -### -class OptionUniverseRemovedAndReAddedRegressionAlgorithm(QCAlgorithm): - - def initialize(self): - self.set_start_date(2014, 6, 6) - self.set_end_date(2014, 6, 9) - self.set_cash(100000) - - self._canonical = self._add_aapl_option() - self._readded = False - self._chains_before = 0 - self._chains_after = 0 - - def _add_aapl_option(self) -> Symbol: - option = self.add_option("AAPL", Resolution.MINUTE) - option.set_filter(-2, 2, 0, 180) - return option.symbol - - def on_data(self, data): - chain = data.option_chains.get(self._canonical) - has_chain = chain is not None and len(chain) > 0 - - if not self._readded: - if has_chain: - self._chains_before += 1 - - if self.time.hour >= 10: - # Remove the option universe and re-add one for the same underlying in the same time step - self.remove_security(self._canonical) - self._canonical = self._add_aapl_option() - self._readded = True - elif has_chain: - self._chains_after += 1 - - def on_end_of_algorithm(self): - if not self._readded: - raise RegressionTestException("The option universe was never removed and re-added") - if self._chains_before == 0: - raise RegressionTestException("Expected option chain data before the universe was removed") - if self._chains_after == 0: - raise RegressionTestException("Expected option chain data after the universe was re-added") diff --git a/Engine/DataFeeds/DataManager.cs b/Engine/DataFeeds/DataManager.cs index a280ccfd3426..536fa447ac0b 100644 --- a/Engine/DataFeeds/DataManager.cs +++ b/Engine/DataFeeds/DataManager.cs @@ -287,8 +287,7 @@ public bool AddSubscription(SubscriptionRequest request) } } - Subscription subscription; - if (DataFeedSubscriptions.TryGetValue(request.Configuration, out subscription)) + if (DataFeedSubscriptions.TryGetValue(request.Configuration, out var subscription)) { if (!subscription.EndOfStream) { @@ -390,9 +389,7 @@ public bool RemoveSubscription(SubscriptionDataConfig configuration, Universe un private bool RemoveSubscriptionInternal(SubscriptionDataConfig configuration, Universe universe, bool forceSubscriptionRemoval) { // remove the subscription from our collection, if it exists - Subscription subscription; - - if (DataFeedSubscriptions.TryGetValue(configuration, out subscription)) + if (DataFeedSubscriptions.TryGetValue(configuration, out var subscription)) { // we remove the subscription when there are no other requests left if (subscription.RemoveSubscriptionRequest(universe)) diff --git a/Engine/DataFeeds/SubscriptionSynchronizer.cs b/Engine/DataFeeds/SubscriptionSynchronizer.cs index 4ceee2045c4a..e6e1404f9b85 100644 --- a/Engine/DataFeeds/SubscriptionSynchronizer.cs +++ b/Engine/DataFeeds/SubscriptionSynchronizer.cs @@ -183,20 +183,16 @@ public IEnumerable Sync(IEnumerable subscriptions, ? packet.Data : packetBaseDataCollection.Data; - BaseDataCollection collection; if (universeData != null - && universeData.TryGetValue(subscription.Universes.Single(), out collection)) + && universeData.TryGetValue(subscription.Universes.Single(), out var collection)) { collection.AddRange(packetData); } else { - collection = new BaseDataCollection(frontierUtc, frontierUtc, subscription.Configuration.Symbol, packetData, packetBaseDataCollection?.Underlying, packetBaseDataCollection?.FilteredContracts); - if (universeData == null) - { - universeData = new Dictionary(); - } - universeData[subscription.Universes.Single()] = collection; + AddUniverseData(ref universeData, subscription.Universes.Single(), + new BaseDataCollection(frontierUtc, frontierUtc, subscription.Configuration.Symbol, packetData, + packetBaseDataCollection?.Underlying, packetBaseDataCollection?.FilteredContracts)); } } } @@ -204,17 +200,7 @@ public IEnumerable Sync(IEnumerable subscriptions, if (subscription.IsUniverseSelectionSubscription && subscription.Universes.Single().DisposeRequested) { - var universe = subscription.Universes.Single(); - // check if a universe selection isn't already scheduled for this disposed universe - if (universeData == null || !universeData.ContainsKey(universe)) - { - if (universeData == null) - { - universeData = new Dictionary(); - } - // we force trigger one last universe selection for this disposed universe, so it deselects all subscriptions it added - universeData[universe] = new BaseDataCollection(frontierUtc, subscription.Configuration.Symbol); - } + ScheduleDisposedUniverseFinalSelection(ref universeData, subscription.Universes.Single(), frontierUtc); // we need to do this after all usages of subscription.Universes OnSubscriptionFinished(subscription); @@ -225,17 +211,9 @@ public IEnumerable Sync(IEnumerable subscriptions, // process any disposed universe whose subscription was removed before we could trigger // its last selection above, e.g. a universe removed and re-added in the same time step, // so that it deselects all the members it added - Universe pendingDisposedUniverse; - while (_universeSelection.TryGetPendingFinalSelection(out pendingDisposedUniverse)) + while (_universeSelection.TryGetPendingFinalSelection(out var pendingDisposedUniverse)) { - if (universeData == null) - { - universeData = new Dictionary(); - } - if (!universeData.ContainsKey(pendingDisposedUniverse)) - { - universeData[pendingDisposedUniverse] = new BaseDataCollection(frontierUtc, pendingDisposedUniverse.Configuration.Symbol); - } + ScheduleDisposedUniverseFinalSelection(ref universeData, pendingDisposedUniverse, frontierUtc); } if (universeData != null && universeData.Count > 0) @@ -279,6 +257,33 @@ public IEnumerable Sync(IEnumerable subscriptions, } } + /// + /// Schedules one last universe selection for the given disposed universe, if not already scheduled, + /// so that it deselects all the members it added + /// + private static void ScheduleDisposedUniverseFinalSelection(ref Dictionary universeData, + Universe universe, DateTime frontierUtc) + { + // check if a universe selection isn't already scheduled for this disposed universe + if (universeData == null || !universeData.ContainsKey(universe)) + { + AddUniverseData(ref universeData, universe, new BaseDataCollection(frontierUtc, universe.Configuration.Symbol)); + } + } + + /// + /// Adds the given universe data collection to perform selection on, lazily creating the holding dictionary + /// + private static void AddUniverseData(ref Dictionary universeData, + Universe universe, BaseDataCollection collection) + { + if (universeData == null) + { + universeData = new Dictionary(); + } + universeData[universe] = collection; + } + /// /// Event invocator for the event /// diff --git a/Tests/Engine/DataFeeds/DataManagerTests.cs b/Tests/Engine/DataFeeds/DataManagerTests.cs index b1fba1c4061e..7a383ef68c6a 100644 --- a/Tests/Engine/DataFeeds/DataManagerTests.cs +++ b/Tests/Engine/DataFeeds/DataManagerTests.cs @@ -297,86 +297,6 @@ public void ReplacesSubscriptionOfUniverseRemovedAndReAddedInSameTimeStep() dataManager.RemoveAllSubscriptions(); } - [Test] - public void OptionUniverseCanBeRemovedAndReAddedInSameTimeStep() - { - AlgorithmRunner.RunLocalBacktest("OptionUniverseRemovedAndReAddedRegressionAlgorithm", - new Dictionary - { - {"Total Orders", "0"}, - {"Average Win", "0%"}, - {"Average Loss", "0%"}, - {"Compounding Annual Return", "0%"}, - {"Drawdown", "0%"}, - {"Expectancy", "0"}, - {"Start Equity", "100000"}, - {"End Equity", "100000"}, - {"Net Profit", "0%"}, - {"Sharpe Ratio", "0"}, - {"Sortino Ratio", "0"}, - {"Probabilistic Sharpe Ratio", "0%"}, - {"Loss Rate", "0%"}, - {"Win Rate", "0%"}, - {"Profit-Loss Ratio", "0"}, - {"Alpha", "0"}, - {"Beta", "0"}, - {"Annual Standard Deviation", "0"}, - {"Annual Variance", "0"}, - {"Information Ratio", "-9.486"}, - {"Tracking Error", "0.008"}, - {"Treynor Ratio", "0"}, - {"Total Fees", "$0.00"}, - {"Estimated Strategy Capacity", "$0"}, - {"Lowest Capacity Asset", ""}, - {"Portfolio Turnover", "0%"}, - {"Drawdown Recovery", "0"}, - {"OrderListHash", "d41d8cd98f00b204e9800998ecf8427e"} - }, - Language.Python, - AlgorithmStatus.Completed, - algorithmLocation: "../../../Algorithm.Python/OptionUniverseRemovedAndReAddedRegressionAlgorithm.py"); - } - - [Test] - public void UniverseMembersAreCleanedUpWhenUniverseIsRemovedAndReAddedInSameTimeStep() - { - AlgorithmRunner.RunLocalBacktest("OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm", - new Dictionary - { - {"Total Orders", "0"}, - {"Average Win", "0%"}, - {"Average Loss", "0%"}, - {"Compounding Annual Return", "0%"}, - {"Drawdown", "0%"}, - {"Expectancy", "0"}, - {"Start Equity", "100000"}, - {"End Equity", "100000"}, - {"Net Profit", "0%"}, - {"Sharpe Ratio", "0"}, - {"Sortino Ratio", "0"}, - {"Probabilistic Sharpe Ratio", "0%"}, - {"Loss Rate", "0%"}, - {"Win Rate", "0%"}, - {"Profit-Loss Ratio", "0"}, - {"Alpha", "0"}, - {"Beta", "0"}, - {"Annual Standard Deviation", "0"}, - {"Annual Variance", "0"}, - {"Information Ratio", "-9.486"}, - {"Tracking Error", "0.008"}, - {"Treynor Ratio", "0"}, - {"Total Fees", "$0.00"}, - {"Estimated Strategy Capacity", "$0"}, - {"Lowest Capacity Asset", ""}, - {"Portfolio Turnover", "0%"}, - {"Drawdown Recovery", "0"}, - {"OrderListHash", "d41d8cd98f00b204e9800998ecf8427e"} - }, - Language.Python, - AlgorithmStatus.Completed, - algorithmLocation: "../../../Algorithm.Python/OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm.py"); - } - [Test] public void ScaledRawNormalizationModeIsNotAllowed() { From 5ed229461e893918b3fe7510c1d0327796a2cbf6 Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Fri, 17 Jul 2026 18:04:20 -0400 Subject: [PATCH 6/8] Simplify fix by parking colliding universe subscription requests and re-issuing them on removal --- ...ReAddedMemberCleanupRegressionAlgorithm.cs | 2 +- ...rseRemovedAndReAddedRegressionAlgorithm.cs | 2 +- Engine/DataFeeds/DataManager.cs | 72 +++++++++++-------- Engine/DataFeeds/Subscription.cs | 9 +-- Engine/DataFeeds/SubscriptionSynchronizer.cs | 59 +++++---------- Engine/DataFeeds/UniverseSelection.cs | 22 ------ Tests/Engine/DataFeeds/DataManagerTests.cs | 16 +++-- 7 files changed, 76 insertions(+), 106 deletions(-) diff --git a/Algorithm.CSharp/OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm.cs b/Algorithm.CSharp/OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm.cs index f385de09d78c..0f97e5695991 100644 --- a/Algorithm.CSharp/OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm.cs +++ b/Algorithm.CSharp/OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm.cs @@ -142,7 +142,7 @@ public override void OnEndOfAlgorithm() /// /// Data Points count of all timeslices of algorithm /// - public long DataPoints => 34112; + public long DataPoints => 34380; /// /// Data Points count of the algorithm history diff --git a/Algorithm.CSharp/OptionUniverseRemovedAndReAddedRegressionAlgorithm.cs b/Algorithm.CSharp/OptionUniverseRemovedAndReAddedRegressionAlgorithm.cs index 338963c60aa5..dd10c3c2d11e 100644 --- a/Algorithm.CSharp/OptionUniverseRemovedAndReAddedRegressionAlgorithm.cs +++ b/Algorithm.CSharp/OptionUniverseRemovedAndReAddedRegressionAlgorithm.cs @@ -102,7 +102,7 @@ public override void OnEndOfAlgorithm() /// /// Data Points count of all timeslices of algorithm /// - public long DataPoints => 108539; + public long DataPoints => 108983; /// /// Data Points count of the algorithm history diff --git a/Engine/DataFeeds/DataManager.cs b/Engine/DataFeeds/DataManager.cs index 536fa447ac0b..9154c3e14b12 100644 --- a/Engine/DataFeeds/DataManager.cs +++ b/Engine/DataFeeds/DataManager.cs @@ -49,6 +49,9 @@ public class DataManager : IAlgorithmSubscriptionManager, IDataFeedSubscriptionM /// so we use ConcurrentDictionary with byte value to minimize memory usage private readonly Dictionary _subscriptionManagerSubscriptions = new(); + /// Universe subscription requests that collided with a subscription pending removal, to be re-issued once it is removed + private readonly Dictionary _pendingUniverseSubscriptionRequests = new(); + /// /// Event fired when a new subscription is added /// @@ -255,6 +258,12 @@ public DataManager( /// public void RemoveAllSubscriptions() { + // drop any parked universe subscription request, we don't want to re-issue them while tearing down + lock (_pendingUniverseSubscriptionRequests) + { + _pendingUniverseSubscriptionRequests.Clear(); + } + // remove each subscription from our collection foreach (var subscription in DataFeedSubscriptions) { @@ -287,7 +296,8 @@ public bool AddSubscription(SubscriptionRequest request) } } - if (DataFeedSubscriptions.TryGetValue(request.Configuration, out var subscription)) + Subscription subscription; + if (DataFeedSubscriptions.TryGetValue(request.Configuration, out subscription)) { if (!subscription.EndOfStream) { @@ -296,39 +306,22 @@ public bool AddSubscription(SubscriptionRequest request) && subscription.Universes.All(universe => universe.DisposeRequested)) { // a universe was removed and a new one with the same configuration was added in the same - // time step. The old subscription removal is performed asynchronously by the synchronizer, - // so we force it out now and carry on creating a new subscription for the new universe - var staleUniverses = subscription.Universes.ToList(); - RemoveSubscriptionInternal(request.Configuration, universe: null, forceSubscriptionRemoval: true); - - // the removal above unregistered the configuration, which the new subscription requires - lock (_subscriptionManagerSubscriptions) - { - if (_subscriptionManagerSubscriptions.TryAdd(request.Configuration, request.Configuration)) - { - _subscriptionDataConfigsEnumerator = null; - } - } - - // the synchronizer will not see the removed subscription, so we schedule the final - // selection it would have triggered for the stale universes to deselect their members - foreach (var staleUniverse in staleUniverses) + // time step. The stale subscription will be removed by the synchronizer in the next loop, + // after performing one last selection for its disposed universes, so we park the new + // universe's request and will re-issue it once the stale subscription is actually removed + lock (_pendingUniverseSubscriptionRequests) { - UniverseSelection.ScheduleFinalSelection(staleUniverse); + _pendingUniverseSubscriptionRequests[request.Configuration] = request; } + return false; } - else - { - // duplicate subscription request - subscription.AddSubscriptionRequest(request); - // only result true if the existing subscription is internal, we actually added something from the users perspective - return subscription.Configuration.IsInternalFeed; - } - } - else - { - DataFeedSubscriptions.TryRemove(request.Configuration, out _); + + // duplicate subscription request + subscription.AddSubscriptionRequest(request); + // only result true if the existing subscription is internal, we actually added something from the users perspective + return subscription.Configuration.IsInternalFeed; } + DataFeedSubscriptions.TryRemove(request.Configuration, out _); } if (request.Configuration.DataNormalizationMode == DataNormalizationMode.ScaledRaw) @@ -389,7 +382,9 @@ public bool RemoveSubscription(SubscriptionDataConfig configuration, Universe un private bool RemoveSubscriptionInternal(SubscriptionDataConfig configuration, Universe universe, bool forceSubscriptionRemoval) { // remove the subscription from our collection, if it exists - if (DataFeedSubscriptions.TryGetValue(configuration, out var subscription)) + Subscription subscription; + + if (DataFeedSubscriptions.TryGetValue(configuration, out subscription)) { // we remove the subscription when there are no other requests left if (subscription.RemoveSubscriptionRequest(universe)) @@ -426,6 +421,21 @@ private bool RemoveSubscriptionInternal(SubscriptionDataConfig configuration, Un // this can be executed many times and its in the algorithm thread Log.Debug($"DataManager.RemoveSubscription(): Removed {configuration}"); } + + // if a new universe with this same configuration was added while this subscription's removal was + // pending, its subscription request was parked: the slot is now free, so we re-issue it. + // We keep the request's original start time, which was already adjusted to the previous tradable + // date when the universe was added, so that its first selection happens right away + SubscriptionRequest pendingRequest; + lock (_pendingUniverseSubscriptionRequests) + { + _pendingUniverseSubscriptionRequests.Remove(configuration, out pendingRequest); + } + if (pendingRequest != null && !pendingRequest.Universe.DisposeRequested) + { + AddSubscription(pendingRequest); + } + return true; } } diff --git a/Engine/DataFeeds/Subscription.cs b/Engine/DataFeeds/Subscription.cs index 3022fca327dd..1644353a374e 100644 --- a/Engine/DataFeeds/Subscription.cs +++ b/Engine/DataFeeds/Subscription.cs @@ -134,13 +134,8 @@ public bool AddSubscriptionRequest(SubscriptionRequest subscriptionRequest) if (IsUniverseSelectionSubscription || subscriptionRequest.IsUniverseSubscription) { - if (subscriptionRequest.Universe is UserDefinedUniverse) - { - // for different reasons a user defined universe can trigger a subscription request, likes additions/removals - return false; - } - throw new Exception("Subscription.AddSubscriptionRequest(): Universe selection" + - " subscriptions should not have more than 1 SubscriptionRequest"); + // for different reasons a user defined universe can trigger a subscription request, likes additions/removals + return false; } // this shouldn't happen but just in case.. diff --git a/Engine/DataFeeds/SubscriptionSynchronizer.cs b/Engine/DataFeeds/SubscriptionSynchronizer.cs index e6e1404f9b85..06c062270208 100644 --- a/Engine/DataFeeds/SubscriptionSynchronizer.cs +++ b/Engine/DataFeeds/SubscriptionSynchronizer.cs @@ -183,16 +183,20 @@ public IEnumerable Sync(IEnumerable subscriptions, ? packet.Data : packetBaseDataCollection.Data; + BaseDataCollection collection; if (universeData != null - && universeData.TryGetValue(subscription.Universes.Single(), out var collection)) + && universeData.TryGetValue(subscription.Universes.Single(), out collection)) { collection.AddRange(packetData); } else { - AddUniverseData(ref universeData, subscription.Universes.Single(), - new BaseDataCollection(frontierUtc, frontierUtc, subscription.Configuration.Symbol, packetData, - packetBaseDataCollection?.Underlying, packetBaseDataCollection?.FilteredContracts)); + collection = new BaseDataCollection(frontierUtc, frontierUtc, subscription.Configuration.Symbol, packetData, packetBaseDataCollection?.Underlying, packetBaseDataCollection?.FilteredContracts); + if (universeData == null) + { + universeData = new Dictionary(); + } + universeData[subscription.Universes.Single()] = collection; } } } @@ -200,7 +204,17 @@ public IEnumerable Sync(IEnumerable subscriptions, if (subscription.IsUniverseSelectionSubscription && subscription.Universes.Single().DisposeRequested) { - ScheduleDisposedUniverseFinalSelection(ref universeData, subscription.Universes.Single(), frontierUtc); + var universe = subscription.Universes.Single(); + // check if a universe selection isn't already scheduled for this disposed universe + if (universeData == null || !universeData.ContainsKey(universe)) + { + if (universeData == null) + { + universeData = new Dictionary(); + } + // we force trigger one last universe selection for this disposed universe, so it deselects all subscriptions it added + universeData[universe] = new BaseDataCollection(frontierUtc, subscription.Configuration.Symbol); + } // we need to do this after all usages of subscription.Universes OnSubscriptionFinished(subscription); @@ -208,14 +222,6 @@ public IEnumerable Sync(IEnumerable subscriptions, } _perfTrackingTool.Stop(PerformanceTarget.Subscriptions); - // process any disposed universe whose subscription was removed before we could trigger - // its last selection above, e.g. a universe removed and re-added in the same time step, - // so that it deselects all the members it added - while (_universeSelection.TryGetPendingFinalSelection(out var pendingDisposedUniverse)) - { - ScheduleDisposedUniverseFinalSelection(ref universeData, pendingDisposedUniverse, frontierUtc); - } - if (universeData != null && universeData.Count > 0) { // if we are going to perform universe selection we emit an empty @@ -257,33 +263,6 @@ public IEnumerable Sync(IEnumerable subscriptions, } } - /// - /// Schedules one last universe selection for the given disposed universe, if not already scheduled, - /// so that it deselects all the members it added - /// - private static void ScheduleDisposedUniverseFinalSelection(ref Dictionary universeData, - Universe universe, DateTime frontierUtc) - { - // check if a universe selection isn't already scheduled for this disposed universe - if (universeData == null || !universeData.ContainsKey(universe)) - { - AddUniverseData(ref universeData, universe, new BaseDataCollection(frontierUtc, universe.Configuration.Symbol)); - } - } - - /// - /// Adds the given universe data collection to perform selection on, lazily creating the holding dictionary - /// - private static void AddUniverseData(ref Dictionary universeData, - Universe universe, BaseDataCollection collection) - { - if (universeData == null) - { - universeData = new Dictionary(); - } - universeData[universe] = collection; - } - /// /// Event invocator for the event /// diff --git a/Engine/DataFeeds/UniverseSelection.cs b/Engine/DataFeeds/UniverseSelection.cs index 98152337c500..e5512ba46122 100644 --- a/Engine/DataFeeds/UniverseSelection.cs +++ b/Engine/DataFeeds/UniverseSelection.cs @@ -14,7 +14,6 @@ */ using System; -using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; using QuantConnect.Benchmarks; @@ -43,7 +42,6 @@ public class UniverseSelection private bool _initializedSecurityBenchmark; private bool _anyDoesNotHaveFundamentalDataWarningLogged; private readonly SecurityChangesConstructor _securityChangesConstructor; - private readonly ConcurrentQueue _pendingDisposedUniverseSelections = new(); /// /// Initializes a new instance of the class @@ -94,26 +92,6 @@ public void SetDataManager(IDataFeedSubscriptionManager dataManager) }; } - /// - /// Schedules a final selection for a disposed universe whose subscription was removed - /// before the synchronizer could trigger it, so that it deselects all the members it added - /// - /// The disposed universe requiring a final selection - public void ScheduleFinalSelection(Universe universe) - { - _pendingDisposedUniverseSelections.Enqueue(universe); - } - - /// - /// Gets the next disposed universe with a pending scheduled final selection, if any - /// - /// The next disposed universe with a pending final selection - /// True if there was a pending disposed universe final selection - public bool TryGetPendingFinalSelection(out Universe universe) - { - return _pendingDisposedUniverseSelections.TryDequeue(out universe); - } - /// /// Applies universe selection the the data feed and algorithm /// diff --git a/Tests/Engine/DataFeeds/DataManagerTests.cs b/Tests/Engine/DataFeeds/DataManagerTests.cs index 7a383ef68c6a..333e92ff6050 100644 --- a/Tests/Engine/DataFeeds/DataManagerTests.cs +++ b/Tests/Engine/DataFeeds/DataManagerTests.cs @@ -264,7 +264,8 @@ public void ReplacesSubscriptionOfUniverseRemovedAndReAddedInSameTimeStep() new DateTime(2019, 1, 1), new DateTime(2019, 1, 2)); using var oldEnumerator = new EnqueueableEnumerator(); - dataFeed.Subscription = new Subscription(oldRequest, oldEnumerator, null); + var oldSubscription = new Subscription(oldRequest, oldEnumerator, null); + dataFeed.Subscription = oldSubscription; Assert.IsTrue(dataManager.AddSubscription(oldRequest)); // the universe is removed: it gets disposed synchronously but its subscription removal @@ -285,10 +286,17 @@ public void ReplacesSubscriptionOfUniverseRemovedAndReAddedInSameTimeStep() var newSubscription = new Subscription(newRequest, newEnumerator, null); dataFeed.Subscription = newSubscription; - Assert.IsTrue(dataManager.AddSubscription(newRequest)); - - // the stale subscription was replaced by one belonging to the new universe + // the new universe's request is parked: the stale subscription stays in place until it is removed + Assert.IsFalse(dataManager.AddSubscription(newRequest)); Assert.IsTrue(dataManager.DataFeedSubscriptions.TryGetValue(newConfig, out var currentSubscription)); + Assert.AreSame(oldSubscription, currentSubscription); + + // the synchronizer will remove the stale subscription in its next loop, + // after performing one last selection for the disposed universe + Assert.IsTrue(dataManager.RemoveSubscription(config)); + + // the parked request was re-issued: the new universe's subscription is now in place + Assert.IsTrue(dataManager.DataFeedSubscriptions.TryGetValue(newConfig, out currentSubscription)); Assert.AreSame(newSubscription, currentSubscription); CollectionAssert.AreEqual(new[] { newUniverse }, currentSubscription.Universes); // the configuration is still registered for the new universe From b17b7415778273c94c940284c8a13a5ce82236a1 Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Mon, 20 Jul 2026 16:10:13 -0400 Subject: [PATCH 7/8] Restore universe subscription request invariant exception --- Engine/DataFeeds/Subscription.cs | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/Engine/DataFeeds/Subscription.cs b/Engine/DataFeeds/Subscription.cs index 1644353a374e..d8a36286050b 100644 --- a/Engine/DataFeeds/Subscription.cs +++ b/Engine/DataFeeds/Subscription.cs @@ -134,8 +134,13 @@ public bool AddSubscriptionRequest(SubscriptionRequest subscriptionRequest) if (IsUniverseSelectionSubscription || subscriptionRequest.IsUniverseSubscription) { - // for different reasons a user defined universe can trigger a subscription request, likes additions/removals - return false; + if (subscriptionRequest.Universe is UserDefinedUniverse) + { + // for different reasons a user defined universe can trigger a subscription request, likes additions/removals + return true; + } + throw new Exception("Subscription.AddSubscriptionRequest(): Universe selection" + + " subscriptions should not have more than 1 SubscriptionRequest"); } // this shouldn't happen but just in case.. From e1aa7af8178911b16da0f54010ef35ae3c3bcb96 Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Mon, 20 Jul 2026 16:46:45 -0400 Subject: [PATCH 8/8] Report parked universe subscription requests as successfully added --- Engine/DataFeeds/DataManager.cs | 2 +- Engine/DataFeeds/Subscription.cs | 2 +- Tests/Engine/DataFeeds/DataManagerTests.cs | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/Engine/DataFeeds/DataManager.cs b/Engine/DataFeeds/DataManager.cs index 9154c3e14b12..9b76ef85eec2 100644 --- a/Engine/DataFeeds/DataManager.cs +++ b/Engine/DataFeeds/DataManager.cs @@ -313,7 +313,7 @@ public bool AddSubscription(SubscriptionRequest request) { _pendingUniverseSubscriptionRequests[request.Configuration] = request; } - return false; + return true; } // duplicate subscription request diff --git a/Engine/DataFeeds/Subscription.cs b/Engine/DataFeeds/Subscription.cs index d8a36286050b..3022fca327dd 100644 --- a/Engine/DataFeeds/Subscription.cs +++ b/Engine/DataFeeds/Subscription.cs @@ -137,7 +137,7 @@ public bool AddSubscriptionRequest(SubscriptionRequest subscriptionRequest) if (subscriptionRequest.Universe is UserDefinedUniverse) { // for different reasons a user defined universe can trigger a subscription request, likes additions/removals - return true; + return false; } throw new Exception("Subscription.AddSubscriptionRequest(): Universe selection" + " subscriptions should not have more than 1 SubscriptionRequest"); diff --git a/Tests/Engine/DataFeeds/DataManagerTests.cs b/Tests/Engine/DataFeeds/DataManagerTests.cs index 333e92ff6050..53f341b83701 100644 --- a/Tests/Engine/DataFeeds/DataManagerTests.cs +++ b/Tests/Engine/DataFeeds/DataManagerTests.cs @@ -287,7 +287,7 @@ public void ReplacesSubscriptionOfUniverseRemovedAndReAddedInSameTimeStep() dataFeed.Subscription = newSubscription; // the new universe's request is parked: the stale subscription stays in place until it is removed - Assert.IsFalse(dataManager.AddSubscription(newRequest)); + Assert.IsTrue(dataManager.AddSubscription(newRequest)); Assert.IsTrue(dataManager.DataFeedSubscriptions.TryGetValue(newConfig, out var currentSubscription)); Assert.AreSame(oldSubscription, currentSubscription);