Improve consolidator registration ergonomics: quote-to-trade bar adaptation, registration-time period validation, calendar periods in RegisterIndicator

Algorithm.CSharp/ConsolidatorAutoAdaptationRegressionAlgorithm.cs

@@ -0,0 +1,158 @@
+/*
+ * 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.Indicators;
+using QuantConnect.Interfaces;
+
+namespace QuantConnect.Algorithm.CSharp
+{
+    /// <summary>
+    /// Regression algorithm asserting consolidator registration ergonomics on a quote-only feed (forex):
+    /// trade bar consolidators and indicators are fed collapsed quote bars instead of being rejected,
+    /// <see cref="QCAlgorithm.RegisterIndicator{T}(Symbol,IndicatorBase{T},Func{DateTime,CalendarInfo},Func{IBaseData,T})"/>
+    /// accepts calendar periods, and consolidator periods smaller than the subscription period are
+    /// rejected at registration time instead of when the first data point arrives.
+    /// </summary>
+    public class ConsolidatorAutoAdaptationRegressionAlgorithm : QCAlgorithm, IRegressionAlgorithmDefinition
+    {
+        private TradeBarConsolidator _consolidator;
+        private OnBalanceVolume _obv;
+        private RelativeStrengthIndex _weeklyRsi;
+        private int _adaptedTradeBars;
+
+        /// <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(2014, 5, 5);
+            SetEndDate(2014, 5, 12);
+
+            var eurusd = AddForex("EURUSD", Resolution.Minute, Market.Oanda).Symbol;
+
+            // a trade bar consolidator on a quote-only feed: the engine feeds it quote bars collapsed
+            // into mid-point trade bars with zero volume
+            _consolidator = new TradeBarConsolidator(TimeSpan.FromHours(1));
+            _consolidator.DataConsolidated += (_, bar) =>
+            {
+                _adaptedTradeBars++;
+                if (bar.Volume != 0)
+                {
+                    throw new RegressionTestException("Expected trade bars collapsed from quote bars to have zero volume");
+                }
+            };
+            SubscriptionManager.AddConsolidator(eurusd, _consolidator);
+
+            // a trade bar indicator on a quote-only feed is fed collapsed trade bars as well
+            _obv = OBV(eurusd, Resolution.Hour);
+
+            // RegisterIndicator accepts calendar periods, like Consolidate does
+            _weeklyRsi = new RelativeStrengthIndex(2);
+            RegisterIndicator(eurusd, _weeklyRsi, Calendar.Weekly);
+
+            // a consolidator period smaller than the subscription period is rejected at registration time,
+            // instead of when the first data point arrives
+            try
+            {
+                SubscriptionManager.AddConsolidator(eurusd, new TradeBarConsolidator(TimeSpan.FromSeconds(10)));
+                throw new RegressionTestException($"Expected {nameof(ArgumentException)} for a consolidator period smaller than the subscription period");
+            }
+            catch (ArgumentException)
+            {
+                // expected, all the required information is available at registration time
+            }
+        }
+
+        public override void OnEndOfAlgorithm()
+        {
+            if (_adaptedTradeBars == 0)
+            {
+                throw new RegressionTestException("Expected the adapted trade bar consolidator to receive data");
+            }
+            if (_obv.Samples == 0)
+            {
+                throw new RegressionTestException("Expected the OnBalanceVolume indicator to be updated with collapsed trade bars");
+            }
+            if (_weeklyRsi.Samples == 0)
+            {
+                throw new RegressionTestException("Expected the weekly RSI to be updated when the week boundary was crossed");
+            }
+        }
+
+        /// <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 => 8661;
+
+        /// <summary>
+        /// Data Points count of the algorithm history
+        /// </summary>
+        public int AlgorithmHistoryDataPoints => 3;
+
+        /// <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.00"},
+            {"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", "-3.328"},
+            {"Tracking Error", "0.091"},
+            {"Treynor Ratio", "0"},
+            {"Total Fees", "$0.00"},
+            {"Estimated Strategy Capacity", "$0"},
+            {"Lowest Capacity Asset", ""},
+            {"Portfolio Turnover", "0%"},
+            {"Drawdown Recovery", "0"},
+            {"OrderListHash", "d41d8cd98f00b204e9800998ecf8427e"}
+        };
+    }
+}

Algorithm.Python/ConsolidatorAutoAdaptationRegressionAlgorithm.py

@@ -0,0 +1,65 @@
+# 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 consolidator registration ergonomics on a quote-only feed (forex):
+### trade bar consolidators and indicators are fed collapsed quote bars instead of being rejected,
+### register_indicator accepts calendar periods, and consolidator periods smaller than the subscription
+### period are rejected at registration time instead of when the first data point arrives.
+### </summary>
+class ConsolidatorAutoAdaptationRegressionAlgorithm(QCAlgorithm):
+    def initialize(self):
+        self.set_start_date(2014, 5, 5)
+        self.set_end_date(2014, 5, 12)
+
+        eurusd = self.add_forex("EURUSD", Resolution.MINUTE, Market.OANDA).symbol
+        self._adapted_trade_bars = 0
+
+        # a trade bar consolidator on a quote-only feed: the engine feeds it quote bars collapsed
+        # into mid-point trade bars with zero volume
+        self._consolidator = TradeBarConsolidator(timedelta(hours=1))
+        self._consolidator.data_consolidated += self._on_trade_bar
+        self.subscription_manager.add_consolidator(eurusd, self._consolidator)
+
+        # a trade bar indicator on a quote-only feed is fed collapsed trade bars as well
+        self._obv = self.obv(eurusd, Resolution.HOUR)
+
+        # register_indicator accepts calendar periods, like self.consolidate does
+        self._weekly_rsi = RelativeStrengthIndex(2)
+        self.register_indicator(eurusd, self._weekly_rsi, Calendar.WEEKLY)
+
+        # a consolidator period smaller than the subscription period is rejected at registration time,
+        # instead of when the first data point arrives
+        rejected = False
+        try:
+            self.subscription_manager.add_consolidator(eurusd, TradeBarConsolidator(timedelta(seconds=10)))
+        except:
+            # expected, all the required information is available at registration time
+            rejected = True
+        if not rejected:
+            raise AssertionError("Expected an error for a consolidator period smaller than the subscription period")
+
+    def _on_trade_bar(self, sender, bar):
+        self._adapted_trade_bars += 1
+        if bar.volume != 0:
+            raise AssertionError("Expected trade bars collapsed from quote bars to have zero volume")
+
+    def on_end_of_algorithm(self):
+        if self._adapted_trade_bars == 0:
+            raise AssertionError("Expected the adapted trade bar consolidator to receive data")
+        if self._obv.samples == 0:
+            raise AssertionError("Expected the OnBalanceVolume indicator to be updated with collapsed trade bars")
+        if self._weekly_rsi.samples == 0:
+            raise AssertionError("Expected the weekly RSI to be updated when the week boundary was crossed")

Algorithm.Python/QuantConnect.Algorithm.Python.csproj

@@ -53,6 +53,7 @@
     <Content Include="AccumulativeInsightPortfolioRegressionAlgorithm.py" />
     <Content Include="CustomDataTypeHistoryAlgorithm.py" />
     <Content Include="CorrectConsolidatedBarTypeForTickTypesAlgorithm.py" />
+    <Content Include="ConsolidatorAutoAdaptationRegressionAlgorithm.py" />
     <Content Include="IndustryStandardSecurityIdentifiersRegressionAlgorithm.py" />
     <Content Include="AddAlphaModelAlgorithm.py" />
     <Content Include="AddFutureOptionContractDataStreamingRegressionAlgorithm.py" />

Algorithm/QCAlgorithm.Indicators.cs

@@ -3160,6 +3160,14 @@ public string CreateIndicatorName(Symbol symbol, string type, Resolution? resolu
         /// <returns>The SubscriptionDataConfig for the specified symbol</returns>
         private SubscriptionDataConfig GetSubscription(Symbol symbol, TickType? tickType = null)
         {
+            if (symbol == null)
+            {
+                // fail with a clear message instead of a null reference below, commonly a Future.Mapped
+                // or similar property that is still null when the consolidator/indicator is registered
+                throw new ArgumentNullException(nameof(symbol), "Cannot resolve a subscription because the given symbol is null. " +
+                    "If the symbol comes from a property like Future.Mapped, note it can be null until the security receives data.");
+            }
+
             if (!TryGetSubscription(symbol, tickType, out var subscription))
             {
                 // The symbol was not manually subscribed to. Mirror the behavior of order submission
@@ -3231,6 +3239,21 @@ public void RegisterIndicator(Symbol symbol, IndicatorBase<IndicatorDataPoint> i
             RegisterIndicator(symbol, indicator, ResolveConsolidator(symbol, resolution), selector ?? (x => x.Value));
         }
 
+        /// <summary>
+        /// Creates and registers a new consolidator to receive automatic updates on the given calendar as well as configures
+        /// the indicator to receive updates from the consolidator.
+        /// </summary>
+        /// <param name="symbol">The symbol to register against</param>
+        /// <param name="indicator">The indicator to receive data from the consolidator</param>
+        /// <param name="calendar">The consolidation calendar, for example <see cref="Calendar.Weekly"/></param>
+        /// <param name="selector">Selects a value from the BaseData to send into the indicator, if null defaults to the Value property of BaseData (x => x.Value)</param>
+        [DocumentationAttribute(ConsolidatingData)]
+        [DocumentationAttribute(Indicators)]
+        public void RegisterIndicator(Symbol symbol, IndicatorBase<IndicatorDataPoint> indicator, Func<DateTime, CalendarInfo> calendar, Func<IBaseData, decimal> selector = null)
+        {
+            RegisterIndicator(symbol, indicator, ResolveConsolidator(symbol, calendar), selector ?? (x => x.Value));
+        }
+
         /// <summary>
         /// Registers the consolidator to receive automatic updates as well as configures the indicator to receive updates
         /// from the consolidator.
@@ -3303,6 +3326,22 @@ public void RegisterIndicator<T>(Symbol symbol, IndicatorBase<T> indicator, Time
             RegisterIndicator(symbol, indicator, ResolveConsolidator(symbol, resolution, typeof(T)), selector);
         }
 
+        /// <summary>
+        /// Creates and registers a new consolidator to receive automatic updates on the given calendar as well as configures
+        /// the indicator to receive updates from the consolidator.
+        /// </summary>
+        /// <param name="symbol">The symbol to register against</param>
+        /// <param name="indicator">The indicator to receive data from the consolidator</param>
+        /// <param name="calendar">The consolidation calendar, for example <see cref="Calendar.Weekly"/></param>
+        /// <param name="selector">Selects a value from the BaseData send into the indicator, if null defaults to a cast (x => (T)x)</param>
+        [DocumentationAttribute(ConsolidatingData)]
+        [DocumentationAttribute(Indicators)]
+        public void RegisterIndicator<T>(Symbol symbol, IndicatorBase<T> indicator, Func<DateTime, CalendarInfo> calendar, Func<IBaseData, T> selector = null)
+            where T : IBaseData
+        {
+            RegisterIndicator(symbol, indicator, ResolveConsolidator(symbol, calendar, typeof(T)), selector);
+        }
+
         /// <summary>
         /// Registers the consolidator to receive automatic updates as well as configures the indicator to receive updates
         /// from the consolidator.
@@ -3331,6 +3370,12 @@ public void RegisterIndicator<T>(Symbol symbol, IndicatorBase<T> indicator, IDat
                     // so we use a smarter selector as in other API methods
                     selectorToUse = consolidated => (T)(object)new IndicatorDataPoint(consolidated.Symbol, consolidated.EndTime, consolidated.Value);
                 }
+                else if (type == typeof(TradeBar) && consolidator.OutputType == typeof(QuoteBar) && selector == null)
+                {
+                    // trade bar indicator registered against a quote-only feed (forex, cfd): collapse each quote bar
+                    // into a mid-point trade bar with zero volume instead of failing
+                    selectorToUse = consolidated => (T)(object)((QuoteBar)consolidated).Collapse();
+                }
                 else
                 {
                     throw new ArgumentException($"Type mismatch found between consolidator and indicator for symbol: {symbol}." +
@@ -3694,6 +3739,21 @@ public IDataConsolidator ResolveConsolidator(Symbol symbol, TimeSpan? timeSpan,
             return CreateConsolidator(symbol, null, tickType, timeSpan, null, null);
         }
 
+        /// <summary>
+        /// Gets the default consolidator for the specified symbol and consolidation calendar
+        /// </summary>
+        /// <param name="symbol">The symbol whose data is to be consolidated</param>
+        /// <param name="calendar">The consolidation calendar, for example <see cref="Calendar.Weekly"/></param>
+        /// <param name="dataType">The data type for this consolidator, if null, uses TradeBar over QuoteBar if present</param>
+        /// <returns>The new default consolidator</returns>
+        [DocumentationAttribute(ConsolidatingData)]
+        [DocumentationAttribute(Indicators)]
+        public IDataConsolidator ResolveConsolidator(Symbol symbol, Func<DateTime, CalendarInfo> calendar, Type dataType = null)
+        {
+            var tickType = dataType != null ? LeanData.GetCommonTickTypeForCommonDataTypes(dataType, symbol.SecurityType) : (TickType?)null;
+            return CreateConsolidator(symbol, calendar, tickType, null, null, null);
+        }
+
         /// <summary>
         /// Creates a new consolidator for the specified period, generating the requested output type.
         /// </summary>

Algorithm/QCAlgorithm.Python.cs

@@ -621,7 +621,7 @@ public void RegisterIndicator(Symbol symbol, PyObject indicator, TimeSpan? resol
         /// </summary>
         /// <param name="symbol">The symbol to register against</param>
         /// <param name="indicator">The indicator to receive data from the consolidator</param>
-        /// <param name="pyObject">The python object that it is trying to register with, could be consolidator or a timespan</param>
+        /// <param name="pyObject">The python object that it is trying to register with, could be consolidator, timespan or a calendar</param>
         /// <param name="selector">Selects a value from the BaseData send into the indicator, if null defaults to a cast (x => (T)x)</param>
         [DocumentationAttribute(Indicators)]
         [DocumentationAttribute(ConsolidatingData)]
@@ -641,7 +641,7 @@ public void RegisterIndicator(Symbol symbol, PyObject indicator, PyObject pyObje
             }
             catch
             {
-                // Finally, since above didn't work, just try it as a timespan
+                // Since above didn't work, just try it as a timespan
                 // Issue #4668 Fix
                 using (Py.GIL())
                 {
@@ -657,7 +657,15 @@ public void RegisterIndicator(Symbol symbol, PyObject indicator, PyObject pyObje
                     }
                     catch (Exception e)
                     {
-                        throw new ArgumentException("Invalid third argument, should be either a valid consolidator or timedelta object. The following exception was thrown: ", e);
+                        // Finally, try it as a consolidation calendar, e.g. Calendar.WEEKLY, which Consolidate() also accepts
+                        if (pyObject.TrySafeAs(out Func<DateTime, CalendarInfo> calendar))
+                        {
+                            RegisterIndicator(symbol, indicator, ResolveConsolidator(symbol, calendar), selector);
+                            return;
+                        }
+
+                        throw new ArgumentException("Invalid third argument, should be either a valid consolidator, timedelta or " +
+                            "calendar (e.g. Calendar.WEEKLY) object. The following exception was thrown: ", e);
                     }
                 }
             }