diff --git a/Technical/MultiMarketPower.cs b/Technical/MultiMarketPower.cs index 9a532d6c..aa993da4 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; @@ -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; @@ -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,8 +446,6 @@ protected override void OnCumulativeTradesResponse(CumulativeTradesRequest reque ClearValues(); CalculateHistory(cumulativeTrades); - - _bigTradesIsReceived = true; } protected override void OnNewTrade(MarketDataArg trade) @@ -450,10 +453,13 @@ 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 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 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 trades) { orderedTrades?.Clear(); orderedTicks?.Clear(); - _trades.Clear(); - _ticks.Clear(); + } + } + + private void DrainBufferedData() + { + while (true) + { + List<(CumulativeTrade Trade, bool IsUpdate)> tradeBatch = null; + List 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(_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); + } } }