Move the backup universe file fallback logic into the live custom data enumerator factory

Author: jhonabreul

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
     /// </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 Func<SubscriptionDataSource, DateTime, SubscriptionDataSource> _sourceAdjustment;
+        private readonly bool _fallBackToBackupUniverseFiles;
         private readonly IObjectStore _objectStore;
 
         /// <summary>
@@ -43,17 +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="sourceAdjustment">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</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<SubscriptionDataSource, DateTime, SubscriptionDataSource> 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<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>(() =>
@@ -87,9 +95,9 @@ 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)
+                if (sourceAdjustment != null)
                 {
-                    source = _sourceAdjustment(source, utcNow);
+                    source = sourceAdjustment(source, utcNow);
                 }
 
                 // fetch the new source and enumerate the data source reader
@@ -207,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

@@ -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;
         }
 
-        /// <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
-        /// </summary>
-        private Func<SubscriptionDataSource, DateTime, SubscriptionDataSource> 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;
-        }
-
         /// <summary>
         /// Build and apply the warmup enumerators when required
         /// </summary>