From 22c8ee4c236172367ef76ea13f7d94fa52942cb2 Mon Sep 17 00:00:00 2001 From: Jhonathan Abreu Date: Wed, 12 Aug 2026 09:39:04 -0400 Subject: [PATCH] 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 e6e01bc01067..2958a4e8dd16 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); diff --git a/Engine/DataFeeds/LiveTradingDataFeed.cs b/Engine/DataFeeds/LiveTradingDataFeed.cs index b6bf32b3f71c..669bfe743a5d 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 675d8de49efd..7638596db30d 100644 --- a/Tests/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactoryTests.cs +++ b/Tests/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactoryTests.cs @@ -520,6 +520,51 @@ public void AllowsSpecifyingIntervalCheck(int intervalCheck) VerifyGetSourceInvocationCount(dataSourceReader, 2, "local.file.source", SubscriptionTransportMedium.LocalFile, FileFormat.Csv); } + [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 => @@ -593,8 +638,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() {