/* * 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 System.Threading; using QuantConnect.Configuration; using QuantConnect.Interfaces; using QuantConnect.Lean.Engine.Setup; using QuantConnect.Lean.Engine.TransactionHandlers; using QuantConnect.Logging; using QuantConnect.Orders; using QuantConnect.Packets; using QuantConnect.Statistics; using QuantConnect.Util; using System.IO; using QuantConnect.Lean.Engine.Alphas; using QuantConnect.Securities; namespace QuantConnect.Lean.Engine.Results { /// /// Backtesting result handler passes messages back from the Lean to the User. /// public class BacktestingResultHandler : BaseResultsHandler, IResultHandler { // used for resetting out/error upon completion private static readonly TextWriter StandardOut = Console.Out; private static readonly TextWriter StandardError = Console.Error; private bool _exitTriggered; private BacktestNodePacket _job; private int _jobDays; private string _compileId = ""; private string _backtestId = ""; private DateTime _nextUpdate; private DateTime _nextS3Update; private DateTime _lastUpdate; private readonly List _log = new List(); private string _errorMessage = ""; private readonly object _chartLock = new object(); private readonly object _runtimeLock = new object(); private readonly Dictionary _runtimeStatistics = new Dictionary(); private double _daysProcessed; private double _daysProcessedFrontier; private bool _processingFinalPacket; private readonly HashSet _chartSeriesExceededDataPoints = new HashSet(); //Processing Time: private readonly DateTime _startTime = DateTime.UtcNow; private DateTime _nextSample; private IMessagingHandler _messagingHandler; private ITransactionHandler _transactionHandler; private ISetupHandler _setupHandler; private string _algorithmId; private int _projectId; private const double Samples = 4000; private const double MinimumSamplePeriod = 4; /// /// Packeting message queue to temporarily store packets and then pull for processing. /// public ConcurrentQueue Messages { get; set; } = new ConcurrentQueue(); /// /// Local object access to the algorithm for the underlying Debug and Error messaging. /// public IAlgorithm Algorithm { get; set; } /// /// Charts collection for storing the master copy of user charting data. /// public ConcurrentDictionary Charts { get; set; } = new ConcurrentDictionary(); /// /// Boolean flag indicating the result hander thread is completely finished and ready to dispose. /// public bool IsActive { get; private set; } = true; /// /// Sampling period for timespans between resamples of the charting equity. /// /// Specifically critical for backtesting since with such long timeframes the sampled data can get extreme. public TimeSpan ResamplePeriod { get; private set; } = TimeSpan.FromMinutes(4); /// /// How frequently the backtests push messages to the browser. /// /// Update frequency of notification packets public TimeSpan NotificationPeriod { get; } = TimeSpan.FromSeconds(2); /// /// A dictionary containing summary statistics /// public Dictionary FinalStatistics { get; private set; } /// /// Default initializer for /// public BacktestingResultHandler() { // 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 /// /// public virtual void Initialize(AlgorithmNodePacket job, IMessagingHandler messagingHandler, IApi api, ISetupHandler setupHandler, ITransactionHandler transactionHandler) { _algorithmId = job.AlgorithmId; _projectId = job.ProjectId; _messagingHandler = messagingHandler; _transactionHandler = transactionHandler; _setupHandler = setupHandler; _job = (BacktestNodePacket)job; if (_job == null) throw new Exception("BacktestingResultHandler.Constructor(): Submitted Job type invalid."); _compileId = _job.CompileId; _backtestId = _job.BacktestId; } /// /// The main processing method steps through the messaging queue and processes the messages one by one. /// public void Run() { //Setup minimum result arrays: //SampleEquity(job.periodStart, job.startingCapital); //SamplePerformance(job.periodStart, 0); 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..."); IsActive = false; // reset standard out/error Console.SetOut(StandardOut); Console.SetError(StandardError); } // End Run(); /// /// Send a backtest update to the browser taking a latest snapshot of the charting data. /// public void Update() { try { //Sometimes don't run the update, if not ready or we're ending. if (Algorithm?.Transactions == null || _processingFinalPacket) { return; } if (DateTime.UtcNow <= _nextUpdate || _daysProcessed < _daysProcessedFrontier) return; //Extract the orders since last update var deltaOrders = new Dictionary(); try { deltaOrders = (from order in _transactionHandler.Orders where order.Value.Time.Date >= _lastUpdate && order.Value.Status == OrderStatus.Filled select order).ToDictionary(t => t.Key, t => t.Value); } catch (Exception err) { Log.Error(err, "Transactions"); } //Limit length of orders we pass back dynamically to avoid flooding. if (deltaOrders.Count > 50) deltaOrders.Clear(); //Reset loop variables: try { _lastUpdate = Algorithm.UtcTime.Date; _daysProcessedFrontier = _daysProcessed + 1; _nextUpdate = DateTime.UtcNow.AddSeconds(2); } catch (Exception err) { Log.Error(err, "Can't update variables"); } var deltaCharts = new Dictionary(); lock (_chartLock) { //Get the updates since the last chart foreach (var kvp in Charts) { var chart = kvp.Value; deltaCharts.Add(chart.Name, chart.GetUpdates()); } } //Get the runtime statistics from the user algorithm: var runtimeStatistics = new Dictionary(); lock (_runtimeLock) { foreach (var pair in _runtimeStatistics) { runtimeStatistics.Add(pair.Key, pair.Value); } } runtimeStatistics.Add("Unrealized", "$" + Algorithm.Portfolio.TotalUnrealizedProfit.ToString("N2")); runtimeStatistics.Add("Fees", "-$" + Algorithm.Portfolio.TotalFees.ToString("N2")); runtimeStatistics.Add("Net Profit", "$" + (Algorithm.Portfolio.TotalProfit - Algorithm.Portfolio.TotalFees).ToString("N2")); // when there is an initialization error StartingPortfolioValue is 0, want to avoid dividing by zero if (_setupHandler.StartingPortfolioValue != 0) { runtimeStatistics.Add("Return", ((Algorithm.Portfolio.TotalPortfolioValue - _setupHandler.StartingPortfolioValue) / _setupHandler.StartingPortfolioValue).ToString("P")); } runtimeStatistics.Add("Equity", "$" + Algorithm.Portfolio.TotalPortfolioValue.ToString("N2")); //Profit Loss Changes: var progress = Convert.ToDecimal(_daysProcessed / _jobDays); if (progress > 0.999m) progress = 0.999m; //1. Cloud Upload -> Upload the whole packet to S3 Immediately: if (DateTime.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( Charts, orderCount > maxOrders ? _transactionHandler.Orders.Skip(orderCount - maxOrders).ToDictionary() : _transactionHandler.Orders.ToDictionary(), Algorithm.Transactions.TransactionRecord, new Dictionary(), runtimeStatistics, new Dictionary()); StoreResult(new BacktestResultPacket(_job, completeResult, progress)); _nextS3Update = DateTime.UtcNow.AddSeconds(30); } //2. Backtest Update -> Send the truncated packet to the backtester: var splitPackets = SplitPackets(deltaCharts, deltaOrders, runtimeStatistics, progress); foreach (var backtestingPacket in splitPackets) { _messagingHandler.Send(backtestingPacket); } } 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 IEnumerable SplitPackets(Dictionary deltaCharts, Dictionary deltaOrders, Dictionary runtimeStatistics, decimal progress) { // break the charts into groups var splitPackets = new List(); foreach (var chart in deltaCharts.Values) { //Don't add packet if the series is empty: if (chart.Series.Values.Aggregate(0, (i, x) => i + x.Values.Count) == 0) continue; splitPackets.Add(new BacktestResultPacket(_job, new BacktestResult { Charts = new Dictionary() { {chart.Name, chart} } }, progress)); } // Send alpha run time statistics splitPackets.Add(new BacktestResultPacket(_job, new BacktestResult { AlphaRuntimeStatistics = AlphaRuntimeStatistics}, progress)); // Add the orders into the charting packet: splitPackets.Add(new BacktestResultPacket(_job, new BacktestResult { Orders = deltaOrders }, progress)); //Add any user runtime statistics into the backtest. splitPackets.Add(new BacktestResultPacket(_job, new BacktestResult { RuntimeStatistics = runtimeStatistics }, progress)); return splitPackets; } /// /// Save the snapshot of the total results to storage. /// /// Packet to store. /// Store the packet asyncronously to speed up the thread. /// Async creates crashes in Mono 3.10 if the thread disappears before the upload is complete so it is disabled for now. public void StoreResult(Packet packet, bool async = false) { 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 = _job.BacktestId + ".json"; BacktestResult results; lock (_chartLock) { results = new BacktestResult( 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, result.Results.TotalPerformance ) // Set Alpha Runtime Statistics { AlphaRuntimeStatistics = result.Results.AlphaRuntimeStatistics }; } // Save results SaveResults(key, results); } else { Log.Error("BacktestingResultHandler.StoreResult(): Result Null."); } } catch (Exception err) { Log.Error(err); } } /// /// Send a final analysis result back to the IDE. /// /// Lean AlgorithmJob task /// Collection of orders from the algorithm /// Collection of time-profit values for the algorithm /// Current holdings state for the algorithm /// Cashbook for the holdingss /// Statistics information for the algorithm (empty if not finished) /// Runtime statistics banner information public void SendFinalResult(AlgorithmNodePacket job, Dictionary orders, Dictionary profitLoss, Dictionary holdings, CashBook cashbook, StatisticsResults statisticsResults, Dictionary banner) { try { FinalStatistics = statisticsResults.Summary; //Convert local dictionary: var charts = new Dictionary(Charts); _processingFinalPacket = true; // clear the trades collection before placing inside the backtest result foreach (var ap in statisticsResults.RollingPerformances.Values) { ap.ClosedTrades.Clear(); } //Create a result packet to send to the browser. var result = new BacktestResultPacket((BacktestNodePacket) job, new BacktestResult(charts, orders, profitLoss, statisticsResults.Summary, banner, statisticsResults.RollingPerformances, statisticsResults.TotalPerformance) { AlphaRuntimeStatistics = AlphaRuntimeStatistics }) { ProcessingTime = (DateTime.UtcNow - _startTime).TotalSeconds, DateFinished = DateTime.Now, 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. /// While setting the algorithm the backtest result handler. public void SetAlgorithm(IAlgorithm algorithm) { Algorithm = algorithm; //Get the resample period: var totalMinutes = (_job.PeriodFinish - _job.PeriodStart).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.ToString("00.00")); //Setup the sampling periods: _jobDays = Algorithm.Securities.Count > 0 ? Time.TradeableDates(Algorithm.Securities.Values, _job.PeriodStart, _job.PeriodFinish) : Convert.ToInt32((_job.PeriodFinish.Date - _job.PeriodStart.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); 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))); } } /// /// Send a debug message back to the browser console. /// /// Message we'd like shown in console. public void DebugMessage(string message) { Messages.Enqueue(new DebugPacket(_projectId, _backtestId, _compileId, message)); //Save last message sent: if (Algorithm != null) { _log.Add(Algorithm.Time.ToString(DateFormat.UI) + " " + message); } } /// /// Send a system debug message back to the browser console. /// /// Message we'd like shown in console. public void SystemDebugMessage(string message) { Messages.Enqueue(new SystemDebugPacket(_projectId, _backtestId, _compileId, message)); //Save last message sent: if (Algorithm != null) { _log.Add(Algorithm.Time.ToString(DateFormat.UI) + " " + message); } } /// /// Send a logging message to the log list for storage. /// /// Message we'd in the log. public void LogMessage(string message) { Messages.Enqueue(new LogPacket(_backtestId, message)); if (Algorithm != null) { _log.Add(Algorithm.Time.ToString(DateFormat.UI) + " " + message); } } /// /// Send list of security asset types the algortihm uses to browser. /// public 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 void ErrorMessage(string message, string stacktrace = "") { if (message == _errorMessage) return; if (Messages.Count > 500) return; Messages.Enqueue(new HandledErrorPacket(_backtestId, message, stacktrace)); _errorMessage = message; } /// /// Send a runtime error message back to the browser highlighted with in red /// /// Error message. /// Stacktrace information string public void RuntimeError(string message, string stacktrace = "") { PurgeQueue(); Messages.Enqueue(new RuntimeErrorPacket(_job.UserId, _backtestId, 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. public 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.Values.Add(new ChartPoint(time, value)); } } } /// /// Sample the current equity of the strategy directly with time-value pair. /// /// Current backtest time. /// Current equity value. public void SampleEquity(DateTime time, decimal value) { //Sample the Equity Value: Sample("Strategy Equity", "Equity", 0, SeriesType.Candle, time, value); //Recalculate the days processed: _daysProcessed = (time - Algorithm.StartDate).TotalDays; } /// /// Sample the current daily performance directly with a time-value pair. /// /// Current backtest date. /// Current daily performance value. public void SamplePerformance(DateTime time, decimal value) { //Added a second chart to equity plot - daily perforamnce: Sample("Strategy Equity", "Daily Performance", 1, SeriesType.Bar, time, value, "%"); } /// /// Sample the current benchmark performance directly with a time-value pair. /// /// Current backtest date. /// Current benchmark value. /// public void SampleBenchmark(DateTime time, decimal value) { Sample("Benchmark", "Benchmark", 0, SeriesType.Line, time, value); } /// /// Add a range of samples from the users algorithms to the end of our current list. /// /// Chart updates since the last request. public 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. /// public virtual void Exit() { // Only process the logs once if (!_exitTriggered) { ProcessSynchronousEvents(true); var logLocation = SaveLogs(_algorithmId, _log); SystemDebugMessage("Your log was successfully created and can be retrieved from: " + logLocation); } //Set exit flag, and wait for the messages to send: _exitTriggered = true; } /// /// Send a new order event to the browser. /// /// In backtesting the order events are not sent because it would generate a high load of messaging. /// New order event details public virtual void OrderEvent(OrderEvent newEvent) { // NOP. Don't do any order event processing for results in backtest mode. } /// /// 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); } /// /// Sample the asset prices to generate plots. /// /// Symbol we're sampling. /// Time of sample /// Value of the asset price public void SampleAssetPrices(Symbol symbol, DateTime time, decimal value) { //NOP. Don't sample asset prices in console. } /// /// Purge/clear any outstanding messages in message queue. /// public void PurgeQueue() { Messages.Clear(); } /// /// 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 void RuntimeStatistic(string key, string value) { lock (_runtimeLock) { _runtimeStatistics[key] = value; } } /// /// Set the chart subscription we want data for. Not used in backtesting. /// public void SetChartSubscription(string symbol) { //NOP. } /// /// 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 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()); //Sample the asset pricing: foreach (var kvp in Algorithm.Securities) { var security = kvp.Value; SampleAssetPrices(security.Symbol, time, security.Price); } } long endTime; // avoid calling utcNow if not required if (Algorithm.DebugMessages.Count > 0) { //Send out the debug messages: endTime = DateTime.UtcNow.AddMilliseconds(250).Ticks; while (Algorithm.DebugMessages.Count > 0 && DateTime.UtcNow.Ticks < endTime) { string message; if (Algorithm.DebugMessages.TryDequeue(out message)) { DebugMessage(message); } } } // avoid calling utcNow if not required if (Algorithm.ErrorMessages.Count > 0) { //Send out the error messages: endTime = DateTime.UtcNow.AddMilliseconds(250).Ticks; while (Algorithm.ErrorMessages.Count > 0 && DateTime.UtcNow.Ticks < endTime) { string message; if (Algorithm.ErrorMessages.TryDequeue(out message)) { ErrorMessage(message); } } } // avoid calling utcNow if not required if (Algorithm.LogMessages.Count > 0) { //Send out the log messages: endTime = DateTime.UtcNow.AddMilliseconds(250).Ticks; while (Algorithm.LogMessages.Count > 0 && DateTime.UtcNow.Ticks < endTime) { string message; if (Algorithm.LogMessages.TryDequeue(out message)) { LogMessage(message); } } } //Set the running statistics: foreach (var pair in Algorithm.RuntimeStatistics) { RuntimeStatistic(pair.Key, pair.Value); } } } }