Delay reading option chain universe files until close to market open in live trading

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<DateTime, DateTime> _dateAdjustment;
+        private readonly Func<DateTime, bool> _canRefresh;
         private readonly IObjectStore _objectStore;
 
         /// <summary>
@@ -42,12 +43,15 @@ public class LiveCustomDataSubscriptionEnumeratorFactory : ISubscriptionEnumerat
         /// <param name="objectStore">The object store to use</param>
         /// <param name="dateAdjustment">Func that allows adjusting the datetime to use</param>
         /// <param name="minimumIntervalCheck">Allows specifying the minimum interval between each enumerator refresh and data check, default is 30 minutes</param>
+        /// <param name="canRefresh">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</param>
         public LiveCustomDataSubscriptionEnumeratorFactory(ITimeProvider timeProvider, IObjectStore objectStore,
-            Func<DateTime, DateTime> dateAdjustment = null, TimeSpan? minimumIntervalCheck = null)
+            Func<DateTime, DateTime> dateAdjustment = null, TimeSpan? minimumIntervalCheck = null, Func<DateTime, bool> canRefresh = null)
         {
             _timeProvider = timeProvider;
             _dateAdjustment = dateAdjustment;
             _minimumIntervalCheck = minimumIntervalCheck ?? TimeSpan.FromMinutes(30);
+            _canRefresh = canRefresh;
             _objectStore = objectStore;
         }
 
@@ -79,6 +83,13 @@ public IEnumerator<BaseData> 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<BaseData>().GetEnumerator();
+                }
+
                 var localDate = _dateAdjustment?.Invoke(utcNow.ConvertFromUtc(config.ExchangeTimeZone).Date) ?? utcNow.ConvertFromUtc(config.ExchangeTimeZone).Date;
                 var source = sourceFactory.GetSource(config, localDate, true);
 

Engine/DataFeeds/LiveTradingDataFeed.cs

@@ -61,6 +61,10 @@ public class LiveTradingDataFeed : FileSystemDataFeed
         private static ReferenceWrapper<DateTime> _lastUtcDateShiftUpdate;
         private static ReferenceWrapper<TimeSpan> _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);
+
         /// <summary>
         /// Public flag indicator that the thread is still busy.
         /// </summary>
@@ -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;
         }
 
+        /// <summary>
+        /// 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 <see cref="PreOpenUniverseFileRefreshWindow"/> of the next market open
+        /// </summary>
+        private static Func<DateTime, bool> 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;
+            };
+        }
+
         /// <summary>
         /// Build and apply the warmup enumerators when required
         /// </summary>

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<ISubscriptionDataSourceReader>();
+            dataSourceReader.Setup(dsr => dsr.Read(It.IsAny<SubscriptionDataSource>()))
+                .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<SubscriptionDataSource>()), 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<SubscriptionDataSource>()), 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<SubscriptionDataSource>()), Times.Once);
+        }
+
         private static void VerifyGetSourceInvocationCount(Mock<ISubscriptionDataSourceReader> dataSourceReader, int count, string source, SubscriptionTransportMedium medium, FileFormat fileFormat)
         {
             dataSourceReader.Verify(dsr => dsr.Read(It.Is<SubscriptionDataSource>(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<DateTime, bool> canRefresh = null)
+                : base(timeProvider, null, minimumIntervalCheck: minimumIntervalCheck, canRefresh: canRefresh)
             {
                 _dataSourceReader = dataSourceReader;
             }

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<Symbol> 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<Symbol>)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()
         {