Enable trades results streaming

Common/Statistics/AlgorithmPerformance.cs

@@ -83,5 +83,16 @@ 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>
+        /// <param name="trades">The list of closed trades</param>
+        public AlgorithmPerformance(AlgorithmPerformance other, List<Trade> trades)
+        {
+            TradeStatistics = new TradeStatistics(trades);
+            PortfolioStatistics = other.PortfolioStatistics;
+            ClosedTrades = trades;
+        }
     }
 }

Common/Statistics/Trade.cs

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

Common/Statistics/TradeBuilder.cs

@@ -58,6 +58,7 @@ public Position()
         private readonly FillMatchingMethod _matchingMethod;
         private SecurityManager _securities;
         private bool _liveMode;
+        private long _nextTradeId = 1;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="TradeBuilder"/> class
@@ -561,6 +562,7 @@ private void AddNewTrade(Trade trade, OrderEvent fill)
                     ? fill.IsWin(security, trade.ProfitLoss)
                     : trade.ProfitLoss > 0;
 
+                trade.Id = _nextTradeId++;
                 _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,13 @@ private void Update()
                     _nextS3Update = DateTime.UtcNow.AddSeconds(30);
                 }
 
+                var deltaTrades = GetDeltaTrades(statisticsResult.TotalPerformance.ClosedTrades, LastDeltaTradePosition, shouldStop: tradeCount => tradeCount >= 50);
+                // Deliberately skip to the end of trade collection to prevent overloading backtesting UX
+                LastDeltaTradePosition = statisticsResult.TotalPerformance.ClosedTrades[0].Id;
+                var algorithmPerformance = new AlgorithmPerformance(statisticsResult.TotalPerformance, 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 +250,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 +274,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.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,9 @@ public abstract class BaseResultsHandler
 
         private Bar _currentAlgorithmEquity;
 
+        private List<ISeriesPoint> _temporaryPerformanceValues;
+        private List<ISeriesPoint> _temporaryBenchmarkValues;
+
         /// <summary>
         /// String message saying: Strategy Equity
         /// </summary>
@@ -114,6 +117,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 long LastDeltaTradePosition { get; set; }
+
         /// <summary>
         /// The last position consumed from the <see cref="ITransactionHandler.OrderEvents"/> while determining delta order events
         /// </summary>
@@ -122,7 +130,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 +454,26 @@ 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="LastDeltaTradePosition"/> and the <see cref="Trade.Id"/>
+        /// </summary>
+        /// <returns>The delta trades</returns>
+        protected virtual List<Trade> GetDeltaTrades(List<Trade> trades, long tradesStartId, Func<int, bool> shouldStop)
+        {
+            var deltaTrades = new List<Trade>();
+            foreach (var trade in trades.OrderBy(x => x.Id).Where(x => x.Id > tradesStartId))
+            {
+                LastDeltaTradePosition = trade.Id;
+                deltaTrades.Add(trade);
+                if (shouldStop(deltaTrades.Count))
+                {
+                    break;
+                }
+            }
+            return deltaTrades;
+        }
+
         /// <summary>
         /// Initialize the result handler with this result packet.
         /// </summary>
@@ -466,7 +494,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 +664,7 @@ 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);
+            GetPortfolioPerformance(out var currentPortfolioValue, out var portfolioPerformance);
 
             // Update our max portfolio value
             CumulativeMaxPortfolioValue = Math.Max(currentPortfolioValue, CumulativeMaxPortfolioValue);
@@ -659,6 +685,12 @@ public virtual void Sample(DateTime time)
             DailyPortfolioValue = currentPortfolioValue;
         }
 
+        private void GetPortfolioPerformance(out decimal currentPortfolioValue, out decimal portfolioPerformance)
+        {
+            currentPortfolioValue = GetPortfolioValue();
+            portfolioPerformance = 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 +991,66 @@ 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;
+                    List<ISeriesPoint> benchmarkValues;
+                    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
+                        _temporaryPerformanceValues = null;
+                        _temporaryBenchmarkValues = null;
                     }
                     else
                     {
-                        portfolioTurnover = new Series();
+                        // 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
+
+                        if (_temporaryPerformanceValues == null || _temporaryBenchmarkValues == null)
+                        {
+                            // Let's force update and sample both performance and benchmark at the current time since they need to be aligned
+
+                            UpdatePortfolioValues(Algorithm.UtcTime);
+                            UpdateBenchmarkValue(Algorithm.UtcTime);
+                            GetPortfolioPerformance(out _, out var portfolioPerformance);
+
+                            if (portfolioPerformance != 0)
+                            {
+                                _temporaryPerformanceValues = new List<ISeriesPoint> { new ChartPoint(Algorithm.UtcTime, portfolioPerformance) };
+                                _temporaryBenchmarkValues = new List<ISeriesPoint> { new ChartPoint(Algorithm.UtcTime, GetBechmarkValue(Algorithm.UtcTime)) };
+                            }
+                        }
+
+                        performanceValues = _temporaryPerformanceValues;
+                        benchmarkValues = _temporaryBenchmarkValues;
                     }
 
-                    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
-                        );
+                    if (performanceValues != null && benchmarkValues != null)
+                    {
+                        var trades = Algorithm.TradeBuilder.ClosedTrades;
+
+                        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
+                            );
+                    }
                 }
 
                 statisticsResults.AddCustomSummaryStatistics(_customSummaryStatistics);
@@ -1142,5 +1213,10 @@ protected virtual void UpdateBenchmarkValue(DateTime time, bool force = false)
                 _benchmarkValue = new ReferenceWrapper<decimal>(Algorithm.Benchmark.Evaluate(time).SmartRounding());
             }
         }
+
+        private decimal GetBechmarkValue(DateTime time)
+        {
+            return Algorithm.Benchmark.Evaluate(time).SmartRounding();
+        }
     }
 }

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, LastDeltaTradePosition, shouldStop: _ => stopwatch.ElapsedMilliseconds > 15);
+                        algorithmPerformance = new AlgorithmPerformance(statistics.TotalPerformance, 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.Count > 0)
+            {
+                result = result.Concat(new[] { new LiveResultPacket(_job, new LiveResult { TotalPerformance = algorithmPerformance }) });
+            }
+
             return result;
         }