Fall back to the backup coarse universe file for fundamental properties in live trading

Common/Data/UniverseSelection/BackupUniverseFileDataProvider.cs

@@ -0,0 +1,109 @@
+/*
+ * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
+ * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+*/
+
+using System;
+using System.IO;
+using QuantConnect.Interfaces;
+using QuantConnect.Logging;
+
+namespace QuantConnect.Data.UniverseSelection
+{
+    /// <summary>
+    /// Data provider wrapper that falls back to the backup universe file ("*.backup"), if any,
+    /// when the expected universe file can't be fetched, as a last resort
+    /// </summary>
+    public class BackupUniverseFileDataProvider : IDataProvider
+    {
+        // the fallback is retried on every universe refresh for as long as the expected file is missing,
+        // so the fallback trace is paced to avoid flooding the logs
+        private const int MaximumLogsPerWindow = 30;
+        private static readonly TimeSpan LogWindow = TimeSpan.FromMinutes(5);
+        private static readonly object LogLock = new();
+        private DateTime _logWindowStartUtc;
+        private int _logCount;
+
+        private IDataProvider _dataProvider;
+
+        /// <summary>
+        /// Event raised each time data fetch is finished (successfully or not)
+        /// </summary>
+        public event EventHandler<DataProviderNewDataRequestEventArgs> NewDataRequest
+        {
+            add => _dataProvider?.NewDataRequest += value;
+            remove => _dataProvider?.NewDataRequest -= value;
+        }
+
+        /// <summary>
+        /// Creates a new instance
+        /// </summary>
+        /// <param name="dataProvider">The data provider to wrap, can be set later with <see cref="SetDataProvider"/></param>
+        public BackupUniverseFileDataProvider(IDataProvider dataProvider = null)
+        {
+            _dataProvider = dataProvider;
+        }
+
+        /// <summary>
+        /// Sets the data provider to wrap, forwarding its <see cref="IDataProvider.NewDataRequest"/> events
+        /// </summary>
+        /// <param name="dataProvider">The data provider to wrap</param>
+        public void SetDataProvider(IDataProvider dataProvider)
+        {
+            _dataProvider = dataProvider;
+        }
+
+        /// <summary>
+        /// Retrieves data to be used in an algorithm, falling back to the backup universe file, if any,
+        /// when the requested file is not available
+        /// </summary>
+        /// <param name="key">A string representing where the data is stored</param>
+        /// <returns>A <see cref="Stream"/> of the data requested, or null if none is available</returns>
+        public Stream Fetch(string key)
+        {
+            var stream = _dataProvider.Fetch(key);
+            if (stream != null)
+            {
+                return stream;
+            }
+
+            var backupKey = key + ".backup";
+            stream = _dataProvider.Fetch(backupKey);
+            if (stream != null && ShouldLog())
+            {
+                Log.Trace($"BackupUniverseFileDataProvider.Fetch(): universe file '{key}' is not available, " +
+                    $"falling back to backup universe file '{backupKey}'");
+            }
+
+            return stream;
+        }
+
+        /// <summary>
+        /// Determines whether the fallback should be logged, allowing up to <see cref="MaximumLogsPerWindow"/> logs per <see cref="LogWindow"/>
+        /// </summary>
+        private bool ShouldLog()
+        {
+            lock (LogLock)
+            {
+                var utcNow = DateTime.UtcNow;
+                if (utcNow - _logWindowStartUtc >= LogWindow)
+                {
+                    _logWindowStartUtc = utcNow;
+                    _logCount = 0;
+                }
+
+                return _logCount++ < MaximumLogsPerWindow;
+            }
+        }
+    }
+}

Common/Data/UniverseSelection/CoarseFundamentalDataProvider.cs

@@ -17,6 +17,7 @@
 using System.IO;
 using System.Collections.Generic;
 using QuantConnect.Data.Fundamental;
+using QuantConnect.Interfaces;
 
 namespace QuantConnect.Data.UniverseSelection
 {
@@ -28,6 +29,22 @@ public class CoarseFundamentalDataProvider : BaseFundamentalDataProvider
         private DateTime _date;
         private readonly Dictionary<SecurityIdentifier, CoarseFundamental> _coarseFundamental = new();
 
+        /// <summary>
+        /// Initializes the service
+        /// </summary>
+        /// <param name="dataProvider">The data provider instance to use</param>
+        /// <param name="liveMode">True if running in live mode</param>
+        public override void Initialize(IDataProvider dataProvider, bool liveMode)
+        {
+            base.Initialize(dataProvider, liveMode);
+            if (liveMode)
+            {
+                // in live trading, fall back to the backup coarse universe file, if any, as a last resort,
+                // consistent with the universe selection data itself
+                DataProvider = new BackupUniverseFileDataProvider(dataProvider);
+            }
+        }
+
         /// <summary>
         /// Will fetch the requested fundamental information for the requested time and symbol
         /// </summary>

Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactory.cs

@@ -16,14 +16,12 @@
 
 using System;
 using System.Collections.Generic;
-using System.IO;
 using System.Linq;
 using Python.Runtime;
 using QuantConnect.Configuration;
 using QuantConnect.Data;
 using QuantConnect.Data.UniverseSelection;
 using QuantConnect.Interfaces;
-using QuantConnect.Logging;
 using QuantConnect.Securities;
 using QuantConnect.Util;
 
@@ -261,50 +259,5 @@ private static TimeSpan GetMaximumDataAge(TimeSpan increment)
         {
             return TimeSpan.FromTicks(Math.Max(increment.Ticks, TimeSpan.FromSeconds(5).Ticks));
         }
-
-        /// <summary>
-        /// Data provider wrapper that falls back to the backup universe file ("*.backup"), if any,
-        /// when the expected universe file can't be fetched, as a last resort
-        /// </summary>
-        private sealed class BackupUniverseFileDataProvider : IDataProvider
-        {
-            private IDataProvider _dataProvider;
-
-            /// <summary>
-            /// Event raised each time data fetch is finished (successfully or not)
-            /// </summary>
-            public event EventHandler<DataProviderNewDataRequestEventArgs> NewDataRequest
-            {
-                add => _dataProvider?.NewDataRequest += value;
-                remove => _dataProvider?.NewDataRequest -= value;
-            }
-
-            /// <summary>
-            /// Sets the data provider to wrap, forwarding its <see cref="IDataProvider.NewDataRequest"/> events
-            /// </summary>
-            public void SetDataProvider(IDataProvider dataProvider)
-            {
-                _dataProvider = dataProvider;
-            }
-
-            public Stream Fetch(string key)
-            {
-                var stream = _dataProvider.Fetch(key);
-                if (stream != null)
-                {
-                    return stream;
-                }
-
-                var backupKey = key + ".backup";
-                stream = _dataProvider.Fetch(backupKey);
-                if (stream != null)
-                {
-                    Log.Trace($"LiveCustomDataSubscriptionEnumeratorFactory.BackupUniverseFileDataProvider.Fetch(): universe file '{key}' is not available, " +
-                        $"falling back to backup universe file '{backupKey}'");
-                }
-
-                return stream;
-            }
-        }
     }
 }

Tests/Common/Data/UniverseSelection/BackupUniverseFileDataProviderTests.cs

@@ -0,0 +1,135 @@
+/*
+ * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
+ * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+*/
+
+using System;
+using System.IO;
+using System.Linq;
+using Moq;
+using NUnit.Framework;
+using QuantConnect.Data.UniverseSelection;
+using QuantConnect.Interfaces;
+using QuantConnect.Logging;
+
+namespace QuantConnect.Tests.Common.Data.UniverseSelection
+{
+    [TestFixture]
+    public class BackupUniverseFileDataProviderTests
+    {
+        private const string Key = "universes/20250815.csv";
+        private const string BackupKey = "universes/20250815.csv.backup";
+
+        [Test]
+        public void ReturnsTheExpectedFileWithoutTouchingTheBackupFile()
+        {
+            var dataProvider = new Mock<IDataProvider>();
+            dataProvider.Setup(dp => dp.Fetch(Key)).Returns(() => new MemoryStream());
+            var backupDataProvider = new BackupUniverseFileDataProvider(dataProvider.Object);
+
+            using var stream = backupDataProvider.Fetch(Key);
+
+            Assert.IsNotNull(stream);
+            dataProvider.Verify(dp => dp.Fetch(Key), Times.Once);
+            dataProvider.Verify(dp => dp.Fetch(BackupKey), Times.Never);
+        }
+
+        [Test]
+        public void FallsBackToTheBackupFileWhenTheExpectedFileIsNotAvailable()
+        {
+            var dataProvider = new Mock<IDataProvider>();
+            dataProvider.Setup(dp => dp.Fetch(BackupKey)).Returns(() => new MemoryStream());
+            var backupDataProvider = new BackupUniverseFileDataProvider(dataProvider.Object);
+
+            using var stream = backupDataProvider.Fetch(Key);
+
+            Assert.IsNotNull(stream);
+            dataProvider.Verify(dp => dp.Fetch(Key), Times.Once);
+            dataProvider.Verify(dp => dp.Fetch(BackupKey), Times.Once);
+        }
+
+        [Test]
+        public void ReturnsNullWhenNeitherTheExpectedNorTheBackupFilesAreAvailable()
+        {
+            var dataProvider = new Mock<IDataProvider>();
+            var backupDataProvider = new BackupUniverseFileDataProvider(dataProvider.Object);
+
+            using var stream = backupDataProvider.Fetch(Key);
+
+            Assert.IsNull(stream);
+            dataProvider.Verify(dp => dp.Fetch(Key), Times.Once);
+            dataProvider.Verify(dp => dp.Fetch(BackupKey), Times.Once);
+        }
+
+        [Test]
+        public void UsesTheDataProviderSetAfterConstruction()
+        {
+            var dataProvider = new Mock<IDataProvider>();
+            dataProvider.Setup(dp => dp.Fetch(BackupKey)).Returns(() => new MemoryStream());
+            var backupDataProvider = new BackupUniverseFileDataProvider();
+            backupDataProvider.SetDataProvider(dataProvider.Object);
+
+            using var stream = backupDataProvider.Fetch(Key);
+
+            Assert.IsNotNull(stream);
+            dataProvider.Verify(dp => dp.Fetch(BackupKey), Times.Once);
+        }
+
+        [Test]
+        public void PacesTheFallbackLogging()
+        {
+            var dataProvider = new Mock<IDataProvider>();
+            dataProvider.Setup(dp => dp.Fetch(It.Is<string>(key => key.EndsWith(".backup")))).Returns(() => new MemoryStream());
+            var backupDataProvider = new BackupUniverseFileDataProvider(dataProvider.Object);
+
+            var previousLogHandler = Log.LogHandler;
+            var logHandler = new QueueLogHandler();
+            Log.LogHandler = logHandler;
+            try
+            {
+                // distinct keys so that the Log's own identical consecutive message protection does not kick in
+                for (var i = 0; i < 100; i++)
+                {
+                    using var stream = backupDataProvider.Fetch($"universes/{i:D8}.csv");
+                    Assert.IsNotNull(stream);
+                }
+            }
+            finally
+            {
+                Log.LogHandler = previousLogHandler;
+            }
+
+            // every fetch fell back to the backup file, but only the first few of them were logged
+            dataProvider.Verify(dp => dp.Fetch(It.Is<string>(key => key.EndsWith(".backup"))), Times.Exactly(100));
+            var fallbackLogs = logHandler.Logs.Count(log => log.Message.Contains("falling back to backup universe file", StringComparison.InvariantCulture));
+            Assert.AreEqual(30, fallbackLogs);
+        }
+
+        [Test]
+        public void ForwardsNewDataRequestEventsToTheWrappedDataProvider()
+        {
+            var dataProvider = new Mock<IDataProvider>();
+            var backupDataProvider = new BackupUniverseFileDataProvider(dataProvider.Object);
+
+            var raised = 0;
+            EventHandler<DataProviderNewDataRequestEventArgs> handler = (_, _) => raised++;
+            backupDataProvider.NewDataRequest += handler;
+            dataProvider.Raise(dp => dp.NewDataRequest += null, new DataProviderNewDataRequestEventArgs(Key, true, ""));
+            Assert.AreEqual(1, raised);
+
+            backupDataProvider.NewDataRequest -= handler;
+            dataProvider.Raise(dp => dp.NewDataRequest += null, new DataProviderNewDataRequestEventArgs(Key, true, ""));
+            Assert.AreEqual(1, raised);
+        }
+    }
+}