Feature: Generic brokerage downloader wrapper (#8235)

* feat: new constructor of AlgorithmNodePacket

* refactor: extract JobQueue configuration

* remove: not used `using` in IDataDownloader

* feat: create BrokerageDataDownloader

* Revert "refactor: extract JobQueue configuration"

This reverts commit 5778936.

* Revert "feat: new constructor of AlgorithmNodePacket"

This reverts commit d7a565f.

* feat: new config `data-download-brokerage` in DataDownloadProvider

* refactor: initialize in BrokerageDataDownloader

* remove: not used `using` in Program's DataDownloadProvider

* remove: not used ref on QuantConnect.Queue proj

* refactor: use default market based on SecurityType

* refactor: MarketName in DataDownloadConfig struct
test:feat: validate MarketName

* feat: support Canonical Symbols in BrokerageDataDownloader

* remove: not used command arguments

* feat: init CacheProvider of IOptionChainProvider in Downloader

* feat: add brokerage message event in BrokerageDataDownloader

Common/IDataDownloader.cs

@@ -13,7 +13,6 @@
  * limitations under the License.
 */
 
-using System;
 using QuantConnect.Data;
 using System.Collections.Generic;
 

DownloaderDataProvider/DataDownloadConfig.cs

@@ -15,6 +15,8 @@
 */
 
 using System.Globalization;
+using QuantConnect.Logging;
+using QuantConnect.Brokerages;
 using QuantConnect.Configuration;
 using QuantConnect.DownloaderDataProvider.Launcher.Models.Constants;
 
@@ -74,8 +76,15 @@ public DataDownloadConfig()
             EndDate = DateTime.ParseExact(Config.Get(DownloaderCommandArguments.CommandEndDate).ToString(), DateFormat.EightCharacter, CultureInfo.InvariantCulture);
 
 #pragma warning disable CA1308 // class Market keeps all name in lowercase
-            MarketName = Config.Get(DownloaderCommandArguments.CommandMarketName).ToString()?.ToLower(CultureInfo.InvariantCulture) ?? Market.USA;
+            MarketName = Config.Get(DownloaderCommandArguments.CommandMarketName).ToString().ToLower(CultureInfo.InvariantCulture);
 #pragma warning restore CA1308
+
+            if (string.IsNullOrEmpty(MarketName))
+            {
+                MarketName = DefaultBrokerageModel.DefaultMarketMap[SecurityType];
+                Log.Trace($"{nameof(DataDownloadConfig)}: Default market '{MarketName}' applied for SecurityType '{SecurityType}'");
+            }
+
             if (!Market.SupportedMarkets().Contains(MarketName))
             {
                 throw new ArgumentException($"The specified market '{MarketName}' is not supported. Supported markets are: {string.Join(", ", Market.SupportedMarkets())}.");

DownloaderDataProvider/Models/BrokerageDataDownloader.cs

@@ -0,0 +1,134 @@
+/*
+ * 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.Util;
+using QuantConnect.Data;
+using QuantConnect.Packets;
+using QuantConnect.Interfaces;
+using QuantConnect.Securities;
+using QuantConnect.Configuration;
+
+namespace QuantConnect.DownloaderDataProvider.Launcher.Models
+{
+    /// <summary>
+    /// Class for downloading data from a brokerage.
+    /// </summary>
+    public class BrokerageDataDownloader : IDataDownloader
+    {
+        /// <summary>
+        /// Represents the Brokerage implementation.
+        /// </summary>
+        private IBrokerage _brokerage;
+
+        /// <summary>
+        /// Provides access to exchange hours and raw data times zones in various markets
+        /// </summary>
+        private readonly MarketHoursDatabase _marketHoursDatabase = MarketHoursDatabase.FromDataFolder();
+
+        /// <summary>
+        /// Initializes a new instance of the <see cref="BrokerageDataDownloader"/> class.
+        /// </summary>
+        public BrokerageDataDownloader()
+        {
+            var liveNodeConfiguration = new LiveNodePacket() { Brokerage = Config.Get("data-downloader-brokerage") };
+
+            try
+            {
+                // import the brokerage data for the configured brokerage
+                var brokerageFactory = Composer.Instance.Single<IBrokerageFactory>(factory => factory.BrokerageType.MatchesTypeName(liveNodeConfiguration.Brokerage));
+                liveNodeConfiguration.BrokerageData = brokerageFactory.BrokerageData;
+            }
+            catch (InvalidOperationException error)
+            {
+                throw new InvalidOperationException($"{nameof(BrokerageDataDownloader)}.An error occurred while resolving brokerage data for a live job. Brokerage: {liveNodeConfiguration.Brokerage}.", error);
+            }
+
+            _brokerage = Composer.Instance.GetExportedValueByTypeName<IBrokerage>(liveNodeConfiguration.Brokerage);
+
+            _brokerage.Message += (object _, Brokerages.BrokerageMessageEvent e) =>
+            {
+                if (e.Type == Brokerages.BrokerageMessageType.Error)
+                {
+                    Logging.Log.Error(e.Message);
+                }
+                else
+                {
+                    Logging.Log.Trace(e.Message);
+                }
+            };
+
+            ((IDataQueueHandler)_brokerage).SetJob(liveNodeConfiguration);
+        }
+
+        /// <summary>
+        /// Get historical data enumerable for a single symbol, type and resolution given this start and end time (in UTC).
+        /// </summary>
+        /// <param name="dataDownloaderGetParameters">model class for passing in parameters for historical data</param>
+        /// <returns>Enumerable of base data for this symbol</returns>
+        public IEnumerable<BaseData>? Get(DataDownloaderGetParameters dataDownloaderGetParameters)
+        {
+            var symbol = dataDownloaderGetParameters.Symbol;
+            var resolution = dataDownloaderGetParameters.Resolution;
+            var startUtc = dataDownloaderGetParameters.StartUtc;
+            var endUtc = dataDownloaderGetParameters.EndUtc;
+            var tickType = dataDownloaderGetParameters.TickType;
+
+            var dataType = LeanData.GetDataType(resolution, tickType);
+            var exchangeHours = _marketHoursDatabase.GetExchangeHours(symbol.ID.Market, symbol, symbol.SecurityType);
+            var dataTimeZone = _marketHoursDatabase.GetDataTimeZone(symbol.ID.Market, symbol, symbol.SecurityType);
+
+            var symbols = new List<Symbol> { symbol };
+            if (symbol.IsCanonical())
+            {
+                symbols = GetChainSymbols(symbol, true).ToList();
+            }
+
+            return symbols
+                .Select(symbol =>
+                {
+                    var request = new Data.HistoryRequest(startUtc, endUtc, dataType, symbol, resolution, exchangeHours: exchangeHours, dataTimeZone: dataTimeZone, resolution,
+                        includeExtendedMarketHours: true, false, DataNormalizationMode.Raw, tickType);
+
+                    var history = _brokerage.GetHistory(request);
+
+                    if (history == null)
+                    {
+                        Logging.Log.Trace($"{nameof(BrokerageDataDownloader)}.{nameof(Get)}: Ignoring history request for unsupported symbol {symbol}");
+                    }
+
+                    return history;
+                })
+                .Where(history => history != null)
+                .SelectMany(history => history);
+        }
+
+        /// <summary>
+        /// Returns an IEnumerable of Future/Option contract symbols for the given root ticker
+        /// </summary>
+        /// <param name="symbol">The Symbol to get futures/options chain for</param>
+        /// <param name="includeExpired">Include expired contracts</param>
+        private IEnumerable<Symbol> GetChainSymbols(Symbol symbol, bool includeExpired)
+        {
+            if (_brokerage is IDataQueueUniverseProvider universeProvider)
+            {
+                return universeProvider.LookupSymbols(symbol, includeExpired);
+            }
+            else
+            {
+                throw new InvalidOperationException($"{nameof(BrokerageDataDownloader)}.{nameof(GetChainSymbols)}: The current brokerage does not support fetching canonical symbols. Please ensure your brokerage instance supports this feature.");
+            }
+        }
+    }
+}

DownloaderDataProvider/Program.cs

@@ -15,13 +15,13 @@
 */
 
 using NodaTime;
-using System.Timers;
 using QuantConnect.Util;
 using QuantConnect.Data;
 using QuantConnect.Logging;
 using QuantConnect.Interfaces;
 using QuantConnect.Securities;
 using QuantConnect.Configuration;
+using QuantConnect.Lean.Engine.DataFeeds;
 using DataFeeds = QuantConnect.Lean.Engine.DataFeeds;
 using QuantConnect.DownloaderDataProvider.Launcher.Models.Constants;
 
@@ -181,6 +181,13 @@ public static void InitializeConfigurations()
         var mapFileProvider = Composer.Instance.GetExportedValueByTypeName<IMapFileProvider>(Config.Get("map-file-provider", "LocalDiskMapFileProvider"));
         var factorFileProvider = Composer.Instance.GetExportedValueByTypeName<IFactorFileProvider>(Config.Get("factor-file-provider", "LocalDiskFactorFileProvider"));
 
+        var optionChainProvider = Composer.Instance.GetPart<IOptionChainProvider>();
+        if (optionChainProvider == null)
+        {
+            optionChainProvider = new CachingOptionChainProvider(new LiveOptionChainProvider(new ZipDataCacheProvider(dataProvider, false), mapFileProvider));
+            Composer.Instance.AddPart(optionChainProvider);
+        }
+
         mapFileProvider.Initialize(dataProvider);
         factorFileProvider.Initialize(mapFileProvider, dataProvider);
     }

DownloaderDataProvider/config.example.json

@@ -13,5 +13,8 @@
   "job-organization-id": "",
 
   // Data downloader provider
-  "data-downloader": ""
+  "data-downloader": "",
+
+  // Specifies the name of the brokerage service used for data downloading (optional).
+  "data-downloader-brokerage": ""
 }

Tests/DownloaderDataProvider/DataDownloadConfigTests.cs

@@ -0,0 +1,57 @@
+/*
+ * 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 NUnit.Framework;
+using QuantConnect.Configuration;
+using QuantConnect.DownloaderDataProvider.Launcher;
+
+namespace QuantConnect.Tests.DownloaderDataProvider
+{
+    [TestFixture]
+    public class DataDownloadConfigTests
+    {
+        [TestCase(null, "BTCUSDT", SecurityType.Crypto, "coinbase", false)]
+        [TestCase(null, "BTCUSDT", SecurityType.Crypto, "coinbase", true)]
+        [TestCase("", "ETHUSDT", SecurityType.Crypto, "coinbase", false)]
+        [TestCase("", "ETHUSDT", SecurityType.Crypto, "coinbase", true)]
+        [TestCase(null, "AAPL", SecurityType.Equity, "usa", false)]
+        [TestCase(null, "AAPL", SecurityType.Equity, "usa", true)]
+        [TestCase("", "AAPL", SecurityType.Equity, "usa", false)]
+        [TestCase("", "AAPL", SecurityType.Equity, "usa", true)]
+        [TestCase("USA", "AAPL", SecurityType.Equity, "usa")]
+        [TestCase("ICE", "AAPL", SecurityType.Equity, "ice")]
+        public void ValidateMarketArguments(string market, string ticker, SecurityType securityType, string expectedMarket, bool skipConfigMarket = false)
+        {
+            Config.Set("data-type", "Trade");
+            Config.Set("resolution", "Daily");
+            Config.Set("security-type", $"{securityType}");
+            Config.Set("tickers", $"{{\"{ticker}\": \"\"}}");
+            Config.Set("start-date", "20240101");
+            Config.Set("end-date", "20240202");
+
+            if (!skipConfigMarket)
+            {
+                Config.Set("market", market);
+            }
+
+            var dataDownloadConfig = new DataDownloadConfig();
+
+            Assert.That(dataDownloadConfig.MarketName, Is.EqualTo(expectedMarket));
+
+            Config.Reset();
+        }
+    }
+}