Enable trades results streaming (#9234)

* Add a temporary benchmark stats sample in live

Allow trades to be written to result file on first day of the deployment before the first daily sample is done

* Enable trades results streaming

* Minor changes

* Make Trade.Id a Guid

* Cleanup

* Sample temporary charts once per hour

* Minor change

* Minor fix

* Minor fix

* Minor fix

* Add fix for potential race condition

Author: jhonabreul

Common/Statistics/AlgorithmPerformance.cs

@@ -83,5 +83,15 @@ public AlgorithmPerformance()
             ClosedTrades = new List<Trade>();
         }
 
+        /// <summary>
+        /// Initializes a new instance of the <see cref="AlgorithmPerformance"/> class
+        /// </summary>
+        /// <param name="other">The performance instance to use as a base</param>
+        public AlgorithmPerformance(AlgorithmPerformance other)
+        {
+            TradeStatistics = other.TradeStatistics;
+            PortfolioStatistics = other.PortfolioStatistics;
+            ClosedTrades = other.ClosedTrades;
+        }
     }
 }

Common/Statistics/Trade.cs

@@ -26,6 +26,11 @@ public class Trade
     {
         private List<Symbol> _symbols;
 
+        /// <summary>
+        /// A unique identifier for the trade
+        /// </summary>
+        public string Id { get; set; }
+
         /// <summary>
         /// The symbol of the traded instrument
         /// </summary>

Common/Statistics/TradeBuilder.cs

@@ -615,6 +615,7 @@ private void AddNewTrade(Trade trade, OrderEvent fill)
                     ? fill.IsWin(security, trade.ProfitLoss)
                     : trade.ProfitLoss > 0;
 
+                trade.Id = Guid.NewGuid().ToString();
                 _closedTrades.Add(trade);
 
                 // Due to memory constraints in live mode, we cap the number of trades

Engine/Results/BacktestingResultHandler.cs

@@ -197,8 +197,8 @@ private void Update()
                 }
 
                 //Get the runtime statistics from the user algorithm:
-                var summary = GenerateStatisticsResults(performanceCharts, estimatedStrategyCapacity: _capacityEstimate).Summary;
-                var runtimeStatistics = GetAlgorithmRuntimeStatistics(summary, _capacityEstimate);
+                var statisticsResult = GenerateStatisticsResults(performanceCharts, estimatedStrategyCapacity: _capacityEstimate);
+                var runtimeStatistics = GetAlgorithmRuntimeStatistics(statisticsResult.Summary, _capacityEstimate);
 
                 var progress = _progressMonitor.Progress;
 
@@ -225,8 +225,16 @@ private void Update()
                     _nextS3Update = DateTime.UtcNow.AddSeconds(30);
                 }
 
+                var deltaTrades = GetDeltaTrades(statisticsResult.TotalPerformance.ClosedTrades, LastTradeId, shouldStop: tradeCount => tradeCount >= 50);
+                // Deliberately skip to the end of trade collection to prevent overloading backtesting UX
+                if (statisticsResult.TotalPerformance.ClosedTrades.Count > 0)
+                {
+                    LastTradeId = statisticsResult.TotalPerformance.ClosedTrades[^1].Id;
+                }
+                var algorithmPerformance = new AlgorithmPerformance(statisticsResult.TotalPerformance) { ClosedTrades = deltaTrades };
+
                 //2. Backtest Update -> Send the truncated packet to the backtester:
-                var splitPackets = SplitPackets(deltaCharts, deltaOrders, runtimeStatistics, progress, serverStatistics);
+                var splitPackets = SplitPackets(deltaCharts, deltaOrders, runtimeStatistics, progress, serverStatistics, algorithmPerformance);
 
                 foreach (var backtestingPacket in splitPackets)
                 {
@@ -245,7 +253,9 @@ private void Update()
         /// <summary>
         /// Run over all the data and break it into smaller packets to ensure they all arrive at the terminal
         /// </summary>
-        public virtual IEnumerable<BacktestResultPacket> SplitPackets(Dictionary<string, Chart> deltaCharts, Dictionary<int, Order> deltaOrders, SortedDictionary<string, string> runtimeStatistics, decimal progress, Dictionary<string, string> serverStatistics)
+        public virtual IEnumerable<BacktestResultPacket> SplitPackets(Dictionary<string, Chart> deltaCharts, Dictionary<int, Order> deltaOrders, 
+            SortedDictionary<string, string> runtimeStatistics, decimal progress, Dictionary<string, string> serverStatistics, 
+            AlgorithmPerformance algorithmPerformance)
         {
             // break the charts into groups
             var splitPackets = new List<BacktestResultPacket>();
@@ -267,6 +277,13 @@ public virtual IEnumerable<BacktestResultPacket> SplitPackets(Dictionary<string,
                 splitPackets.Add(new BacktestResultPacket(_job, new BacktestResult { Orders = deltaOrders }, Algorithm.EndDate, Algorithm.StartDate, progress));
             }
 
+            // only send trades if there is actually any update
+            if (algorithmPerformance.ClosedTrades != null && algorithmPerformance.ClosedTrades.Count > 0)
+            {
+                // Add the trades into the charting packet:
+                splitPackets.Add(new BacktestResultPacket(_job, new BacktestResult { TotalPerformance = algorithmPerformance }, 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));
 

Engine/Results/BaseResultsHandler.cs

@@ -14,12 +14,6 @@
  *
 */
 
-using System;
-using System.Collections.Concurrent;
-using System.Collections.Generic;
-using System.IO;
-using System.Linq;
-using System.Threading;
 using Newtonsoft.Json;
 using Newtonsoft.Json.Serialization;
 using QuantConnect.Data.Market;
@@ -34,6 +28,12 @@
 using QuantConnect.Securities.Positions;
 using QuantConnect.Statistics;
 using QuantConnect.Util;
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Threading;
 
 namespace QuantConnect.Lean.Engine.Results
 {
@@ -59,6 +59,11 @@ public abstract class BaseResultsHandler
 
         private Bar _currentAlgorithmEquity;
 
+        private List<ISeriesPoint> _temporaryPerformanceValues;
+        private List<ISeriesPoint> _temporaryBenchmarkValues;
+        private DateTime _temporaryChartsLastSampleTime;
+        private object _temporaryChartsLock = new();
+
         /// <summary>
         /// String message saying: Strategy Equity
         /// </summary>
@@ -114,6 +119,11 @@ public abstract class BaseResultsHandler
         /// </summary>
         protected int LastDeltaOrderPosition { get; set; }
 
+        /// <summary>
+        /// The last position consumed from the <see cref="TradeBuilder.ClosedTrades"/> by <see cref="GetDeltaTrades"/>
+        /// </summary>
+        protected string LastTradeId { get; set; }
+
         /// <summary>
         /// The last position consumed from the <see cref="ITransactionHandler.OrderEvents"/> while determining delta order events
         /// </summary>
@@ -122,7 +132,7 @@ public abstract class BaseResultsHandler
         /// <summary>
         /// Serializer settings to use
         /// </summary>
-        protected JsonSerializerSettings SerializerSettings { get; set; } = new ()
+        protected JsonSerializerSettings SerializerSettings { get; set; } = new()
         {
             ContractResolver = new DefaultContractResolver
             {
@@ -446,6 +456,29 @@ protected virtual Dictionary<int, Order> GetDeltaOrders(int orderEventsStartPosi
             return deltaOrders;
         }
 
+        /// <summary>
+        /// Gets the trades generated starting from the provided <see cref="TradeBuilder.ClosedTrades"/> position,
+        /// which is determined by the <see cref="LastTradeId"/> and the <see cref="Trade.Id"/>
+        /// </summary>
+        /// <returns>The delta trades</returns>
+        protected virtual List<Trade> GetDeltaTrades(List<Trade> trades, string lastTradeId, Func<int, bool> shouldStop)
+        {
+            var lastTradeIndex = trades.FindIndex(x => x.Id == lastTradeId);
+            List<Trade> deltaTrades = null;
+            foreach (var trade in trades.Skip(lastTradeIndex + 1))
+            {
+                LastTradeId = trade.Id;
+                deltaTrades ??= new List<Trade>();
+                deltaTrades.Add(trade);
+                if (shouldStop(deltaTrades.Count))
+                {
+                    break;
+                }
+            }
+
+            return deltaTrades;
+        }
+
         /// <summary>
         /// Initialize the result handler with this result packet.
         /// </summary>
@@ -466,7 +499,7 @@ public virtual void Initialize(ResultHandlerInitializeParameters parameters)
 
             SerializerSettings = new()
             {
-                Converters = new [] { new OrderEventJsonConverter(AlgorithmId) },
+                Converters = new[] { new OrderEventJsonConverter(AlgorithmId) },
                 ContractResolver = new DefaultContractResolver
                 {
                     NamingStrategy = new CamelCaseNamingStrategy
@@ -636,9 +669,8 @@ public virtual void Sample(DateTime time)
             // Force an update for our values before doing our daily sample
             UpdatePortfolioValues(time);
             UpdateBenchmarkValue(time);
-
             var currentPortfolioValue = GetPortfolioValue();
-            var portfolioPerformance = DailyPortfolioValue == 0 ? 0 : Math.Round((currentPortfolioValue - DailyPortfolioValue) * 100 / DailyPortfolioValue, 10);
+            var portfolioPerformance = GetPortfolioPerformance(currentPortfolioValue);
 
             // Update our max portfolio value
             CumulativeMaxPortfolioValue = Math.Max(currentPortfolioValue, CumulativeMaxPortfolioValue);
@@ -659,6 +691,11 @@ public virtual void Sample(DateTime time)
             DailyPortfolioValue = currentPortfolioValue;
         }
 
+        private decimal GetPortfolioPerformance(decimal currentPortfolioValue)
+        {
+            return DailyPortfolioValue == 0 ? 0 : Math.Round((currentPortfolioValue - DailyPortfolioValue) * 100 / DailyPortfolioValue, 10);
+        }
+
         private void SamplePortfolioMargin(DateTime algorithmUtcTime, decimal currentPortfolioValue)
         {
             var state = PortfolioState.Create(Algorithm.Portfolio, algorithmUtcTime, currentPortfolioValue);
@@ -959,27 +996,82 @@ protected StatisticsResults GenerateStatisticsResults(Dictionary<string, Chart>
                 // make sure we've taken samples for these series before just blindly requesting them
                 if (charts.TryGetValue(StrategyEquityKey, out var strategyEquity) &&
                     strategyEquity.Series.TryGetValue(EquityKey, out var equity) &&
-                    strategyEquity.Series.TryGetValue(ReturnKey, out var performance) &&
-                    charts.TryGetValue(BenchmarkKey, out var benchmarkChart) &&
-                    benchmarkChart.Series.TryGetValue(BenchmarkKey, out var benchmark))
+                    equity.Values.Count > 0)
                 {
-                    var trades = Algorithm.TradeBuilder.ClosedTrades;
-
-                    BaseSeries portfolioTurnover;
-                    if (charts.TryGetValue(PortfolioTurnoverKey, out var portfolioTurnoverChart))
+                    List<ISeriesPoint> performanceValues = null;
+                    List<ISeriesPoint> benchmarkValues = null;
+                    if (strategyEquity.Series.TryGetValue(ReturnKey, out var performance) &&
+                        charts.TryGetValue(BenchmarkKey, out var benchmarkChart) &&
+                        benchmarkChart.Series.TryGetValue(BenchmarkKey, out var benchmark))
                     {
-                        portfolioTurnoverChart.Series.TryGetValue(PortfolioTurnoverKey, out portfolioTurnover);
+                        performanceValues = performance.Values;
+                        benchmarkValues = benchmark.Values;
+
+                        // Clear temporary values, free memory. We don't need them anymore
+                        if (_temporaryPerformanceValues != null && _temporaryBenchmarkValues != null)
+                        {
+                            lock (_temporaryChartsLock)
+                            {
+                                _temporaryPerformanceValues = null;
+                                _temporaryBenchmarkValues = null;
+                            }
+                        }
                     }
                     else
                     {
-                        portfolioTurnover = new Series();
+                        lock (_temporaryChartsLock)
+                        {
+                            if (Algorithm.UtcTime - _temporaryChartsLastSampleTime >= TimeSpan.FromHours(1))
+                            {
+                                // We don't have performance and/or benchmark values sampled, likely because we are on the first day of the algo
+                                // and we only sample at the end of the day. In this case we will create temporary values for performance and benchmark
+                                // so that we can generate statistics and write trades to the result files
+
+                                // Let's force update and sample both performance and benchmark at the current time since they need to be aligned
+                                var currentPortfolioValue = GetPortfolioValue();
+                                var portfolioPerformance = GetPortfolioPerformance(currentPortfolioValue);
+
+                                if (portfolioPerformance != 0)
+                                {
+                                    performanceValues = _temporaryPerformanceValues ??= new List<ISeriesPoint>();
+                                    performanceValues.Add(new ChartPoint(Algorithm.UtcTime, portfolioPerformance));
+                                    benchmarkValues = _temporaryBenchmarkValues ??= new List<ISeriesPoint>();
+                                    benchmarkValues.Add(new ChartPoint(Algorithm.UtcTime, GetBenchmarkValue()));
+                                    _temporaryChartsLastSampleTime = Algorithm.UtcTime;
+                                }
+                            }
+
+                            if (performanceValues != null && benchmarkValues != null)
+                            {
+                                performanceValues = [.. performanceValues];
+                                benchmarkValues = [.. benchmarkValues];
+                            }
+                        }
                     }
 
-                    statisticsResults = StatisticsBuilder.Generate(trades, profitLoss, equity.Values, performance.Values, benchmark.Values,
-                        portfolioTurnover.Values, StartingPortfolioValue, Algorithm.Portfolio.TotalFees, TotalTradesCount(),
-                        estimatedStrategyCapacity, AlgorithmCurrencySymbol, Algorithm.Transactions, Algorithm.RiskFreeInterestRateModel,
-                        Algorithm.Settings.TradingDaysPerYear.Value // already set in Brokerage|Backtesting-SetupHandler classes
-                        );
+                    var trades = Algorithm.TradeBuilder.ClosedTrades;
+                    if (performanceValues != null && benchmarkValues != null)
+                    {
+                        BaseSeries portfolioTurnover;
+                        if (charts.TryGetValue(PortfolioTurnoverKey, out var portfolioTurnoverChart))
+                        {
+                            portfolioTurnoverChart.Series.TryGetValue(PortfolioTurnoverKey, out portfolioTurnover);
+                        }
+                        else
+                        {
+                            portfolioTurnover = new Series();
+                        }
+
+                        statisticsResults = StatisticsBuilder.Generate(trades, profitLoss, equity.Values, performanceValues, benchmarkValues,
+                            portfolioTurnover.Values, StartingPortfolioValue, Algorithm.Portfolio.TotalFees, TotalTradesCount(),
+                            estimatedStrategyCapacity, AlgorithmCurrencySymbol, Algorithm.Transactions, Algorithm.RiskFreeInterestRateModel,
+                            Algorithm.Settings.TradingDaysPerYear.Value // already set in Brokerage|Backtesting-SetupHandler classes
+                            );
+                    }
+                    else
+                    {
+                        statisticsResults.TotalPerformance.ClosedTrades = trades;
+                    }
                 }
 
                 statisticsResults.AddCustomSummaryStatistics(_customSummaryStatistics);

Engine/Results/LiveTradingResultHandler.cs

@@ -219,9 +219,16 @@ private void Update()
                     var statistics = GenerateStatisticsResults(performanceCharts);
                     var runtimeStatistics = GetAlgorithmRuntimeStatistics(statistics.Summary);
 
+                    AlgorithmPerformance algorithmPerformance;
+                    {
+                        var stopwatch = Stopwatch.StartNew();
+                        var deltaTrades = GetDeltaTrades(statistics.TotalPerformance.ClosedTrades, LastTradeId, shouldStop: _ => stopwatch.ElapsedMilliseconds > 15);
+                        algorithmPerformance = new AlgorithmPerformance(statistics.TotalPerformance) { ClosedTrades = deltaTrades };
+                    }
+
                     // since we're sending multiple packets, let's do it async and forget about it
                     // chart data can get big so let's break them up into groups
-                    var splitPackets = SplitPackets(deltaCharts, deltaOrders, holdings, Algorithm.Portfolio.CashBook, runtimeStatistics, serverStatistics, deltaOrderEvents);
+                    var splitPackets = SplitPackets(deltaCharts, deltaOrders, holdings, Algorithm.Portfolio.CashBook, runtimeStatistics, serverStatistics, deltaOrderEvents, algorithmPerformance);
 
                     foreach (var liveResultPacket in splitPackets)
                     {
@@ -463,7 +470,8 @@ private IEnumerable<LiveResultPacket> SplitPackets(Dictionary<string, Chart> del
             CashBook cashbook,
             SortedDictionary<string, string> runtimeStatistics,
             Dictionary<string, string> serverStatistics,
-            List<OrderEvent> deltaOrderEvents)
+            List<OrderEvent> deltaOrderEvents,
+            AlgorithmPerformance algorithmPerformance)
         {
             // break the charts into groups
             var current = new Dictionary<string, Chart>();
@@ -518,6 +526,12 @@ private IEnumerable<LiveResultPacket> SplitPackets(Dictionary<string, Chart> del
                 result = result.Concat(new[] { new LiveResultPacket(_job, new LiveResult { Orders = deltaOrders, OrderEvents = deltaOrderEvents }) });
             }
 
+            // only send trades packet if there is actually any update
+            if (algorithmPerformance.ClosedTrades != null && algorithmPerformance.ClosedTrades.Count > 0)
+            {
+                result = result.Concat(new[] { new LiveResultPacket(_job, new LiveResult { TotalPerformance = algorithmPerformance }) });
+            }
+
             return result;
         }