diff --git a/Algorithm/QCAlgorithm.History.cs b/Algorithm/QCAlgorithm.History.cs index 077b4cbda32f..afdd00def57a 100644 --- a/Algorithm/QCAlgorithm.History.cs +++ b/Algorithm/QCAlgorithm.History.cs @@ -38,6 +38,8 @@ public partial class QCAlgorithm private bool _dataDictionaryTickWarningSent; + private readonly LargeHistoryRequestDiagnostics _largeHistoryRequestDiagnostics = new(); + /// /// Gets or sets the history provider for the algorithm /// @@ -1019,7 +1021,9 @@ protected IEnumerable> GetDataTypedHistory(IEnumerable History(IEnumerable requests, DateTimeZone timeZone) { // filter out any universe securities that may have made it this far - var filteredRequests = GetFilterestRequests(requests); + var filteredRequests = GetFilterestRequests(requests).ToList(); + + _largeHistoryRequestDiagnostics.WarnOnLargeHistoryRequests(filteredRequests, Debug); // filter out future data to prevent look ahead bias var history = HistoryProvider.GetHistory(filteredRequests, timeZone); @@ -1555,5 +1559,119 @@ private static IEnumerable WrapPythonDataHistory(IEnumerable histo } } } + + /// + /// One-time warnings for history requests likely to stall the algorithm or run it out of memory: + /// a single call over the cells threshold, or repeated large calls with overlapping time windows, + /// e.g. re-fetching a long lookback every day instead of updating incrementally + /// + private sealed class LargeHistoryRequestDiagnostics + { + private const int OverlappingRequestsWarningCount = 30; + private const int OverlappingRequestSizeDivisor = 50; + private const int AssumedTicksPerSecond = 10; + + private readonly long _cellsWarningThreshold = Config.GetInt("history-request-cells-warning-threshold", 5000000); + private bool _largeRequestWarningSent; + private bool _overlappingRequestsWarningSent; + private int _consecutiveOverlappingRequests; + private DateTime _previousRequestStartUtc; + private DateTime _previousRequestEndUtc; + + /// + /// Checks the given requests and emits the applicable warnings through the given debug callback + /// + public void WarnOnLargeHistoryRequests(List requests, Action debug) + { + try + { + if (_largeRequestWarningSent && _overlappingRequestsWarningSent || requests.Count == 0) + { + return; + } + + long estimatedCells = 0; + var startUtc = DateTime.MaxValue; + var endUtc = DateTime.MinValue; + foreach (var request in requests) + { + estimatedCells += EstimateDataCells(request); + if (request.StartTimeUtc < startUtc) + { + startUtc = request.StartTimeUtc; + } + if (request.EndTimeUtc > endUtc) + { + endUtc = request.EndTimeUtc; + } + } + + if (!_largeRequestWarningSent && estimatedCells >= _cellsWarningThreshold) + { + _largeRequestWarningSent = true; + debug($"Warning: large history request, estimated at ~{estimatedCells.ToStringInvariant("N0")} data cells" + + $" across {requests.Count} request(s). This can be slow and memory intensive: request fewer symbols," + + " a shorter period or a coarser resolution."); + } + + if (!_overlappingRequestsWarningSent && estimatedCells >= _cellsWarningThreshold / OverlappingRequestSizeDivisor) + { + var overlaps = startUtc < _previousRequestEndUtc && endUtc > _previousRequestStartUtc; + _consecutiveOverlappingRequests = overlaps ? _consecutiveOverlappingRequests + 1 : 0; + _previousRequestStartUtc = startUtc; + _previousRequestEndUtc = endUtc; + + if (_consecutiveOverlappingRequests >= OverlappingRequestsWarningCount) + { + _overlappingRequestsWarningSent = true; + debug($"Warning: history() has been called {OverlappingRequestsWarningCount}+ consecutive times with" + + " overlapping time windows. Instead of re-fetching a long lookback, fetch it once and keep it updated" + + " with a rolling window, consolidators or indicator warm up."); + } + } + } + catch (Exception exception) + { + // diagnostics must never interfere with the algorithm: log, disable and move on + _largeRequestWarningSent = true; + _overlappingRequestsWarningSent = true; + QuantConnect.Logging.Log.Error(exception); + } + } + + /// + /// Rough order-of-magnitude estimate of the data cells (bars x columns) a history request will produce + /// + private static long EstimateDataCells(HistoryRequest request) + { + var span = request.EndTimeUtc - request.StartTimeUtc; + if (span <= TimeSpan.Zero) + { + return 0; + } + + // ~5 tradable days a week; sub-daily resolutions only have data during regular market hours + var tradableDays = Math.Max(1, span.TotalDays * 5 / 7); + double bars; + switch (request.Resolution) + { + case Resolution.Daily: + bars = tradableDays; + break; + case Resolution.Tick: + // unknowable upfront: liquid symbols see ~10 trades and 100+ quotes per second, + // so 10 data points per second of market time is still on the low side + bars = tradableDays * request.ExchangeHours.RegularMarketDuration.TotalSeconds * AssumedTicksPerSecond; + break; + default: + bars = tradableDays * (request.ExchangeHours.RegularMarketDuration / request.Resolution.ToTimeSpan()); + break; + } + + // approximate data frame column counts per data point + var columns = request.TickType == TickType.Quote ? 10 : request.TickType == TickType.OpenInterest ? 1 : 5; + return (long)(bars * columns); + } + } } } diff --git a/Common/IsolatorLimitResultProvider.cs b/Common/IsolatorLimitResultProvider.cs index 5bf7d47b5f89..d19ecf273774 100644 --- a/Common/IsolatorLimitResultProvider.cs +++ b/Common/IsolatorLimitResultProvider.cs @@ -45,7 +45,7 @@ TimeMonitor timeMonitor } var timeProvider = RealTimeProvider.Instance; - isolatorLimitProvider.Consume(timeProvider, () => scheduledEvent.Scan(scanTimeUtc), timeMonitor); + isolatorLimitProvider.Consume(timeProvider, () => scheduledEvent.Scan(scanTimeUtc), timeMonitor, scheduledEvent.Name); } /// @@ -63,13 +63,15 @@ public static void Consume( this IIsolatorLimitResultProvider isolatorLimitProvider, ITimeProvider timeProvider, Action code, - TimeMonitor timeMonitor + TimeMonitor timeMonitor, + string name = null ) { var consumer = new TimeConsumer { IsolatorLimitProvider = isolatorLimitProvider, - TimeProvider = timeProvider + TimeProvider = timeProvider, + Name = name }; timeMonitor.Add(consumer); code(); diff --git a/Common/Scheduling/TimeConsumer.cs b/Common/Scheduling/TimeConsumer.cs index 0ddf9a2e82bb..6936165a8f20 100644 --- a/Common/Scheduling/TimeConsumer.cs +++ b/Common/Scheduling/TimeConsumer.cs @@ -42,5 +42,16 @@ public class TimeConsumer /// to be /// public DateTime? NextTimeRequest { get; set; } + + /// + /// Name of the work being executed, if any, e.g. the scheduled event's name. Used to name the + /// long-running work in logs when additional time is requested + /// + public string Name { get; set; } + + /// + /// The number of additional minutes that have been requested for this consumer so far + /// + public int AdditionalMinutesRequested { get; set; } } } diff --git a/Common/Scheduling/TimeMonitor.cs b/Common/Scheduling/TimeMonitor.cs index b302257d6c94..3285593ee374 100644 --- a/Common/Scheduling/TimeMonitor.cs +++ b/Common/Scheduling/TimeMonitor.cs @@ -16,6 +16,7 @@ using System; using System.Threading; using QuantConnect.Util; +using QuantConnect.Logging; using System.Collections.Generic; namespace QuantConnect.Scheduling @@ -35,6 +36,12 @@ public class TimeMonitor : IDisposable /// in `IsolatorLimitResultProviderTests.cs protected List TimeConsumers { get; init; } + /// + /// Optional handler used to also surface long-running work warnings to the user, + /// e.g. through the result handler's debug messages. Engine logs alone don't reach the user's logs + /// + public Action UserWarningHandler { get; set; } + /// /// Returns the number of time consumers currently being monitored /// @@ -107,6 +114,22 @@ protected virtual void ProcessConsumer(TimeConsumer consumer) { // pass } + + consumer.AdditionalMinutesRequested++; + if (consumer.Name != null) + { + // name the long-running work: the first minute crossing is the actionable heads-up, later ones are informational + var message = $"'{consumer.Name}' has been executing for over {consumer.AdditionalMinutesRequested} minute(s)"; + if (consumer.AdditionalMinutesRequested == 1) + { + Log.Error($"TimeMonitor.ProcessConsumer(): {message}. It will be stopped once the algorithm time loop limit is exhausted"); + UserWarningHandler?.Invoke($"Warning: {message}. It will be stopped once the algorithm time loop limit is exhausted"); + } + else + { + Log.Trace($"TimeMonitor.ProcessConsumer(): {message}"); + } + } } } diff --git a/Engine/AlgorithmManager.cs b/Engine/AlgorithmManager.cs index 9bfd29cafdba..10d2853ea005 100644 --- a/Engine/AlgorithmManager.cs +++ b/Engine/AlgorithmManager.cs @@ -101,7 +101,8 @@ public AlgorithmManager(bool liveMode, AlgorithmNodePacket job = null) // initialize the time limit manager TimeLimit = new AlgorithmTimeLimitManager( CreateTokenBucket(job?.Controls?.TrainingLimits), - TimeSpan.FromMinutes(Config.GetDouble("algorithm-manager-time-loop-maximum", 20)) + TimeSpan.FromMinutes(Config.GetDouble("algorithm-manager-time-loop-maximum", 20)), + TimeSpan.FromMinutes(Config.GetDouble("algorithm-manager-time-loop-warning", 3)) ); } @@ -123,6 +124,8 @@ public void Run(AlgorithmNodePacket job, IAlgorithm algorithm, ISynchronizer syn //Initialize: _algorithm = algorithm; _performanceTrackingTool = performanceTrackingTool; + // surface time limit warnings in the user's logs, not just the engine log + TimeLimit.UserWarningHandler = results.DebugMessage; var token = cancellationTokenSource.Token; _cancellationTokenSource = cancellationTokenSource; diff --git a/Engine/AlgorithmTimeLimitManager.cs b/Engine/AlgorithmTimeLimitManager.cs index e564ad5a75af..12d65bf705e3 100644 --- a/Engine/AlgorithmTimeLimitManager.cs +++ b/Engine/AlgorithmTimeLimitManager.cs @@ -1,4 +1,4 @@ -/* +/* * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation. * @@ -35,6 +35,8 @@ public class AlgorithmTimeLimitManager : IIsolatorLimitResultProvider private volatile ReferenceWrapper _currentTimeStepTime; private readonly TimeSpan _timeLoopMaximum; + private readonly TimeSpan _timeLoopWarningThreshold; + private volatile bool _timeStepWarningSent; /// /// Gets the additional time bucket which is responsible for tracking additional time requested @@ -42,6 +44,12 @@ public class AlgorithmTimeLimitManager : IIsolatorLimitResultProvider /// public ITokenBucket AdditionalTimeBucket { get; } + /// + /// Optional handler used to also surface the slow time step warning to the user, + /// e.g. through the result handler's debug messages. Engine logs alone don't reach the user's logs + /// + public Action UserWarningHandler { get; set; } + /// /// Initializes a new instance of to manage the /// creation of instances as it pertains to the @@ -52,9 +60,12 @@ public class AlgorithmTimeLimitManager : IIsolatorLimitResultProvider /// Specifies the maximum amount of time the algorithm is permitted to /// spend in a single time loop. This value can be overriden if certain actions are taken by the /// algorithm, such as invoking the training methods. - public AlgorithmTimeLimitManager(ITokenBucket additionalTimeBucket, TimeSpan timeLoopMaximum) + /// Elapsed time of a single time loop after which a warning is logged, + /// once per time step. Defaults to three minutes; a non positive value disables the warning + public AlgorithmTimeLimitManager(ITokenBucket additionalTimeBucket, TimeSpan timeLoopMaximum, TimeSpan? timeLoopWarningThreshold = null) { _timeLoopMaximum = timeLoopMaximum; + _timeLoopWarningThreshold = timeLoopWarningThreshold ?? TimeSpan.FromMinutes(3); AdditionalTimeBucket = additionalTimeBucket; _currentTimeStepTime = new ReferenceWrapper(DateTime.MinValue); } @@ -80,6 +91,7 @@ public void StartNewTimeStep() // accessing DateTime.UtcNow from the algorithm manager thread to the isolator thread _currentTimeStepTime = new ReferenceWrapper(DateTime.MinValue); Interlocked.Exchange(ref _additionalMinutes, 0L); + _timeStepWarningSent = false; } /// @@ -99,6 +111,22 @@ public IsolatorLimitResult IsWithinLimit() { TimeSpan currentTimeStepElapsed; var message = IsOutOfTime(out currentTimeStepElapsed) ? GetErrorMessage(currentTimeStepElapsed) : string.Empty; + + // warn early about an abnormally long time step: an isolator kill minutes later is opaque, + // the elapsed-time warning is actionable now + if (message.Length == 0 && !_stopped && !_timeStepWarningSent + && _timeLoopWarningThreshold > TimeSpan.Zero && currentTimeStepElapsed > _timeLoopWarningThreshold) + { + _timeStepWarningSent = true; + var warning = $"the current time step has been executing for {currentTimeStepElapsed.TotalMinutes.ToStringInvariant("0.0")} minutes" + + $" and the algorithm will be stopped if it exceeds {_timeLoopMaximum.TotalMinutes.ToStringInvariant()} minutes." + + " Common causes: a slow scheduled event (named in 'TimeMonitor' logs), large history() requests," + + " heavy on_data work or an infinite loop."; + // override flood protection: the same text repeats for each slow time step and each occurrence matters + Log.Error($"AlgorithmTimeLimitManager.IsWithinLimit(): {warning}", overrideMessageFloodProtection: true); + UserWarningHandler?.Invoke($"Warning: {warning}"); + } + return new IsolatorLimitResult(currentTimeStepElapsed, message); } diff --git a/Engine/DataFeeds/UniverseSelection.cs b/Engine/DataFeeds/UniverseSelection.cs index e5512ba46122..4363be9227c0 100644 --- a/Engine/DataFeeds/UniverseSelection.cs +++ b/Engine/DataFeeds/UniverseSelection.cs @@ -17,6 +17,7 @@ using System.Collections.Generic; using System.Linq; using QuantConnect.Benchmarks; +using QuantConnect.Configuration; using QuantConnect.Data; using QuantConnect.Data.UniverseSelection; using QuantConnect.Interfaces; @@ -42,6 +43,19 @@ public class UniverseSelection private bool _initializedSecurityBenchmark; private bool _anyDoesNotHaveFundamentalDataWarningLogged; private readonly SecurityChangesConstructor _securityChangesConstructor; + private bool _universeSelectionSizeWarningSent; + // the cost of a selected symbol grows with the resolution its subscriptions are added at, + // so the warning threshold is defined per resolution instead of as a single flat count + private readonly Dictionary _universeSelectionSizeWarningThresholds = Config.GetValue( + "universe-selection-size-warning-thresholds", + new Dictionary + { + { Resolution.Tick, 100 }, + { Resolution.Second, 250 }, + { Resolution.Minute, 500 }, + { Resolution.Hour, 1000 }, + { Resolution.Daily, 2000 }, + }); /// /// Initializes a new instance of the class @@ -183,6 +197,8 @@ public SecurityChanges ApplyUniverseSelection(Universe universe, DateTime dateTi { // materialize the enumerable into a set for processing universe.Selected = selectSymbolsResult.ToHashSet(); + + WarnOnLargeUniverseSelection(universe); } // first check for no pending removals, even if the universe selection @@ -473,6 +489,110 @@ public SecurityChanges HandleDelisting(BaseData data, bool isInternalFeed) return SecurityChanges.None; } + /// + /// Warns once per algorithm when universe selections grow past the per-resolution symbol thresholds in + /// 'universe-selection-size-warning-thresholds', a recurring cause of out of memory kills and stalls. + /// All universes consume a single shared subscription budget: each resolution's threshold defines the + /// full budget for symbols subscribed at that resolution, and each universe consumes a fraction of it, + /// so load still accumulates across universes of different resolutions. + /// Warn only, never fail: the run may still succeed + /// + internal void WarnOnLargeUniverseSelection(Universe universe) + { + try + { + if (_universeSelectionSizeWarningSent) + { + return; + } + + var load = 0d; + var universeCount = 0; + var selectedByResolution = new SortedDictionary(); + + void Accumulate(Universe target) + { + if (target.Selected == null || !TryGetSizeWarningThreshold(target, out var resolution, out var threshold)) + { + return; + } + var selectionSize = GetSelectionSize(target); + load += selectionSize / (double)threshold; + universeCount++; + selectedByResolution.TryGetValue(resolution, out var resolutionCount); + selectedByResolution[resolution] = resolutionCount + selectionSize; + } + + foreach (var kvp in _algorithm.UniverseManager) + { + Accumulate(kvp.Value); + } + if (universeCount == 0) + { + // not registered in the universe manager, count the given universe only + Accumulate(universe); + } + + if (load < 1) + { + return; + } + + _universeSelectionSizeWarningSent = true; + var counts = string.Join(" and ", selectedByResolution.Select(pair => $"~{pair.Value} symbols at {pair.Key} resolution")); + var suggestion = universe is OptionChainUniverse + ? "Narrow the filter (SetFilter/set_filter) or add specific contracts (AddOptionContract/add_option_contract)." + : "Select fewer symbols or use a coarser universe resolution (UniverseSettings.Resolution/universe_settings.resolution)."; + _algorithm.Debug($"Warning: universe selections have reached {counts} across {universeCount} universe(s)," + + $" latest: {GetSelectionSize(universe)} from {GetUniverseName(universe)}. Each selected symbol adds data" + + $" subscriptions, increasing time and memory usage. {suggestion}"); + } + catch (Exception exception) + { + // diagnostics must never interfere with the algorithm: log, disable and move on + _universeSelectionSizeWarningSent = true; + Log.Error(exception); + } + } + + /// + /// Gets the selection size warning threshold for the resolution the universe's members subscribe at. + /// False when the resolution is unavailable or warnings are disabled for it + /// + private bool TryGetSizeWarningThreshold(Universe universe, out Resolution resolution, out int threshold) + { + threshold = 0; + resolution = default; + var settings = universe.UniverseSettings; + if (settings == null) + { + return false; + } + resolution = settings.Resolution; + return _universeSelectionSizeWarningThresholds.TryGetValue(resolution, out threshold) && threshold > 0; + } + + /// + /// Number of symbols a universe selected. Option universe selections have the underlying + /// symbol prepended, which is not a contract + /// + private static int GetSelectionSize(Universe universe) + { + return universe is OptionChainUniverse + ? Math.Max(0, universe.Selected.Count - 1) + : universe.Selected.Count; + } + + /// + /// User-recognizable name for the universe emitting the warning + /// + private static string GetUniverseName(Universe universe) + { + return universe is OptionChainUniverse optionUniverse + ? optionUniverse.Option.Symbol.Underlying.Value + : universe.Configuration.Symbol.Value; + } + private void RemoveSecurityFromUniverse( List removedMembers, DateTime dateTimeUtc, diff --git a/Engine/DataFeeds/ZipDataCacheProvider.cs b/Engine/DataFeeds/ZipDataCacheProvider.cs index ca7ce83a5dba..e8554aabfcec 100644 --- a/Engine/DataFeeds/ZipDataCacheProvider.cs +++ b/Engine/DataFeeds/ZipDataCacheProvider.cs @@ -97,6 +97,7 @@ public Stream Fetch(string key) } catch (Exception exception) { + RethrowIfOutOfMemory(exception, filename, entryName); if (exception is ZipException || exception is ZlibException) { Log.Error("ZipDataCacheProvider.Fetch(): Corrupt zip file/entry: " + filename + "#" + entryName + " Error: " + exception); @@ -109,8 +110,13 @@ public Stream Fetch(string key) } catch (Exception err) { - Log.Error(err, "Inner try/catch"); stream?.DisposeSafely(); + if (err is OutOfMemoryException) + { + // let the memory diagnostic bubble up instead of resurfacing as a confusing downstream failure + throw; + } + Log.Error(err, "Inner try/catch"); return null; } } @@ -290,6 +296,7 @@ private Stream CacheAndCreateEntryStream(string filename, string entryName) { // don't leak the file stream! dataStream.DisposeSafely(); + RethrowIfOutOfMemory(exception, filename, entryName); if (exception is ZipException || exception is ZlibException) { Log.Error("ZipDataCacheProvider.Fetch(): Corrupt zip file/entry: " + filename + "#" + entryName + " Error: " + exception); @@ -412,6 +419,7 @@ private bool Cache(string filename, out CachedZipFile cachedZip) } catch (Exception exception) { + RethrowIfOutOfMemory(exception, filename, entryName: null); if (exception is ZipException || exception is ZlibException) { Log.Error("ZipDataCacheProvider.Fetch(): Corrupt zip file/entry: " + filename + " Error: " + exception); @@ -425,6 +433,25 @@ private bool Cache(string filename, out CachedZipFile cachedZip) return false; } + /// + /// Rethrows as a memory diagnostic if the exception is, or wraps, an . + /// Ionic wraps allocation failures into 'ZipException: Cannot read that as a ZipFile', which would otherwise + /// be logged as a corrupt zip file and swallowed + /// + private static void RethrowIfOutOfMemory(Exception exception, string filename, string entryName) + { + for (var inner = exception; inner != null; inner = inner.InnerException) + { + if (inner is OutOfMemoryException) + { + var entry = entryName != null ? "#" + entryName : string.Empty; + throw new OutOfMemoryException("ZipDataCacheProvider.Fetch(): ran out of memory reading " + filename + entry + + ". This is memory exhaustion, not a corrupt data file: reduce the number of subscriptions" + + " (e.g. narrow option universe filters) or the amount of data requested.", exception); + } + } + } + /// /// Type for storing zipfile in cache diff --git a/Engine/RealTime/BaseRealTimeHandler.cs b/Engine/RealTime/BaseRealTimeHandler.cs index d106b73bef4f..ef5a0b3f722d 100644 --- a/Engine/RealTime/BaseRealTimeHandler.cs +++ b/Engine/RealTime/BaseRealTimeHandler.cs @@ -119,7 +119,8 @@ public virtual void Setup(IAlgorithm algorithm, AlgorithmNodePacket job, IResult { Algorithm = algorithm; ResultHandler = resultHandler; - TimeMonitor = new TimeMonitor(GetTimeMonitorTimeout()); + // surface long-running scheduled event warnings in the user's logs, not just the engine log + TimeMonitor = new TimeMonitor(GetTimeMonitorTimeout()) { UserWarningHandler = message => ResultHandler?.DebugMessage(message) }; IsolatorLimitProvider = isolatorLimitProvider; if (job.Language == Language.CSharp) diff --git a/Tests/Algorithm/AlgorithmHistoryTests.cs b/Tests/Algorithm/AlgorithmHistoryTests.cs index e6c615377946..f4454d9b9158 100644 --- a/Tests/Algorithm/AlgorithmHistoryTests.cs +++ b/Tests/Algorithm/AlgorithmHistoryTests.cs @@ -38,6 +38,7 @@ using QuantConnect.Data.UniverseSelection; using QuantConnect.Tests.Common.Data.Fundamental; using QuantConnect.Logging; +using QuantConnect.Configuration; namespace QuantConnect.Tests.Algorithm { @@ -4287,6 +4288,100 @@ private static TestCaseData[] GetHistoryWithDataNormalizationModeTestCases() }).ToArray(); } + [Test] + public void WarnsOnLargeHistoryRequest() + { + Config.Set("history-request-cells-warning-threshold", "1000"); + try + { + var algorithm = CreateAlgorithmForHistoryWarningTests(); + var symbol = algorithm.AddEquity("SPY", Resolution.Daily).Symbol; + + // ~365 daily bars x 5 columns > 1000 cells + algorithm.History(new[] { symbol }, 365, Resolution.Daily); + + Assert.AreEqual(1, algorithm.DebugMessages.Count(x => x.Contains("large history request"))); + + // the warning is only sent once per algorithm + algorithm.History(new[] { symbol }, 365, Resolution.Daily); + Assert.AreEqual(1, algorithm.DebugMessages.Count(x => x.Contains("large history request"))); + } + finally + { + Config.Reset(); + } + } + + [Test] + public void WarnsOnLargeTickHistoryRequest() + { + Config.Set("history-request-cells-warning-threshold", "1000000"); + try + { + var algorithm = CreateAlgorithmForHistoryWarningTests(); + var symbol = algorithm.AddEquity("SPY", Resolution.Tick).Symbol; + + // one market day estimated at 10 ticks per second: 234,000 bars x 5 columns > 1,000,000 cells. + // At 1 tick per second the estimate would stay under the threshold and miss the warning + algorithm.History(new[] { symbol }, TimeSpan.FromDays(1), Resolution.Tick); + + Assert.AreEqual(1, algorithm.DebugMessages.Count(x => x.Contains("large history request"))); + } + finally + { + Config.Reset(); + } + } + + [Test] + public void DoesNotWarnOnSmallHistoryRequest() + { + // default threshold + var algorithm = CreateAlgorithmForHistoryWarningTests(); + var symbol = algorithm.AddEquity("SPY", Resolution.Daily).Symbol; + + algorithm.History(new[] { symbol }, 30, Resolution.Daily); + + Assert.AreEqual(0, algorithm.DebugMessages.Count(x => x.Contains("large history request"))); + } + + [Test] + public void WarnsOnRepeatedOverlappingHistoryRequests() + { + // the "large call" floor tracked for overlap detection is threshold / 50 = 2000 cells + Config.Set("history-request-cells-warning-threshold", "100000"); + try + { + var algorithm = CreateAlgorithmForHistoryWarningTests(); + var symbol = algorithm.AddEquity("SPY", Resolution.Daily).Symbol; + + for (var i = 0; i < 40; i++) + { + // ~800 daily bars x 5 columns = ~4000 cells: over the large-call floor, under the size warning threshold + algorithm.History(new[] { symbol }, 800, Resolution.Daily); + } + + Assert.AreEqual(0, algorithm.DebugMessages.Count(x => x.Contains("large history request"))); + Assert.AreEqual(1, algorithm.DebugMessages.Count(x => x.Contains("overlapping"))); + } + finally + { + Config.Reset(); + } + } + + private static QCAlgorithm CreateAlgorithmForHistoryWarningTests() + { + // the warning thresholds are read from the config when the algorithm is created, + // so we create a dedicated instance instead of using the fixture's algorithm + var algorithm = new QCAlgorithm(); + algorithm.SubscriptionManager.SetDataManager(new DataManagerStub(algorithm)); + algorithm.HistoryProvider = new TestHistoryProvider(); + algorithm.SetStartDate(2013, 10, 07); + algorithm.Settings.SeedInitialPrices = false; + return algorithm; + } + private QCAlgorithm GetAlgorithm(DateTime dateTime) { var algorithm = new QCAlgorithm(); diff --git a/Tests/Common/IsolatorLimitResultProviderTests.cs b/Tests/Common/IsolatorLimitResultProviderTests.cs index 8dea41832a28..a6edcef53028 100644 --- a/Tests/Common/IsolatorLimitResultProviderTests.cs +++ b/Tests/Common/IsolatorLimitResultProviderTests.cs @@ -20,6 +20,7 @@ using System.Threading.Tasks; using NUnit.Framework; using QuantConnect.Lean.Engine.DataFeeds; +using QuantConnect.Logging; using QuantConnect.Scheduling; using QuantConnect.Util; @@ -144,6 +145,49 @@ public void ConsumesMultipleMinutes() Assert.AreEqual(0, _timeMonitor.Count); } + [Test] + public void ConsumeNamesTheWorkWhenRequestingAdditionalTime() + { + var previousLogHandler = Log.LogHandler; + try + { + var logHandler = new QueueLogHandler(); + Log.LogHandler = logHandler; + var userWarnings = new List(); + _timeMonitor.UserWarningHandler = userWarnings.Add; + + var timeProvider = new ManualTimeProvider(new DateTime(2000, 01, 01)); + var provider = new FakeIsolatorLimitResultProvider(); + + Action code = () => + { + // lets give the monitor time to register the initial time + _timeMonitorEvent.WaitOne(); + timeProvider.Advance(TimeSpan.FromSeconds(65)); + // give the monitoring task time to request more time + _timeMonitorEvent.WaitOne(); + }; + + provider.Consume(timeProvider, code, _timeMonitor, "My Slow Scheduled Event"); + + Assert.AreEqual(1, provider.Invocations.Count); + Assert.AreEqual(1, logHandler.Logs.Count( + entry => entry.Message.Contains("'My Slow Scheduled Event' has been executing for over 1 minute"))); + // the warning also reaches the user + Assert.AreEqual(1, userWarnings.Count( + message => message.Contains("'My Slow Scheduled Event' has been executing for over 1 minute"))); + + // give time to the monitor to register the time consumer ended + _timeMonitorEvent.WaitOne(); + Assert.AreEqual(0, _timeMonitor.Count); + } + finally + { + Log.LogHandler = previousLogHandler; + _timeMonitor.UserWarningHandler = null; + } + } + private class FakeIsolatorLimitResultProvider : IIsolatorLimitResultProvider { private List _ivocations = new List(); diff --git a/Tests/Engine/AlgorithmTimeLimitManagerTests.cs b/Tests/Engine/AlgorithmTimeLimitManagerTests.cs index 580aa1b76646..144c9c944594 100644 --- a/Tests/Engine/AlgorithmTimeLimitManagerTests.cs +++ b/Tests/Engine/AlgorithmTimeLimitManagerTests.cs @@ -16,11 +16,14 @@ using System; using System.Collections.Generic; +using System.Linq; +using System.Threading; using System.Threading.Tasks; using NUnit.Framework; using QuantConnect.Algorithm.CSharp; using QuantConnect.Configuration; using QuantConnect.Lean.Engine; +using QuantConnect.Logging; using QuantConnect.Util.RateLimit; namespace QuantConnect.Tests.Engine @@ -57,6 +60,82 @@ public void StopsAlgorithm() parameter.ExpectedFinalStatus); } + [Test] + public void WarnsOnceOnLongTimeStepWithoutFailing() + { + var previousLogHandler = Log.LogHandler; + try + { + var logHandler = new QueueLogHandler(); + Log.LogHandler = logHandler; + + var userWarnings = new List(); + var timeManager = new AlgorithmTimeLimitManager(TokenBucket.Null, TimeSpan.FromMinutes(20), + timeLoopWarningThreshold: TimeSpan.FromMilliseconds(5)); + timeManager.UserWarningHandler = userWarnings.Add; + timeManager.StartNewTimeStep(); + // the first call initializes the current time step start time + Assert.IsTrue(timeManager.IsWithinLimit().IsWithinCustomLimits); + Thread.Sleep(50); + + // the time step is over the warning threshold: it warns but does not fail + var result = timeManager.IsWithinLimit(); + Assert.IsTrue(result.IsWithinCustomLimits, result.ErrorMessage); + Assert.AreEqual(1, WarningCount(logHandler)); + Assert.AreEqual(1, userWarnings.Count(x => x.Contains("time step has been executing"))); + + // only warns once per time step + Thread.Sleep(20); + Assert.IsTrue(timeManager.IsWithinLimit().IsWithinCustomLimits); + Assert.AreEqual(1, WarningCount(logHandler)); + Assert.AreEqual(1, userWarnings.Count); + + // a new slow time step warns again + timeManager.StartNewTimeStep(); + Assert.IsTrue(timeManager.IsWithinLimit().IsWithinCustomLimits); + Thread.Sleep(50); + Assert.IsTrue(timeManager.IsWithinLimit().IsWithinCustomLimits); + Assert.AreEqual(2, WarningCount(logHandler)); + Assert.AreEqual(2, userWarnings.Count); + } + finally + { + Log.LogHandler = previousLogHandler; + } + } + + [Test] + public void DoesNotWarnOnFastTimeStep() + { + var previousLogHandler = Log.LogHandler; + try + { + var logHandler = new QueueLogHandler(); + Log.LogHandler = logHandler; + + // default three minute warning threshold + var userWarnings = new List(); + var timeManager = new AlgorithmTimeLimitManager(TokenBucket.Null, TimeSpan.FromMinutes(20)); + timeManager.UserWarningHandler = userWarnings.Add; + timeManager.StartNewTimeStep(); + Assert.IsTrue(timeManager.IsWithinLimit().IsWithinCustomLimits); + Thread.Sleep(20); + Assert.IsTrue(timeManager.IsWithinLimit().IsWithinCustomLimits); + + Assert.AreEqual(0, WarningCount(logHandler)); + Assert.AreEqual(0, userWarnings.Count); + } + finally + { + Log.LogHandler = previousLogHandler; + } + } + + private static int WarningCount(QueueLogHandler logHandler) + { + return logHandler.Logs.Count(entry => entry.Message.Contains("time step has been executing")); + } + [Test] public void RaceCondition() { diff --git a/Tests/Engine/DataCacheProviders/ZipDataCacheProviderTests.cs b/Tests/Engine/DataCacheProviders/ZipDataCacheProviderTests.cs index 0bd1a79174c4..842543d1f1ed 100644 --- a/Tests/Engine/DataCacheProviders/ZipDataCacheProviderTests.cs +++ b/Tests/Engine/DataCacheProviders/ZipDataCacheProviderTests.cs @@ -51,6 +51,35 @@ public void MultiThreadReadWriteTest() dataCacheProvider.Dispose(); } + [Test] + public void FetchRethrowsOutOfMemoryAsMemoryDiagnostic() + { + // Ionic wraps the allocation failure into ZipException("Cannot read that as a ZipFile"), which used to be + // logged as a corrupt zip file and swallowed. We want an honest memory diagnostic instead. + using var dataCacheProvider = new ZipDataCacheProvider(new OutOfMemoryDataProvider()); + + var exception = Assert.Throws( + () => dataCacheProvider.Fetch("/data/option/usa/daily/pep_2026_trade_american.zip#entry.csv")); + + StringAssert.Contains("ran out of memory", exception.Message); + StringAssert.DoesNotContain("Corrupt zip", exception.Message); + Assert.IsNotNull(exception.InnerException); + } + + [Test] + public void FetchDoesNotThrowOnCorruptZipFile() + { + // a truly corrupt file must keep the previous behavior: log and return null instead of throwing + using var dataCacheProvider = new ZipDataCacheProvider(TestGlobals.DataProvider, cacheTimer: 0.1); + + var tempZipFile = Path.GetTempFileName().Replace(".tmp", ".zip", StringComparison.InvariantCulture); + File.WriteAllText(tempZipFile, "corrupted zip"); + + Stream result = null; + Assert.DoesNotThrow(() => result = dataCacheProvider.Fetch(tempZipFile + "#testEntry.csv")); + Assert.IsNull(result); + } + [Test] public void StoreFailsCorruptedFile() { @@ -72,5 +101,34 @@ private void ReadAndWrite(IDataCacheProvider dataCacheProvider, byte[] data) dataCacheProvider.Fetch(_tempZipFileEntry); dataCacheProvider.Store(_tempZipFileEntry, data); } + + /// + /// Provider whose streams fail allocation-style, simulating reading a zip while the process is out of memory + /// + private class OutOfMemoryDataProvider : IDataProvider + { + public event EventHandler NewDataRequest; + + public Stream Fetch(string key) + { + NewDataRequest?.Invoke(this, null); + return new OutOfMemoryStream(); + } + + private class OutOfMemoryStream : Stream + { + public override bool CanRead => true; + public override bool CanSeek => true; + public override bool CanWrite => false; + public override long Length => 1024; + public override long Position { get; set; } + + public override void Flush() { } + public override int Read(byte[] buffer, int offset, int count) => throw new OutOfMemoryException(); + public override long Seek(long offset, SeekOrigin origin) => Position; + public override void SetLength(long value) => throw new NotSupportedException(); + public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + } + } } } diff --git a/Tests/Engine/DataFeeds/UniverseSelectionTests.cs b/Tests/Engine/DataFeeds/UniverseSelectionTests.cs index 5f2e47d43625..ec9831316509 100644 --- a/Tests/Engine/DataFeeds/UniverseSelectionTests.cs +++ b/Tests/Engine/DataFeeds/UniverseSelectionTests.cs @@ -18,6 +18,7 @@ using System.Linq; using Moq; using NUnit.Framework; +using QuantConnect.Configuration; using QuantConnect.Data; using QuantConnect.Data.Fundamental; using QuantConnect.Data.UniverseSelection; @@ -33,6 +34,133 @@ namespace QuantConnect.Tests.Engine.DataFeeds [TestFixture] public class UniverseSelectionTests { + [Test] + public void WarnsOnLargeOptionUniverseSelection() + { + // option universe subscriptions are added at minute resolution + Config.Set("universe-selection-size-warning-thresholds", "{\"Minute\": 10}"); + try + { + var algorithm = new AlgorithmStub(new MockDataFeed()); + algorithm.SetStartDate(2014, 6, 6); + var option = algorithm.AddOption("AAPL"); + // OnEndOfTimeStep will add all pending universe additions + algorithm.OnEndOfTimeStep(); + var universe = algorithm.UniverseManager.Values.OfType().Single(); + + // below the threshold: no warning + universe.Selected = CreateOptionSelection(option.Symbol.Underlying, 5); + algorithm.DataManager.UniverseSelection.WarnOnLargeUniverseSelection(universe); + Assert.AreEqual(0, UniverseSizeWarningCount(algorithm)); + + // above the threshold: warns + universe.Selected = CreateOptionSelection(option.Symbol.Underlying, 15); + algorithm.DataManager.UniverseSelection.WarnOnLargeUniverseSelection(universe); + Assert.AreEqual(1, UniverseSizeWarningCount(algorithm)); + Assert.AreEqual(1, algorithm.DebugMessages.Count(x => x.Contains("SetFilter"))); + + // only warns once per algorithm + algorithm.DataManager.UniverseSelection.WarnOnLargeUniverseSelection(universe); + Assert.AreEqual(1, UniverseSizeWarningCount(algorithm)); + } + finally + { + Config.Reset(); + } + } + + [Test] + public void LargeUniverseSelectionWarningIsResolutionAware() + { + Config.Set("universe-selection-size-warning-thresholds", "{\"Minute\": 10, \"Daily\": 50}"); + try + { + var algorithm = new AlgorithmStub(new MockDataFeed()); + algorithm.SetEndDate(new DateTime(2024, 12, 13)); + algorithm.SetStartDate(algorithm.EndDate.Subtract(TimeSpan.FromDays(10))); + algorithm.UniverseSettings.Resolution = Resolution.Daily; + algorithm.AddUniverse(coarse => Enumerable.Empty()); + // OnEndOfTimeStep will add all pending universe additions + algorithm.OnEndOfTimeStep(); + var universe = algorithm.UniverseManager.Values.First(); + + // over the minute threshold but under this universe's daily threshold: no warning + universe.Selected = CreateEquitySelection(20); + algorithm.DataManager.UniverseSelection.WarnOnLargeUniverseSelection(universe); + Assert.AreEqual(0, UniverseSizeWarningCount(algorithm)); + + // over the daily threshold: warns with the generic suggestion + universe.Selected = CreateEquitySelection(60); + algorithm.DataManager.UniverseSelection.WarnOnLargeUniverseSelection(universe); + Assert.AreEqual(1, UniverseSizeWarningCount(algorithm)); + Assert.AreEqual(1, algorithm.DebugMessages.Count(x => x.Contains("Daily") && x.Contains("UniverseSettings.Resolution"))); + } + finally + { + Config.Reset(); + } + } + + [Test] + public void LargeUniverseSelectionWarningAggregatesAcrossResolutions() + { + Config.Set("universe-selection-size-warning-thresholds", "{\"Minute\": 10, \"Daily\": 50}"); + try + { + var algorithm = new AlgorithmStub(new MockDataFeed()); + algorithm.SetStartDate(2014, 6, 6); + var option = algorithm.AddOption("AAPL"); + algorithm.AddUniverse(fundamentals => Enumerable.Empty()); + // OnEndOfTimeStep will add all pending universe additions + algorithm.OnEndOfTimeStep(); + var optionUniverse = algorithm.UniverseManager.Values.OfType().Single(); + var equityUniverse = algorithm.UniverseManager.Values.Single(x => x is FundamentalUniverseFactory); + optionUniverse.UniverseSettings = new UniverseSettings(optionUniverse.UniverseSettings) { Resolution = Resolution.Minute }; + equityUniverse.UniverseSettings = new UniverseSettings(equityUniverse.UniverseSettings) { Resolution = Resolution.Daily }; + + // 8/10 minute contracts: under budget + optionUniverse.Selected = CreateOptionSelection(option.Symbol.Underlying, 8); + algorithm.DataManager.UniverseSelection.WarnOnLargeUniverseSelection(optionUniverse); + Assert.AreEqual(0, UniverseSizeWarningCount(algorithm)); + + // 8/10 minute contracts + 30/50 daily symbols = 1.4 of the shared budget: warns, + // even though each universe is under its own resolution threshold + equityUniverse.Selected = CreateEquitySelection(30); + algorithm.DataManager.UniverseSelection.WarnOnLargeUniverseSelection(equityUniverse); + Assert.AreEqual(1, UniverseSizeWarningCount(algorithm)); + Assert.AreEqual(1, algorithm.DebugMessages.Count(x => x.Contains("~8 symbols at Minute resolution") && + x.Contains("~30 symbols at Daily resolution"))); + } + finally + { + Config.Reset(); + } + } + + private static int UniverseSizeWarningCount(AlgorithmStub algorithm) + { + return algorithm.DebugMessages.Count(x => x.Contains("universe selections")); + } + + private static HashSet CreateEquitySelection(int count) + { + return Enumerable.Range(0, count) + .Select(i => Symbol.Create($"SYM{i}", SecurityType.Equity, Market.USA)) + .ToHashSet(); + } + + private static HashSet CreateOptionSelection(Symbol underlying, int contractCount) + { + // option universe selections have the underlying symbol prepended + var selected = new HashSet { underlying }; + var expiry = new DateTime(2014, 7, 19); + for (var i = 0; i < contractCount; i++) + { + selected.Add(Symbol.CreateOption(underlying, Market.USA, OptionStyle.American, OptionRight.Call, 100 + i, expiry)); + } + return selected; + } + [Test] public void CreatedEquityIsNotAddedToSymbolCache() {