From e25539f18283fb84507208f13bf78ca6e5f45bdb Mon Sep 17 00:00:00 2001 From: Josue Nina Date: Tue, 7 Jul 2026 18:14:27 -0500 Subject: [PATCH 1/3] Fix consolidator rolling window event ordering and out of order handling --- .../Consolidators/BaseTimelessConsolidator.cs | 8 ++- Common/Data/Consolidators/ConsolidatorBase.cs | 35 +++++++++---- Common/Data/Consolidators/DataConsolidator.cs | 5 +- .../MarketHourAwareConsolidator.cs | 10 +++- .../Data/Consolidators/RenkoConsolidator.cs | 11 ++-- .../Consolidators/SequentialConsolidator.cs | 6 +-- .../Python/DataConsolidatorPythonWrapper.cs | 26 ++++++++-- Tests/Common/Data/ConsolidatorBaseTests.cs | 50 +++++++++++++++++++ 8 files changed, 117 insertions(+), 34 deletions(-) diff --git a/Common/Data/Consolidators/BaseTimelessConsolidator.cs b/Common/Data/Consolidators/BaseTimelessConsolidator.cs index a68aeddd03a0..60aba0f26596 100644 --- a/Common/Data/Consolidators/BaseTimelessConsolidator.cs +++ b/Common/Data/Consolidators/BaseTimelessConsolidator.cs @@ -158,14 +158,12 @@ public override void Update(IBaseData data) protected abstract void CreateNewBar(IBaseData data, decimal currentValue, decimal volume); /// - /// 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 /// /// The newly consolidated data - protected void OnDataConsolidated(T consolidated) + protected override void FireDataConsolidated(IBaseData consolidated) { - DataConsolidated?.Invoke(this, consolidated); - base.OnDataConsolidated(consolidated); + DataConsolidated?.Invoke(this, (T)consolidated); } /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources. diff --git a/Common/Data/Consolidators/ConsolidatorBase.cs b/Common/Data/Consolidators/ConsolidatorBase.cs index df4f9eb8430e..05b4f8516746 100644 --- a/Common/Data/Consolidators/ConsolidatorBase.cs +++ b/Common/Data/Consolidators/ConsolidatorBase.cs @@ -30,8 +30,7 @@ public abstract class ConsolidatorBase : WindowBase, IDataConsolidato /// /// 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. /// public IBaseData Consolidated { @@ -42,14 +41,6 @@ public IBaseData Consolidated protected set { _consolidated = value; - if (value == null) - { - ResetWindow(); - } - else - { - Current = value; - } } } @@ -90,10 +81,22 @@ event DataConsolidatedHandler IDataConsolidator.DataConsolidated } /// - /// 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 property. /// protected virtual void OnDataConsolidated(IBaseData 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, @@ -102,6 +105,15 @@ protected virtual void OnDataConsolidated(IBaseData consolidated) Consolidated = consolidated; } + /// + /// Raises the strongly typed DataConsolidated event exposed by derived consolidators. Invoked after + /// the rolling window is populated and before the interface event so every handler sees the same window. + /// + /// The newly consolidated data + protected virtual void FireDataConsolidated(IBaseData consolidated) + { + } + /// /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources. /// @@ -116,6 +128,7 @@ public virtual void Dispose() public virtual void Reset() { Consolidated = null; + ResetWindow(); } } } diff --git a/Common/Data/Consolidators/DataConsolidator.cs b/Common/Data/Consolidators/DataConsolidator.cs index 2bfaec1017a9..33bdd23a8cea 100644 --- a/Common/Data/Consolidators/DataConsolidator.cs +++ b/Common/Data/Consolidators/DataConsolidator.cs @@ -83,12 +83,11 @@ public override void Dispose() } /// - /// Event invocator for the DataConsolidated event + /// Raises the strongly typed DataConsolidated event /// - protected override void OnDataConsolidated(IBaseData consolidated) + protected override void FireDataConsolidated(IBaseData consolidated) { DataConsolidated?.Invoke(this, consolidated); - base.OnDataConsolidated(consolidated); } } } \ No newline at end of file diff --git a/Common/Data/Consolidators/MarketHourAwareConsolidator.cs b/Common/Data/Consolidators/MarketHourAwareConsolidator.cs index e844439b278e..f6ec272df5c4 100644 --- a/Common/Data/Consolidators/MarketHourAwareConsolidator.cs +++ b/Common/Data/Consolidators/MarketHourAwareConsolidator.cs @@ -276,9 +276,17 @@ protected virtual bool UseStrictEndTime(Symbol symbol) /// This wrapper keeps its own rolling window in addition to the inner consolidator's window. /// protected virtual void ForwardConsolidatedBar(object sender, IBaseData consolidated) + { + OnDataConsolidated(consolidated); + } + + /// + /// Raises the strongly typed DataConsolidated event + /// + /// The newly consolidated data + protected override void FireDataConsolidated(IBaseData consolidated) { DataConsolidated?.Invoke(this, consolidated); - base.OnDataConsolidated(consolidated); } } } diff --git a/Common/Data/Consolidators/RenkoConsolidator.cs b/Common/Data/Consolidators/RenkoConsolidator.cs index b59042290e6a..1f280db2b8c5 100644 --- a/Common/Data/Consolidators/RenkoConsolidator.cs +++ b/Common/Data/Consolidators/RenkoConsolidator.cs @@ -238,15 +238,14 @@ public override void Reset() } /// - /// 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 /// /// The newly consolidated data - protected void OnDataConsolidated(RenkoBar consolidated) + protected override void FireDataConsolidated(IBaseData consolidated) { - DataConsolidated?.Invoke(this, consolidated); - _currentBar = consolidated; - base.OnDataConsolidated(consolidated); + var bar = (RenkoBar)consolidated; + _currentBar = bar; + DataConsolidated?.Invoke(this, bar); } private void Rising(IBaseData data) diff --git a/Common/Data/Consolidators/SequentialConsolidator.cs b/Common/Data/Consolidators/SequentialConsolidator.cs index 485cebe75a47..052ee353ef83 100644 --- a/Common/Data/Consolidators/SequentialConsolidator.cs +++ b/Common/Data/Consolidators/SequentialConsolidator.cs @@ -112,14 +112,12 @@ public SequentialConsolidator(IDataConsolidator first, IDataConsolidator second) } /// - /// 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 /// /// The newly consolidated data - protected override void OnDataConsolidated(IBaseData consolidated) + protected override void FireDataConsolidated(IBaseData consolidated) { DataConsolidated?.Invoke(this, consolidated); - base.OnDataConsolidated(consolidated); } /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources. diff --git a/Common/Python/DataConsolidatorPythonWrapper.cs b/Common/Python/DataConsolidatorPythonWrapper.cs index cde0b04da3ed..199b22e50292 100644 --- a/Common/Python/DataConsolidatorPythonWrapper.cs +++ b/Common/Python/DataConsolidatorPythonWrapper.cs @@ -27,6 +27,7 @@ public class DataConsolidatorPythonWrapper : ConsolidatorBase { private readonly BasePythonWrapper _pythonWrapper; private readonly DataConsolidatedHandler _pythonDataConsolidated; + private bool _disposed; /// /// Gets a clone of the data being currently consolidated @@ -60,8 +61,14 @@ public DataConsolidatorPythonWrapper(PyObject consolidator) { _pythonWrapper = new BasePythonWrapper(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; + } } /// @@ -96,8 +103,19 @@ public override void Reset() /// 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(); } diff --git a/Tests/Common/Data/ConsolidatorBaseTests.cs b/Tests/Common/Data/ConsolidatorBaseTests.cs index 30374bca0871..2149445dcf83 100644 --- a/Tests/Common/Data/ConsolidatorBaseTests.cs +++ b/Tests/Common/Data/ConsolidatorBaseTests.cs @@ -79,6 +79,56 @@ public void HandlerSeesPreviousConsolidatedBarWhileReceivingTheNewOne() consolidator.Dispose(); } + [Test] + public void WindowHoldsTheNewBarInsideTypedHandler() + { + // regression for consolidators that fired their typed event before populating the window: + // inside the handler consolidator[0] must be the bar that was just produced + var reference = new DateTime(2015, 4, 13); + var spy = Symbols.SPY; + var consolidator = new RenkoConsolidator(1m); + var windowConsolidator = (ConsolidatorBase)consolidator; + + var handlerCalls = 0; + consolidator.DataConsolidated += (_, bar) => + { + handlerCalls++; + Assert.AreEqual(bar, windowConsolidator[0]); + Assert.AreEqual(bar, windowConsolidator.Current); + Assert.AreEqual(bar.Value, windowConsolidator.Current.Value); + }; + + consolidator.Update(new IndicatorDataPoint(spy, reference, 10m)); + consolidator.Update(new IndicatorDataPoint(spy, reference.AddMinutes(1), 12.1m)); + + Assert.Greater(handlerCalls, 0); + consolidator.Dispose(); + } + + [Test] + public void OutOfOrderDataDoesNotClearWindow() + { + // regression for count mode emitting a null bar on an out of order data point, which + // previously reset the whole rolling window through the Consolidated setter + var reference = new DateTime(2015, 4, 13); + var spy = Symbols.SPY; + var consolidator = new TradeBarConsolidator(1); + var windowConsolidator = (ConsolidatorBase)consolidator; + + consolidator.Update(new TradeBar { Symbol = spy, Time = reference, Close = 10m, Value = 10m, Period = Time.OneMinute }); + consolidator.Update(new TradeBar { Symbol = spy, Time = reference.AddMinutes(1), Close = 20m, Value = 20m, Period = Time.OneMinute }); + + Assert.AreEqual(2, windowConsolidator.Window.Count); + + consolidator.Update(new TradeBar { Symbol = spy, Time = reference.AddMinutes(-5), Close = 30m, Value = 30m, Period = Time.OneMinute }); + + Assert.AreEqual(2, windowConsolidator.Window.Count); + Assert.AreEqual(20m, windowConsolidator.Window[0].Value); + Assert.AreEqual(10m, windowConsolidator.Window[1].Value); + + consolidator.Dispose(); + } + [Test] public void CurrentAndPreviousAreNullBeforeFirstConsolidation() { From 3fabde1376da6d91ebf000dd42248d74f17e1233 Mon Sep 17 00:00:00 2001 From: Josue Nina Date: Tue, 7 Jul 2026 18:14:33 -0500 Subject: [PATCH 2/3] Unify DataConsolidated event and expose rolling window to Python consolidators --- Common/Data/Consolidators/ConsolidatorBase.cs | 22 +++++------ Common/Data/Consolidators/DataConsolidator.cs | 20 ---------- .../MarketHourAwareConsolidator.cs | 14 ------- .../Consolidators/SequentialConsolidator.cs | 15 ------- Common/Data/SubscriptionManager.cs | 31 ++++++++++++++- Common/Python/PythonConsolidator.cs | 11 +++++- Tests/Common/Data/ConsolidatorBaseTests.cs | 22 +++++++++++ .../DataConsolidatorPythonWrapperTests.cs | 39 +++++++++++++++++++ 8 files changed, 111 insertions(+), 63 deletions(-) diff --git a/Common/Data/Consolidators/ConsolidatorBase.cs b/Common/Data/Consolidators/ConsolidatorBase.cs index 05b4f8516746..0b853310efcc 100644 --- a/Common/Data/Consolidators/ConsolidatorBase.cs +++ b/Common/Data/Consolidators/ConsolidatorBase.cs @@ -24,8 +24,6 @@ namespace QuantConnect.Data.Consolidators /// public abstract class ConsolidatorBase : WindowBase, IDataConsolidator { - private DataConsolidatedHandler _dataConsolidated; - private IBaseData _consolidated; /// @@ -72,13 +70,11 @@ protected set public abstract void Scan(DateTime currentLocalTime); /// - /// 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 interface and by derived consolidators whose + /// output is a base data bar, so subscribing and unsubscribing always target the same handler list. /// - event DataConsolidatedHandler IDataConsolidator.DataConsolidated - { - add { _dataConsolidated += value; } - remove { _dataConsolidated -= value; } - } + public event DataConsolidatedHandler DataConsolidated; /// /// Event invocator for the DataConsolidated event. Populates the rolling window, raises the @@ -97,7 +93,7 @@ protected virtual void OnDataConsolidated(IBaseData consolidated) // let derived consolidators raise their strongly typed DataConsolidated event FireDataConsolidated(consolidated); - _dataConsolidated?.Invoke(this, 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 @@ -106,8 +102,10 @@ protected virtual void OnDataConsolidated(IBaseData consolidated) } /// - /// Raises the strongly typed DataConsolidated event exposed by derived consolidators. Invoked after - /// the rolling window is populated and before the interface event so every handler sees the same window. + /// 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 event already carries their bar. /// /// The newly consolidated data protected virtual void FireDataConsolidated(IBaseData consolidated) @@ -119,7 +117,7 @@ protected virtual void FireDataConsolidated(IBaseData consolidated) /// public virtual void Dispose() { - _dataConsolidated = null; + DataConsolidated = null; } /// diff --git a/Common/Data/Consolidators/DataConsolidator.cs b/Common/Data/Consolidators/DataConsolidator.cs index 33bdd23a8cea..045eece3b3d8 100644 --- a/Common/Data/Consolidators/DataConsolidator.cs +++ b/Common/Data/Consolidators/DataConsolidator.cs @@ -47,11 +47,6 @@ public override void Update(IBaseData data) /// The current time in the local time zone (same as ) public abstract override void Scan(DateTime currentLocalTime); - /// - /// Event handler that fires when a new piece of data is produced - /// - public event DataConsolidatedHandler DataConsolidated; - /// /// Gets a clone of the data being currently consolidated /// @@ -74,20 +69,5 @@ public override void Update(IBaseData data) /// The new data for the consolidator public abstract void Update(TInput data); - /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources. - /// 2 - public override void Dispose() - { - DataConsolidated = null; - base.Dispose(); - } - - /// - /// Raises the strongly typed DataConsolidated event - /// - protected override void FireDataConsolidated(IBaseData consolidated) - { - DataConsolidated?.Invoke(this, consolidated); - } } } \ No newline at end of file diff --git a/Common/Data/Consolidators/MarketHourAwareConsolidator.cs b/Common/Data/Consolidators/MarketHourAwareConsolidator.cs index f6ec272df5c4..8cdda4175d2b 100644 --- a/Common/Data/Consolidators/MarketHourAwareConsolidator.cs +++ b/Common/Data/Consolidators/MarketHourAwareConsolidator.cs @@ -166,11 +166,6 @@ protected virtual IDataConsolidator CreateConsolidator(Func - /// Event handler that fires when a new piece of data is produced - /// - public event DataConsolidatedHandler DataConsolidated; - /// /// Updates this consolidator with the specified data /// @@ -279,14 +274,5 @@ protected virtual void ForwardConsolidatedBar(object sender, IBaseData consolida { OnDataConsolidated(consolidated); } - - /// - /// Raises the strongly typed DataConsolidated event - /// - /// The newly consolidated data - protected override void FireDataConsolidated(IBaseData consolidated) - { - DataConsolidated?.Invoke(this, consolidated); - } } } diff --git a/Common/Data/Consolidators/SequentialConsolidator.cs b/Common/Data/Consolidators/SequentialConsolidator.cs index 052ee353ef83..04746c9262fa 100644 --- a/Common/Data/Consolidators/SequentialConsolidator.cs +++ b/Common/Data/Consolidators/SequentialConsolidator.cs @@ -83,11 +83,6 @@ public override void Scan(DateTime currentLocalTime) First.Scan(currentLocalTime); } - /// - /// Event handler that fires when a new piece of data is produced - /// - public event DataConsolidatedHandler DataConsolidated; - /// /// 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,22 +106,12 @@ public SequentialConsolidator(IDataConsolidator first, IDataConsolidator second) second.DataConsolidated += (sender, consolidated) => OnDataConsolidated(consolidated); } - /// - /// Raises the strongly typed DataConsolidated event - /// - /// The newly consolidated data - protected override void FireDataConsolidated(IBaseData consolidated) - { - DataConsolidated?.Invoke(this, consolidated); - } - /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources. /// 2 public override void Dispose() { First.Dispose(); Second.Dispose(); - DataConsolidated = null; base.Dispose(); } diff --git a/Common/Data/SubscriptionManager.cs b/Common/Data/SubscriptionManager.cs index 3fe7f90822c7..b5a794e85e45 100644 --- a/Common/Data/SubscriptionManager.cs +++ b/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); } + /// + /// Finds the 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. + /// + 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; + } + /// /// Will trigger past consolidator scans /// diff --git a/Common/Python/PythonConsolidator.cs b/Common/Python/PythonConsolidator.cs index 4c461639b86b..aade6d3d8ce5 100644 --- a/Common/Python/PythonConsolidator.cs +++ b/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 { /// /// 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. /// - public class PythonConsolidator : IDataConsolidator + public class PythonConsolidator : WindowBase, IDataConsolidator { /// /// Gets the most recently consolidated piece of data. This will be null if this consolidator @@ -70,6 +73,11 @@ public Type OutputType /// The finished data from the consolidator 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(); } /// diff --git a/Tests/Common/Data/ConsolidatorBaseTests.cs b/Tests/Common/Data/ConsolidatorBaseTests.cs index 2149445dcf83..7ac15ee392ae 100644 --- a/Tests/Common/Data/ConsolidatorBaseTests.cs +++ b/Tests/Common/Data/ConsolidatorBaseTests.cs @@ -105,6 +105,28 @@ public void WindowHoldsTheNewBarInsideTypedHandler() consolidator.Dispose(); } + [Test] + public void InterfaceAndConcreteDataConsolidatedShareOneSubscriptionList() + { + // regression for the interface event and the concrete event being two separate handler lists: + // subscribing through one and unsubscribing through the other must cancel out + var reference = new DateTime(2015, 4, 13); + var spy = Symbols.SPY; + var consolidator = new IdentityDataConsolidator(); + IDataConsolidator asInterface = consolidator; + + var calls = 0; + DataConsolidatedHandler handler = (_, __) => calls++; + + asInterface.DataConsolidated += handler; + consolidator.DataConsolidated -= handler; + + consolidator.Update(new TradeBar { Symbol = spy, Time = reference, Close = 10m, Value = 10m, Period = Time.OneMinute }); + + Assert.AreEqual(0, calls); + consolidator.Dispose(); + } + [Test] public void OutOfOrderDataDoesNotClearWindow() { diff --git a/Tests/Python/DataConsolidatorPythonWrapperTests.cs b/Tests/Python/DataConsolidatorPythonWrapperTests.cs index 914e265feb11..540527d875d7 100644 --- a/Tests/Python/DataConsolidatorPythonWrapperTests.cs +++ b/Tests/Python/DataConsolidatorPythonWrapperTests.cs @@ -281,6 +281,45 @@ private static void FeedConsolidation(DataConsolidatorPythonWrapper wrapper, int } } + [Test] + public void PythonConsolidatorSubclassExposesWindow() + { + // custom consolidators that inherit PythonConsolidator (the documented path) must expose the + // rolling window just like the C# consolidators, not only the ones routed through the wrapper + using (Py.GIL()) + { + var module = PyModule.FromString(Guid.NewGuid().ToString(), + "from AlgorithmImports import *\n" + + "class CustomConsolidator(PythonConsolidator):\n" + + " def __init__(self):\n" + + " self.input_type = QuoteBar\n" + + " self.output_type = QuoteBar\n" + + " self.consolidated = None\n" + + " self.working_data = None\n" + + " def update(self, data):\n" + + " pass\n" + + " def scan(self, time):\n" + + " pass\n"); + + var customConsolidator = module.GetAttr("CustomConsolidator").Invoke(); + var consolidator = customConsolidator.As(); + + var time = DateTime.Today; + var bar1 = new QuoteBar { Time = time, Symbol = Symbols.SPY, Bid = new Bar(1, 2, 0.75m, 1.25m), Value = 1, Period = TimeSpan.FromMinutes(1) }; + var bar2 = new QuoteBar { Time = time.AddMinutes(1), Symbol = Symbols.SPY, Bid = new Bar(1, 2, 0.75m, 1.25m), Value = 2, Period = TimeSpan.FromMinutes(1) }; + + consolidator.OnDataConsolidated(customConsolidator, bar1); + consolidator.OnDataConsolidated(customConsolidator, bar2); + + Assert.AreEqual(2, consolidator.Window.Count); + Assert.AreEqual(bar2, consolidator[0]); + Assert.AreEqual(bar1, consolidator.Previous); + + consolidator.Reset(); + Assert.AreEqual(0, consolidator.Window.Count); + } + } + [Test] public void AttachAndTriggerEvent() { From 5a320fafd49601c690131945e35a2b28a79b32c5 Mon Sep 17 00:00:00 2001 From: Martin Molinero Date: Wed, 8 Jul 2026 10:00:42 -0300 Subject: [PATCH 3/3] Minor fixes --- Common/Data/Consolidators/BaseTimelessConsolidator.cs | 2 +- Common/Data/Consolidators/RenkoConsolidator.cs | 6 ++++-- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/Common/Data/Consolidators/BaseTimelessConsolidator.cs b/Common/Data/Consolidators/BaseTimelessConsolidator.cs index 60aba0f26596..e763d63282a0 100644 --- a/Common/Data/Consolidators/BaseTimelessConsolidator.cs +++ b/Common/Data/Consolidators/BaseTimelessConsolidator.cs @@ -60,7 +60,7 @@ public abstract class BaseTimelessConsolidator : ConsolidatorBase /// /// Typed event handler that fires when a new piece of data is produced /// - public event EventHandler DataConsolidated; + public new event EventHandler DataConsolidated; /// /// Initializes a new instance of the class. diff --git a/Common/Data/Consolidators/RenkoConsolidator.cs b/Common/Data/Consolidators/RenkoConsolidator.cs index 1f280db2b8c5..c7d88c3adb70 100644 --- a/Common/Data/Consolidators/RenkoConsolidator.cs +++ b/Common/Data/Consolidators/RenkoConsolidator.cs @@ -95,7 +95,7 @@ public class RenkoConsolidator : ConsolidatorBase /// /// Typed event handler that fires when a new piece of data is produced /// - public event EventHandler DataConsolidated; + public new event EventHandler DataConsolidated; /// /// Initializes a new instance of the class using the specified . @@ -244,8 +244,10 @@ public override void Reset() protected override void FireDataConsolidated(IBaseData consolidated) { var bar = (RenkoBar)consolidated; - _currentBar = bar; + // 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)