Update IdentityDataConsolidator to track last processed time (#8510)

Common/Data/Consolidators/IdentityDataConsolidator.cs

@@ -26,10 +26,15 @@ namespace QuantConnect.Data.Consolidators
     public class IdentityDataConsolidator<T> : DataConsolidator<T>
         where T : IBaseData
     {
-        private static readonly bool IsTick = typeof (T) == typeof (Tick);
+        private static readonly bool IsTick = typeof(T) == typeof(Tick);
 
         private T _last;
 
+        /// <summary>
+        /// Stores the timestamp of the last processed data item.
+        /// </summary>
+        private DateTime _lastTime;
+
         /// <summary>
         /// Gets a clone of the data being currently consolidated
         /// </summary>
@@ -43,7 +48,7 @@ public override IBaseData WorkingData
         /// </summary>
         public override Type OutputType
         {
-            get { return typeof (T); }
+            get { return typeof(T); }
         }
 
         /// <summary>
@@ -52,10 +57,11 @@ public override Type OutputType
         /// <param name="data">The new data for the consolidator</param>
         public override void Update(T data)
         {
-            if (IsTick || _last == null || _last.EndTime != data.EndTime)
+            if (IsTick || _last == null || data.EndTime != _lastTime)
             {
                 OnDataConsolidated(data);
                 _last = data;
+                _lastTime = data.EndTime;
             }
         }