From 323102d20326cc159969048edadf9e438b66a0a4 Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Wed, 12 Aug 2026 09:39:04 -0400 Subject: [PATCH 1/6] Delay reading option chain universe files until close to market open in live trading --- ...CustomDataSubscriptionEnumeratorFactory.cs | 13 +++- Engine/DataFeeds/LiveTradingDataFeed.cs | 29 ++++++++- ...mDataSubscriptionEnumeratorFactoryTests.cs | 50 +++++++++++++++- .../DataFeeds/LiveTradingDataFeedTests.cs | 59 +++++++++++++++++++ 4 files changed, 147 insertions(+), 4 deletions(-) diff --git a/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactory.cs b/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactory.cs index 8de3484b5aa0..556034d72a88 100644 --- a/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactory.cs +++ b/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactory.cs @@ -33,6 +33,7 @@ public class LiveCustomDataSubscriptionEnumeratorFactory : ISubscriptionEnumerat private readonly TimeSpan _minimumIntervalCheck; private readonly ITimeProvider _timeProvider; private readonly Func _dateAdjustment; + private readonly Func _canRefresh; private readonly IObjectStore _objectStore; /// @@ -42,12 +43,15 @@ public class LiveCustomDataSubscriptionEnumeratorFactory : ISubscriptionEnumerat /// The object store to use /// Func that allows adjusting the datetime to use /// Allows specifying the minimum interval between each enumerator refresh and data check, default is 30 minutes + /// Optional predicate that determines, given the current utc time, whether the source can be refreshed and read. + /// It is evaluated at the same cadence as the refresh interval, so once it returns true the source will be read within one interval public LiveCustomDataSubscriptionEnumeratorFactory(ITimeProvider timeProvider, IObjectStore objectStore, - Func dateAdjustment = null, TimeSpan? minimumIntervalCheck = null) + Func dateAdjustment = null, TimeSpan? minimumIntervalCheck = null, Func canRefresh = null) { _timeProvider = timeProvider; _dateAdjustment = dateAdjustment; _minimumIntervalCheck = minimumIntervalCheck ?? TimeSpan.FromMinutes(30); + _canRefresh = canRefresh; _objectStore = objectStore; } @@ -79,6 +83,13 @@ public IEnumerator CreateEnumerator(SubscriptionRequest request, IData } lastSourceRefreshTime = utcNow; + + // the refresh gate, if any, is rate limited like the source refreshes so it's not evaluated in a tight loop + if (_canRefresh != null && !_canRefresh(utcNow)) + { + return Enumerable.Empty().GetEnumerator(); + } + var localDate = _dateAdjustment?.Invoke(utcNow.ConvertFromUtc(config.ExchangeTimeZone).Date) ?? utcNow.ConvertFromUtc(config.ExchangeTimeZone).Date; var source = sourceFactory.GetSource(config, localDate, true); if (source == null) diff --git a/Engine/DataFeeds/LiveTradingDataFeed.cs b/Engine/DataFeeds/LiveTradingDataFeed.cs index 0dfc824171f4..b37916e7c302 100644 --- a/Engine/DataFeeds/LiveTradingDataFeed.cs +++ b/Engine/DataFeeds/LiveTradingDataFeed.cs @@ -61,6 +61,10 @@ public class LiveTradingDataFeed : FileSystemDataFeed private static ReferenceWrapper _lastUtcDateShiftUpdate; private static ReferenceWrapper _scheduledUniverseUtcTimeShift; + // option chain universe files can be big, so we delay reading them until the market is open or close to opening + // instead of reading them around the clock + private static readonly TimeSpan PreOpenUniverseFileRefreshWindow = TimeSpan.FromHours(1); + /// /// Public flag indicator that the thread is still busy. /// @@ -356,7 +360,8 @@ request.Universe is OptionChainUniverse || _algorithm.ObjectStore, // we adjust time to the previous tradable date time => Time.GetStartTimeForTradeBars(request.Security.Exchange.Hours, time, Time.OneDay, 1, false, config.DataTimeZone, _algorithm.Settings.DailyPreciseEndTime), - TimeSpan.FromMinutes(10) + TimeSpan.FromMinutes(10), + canRefresh: GetUniverseFileRefreshGate(request) ); var enumeratorStack = factory.CreateEnumerator(request, _dataProvider); @@ -415,6 +420,28 @@ public static TimeSpan GetScheduledUniverseUtcTimeShift(DateTime currentUtcDateT return _scheduledUniverseUtcTimeShift.Value; } + /// + /// Gets a gate for reading a universe file, for universes whose files should not be read around the clock, + /// like option chains, which can be big: they are only read while the market is open + /// or within of the next market open + /// + private static Func GetUniverseFileRefreshGate(SubscriptionRequest request) + { + if (request.Universe is not OptionChainUniverse) + { + return null; + } + + var exchangeHours = request.Security.Exchange.Hours; + return utcNow => + { + var localTime = utcNow.ConvertFromUtc(exchangeHours.TimeZone); + return exchangeHours.IsOpen(localTime, extendedMarketHours: false) + // if the market is closed, GetNextMarketOpen returns the next day open + || exchangeHours.GetNextMarketOpen(localTime, extendedMarketHours: false) - localTime <= PreOpenUniverseFileRefreshWindow; + }; + } + /// /// Build and apply the warmup enumerators when required /// diff --git a/Tests/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactoryTests.cs b/Tests/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactoryTests.cs index a4345c17c6e2..d47f2c2d18c8 100644 --- a/Tests/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactoryTests.cs +++ b/Tests/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactoryTests.cs @@ -549,6 +549,51 @@ public void ToleratesNullSource() enumerator.DisposeSafely(); } + [Test] + public void RespectsRefreshGate() + { + var referenceLocal = new DateTime(2017, 10, 12); + var referenceUtc = referenceLocal.ConvertToUtc(TimeZones.NewYork); + + var timeProvider = new ManualTimeProvider(referenceUtc); + + var dataSourceReader = new Mock(); + dataSourceReader.Setup(dsr => dsr.Read(It.IsAny())) + .Returns(() => new[] { new LocalFileData { EndTime = timeProvider.GetUtcNow().ConvertFromUtc(TimeZones.NewYork) } }) + .Verifiable(); + + var config = new SubscriptionDataConfig(typeof(LocalFileData), Symbols.SPY, Resolution.Daily, TimeZones.NewYork, TimeZones.NewYork, false, false, false); + var request = GetSubscriptionRequest(config, referenceUtc.AddDays(-1), referenceUtc.AddDays(1)); + + var interval = TimeSpan.FromMinutes(30); + var canRefresh = false; + var factory = new TestableLiveCustomDataSubscriptionEnumeratorFactory(timeProvider, dataSourceReader.Object, interval, utcTime => canRefresh); + using var enumerator = factory.CreateEnumerator(request, null); + + // while the gate is closed the source is never read, regardless of how much time passes + Assert.IsTrue(enumerator.MoveNext()); + Assert.IsNull(enumerator.Current); + for (var i = 0; i < 5; i++) + { + timeProvider.Advance(interval); + Assert.IsTrue(enumerator.MoveNext()); + Assert.IsNull(enumerator.Current); + } + dataSourceReader.Verify(dsr => dsr.Read(It.IsAny()), Times.Never); + + // the gate checks are rate limited like source refreshes, so within the same interval nothing is read either + canRefresh = true; + Assert.IsTrue(enumerator.MoveNext()); + Assert.IsNull(enumerator.Current); + dataSourceReader.Verify(dsr => dsr.Read(It.IsAny()), Times.Never); + + // once the gate is open, the next refresh reads the source + timeProvider.Advance(interval); + Assert.IsTrue(enumerator.MoveNext()); + Assert.IsNotNull(enumerator.Current); + dataSourceReader.Verify(dsr => dsr.Read(It.IsAny()), Times.Once); + } + private static void VerifyGetSourceInvocationCount(Mock dataSourceReader, int count, string source, SubscriptionTransportMedium medium, FileFormat fileFormat) { dataSourceReader.Verify(dsr => dsr.Read(It.Is(sds => @@ -630,8 +675,9 @@ class TestableLiveCustomDataSubscriptionEnumeratorFactory : LiveCustomDataSubscr { private readonly ISubscriptionDataSourceReader _dataSourceReader; - public TestableLiveCustomDataSubscriptionEnumeratorFactory(ITimeProvider timeProvider, ISubscriptionDataSourceReader dataSourceReader, TimeSpan? minimumIntervalCheck = null) - : base(timeProvider, null, minimumIntervalCheck: minimumIntervalCheck) + public TestableLiveCustomDataSubscriptionEnumeratorFactory(ITimeProvider timeProvider, ISubscriptionDataSourceReader dataSourceReader, + TimeSpan? minimumIntervalCheck = null, Func canRefresh = null) + : base(timeProvider, null, minimumIntervalCheck: minimumIntervalCheck, canRefresh: canRefresh) { _dataSourceReader = dataSourceReader; } diff --git a/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs b/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs index 74c4d2ad2df5..22012cabd8ac 100644 --- a/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs +++ b/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs @@ -621,6 +621,65 @@ public void OptionChainImmediateSelection(SecurityType securityType) Assert.IsNotEmpty(selectedSymbols); } + [TestCase(SecurityType.Option)] + [TestCase(SecurityType.IndexOption)] + public void OptionChainSelectionIsDelayedUntilCloseToMarketOpen(SecurityType securityType) + { + // start the algorithm during the night: the universe file should not be read until close to the market open + _startDate = securityType == SecurityType.Option + ? new DateTime(2015, 12, 24, 2, 0, 0) + : new DateTime(2021, 01, 04, 2, 0, 0); + var startDateUtc = _startDate.ConvertToUtc(_algorithm.TimeZone); + _manualTimeProvider.SetCurrentTimeUtc(startDateUtc); + var endDate = _startDate.AddDays(1); + + _algorithm.SetBenchmark(x => 1); + + var feed = RunDataFeed(runPostInitialize: false); + + var firstSelectionTimeUtc = DateTime.MinValue; + List selectedSymbols = null; + + var option = securityType == SecurityType.Option + ? _algorithm.AddOption("GOOG") + : _algorithm.AddIndexOption("SPX"); + option.SetFilter(universe => + { + firstSelectionTimeUtc = universe.LocalTime.ConvertToUtc(option.Exchange.TimeZone); + selectedSymbols = (List)universe; + + return universe; + }); + + _algorithm.PostInitialize(); + + // allow time for the exchange to pick up the selection point + Thread.Sleep(50); + + ConsumeBridge(feed, TimeSpan.FromSeconds(15), true, ts => + { + if (firstSelectionTimeUtc != default) + { + // we got what we wanted shortcut unit test + _manualTimeProvider.SetCurrentTimeUtc(Time.EndOfTime); + } + }, + endDate: endDate, + secondsTimeStep: 60); + + var exchangeHours = option.Exchange.Hours; + var marketOpenUtc = exchangeHours + .GetNextMarketOpen(startDateUtc.ConvertFromUtc(exchangeHours.TimeZone), extendedMarketHours: false) + .ConvertToUtc(exchangeHours.TimeZone); + + Assert.AreNotEqual(DateTime.MinValue, firstSelectionTimeUtc); + // selection should have been delayed to at most one hour before the market open, instead of happening right away + Assert.GreaterOrEqual(firstSelectionTimeUtc, marketOpenUtc.AddHours(-1)); + Assert.LessOrEqual(firstSelectionTimeUtc, marketOpenUtc); + Assert.IsNotNull(selectedSymbols); + Assert.IsNotEmpty(selectedSymbols); + } + [Test] public void CustomUniverseImmediateSelection() { From 9b547b58e80981a98cd58d474eada01d3772c253 Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Thu, 13 Aug 2026 16:19:43 -0400 Subject: [PATCH 2/6] Make live chain selection unit tests deterministic with delayed option universe file reads --- .../DataFeeds/LiveTradingDataFeedTests.cs | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs b/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs index 22012cabd8ac..f6ebf0b3c9d9 100644 --- a/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs +++ b/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs @@ -229,6 +229,12 @@ public void EmitsLeanAggregatedDailyData(bool strictEndTimes, bool warmup) public void LiveChainSelection(SecurityType securityType, Resolution resolution, int expirationDatesFilter, bool strictEndTimes) { _startDate = securityType == SecurityType.IndexOption ? new DateTime(2021, 1, 4) : new DateTime(2014, 6, 9); + if (securityType.IsOption()) + { + // option chain universe files are only read while the market is open or close to opening, + // so we start within that window, half an hour before the market open (9:30 NY) + _startDate = _startDate.AddHours(securityType == SecurityType.IndexOption ? 14 : 13); + } _manualTimeProvider.SetCurrentTimeUtc(_startDate); var endDate = _startDate.AddDays(securityType == SecurityType.Future ? 5 : 1); @@ -265,20 +271,23 @@ public void LiveChainSelection(SecurityType securityType, Resolution resolution, } _algorithm.OnEndOfTimeStep(); + var expectedSelections = securityType == SecurityType.Future ? 2 : 1; + // allow time for the exchange to pick up the selection point Thread.Sleep(50); ConsumeBridge(feed, TimeSpan.FromSeconds(5), true, ts => { - if (selectionHappened == 2) + if (selectionHappened == expectedSelections) { // we got what we wanted shortcut unit test _manualTimeProvider.SetCurrentTimeUtc(Time.EndOfTime); } }, endDate: endDate, - secondsTimeStep: 60 * 60); + // a slower time step for options so the simulated clock doesn't race past the universe file refresh window + // between the real-time custom exchange polls + secondsTimeStep: securityType == SecurityType.Future ? 60 * 60 : 60); - var expectedSelections = securityType == SecurityType.Future ? 2 : 1; Assert.AreEqual(expectedSelections, selectionHappened); } @@ -656,7 +665,9 @@ public void OptionChainSelectionIsDelayedUntilCloseToMarketOpen(SecurityType sec // allow time for the exchange to pick up the selection point Thread.Sleep(50); - ConsumeBridge(feed, TimeSpan.FromSeconds(15), true, ts => + // the timeout needs to be generous: the simulated clock advances one minute per loop iteration + // and it has several simulated hours to go through before the universe file refresh window is reached + ConsumeBridge(feed, TimeSpan.FromSeconds(30), true, ts => { if (firstSelectionTimeUtc != default) { From 8c552327884f20fb42ea2373211265e5d6e460ca Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Thu, 13 Aug 2026 16:21:06 -0400 Subject: [PATCH 3/6] Revert PR changes to start over from master --- ...CustomDataSubscriptionEnumeratorFactory.cs | 13 +--- Engine/DataFeeds/LiveTradingDataFeed.cs | 29 +------ ...mDataSubscriptionEnumeratorFactoryTests.cs | 50 +----------- .../DataFeeds/LiveTradingDataFeedTests.cs | 76 +------------------ 4 files changed, 7 insertions(+), 161 deletions(-) diff --git a/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactory.cs b/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactory.cs index 556034d72a88..8de3484b5aa0 100644 --- a/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactory.cs +++ b/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactory.cs @@ -33,7 +33,6 @@ public class LiveCustomDataSubscriptionEnumeratorFactory : ISubscriptionEnumerat private readonly TimeSpan _minimumIntervalCheck; private readonly ITimeProvider _timeProvider; private readonly Func _dateAdjustment; - private readonly Func _canRefresh; private readonly IObjectStore _objectStore; /// @@ -43,15 +42,12 @@ public class LiveCustomDataSubscriptionEnumeratorFactory : ISubscriptionEnumerat /// The object store to use /// Func that allows adjusting the datetime to use /// Allows specifying the minimum interval between each enumerator refresh and data check, default is 30 minutes - /// Optional predicate that determines, given the current utc time, whether the source can be refreshed and read. - /// It is evaluated at the same cadence as the refresh interval, so once it returns true the source will be read within one interval public LiveCustomDataSubscriptionEnumeratorFactory(ITimeProvider timeProvider, IObjectStore objectStore, - Func dateAdjustment = null, TimeSpan? minimumIntervalCheck = null, Func canRefresh = null) + Func dateAdjustment = null, TimeSpan? minimumIntervalCheck = null) { _timeProvider = timeProvider; _dateAdjustment = dateAdjustment; _minimumIntervalCheck = minimumIntervalCheck ?? TimeSpan.FromMinutes(30); - _canRefresh = canRefresh; _objectStore = objectStore; } @@ -83,13 +79,6 @@ public IEnumerator CreateEnumerator(SubscriptionRequest request, IData } lastSourceRefreshTime = utcNow; - - // the refresh gate, if any, is rate limited like the source refreshes so it's not evaluated in a tight loop - if (_canRefresh != null && !_canRefresh(utcNow)) - { - return Enumerable.Empty().GetEnumerator(); - } - var localDate = _dateAdjustment?.Invoke(utcNow.ConvertFromUtc(config.ExchangeTimeZone).Date) ?? utcNow.ConvertFromUtc(config.ExchangeTimeZone).Date; var source = sourceFactory.GetSource(config, localDate, true); if (source == null) diff --git a/Engine/DataFeeds/LiveTradingDataFeed.cs b/Engine/DataFeeds/LiveTradingDataFeed.cs index b37916e7c302..0dfc824171f4 100644 --- a/Engine/DataFeeds/LiveTradingDataFeed.cs +++ b/Engine/DataFeeds/LiveTradingDataFeed.cs @@ -61,10 +61,6 @@ public class LiveTradingDataFeed : FileSystemDataFeed private static ReferenceWrapper _lastUtcDateShiftUpdate; private static ReferenceWrapper _scheduledUniverseUtcTimeShift; - // option chain universe files can be big, so we delay reading them until the market is open or close to opening - // instead of reading them around the clock - private static readonly TimeSpan PreOpenUniverseFileRefreshWindow = TimeSpan.FromHours(1); - /// /// Public flag indicator that the thread is still busy. /// @@ -360,8 +356,7 @@ request.Universe is OptionChainUniverse || _algorithm.ObjectStore, // we adjust time to the previous tradable date time => Time.GetStartTimeForTradeBars(request.Security.Exchange.Hours, time, Time.OneDay, 1, false, config.DataTimeZone, _algorithm.Settings.DailyPreciseEndTime), - TimeSpan.FromMinutes(10), - canRefresh: GetUniverseFileRefreshGate(request) + TimeSpan.FromMinutes(10) ); var enumeratorStack = factory.CreateEnumerator(request, _dataProvider); @@ -420,28 +415,6 @@ public static TimeSpan GetScheduledUniverseUtcTimeShift(DateTime currentUtcDateT return _scheduledUniverseUtcTimeShift.Value; } - /// - /// Gets a gate for reading a universe file, for universes whose files should not be read around the clock, - /// like option chains, which can be big: they are only read while the market is open - /// or within of the next market open - /// - private static Func GetUniverseFileRefreshGate(SubscriptionRequest request) - { - if (request.Universe is not OptionChainUniverse) - { - return null; - } - - var exchangeHours = request.Security.Exchange.Hours; - return utcNow => - { - var localTime = utcNow.ConvertFromUtc(exchangeHours.TimeZone); - return exchangeHours.IsOpen(localTime, extendedMarketHours: false) - // if the market is closed, GetNextMarketOpen returns the next day open - || exchangeHours.GetNextMarketOpen(localTime, extendedMarketHours: false) - localTime <= PreOpenUniverseFileRefreshWindow; - }; - } - /// /// Build and apply the warmup enumerators when required /// diff --git a/Tests/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactoryTests.cs b/Tests/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactoryTests.cs index d47f2c2d18c8..a4345c17c6e2 100644 --- a/Tests/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactoryTests.cs +++ b/Tests/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactoryTests.cs @@ -549,51 +549,6 @@ public void ToleratesNullSource() enumerator.DisposeSafely(); } - [Test] - public void RespectsRefreshGate() - { - var referenceLocal = new DateTime(2017, 10, 12); - var referenceUtc = referenceLocal.ConvertToUtc(TimeZones.NewYork); - - var timeProvider = new ManualTimeProvider(referenceUtc); - - var dataSourceReader = new Mock(); - dataSourceReader.Setup(dsr => dsr.Read(It.IsAny())) - .Returns(() => new[] { new LocalFileData { EndTime = timeProvider.GetUtcNow().ConvertFromUtc(TimeZones.NewYork) } }) - .Verifiable(); - - var config = new SubscriptionDataConfig(typeof(LocalFileData), Symbols.SPY, Resolution.Daily, TimeZones.NewYork, TimeZones.NewYork, false, false, false); - var request = GetSubscriptionRequest(config, referenceUtc.AddDays(-1), referenceUtc.AddDays(1)); - - var interval = TimeSpan.FromMinutes(30); - var canRefresh = false; - var factory = new TestableLiveCustomDataSubscriptionEnumeratorFactory(timeProvider, dataSourceReader.Object, interval, utcTime => canRefresh); - using var enumerator = factory.CreateEnumerator(request, null); - - // while the gate is closed the source is never read, regardless of how much time passes - Assert.IsTrue(enumerator.MoveNext()); - Assert.IsNull(enumerator.Current); - for (var i = 0; i < 5; i++) - { - timeProvider.Advance(interval); - Assert.IsTrue(enumerator.MoveNext()); - Assert.IsNull(enumerator.Current); - } - dataSourceReader.Verify(dsr => dsr.Read(It.IsAny()), Times.Never); - - // the gate checks are rate limited like source refreshes, so within the same interval nothing is read either - canRefresh = true; - Assert.IsTrue(enumerator.MoveNext()); - Assert.IsNull(enumerator.Current); - dataSourceReader.Verify(dsr => dsr.Read(It.IsAny()), Times.Never); - - // once the gate is open, the next refresh reads the source - timeProvider.Advance(interval); - Assert.IsTrue(enumerator.MoveNext()); - Assert.IsNotNull(enumerator.Current); - dataSourceReader.Verify(dsr => dsr.Read(It.IsAny()), Times.Once); - } - private static void VerifyGetSourceInvocationCount(Mock dataSourceReader, int count, string source, SubscriptionTransportMedium medium, FileFormat fileFormat) { dataSourceReader.Verify(dsr => dsr.Read(It.Is(sds => @@ -675,9 +630,8 @@ class TestableLiveCustomDataSubscriptionEnumeratorFactory : LiveCustomDataSubscr { private readonly ISubscriptionDataSourceReader _dataSourceReader; - public TestableLiveCustomDataSubscriptionEnumeratorFactory(ITimeProvider timeProvider, ISubscriptionDataSourceReader dataSourceReader, - TimeSpan? minimumIntervalCheck = null, Func canRefresh = null) - : base(timeProvider, null, minimumIntervalCheck: minimumIntervalCheck, canRefresh: canRefresh) + public TestableLiveCustomDataSubscriptionEnumeratorFactory(ITimeProvider timeProvider, ISubscriptionDataSourceReader dataSourceReader, TimeSpan? minimumIntervalCheck = null) + : base(timeProvider, null, minimumIntervalCheck: minimumIntervalCheck) { _dataSourceReader = dataSourceReader; } diff --git a/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs b/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs index f6ebf0b3c9d9..74c4d2ad2df5 100644 --- a/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs +++ b/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs @@ -229,12 +229,6 @@ public void EmitsLeanAggregatedDailyData(bool strictEndTimes, bool warmup) public void LiveChainSelection(SecurityType securityType, Resolution resolution, int expirationDatesFilter, bool strictEndTimes) { _startDate = securityType == SecurityType.IndexOption ? new DateTime(2021, 1, 4) : new DateTime(2014, 6, 9); - if (securityType.IsOption()) - { - // option chain universe files are only read while the market is open or close to opening, - // so we start within that window, half an hour before the market open (9:30 NY) - _startDate = _startDate.AddHours(securityType == SecurityType.IndexOption ? 14 : 13); - } _manualTimeProvider.SetCurrentTimeUtc(_startDate); var endDate = _startDate.AddDays(securityType == SecurityType.Future ? 5 : 1); @@ -271,23 +265,20 @@ public void LiveChainSelection(SecurityType securityType, Resolution resolution, } _algorithm.OnEndOfTimeStep(); - var expectedSelections = securityType == SecurityType.Future ? 2 : 1; - // allow time for the exchange to pick up the selection point Thread.Sleep(50); ConsumeBridge(feed, TimeSpan.FromSeconds(5), true, ts => { - if (selectionHappened == expectedSelections) + if (selectionHappened == 2) { // we got what we wanted shortcut unit test _manualTimeProvider.SetCurrentTimeUtc(Time.EndOfTime); } }, endDate: endDate, - // a slower time step for options so the simulated clock doesn't race past the universe file refresh window - // between the real-time custom exchange polls - secondsTimeStep: securityType == SecurityType.Future ? 60 * 60 : 60); + secondsTimeStep: 60 * 60); + var expectedSelections = securityType == SecurityType.Future ? 2 : 1; Assert.AreEqual(expectedSelections, selectionHappened); } @@ -630,67 +621,6 @@ public void OptionChainImmediateSelection(SecurityType securityType) Assert.IsNotEmpty(selectedSymbols); } - [TestCase(SecurityType.Option)] - [TestCase(SecurityType.IndexOption)] - public void OptionChainSelectionIsDelayedUntilCloseToMarketOpen(SecurityType securityType) - { - // start the algorithm during the night: the universe file should not be read until close to the market open - _startDate = securityType == SecurityType.Option - ? new DateTime(2015, 12, 24, 2, 0, 0) - : new DateTime(2021, 01, 04, 2, 0, 0); - var startDateUtc = _startDate.ConvertToUtc(_algorithm.TimeZone); - _manualTimeProvider.SetCurrentTimeUtc(startDateUtc); - var endDate = _startDate.AddDays(1); - - _algorithm.SetBenchmark(x => 1); - - var feed = RunDataFeed(runPostInitialize: false); - - var firstSelectionTimeUtc = DateTime.MinValue; - List selectedSymbols = null; - - var option = securityType == SecurityType.Option - ? _algorithm.AddOption("GOOG") - : _algorithm.AddIndexOption("SPX"); - option.SetFilter(universe => - { - firstSelectionTimeUtc = universe.LocalTime.ConvertToUtc(option.Exchange.TimeZone); - selectedSymbols = (List)universe; - - return universe; - }); - - _algorithm.PostInitialize(); - - // allow time for the exchange to pick up the selection point - Thread.Sleep(50); - - // the timeout needs to be generous: the simulated clock advances one minute per loop iteration - // and it has several simulated hours to go through before the universe file refresh window is reached - ConsumeBridge(feed, TimeSpan.FromSeconds(30), true, ts => - { - if (firstSelectionTimeUtc != default) - { - // we got what we wanted shortcut unit test - _manualTimeProvider.SetCurrentTimeUtc(Time.EndOfTime); - } - }, - endDate: endDate, - secondsTimeStep: 60); - - var exchangeHours = option.Exchange.Hours; - var marketOpenUtc = exchangeHours - .GetNextMarketOpen(startDateUtc.ConvertFromUtc(exchangeHours.TimeZone), extendedMarketHours: false) - .ConvertToUtc(exchangeHours.TimeZone); - - Assert.AreNotEqual(DateTime.MinValue, firstSelectionTimeUtc); - // selection should have been delayed to at most one hour before the market open, instead of happening right away - Assert.GreaterOrEqual(firstSelectionTimeUtc, marketOpenUtc.AddHours(-1)); - Assert.LessOrEqual(firstSelectionTimeUtc, marketOpenUtc); - Assert.IsNotNull(selectedSymbols); - Assert.IsNotEmpty(selectedSymbols); - } - [Test] public void CustomUniverseImmediateSelection() { From da046860217a64a70820bb978ac8a47fb468483c Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Thu, 13 Aug 2026 17:45:39 -0400 Subject: [PATCH 4/6] Fall back to backup chain universe files in live trading when the expected ones are unavailable --- ...CustomDataSubscriptionEnumeratorFactory.cs | 13 +- Engine/DataFeeds/LiveTradingDataFeed.cs | 60 ++++++- ...mDataSubscriptionEnumeratorFactoryTests.cs | 50 +++++- .../DataFeeds/LiveTradingDataFeedTests.cs | 146 +++++++++++++++++- 4 files changed, 263 insertions(+), 6 deletions(-) diff --git a/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactory.cs b/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactory.cs index 8de3484b5aa0..51213256447a 100644 --- a/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactory.cs +++ b/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactory.cs @@ -33,6 +33,7 @@ public class LiveCustomDataSubscriptionEnumeratorFactory : ISubscriptionEnumerat private readonly TimeSpan _minimumIntervalCheck; private readonly ITimeProvider _timeProvider; private readonly Func _dateAdjustment; + private readonly Func _sourceAdjustment; private readonly IObjectStore _objectStore; /// @@ -42,12 +43,17 @@ public class LiveCustomDataSubscriptionEnumeratorFactory : ISubscriptionEnumerat /// The object store to use /// Func that allows adjusting the datetime to use /// Allows specifying the minimum interval between each enumerator refresh and data check, default is 30 minutes + /// Optional func that allows adjusting the data source to read from, given the source and the current utc time, + /// e.g. to fall back to an alternative source when the expected one is not available. + /// It is evaluated at the same cadence as the enumerator refreshes public LiveCustomDataSubscriptionEnumeratorFactory(ITimeProvider timeProvider, IObjectStore objectStore, - Func dateAdjustment = null, TimeSpan? minimumIntervalCheck = null) + Func dateAdjustment = null, TimeSpan? minimumIntervalCheck = null, + Func sourceAdjustment = null) { _timeProvider = timeProvider; _dateAdjustment = dateAdjustment; _minimumIntervalCheck = minimumIntervalCheck ?? TimeSpan.FromMinutes(30); + _sourceAdjustment = sourceAdjustment; _objectStore = objectStore; } @@ -87,6 +93,11 @@ public IEnumerator CreateEnumerator(SubscriptionRequest request, IData return Enumerable.Empty().GetEnumerator(); } + if (_sourceAdjustment != null) + { + source = _sourceAdjustment(source, utcNow); + } + // fetch the new source and enumerate the data source reader var enumerator = EnumerateDataSourceReader(config, dataProvider, frontier, source, localDate, sourceFactory); diff --git a/Engine/DataFeeds/LiveTradingDataFeed.cs b/Engine/DataFeeds/LiveTradingDataFeed.cs index 0dfc824171f4..50c63fb8856a 100644 --- a/Engine/DataFeeds/LiveTradingDataFeed.cs +++ b/Engine/DataFeeds/LiveTradingDataFeed.cs @@ -42,6 +42,11 @@ public class LiveTradingDataFeed : FileSystemDataFeed { private static readonly int MaximumWarmupHistoryDaysLookBack = Config.GetInt("maximum-warmup-history-days-look-back", 5); + // when the expected chain universe file is not available yet, we fall back to the backup universe file ("*.backup"), + // if any, as a last resort, when the market is open or within this time span before the next market open + private static readonly TimeSpan UniverseFileBackupFallbackWindow = + TimeSpan.FromMinutes(Config.GetInt("universe-file-backup-fallback-minutes", 30)); + private LiveNodePacket _job; // used to get current time @@ -356,7 +361,8 @@ request.Universe is OptionChainUniverse || _algorithm.ObjectStore, // we adjust time to the previous tradable date time => Time.GetStartTimeForTradeBars(request.Security.Exchange.Hours, time, Time.OneDay, 1, false, config.DataTimeZone, _algorithm.Settings.DailyPreciseEndTime), - TimeSpan.FromMinutes(10) + TimeSpan.FromMinutes(10), + sourceAdjustment: GetUniverseFileBackupSourceAdjustment(request) ); var enumeratorStack = factory.CreateEnumerator(request, _dataProvider); @@ -415,6 +421,58 @@ public static TimeSpan GetScheduledUniverseUtcTimeShift(DateTime currentUtcDateT return _scheduledUniverseUtcTimeShift.Value; } + /// + /// Gets a source adjustment for chain universe files as a safety net for when the expected universe file + /// is not available yet: when the market is open or close to opening (within + /// of the next market open), it falls back to the backup universe file ("*.backup") if present, as a last resort + /// + private Func GetUniverseFileBackupSourceAdjustment(SubscriptionRequest request) + { + if (request.Universe is not (OptionChainUniverse or FuturesChainUniverse)) + { + return null; + } + + var exchangeHours = request.Security.Exchange.Hours; + return (source, utcNow) => + { + if (source.TransportMedium != SubscriptionTransportMedium.LocalFile) + { + return source; + } + + var localTime = utcNow.ConvertFromUtc(exchangeHours.TimeZone); + // only fall back when the market is open or close to opening, when the expected universe file should already be available + if (!exchangeHours.IsOpen(localTime, extendedMarketHours: false) + // if the market is closed, GetNextMarketOpen returns the next day open + && exchangeHours.GetNextMarketOpen(localTime, extendedMarketHours: false) - localTime > UniverseFileBackupFallbackWindow) + { + return source; + } + + if (CanFetchDataSource(source)) + { + return source; + } + + var backupSource = new SubscriptionDataSource(source.Source + ".backup", source.TransportMedium, source.Format); + if (CanFetchDataSource(backupSource)) + { + Log.Trace($"LiveTradingDataFeed.GetUniverseFileBackupSourceAdjustment(): universe file '{source.Source}' is not available, " + + $"falling back to backup universe file '{backupSource.Source}'"); + return backupSource; + } + + return source; + }; + } + + private bool CanFetchDataSource(SubscriptionDataSource source) + { + using var stream = _dataProvider.Fetch(source.Source); + return stream != null; + } + /// /// Build and apply the warmup enumerators when required /// diff --git a/Tests/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactoryTests.cs b/Tests/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactoryTests.cs index a4345c17c6e2..6f8f1a6fb830 100644 --- a/Tests/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactoryTests.cs +++ b/Tests/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactoryTests.cs @@ -549,6 +549,51 @@ public void ToleratesNullSource() enumerator.DisposeSafely(); } + [Test] + public void AllowsAdjustingTheDataSource() + { + var referenceLocal = new DateTime(2017, 10, 12); + var referenceUtc = new DateTime(2017, 10, 12).ConvertToUtc(TimeZones.NewYork); + + var timeProvider = new ManualTimeProvider(referenceUtc); + + var dataSourceReader = new Mock(); + dataSourceReader.Setup(dsr => dsr.Read(It.IsAny())) + .Returns(() => new[] { new LocalFileData { EndTime = referenceLocal.AddSeconds(1) } }) + .Verifiable(); + + var config = new SubscriptionDataConfig(typeof(LocalFileData), Symbols.SPY, Resolution.Daily, TimeZones.NewYork, TimeZones.NewYork, false, false, false); + var request = GetSubscriptionRequest(config, referenceUtc.AddSeconds(-1), referenceUtc.AddDays(1)); + + var sourceAdjustmentTimesUtc = new List(); + var factory = new TestableLiveCustomDataSubscriptionEnumeratorFactory(timeProvider, dataSourceReader.Object, + sourceAdjustment: (source, utcNow) => + { + sourceAdjustmentTimesUtc.Add(utcNow); + return new SubscriptionDataSource(source.Source + ".backup", source.TransportMedium, source.Format); + }); + using var enumerator = factory.CreateEnumerator(request, null); + + Assert.IsTrue(enumerator.MoveNext()); + Assert.IsNotNull(enumerator.Current); + + // the adjusted source is the one that gets read, not the original one + VerifyGetSourceInvocationCount(dataSourceReader, 1, "local.file.source.backup", SubscriptionTransportMedium.LocalFile, FileFormat.Csv); + VerifyGetSourceInvocationCount(dataSourceReader, 0, "local.file.source", SubscriptionTransportMedium.LocalFile, FileFormat.Csv); + + CollectionAssert.AreEqual(new[] { referenceUtc }, sourceAdjustmentTimesUtc); + + // the source adjustment is rate limited like the source refreshes + Assert.IsTrue(enumerator.MoveNext()); + Assert.IsNull(enumerator.Current); + Assert.AreEqual(1, sourceAdjustmentTimesUtc.Count); + + timeProvider.Advance(TimeSpan.FromMinutes(30)); + Assert.IsTrue(enumerator.MoveNext()); + Assert.AreEqual(2, sourceAdjustmentTimesUtc.Count); + Assert.AreEqual(referenceUtc.AddMinutes(30), sourceAdjustmentTimesUtc[1]); + } + private static void VerifyGetSourceInvocationCount(Mock dataSourceReader, int count, string source, SubscriptionTransportMedium medium, FileFormat fileFormat) { dataSourceReader.Verify(dsr => dsr.Read(It.Is(sds => @@ -630,8 +675,9 @@ class TestableLiveCustomDataSubscriptionEnumeratorFactory : LiveCustomDataSubscr { private readonly ISubscriptionDataSourceReader _dataSourceReader; - public TestableLiveCustomDataSubscriptionEnumeratorFactory(ITimeProvider timeProvider, ISubscriptionDataSourceReader dataSourceReader, TimeSpan? minimumIntervalCheck = null) - : base(timeProvider, null, minimumIntervalCheck: minimumIntervalCheck) + public TestableLiveCustomDataSubscriptionEnumeratorFactory(ITimeProvider timeProvider, ISubscriptionDataSourceReader dataSourceReader, + TimeSpan? minimumIntervalCheck = null, Func sourceAdjustment = null) + : base(timeProvider, null, minimumIntervalCheck: minimumIntervalCheck, sourceAdjustment: sourceAdjustment) { _dataSourceReader = dataSourceReader; } diff --git a/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs b/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs index 74c4d2ad2df5..b6c61439101c 100644 --- a/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs +++ b/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs @@ -17,6 +17,7 @@ using System; using System.Collections.Generic; using System.Diagnostics; +using System.IO; using System.Linq; using System.Threading; using Microsoft.CodeAnalysis; @@ -282,6 +283,104 @@ public void LiveChainSelection(SecurityType securityType, Resolution resolution, Assert.AreEqual(expectedSelections, selectionHappened); } + [TestCase(SecurityType.Option, false)] + [TestCase(SecurityType.Option, true)] + [TestCase(SecurityType.IndexOption, false)] + [TestCase(SecurityType.IndexOption, true)] + public void ChainSelectionFallsBackToBackupUniverseFileCloseToMarketOpen(SecurityType securityType, bool universeFileAvailable) + { + // start close to the market open (9:15 NY), within the backup universe file fallback window (30 minutes before the open by default) + _startDate = securityType == SecurityType.Option + ? new DateTime(2014, 6, 9, 13, 15, 0) + : new DateTime(2021, 1, 4, 14, 15, 0); + _manualTimeProvider.SetCurrentTimeUtc(_startDate); + var endDate = _startDate.AddDays(1); + + _algorithm.SetBenchmark(x => 1); + + var dataProvider = new BackupUniverseFileDataProvider(hideUniverseFiles: !universeFileAvailable); + var feed = RunDataFeed(runPostInitialize: false, dataProvider: dataProvider); + + var selectionHappened = 0; + List selectedContracts = null; + var option = securityType == SecurityType.Option + ? _algorithm.AddOption("AAPL") + : _algorithm.AddIndexOption("SPX"); + option.SetFilter(universe => + { + selectionHappened++; + selectedContracts = universe.ToList(); + + return universe; + }); + + _algorithm.PostInitialize(); + + // allow time for the exchange to pick up the selection point + Thread.Sleep(50); + + ConsumeBridge(feed, TimeSpan.FromSeconds(30), true, ts => + { + if (selectionHappened > 0) + { + // we got what we wanted shortcut unit test + _manualTimeProvider.SetCurrentTimeUtc(Time.EndOfTime); + } + }, + endDate: endDate, + secondsTimeStep: 60); + + Assert.AreEqual(1, selectionHappened); + Assert.IsNotNull(selectedContracts); + Assert.IsNotEmpty(selectedContracts); + + if (universeFileAvailable) + { + // the universe file was available, so the backup file should not have even been checked + Assert.AreEqual(0, dataProvider.BackupUniverseFileRequests); + } + else + { + Assert.AreNotEqual(0, dataProvider.UniverseFileRequests); + Assert.AreNotEqual(0, dataProvider.BackupUniverseFileRequests); + } + } + + [Test] + public void ChainSelectionDoesNotFallBackToBackupUniverseFileFarFromMarketOpen() + { + // start during the night: far from the market open, the missing universe file should not fall back to the backup file + _startDate = new DateTime(2014, 6, 9, 6, 0, 0); + _manualTimeProvider.SetCurrentTimeUtc(_startDate); + // stop before entering the fallback window, 30 minutes (by default) before the 9:30 NY market open + var endDate = new DateTime(2014, 6, 9, 12, 0, 0); + + _algorithm.SetBenchmark(x => 1); + + var dataProvider = new BackupUniverseFileDataProvider(hideUniverseFiles: true); + var feed = RunDataFeed(runPostInitialize: false, dataProvider: dataProvider); + + var selectionHappened = 0; + var option = _algorithm.AddOption("AAPL"); + option.SetFilter(universe => + { + selectionHappened++; + return universe; + }); + + _algorithm.PostInitialize(); + + // allow time for the exchange to pick up the selection point + Thread.Sleep(50); + + ConsumeBridge(feed, TimeSpan.FromSeconds(30), true, ts => { }, endDate: endDate, secondsTimeStep: 60); + + // the universe file was tried but never available, and the backup file should not have been used + Assert.AreEqual(0, selectionHappened); + Assert.AreNotEqual(0, dataProvider.UniverseFileRequests); + Assert.AreEqual(0, dataProvider.BackupUniverseFileRequests); + } + [Test] public void ContinuousFuturesImmediateSelection() { @@ -2914,7 +3013,7 @@ private IDataFeed RunDataFeed(Resolution resolution = Resolution.Second, List> getNextTicksFunction = null, Func> lookupSymbolsFunction = null, Func canPerformSelection = null, IDataQueueHandler dataQueueHandler = null, - bool runPostInitialize = true) + bool runPostInitialize = true, IDataProvider dataProvider = null) { _algorithm.SetStartDate(_startDate); _algorithm.SetDateTime(_manualTimeProvider.GetUtcNow()); @@ -2988,7 +3087,7 @@ private IDataFeed RunDataFeed(Resolution resolution = Resolution.Second, List _universeFileRequests; + public int BackupUniverseFileRequests => _backupUniverseFileRequests; + + public event EventHandler NewDataRequest; + + public BackupUniverseFileDataProvider(bool hideUniverseFiles) + { + _hideUniverseFiles = hideUniverseFiles; + } + + public Stream Fetch(string key) + { + if (key.Contains("universes", StringComparison.InvariantCulture)) + { + if (key.EndsWith(".csv.backup", StringComparison.InvariantCulture)) + { + Interlocked.Increment(ref _backupUniverseFileRequests); + // serve the backup universe file contents from the actual universe file + return _dataProvider.Fetch(key.Substring(0, key.Length - ".backup".Length)); + } + + if (key.EndsWith(".csv", StringComparison.InvariantCulture)) + { + Interlocked.Increment(ref _universeFileRequests); + if (_hideUniverseFiles) + { + return null; + } + } + } + + return _dataProvider.Fetch(key); + } + } + private static IEnumerable ProduceBenchmarkTicks(FuncDataQueueHandler fdqh, Count count) { for (int i = 0; i < 10000; i++) From 4e8027bede7b9c1ba68ebedfbcd4997d9c4bec1d Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Thu, 13 Aug 2026 18:04:06 -0400 Subject: [PATCH 5/6] Extend the backup universe file fallback to all file-based universes --- Engine/DataFeeds/LiveTradingDataFeed.cs | 9 +-- .../DataFeeds/LiveTradingDataFeedTests.cs | 78 ++++++++++++++----- 2 files changed, 61 insertions(+), 26 deletions(-) diff --git a/Engine/DataFeeds/LiveTradingDataFeed.cs b/Engine/DataFeeds/LiveTradingDataFeed.cs index 50c63fb8856a..2d3deca489a4 100644 --- a/Engine/DataFeeds/LiveTradingDataFeed.cs +++ b/Engine/DataFeeds/LiveTradingDataFeed.cs @@ -42,7 +42,7 @@ public class LiveTradingDataFeed : FileSystemDataFeed { private static readonly int MaximumWarmupHistoryDaysLookBack = Config.GetInt("maximum-warmup-history-days-look-back", 5); - // when the expected chain universe file is not available yet, we fall back to the backup universe file ("*.backup"), + // when the expected universe file is not available yet, we fall back to the backup universe file ("*.backup"), // if any, as a last resort, when the market is open or within this time span before the next market open private static readonly TimeSpan UniverseFileBackupFallbackWindow = TimeSpan.FromMinutes(Config.GetInt("universe-file-backup-fallback-minutes", 30)); @@ -422,17 +422,12 @@ public static TimeSpan GetScheduledUniverseUtcTimeShift(DateTime currentUtcDateT } /// - /// Gets a source adjustment for chain universe files as a safety net for when the expected universe file + /// Gets a source adjustment for universe files as a safety net for when the expected universe file /// is not available yet: when the market is open or close to opening (within /// of the next market open), it falls back to the backup universe file ("*.backup") if present, as a last resort /// private Func GetUniverseFileBackupSourceAdjustment(SubscriptionRequest request) { - if (request.Universe is not (OptionChainUniverse or FuturesChainUniverse)) - { - return null; - } - var exchangeHours = request.Security.Exchange.Hours; return (source, utcNow) => { diff --git a/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs b/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs index b6c61439101c..4c4cc80e799a 100644 --- a/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs +++ b/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs @@ -283,16 +283,25 @@ public void LiveChainSelection(SecurityType securityType, Resolution resolution, Assert.AreEqual(expectedSelections, selectionHappened); } - [TestCase(SecurityType.Option, false)] - [TestCase(SecurityType.Option, true)] - [TestCase(SecurityType.IndexOption, false)] - [TestCase(SecurityType.IndexOption, true)] - public void ChainSelectionFallsBackToBackupUniverseFileCloseToMarketOpen(SecurityType securityType, bool universeFileAvailable) + [TestCase("OptionChain", false)] + [TestCase("OptionChain", true)] + [TestCase("IndexOptionChain", false)] + [TestCase("IndexOptionChain", true)] + [TestCase("CoarseFundamental", false)] + [TestCase("CoarseFundamental", true)] + [TestCase("EtfConstituents", false)] + [TestCase("EtfConstituents", true)] + public void UniverseSelectionFallsBackToBackupUniverseFileCloseToMarketOpen(string universeKind, bool universeFileAvailable) { // start close to the market open (9:15 NY), within the backup universe file fallback window (30 minutes before the open by default) - _startDate = securityType == SecurityType.Option - ? new DateTime(2014, 6, 9, 13, 15, 0) - : new DateTime(2021, 1, 4, 14, 15, 0); + _startDate = universeKind switch + { + "OptionChain" => new DateTime(2014, 6, 9, 13, 15, 0), + "IndexOptionChain" => new DateTime(2021, 1, 4, 14, 15, 0), + "CoarseFundamental" => new DateTime(2014, 3, 26, 13, 15, 0), + "EtfConstituents" => new DateTime(2020, 12, 1, 14, 15, 0), + _ => throw new ArgumentException($"Unexpected universe kind: {universeKind}") + }; _manualTimeProvider.SetCurrentTimeUtc(_startDate); var endDate = _startDate.AddDays(1); @@ -302,17 +311,47 @@ public void ChainSelectionFallsBackToBackupUniverseFileCloseToMarketOpen(Securit var feed = RunDataFeed(runPostInitialize: false, dataProvider: dataProvider); var selectionHappened = 0; - List selectedContracts = null; - var option = securityType == SecurityType.Option - ? _algorithm.AddOption("AAPL") - : _algorithm.AddIndexOption("SPX"); - option.SetFilter(universe => + var selectedCount = 0; + + IEnumerable CoarseFilter(IEnumerable coarse) { selectionHappened++; - selectedContracts = universe.ToList(); + var symbols = coarse.Select(x => x.Symbol).ToList(); + selectedCount = symbols.Count; + return symbols; + } - return universe; - }); + switch (universeKind) + { + case "OptionChain": + case "IndexOptionChain": + var option = universeKind == "OptionChain" + ? _algorithm.AddOption("AAPL") + : _algorithm.AddIndexOption("SPX"); + option.SetFilter(universe => + { + selectionHappened++; + selectedCount = universe.Count(); + return universe; + }); + break; + + case "CoarseFundamental": + _algorithm.UniverseSettings.Resolution = Resolution.Daily; + _algorithm.AddUniverse(CoarseFilter); + break; + + case "EtfConstituents": + var spy = _algorithm.AddEquity("SPY").Symbol; + _algorithm.AddUniverse(_algorithm.Universe.ETF(spy, constituentsData => + { + selectionHappened++; + var symbols = constituentsData.Select(x => x.Symbol).ToList(); + selectedCount = symbols.Count; + return symbols; + })); + break; + } _algorithm.PostInitialize(); @@ -331,8 +370,7 @@ public void ChainSelectionFallsBackToBackupUniverseFileCloseToMarketOpen(Securit secondsTimeStep: 60); Assert.AreEqual(1, selectionHappened); - Assert.IsNotNull(selectedContracts); - Assert.IsNotEmpty(selectedContracts); + Assert.AreNotEqual(0, selectedCount); if (universeFileAvailable) { @@ -3206,7 +3244,9 @@ public BackupUniverseFileDataProvider(bool hideUniverseFiles) public Stream Fetch(string key) { - if (key.Contains("universes", StringComparison.InvariantCulture)) + // coarse fundamental files are universe files too, they just don't live under a "universes" folder + if (key.Contains("universes", StringComparison.InvariantCulture) + || key.Replace('\\', '/').Contains("fundamental/coarse", StringComparison.InvariantCulture)) { if (key.EndsWith(".csv.backup", StringComparison.InvariantCulture)) { From d49e1fd466ed6bea0f70df5866d6eb59b4ab867f Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Thu, 13 Aug 2026 18:28:39 -0400 Subject: [PATCH 6/6] Move the backup universe file fallback logic into the live custom data enumerator factory --- ...CustomDataSubscriptionEnumeratorFactory.cs | 73 +++++++++++-- Engine/DataFeeds/LiveTradingDataFeed.cs | 55 +--------- ...mDataSubscriptionEnumeratorFactoryTests.cs | 100 ++++++++++++++---- 3 files changed, 148 insertions(+), 80 deletions(-) diff --git a/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactory.cs b/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactory.cs index 51213256447a..3be25275d20d 100644 --- a/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactory.cs +++ b/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactory.cs @@ -18,9 +18,11 @@ using System.Collections.Generic; using System.Linq; using Python.Runtime; +using QuantConnect.Configuration; using QuantConnect.Data; using QuantConnect.Data.UniverseSelection; using QuantConnect.Interfaces; +using QuantConnect.Logging; using QuantConnect.Util; namespace QuantConnect.Lean.Engine.DataFeeds.Enumerators.Factories @@ -30,10 +32,15 @@ namespace QuantConnect.Lean.Engine.DataFeeds.Enumerators.Factories /// public class LiveCustomDataSubscriptionEnumeratorFactory : ISubscriptionEnumeratorFactory { + // when the expected universe file is not available yet, we fall back to the backup universe file ("*.backup"), + // if any, as a last resort, when the market is open or within this time span before the next market open + private static readonly TimeSpan UniverseFileBackupFallbackWindow = + TimeSpan.FromMinutes(Config.GetInt("universe-file-backup-fallback-minutes", 30)); + private readonly TimeSpan _minimumIntervalCheck; private readonly ITimeProvider _timeProvider; private readonly Func _dateAdjustment; - private readonly Func _sourceAdjustment; + private readonly bool _fallBackToBackupUniverseFiles; private readonly IObjectStore _objectStore; /// @@ -43,17 +50,17 @@ public class LiveCustomDataSubscriptionEnumeratorFactory : ISubscriptionEnumerat /// The object store to use /// Func that allows adjusting the datetime to use /// Allows specifying the minimum interval between each enumerator refresh and data check, default is 30 minutes - /// Optional func that allows adjusting the data source to read from, given the source and the current utc time, - /// e.g. to fall back to an alternative source when the expected one is not available. - /// It is evaluated at the same cadence as the enumerator refreshes + /// Whether to fall back to the backup universe file ("*.backup"), if any, as a last resort + /// when the expected universe file is not available and the market is open or close to opening. + /// Only meaningful for universe subscriptions backed by local files public LiveCustomDataSubscriptionEnumeratorFactory(ITimeProvider timeProvider, IObjectStore objectStore, Func dateAdjustment = null, TimeSpan? minimumIntervalCheck = null, - Func sourceAdjustment = null) + bool fallBackToBackupUniverseFiles = false) { _timeProvider = timeProvider; _dateAdjustment = dateAdjustment; _minimumIntervalCheck = minimumIntervalCheck ?? TimeSpan.FromMinutes(30); - _sourceAdjustment = sourceAdjustment; + _fallBackToBackupUniverseFiles = fallBackToBackupUniverseFiles; _objectStore = objectStore; } @@ -72,6 +79,7 @@ public IEnumerator CreateEnumerator(SubscriptionRequest request, IData var frontier = Ref.Create(_dateAdjustment?.Invoke(request.StartTimeLocal) ?? request.StartTimeLocal); var lastSourceRefreshTime = DateTime.MinValue; var sourceFactory = config.GetBaseDataInstance(); + var sourceAdjustment = _fallBackToBackupUniverseFiles ? GetUniverseFileBackupSourceAdjustment(request, dataProvider) : null; // this is refreshing the enumerator stack for each new source var refresher = new RefreshEnumerator(() => @@ -93,9 +101,9 @@ public IEnumerator CreateEnumerator(SubscriptionRequest request, IData return Enumerable.Empty().GetEnumerator(); } - if (_sourceAdjustment != null) + if (sourceAdjustment != null) { - source = _sourceAdjustment(source, utcNow); + source = sourceAdjustment(source, utcNow); } // fetch the new source and enumerate the data source reader @@ -213,6 +221,55 @@ IDataProvider dataProvider return SubscriptionDataSourceReader.ForSource(source, dataCacheProvider, config, date, true, baseDataInstance, dataProvider, _objectStore); } + /// + /// Gets a source adjustment for universe files as a safety net for when the expected universe file + /// is not available yet: when the market is open or close to opening (within + /// of the next market open), it falls back to the backup universe file ("*.backup") if present, as a last resort. + /// It is evaluated at the same cadence as the enumerator refreshes + /// + private static Func GetUniverseFileBackupSourceAdjustment( + SubscriptionRequest request, IDataProvider dataProvider) + { + var exchangeHours = request.Security.Exchange.Hours; + return (source, utcNow) => + { + if (source.TransportMedium != SubscriptionTransportMedium.LocalFile) + { + return source; + } + + var localTime = utcNow.ConvertFromUtc(exchangeHours.TimeZone); + // only fall back when the market is open or close to opening, when the expected universe file should already be available + if (!exchangeHours.IsOpen(localTime, extendedMarketHours: false) + // if the market is closed, GetNextMarketOpen returns the next day open + && exchangeHours.GetNextMarketOpen(localTime, extendedMarketHours: false) - localTime > UniverseFileBackupFallbackWindow) + { + return source; + } + + if (CanFetchDataSource(dataProvider, source)) + { + return source; + } + + var backupSource = new SubscriptionDataSource(source.Source + ".backup", source.TransportMedium, source.Format); + if (CanFetchDataSource(dataProvider, backupSource)) + { + Log.Trace($"LiveCustomDataSubscriptionEnumeratorFactory.GetUniverseFileBackupSourceAdjustment(): universe file '{source.Source}' is not available, " + + $"falling back to backup universe file '{backupSource.Source}'"); + return backupSource; + } + + return source; + }; + } + + private static bool CanFetchDataSource(IDataProvider dataProvider, SubscriptionDataSource source) + { + using var stream = dataProvider.Fetch(source.Source); + return stream != null; + } + private bool SourceRequiresFastForward(SubscriptionDataSource source) { return source.TransportMedium == SubscriptionTransportMedium.LocalFile diff --git a/Engine/DataFeeds/LiveTradingDataFeed.cs b/Engine/DataFeeds/LiveTradingDataFeed.cs index 2d3deca489a4..4884eaa68ca6 100644 --- a/Engine/DataFeeds/LiveTradingDataFeed.cs +++ b/Engine/DataFeeds/LiveTradingDataFeed.cs @@ -42,11 +42,6 @@ public class LiveTradingDataFeed : FileSystemDataFeed { private static readonly int MaximumWarmupHistoryDaysLookBack = Config.GetInt("maximum-warmup-history-days-look-back", 5); - // when the expected universe file is not available yet, we fall back to the backup universe file ("*.backup"), - // if any, as a last resort, when the market is open or within this time span before the next market open - private static readonly TimeSpan UniverseFileBackupFallbackWindow = - TimeSpan.FromMinutes(Config.GetInt("universe-file-backup-fallback-minutes", 30)); - private LiveNodePacket _job; // used to get current time @@ -362,7 +357,8 @@ request.Universe is OptionChainUniverse || // we adjust time to the previous tradable date time => Time.GetStartTimeForTradeBars(request.Security.Exchange.Hours, time, Time.OneDay, 1, false, config.DataTimeZone, _algorithm.Settings.DailyPreciseEndTime), TimeSpan.FromMinutes(10), - sourceAdjustment: GetUniverseFileBackupSourceAdjustment(request) + // when the expected universe file is not available yet, fall back to the backup universe file as a last resort + fallBackToBackupUniverseFiles: true ); var enumeratorStack = factory.CreateEnumerator(request, _dataProvider); @@ -421,53 +417,6 @@ public static TimeSpan GetScheduledUniverseUtcTimeShift(DateTime currentUtcDateT return _scheduledUniverseUtcTimeShift.Value; } - /// - /// Gets a source adjustment for universe files as a safety net for when the expected universe file - /// is not available yet: when the market is open or close to opening (within - /// of the next market open), it falls back to the backup universe file ("*.backup") if present, as a last resort - /// - private Func GetUniverseFileBackupSourceAdjustment(SubscriptionRequest request) - { - var exchangeHours = request.Security.Exchange.Hours; - return (source, utcNow) => - { - if (source.TransportMedium != SubscriptionTransportMedium.LocalFile) - { - return source; - } - - var localTime = utcNow.ConvertFromUtc(exchangeHours.TimeZone); - // only fall back when the market is open or close to opening, when the expected universe file should already be available - if (!exchangeHours.IsOpen(localTime, extendedMarketHours: false) - // if the market is closed, GetNextMarketOpen returns the next day open - && exchangeHours.GetNextMarketOpen(localTime, extendedMarketHours: false) - localTime > UniverseFileBackupFallbackWindow) - { - return source; - } - - if (CanFetchDataSource(source)) - { - return source; - } - - var backupSource = new SubscriptionDataSource(source.Source + ".backup", source.TransportMedium, source.Format); - if (CanFetchDataSource(backupSource)) - { - Log.Trace($"LiveTradingDataFeed.GetUniverseFileBackupSourceAdjustment(): universe file '{source.Source}' is not available, " + - $"falling back to backup universe file '{backupSource.Source}'"); - return backupSource; - } - - return source; - }; - } - - private bool CanFetchDataSource(SubscriptionDataSource source) - { - using var stream = _dataProvider.Fetch(source.Source); - return stream != null; - } - /// /// Build and apply the warmup enumerators when required /// diff --git a/Tests/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactoryTests.cs b/Tests/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactoryTests.cs index 6f8f1a6fb830..8ddcd132b45c 100644 --- a/Tests/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactoryTests.cs +++ b/Tests/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactoryTests.cs @@ -16,6 +16,7 @@ using System; using System.Collections.Generic; +using System.IO; using System.Linq; using Moq; using NUnit.Framework; @@ -550,10 +551,11 @@ public void ToleratesNullSource() } [Test] - public void AllowsAdjustingTheDataSource() + public void FallsBackToBackupUniverseFileWhenExpectedSourceIsNotAvailable() { - var referenceLocal = new DateTime(2017, 10, 12); - var referenceUtc = new DateTime(2017, 10, 12).ConvertToUtc(TimeZones.NewYork); + // 10 am, the market is open, so the backup fallback is active + var referenceLocal = new DateTime(2017, 10, 12, 10, 0, 0); + var referenceUtc = referenceLocal.ConvertToUtc(TimeZones.NewYork); var timeProvider = new ManualTimeProvider(referenceUtc); @@ -562,36 +564,96 @@ public void AllowsAdjustingTheDataSource() .Returns(() => new[] { new LocalFileData { EndTime = referenceLocal.AddSeconds(1) } }) .Verifiable(); + var expectedSourceAvailable = false; + var dataProvider = new Mock(); + dataProvider.Setup(dp => dp.Fetch("local.file.source")).Returns(() => expectedSourceAvailable ? new MemoryStream() : null); + dataProvider.Setup(dp => dp.Fetch("local.file.source.backup")).Returns(() => new MemoryStream()); + var config = new SubscriptionDataConfig(typeof(LocalFileData), Symbols.SPY, Resolution.Daily, TimeZones.NewYork, TimeZones.NewYork, false, false, false); var request = GetSubscriptionRequest(config, referenceUtc.AddSeconds(-1), referenceUtc.AddDays(1)); - var sourceAdjustmentTimesUtc = new List(); var factory = new TestableLiveCustomDataSubscriptionEnumeratorFactory(timeProvider, dataSourceReader.Object, - sourceAdjustment: (source, utcNow) => - { - sourceAdjustmentTimesUtc.Add(utcNow); - return new SubscriptionDataSource(source.Source + ".backup", source.TransportMedium, source.Format); - }); - using var enumerator = factory.CreateEnumerator(request, null); + fallBackToBackupUniverseFiles: true); + using var enumerator = factory.CreateEnumerator(request, dataProvider.Object); Assert.IsTrue(enumerator.MoveNext()); Assert.IsNotNull(enumerator.Current); - // the adjusted source is the one that gets read, not the original one + // the expected source is not available, so the backup source is the one that gets read VerifyGetSourceInvocationCount(dataSourceReader, 1, "local.file.source.backup", SubscriptionTransportMedium.LocalFile, FileFormat.Csv); VerifyGetSourceInvocationCount(dataSourceReader, 0, "local.file.source", SubscriptionTransportMedium.LocalFile, FileFormat.Csv); + dataProvider.Verify(dp => dp.Fetch(It.IsAny()), Times.Exactly(2)); - CollectionAssert.AreEqual(new[] { referenceUtc }, sourceAdjustmentTimesUtc); - - // the source adjustment is rate limited like the source refreshes + // the fallback checks are rate limited like the source refreshes Assert.IsTrue(enumerator.MoveNext()); Assert.IsNull(enumerator.Current); - Assert.AreEqual(1, sourceAdjustmentTimesUtc.Count); + dataProvider.Verify(dp => dp.Fetch(It.IsAny()), Times.Exactly(2)); + // the expected source is re-checked and preferred on the next refresh once it becomes available + expectedSourceAvailable = true; timeProvider.Advance(TimeSpan.FromMinutes(30)); Assert.IsTrue(enumerator.MoveNext()); - Assert.AreEqual(2, sourceAdjustmentTimesUtc.Count); - Assert.AreEqual(referenceUtc.AddMinutes(30), sourceAdjustmentTimesUtc[1]); + VerifyGetSourceInvocationCount(dataSourceReader, 1, "local.file.source", SubscriptionTransportMedium.LocalFile, FileFormat.Csv); + VerifyGetSourceInvocationCount(dataSourceReader, 1, "local.file.source.backup", SubscriptionTransportMedium.LocalFile, FileFormat.Csv); + } + + [Test] + public void DoesNotFallBackToBackupUniverseFileFarFromMarketOpen() + { + // midnight, more than the fallback window away from the next market open, so no backup probing happens + var referenceLocal = new DateTime(2017, 10, 12); + var referenceUtc = referenceLocal.ConvertToUtc(TimeZones.NewYork); + + var timeProvider = new ManualTimeProvider(referenceUtc); + + var dataSourceReader = new Mock(); + dataSourceReader.Setup(dsr => dsr.Read(It.IsAny())) + .Returns(() => new[] { new LocalFileData { EndTime = referenceLocal.AddSeconds(1) } }) + .Verifiable(); + + var dataProvider = new Mock(); + + var config = new SubscriptionDataConfig(typeof(LocalFileData), Symbols.SPY, Resolution.Daily, TimeZones.NewYork, TimeZones.NewYork, false, false, false); + var request = GetSubscriptionRequest(config, referenceUtc.AddSeconds(-1), referenceUtc.AddDays(1)); + + var factory = new TestableLiveCustomDataSubscriptionEnumeratorFactory(timeProvider, dataSourceReader.Object, + fallBackToBackupUniverseFiles: true); + using var enumerator = factory.CreateEnumerator(request, dataProvider.Object); + + Assert.IsTrue(enumerator.MoveNext()); + + // the expected source is read without any availability probing + VerifyGetSourceInvocationCount(dataSourceReader, 1, "local.file.source", SubscriptionTransportMedium.LocalFile, FileFormat.Csv); + dataProvider.Verify(dp => dp.Fetch(It.IsAny()), Times.Never); + } + + [Test] + public void DoesNotFallBackToBackupUniverseFileWhenNotConfigured() + { + // 10 am, the market is open, but the factory is not configured to fall back to backup universe files + var referenceLocal = new DateTime(2017, 10, 12, 10, 0, 0); + var referenceUtc = referenceLocal.ConvertToUtc(TimeZones.NewYork); + + var timeProvider = new ManualTimeProvider(referenceUtc); + + var dataSourceReader = new Mock(); + dataSourceReader.Setup(dsr => dsr.Read(It.IsAny())) + .Returns(() => new[] { new LocalFileData { EndTime = referenceLocal.AddSeconds(1) } }) + .Verifiable(); + + var dataProvider = new Mock(); + + var config = new SubscriptionDataConfig(typeof(LocalFileData), Symbols.SPY, Resolution.Daily, TimeZones.NewYork, TimeZones.NewYork, false, false, false); + var request = GetSubscriptionRequest(config, referenceUtc.AddSeconds(-1), referenceUtc.AddDays(1)); + + var factory = new TestableLiveCustomDataSubscriptionEnumeratorFactory(timeProvider, dataSourceReader.Object); + using var enumerator = factory.CreateEnumerator(request, dataProvider.Object); + + Assert.IsTrue(enumerator.MoveNext()); + + // the expected source is read without any availability probing + VerifyGetSourceInvocationCount(dataSourceReader, 1, "local.file.source", SubscriptionTransportMedium.LocalFile, FileFormat.Csv); + dataProvider.Verify(dp => dp.Fetch(It.IsAny()), Times.Never); } private static void VerifyGetSourceInvocationCount(Mock dataSourceReader, int count, string source, SubscriptionTransportMedium medium, FileFormat fileFormat) @@ -676,8 +738,8 @@ class TestableLiveCustomDataSubscriptionEnumeratorFactory : LiveCustomDataSubscr private readonly ISubscriptionDataSourceReader _dataSourceReader; public TestableLiveCustomDataSubscriptionEnumeratorFactory(ITimeProvider timeProvider, ISubscriptionDataSourceReader dataSourceReader, - TimeSpan? minimumIntervalCheck = null, Func sourceAdjustment = null) - : base(timeProvider, null, minimumIntervalCheck: minimumIntervalCheck, sourceAdjustment: sourceAdjustment) + TimeSpan? minimumIntervalCheck = null, bool fallBackToBackupUniverseFiles = false) + : base(timeProvider, null, minimumIntervalCheck: minimumIntervalCheck, fallBackToBackupUniverseFiles: fallBackToBackupUniverseFiles) { _dataSourceReader = dataSourceReader; }