/* * 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; private readonly IRegisteredSecurityDataTypesProvider _registeredTypesProvider; private readonly IDataPermissionManager _dataPermissionManager; /// 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, IRegisteredSecurityDataTypesProvider registeredTypesProvider, IDataPermissionManager dataPermissionManager) { _dataFeed = dataFeed; UniverseSelection = universeSelection; UniverseSelection.SetDataManager(this); _algorithmSettings = algorithm.Settings; AvailableDataTypes = SubscriptionManager.DefaultDataTypes(); _timeKeeper = timeKeeper; _marketHoursDatabase = marketHoursDatabase; _liveMode = liveMode; _registeredTypesProvider = registeredTypesProvider; _dataPermissionManager = dataPermissionManager; // 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, RegisteredSecurityDataTypesProvider.Null, new SecurityCache() ); } 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) { // guarantee the configuration is present in our config collection // this is related to GH issue 3877: where we added a configuration which we also removed _subscriptionManagerSubscriptions.TryAdd(request.Configuration, request.Configuration); Subscription subscription; if (DataFeedSubscriptions.TryGetValue(request.Configuration, out subscription)) { // duplicate subscription request subscription.AddSubscriptionRequest(request); // only result true if the existing subscription is internal, we actually added something from the users perspective return subscription.Configuration.IsInternalFeed; } // before adding the configuration to the data feed let's assert it's valid _dataPermissionManager.AssertConfiguration(request.Configuration); 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; } } else if (universe != null) { // a universe requested removal of a subscription which wasn't present anymore, this can happen when a subscription ends // it will get removed from the data feed subscription list, but the configuration will remain until the universe removes it // why? the effect I found is that the fill models are using these subscriptions to determine which data they could use SubscriptionDataConfig config; _subscriptionManagerSubscriptions.TryRemove(configuration, out config); } 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 /// /// 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)) { // 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 if (Log.DebuggingEnabled) { 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); } return config; } /// /// Will try to remove a and update the corresponding /// consumers accordingly /// /// The owning the configuration to remove private void RemoveSubscriptionDataConfig(Subscription subscription) { // the subscription could of ended but might still be part of the universe if (subscription.RemovedFromUniverse.Value) { SubscriptionDataConfig config; _subscriptionManagerSubscriptions.TryRemove(subscription.Configuration, out config); } } /// /// 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 = null, 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 = null, bool fillForward = true, bool extendedMarketHours = false, bool isFilteredSubscription = true, bool isInternalFeed = false, bool isCustomData = false, List> subscriptionDataTypes = null, DataNormalizationMode dataNormalizationMode = DataNormalizationMode.Adjusted ) { var dataTypes = subscriptionDataTypes ?? LookupSubscriptionConfigDataTypes(symbol.SecurityType, resolution ?? Resolution.Minute, symbol.IsCanonical()); if (!dataTypes.Any()) { throw new ArgumentNullException(nameof(dataTypes), "At least one type needed to create new subscriptions"); } var resolutionWasProvided = resolution.HasValue; foreach (var typeTuple in dataTypes) { var baseInstance = typeTuple.Item1.GetBaseDataInstance(); baseInstance.Symbol = symbol; if (!resolutionWasProvided) { var defaultResolution = baseInstance.DefaultResolution(); if (resolution.HasValue && resolution != defaultResolution) { // we are here because there are multiple 'dataTypes'. // if we get different default resolutions lets throw, this shouldn't happen throw new InvalidOperationException( $"Different data types ({string.Join(",", dataTypes.Select(tuple => tuple.Item1))})" + $" provided different default resolutions {defaultResolution} and {resolution}, this is an unexpected invalid operation."); } resolution = defaultResolution; } else { // only assert resolution in backtesting, live can use other data source // for example daily data for options if (!_liveMode) { var supportedResolutions = baseInstance.SupportedResolutions(); if (supportedResolutions.Contains(resolution.Value)) { continue; } throw new ArgumentException($"Sorry {resolution.ToStringInvariant()} is not a supported resolution for {typeTuple.Item1.Name}" + $" and SecurityType.{symbol.SecurityType.ToStringInvariant()}." + $" Please change your AddData to use one of the supported resolutions ({string.Join(",", supportedResolutions)})."); } } } MarketHoursDatabase.Entry marketHoursDbEntry; if (!_marketHoursDatabase.TryGetEntry(symbol.ID.Market, symbol, symbol.ID.SecurityType, out marketHoursDbEntry)) { if (symbol.SecurityType == SecurityType.Base) { var baseInstance = dataTypes.Single().Item1.GetBaseDataInstance(); baseInstance.Symbol = symbol; _marketHoursDatabase.SetEntryAlwaysOpen(symbol.ID.Market, null, SecurityType.Base, baseInstance.DataTimeZone()); } marketHoursDbEntry = _marketHoursDatabase.GetEntry(symbol.ID.Market, symbol, symbol.ID.SecurityType); } var exchangeHours = marketHoursDbEntry.ExchangeHours; if (symbol.ID.SecurityType == SecurityType.Option || symbol.ID.SecurityType == SecurityType.FutureOption || 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."); } var result = (from subscriptionDataType in dataTypes let dataType = subscriptionDataType.Item1 let tickType = subscriptionDataType.Item2 select new SubscriptionDataConfig( dataType, symbol, resolution.Value, 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]); // track all registered data types _registeredTypesProvider.RegisterType(result[i].Type); } 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) }; } IEnumerable availableDataType = AvailableDataTypes[symbolSecurityType]; // Equities will only look for trades in case of low resolutions. if (symbolSecurityType == SecurityType.Equity && (resolution == Resolution.Daily || resolution == Resolution.Hour)) { // we filter out quote tick type availableDataType = availableDataType.Where(t => t != TickType.Quote); } return availableDataType .Select(tickType => new Tuple(LeanData.GetDataType(resolution, tickType), tickType)).ToList(); } /// /// Gets a list of all registered for a given /// /// Will not return internal subscriptions by default public List GetSubscriptionDataConfigs(Symbol symbol, bool includeInternalConfigs = false) { return SubscriptionManagerSubscriptions.Where(x => x.Symbol == symbol && (includeInternalConfigs || !x.IsInternalFeed)).ToList(); } #endregion #endregion #region IDataManager /// /// Get the universe selection instance /// public UniverseSelection UniverseSelection { get; } #endregion } }