/* * 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.Linq; using QuantConnect.Logging; using QuantConnect.Util; using static QuantConnect.StringExtensions; namespace QuantConnect.Algorithm.Framework.Alphas.Analysis { /// /// Encapsulates the storage and on-line scoring of insights. /// /// /// This type assumes a forward progression of time, and as such, methods invoked will /// return data that is current as of the last update time. /// This type is designed to be invoked from two separate threads. That does not mean this type /// is thread-safe in the general sense, but given a particular invocation pattern. The goal is /// to allow the algorithm thread to continue while scoring of the insights happens on a separate /// thread. Insights are added on the algorithm thread via AddInsights and scoring updates are pushed /// on the insight thrad via UpdateScores. This means that the various collections (open, closed, updated) /// are potentially at different frontiers. In fact, it is the common case where the openInsightContexts /// collection is ahead of everything else. /// public class InsightManager : IInsightManager, IDisposable { /// /// Gets all insight score types /// public static readonly IReadOnlyCollection ScoreTypes = Enum.GetValues(typeof(InsightScoreType)).Cast().ToArray(); private readonly double _extraAnalysisPeriodRatio; private readonly List _extensions; private readonly IInsightScoreFunctionProvider _scoreFunctionProvider; private readonly object _lock; private readonly HashSet _updatedInsightContexts; private readonly HashSet _openInsightContexts; private readonly ConcurrentDictionary _closedInsightContexts; /// /// Enumerable of insights still under analysis /// public IEnumerable OpenInsights { get { lock (_lock) { return _openInsightContexts.Select(context => context.Insight).ToList(); } } } /// /// Enumerable of insights who's analysis has been completed /// public IEnumerable ClosedInsights => _closedInsightContexts.Select(kvp => kvp.Value.Insight); /// /// Enumerable of all internally maintained insights /// public IEnumerable AllInsights => OpenInsights.Concat(ClosedInsights); /// /// Gets the unique set of symbols from analysis contexts that will /// /// /// public IEnumerable ContextsOpenAt(DateTime frontierTimeUtc) { lock (_lock) { return _openInsightContexts.Where(context => context.AnalysisEndTimeUtc <= frontierTimeUtc).ToList(); } } /// /// Initializes a new instance of the class /// /// Provides scoring functions by insight type/score type /// Ratio of the insight period to keep the analysis open /// Extensions used to perform tasks at certain events public InsightManager(IInsightScoreFunctionProvider scoreFunctionProvider, double extraAnalysisPeriodRatio, params IInsightManagerExtension[] extensions) { if (extraAnalysisPeriodRatio < 0) { throw new ArgumentOutOfRangeException(nameof(extraAnalysisPeriodRatio), "extraAnalysisPeriodRatio must be greater than or equal to zero."); } _scoreFunctionProvider = scoreFunctionProvider; _extraAnalysisPeriodRatio = extraAnalysisPeriodRatio; _extensions = extensions?.ToList() ?? new List(); _lock = new object(); _openInsightContexts = new HashSet(); _updatedInsightContexts = new HashSet(); _closedInsightContexts = new ConcurrentDictionary(); } /// /// Add an extension to this manager /// /// The extension to be added public void AddExtension(IInsightManagerExtension extension) { _extensions.Add(extension); } /// /// Initializes any extensions for the specified backtesting range /// /// The start date of the backtest (current time in live mode) /// The end date of the backtest ( in live mode) /// The algorithm's current utc time public void InitializeExtensionsForRange(DateTime start, DateTime end, DateTime current) { foreach (var extension in _extensions) { extension.InitializeForRange(start, end, current); } } /// /// Steps the manager forward in time, accepting new state information and potentialy newly generated insights /// /// The frontier time of the insight analysis /// Snap shot of the securities at the frontier time /// Any insight generated by the algorithm at the frontier time public void Step(DateTime frontierTimeUtc, ReadOnlySecurityValuesCollection securityValuesCollection, GeneratedInsightsCollection generatedInsights) { lock (_lock) { if (generatedInsights != null && generatedInsights.Insights.Count > 0) { foreach (var insight in generatedInsights.Insights) { // save initial security values and deterine analysis period var initialValues = securityValuesCollection[insight.Symbol]; var analysisPeriod = insight.Period + TimeSpan.FromTicks((long)(_extraAnalysisPeriodRatio * insight.Period.Ticks)); // set this as an open analysis context var context = new InsightAnalysisContext(insight, initialValues, analysisPeriod); _openInsightContexts.Add(context); if (context.InitialValues.Price == 0) { Log.Error(Invariant($"InsightManager.Step(): Warning {frontierTimeUtc} UTC: insight {insight} initial price value is 0")); } // let everyone know we've received an insight _extensions.ForEach(e => e.OnInsightGenerated(context)); } } UpdateScores(securityValuesCollection); foreach (var extension in _extensions) { extension.Step(frontierTimeUtc); } } } /// /// Removes insights from the manager with the specified ids /// /// The insights ids to be removed public void RemoveInsights(IEnumerable insightIds) { foreach (var id in insightIds) { InsightAnalysisContext context; _closedInsightContexts.TryRemove(id, out context); } } /// /// Gets all insight analysis contexts that have been updated since this method's last invocation. /// Contexts are marked as not updated during the enumeration, so in order to remove a context from /// the updated set, the enumerable must be enumerated. /// /// public IEnumerable GetUpdatedContexts() { lock (_lock) { var copy = _updatedInsightContexts.ToList(); _updatedInsightContexts.Clear(); return copy; } } /// /// Updates all open insight scores /// private void UpdateScores(ReadOnlySecurityValuesCollection securityValuesCollection) { // for performance be lazy to initialize collection List removals = null; foreach (var context in _openInsightContexts) { // was this insight period closed before we update the times? var previouslyClosed = context.InsightPeriodClosed; // update the security values: price/volatility context.SetCurrentValues(securityValuesCollection[context.Symbol]); // update scores for each score type var currentTimeUtc = context.CurrentValues.TimeUtc; foreach (var scoreType in ScoreTypes) { if (!context.ShouldAnalyze(scoreType)) { // not all insights can receive every score type, for example, insight.Magnitude==null, not point in doing magnitude scoring continue; } // resolve and evaluate the scoring function, storing the result in the context var function = _scoreFunctionProvider.GetScoreFunction(context.Insight.Type, scoreType); var score = function.Evaluate(context, scoreType); context.Score.SetScore(scoreType, score, currentTimeUtc); } // it wasn't closed and now it is closed, fire the event. if (!previouslyClosed && context.InsightPeriodClosed) { _extensions.ForEach(e => e.OnInsightClosed(context)); } // if this score has been finalized, remove it from the open set if (currentTimeUtc >= context.AnalysisEndTimeUtc) { context.Score.Finalize(currentTimeUtc); // set the last value used for scoring context.Insight.ReferenceValueFinal = context.CurrentValues.Get(context.Insight.Type); _extensions.ForEach(e => e.OnInsightAnalysisCompleted(context)); var id = context.Insight.Id; _closedInsightContexts[id] = context; if (removals == null) { removals = new List(); } removals.Add(context); } // mark the context as having been updated _updatedInsightContexts.Add(context); } if (removals != null) { _openInsightContexts.RemoveWhere(removals.Contains); } } /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources. /// 2 public void Dispose() { foreach (var ext in _extensions) { (ext as IDisposable)?.DisposeSafely(); } } } }