Add built-in rolling window to all consolidators

Algorithm.CSharp/ConsolidatorRollingWindowRegressionAlgorithm.cs

@@ -0,0 +1,162 @@
+/*
+ * 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++;
+
+            if (_consolidator.Current != _consolidator[0])
+            {
+                throw new RegressionTestException("Expected Current to be the same as Window[0]");
+            }
+
+            // 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 (_consolidator.Previous != _consolidator[1])
+                {
+                    throw new RegressionTestException("Expected Previous to be the same as Window[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,67 @@
+# 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
+
+        if self._consolidator.current != self._consolidator[0]:
+            raise AssertionError("Expected current to be the same as window[0]")
+
+        # 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 self._consolidator.previous != self._consolidator[1]:
+                raise AssertionError("Expected previous to be the same as window[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
         where T : IBaseData
     {
         /// <summary>
@@ -37,51 +37,31 @@ public abstract class BaseTimelessConsolidator<T> : IDataConsolidator
         /// </summary>
         protected Func<IBaseData, decimal> VolumeSelector { get; set; }
 
-        /// <summary>
-        /// Event handler type for the IDataConsolidator.DataConsolidated event
-        /// </summary>
-        protected DataConsolidatedHandler DataConsolidatedHandler { get; set; }
-
         /// <summary>
         /// Bar being created
         /// </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>
-        public abstract IBaseData WorkingData { get; }
+        public abstract override IBaseData WorkingData { get; }
 
         /// <summary>
         /// Gets the type consumed by this consolidator
         /// </summary>
-        public Type InputType => typeof(IBaseData);
+        public override Type InputType => typeof(IBaseData);
 
         /// <summary>
         /// Gets <see cref="T"/> which is the type emitted in the <see cref="IDataConsolidator.DataConsolidated"/> event.
         /// </summary>
-        public virtual Type OutputType => typeof(T);
+        public override Type OutputType => typeof(T);
 
         /// <summary>
-        /// Event handler that fires when a new piece of data is produced
+        /// Typed event handler that fires when a new piece of data is produced
         /// </summary>
         public event EventHandler<T> DataConsolidated;
 
-        /// <summary>
-        /// Event handler that fires when a new piece of data is produced
-        /// </summary>
-        event DataConsolidatedHandler IDataConsolidator.DataConsolidated
-        {
-            add { DataConsolidatedHandler += value; }
-            remove { DataConsolidatedHandler -= value; }
-        }
-
         /// <summary>
         /// Initializes a new instance of the <see cref="BaseTimelessConsolidator{T}" /> class.
         /// </summary>
@@ -141,7 +121,7 @@ private static Func<IBaseData, decimal> TryToConvertSelector(PyObject selector,
         /// Updates this consolidator with the specified data
         /// </summary>
         /// <param name="data">The new data for the consolidator</param>
-        public void Update(IBaseData data)
+        public override void Update(IBaseData data)
         {
             var currentValue = Selector(data);
             var volume = VolumeSelector(data);
@@ -185,34 +165,31 @@ public void Update(IBaseData data)
         protected void OnDataConsolidated(T consolidated)
         {
             DataConsolidated?.Invoke(this, consolidated);
-
-            DataConsolidatedHandler?.Invoke(this, consolidated);
-
-            Consolidated = consolidated;
+            base.OnDataConsolidated(consolidated);
         }
 
         /// <summary>Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.</summary>
         /// <filterpriority>2</filterpriority>
-        public virtual void Dispose()
+        public override void Dispose()
         {
             DataConsolidated = null;
-            DataConsolidatedHandler = null;
+            base.Dispose();
         }
 
         /// <summary>
         /// Resets the consolidator
         /// </summary>
-        public virtual void Reset()
+        public override void Reset()
         {
-            Consolidated = null;
             CurrentBar = default(T);
+            base.Reset();
         }
 
         /// <summary>
         /// Scans this consolidator to see if it should emit a bar due to time passing
         /// </summary>
         /// <param name="currentLocalTime">The current time in the local time zone (same as <see cref="BaseData.Time"/>)</param>
-        public void Scan(DateTime currentLocalTime)
+        public override void Scan(DateTime currentLocalTime)
         {
         }
     }