WIP: Add built-in rolling window to all consolidators

Algorithm.CSharp/ConsolidatorRollingWindowRegressionAlgorithm.cs

@@ -0,0 +1,153 @@
+/*
+ * 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;
+using QuantConnect.Data.Consolidators;
+using QuantConnect.Data.Market;
+using QuantConnect.Interfaces;
+
+namespace QuantConnect.Algorithm.CSharp
+{
+    /// <summary>
+    /// Regression algorithm asserting that consolidators expose a built-in rolling window
+    /// </summary>
+    public class ConsolidatorRollingWindowRegressionAlgorithm : QCAlgorithm, IRegressionAlgorithmDefinition
+    {
+        private TradeBarConsolidator _consolidator;
+        private int _consolidationCount;
+
+        /// <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, 10, 07);
+            SetEndDate(2013, 10, 11);
+
+            AddEquity("SPY", Resolution.Minute);
+
+            _consolidator = new TradeBarConsolidator(TimeSpan.FromMinutes(10));
+            _consolidator.DataConsolidated += OnDataConsolidated;
+            SubscriptionManager.AddConsolidator("SPY", _consolidator);
+        }
+
+        private void OnDataConsolidated(object sender, TradeBar bar)
+        {
+            _consolidationCount++;
+
+            // Window[0] must always be the bar just consolidated
+            var currentBar = (TradeBar)_consolidator[0];
+            if (currentBar.Time != bar.Time)
+            {
+                throw new RegressionTestException($"Expected consolidator[0].Time == {bar.Time} but was {currentBar.Time}");
+            }
+            if (currentBar.Close != bar.Close)
+            {
+                throw new RegressionTestException($"Expected consolidator[0].Close == {bar.Close} but was {currentBar.Close}");
+            }
+
+            // After the second consolidation the previous bar must be accessible at index 1
+            if (_consolidator.Window.Count >= 2)
+            {
+                var previous = (TradeBar)_consolidator[1];
+                if (previous.Time >= bar.Time)
+                {
+                    throw new RegressionTestException($"consolidator[1].Time ({previous.Time}) should be earlier than consolidator[0].Time ({bar.Time})");
+                }
+                if (previous.Close <= 0)
+                {
+                    throw new RegressionTestException("consolidator[1].Close should be greater than zero");
+                }
+            }
+        }
+
+        public override void OnEndOfAlgorithm()
+        {
+            if (_consolidationCount == 0)
+            {
+                throw new RegressionTestException("Expected at least one consolidation but got zero");
+            }
+
+            // Default window size is 2, it must be full
+            if (_consolidator.Window.Count != 2)
+            {
+                throw new RegressionTestException(
+                    $"Expected window count of 2 but was {_consolidator.Window.Count}");
+            }
+        }
+
+        /// <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, Language.Python };
+
+        /// <summary>
+        /// Data Points count of all timeslices of algorithm
+        /// </summary>
+        public long DataPoints => 3943;
+
+        /// <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", "-8.91"},
+            {"Tracking Error", "0.223"},
+            {"Treynor Ratio", "0"},
+            {"Total Fees", "$0.00"},
+            {"Estimated Strategy Capacity", "$0"},
+            {"Lowest Capacity Asset", ""},
+            {"Portfolio Turnover", "0%"},
+            {"Drawdown Recovery", "0"},
+            {"OrderListHash", "d41d8cd98f00b204e9800998ecf8427e"}
+        };
+    }
+}

Algorithm.Python/ConsolidatorRollingWindowRegressionAlgorithm.py

@@ -0,0 +1,62 @@
+# 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.
+
+from AlgorithmImports import *
+
+### <summary>
+### Regression algorithm asserting that consolidators expose a built-in rolling window
+### </summary>
+class ConsolidatorRollingWindowRegressionAlgorithm(QCAlgorithm):
+
+    def initialize(self):
+        self.set_start_date(2013, 10, 7)
+        self.set_end_date(2013, 10, 11)
+
+        self.add_equity("SPY", Resolution.MINUTE)
+
+        self._consolidation_count = 0
+        self._consolidator = TradeBarConsolidator(timedelta(minutes=10))
+        self._consolidator.data_consolidated += self._on_data_consolidated
+        self.subscription_manager.add_consolidator("SPY", self._consolidator)
+
+    def _on_data_consolidated(self, sender, bar):
+        self._consolidation_count += 1
+
+        # consolidator[0] must always match the bar just fired
+        currentBar = self._consolidator[0]
+        if currentBar.time != bar.time:
+            raise AssertionError(f"Expected consolidator[0].time == {bar.time} but was {currentBar.time}")
+        if currentBar.value != bar.close:
+            raise AssertionError(f"Expected consolidator[0].value == {bar.close} but was {currentBar.value}")
+
+        # After the second consolidation the previous bar must be at index 1
+        if self._consolidator.window.count >= 2:
+            previous = self._consolidator[1]
+            if previous.time >= bar.time:
+                raise AssertionError(
+                    f"consolidator[1].time ({previous.time}) should be earlier "
+                    f"than consolidator[0].time ({bar.time})"
+                )
+            if previous.value <= 0:
+                raise AssertionError("consolidator[1].value should be greater than zero")
+
+    def on_data(self, data):
+        pass
+
+    def on_end_of_algorithm(self):
+        if self._consolidation_count == 0:
+            raise AssertionError("Expected at least one consolidation but got zero")
+
+        # Default window size is 2, it must be full
+        if self._consolidator.window.count != 2:
+            raise AssertionError(f"Expected window count of 2 but was {self._consolidator.window.count}")

Common/Data/Consolidators/BaseTimelessConsolidator.cs

@@ -23,7 +23,7 @@ namespace QuantConnect.Data.Consolidators
     /// Represents a timeless consolidator which depends on the given values. This consolidator
     /// is meant to consolidate data into bars that do not depend on time, e.g., RangeBar's.
     /// </summary>
-    public abstract class BaseTimelessConsolidator<T> : IDataConsolidator
+    public abstract class BaseTimelessConsolidator<T> : ConsolidatorBase, IDataConsolidator
         where T : IBaseData
     {
         /// <summary>
@@ -47,12 +47,6 @@ public abstract class BaseTimelessConsolidator<T> : IDataConsolidator
         /// </summary>
         protected virtual T CurrentBar {  get; set; }
 
-        /// <summary>
-        /// Gets the most recently consolidated piece of data. This will be null if this consolidator
-        /// has not produced any data yet.
-        /// </summary>
-        public IBaseData Consolidated { get; protected set; }
-
         /// <summary>
         /// Gets a clone of the data being currently consolidated
         /// </summary>
@@ -188,7 +182,7 @@ protected void OnDataConsolidated(T consolidated)
 
             DataConsolidatedHandler?.Invoke(this, consolidated);
 
-            Consolidated = consolidated;
+            UpdateConsolidated(consolidated);
         }
 
         /// <summary>Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.</summary>
@@ -202,10 +196,10 @@ public virtual void Dispose()
         /// <summary>
         /// Resets the consolidator
         /// </summary>
-        public virtual void Reset()
+        public override void Reset()
         {
-            Consolidated = null;
             CurrentBar = default(T);
+            base.Reset();
         }
 
         /// <summary>

Common/Data/Consolidators/ConsolidatorBase.cs

@@ -0,0 +1,91 @@
+/*
+ * 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.Collections;
+using System.Collections.Generic;
+using QuantConnect.Indicators;
+
+namespace QuantConnect.Data.Consolidators
+{
+    /// <summary>
+    /// Provides a base implementation for consolidators, including a built-in rolling window
+    /// that stores the history of consolidated bars.
+    /// </summary>
+    public abstract class ConsolidatorBase : IEnumerable<IBaseData>
+    {
+        /// <summary>
+        /// The default number of consolidated bars to keep in the rolling window history
+        /// </summary>
+        public static int DefaultWindowSize { get; } = 2;
+
+        private RollingWindow<IBaseData> _window;
+
+        /// <summary>
+        /// A rolling window keeping a history of the consolidated bars. The most recent bar is at index 0.
+        /// </summary>
+        public RollingWindow<IBaseData> Window
+        {
+            get
+            {
+                if (_window == null)
+                {
+                    _window = new RollingWindow<IBaseData>(DefaultWindowSize);
+                }
+                return _window;
+            }
+        }
+
+        /// <summary>
+        /// Gets the most recently consolidated piece of data. This will be null if this consolidator
+        /// has not produced any data yet.
+        /// </summary>
+        public IBaseData Consolidated { get; protected set; }
+
+        /// <summary>
+        /// Indexes the history window, where index 0 is the most recently consolidated bar.
+        /// </summary>
+        /// <param name="i">The index</param>
+        /// <returns>The ith most recently consolidated bar</returns>
+        public IBaseData this[int i] => Window[i];
+
+        /// <summary>
+        /// Returns an enumerator that iterates through the history window.
+        /// </summary>
+        public IEnumerator<IBaseData> GetEnumerator() => Window.GetEnumerator();
+
+        /// <summary>
+        /// Returns an enumerator that iterates through the history window.
+        /// </summary>
+        IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
+
+        /// <summary>
+        /// Updates <see cref="Consolidated"/> and adds the bar to the rolling window.
+        /// </summary>
+        protected void UpdateConsolidated(IBaseData consolidated)
+        {
+            Consolidated = consolidated;
+            Window.Add(consolidated);
+        }
+
+        /// <summary>
+        /// Resets this consolidator, clearing consolidated data and the rolling window.
+        /// </summary>
+        public virtual void Reset()
+        {
+            Consolidated = null;
+            _window?.Reset();
+        }
+    }
+}