From 04afe1ebb85a675532605367f5a950d693fb4473 Mon Sep 17 00:00:00 2001 From: AlbertoAmadorBelchistim Date: Sun, 14 Jun 2026 07:04:11 +0200 Subject: [PATCH 1/3] fix(MultiMarketPower): synchronize realtime trade buffers with the existing lock --- Technical/MultiMarketPower.cs | 39 ++++++++++++++++++++++++++--------- 1 file changed, 29 insertions(+), 10 deletions(-) diff --git a/Technical/MultiMarketPower.cs b/Technical/MultiMarketPower.cs index 9a532d6cb..ac8aa2103 100644 --- a/Technical/MultiMarketPower.cs +++ b/Technical/MultiMarketPower.cs @@ -414,8 +414,11 @@ protected override void OnFinishRecalculate() { _bigTradesIsReceived = false; - _ticks.Clear(); - _trades.Clear(); + lock (_locker) + { + _ticks.Clear(); + _trades.Clear(); + } var totalBars = CurrentBar - 1; _sessionBegin = totalBars; _lastBar = totalBars; @@ -452,7 +455,8 @@ protected override void OnNewTrade(MarketDataArg trade) if (!_bigTradesIsReceived) { - _ticks.Add(trade); + lock (_locker) + _ticks.Add(trade); return; } @@ -471,7 +475,8 @@ protected override void OnCumulativeTrade(CumulativeTrade trade) if (!_bigTradesIsReceived) { - _trades.Add(trade); + lock (_locker) + _trades.Add(trade); return; } @@ -490,8 +495,11 @@ protected override void OnUpdateCumulativeTrade(CumulativeTrade trade) if (!_bigTradesIsReceived) { - if (_trades.Count != 0) - _trades[^1] = trade; + lock (_locker) + { + if (_trades.Count != 0) + _trades[^1] = trade; + } return; } @@ -613,7 +621,11 @@ private void CalculateHistory(IEnumerable trades) for (var i = _sessionBegin; i <= CurrentBar - 1; i++) CalculateBarTrades(orderedTrades, i, ref searchIdx); - foreach (var trade in _trades) + List bufferedTrades; + lock (_locker) + bufferedTrades = new List(_trades); + + foreach (var trade in bufferedTrades) CalculateTrade(trade, false, false); } else @@ -629,7 +641,11 @@ private void CalculateHistory(IEnumerable trades) for (var i = _sessionBegin; i <= CurrentBar - 1; i++) CalculateBarTicks(orderedTicks, i, ref searchIdx); - foreach (var tick in _ticks) + List bufferedTicks; + lock (_locker) + bufferedTicks = new List(_ticks); + + foreach (var tick in bufferedTicks) CalculateTick(tick); } @@ -643,8 +659,11 @@ private void CalculateHistory(IEnumerable trades) { orderedTrades?.Clear(); orderedTicks?.Clear(); - _trades.Clear(); - _ticks.Clear(); + lock (_locker) + { + _trades.Clear(); + _ticks.Clear(); + } } } From 33b00b8e8df21e4392c18348427f49a22967f113 Mon Sep 17 00:00:00 2001 From: AlbertoAmadorBelchistim Date: Sat, 4 Jul 2026 07:52:32 +0200 Subject: [PATCH 2/3] fix(MultiMarketPower): drain buffered trades atomically before enabling realtime The previous version snapshotted _ticks/_trades under the lock, but _bigTradesIsReceived was only set after CalculateHistory returned. Trades arriving between the snapshot and the flag flip were appended to the original buffers and then discarded by the finally-block clear. Close the handoff window: - Trade handlers now check the flag inside the lock: a trade is either buffered under the lock or processed as realtime, never dropped. - CalculateHistory no longer clears the buffers in finally. After the historical replay it calls DrainBufferedData, which repeatedly swaps out the buffers under the lock and replays each batch, and flips _bigTradesIsReceived inside the lock only once both buffers are empty, so no trade can be buffered after the flip. - The flag becomes volatile (it is still read lock-free in OnCalculate) and is reset inside the lock in OnFinishRecalculate. - Zero-history responses now also drain buffered live trades instead of returning early. --- Technical/MultiMarketPower.cs | 107 +++++++++++++++++++++------------- 1 file changed, 67 insertions(+), 40 deletions(-) diff --git a/Technical/MultiMarketPower.cs b/Technical/MultiMarketPower.cs index ac8aa2103..af17b74c0 100644 --- a/Technical/MultiMarketPower.cs +++ b/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; @@ -412,10 +412,9 @@ protected override void OnCalculate(int bar, decimal value) protected override void OnFinishRecalculate() { - _bigTradesIsReceived = false; - lock (_locker) { + _bigTradesIsReceived = false; _ticks.Clear(); _trades.Clear(); } @@ -444,8 +443,6 @@ protected override void OnCumulativeTradesResponse(CumulativeTradesRequest reque ClearValues(); CalculateHistory(cumulativeTrades); - - _bigTradesIsReceived = true; } protected override void OnNewTrade(MarketDataArg trade) @@ -453,11 +450,13 @@ protected override void OnNewTrade(MarketDataArg trade) if (CumulativeTrades || ChartInfo is null) return; - if (!_bigTradesIsReceived) + lock (_locker) { - lock (_locker) + if (!_bigTradesIsReceived) + { _ticks.Add(trade); - return; + return; + } } var newBar = _lastBar < CurrentBar - 1; @@ -473,11 +472,13 @@ protected override void OnCumulativeTrade(CumulativeTrade trade) if (!CumulativeTrades) return; - if (!_bigTradesIsReceived) + lock (_locker) { - lock (_locker) + if (!_bigTradesIsReceived) + { _trades.Add(trade); - return; + return; + } } var newBar = _lastBar < CurrentBar - 1; @@ -493,14 +494,14 @@ protected override void OnUpdateCumulativeTrade(CumulativeTrade trade) if (!CumulativeTrades) return; - if (!_bigTradesIsReceived) + lock (_locker) { - lock (_locker) + if (!_bigTradesIsReceived) { if (_trades.Count != 0) _trades[^1] = trade; + return; } - return; } var newBar = _lastBar < CurrentBar - 1; @@ -615,18 +616,11 @@ private void CalculateHistory(IEnumerable 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); - - List bufferedTrades; - lock (_locker) - bufferedTrades = new List(_trades); - - foreach (var trade in bufferedTrades) - CalculateTrade(trade, false, false); + if (orderedTrades.Count > 0) + { + for (var i = _sessionBegin; i <= CurrentBar - 1; i++) + CalculateBarTrades(orderedTrades, i, ref searchIdx); + } } else { @@ -635,20 +629,14 @@ private void CalculateHistory(IEnumerable 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); - - List bufferedTicks; - lock (_locker) - bufferedTicks = new List(_ticks); - - foreach (var tick in bufferedTicks) - CalculateTick(tick); + if (orderedTicks.Count > 0) + { + for (var i = _sessionBegin; i <= CurrentBar - 1; i++) + CalculateBarTicks(orderedTicks, i, ref searchIdx); + } } + DrainBufferedData(); RedrawChart(); } catch (NullReferenceException) @@ -659,10 +647,49 @@ private void CalculateHistory(IEnumerable trades) { orderedTrades?.Clear(); orderedTicks?.Clear(); + } + } + + private void DrainBufferedData() + { + while (true) + { + List tradeBatch = null; + List tickBatch = null; + lock (_locker) { - _trades.Clear(); - _ticks.Clear(); + 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(_trades); + _trades.Clear(); + } + + if (_ticks.Count > 0) + { + tickBatch = new List(_ticks); + _ticks.Clear(); + } + } + + if (tradeBatch is not null) + { + foreach (var trade in tradeBatch) + CalculateTrade(trade, false, false); + } + + if (tickBatch is not null) + { + foreach (var tick in tickBatch) + CalculateTick(tick); } } } From 3d7d0f9a1e76dc50050aeb3f31d940b48dfbed67 Mon Sep 17 00:00:00 2001 From: AlbertoAmadorBelchistim Date: Sat, 19 Sep 2026 21:57:18 +0200 Subject: [PATCH 3/3] fix(MultiMarketPower): keep cumulative-trade updates across replay batches Buffered updates used to overwrite the last buffered trade. When an update arrived after DrainBufferedData had detached the batch, the buffer was empty and the update was dropped, so the trade was replayed with an outdated volume. Buffer new trades and updates as ordered events and replay each one through CalculateTrade with its own update flag. An update for a trade that is not the last processed one (for example, a trade already contained in the history response) is skipped as before, so its volume is not counted twice. _lastTrade is reset together with the accumulated deltas. --- Technical/MultiMarketPower.cs | 27 +++++++++++++++++++-------- 1 file changed, 19 insertions(+), 8 deletions(-) diff --git a/Technical/MultiMarketPower.cs b/Technical/MultiMarketPower.cs index af17b74c0..aa993da4e 100644 --- a/Technical/MultiMarketPower.cs +++ b/Technical/MultiMarketPower.cs @@ -93,7 +93,10 @@ public class MultiMarketPower : Indicator private int _sessionBegin; private List _ticks = new(); - private List _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; @@ -476,7 +479,7 @@ protected override void OnCumulativeTrade(CumulativeTrade trade) { if (!_bigTradesIsReceived) { - _trades.Add(trade); + _trades.Add((trade, false)); return; } } @@ -498,8 +501,7 @@ protected override void OnUpdateCumulativeTrade(CumulativeTrade trade) { if (!_bigTradesIsReceived) { - if (_trades.Count != 0) - _trades[^1] = trade; + _trades.Add((trade, true)); return; } } @@ -521,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) @@ -654,7 +657,7 @@ private void DrainBufferedData() { while (true) { - List tradeBatch = null; + List<(CumulativeTrade Trade, bool IsUpdate)> tradeBatch = null; List tickBatch = null; lock (_locker) @@ -669,7 +672,7 @@ private void DrainBufferedData() if (_trades.Count > 0) { - tradeBatch = new List(_trades); + tradeBatch = new List<(CumulativeTrade Trade, bool IsUpdate)>(_trades); _trades.Clear(); } @@ -682,8 +685,16 @@ private void DrainBufferedData() if (tradeBatch is not null) { - foreach (var trade in tradeBatch) - CalculateTrade(trade, false, false); + 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)