/* * 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.Linq; using QuantConnect.Util; using QuantConnect.Data; using QuantConnect.Packets; using QuantConnect.Logging; using QuantConnect.Interfaces; using System.Collections.Generic; namespace QuantConnect.Lean.Engine.DataFeeds { /// /// This is an implementation of used to handle multiple live datafeeds /// public class DataQueueHandlerManager : IDataQueueHandler, IDataQueueUniverseProvider { private readonly Dictionary _dataConfigAndDataHandler = new(); /// /// Collection of data queue handles being used /// /// Protected for testing purposes protected List DataHandlers { get; } = new(); /// /// True if the composite queue handler has any instance /// public bool HasUniverseProvider => DataHandlers.OfType().Any(); /// /// Subscribe to the specified configuration /// /// defines the parameters to subscribe to a data feed /// handler to be fired on new data available /// The new enumerator for this subscription request public IEnumerator Subscribe(SubscriptionDataConfig dataConfig, EventHandler newDataAvailableHandler) { foreach (var dataHandler in DataHandlers) { var enumerator = dataHandler.Subscribe(dataConfig, newDataAvailableHandler); // Check if the enumerator is not empty if (enumerator != null) { _dataConfigAndDataHandler.Add(dataConfig, dataHandler); return enumerator; } } return null; } /// /// Removes the specified configuration /// /// Subscription config to be removed public virtual void Unsubscribe(SubscriptionDataConfig dataConfig) { if (_dataConfigAndDataHandler.Remove(dataConfig, out var dataHandler)) { dataHandler.Unsubscribe(dataConfig); } } /// /// Sets the job we're subscribing for /// /// Job we're subscribing for public void SetJob(LiveNodePacket job) { var dataHandlersConfig = job.DataQueueHandler; Log.Trace($"CompositeDataQueueHandler.SetJob(): will use {dataHandlersConfig}"); foreach (var dataHandlerName in dataHandlersConfig.DeserializeList()) { var dataHandler = Composer.Instance.GetExportedValueByTypeName(dataHandlerName); dataHandler.SetJob(job); DataHandlers.Add(dataHandler); } } /// /// Returns whether the data provider is connected /// /// true if the data provider is connected public bool IsConnected => true; /// /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources. /// public void Dispose() { foreach (var dataHandler in DataHandlers) { dataHandler.Dispose(); } } /// /// Method returns a collection of Symbols that are available at the data source. /// /// Symbol to lookup /// Include expired contracts /// Expected security currency(if any) /// Enumerable of Symbols, that are associated with the provided Symbol public IEnumerable LookupSymbols(Symbol symbol, bool includeExpired, string securityCurrency = null) { foreach (var dataHandler in GetUniverseProviders()) { var result = dataHandler.LookupSymbols(symbol, includeExpired, securityCurrency).ToList(); if (result.Any()) { return result; } } return Enumerable.Empty(); } /// /// Returns whether selection can take place or not. /// /// This is useful to avoid a selection taking place during invalid times, for example IB reset times or when not connected, /// because if allowed selection would fail since IB isn't running and would kill the algorithm /// True if selection can take place public bool CanPerformSelection() { return GetUniverseProviders().Any(provider => provider.CanPerformSelection()); } private IEnumerable GetUniverseProviders() { var yielded = false; foreach (var universeProvider in DataHandlers.OfType()) { yielded = true; yield return universeProvider; } if (!yielded) { throw new NotSupportedException("The DataQueueHandler does not support Options and Futures."); } } } }