Force live results store once holdings changes settle

Engine/Results/LiveTradingResultHandler.cs

@@ -58,6 +58,8 @@ public class LiveTradingResultHandler : BaseResultsHandler, IResultHandler
 
         private readonly TimeSpan _storeInsightPeriod;
 
+        private readonly HoldingsChangeMonitor _holdingsChangeMonitor = new();
+
         private DateTime _nextPortfolioMarginUpdate;
         private DateTime _previousPortfolioMarginUpdate;
         private readonly TimeSpan _samplePortfolioPeriod;
@@ -108,6 +110,9 @@ public override void Initialize(ResultHandlerInitializeParameters parameters)
             _currentUtcDate = utcNow.Date;
 
             _nextPortfolioMarginUpdate = utcNow.RoundDown(_samplePortfolioPeriod).Add(_samplePortfolioPeriod);
+
+            _holdingsChangeMonitor.Monitor(parameters.TransactionHandler);
+
             base.Initialize(parameters);
         }
 
@@ -235,8 +240,10 @@ private void Update()
                         MessagingHandler.Send(liveResultPacket);
                     }
 
-                    //Send full packet to storage.
-                    if (utcNow > _nextChartsUpdate)
+                    // Send full packet to storage. Holdings changes not yet persisted force a store once they
+                    // have settled for the configured delay, so the stored results reflect fills quickly
+                    // instead of waiting for the next scheduled store
+                    if (utcNow > _nextChartsUpdate || _holdingsChangeMonitor.ShouldForceStore(utcNow))
                     {
                         Log.Debug("LiveTradingResultHandler.Update(): Pre-store result");
                         var chartComplete = new Dictionary<string, Chart>();
@@ -258,6 +265,7 @@ private void Update()
                             Algorithm.Transactions.TransactionRecord, holdings, Algorithm.Portfolio.CashBook, deltaStatistics,
                             runtimeStatistics, orderEvents, statistics.TotalPerformance, serverStatistics, state: GetAlgorithmState())));
                         StoreResult(complete);
+                        _holdingsChangeMonitor.MarkStored();
                         _nextChartsUpdate = DateTime.UtcNow.Add(ChartUpdateInterval);
                         Log.Debug("LiveTradingResultHandler.Update(): End-store result");
                     }
@@ -1340,5 +1348,52 @@ public void SetSummaryStatistic(string name, string value)
         {
             SummaryStatistic(name, value);
         }
+
+        /// <summary>
+        /// Monitors the order events, keeping track of the last time the holdings changed by filtering fills,
+        /// so the update loop can force a store of the full results once the changes settle,
+        /// without waiting for the next scheduled store
+        /// </summary>
+        private class HoldingsChangeMonitor
+        {
+            private readonly TimeSpan _storeDelay = TimeSpan.FromSeconds(Config.GetDouble("holdings-changed-store-delay", 10));
+
+            private long _lastChangedTicks;
+            private long _lastStoredTicks;
+
+            /// <summary>
+            /// Monitors the order events to keep track of the last time the holdings changed
+            /// </summary>
+            public void Monitor(IOrderEventProvider orderEventProvider)
+            {
+                orderEventProvider.NewOrderEvent += OnNewOrderEvent;
+            }
+
+            /// <summary>
+            /// Determines whether the latest holdings changes have not been stored yet and have
+            /// settled for the configured delay, in which case a results store should be forced
+            /// </summary>
+            public bool ShouldForceStore(DateTime utcNow)
+            {
+                var lastChangedTicks = Interlocked.Read(ref _lastChangedTicks);
+                return lastChangedTicks > _lastStoredTicks && utcNow.Ticks >= lastChangedTicks + _storeDelay.Ticks;
+            }
+
+            /// <summary>
+            /// Marks the latest holdings changes as stored
+            /// </summary>
+            public void MarkStored()
+            {
+                _lastStoredTicks = Interlocked.Read(ref _lastChangedTicks);
+            }
+
+            private void OnNewOrderEvent(object sender, OrderEvent orderEvent)
+            {
+                if (orderEvent.Status.IsFill())
+                {
+                    Interlocked.Exchange(ref _lastChangedTicks, orderEvent.UtcTime.Ticks);
+                }
+            }
+        }
     }
 }

Tests/Engine/Results/LiveTradingResultHandlerTests.cs

@@ -16,14 +16,22 @@
 
 using System;
 using System.Linq;
+using System.Threading;
+using System.Diagnostics;
 using NUnit.Framework;
+using QuantConnect.Orders;
 using QuantConnect.Packets;
+using QuantConnect.Logging;
 using QuantConnect.Securities;
 using QuantConnect.Interfaces;
+using QuantConnect.Data.Market;
+using QuantConnect.Orders.Fees;
+using QuantConnect.Configuration;
 using QuantConnect.Lean.Engine.Results;
 using QuantConnect.Lean.Engine.DataFeeds;
 using QuantConnect.Data.UniverseSelection;
 using QuantConnect.Tests.Engine.DataFeeds;
+using QuantConnect.Brokerages.Backtesting;
 using QuantConnect.Lean.Engine.TransactionHandlers;
 using QuantConnect.Tests.Common.Data.UniverseSelection;
 using QuantConnect.Data.Custom.IconicTypes;
@@ -313,11 +321,141 @@ public void TrimChartsKeepsDailySampleOfStatisticsSeries()
                 .Select(v => (v.Time, v.Open, v.High, v.Low, v.Close)).ToList());
         }
 
+        [Test]
+        public void HoldingsChangeForcesResultsStoreOnceSettled()
+        {
+            var settleDelay = TimeSpan.FromSeconds(2);
+            Config.Set("holdings-changed-store-delay", settleDelay.TotalSeconds.ToStringInvariant());
+            using var api = new Api.Api();
+            using var messaging = new QuantConnect.Messaging.Messaging();
+            var resultHandler = new TestableForceStoreOnSettledHoldingsResultHandler();
+
+            try
+            {
+                var algorithm = new AlgorithmStub(createDataManager: false);
+                var dataManager = new DataManagerStub(new TestDataFeed(), algorithm);
+                algorithm.SubscriptionManager.SetDataManager(dataManager);
+                // live algorithms run on wall-clock time, which the order event times the monitor tracks are in sync with
+                algorithm.SetDateTime(DateTime.UtcNow);
+                // normally initialized by the setup handlers, required for statistics generation
+                algorithm.Settings.TradingDaysPerYear = 365;
+                // crypto markets are always open, so the market orders fill right away regardless of when the test runs
+                var btc = algorithm.AddCrypto("BTCUSD", Resolution.Minute, Market.Coinbase);
+                btc.SetFeeModel(new ConstantFeeModel(0));
+                // the price time must be fresh relative to the order submission time (algorithm utc time) for the fills to happen
+                btc.SetMarketPrice(new TradeBar(algorithm.UtcTime.ConvertFromUtc(btc.Exchange.TimeZone), btc.Symbol, 100, 100, 100, 100, 100));
+                algorithm.PostInitialize();
+
+                var transactionHandler = new BacktestingTransactionHandler();
+                using var brokerage = new BacktestingBrokerage(algorithm);
+                transactionHandler.Initialize(algorithm, brokerage, resultHandler);
+                algorithm.Transactions.SetOrderProcessor(transactionHandler);
+
+                resultHandler.Initialize(new(new LiveNodePacket(), messaging, api, transactionHandler, null));
+                resultHandler.SetAlgorithm(algorithm, 100000);
+                algorithm.SetLocked();
+
+                // places an order for the given quantity, changing the holdings when it fills
+                OrderTicket Trade(decimal quantity, OrderType orderType = OrderType.Market, decimal limitPrice = 0)
+                {
+                    // keep the algorithm clock in sync with wall-clock time like in live trading,
+                    // so the order events are stamped with current times
+                    algorithm.SetDateTime(DateTime.UtcNow);
+                    var ticket = algorithm.Transactions.ProcessRequest(new SubmitOrderRequest(orderType, btc.Symbol.SecurityType,
+                        btc.Symbol, quantity, 0, limitPrice, algorithm.UtcTime, string.Empty));
+                    brokerage.Scan();
+                    return ticket;
+                }
+
+                var expectedStores = 1;
+                // the first update pass always stores because the scheduled store is due right away
+                Assert.IsTrue(resultHandler.WaitForStore(expectedStores, TimeSpan.FromSeconds(30)), "Initial scheduled store did not happen");
+
+                // without holdings changes, no store is forced even after the settle delay elapses
+                Assert.IsFalse(resultHandler.WaitForStore(expectedStores + 1, settleDelay + TimeSpan.FromSeconds(1)), "A store happened without holdings changes");
+
+                // a position is opened. The stopwatch is started before trading so the elapsed time
+                // is measured from no later than the order fill time
+                var stopwatch = Stopwatch.StartNew();
+                Trade(10);
+
+                // no store is forced before the holdings settle
+                Assert.IsFalse(resultHandler.WaitForStore(expectedStores + 1, TimeSpan.FromSeconds(1)), "A store was forced before the holdings settled");
+                // but once they settle, a store is forced without waiting for the scheduled one
+                Assert.IsTrue(resultHandler.WaitForStore(++expectedStores, settleDelay + TimeSpan.FromSeconds(5)), "No store was forced after the opened position settled");
+                Assert.GreaterOrEqual(stopwatch.Elapsed, settleDelay);
+
+                // the position is increased
+                Trade(10);
+                Assert.IsTrue(resultHandler.WaitForStore(++expectedStores, settleDelay + TimeSpan.FromSeconds(5)), "No store was forced after the increased position settled");
+
+                // the position is reduced
+                Trade(-5);
+                Assert.IsTrue(resultHandler.WaitForStore(++expectedStores, settleDelay + TimeSpan.FromSeconds(5)), "No store was forced after the reduced position settled");
+
+                // order events without fills don't force stores: a limit order far from the market price
+                // is submitted and then canceled without ever changing the holdings
+                var ticket = Trade(1, OrderType.Limit, limitPrice: 1);
+                ticket.Cancel();
+                Assert.IsFalse(resultHandler.WaitForStore(expectedStores + 1, settleDelay + TimeSpan.FromSeconds(1)), "A store was forced for order events without fills");
+
+                // the position is liquidated
+                Trade(-15);
+                Assert.IsTrue(resultHandler.WaitForStore(++expectedStores, settleDelay + TimeSpan.FromSeconds(5)), "No store was forced after the liquidation settled");
+
+                // a single settled change forces a single store
+                Assert.IsFalse(resultHandler.WaitForStore(expectedStores + 1, settleDelay + TimeSpan.FromSeconds(1)), "More than one store was forced for a single holdings change");
+            }
+            finally
+            {
+                resultHandler.Exit();
+                Config.Reset();
+            }
+        }
+
         private class TestableLiveTradingResultHandler : LiveTradingResultHandler
         {
             public void PublicTrimCharts(DateTime utcNow) => TrimCharts(utcNow);
         }
 
+        private class TestableForceStoreOnSettledHoldingsResultHandler : LiveTradingResultHandler
+        {
+            private int _storeCount;
+
+            // speed up the update loop so the test can use short settle delays
+            protected override TimeSpan MainUpdateInterval => TimeSpan.FromMilliseconds(100);
+
+            public TestableForceStoreOnSettledHoldingsResultHandler()
+            {
+                // keep the scheduled store out of the way so only forced stores happen after the initial one
+                ChartUpdateInterval = TimeSpan.FromMinutes(10);
+            }
+
+            public bool WaitForStore(int count, TimeSpan timeout)
+            {
+                var start = DateTime.UtcNow;
+                while (DateTime.UtcNow - start < timeout)
+                {
+                    if (Interlocked.CompareExchange(ref _storeCount, 0, 0) >= count)
+                    {
+                        return true;
+                    }
+                    Thread.Sleep(50);
+                }
+                return Interlocked.CompareExchange(ref _storeCount, 0, 0) >= count;
+            }
+
+            protected override void StoreResult(Packet packet)
+            {
+                Interlocked.Increment(ref _storeCount);
+            }
+
+            public override string SaveLogs(string id, List<LogEntry> logs)
+            {
+                return string.Empty;
+            }
+        }
+
         private class TestDataFeed : IDataFeed
         {
             public bool IsActive { get; }