diff --git a/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactory.cs b/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactory.cs index e6e01bc01067..2963d1c44f7b 100644 --- a/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactory.cs +++ b/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 /// 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 _dateAdjustment; + private readonly bool _fallBackToBackupUniverseFiles; private readonly IObjectStore _objectStore; /// @@ -42,12 +50,17 @@ 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 + /// 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 public LiveCustomDataSubscriptionEnumeratorFactory(ITimeProvider timeProvider, IObjectStore objectStore, - Func dateAdjustment = null, TimeSpan? minimumIntervalCheck = null) + Func 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 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(() => @@ -81,6 +95,10 @@ public IEnumerator 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); } + /// + /// 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 + /// 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 + /// + private static Func 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 diff --git a/Engine/DataFeeds/LiveTradingDataFeed.cs b/Engine/DataFeeds/LiveTradingDataFeed.cs index b6bf32b3f71c..b57df8936b80 100644 --- a/Engine/DataFeeds/LiveTradingDataFeed.cs +++ b/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); diff --git a/Tests/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactoryTests.cs b/Tests/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactoryTests.cs index 675d8de49efd..dd3d900d66a9 100644 --- a/Tests/Engine/DataFeeds/Enumerators/Factories/LiveCustomDataSubscriptionEnumeratorFactoryTests.cs +++ b/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(); + dataSourceReader.Setup(dsr => dsr.Read(It.IsAny())) + .Returns(() => new[] { new LocalFileData { EndTime = referenceLocal.AddSeconds(1) } }) + .Verifiable(); + + var expectedSourceAvailable = false; + var dataProvider = new Mock(); + 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()), 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()), 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(); + dataSourceReader.Setup(dsr => dsr.Read(It.IsAny())) + .Returns(() => new[] { new LocalFileData { EndTime = referenceLocal.AddSeconds(1) } }) + .Verifiable(); + + var dataProvider = new Mock(); + + 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()), 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(); + dataSourceReader.Setup(dsr => dsr.Read(It.IsAny())) + .Returns(() => new[] { new LocalFileData { EndTime = referenceLocal.AddSeconds(1) } }) + .Verifiable(); + + var dataProvider = new Mock(); + + 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()), Times.Never); + } + private static void VerifyGetSourceInvocationCount(Mock dataSourceReader, int count, string source, SubscriptionTransportMedium medium, FileFormat fileFormat) { dataSourceReader.Verify(dsr => dsr.Read(It.Is(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; } diff --git a/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs b/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs index 74c4d2ad2df5..4c4cc80e799a 100644 --- a/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs +++ b/Tests/Engine/DataFeeds/LiveTradingDataFeedTests.cs @@ -17,6 +17,7 @@ using System; using System.Collections.Generic; using System.Diagnostics; +using System.IO; using System.Linq; using System.Threading; using Microsoft.CodeAnalysis; @@ -282,6 +283,142 @@ public void LiveChainSelection(SecurityType securityType, Resolution resolution, Assert.AreEqual(expectedSelections, selectionHappened); } + [TestCase("OptionChain", false)] + [TestCase("OptionChain", true)] + [TestCase("IndexOptionChain", false)] + [TestCase("IndexOptionChain", true)] + [TestCase("CoarseFundamental", false)] + [TestCase("CoarseFundamental", true)] + [TestCase("EtfConstituents", false)] + [TestCase("EtfConstituents", true)] + public void UniverseSelectionFallsBackToBackupUniverseFileCloseToMarketOpen(string universeKind, bool universeFileAvailable) + { + // start close to the market open (9:15 NY), within the backup universe file fallback window (30 minutes before the open by default) + _startDate = universeKind switch + { + "OptionChain" => new DateTime(2014, 6, 9, 13, 15, 0), + "IndexOptionChain" => new DateTime(2021, 1, 4, 14, 15, 0), + "CoarseFundamental" => new DateTime(2014, 3, 26, 13, 15, 0), + "EtfConstituents" => new DateTime(2020, 12, 1, 14, 15, 0), + _ => throw new ArgumentException($"Unexpected universe kind: {universeKind}") + }; + _manualTimeProvider.SetCurrentTimeUtc(_startDate); + var endDate = _startDate.AddDays(1); + + _algorithm.SetBenchmark(x => 1); + + var dataProvider = new BackupUniverseFileDataProvider(hideUniverseFiles: !universeFileAvailable); + var feed = RunDataFeed(runPostInitialize: false, dataProvider: dataProvider); + + var selectionHappened = 0; + var selectedCount = 0; + + IEnumerable CoarseFilter(IEnumerable coarse) + { + selectionHappened++; + var symbols = coarse.Select(x => x.Symbol).ToList(); + selectedCount = symbols.Count; + return symbols; + } + + switch (universeKind) + { + case "OptionChain": + case "IndexOptionChain": + var option = universeKind == "OptionChain" + ? _algorithm.AddOption("AAPL") + : _algorithm.AddIndexOption("SPX"); + option.SetFilter(universe => + { + selectionHappened++; + selectedCount = universe.Count(); + return universe; + }); + break; + + case "CoarseFundamental": + _algorithm.UniverseSettings.Resolution = Resolution.Daily; + _algorithm.AddUniverse(CoarseFilter); + break; + + case "EtfConstituents": + var spy = _algorithm.AddEquity("SPY").Symbol; + _algorithm.AddUniverse(_algorithm.Universe.ETF(spy, constituentsData => + { + selectionHappened++; + var symbols = constituentsData.Select(x => x.Symbol).ToList(); + selectedCount = symbols.Count; + return symbols; + })); + break; + } + + _algorithm.PostInitialize(); + + // allow time for the exchange to pick up the selection point + Thread.Sleep(50); + + ConsumeBridge(feed, TimeSpan.FromSeconds(30), true, ts => + { + if (selectionHappened > 0) + { + // we got what we wanted shortcut unit test + _manualTimeProvider.SetCurrentTimeUtc(Time.EndOfTime); + } + }, + endDate: endDate, + secondsTimeStep: 60); + + Assert.AreEqual(1, selectionHappened); + Assert.AreNotEqual(0, selectedCount); + + if (universeFileAvailable) + { + // the universe file was available, so the backup file should not have even been checked + Assert.AreEqual(0, dataProvider.BackupUniverseFileRequests); + } + else + { + Assert.AreNotEqual(0, dataProvider.UniverseFileRequests); + Assert.AreNotEqual(0, dataProvider.BackupUniverseFileRequests); + } + } + + [Test] + public void ChainSelectionDoesNotFallBackToBackupUniverseFileFarFromMarketOpen() + { + // start during the night: far from the market open, the missing universe file should not fall back to the backup file + _startDate = new DateTime(2014, 6, 9, 6, 0, 0); + _manualTimeProvider.SetCurrentTimeUtc(_startDate); + // stop before entering the fallback window, 30 minutes (by default) before the 9:30 NY market open + var endDate = new DateTime(2014, 6, 9, 12, 0, 0); + + _algorithm.SetBenchmark(x => 1); + + var dataProvider = new BackupUniverseFileDataProvider(hideUniverseFiles: true); + var feed = RunDataFeed(runPostInitialize: false, dataProvider: dataProvider); + + var selectionHappened = 0; + var option = _algorithm.AddOption("AAPL"); + option.SetFilter(universe => + { + selectionHappened++; + return universe; + }); + + _algorithm.PostInitialize(); + + // allow time for the exchange to pick up the selection point + Thread.Sleep(50); + + ConsumeBridge(feed, TimeSpan.FromSeconds(30), true, ts => { }, endDate: endDate, secondsTimeStep: 60); + + // the universe file was tried but never available, and the backup file should not have been used + Assert.AreEqual(0, selectionHappened); + Assert.AreNotEqual(0, dataProvider.UniverseFileRequests); + Assert.AreEqual(0, dataProvider.BackupUniverseFileRequests); + } + [Test] public void ContinuousFuturesImmediateSelection() { @@ -2914,7 +3051,7 @@ private IDataFeed RunDataFeed(Resolution resolution = Resolution.Second, List> getNextTicksFunction = null, Func> lookupSymbolsFunction = null, Func canPerformSelection = null, IDataQueueHandler dataQueueHandler = null, - bool runPostInitialize = true) + bool runPostInitialize = true, IDataProvider dataProvider = null) { _algorithm.SetStartDate(_startDate); _algorithm.SetDateTime(_manualTimeProvider.GetUtcNow()); @@ -2988,7 +3125,7 @@ private IDataFeed RunDataFeed(Resolution resolution = Resolution.Second, List _universeFileRequests; + public int BackupUniverseFileRequests => _backupUniverseFileRequests; + + public event EventHandler NewDataRequest; + + public BackupUniverseFileDataProvider(bool hideUniverseFiles) + { + _hideUniverseFiles = hideUniverseFiles; + } + + public Stream Fetch(string key) + { + // coarse fundamental files are universe files too, they just don't live under a "universes" folder + if (key.Contains("universes", StringComparison.InvariantCulture) + || key.Replace('\\', '/').Contains("fundamental/coarse", StringComparison.InvariantCulture)) + { + if (key.EndsWith(".csv.backup", StringComparison.InvariantCulture)) + { + Interlocked.Increment(ref _backupUniverseFileRequests); + // serve the backup universe file contents from the actual universe file + return _dataProvider.Fetch(key.Substring(0, key.Length - ".backup".Length)); + } + + if (key.EndsWith(".csv", StringComparison.InvariantCulture)) + { + Interlocked.Increment(ref _universeFileRequests); + if (_hideUniverseFiles) + { + return null; + } + } + } + + return _dataProvider.Fetch(key); + } + } + private static IEnumerable ProduceBenchmarkTicks(FuncDataQueueHandler fdqh, Count count) { for (int i = 0; i < 10000; i++)