Ensure AdvanceTime() runs only after full algorithm initialization

Algorithm.CSharp/ConsolidatorAndAlgorithmTimeConsistencyWithWarmupRegressionAlgorithm.cs

@@ -0,0 +1,116 @@
+/*
+ * 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.Collections.Generic;
+using QuantConnect.Data.Market;
+using QuantConnect.Interfaces;
+
+namespace QuantConnect.Algorithm.CSharp
+{
+    /// <summary>
+    /// This algorithm tests the time consistency of the consolidator and the algorithm
+    /// It is expected to fail if the consolidator and the algorithm time are not in sync
+    /// </summary>
+    public class ConsolidatorAndAlgorithmTimeConsistencyWithWarmupRegressionAlgorithm : QCAlgorithm, IRegressionAlgorithmDefinition
+    {
+        /// <summary>
+        /// Initialise the data and resolution required, as well as the cash and start-end dates for your algorithm. All algorithms must initialized.
+        /// </summary>
+        public override void Initialize()
+        {
+            SetStartDate(2013, 04, 07);
+            SetEndDate(2013, 10, 31);
+
+            AddEquity("SPY", Resolution.Hour);
+            SetWarmup(TimeSpan.FromDays(10), Resolution.Hour);
+            Consolidate("SPY", TimeSpan.FromDays(1), OnConsolidated);
+        }
+
+        public void OnConsolidated(TradeBar bar)
+        {
+            if (Time != bar.EndTime)
+            {
+                throw new RegressionTestException($"Mismatches between consolidation time and algorithm time: {Time} != {bar.EndTime}");
+            }
+
+            // check if the consolidation time is midnight
+            if (Time.TimeOfDay != TimeSpan.Zero)
+            {
+                throw new RegressionTestException($"Unexpected consolidation time {bar.EndTime}");
+            }
+        }
+
+        /// <summary>
+        /// This is used by the regression test system to indicate if the open source Lean repository has the required data to run this algorithm.
+        /// </summary>
+        public bool CanRunLocally { get; } = true;
+
+        /// <summary>
+        /// This is used by the regression test system to indicate which languages this algorithm is written in.
+        /// </summary>
+        public List<Language> Languages { get; } = new() { Language.CSharp };
+
+        /// <summary>
+        /// Data Points count of all timeslices of algorithm
+        /// </summary>
+        public long DataPoints => 2132;
+
+        /// <summary>
+        /// Data Points count of the algorithm history
+        /// </summary>
+        public int AlgorithmHistoryDataPoints => 0;
+
+        /// <summary>
+        /// Final status of the algorithm
+        /// </summary>
+        public AlgorithmStatus AlgorithmStatus => AlgorithmStatus.Completed;
+
+        /// <summary>
+        /// This is used by the regression test system to indicate what the expected statistics are from running the algorithm
+        /// </summary>
+        public Dictionary<string, string> ExpectedStatistics => new Dictionary<string, string>
+        {
+            {"Total Orders", "0"},
+            {"Average Win", "0%"},
+            {"Average Loss", "0%"},
+            {"Compounding Annual Return", "0%"},
+            {"Drawdown", "0%"},
+            {"Expectancy", "0"},
+            {"Start Equity", "100000"},
+            {"End Equity", "100000"},
+            {"Net Profit", "0%"},
+            {"Sharpe Ratio", "0"},
+            {"Sortino Ratio", "0"},
+            {"Probabilistic Sharpe Ratio", "0%"},
+            {"Loss Rate", "0%"},
+            {"Win Rate", "0%"},
+            {"Profit-Loss Ratio", "0"},
+            {"Alpha", "0"},
+            {"Beta", "0"},
+            {"Annual Standard Deviation", "0"},
+            {"Annual Variance", "0"},
+            {"Information Ratio", "-1.81"},
+            {"Tracking Error", "0.102"},
+            {"Treynor Ratio", "0"},
+            {"Total Fees", "$0.00"},
+            {"Estimated Strategy Capacity", "$0"},
+            {"Lowest Capacity Asset", ""},
+            {"Portfolio Turnover", "0%"},
+            {"Drawdown Recovery", "0"},
+            {"OrderListHash", "d41d8cd98f00b204e9800998ecf8427e"}
+        };
+    }
+}

Common/Data/ConsolidatorWrapper.cs

@@ -49,7 +49,7 @@ internal class ConsolidatorWrapper : IDisposable
         /// <summary>
         /// Get enqueue time
         /// </summary>
-        public ConsolidatorScanPriority Priority => new (UtcScanTime, _id);
+        public ConsolidatorScanPriority Priority => new(UtcScanTime, _id);
 
         /// <summary>
         /// Creates a new instance
@@ -65,7 +65,6 @@ public ConsolidatorWrapper(IDataConsolidator consolidator, TimeSpan configIncrem
             _minimumIncrement = configIncrement < Time.OneSecond ? Time.OneSecond : configIncrement;
 
             _consolidator.DataConsolidated += AdvanceScanTime;
-            AdvanceScanTime();
         }
 
         /// <summary>
@@ -89,7 +88,7 @@ public void Dispose()
         /// <summary>
         /// Helper method to set the next scan time
         /// </summary>
-        private void AdvanceScanTime(object _ = null, IBaseData consolidated = null)
+        public void AdvanceScanTime(object _ = null, IBaseData consolidated = null)
         {
             if (consolidated == null && UtcScanTime > _timeKeeper.UtcTime)
             {

Common/Data/SubscriptionManager.cs

@@ -33,7 +33,7 @@ public class SubscriptionManager
     {
         private readonly PriorityQueue<ConsolidatorWrapper, ConsolidatorScanPriority> _consolidatorsSortedByScanTime;
         private readonly Dictionary<IDataConsolidator, ConsolidatorWrapper> _consolidators;
-        private List<Tuple<ConsolidatorWrapper, ConsolidatorScanPriority>> _consolidatorsToAdd;
+        private List<ConsolidatorWrapper> _consolidatorsToAdd;
         private readonly object _threadSafeCollectionLock;
         private readonly ITimeKeeper _timeKeeper;
         private IAlgorithmSubscriptionManager _subscriptionManager;
@@ -192,7 +192,7 @@ public void AddConsolidator(Symbol symbol, IDataConsolidator consolidator, TickT
                     lock (_threadSafeCollectionLock)
                     {
                         _consolidatorsToAdd ??= new();
-                        _consolidatorsToAdd.Add(new(wrapper, wrapper.Priority));
+                        _consolidatorsToAdd.Add(wrapper);
                     }
                     return;
                 }
@@ -275,7 +275,13 @@ public void ScanPastConsolidators(DateTime newUtcTime, IAlgorithm algorithm)
             {
                 lock (_threadSafeCollectionLock)
                 {
-                    _consolidatorsToAdd.DoForEach(x => _consolidatorsSortedByScanTime.Enqueue(x.Item1, x.Item2));
+                    foreach (var consolidator in _consolidatorsToAdd)
+                    {
+                        // At this point we already calculate the warm up start time, so we can reset the UtcScanTime property
+                        // To ensure correct scan times
+                        consolidator.AdvanceScanTime();
+                        _consolidatorsSortedByScanTime.Enqueue(consolidator, consolidator.Priority);
+                    }
                     _consolidatorsToAdd = null;
                 }
             }

Tests/Common/Data/ConsolidatorWrapperTests.cs

@@ -35,6 +35,7 @@ public void InitialScanTime(int seconds)
             var increment = TimeSpan.FromSeconds(seconds);
             using var consolidator = new TestConsolidator();
             using var wrapper = new ConsolidatorWrapper(consolidator, increment, timeKeeper, localtime);
+            wrapper.AdvanceScanTime();
 
             Assert.AreEqual(time.Add(increment < Time.OneSecond ? Time.OneSecond : increment), wrapper.UtcScanTime);
         }
@@ -49,6 +50,7 @@ public void ScanTimeAfterScanUtcTimeInPast(int seconds)
             var increment = TimeSpan.FromSeconds(seconds);
             using var consolidator = new TestConsolidator();
             using var wrapper = new ConsolidatorWrapper(consolidator, increment, timeKeeper, localtime);
+            wrapper.AdvanceScanTime();
 
             var expected = time.Add(increment < Time.OneSecond ? Time.OneSecond : increment);
             Assert.AreEqual(expected, wrapper.UtcScanTime);
@@ -68,6 +70,7 @@ public void ScanTimeAfterScanUtcTimeInFuture(int seconds)
             var increment = TimeSpan.FromSeconds(seconds);
             using var consolidator = new TestConsolidator();
             using var wrapper = new ConsolidatorWrapper(consolidator, increment, timeKeeper, localtime);
+            wrapper.AdvanceScanTime();
 
             var expected = time.Add(increment < Time.OneSecond ? Time.OneSecond : increment);
             Assert.AreEqual(expected, wrapper.UtcScanTime);
@@ -109,6 +112,7 @@ public void ScanTimeAfterConsolidationDayLightSavings(int hoursShift, bool savin
             var increment = Time.OneHour;
             using var consolidator = new TestConsolidator();
             using var wrapper = new ConsolidatorWrapper(consolidator, increment, timeKeeper, localtime);
+            wrapper.AdvanceScanTime();
 
             var expected = time.Add(Time.OneHour);
             Assert.AreEqual(expected, wrapper.UtcScanTime);
@@ -149,6 +153,7 @@ public void ScanTimeOnWorkingBarDayLightSavings(int hoursShift, bool savingsStar
             var increment = Time.OneHour;
             using var consolidator = new TestConsolidator();
             using var wrapper = new ConsolidatorWrapper(consolidator, increment, timeKeeper, localtime);
+            wrapper.AdvanceScanTime();
 
             var expected = time.Add(Time.OneHour);
             Assert.AreEqual(expected, wrapper.UtcScanTime);
@@ -168,7 +173,7 @@ public void ConsolidatorScanPriorityComparerComparesByUtcScanDate()
         {
             const int id = 1;
             var utcScanTime = new DateTime(2024, 12, 10, 0, 0, 0, DateTimeKind.Utc);
-            
+
             var priority1 = new ConsolidatorScanPriority(utcScanTime, id);
             var priority2 = new ConsolidatorScanPriority(utcScanTime.AddSeconds(1), id + 1);
             var priority3 = new ConsolidatorScanPriority(utcScanTime, id + 1);