Fix bug when consolidating hourly bars into daily ones (#8442)

* First draft of the solution

* Add regression tests

* Improve solution and add unit test

* Address suggestions

* Improve unit tests

* Nit changes

* Fix regression algorithms

Algorithm.CSharp/ConsolidateHourBarsIntoDailyBarsRegressionAlgorithm.cs

@@ -0,0 +1,166 @@
+/*
+ * 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 QuantConnect.Data;
+using QuantConnect.Indicators;
+using QuantConnect.Interfaces;
+using System;
+using System.Collections.Generic;
+using QuantConnect.Data.Market;
+
+namespace QuantConnect.Algorithm.CSharp
+{
+    /// <summary>
+    /// This regression algorithm asserts the consolidated US equity daily bars from the hour bars exactly matches
+    /// the daily bars returned from the database
+    /// </summary>
+    public class ConsolidateHourBarsIntoDailyBarsRegressionAlgorithm : QCAlgorithm, IRegressionAlgorithmDefinition
+    {
+        private Symbol _spy;
+        private RelativeStrengthIndex _rsi;
+        private RelativeStrengthIndex _rsiTimeDelta;
+        private Dictionary<DateTime, decimal> _values = new();
+        private int _count;
+        private bool _indicatorsCompared;
+
+        public override void Initialize()
+        {
+            SetStartDate(2020, 5, 1);
+            SetEndDate(2020, 6, 5);
+
+            _spy = AddEquity("SPY", Resolution.Hour).Symbol;
+
+            // We will use these two indicators to compare the daily consolidated bars equals
+            // the ones returned from the database. We use this specific type of indicator as
+            // it depends on its previous values. Thus, if at some point the bars received by
+            // the indicators differ, so will their final values
+            _rsi = new RelativeStrengthIndex("FIRST", 15, MovingAverageType.Wilders);
+            RegisterIndicator(_spy, _rsi, Resolution.Daily, selector: (bar) =>
+            {
+                var tradeBar = (TradeBar)bar;
+                return (tradeBar.Close + tradeBar.Open) / 2;
+            });
+
+            // We won't register this indicator as we will update it manually at the end of the
+            // month, so that we can compare the values of the indicator that received consolidated
+            // bars and the values of this one
+            _rsiTimeDelta = new RelativeStrengthIndex("SECOND" ,15, MovingAverageType.Wilders);
+        }
+
+        public override void OnData(Slice slice)
+        {
+            if (IsWarmingUp) return;
+
+            if (slice.ContainsKey(_spy) && slice[_spy] != null)
+            {
+                if (Time.Month == EndDate.Month)
+                {
+                    var history = History(_spy, _count, Resolution.Daily);
+                    foreach (var bar in history)
+                    {
+                        var time = bar.EndTime.Date;
+                        var average = (bar.Close + bar.Open) / 2;
+                        _rsiTimeDelta.Update(bar.EndTime, average);
+                        if (_rsiTimeDelta.Current.Value != _values[time])
+                        {
+                            throw new RegressionTestException($"Both {_rsi.Name} and {_rsiTimeDelta.Name} should have the same values, but they differ. {_rsi.Name}: {_values[time]} | {_rsiTimeDelta.Name}: {_rsiTimeDelta.Current.Value}");
+                        }
+                    }
+                    _indicatorsCompared = true;
+                    Quit();
+                }
+                else
+                {
+                    _values[Time.Date] = _rsi.Current.Value;
+
+                    // Since the symbol resolution is hour and the symbol is equity, we know the last bar received in a day will
+                    // be at the market close, this is 16h. We need to count how many daily bars were consolidated in order to know
+                    // how many we need to request from the history
+                    if (Time.Hour == 16)
+                    {
+                        _count++;
+                    }
+                }
+            }
+        }
+
+        public override void OnEndOfAlgorithm()
+        {
+            if (!_indicatorsCompared)
+            {
+                throw new RegressionTestException($"Indicators {_rsi.Name} and {_rsiTimeDelta.Name} should have been compared, but they were not. Please make sure the indicators are getting SPY data");
+            }
+        }
+
+        /// <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 => 290;
+
+        /// <summary>
+        /// Data Points count of the algorithm history
+        /// </summary>
+        public int AlgorithmHistoryDataPoints => 20;
+
+        /// <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", "-5.215"},
+            {"Tracking Error", "0.159"},
+            {"Treynor Ratio", "0"},
+            {"Total Fees", "$0.00"},
+            {"Estimated Strategy Capacity", "$0"},
+            {"Lowest Capacity Asset", ""},
+            {"Portfolio Turnover", "0%"},
+            {"OrderListHash", "d41d8cd98f00b204e9800998ecf8427e"}
+        };
+    }
+}

Algorithm.Python/ConsolidateHourBarsIntoDailyBarsRegressionAlgorithm.py

@@ -0,0 +1,69 @@
+# 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>
+### This regression algorithm asserts the consolidated US equity daily bars from the hour bars exactly matches
+### the daily bars returned from the database
+### </summary>
+class ConsolidateHourBarsIntoDailyBarsRegressionAlgorithm(QCAlgorithm):
+    def initialize(self):
+        self.set_start_date(2020, 5, 1)
+        self.set_end_date(2020, 6, 5)
+
+        self.spy = self.add_equity("SPY", Resolution.HOUR).symbol
+
+        # We will use these two indicators to compare the daily consolidated bars equals
+        # the ones returned from the database. We use this specific type of indicator as
+        # it depends on its previous values. Thus, if at some point the bars received by
+        # the indicators differ, so will their final values
+        self._rsi = RelativeStrengthIndex("First", 15, MovingAverageType.WILDERS)
+        self.register_indicator(self.spy, self._rsi, Resolution.DAILY, selector= lambda bar: (bar.close + bar.open) / 2)
+
+        # We won't register this indicator as we will update it manually at the end of the
+        # month, so that we can compare the values of the indicator that received consolidated
+        # bars and the values of this one
+        self._rsi_timedelta = RelativeStrengthIndex("Second", 15, MovingAverageType.WILDERS)
+        self._values = {}
+        self.count = 0;
+        self._indicators_compared = False;
+
+    def on_data(self, data: Slice):
+        if self.is_warming_up:
+            return
+
+        if data.contains_key(self.spy) and data[self.spy] != None:
+            if self.time.month == self.end_date.month:
+                history = self.history[TradeBar](self.spy, self.count, Resolution.DAILY)
+                for bar in history:
+                    time = bar.end_time.strftime('%Y-%m-%d')
+                    average = (bar.close + bar.open) / 2
+                    self._rsi_timedelta.update(bar.end_time, average)
+                    if self._rsi_timedelta.current.value != self._values[time]:
+                        raise Exception(f"Both {self._rsi.name} and {self._rsi_timedelta.name} should have the same values, but they differ. {self._rsi.name}: {self._values[time]} | {self._rsi_timedelta.name}: {self._rsi_timedelta.current.value}")
+                self._indicators_compared = True
+                self.quit()
+            else:
+                time = self.time.strftime('%Y-%m-%d')
+                self._values[time] = self._rsi.current.value
+
+                # Since the symbol resolution is hour and the symbol is equity, we know the last bar received in a day will
+                # be at the market close, this is 16h. We need to count how many daily bars were consolidated in order to know
+                # how many we need to request from the history
+                if self.time.hour == 16:
+                    self.count += 1
+
+    def on_end_of_algorithm(self):
+        if not self._indicators_compared:
+            raise Exception(f"Indicators {self._rsi.name} and {self._rsi_timedelta.name} should have been compared, but they were not. Please make sure the indicators are getting SPY data")

Common/Data/Consolidators/MarketHourAwareConsolidator.cs

@@ -132,7 +132,12 @@ public virtual void Update(IBaseData data)
         {
             Initialize(data);
 
-            if (_extendedMarketHours || ExchangeHours.IsOpen(data.Time, false))
+            // US equity hour data from the database starts at 9am but the exchange opens at 9:30am. Thus, we need to handle
+            // this case specifically to avoid skipping the first hourly bar. To avoid this, we assert the period is daily,
+            // the data resolution is hour and the exchange opens at any point in time over the data.Time to data.EndTime interval
+            if (_extendedMarketHours ||
+                ExchangeHours.IsOpen(data.Time, false) ||
+                (Period == Time.OneDay && (data.EndTime - data.Time == Time.OneHour) && ExchangeHours.IsOpen(data.Time, data.EndTime, false)))
             {
                 Consolidator.Update(data);
             }

Common/Data/Consolidators/PeriodCountConsolidatorBase.cs

@@ -322,10 +322,21 @@ protected DateTime GetRoundedBarTime(DateTime time)
         protected DateTime GetRoundedBarTime(IBaseData inputData)
         {
             var potentialStartTime = GetRoundedBarTime(inputData.Time);
-            if(_period.HasValue && potentialStartTime + _period < inputData.EndTime)
+            if (_period.HasValue && potentialStartTime + _period < inputData.EndTime)
             {
-                // whops! the end time we were giving is beyond our potential end time, so let's use the giving bars star time instead
-                potentialStartTime = inputData.Time;
+                // US equity hour bars from the database starts at 9am but the exchange opens at 9:30am. Thus, the method
+                // GetRoundedBarTime(inputData.Time) returns the market open of the previous day, which is not consistent
+                // with the given end time. For that reason we need to handle this case specifically, by calling
+                // GetRoundedBarTime(inputData.EndTime) as it will return our expected start time: 9:30am
+                if (inputData.EndTime - inputData.Time == Time.OneHour && potentialStartTime.Date < inputData.Time.Date)
+                {
+                    potentialStartTime = GetRoundedBarTime(inputData.EndTime);
+                }
+                else
+                {
+                    // whops! the end time we were giving is beyond our potential end time, so let's use the giving bars star time instead
+                    potentialStartTime = inputData.Time;
+                }
             }
 
             return potentialStartTime;

Tests/Common/Data/MarketHourAwareConsolidatorTests.cs

@@ -117,6 +117,71 @@ public void Daily(bool strictEndTime)
             Assert.AreEqual(1, latestBar.Low);
         }
 
+        [Test]
+        public void BarIsSkippedWhenDataResolutionIsNotHourAndMarketIsClose()
+        {
+            var symbol = Symbols.SPY;
+            using var consolidator = new MarketHourAwareConsolidator(true, Resolution.Daily, typeof(TradeBar), TickType.Trade, false);
+            var consolidatedBarsCount = 0;
+            TradeBar latestBar = null;
+
+            consolidator.DataConsolidated += (sender, bar) =>
+            {
+                latestBar = (TradeBar)bar;
+                consolidatedBarsCount++;
+            };
+
+            var time = new DateTime(2020, 05, 01, 09, 30, 0);
+            // this bar will be ignored because it's during market closed hours and the bar resolution is not Hour
+            consolidator.Update(new TradeBar() { Time = time.Subtract(Time.OneMinute), Period = Time.OneMinute, Symbol = symbol, Open = 1 });
+            Assert.IsNull(latestBar);
+            Assert.AreEqual(0, consolidatedBarsCount);
+        }
+
+        [Test]
+        public void DailyBarCanBeConsolidatedFromHourData()
+        {
+            var symbol =  Symbols.SPY;
+            using var consolidator = new MarketHourAwareConsolidator(true, Resolution.Daily, typeof(TradeBar), TickType.Trade, false);
+            var consolidatedBarsCount = 0;
+            TradeBar latestBar = null;
+
+            consolidator.DataConsolidated += (sender, bar) =>
+            {
+                latestBar = (TradeBar)bar;
+                consolidatedBarsCount++;
+            };
+
+            var time = new DateTime(2020, 05, 01, 09, 0, 0);
+            var hourBars = new List<TradeBar>()
+            {
+                new TradeBar() { Time = time, Period = Time.OneHour, Symbol = symbol, Open = 2 },
+                new TradeBar() { Time = time.AddHours(1), Period = Time.OneHour, Symbol = symbol, High = 200 },
+                new TradeBar() { Time = time.AddHours(2), Period = Time.OneHour, Symbol = symbol, Low = 0.02m },
+                new TradeBar() { Time = time.AddHours(3), Period = Time.OneHour, Symbol = symbol, Close = 20 },
+                new TradeBar() { Time = time.AddHours(4), Period = Time.OneHour, Symbol = symbol, Open = 3 },
+                new TradeBar() { Time = time.AddHours(5), Period = Time.OneHour, Symbol = symbol, High = 300 },
+                new TradeBar() { Time = time.AddHours(6), Period = Time.OneHour, Symbol = symbol, Low = 0.03m, Close = 30 },
+            };
+
+            foreach (var bar in hourBars)
+            {
+                consolidator.Update(bar);
+            }
+
+            consolidator.Scan(time.AddHours(7));
+
+            // Assert that the bar emitted
+            Assert.IsNotNull(latestBar);
+            Assert.AreEqual(time.AddHours(7), latestBar.EndTime);
+            Assert.AreEqual(time.AddMinutes(30), latestBar.Time);
+            Assert.AreEqual(1, consolidatedBarsCount);
+            Assert.AreEqual(2, latestBar.Open);
+            Assert.AreEqual(300, latestBar.High);
+            Assert.AreEqual(0.02, latestBar.Low);
+            Assert.AreEqual(30, latestBar.Close);
+        }
+
         [TestCase(true)]
         [TestCase(false)]
         public void DailyExtendedMarketHours(bool strictEndTime)