/*
* 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.");
}
}
}
}