diff --git a/Algorithm.CSharp/OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm.cs b/Algorithm.CSharp/OptionUniverseRemovedAndReAddedMemberCleanupRegressionAlgorithm.cs new file mode 100644 index 000000000000..0f97e5695991 --- /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 => 34380; + + /// + /// 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..dd10c3c2d11e --- /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 => 108983; + + /// + /// 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/Engine/DataFeeds/DataManager.cs b/Engine/DataFeeds/DataManager.cs index b634111f4287..9b76ef85eec2 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) { @@ -292,6 +301,21 @@ public bool AddSubscription(SubscriptionRequest request) { if (!subscription.EndOfStream) { + if (request.IsUniverseSubscription + && subscription.IsUniverseSelectionSubscription + && 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 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) + { + _pendingUniverseSubscriptionRequests[request.Configuration] = request; + } + return true; + } + // duplicate subscription request subscription.AddSubscriptionRequest(request); // only result true if the existing subscription is internal, we actually added something from the users perspective @@ -397,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/Tests/Engine/DataFeeds/DataManagerTests.cs b/Tests/Engine/DataFeeds/DataManagerTests.cs index ee8755b87045..53f341b83701 100644 --- a/Tests/Engine/DataFeeds/DataManagerTests.cs +++ b/Tests/Engine/DataFeeds/DataManagerTests.cs @@ -221,6 +221,90 @@ public void ConfigurationForAddedSubscriptionIsAlwaysPresent() dataManager.RemoveAllSubscriptions(); } + [Test] + 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(); + 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 + // 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; + + // the new universe's request is parked: the stale subscription stays in place until it is removed + Assert.IsTrue(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 + Assert.AreEqual(1, dataManager.GetSubscriptionDataConfigs(config.Symbol).Count); + + dataManager.RemoveAllSubscriptions(); + } + [Test] public void ScaledRawNormalizationModeIsNotAllowed() {