Fix MultimarketPower: bar-0 forward-fill guard and thread-safe trade buffers

Technical/MultiMarketPower.cs

@@ -63,7 +63,7 @@ public class MultiMarketPower : Indicator
 		UseMinimizedModeIfEnabled = true
 	};
 
-	private bool _bigTradesIsReceived;
+	private volatile bool _bigTradesIsReceived;
 	private bool _cumulativeTrades = true;
 	private decimal _delta1;
 	private decimal _delta2;
@@ -93,7 +93,10 @@ public class MultiMarketPower : Indicator
 	private int _sessionBegin;
 
 	private List<MarketDataArg> _ticks = new();
-	private List<CumulativeTrade> _trades = new();
+
+	// Cumulative-trade events received while history is being calculated, in arrival order.
+	// Updates are kept as separate events so they can be replayed with update semantics.
+	private List<(CumulativeTrade Trade, bool IsUpdate)> _trades = new();
 
 	private bool _useFilter1 = true;
 	private bool _useFilter2 = true;
@@ -412,10 +415,12 @@ protected override void OnCalculate(int bar, decimal value)
 
 	protected override void OnFinishRecalculate()
 	{
-		_bigTradesIsReceived = false;
-
-        _ticks.Clear();
-		_trades.Clear();
+		lock (_locker)
+		{
+			_bigTradesIsReceived = false;
+			_ticks.Clear();
+			_trades.Clear();
+		}
 		var totalBars = CurrentBar - 1;
 		_sessionBegin = totalBars;
 		_lastBar = totalBars;
@@ -441,19 +446,20 @@ protected override void OnCumulativeTradesResponse(CumulativeTradesRequest reque
 
 		ClearValues();
 		CalculateHistory(cumulativeTrades);
-
-		_bigTradesIsReceived = true;
 	}
 
 	protected override void OnNewTrade(MarketDataArg trade)
 	{
 		if (CumulativeTrades || ChartInfo is null)
 			return;
 
-		if (!_bigTradesIsReceived)
+		lock (_locker)
 		{
-			_ticks.Add(trade);
-			return;
+			if (!_bigTradesIsReceived)
+			{
+				_ticks.Add(trade);
+				return;
+			}
 		}
 
 		var newBar = _lastBar < CurrentBar - 1;
@@ -469,10 +475,13 @@ protected override void OnCumulativeTrade(CumulativeTrade trade)
 		if (!CumulativeTrades)
 			return;
 
-		if (!_bigTradesIsReceived)
+		lock (_locker)
 		{
-			_trades.Add(trade);
-			return;
+			if (!_bigTradesIsReceived)
+			{
+				_trades.Add((trade, false));
+				return;
+			}
 		}
 
 		var newBar = _lastBar < CurrentBar - 1;
@@ -488,11 +497,13 @@ protected override void OnUpdateCumulativeTrade(CumulativeTrade trade)
 		if (!CumulativeTrades)
 			return;
 
-		if (!_bigTradesIsReceived)
+		lock (_locker)
 		{
-			if (_trades.Count != 0)
-				_trades[^1] = trade;
-			return;
+			if (!_bigTradesIsReceived)
+			{
+				_trades.Add((trade, true));
+				return;
+			}
 		}
 
 		var newBar = _lastBar < CurrentBar - 1;
@@ -512,6 +523,7 @@ private void ClearValues()
 		_bigTradesIsReceived = false;
 		DataSeries.ForEach(x => x.Clear());
 		_delta1 = _delta2 = _delta3 = _delta4 = _delta5 = 0;
+		_lastTrade = null;
 	}
 
 	private void CalculateTrade(CumulativeTrade trade, bool isUpdate, bool newBar)
@@ -607,14 +619,11 @@ private void CalculateHistory(IEnumerable<CumulativeTrade> trades)
 			{
 				orderedTrades = trades.OrderBy(t => t.Time).ToList();
 
-				if (orderedTrades.Count is 0)
-					return;
-
-				for (var i = _sessionBegin; i <= CurrentBar - 1; i++)
-					CalculateBarTrades(orderedTrades, i, ref searchIdx);
-
-				foreach (var trade in _trades)
-					CalculateTrade(trade, false, false);
+				if (orderedTrades.Count > 0)
+				{
+					for (var i = _sessionBegin; i <= CurrentBar - 1; i++)
+						CalculateBarTrades(orderedTrades, i, ref searchIdx);
+				}
 			}
 			else
 			{
@@ -623,16 +632,14 @@ private void CalculateHistory(IEnumerable<CumulativeTrade> trades)
 					.OrderBy(t => t.Time)
 					.ToList();
 
-				if (orderedTicks.Count is 0)
-					return;
-
-				for (var i = _sessionBegin; i <= CurrentBar - 1; i++)
-					CalculateBarTicks(orderedTicks, i, ref searchIdx);
-
-				foreach (var tick in _ticks)
-					CalculateTick(tick);
+				if (orderedTicks.Count > 0)
+				{
+					for (var i = _sessionBegin; i <= CurrentBar - 1; i++)
+						CalculateBarTicks(orderedTicks, i, ref searchIdx);
+				}
 			}
 
+			DrainBufferedData();
 			RedrawChart();
 		}
 		catch (NullReferenceException)
@@ -643,8 +650,58 @@ private void CalculateHistory(IEnumerable<CumulativeTrade> trades)
 		{
 			orderedTrades?.Clear();
 			orderedTicks?.Clear();
-			_trades.Clear();
-			_ticks.Clear();
+		}
+	}
+
+	private void DrainBufferedData()
+	{
+		while (true)
+		{
+			List<(CumulativeTrade Trade, bool IsUpdate)> tradeBatch = null;
+			List<MarketDataArg> tickBatch = null;
+
+			lock (_locker)
+			{
+				if (_trades.Count is 0 && _ticks.Count is 0)
+				{
+					// Nothing left to replay: enable realtime processing before
+					// releasing the lock, so no trade can be buffered afterwards.
+					_bigTradesIsReceived = true;
+					return;
+				}
+
+				if (_trades.Count > 0)
+				{
+					tradeBatch = new List<(CumulativeTrade Trade, bool IsUpdate)>(_trades);
+					_trades.Clear();
+				}
+
+				if (_ticks.Count > 0)
+				{
+					tickBatch = new List<MarketDataArg>(_ticks);
+					_ticks.Clear();
+				}
+			}
+
+			if (tradeBatch is not null)
+			{
+				foreach (var (trade, isUpdate) in tradeBatch)
+				{
+					// An update is only meaningful for the trade processed last. An update for a trade
+					// that is not being tracked (e.g. already contained in the history response) is skipped,
+					// otherwise its full volume would be counted a second time.
+					if (isUpdate && (_lastTrade is null || !_lastTrade.IsEqual(trade)))
+						continue;
+
+					CalculateTrade(trade, isUpdate, false);
+				}
+			}
+
+			if (tickBatch is not null)
+			{
+				foreach (var tick in tickBatch)
+					CalculateTick(tick);
+			}
 		}
 	}