Consolidator rolling window fixes

Common/Data/Consolidators/BaseTimelessConsolidator.cs

@@ -60,7 +60,7 @@ public abstract class BaseTimelessConsolidator<T> : ConsolidatorBase
         /// <summary>
         /// Typed event handler that fires when a new piece of data is produced
         /// </summary>
-        public event EventHandler<T> DataConsolidated;
+        public new event EventHandler<T> DataConsolidated;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="BaseTimelessConsolidator{T}" /> class.
@@ -158,14 +158,12 @@ public override void Update(IBaseData data)
         protected abstract void CreateNewBar(IBaseData data, decimal currentValue, decimal volume);
 
         /// <summary>
-        /// Event invocator for the DataConsolidated event. This should be invoked
-        /// by derived classes when they have consolidated a new piece of data.
+        /// Raises the strongly typed DataConsolidated event
         /// </summary>
         /// <param name="consolidated">The newly consolidated data</param>
-        protected void OnDataConsolidated(T consolidated)
+        protected override void FireDataConsolidated(IBaseData consolidated)
         {
-            DataConsolidated?.Invoke(this, consolidated);
-            base.OnDataConsolidated(consolidated);
+            DataConsolidated?.Invoke(this, (T)consolidated);
         }
 
         /// <summary>Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.</summary>

Common/Data/Consolidators/ConsolidatorBase.cs

@@ -24,14 +24,11 @@ namespace QuantConnect.Data.Consolidators
     /// </summary>
     public abstract class ConsolidatorBase : WindowBase<IBaseData>, IDataConsolidator
     {
-        private DataConsolidatedHandler _dataConsolidated;
-
         private IBaseData _consolidated;
 
         /// <summary>
         /// Gets the most recently consolidated piece of data. This will be null if this consolidator
-        /// has not produced any data yet. Setting this property adds the value to the rolling window,
-        /// setting it to null clears the window.
+        /// has not produced any data yet.
         /// </summary>
         public IBaseData Consolidated
         {
@@ -42,14 +39,6 @@ public IBaseData Consolidated
             protected set
             {
                 _consolidated = value;
-                if (value == null)
-                {
-                    ResetWindow();
-                }
-                else
-                {
-                    Current = value;
-                }
             }
         }
 
@@ -81,33 +70,54 @@ protected set
         public abstract void Scan(DateTime currentLocalTime);
 
         /// <summary>
-        /// Event handler that fires when a new piece of data is produced
+        /// Event handler that fires when a new piece of data is produced. This is the single subscription
+        /// point, shared by the <see cref="IDataConsolidator"/> interface and by derived consolidators whose
+        /// output is a base data bar, so subscribing and unsubscribing always target the same handler list.
         /// </summary>
-        event DataConsolidatedHandler IDataConsolidator.DataConsolidated
-        {
-            add { _dataConsolidated += value; }
-            remove { _dataConsolidated -= value; }
-        }
+        public event DataConsolidatedHandler DataConsolidated;
 
         /// <summary>
-        /// Event invocator for the DataConsolidated event. Fires the event and updates the rolling window.
+        /// Event invocator for the DataConsolidated event. Populates the rolling window, raises the
+        /// strongly typed and interface events, and finally updates the <see cref="Consolidated"/> property.
         /// </summary>
         protected virtual void OnDataConsolidated(IBaseData consolidated)
         {
-            _dataConsolidated?.Invoke(this, consolidated);
+            // populate the rolling window before firing any event so that, inside a DataConsolidated
+            // handler, consolidator[0] is the bar that was just produced. Skip null bars, an out of order
+            // data point can produce a null bar in count mode, so we never push null nor wipe the history
+            if (consolidated != null)
+            {
+                Current = consolidated;
+            }
+
+            // let derived consolidators raise their strongly typed DataConsolidated event
+            FireDataConsolidated(consolidated);
+
+            DataConsolidated?.Invoke(this, consolidated);
 
             // assign the Consolidated property after the event handlers are fired,
             // this allows the event handlers to look at the new consolidated data
             // and the previous consolidated data at the same time without extra bookkeeping
             Consolidated = consolidated;
         }
 
+        /// <summary>
+        /// Raises the strongly typed DataConsolidated event exposed by derived consolidators that produce a
+        /// more specific bar type. Invoked after the rolling window is populated and before the shared event
+        /// so every handler sees the same window. Consolidators whose output is a base data bar do not need
+        /// to override this, the shared <see cref="DataConsolidated"/> event already carries their bar.
+        /// </summary>
+        /// <param name="consolidated">The newly consolidated data</param>
+        protected virtual void FireDataConsolidated(IBaseData consolidated)
+        {
+        }
+
         /// <summary>
         /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
         /// </summary>
         public virtual void Dispose()
         {
-            _dataConsolidated = null;
+            DataConsolidated = null;
         }
 
         /// <summary>
@@ -116,6 +126,7 @@ public virtual void Dispose()
         public virtual void Reset()
         {
             Consolidated = null;
+            ResetWindow();
         }
     }
 }

Common/Data/Consolidators/DataConsolidator.cs

@@ -47,11 +47,6 @@ public override void Update(IBaseData data)
         /// <param name="currentLocalTime">The current time in the local time zone (same as <see cref="BaseData.Time"/>)</param>
         public abstract override void Scan(DateTime currentLocalTime);
 
-        /// <summary>
-        /// Event handler that fires when a new piece of data is produced
-        /// </summary>
-        public event DataConsolidatedHandler DataConsolidated;
-
         /// <summary>
         /// Gets a clone of the data being currently consolidated
         /// </summary>
@@ -74,21 +69,5 @@ public override void Update(IBaseData data)
         /// <param name="data">The new data for the consolidator</param>
         public abstract void Update(TInput data);
 
-        /// <summary>Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.</summary>
-        /// <filterpriority>2</filterpriority>
-        public override void Dispose()
-        {
-            DataConsolidated = null;
-            base.Dispose();
-        }
-
-        /// <summary>
-        /// Event invocator for the DataConsolidated event
-        /// </summary>
-        protected override void OnDataConsolidated(IBaseData consolidated)
-        {
-            DataConsolidated?.Invoke(this, consolidated);
-            base.OnDataConsolidated(consolidated);
-        }
     }
 }

Common/Data/Consolidators/MarketHourAwareConsolidator.cs

@@ -166,11 +166,6 @@ protected virtual IDataConsolidator CreateConsolidator(Func<DateTime, CalendarIn
             throw new ArgumentNullException(nameof(dataType), $"{dataType.Name} not supported");
         }
 
-        /// <summary>
-        /// Event handler that fires when a new piece of data is produced
-        /// </summary>
-        public event DataConsolidatedHandler DataConsolidated;
-
         /// <summary>
         /// Updates this consolidator with the specified data
         /// </summary>
@@ -277,8 +272,7 @@ protected virtual bool UseStrictEndTime(Symbol symbol)
         /// </summary>
         protected virtual void ForwardConsolidatedBar(object sender, IBaseData consolidated)
         {
-            DataConsolidated?.Invoke(this, consolidated);
-            base.OnDataConsolidated(consolidated);
+            OnDataConsolidated(consolidated);
         }
     }
 }

Common/Data/Consolidators/RenkoConsolidator.cs

@@ -95,7 +95,7 @@ public class RenkoConsolidator : ConsolidatorBase
         /// <summary>
         /// Typed event handler that fires when a new piece of data is produced
         /// </summary>
-        public event EventHandler<RenkoBar> DataConsolidated;
+        public new event EventHandler<RenkoBar> DataConsolidated;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="RenkoConsolidator"/> class using the specified <paramref name="barSize"/>.
@@ -238,15 +238,16 @@ public override void Reset()
         }
 
         /// <summary>
-        /// Event invocator for the DataConsolidated event. This should be invoked
-        /// by derived classes when they have consolidated a new piece of data.
+        /// Raises the strongly typed DataConsolidated event
         /// </summary>
         /// <param name="consolidated">The newly consolidated data</param>
-        protected void OnDataConsolidated(RenkoBar consolidated)
+        protected override void FireDataConsolidated(IBaseData consolidated)
         {
-            DataConsolidated?.Invoke(this, consolidated);
-            _currentBar = consolidated;
-            base.OnDataConsolidated(consolidated);
+            var bar = (RenkoBar)consolidated;
+            // fire the typed event before updating the current bar so handlers reading
+            // WorkingData still see the previous bar, as they did before the rolling window
+            DataConsolidated?.Invoke(this, bar);
+            _currentBar = bar;
         }
 
         private void Rising(IBaseData data)

Common/Data/Consolidators/SequentialConsolidator.cs

@@ -83,11 +83,6 @@ public override void Scan(DateTime currentLocalTime)
             First.Scan(currentLocalTime);
         }
 
-        /// <summary>
-        /// Event handler that fires when a new piece of data is produced
-        /// </summary>
-        public event DataConsolidatedHandler DataConsolidated;
-
         /// <summary>
         /// Creates a new consolidator that will pump date through the first, and then the output
         /// of the first into the second. This enables 'wrapping' or 'composing' of consolidators
@@ -111,24 +106,12 @@ public SequentialConsolidator(IDataConsolidator first, IDataConsolidator second)
             second.DataConsolidated += (sender, consolidated) => OnDataConsolidated(consolidated);
         }
 
-        /// <summary>
-        /// Event invocator for the DataConsolidated event. This should be invoked
-        /// by derived classes when they have consolidated a new piece of data.
-        /// </summary>
-        /// <param name="consolidated">The newly consolidated data</param>
-        protected override void OnDataConsolidated(IBaseData consolidated)
-        {
-            DataConsolidated?.Invoke(this, consolidated);
-            base.OnDataConsolidated(consolidated);
-        }
-
         /// <summary>Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.</summary>
         /// <filterpriority>2</filterpriority>
         public override void Dispose()
         {
             First.Dispose();
             Second.Dispose();
-            DataConsolidated = null;
             base.Dispose();
         }
 

Common/Data/SubscriptionManager.cs

@@ -258,12 +258,41 @@ public void RemoveConsolidator(Symbol symbol, PyObject pyConsolidator)
         {
             if (!pyConsolidator.TryConvert(out IDataConsolidator consolidator))
             {
-                consolidator = new DataConsolidatorPythonWrapper(pyConsolidator);
+                // reuse the wrapper created when this python consolidator was added instead of building a
+                // throwaway one: a new wrapper would subscribe to the live python object's event just to be
+                // disposed again, and would leave the original wrapper's subscription leaked
+                consolidator = FindPythonConsolidator(symbol, pyConsolidator)
+                    ?? new DataConsolidatorPythonWrapper(pyConsolidator);
             }
 
             RemoveConsolidator(symbol, consolidator);
         }
 
+        /// <summary>
+        /// Finds the <see cref="DataConsolidatorPythonWrapper"/> previously created for the given python
+        /// consolidator so it can be removed and disposed, rather than a throwaway wrapper that would churn
+        /// the live python object's event subscription and leak the original one.
+        /// </summary>
+        private IDataConsolidator FindPythonConsolidator(Symbol symbol, PyObject pyConsolidator)
+        {
+            var configs = symbol != null
+                ? _subscriptionManager.GetSubscriptionDataConfigs(symbol)
+                : Subscriptions;
+
+            foreach (var subscription in configs)
+            {
+                foreach (var existing in subscription.Consolidators)
+                {
+                    if (existing is DataConsolidatorPythonWrapper && existing.Equals(pyConsolidator))
+                    {
+                        return existing;
+                    }
+                }
+            }
+
+            return null;
+        }
+
         /// <summary>
         /// Will trigger past consolidator scans
         /// </summary>

Common/Python/DataConsolidatorPythonWrapper.cs

@@ -27,6 +27,7 @@ public class DataConsolidatorPythonWrapper : ConsolidatorBase
     {
         private readonly BasePythonWrapper<IDataConsolidator> _pythonWrapper;
         private readonly DataConsolidatedHandler _pythonDataConsolidated;
+        private bool _disposed;
 
         /// <summary>
         /// Gets a clone of the data being currently consolidated
@@ -60,8 +61,14 @@ public DataConsolidatorPythonWrapper(PyObject consolidator)
         {
             _pythonWrapper = new BasePythonWrapper<IDataConsolidator>(consolidator, true);
             _pythonDataConsolidated = (_, bar) => OnDataConsolidated(bar);
-            var pythonEvent = _pythonWrapper.GetEvent("DataConsolidated");
-            pythonEvent += _pythonDataConsolidated;
+
+            // GetEvent releases the GIL before returning, so the event subscription must
+            // hold it explicitly, otherwise pythonnet mutates the Python object without the GIL
+            using (Py.GIL())
+            {
+                dynamic pythonEvent = _pythonWrapper.GetEvent("DataConsolidated");
+                pythonEvent += _pythonDataConsolidated;
+            }
         }
 
         /// <summary>
@@ -96,8 +103,19 @@ public override void Reset()
         /// </summary>
         public override void Dispose()
         {
-            var pythonEvent = _pythonWrapper.GetEvent("DataConsolidated");
-            pythonEvent -= _pythonDataConsolidated;
+            if (_disposed)
+            {
+                return;
+            }
+            _disposed = true;
+
+            // Dispose can run from engine threads that do not hold the GIL, so acquire it before
+            // unsubscribing, otherwise the pythonnet event mutation can crash the runtime
+            using (Py.GIL())
+            {
+                dynamic pythonEvent = _pythonWrapper.GetEvent("DataConsolidated");
+                pythonEvent -= _pythonDataConsolidated;
+            }
             _pythonWrapper.Dispose();
             base.Dispose();
         }

Common/Python/PythonConsolidator.cs

@@ -16,14 +16,17 @@
 using Python.Runtime;
 using QuantConnect.Data;
 using QuantConnect.Data.Consolidators;
+using QuantConnect.Indicators;
 using System;
 
 namespace QuantConnect.Python
 {
     /// <summary>
     /// Provides a base class for python consolidators, necessary to use event handler.
+    /// Inherits the built-in rolling window so custom Python consolidators can access their
+    /// consolidated history through the indexer, Current, Previous and Window members.
     /// </summary>
-    public class PythonConsolidator : IDataConsolidator
+    public class PythonConsolidator : WindowBase<IBaseData>, IDataConsolidator
     {
         /// <summary>
         /// Gets the most recently consolidated piece of data. This will be null if this consolidator
@@ -70,6 +73,11 @@ public Type OutputType
         /// <param name="data">The finished data from the consolidator</param>
         public void OnDataConsolidated(PyObject consolidator, IBaseData data)
         {
+            // populate the rolling window before firing so a handler sees the new bar at index 0
+            if (data != null)
+            {
+                Current = data;
+            }
             DataConsolidated?.Invoke(consolidator, data);
         }
 
@@ -80,6 +88,7 @@ public virtual void Reset()
         {
             Consolidated = null;
             WorkingData = null;
+            ResetWindow();
         }
 
         /// <summary>