Fall back to backup universe files in live trading when the expected ones are unavailable

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,9 +32,15 @@ namespace QuantConnect.Lean.Engine.DataFeeds.Enumerators.Factories
     /// </summary>
     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<DateTime, DateTime> _dateAdjustment;
+        private readonly bool _fallBackToBackupUniverseFiles;
         private readonly IObjectStore _objectStore;
 
         /// <summary>
@@ -42,12 +50,17 @@ 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="fallBackToBackupUniverseFiles">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</param>
         public LiveCustomDataSubscriptionEnumeratorFactory(ITimeProvider timeProvider, IObjectStore objectStore,
-            Func<DateTime, DateTime> dateAdjustment = null, TimeSpan? minimumIntervalCheck = null)
+            Func<DateTime, DateTime> dateAdjustment = null, TimeSpan? minimumIntervalCheck = null,
+            bool fallBackToBackupUniverseFiles = false)
         {
             _timeProvider = timeProvider;
             _dateAdjustment = dateAdjustment;
             _minimumIntervalCheck = minimumIntervalCheck ?? TimeSpan.FromMinutes(30);
+            _fallBackToBackupUniverseFiles = fallBackToBackupUniverseFiles;
             _objectStore = objectStore;
         }
 
@@ -66,6 +79,7 @@ public IEnumerator<BaseData> 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<BaseData>(() =>
@@ -81,6 +95,10 @@ public IEnumerator<BaseData> CreateEnumerator(SubscriptionRequest request, IData
                 lastSourceRefreshTime = utcNow;
                 var localDate = _dateAdjustment?.Invoke(utcNow.ConvertFromUtc(config.ExchangeTimeZone).Date) ?? utcNow.ConvertFromUtc(config.ExchangeTimeZone).Date;
                 var source = sourceFactory.GetSource(config, localDate, true);
+                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);
@@ -197,6 +215,55 @@ IDataProvider dataProvider
             return SubscriptionDataSourceReader.ForSource(source, dataCacheProvider, config, date, true, baseDataInstance, dataProvider, _objectStore);
         }
 
+        /// <summary>
+        /// 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 <see cref="UniverseFileBackupFallbackWindow"/>
+        /// 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
+        /// </summary>
+        private static Func<SubscriptionDataSource, DateTime, SubscriptionDataSource> 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

Engine/DataFeeds/LiveTradingDataFeed.cs

@@ -356,7 +356,9 @@ 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),
+                    // 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);
 

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;
@@ -520,6 +521,112 @@ public void AllowsSpecifyingIntervalCheck(int intervalCheck)
             VerifyGetSourceInvocationCount(dataSourceReader, 2, "local.file.source", SubscriptionTransportMedium.LocalFile, FileFormat.Csv);
         }
 
+        [Test]
+        public void FallsBackToBackupUniverseFileWhenExpectedSourceIsNotAvailable()
+        {
+            // 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);
+
+            var dataSourceReader = new Mock<ISubscriptionDataSourceReader>();
+            dataSourceReader.Setup(dsr => dsr.Read(It.IsAny<SubscriptionDataSource>()))
+                .Returns(() => new[] { new LocalFileData { EndTime = referenceLocal.AddSeconds(1) } })
+                .Verifiable();
+
+            var expectedSourceAvailable = false;
+            var dataProvider = new Mock<IDataProvider>();
+            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 factory = new TestableLiveCustomDataSubscriptionEnumeratorFactory(timeProvider, dataSourceReader.Object,
+                fallBackToBackupUniverseFiles: true);
+            using var enumerator = factory.CreateEnumerator(request, dataProvider.Object);
+
+            Assert.IsTrue(enumerator.MoveNext());
+            Assert.IsNotNull(enumerator.Current);
+
+            // 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<string>()), Times.Exactly(2));
+
+            // the fallback checks are rate limited like the source refreshes
+            Assert.IsTrue(enumerator.MoveNext());
+            Assert.IsNull(enumerator.Current);
+            dataProvider.Verify(dp => dp.Fetch(It.IsAny<string>()), 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());
+            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<ISubscriptionDataSourceReader>();
+            dataSourceReader.Setup(dsr => dsr.Read(It.IsAny<SubscriptionDataSource>()))
+                .Returns(() => new[] { new LocalFileData { EndTime = referenceLocal.AddSeconds(1) } })
+                .Verifiable();
+
+            var dataProvider = new Mock<IDataProvider>();
+
+            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<string>()), 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<ISubscriptionDataSourceReader>();
+            dataSourceReader.Setup(dsr => dsr.Read(It.IsAny<SubscriptionDataSource>()))
+                .Returns(() => new[] { new LocalFileData { EndTime = referenceLocal.AddSeconds(1) } })
+                .Verifiable();
+
+            var dataProvider = new Mock<IDataProvider>();
+
+            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<string>()), Times.Never);
+        }
+
         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 +700,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, bool fallBackToBackupUniverseFiles = false)
+                : base(timeProvider, null, minimumIntervalCheck: minimumIntervalCheck, fallBackToBackupUniverseFiles: fallBackToBackupUniverseFiles)
             {
                 _dataSourceReader = dataSourceReader;
             }