/*
* 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.Concurrent;
using System.Collections.Generic;
using System.Collections.Specialized;
using System.Linq;
using QuantConnect.Data;
using QuantConnect.Data.Auxiliary;
using QuantConnect.Data.UniverseSelection;
using QuantConnect.Interfaces;
using QuantConnect.Logging;
using QuantConnect.Securities;
using QuantConnect.Util;
namespace QuantConnect.Lean.Engine.DataFeeds
{
///
/// DataManager will manage the subscriptions for both the DataFeeds and the SubscriptionManager
///
public class DataManager : IAlgorithmSubscriptionManager, IDataFeedSubscriptionManager, IDataManager
{
private readonly IAlgorithmSettings _algorithmSettings;
private readonly IDataFeed _dataFeed;
private readonly MarketHoursDatabase _marketHoursDatabase;
private readonly ITimeKeeper _timeKeeper;
private readonly bool _liveMode;
/// There is no ConcurrentHashSet collection in .NET,
/// so we use ConcurrentDictionary with byte value to minimize memory usage
private readonly ConcurrentDictionary _subscriptionManagerSubscriptions
= new ConcurrentDictionary();
///
/// Event fired when a new subscription is added
///
public event EventHandler SubscriptionAdded;
///
/// Event fired when an existing subscription is removed
///
public event EventHandler SubscriptionRemoved;
///
/// Creates a new instance of the DataManager
///
public DataManager(
IDataFeed dataFeed,
UniverseSelection universeSelection,
IAlgorithm algorithm,
ITimeKeeper timeKeeper,
MarketHoursDatabase marketHoursDatabase,
bool liveMode)
{
_dataFeed = dataFeed;
UniverseSelection = universeSelection;
UniverseSelection.SetDataManager(this);
_algorithmSettings = algorithm.Settings;
AvailableDataTypes = SubscriptionManager.DefaultDataTypes();
_timeKeeper = timeKeeper;
_marketHoursDatabase = marketHoursDatabase;
_liveMode = liveMode;
// wire ourselves up to receive notifications when universes are added/removed
algorithm.UniverseManager.CollectionChanged += (sender, args) =>
{
switch (args.Action)
{
case NotifyCollectionChangedAction.Add:
foreach (var universe in args.NewItems.OfType())
{
var config = universe.Configuration;
var start = algorithm.UtcTime;
var end = algorithm.LiveMode ? Time.EndOfTime
: algorithm.EndDate.ConvertToUtc(algorithm.TimeZone);
Security security;
if (!algorithm.Securities.TryGetValue(config.Symbol, out security))
{
// create a canonical security object if it doesn't exist
security = new Security(
_marketHoursDatabase.GetExchangeHours(config),
config,
algorithm.Portfolio.CashBook[algorithm.AccountCurrency],
SymbolProperties.GetDefault(algorithm.AccountCurrency),
algorithm.Portfolio.CashBook
);
}
AddSubscription(
new SubscriptionRequest(true,
universe,
security,
config,
start,
end));
}
break;
case NotifyCollectionChangedAction.Remove:
foreach (var universe in args.OldItems.OfType())
{
// removing the subscription will be handled by the SubscriptionSynchronizer
// in the next loop as well as executing a UniverseSelection one last time.
if (!universe.DisposeRequested)
{
universe.Dispose();
}
}
break;
default:
throw new NotImplementedException("The specified action is not implemented: " + args.Action);
}
};
}
#region IDataFeedSubscriptionManager
///
/// Gets the data feed subscription collection
///
public SubscriptionCollection DataFeedSubscriptions { get; } = new SubscriptionCollection();
///
/// Will remove all current
///
public void RemoveAllSubscriptions()
{
// remove each subscription from our collection
foreach (var subscription in DataFeedSubscriptions)
{
try
{
RemoveSubscription(subscription.Configuration);
}
catch (Exception err)
{
Log.Error(err, "DataManager.RemoveAllSubscriptions():" +
$"Error removing: {subscription.Configuration}");
}
}
}
///
/// Adds a new to provide data for the specified security.
///
/// Defines the to be added
/// True if the subscription was created and added successfully, false otherwise
public bool AddSubscription(SubscriptionRequest request)
{
Subscription subscription;
if (DataFeedSubscriptions.TryGetValue(request.Configuration, out subscription))
{
// duplicate subscription request
return subscription.AddSubscriptionRequest(request);
}
subscription = _dataFeed.CreateSubscription(request);
if (subscription == null)
{
Log.Trace($"DataManager.AddSubscription(): Unable to add subscription for: {request.Configuration}");
// subscription will be null when there's no tradeable dates for the security between the requested times, so
// don't even try to load the data
return false;
}
if (_liveMode)
{
OnSubscriptionAdded(subscription);
Log.Trace($"DataManager.AddSubscription(): Added {request.Configuration}." +
$" Start: {request.StartTimeUtc}. End: {request.EndTimeUtc}");
}
else if(Log.DebuggingEnabled)
{
// for performance lets not create the message string if debugging is not enabled
// this can be executed many times and its in the algorithm thread
Log.Debug($"DataManager.AddSubscription(): Added {request.Configuration}." +
$" Start: {request.StartTimeUtc}. End: {request.EndTimeUtc}");
}
return DataFeedSubscriptions.TryAdd(subscription);
}
///
/// Removes the , if it exists
///
/// The of the subscription to remove
/// Universe requesting to remove .
/// Default value, null, will remove all universes
/// True if the subscription was successfully removed, false otherwise
public bool RemoveSubscription(SubscriptionDataConfig configuration, Universe universe = null)
{
// remove the subscription from our collection, if it exists
Subscription subscription;
if (DataFeedSubscriptions.TryGetValue(configuration, out subscription))
{
// we remove the subscription when there are no other requests left
if (subscription.RemoveSubscriptionRequest(universe))
{
if (!DataFeedSubscriptions.TryRemove(configuration, out subscription))
{
Log.Error($"DataManager.RemoveSubscription(): Unable to remove {configuration}");
return false;
}
_dataFeed.RemoveSubscription(subscription);
if (_liveMode)
{
OnSubscriptionRemoved(subscription);
}
subscription.Dispose();
RemoveSubscriptionDataConfig(subscription);
if (_liveMode)
{
Log.Trace($"DataManager.RemoveSubscription(): Removed {configuration}");
}
else if(Log.DebuggingEnabled)
{
// for performance lets not create the message string if debugging is not enabled
// this can be executed many times and its in the algorithm thread
Log.Debug($"DataManager.RemoveSubscription(): Removed {configuration}");
}
return true;
}
}
return false;
}
///
/// Event invocator for the event
///
/// The added subscription
private void OnSubscriptionAdded(Subscription subscription)
{
SubscriptionAdded?.Invoke(this, subscription);
}
///
/// Event invocator for the event
///
/// The removed subscription
private void OnSubscriptionRemoved(Subscription subscription)
{
SubscriptionRemoved?.Invoke(this, subscription);
}
#endregion
#region IAlgorithmSubscriptionManager
///
/// Flags the existence of custom data in the subscriptions
///
public bool HasCustomData { get; set; }
///
/// Gets all the current data config subscriptions that are being processed for the SubscriptionManager
///
public IEnumerable SubscriptionManagerSubscriptions =>
_subscriptionManagerSubscriptions.Select(x => x.Key);
///
/// Gets existing or adds new
///
/// Returns the SubscriptionDataConfig instance used
public SubscriptionDataConfig SubscriptionManagerGetOrAdd(SubscriptionDataConfig newConfig)
{
var config = _subscriptionManagerSubscriptions.GetOrAdd(newConfig, newConfig);
// if the reference is not the same, means it was already there and we did not add anything new
if (!ReferenceEquals(config, newConfig) && Log.DebuggingEnabled)
{
// for performance lets not create the message string if debugging is not enabled
// this can be executed many times and its in the algorithm thread
Log.Debug("DataManager.SubscriptionManagerGetOrAdd(): subscription already added: " + config);
}
else
{
// for performance, only count if we are above the limit
if (SubscriptionManagerCount() > _algorithmSettings.DataSubscriptionLimit)
{
// count data subscriptions by symbol, ignoring multiple data types.
// this limit was added due to the limits IB places on number of subscriptions
var uniqueCount = SubscriptionManagerSubscriptions
.Where(x => !x.Symbol.IsCanonical())
.DistinctBy(x => x.Symbol.Value)
.Count();
if (uniqueCount > _algorithmSettings.DataSubscriptionLimit)
{
throw new Exception(
$"The maximum number of concurrent market data subscriptions was exceeded ({_algorithmSettings.DataSubscriptionLimit})." +
"Please reduce the number of symbols requested or increase the limit using Settings.DataSubscriptionLimit.");
}
}
// add the time zone to our time keeper
_timeKeeper.AddTimeZone(newConfig.ExchangeTimeZone);
// if is custom data, sets HasCustomData to true
HasCustomData = HasCustomData || newConfig.IsCustomData;
}
return config;
}
///
/// Will try to remove a and update the corresponding
/// consumers accordingly
///
/// The owning the configuration to remove
private void RemoveSubscriptionDataConfig(Subscription subscription)
{
SubscriptionDataConfig config;
if (subscription.RemovedFromUniverse.Value
&& _subscriptionManagerSubscriptions.TryRemove(subscription.Configuration, out config))
{
if (HasCustomData && config.IsCustomData)
{
HasCustomData = _subscriptionManagerSubscriptions.Any(x => x.Key.IsCustomData);
}
}
}
///
/// Returns the amount of data config subscriptions processed for the SubscriptionManager
///
public int SubscriptionManagerCount()
{
return _subscriptionManagerSubscriptions.Skip(0).Count();
}
#region ISubscriptionDataConfigService
///
/// The different each supports
///
public Dictionary> AvailableDataTypes { get; }
///
/// Creates and adds a list of for a given symbol and configuration.
/// Can optionally pass in desired subscription data type to use.
/// If the config already existed will return existing instance instead
///
public SubscriptionDataConfig Add(
Type dataType,
Symbol symbol,
Resolution resolution,
bool fillForward = true,
bool extendedMarketHours = false,
bool isFilteredSubscription = true,
bool isInternalFeed = false,
bool isCustomData = false,
DataNormalizationMode dataNormalizationMode = DataNormalizationMode.Adjusted
)
{
return Add(symbol, resolution, fillForward, extendedMarketHours, isFilteredSubscription, isInternalFeed, isCustomData,
new List> { new Tuple(dataType, LeanData.GetCommonTickTypeForCommonDataTypes(dataType, symbol.SecurityType))}, dataNormalizationMode)
.First();
}
///
/// Creates and adds a list of for a given symbol and configuration.
/// Can optionally pass in desired subscription data types to use.
/// If the config already existed will return existing instance instead
///
public List Add(
Symbol symbol,
Resolution resolution,
bool fillForward,
bool extendedMarketHours,
bool isFilteredSubscription = true,
bool isInternalFeed = false,
bool isCustomData = false,
List> subscriptionDataTypes = null,
DataNormalizationMode dataNormalizationMode = DataNormalizationMode.Adjusted
)
{
var dataTypes = subscriptionDataTypes ??
LookupSubscriptionConfigDataTypes(symbol.SecurityType, resolution, symbol.IsCanonical());
var marketHoursDbEntry = _marketHoursDatabase.GetEntry(symbol.ID.Market, symbol, symbol.ID.SecurityType);
var exchangeHours = marketHoursDbEntry.ExchangeHours;
if (symbol.ID.SecurityType == SecurityType.Option || symbol.ID.SecurityType == SecurityType.Future)
{
dataNormalizationMode = DataNormalizationMode.Raw;
}
if (marketHoursDbEntry.DataTimeZone == null)
{
throw new ArgumentNullException(nameof(marketHoursDbEntry.DataTimeZone),
"DataTimeZone is a required parameter for new subscriptions. Set to the time zone the raw data is time stamped in.");
}
if (exchangeHours.TimeZone == null)
{
throw new ArgumentNullException(nameof(exchangeHours.TimeZone),
"ExchangeTimeZone is a required parameter for new subscriptions. Set to the time zone the security exchange resides in.");
}
if (!dataTypes.Any())
{
throw new ArgumentNullException(nameof(dataTypes), "At least one type needed to create new subscriptions");
}
var result = (from subscriptionDataType in dataTypes
let dataType = subscriptionDataType.Item1
let tickType = subscriptionDataType.Item2
select new SubscriptionDataConfig(
dataType,
symbol,
resolution,
marketHoursDbEntry.DataTimeZone,
exchangeHours.TimeZone,
fillForward,
extendedMarketHours,
isInternalFeed,
isCustomData,
isFilteredSubscription: isFilteredSubscription,
tickType: tickType,
dataNormalizationMode: dataNormalizationMode)).ToList();
for (int i = 0; i < result.Count; i++)
{
result[i] = SubscriptionManagerGetOrAdd(result[i]);
}
return result;
}
///
/// Get the data feed types for a given
///
/// The used to determine the types
/// The resolution of the data requested
/// Indicates whether the security is Canonical (future and options)
/// Types that should be added to the
public List> LookupSubscriptionConfigDataTypes(
SecurityType symbolSecurityType,
Resolution resolution,
bool isCanonical
)
{
if (isCanonical)
{
return new List> { new Tuple(typeof(ZipEntryName), TickType.Quote) };
}
return AvailableDataTypes[symbolSecurityType]
.Select(tickType => new Tuple(LeanData.GetDataType(resolution, tickType), tickType)).ToList();
}
///
/// Gets a list of all registered for a given
///
public List GetSubscriptionDataConfigs(Symbol symbol)
{
return SubscriptionManagerSubscriptions.Where(x => x.Symbol == symbol).ToList();
}
#endregion
#endregion
#region IDataManager
///
/// Get the universe selection instance
///
public UniverseSelection UniverseSelection { get; }
#endregion
}
}