/* * 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.Generic; using System.IO; using System.Linq; using System.Threading; using QuantConnect.Configuration; using QuantConnect.Interfaces; using QuantConnect.Lean.Engine.TransactionHandlers; using QuantConnect.Logging; using QuantConnect.Orders; using QuantConnect.Packets; using QuantConnect.Statistics; using QuantConnect.Util; using QuantConnect.Lean.Engine.Alphas; namespace QuantConnect.Lean.Engine.Results { /// /// Backtesting result handler passes messages back from the Lean to the User. /// public class BacktestingResultHandler : BaseResultsHandler, IResultHandler { private const double Samples = 4000; private const double MinimumSamplePeriod = 4; private BacktestNodePacket _job; private int _jobDays; private DateTime _nextUpdate; private DateTime _nextS3Update; private string _errorMessage; private int _daysProcessed; private int _daysProcessedFrontier; private readonly HashSet _chartSeriesExceededDataPoints; //Processing Time: private DateTime _nextSample; private string _algorithmId; private int _projectId; /// /// A dictionary containing summary statistics /// public Dictionary FinalStatistics { get; private set; } /// /// Creates a new instance /// public BacktestingResultHandler() { ResamplePeriod = TimeSpan.FromMinutes(4); NotificationPeriod = TimeSpan.FromSeconds(2); _chartSeriesExceededDataPoints = new HashSet(); // Delay uploading first packet _nextS3Update = StartTime.AddSeconds(30); //Default charts: Charts.AddOrUpdate("Strategy Equity", new Chart("Strategy Equity")); Charts["Strategy Equity"].Series.Add("Equity", new Series("Equity", SeriesType.Candle, 0, "$")); Charts["Strategy Equity"].Series.Add("Daily Performance", new Series("Daily Performance", SeriesType.Bar, 1, "%")); } /// /// Initialize the result handler with this result packet. /// /// Algorithm job packet for this result handler /// The handler responsible for communicating messages to listeners /// The api instance used for handling logs /// The transaction handler used to get the algorithms information public override void Initialize(AlgorithmNodePacket job, IMessagingHandler messagingHandler, IApi api, ITransactionHandler transactionHandler) { _algorithmId = job.AlgorithmId; _projectId = job.ProjectId; _job = (BacktestNodePacket)job; if (_job == null) throw new Exception("BacktestingResultHandler.Constructor(): Submitted Job type invalid."); base.Initialize(job, messagingHandler, api, transactionHandler); } /// /// The main processing method steps through the messaging queue and processes the messages one by one. /// protected override void Run() { try { while (!(ExitTriggered && Messages.Count == 0)) { //While there's no work to do, go back to the algorithm: if (Messages.Count == 0) { Thread.Sleep(50); } else { //1. Process Simple Messages in Queue Packet packet; if (Messages.TryDequeue(out packet)) { MessagingHandler.Send(packet); } } //2. Update the packet scanner: Update(); } // While !End. } catch (Exception err) { // unexpected error, we need to close down shop Log.Error(err); // quit the algorithm due to error Algorithm.RunTimeError = err; } Log.Trace("BacktestingResultHandler.Run(): Ending Thread..."); } // End Run(); /// /// Send a backtest update to the browser taking a latest snapshot of the charting data. /// private void Update() { try { //Sometimes don't run the update, if not ready or we're ending. if (Algorithm?.Transactions == null || ExitTriggered) { return; } var utcNow = DateTime.UtcNow; if (utcNow <= _nextUpdate || _daysProcessed < _daysProcessedFrontier) return; var deltaOrders = GetDeltaOrders(LastDeltaOrderPosition, shouldStop: orderCount => orderCount >= 50); // Deliberately skip to the end of order event collection to prevent overloading backtesting UX LastDeltaOrderPosition = TransactionHandler.OrderEvents.Count(); //Reset loop variables: try { _daysProcessedFrontier = _daysProcessed + 1; _nextUpdate = utcNow.AddSeconds(3); } catch (Exception err) { Log.Error(err, "Can't update variables"); } var deltaCharts = new Dictionary(); var serverStatistics = GetServerStatistics(utcNow); var performanceCharts = new Dictionary(); lock (ChartLock) { //Get the updates since the last chart foreach (var kvp in Charts) { var chart = kvp.Value; var updates = chart.GetUpdates(); if (!updates.IsEmpty()) { deltaCharts.Add(chart.Name, updates); } if (AlgorithmPerformanceCharts.Contains(kvp.Key)) { performanceCharts[kvp.Key] = chart.Clone(); } } } //Get the runtime statistics from the user algorithm: var runtimeStatistics = new Dictionary(); lock (RuntimeStatistics) { foreach (var pair in RuntimeStatistics) { runtimeStatistics.Add(pair.Key, pair.Value); } } var summary = GenerateStatisticsResults(performanceCharts).Summary; GetAlgorithmRuntimeStatistics(summary, runtimeStatistics); var progress = (decimal)_daysProcessed / _jobDays; if (progress > 0.999m) progress = 0.999m; //1. Cloud Upload -> Upload the whole packet to S3 Immediately: if (utcNow > _nextS3Update) { // For intermediate backtesting results, we truncate the order list to include only the last 100 orders // The final packet will contain the full list of orders. const int maxOrders = 100; var orderCount = TransactionHandler.Orders.Count; var completeResult = new BacktestResult(new BacktestResultParameters( Charts, orderCount > maxOrders ? TransactionHandler.Orders.Skip(orderCount - maxOrders).ToDictionary() : TransactionHandler.Orders.ToDictionary(), Algorithm.Transactions.TransactionRecord, new Dictionary(), runtimeStatistics, new Dictionary(), // we store the last 100 order events, the final packet will contain the full list TransactionHandler.OrderEvents.Reverse().Take(100).ToList())); StoreResult(new BacktestResultPacket(_job, completeResult, Algorithm.EndDate, Algorithm.StartDate, progress)); _nextS3Update = DateTime.UtcNow.AddSeconds(30); } //2. Backtest Update -> Send the truncated packet to the backtester: var splitPackets = SplitPackets(deltaCharts, deltaOrders, runtimeStatistics, progress, serverStatistics); foreach (var backtestingPacket in splitPackets) { MessagingHandler.Send(backtestingPacket); } // let's re update this value after we finish just in case, so we don't re enter in the next loop _nextUpdate = DateTime.UtcNow.Add(MainUpdateInterval); } catch (Exception err) { Log.Error(err); } } /// /// Run over all the data and break it into smaller packets to ensure they all arrive at the terminal /// public virtual IEnumerable SplitPackets(Dictionary deltaCharts, Dictionary deltaOrders, Dictionary runtimeStatistics, decimal progress, Dictionary serverStatistics) { // break the charts into groups var splitPackets = new List(); foreach (var chart in deltaCharts.Values) { splitPackets.Add(new BacktestResultPacket(_job, new BacktestResult { Charts = new Dictionary { {chart.Name, chart} } }, Algorithm.EndDate, Algorithm.StartDate, progress)); } // Send alpha run time statistics splitPackets.Add(new BacktestResultPacket(_job, new BacktestResult { AlphaRuntimeStatistics = AlphaRuntimeStatistics }, Algorithm.EndDate, Algorithm.StartDate, progress)); // only send orders if there is actually any update if (deltaOrders.Count > 0) { // Add the orders into the charting packet: splitPackets.Add(new BacktestResultPacket(_job, new BacktestResult { Orders = deltaOrders }, Algorithm.EndDate, Algorithm.StartDate, progress)); } //Add any user runtime statistics into the backtest. splitPackets.Add(new BacktestResultPacket(_job, new BacktestResult { ServerStatistics = serverStatistics, RuntimeStatistics = runtimeStatistics }, Algorithm.EndDate, Algorithm.StartDate, progress)); return splitPackets; } /// /// Save the snapshot of the total results to storage. /// /// Packet to store. protected override void StoreResult(Packet packet) { try { // Make sure this is the right type of packet: if (packet.Type != PacketType.BacktestResult) return; // Port to packet format: var result = packet as BacktestResultPacket; if (result != null) { // Get Storage Location: var key = $"{AlgorithmId}.json"; BacktestResult results; lock (ChartLock) { results = new BacktestResult(new BacktestResultParameters( result.Results.Charts.ToDictionary(x => x.Key, x => x.Value.Clone()), result.Results.Orders, result.Results.ProfitLoss, result.Results.Statistics, result.Results.RuntimeStatistics, result.Results.RollingWindow, null, // null order events, we store them separately result.Results.TotalPerformance, result.Results.AlphaRuntimeStatistics)); } // Save results SaveResults(key, results); // Store Order Events in a separate file StoreOrderEvents(Algorithm?.UtcTime ?? DateTime.UtcNow, result.Results.OrderEvents); } else { Log.Error("BacktestingResultHandler.StoreResult(): Result Null."); } } catch (Exception err) { Log.Error(err); } } /// /// Send a final analysis result back to the IDE. /// protected void SendFinalResult() { try { BacktestResultPacket result; // could happen if algorithm failed to init if (Algorithm != null) { //Convert local dictionary: var charts = new Dictionary(Charts); var orders = new Dictionary(TransactionHandler.Orders); var profitLoss = new SortedDictionary(Algorithm.Transactions.TransactionRecord); var statisticsResults = GenerateStatisticsResults(charts, profitLoss); var runtime = GetAlgorithmRuntimeStatistics(statisticsResults.Summary); FinalStatistics = statisticsResults.Summary; // clear the trades collection before placing inside the backtest result foreach (var ap in statisticsResults.RollingPerformances.Values) { ap.ClosedTrades.Clear(); } var orderEvents = TransactionHandler.OrderEvents.ToList(); //Create a result packet to send to the browser. result = new BacktestResultPacket(_job, new BacktestResult(new BacktestResultParameters(charts, orders, profitLoss, statisticsResults.Summary, runtime, statisticsResults.RollingPerformances, orderEvents, statisticsResults.TotalPerformance, AlphaRuntimeStatistics)), Algorithm.EndDate, Algorithm.StartDate); } else { result = BacktestResultPacket.CreateEmpty(_job); } result.ProcessingTime = (DateTime.UtcNow - StartTime).TotalSeconds; result.DateFinished = DateTime.Now; result.Progress = 1; //Place result into storage. StoreResult(result); //Second, send the truncated packet: MessagingHandler.Send(result); Log.Trace("BacktestingResultHandler.SendAnalysisResult(): Processed final packet"); } catch (Exception err) { Log.Error(err); } } /// /// Set the Algorithm instance for ths result. /// /// Algorithm we're working on. /// Algorithm starting capital for statistics calculations /// While setting the algorithm the backtest result handler. public virtual void SetAlgorithm(IAlgorithm algorithm, decimal startingPortfolioValue) { Algorithm = algorithm; StartingPortfolioValue = startingPortfolioValue; PreviousUtcSampleTime = Algorithm.UtcTime; DailyPortfolioValue = StartingPortfolioValue; //Get the resample period: var totalMinutes = (algorithm.EndDate - algorithm.StartDate).TotalMinutes; var resampleMinutes = totalMinutes < MinimumSamplePeriod * Samples ? MinimumSamplePeriod : totalMinutes / Samples; // Space out the sampling every ResamplePeriod = TimeSpan.FromMinutes(resampleMinutes); Log.Trace("BacktestingResultHandler(): Sample Period Set: " + resampleMinutes.ToStringInvariant("00.00")); //Setup the sampling periods: _jobDays = Algorithm.Securities.Count > 0 ? Time.TradeableDates(Algorithm.Securities.Values, algorithm.StartDate, algorithm.EndDate) : Convert.ToInt32((algorithm.EndDate.Date - algorithm.StartDate.Date).TotalDays) + 1; //Set the security / market types. var types = new List(); foreach (var kvp in Algorithm.Securities) { var security = kvp.Value; if (!types.Contains(security.Type)) types.Add(security.Type); } SecurityType(types); ConfigureConsoleTextWriter(algorithm); } /// /// Send a debug message back to the browser console. /// /// Message we'd like shown in console. public virtual void DebugMessage(string message) { Messages.Enqueue(new DebugPacket(_projectId, AlgorithmId, CompileId, message)); AddToLogStore(message); } /// /// Send a system debug message back to the browser console. /// /// Message we'd like shown in console. public virtual void SystemDebugMessage(string message) { Messages.Enqueue(new SystemDebugPacket(_projectId, AlgorithmId, CompileId, message)); AddToLogStore(message); } /// /// Send a logging message to the log list for storage. /// /// Message we'd in the log. public virtual void LogMessage(string message) { Messages.Enqueue(new LogPacket(AlgorithmId, message)); AddToLogStore(message); } protected override void AddToLogStore(string message) { lock (LogStore) { var messageToLog = Algorithm != null ? new LogEntry(Algorithm.Time.ToStringInvariant(DateFormat.UI) + " " + message) : new LogEntry("Algorithm Initialization: " + message); LogStore.Add(messageToLog); } } /// /// Send list of security asset types the algortihm uses to browser. /// public virtual void SecurityType(List types) { var packet = new SecurityTypesPacket { Types = types }; Messages.Enqueue(packet); } /// /// Send an error message back to the browser highlighted in red with a stacktrace. /// /// Error message we'd like shown in console. /// Stacktrace information string public virtual void ErrorMessage(string message, string stacktrace = "") { if (message == _errorMessage) return; if (Messages.Count > 500) return; Messages.Enqueue(new HandledErrorPacket(AlgorithmId, message, stacktrace)); _errorMessage = message; } /// /// Send a runtime error message back to the browser highlighted with in red /// /// Error message. /// Stacktrace information string public virtual void RuntimeError(string message, string stacktrace = "") { PurgeQueue(); Messages.Enqueue(new RuntimeErrorPacket(_job.UserId, AlgorithmId, message, stacktrace)); _errorMessage = message; } /// /// Add a sample to the chart specified by the chartName, and seriesName. /// /// String chart name to place the sample. /// Type of chart we should create if it doesn't already exist. /// Series name for the chart. /// Series type for the chart. /// Time for the sample /// Unit of the sample /// Value for the chart sample. protected override void Sample(string chartName, string seriesName, int seriesIndex, SeriesType seriesType, DateTime time, decimal value, string unit = "$") { // Sampling during warming up period skews statistics if (Algorithm.IsWarmingUp) { return; } lock (ChartLock) { //Add a copy locally: Chart chart; if (!Charts.TryGetValue(chartName, out chart)) { chart = new Chart(chartName); Charts.AddOrUpdate(chartName, chart); } //Add the sample to our chart: Series series; if (!chart.Series.TryGetValue(seriesName, out series)) { series = new Series(seriesName, seriesType, seriesIndex, unit); chart.Series.Add(seriesName, series); } //Add our value: if (series.Values.Count == 0 || time > Time.UnixTimeStampToDateTime(series.Values[series.Values.Count - 1].x)) { series.AddPoint(time, value); } } } /// /// Sample the current equity of the strategy directly with time-value pair. /// /// Current backtest time. /// Current equity value. protected override void SampleEquity(DateTime time, decimal value) { base.SampleEquity(time, value); try { //Recalculate the days processed. We use 'int' so it's thread safe _daysProcessed = (int) (time - Algorithm.StartDate).TotalDays; } catch (OverflowException) { } } /// /// Add a range of samples from the users algorithms to the end of our current list. /// /// Chart updates since the last request. protected void SampleRange(List updates) { lock (ChartLock) { foreach (var update in updates) { //Create the chart if it doesn't exist already: Chart chart; if (!Charts.TryGetValue(update.Name, out chart)) { chart = new Chart(update.Name); Charts.AddOrUpdate(update.Name, chart); } // for alpha assets chart, we always create a new series instance (step on previous value) var forceNewSeries = update.Name == ChartingInsightManagerExtension.AlphaAssets; //Add these samples to this chart. foreach (var series in update.Series.Values) { if (series.Values.Count > 0) { var thisSeries = chart.TryAddAndGetSeries(series.Name, series.SeriesType, series.Index, series.Unit, series.Color, series.ScatterMarkerSymbol, forceNewSeries); if (series.SeriesType == SeriesType.Pie) { var dataPoint = series.ConsolidateChartPoints(); if (dataPoint != null) { thisSeries.AddPoint(dataPoint); } } else { var values = thisSeries.Values; if ((values.Count + series.Values.Count) <= _job.Controls.MaximumDataPointsPerChartSeries) // check chart data point limit first { //We already have this record, so just the new samples to the end: values.AddRange(series.Values); } else if (!_chartSeriesExceededDataPoints.Contains(chart.Name + series.Name)) { _chartSeriesExceededDataPoints.Add(chart.Name + series.Name); DebugMessage($"Exceeded maximum data points per series, chart update skipped. Chart Name {update.Name}. Series name {series.Name}. " + $"Limit is currently set at {_job.Controls.MaximumDataPointsPerChartSeries}"); } } } } } } } /// /// Terminate the result thread and apply any required exit procedures like sending final results. /// public override void Exit() { // Only process the logs once if (!ExitTriggered) { Log.Trace("BacktestingResultHandler.Exit(): starting..."); List copy; lock (LogStore) { copy = LogStore.ToList(); } ProcessSynchronousEvents(true); Log.Trace("BacktestingResultHandler.Exit(): Saving logs..."); var logLocation = SaveLogs(_algorithmId, copy); SystemDebugMessage("Your log was successfully created and can be retrieved from: " + logLocation); // Set exit flag, update task will send any message before stopping ExitTriggered = true; StopUpdateRunner(); SendFinalResult(); base.Exit(); } } /// /// Send an algorithm status update to the browser. /// /// Status enum value. /// Additional optional status message. public virtual void SendStatusUpdate(AlgorithmStatus status, string message = "") { var statusPacket = new AlgorithmStatusPacket(_algorithmId, _projectId, status, message); MessagingHandler.Send(statusPacket); } /// /// Set the current runtime statistics of the algorithm. /// These are banner/title statistics which show at the top of the live trading results. /// /// Runtime headline statistic name /// Runtime headline statistic value public virtual void RuntimeStatistic(string key, string value) { lock (RuntimeStatistics) { RuntimeStatistics[key] = value; } } /// /// Process the synchronous result events, sampling and message reading. /// This method is triggered from the algorithm manager thread. /// /// Prime candidate for putting into a base class. Is identical across all result handlers. public virtual void ProcessSynchronousEvents(bool forceProcess = false) { if (Algorithm == null) return; var time = Algorithm.UtcTime; if (time > _nextSample || forceProcess) { //Set next sample time: 4000 samples per backtest _nextSample = time.Add(ResamplePeriod); //Sample the portfolio value over time for chart. SampleEquity(time, Math.Round(Algorithm.Portfolio.TotalPortfolioValue, 4)); //Also add the user samples / plots to the result handler tracking: SampleRange(Algorithm.GetChartUpdates()); } ProcessAlgorithmLogs(); //Set the running statistics: foreach (var pair in Algorithm.RuntimeStatistics) { RuntimeStatistic(pair.Key, pair.Value); } } /// /// Configures the and /// instances. By default, we forward to . /// This is perfect for running in the cloud, but since they're processed asynchronously, the ordering of these /// messages with respect to messages is broken. This can lead to differences in regression /// test logs based solely on the ordering of messages. To disable this forwarding, set "forward-console-messages" /// to false in the configuration. /// protected virtual void ConfigureConsoleTextWriter(IAlgorithm algorithm) { if (Config.GetBool("forward-console-messages", true)) { // we need to forward Console.Write messages to the algorithm's Debug function Console.SetOut(new FuncTextWriter(algorithm.Debug)); Console.SetError(new FuncTextWriter(algorithm.Error)); } else { // we need to forward Console.Write messages to the standard Log functions Console.SetOut(new FuncTextWriter(msg => Log.Trace(msg))); Console.SetError(new FuncTextWriter(msg => Log.Error(msg))); } } } }