using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; using System.Threading; using System.Threading.Tasks; using FinlyticCore.Dtos.TechnicalAnalysis; using FinlyticCore.Services; using FinlyticTechnicals.Timeframe; using FinlyticTechnicals.Util; namespace FinlyticTechnicals.Services; public interface IMultiTimeframeCandleAggregator { /// /// Initializes historical ring buffers for an ISIN with Yahoo/database candles. /// void InitializeHistory(string isin, string timeframe, IEnumerable candles); /// /// Processes an incoming clean tick and updates 1m, 5m, 15m, 1h, and 1d candles. /// void ProcessTick(CleanLiveTick tick); /// /// Gets a snapshot of the ring buffer for an ISIN and timeframe. /// IReadOnlyList GetCandles(string isin, string timeframe); /// /// Gets all multi-timeframe candles (1m, 5m, 15m, 1h, 1d) as a dictionary. /// Dictionary> GetAllTimeframes(string isin); /// /// Event triggered when a timeframe bar completes. /// event Action? OnCandleClosed; } public class MultiTimeframeCandleAggregator : IMultiTimeframeCandleAggregator { private readonly IFinlyticLogger _logger; private readonly ConcurrentDictionary>> _buffers = new(StringComparer.OrdinalIgnoreCase); private readonly ConcurrentDictionary _current1mCandles = new(StringComparer.OrdinalIgnoreCase); private readonly object _aggregationLock = new(); public event Action? OnCandleClosed; public MultiTimeframeCandleAggregator(IFinlyticLogger logger) { _logger = logger; } public void InitializeHistory(string isin, string timeframe, IEnumerable candles) { if (string.IsNullOrWhiteSpace(isin) || string.IsNullOrWhiteSpace(timeframe)) return; var cleanIsin = isin.Trim().ToUpperInvariant(); var cleanTf = timeframe.Trim().ToLowerInvariant(); var isinBuffers = _buffers.GetOrAdd(cleanIsin, _ => new ConcurrentDictionary>(StringComparer.OrdinalIgnoreCase)); var ringBuffer = isinBuffers.GetOrAdd(cleanTf, _ => new CircularRingBuffer(500)); var ordered = candles .Where(c => c.Close > 0m) .OrderBy(c => c.Timestamp) .ToList(); ringBuffer.LoadBulk(ordered); } public void ProcessTick(CleanLiveTick tick) { if (tick == null || string.IsNullOrWhiteSpace(tick.Isin)) return; var isin = tick.Isin.Trim().ToUpperInvariant(); var tickTime = tick.TimestampUtc; var minuteBoundary = new DateTime(tickTime.Year, tickTime.Month, tickTime.Day, tickTime.Hour, tickTime.Minute, 0, DateTimeKind.Utc); lock (_aggregationLock) { var isinBuffers = _buffers.GetOrAdd(isin, _ => new ConcurrentDictionary>(StringComparer.OrdinalIgnoreCase)); var ringBuffer1m = isinBuffers.GetOrAdd("1m", _ => new CircularRingBuffer(500)); if (_current1mCandles.TryGetValue(isin, out var current1m)) { if (current1m.Timestamp == minuteBoundary) { // Update current open 1m bar var updated = current1m with { High = Math.Max(current1m.High, tick.MidPrice), Low = Math.Min(current1m.Low, tick.MidPrice), Close = tick.MidPrice, Volume = current1m.Volume + 1, Bid = tick.Bid, Ask = tick.Ask }; _current1mCandles[isin] = updated; ringBuffer1m.UpdateLast(updated); } else if (minuteBoundary > current1m.Timestamp) { // 1. Close current 1m bar ringBuffer1m.UpdateLast(current1m); OnCandleClosed?.Invoke(isin, "1m", current1m); // 2. Reconnect-Lückenbehandlung (Gap Handling) // If multiple minutes passed without ticks (e.g. disconnect), fill gaps flatly with Volume = 0 var gapStart = current1m.Timestamp.AddMinutes(1); var lastClose = current1m.Close; while (gapStart < minuteBoundary) { var flatBar = new CandleDto( Timestamp: gapStart, Open: lastClose, High: lastClose, Low: lastClose, Close: lastClose, Volume: 0, Bid: tick.Bid, Ask: tick.Ask ); ringBuffer1m.Add(flatBar); OnCandleClosed?.Invoke(isin, "1m", flatBar); gapStart = gapStart.AddMinutes(1); } // 3. Start new 1m bar var new1m = new CandleDto( Timestamp: minuteBoundary, Open: tick.MidPrice, High: tick.MidPrice, Low: tick.MidPrice, Close: tick.MidPrice, Volume: 1, Bid: tick.Bid, Ask: tick.Ask ); _current1mCandles[isin] = new1m; ringBuffer1m.Add(new1m); // 4. Update higher timeframes (5m, 15m, 1h, 1d) RebuildHigherTimeframes(isin, isinBuffers, ringBuffer1m); } } else { // First tick for this ISIN var new1m = new CandleDto( Timestamp: minuteBoundary, Open: tick.MidPrice, High: tick.MidPrice, Low: tick.MidPrice, Close: tick.MidPrice, Volume: 1, Bid: tick.Bid, Ask: tick.Ask ); _current1mCandles[isin] = new1m; ringBuffer1m.Add(new1m); } } } private void RebuildHigherTimeframes(string isin, ConcurrentDictionary> isinBuffers, CircularRingBuffer ringBuffer1m) { var snapshot1m = ringBuffer1m.ToArray(); if (snapshot1m.Length == 0) return; // Build 5m candles AggregatePeriod(isin, isinBuffers, snapshot1m, "5m", 5); // Build 15m candles AggregatePeriod(isin, isinBuffers, snapshot1m, "15m", 15); // Build 1h candles AggregatePeriod(isin, isinBuffers, snapshot1m, "1h", 60); // Build 1d candles AggregateDaily(isin, isinBuffers, snapshot1m); } private void AggregatePeriod(string isin, ConcurrentDictionary> isinBuffers, CandleDto[] candles1m, string tfName, int minutes) { var targetBuffer = isinBuffers.GetOrAdd(tfName, _ => new CircularRingBuffer(500)); targetBuffer.LoadBulk(FinlyticTechnicals.Indicators.CandleResampler.Resample(candles1m, minutes)); } private void AggregateDaily(string isin, ConcurrentDictionary> isinBuffers, CandleDto[] candles1m) { var targetBuffer = isinBuffers.GetOrAdd("1d", _ => new CircularRingBuffer(500)); var aggregated = FinlyticTechnicals.Indicators.CandleResampler.Resample(candles1m, 1440); // If daily buffer already has deep Yahoo history, stitch today's aggregated bar onto the end if (targetBuffer.Count > 0 && aggregated.Count > 0) { var today = aggregated.Last(); var lastHistory = targetBuffer.GetLast(); if (lastHistory != null && lastHistory.Timestamp.Date == today.Timestamp.Date) { targetBuffer.UpdateLast(today); } else { targetBuffer.Add(today); } } else if (aggregated.Count > 0) { targetBuffer.LoadBulk(aggregated); } } public IReadOnlyList GetCandles(string isin, string timeframe) { if (string.IsNullOrWhiteSpace(isin)) return []; var cleanIsin = isin.Trim().ToUpperInvariant(); var cleanTf = (timeframe ?? "15m").Trim().ToLowerInvariant(); if (_buffers.TryGetValue(cleanIsin, out var isinBuffers) && isinBuffers.TryGetValue(cleanTf, out var ringBuffer)) { return ringBuffer.ToArray(); } return []; } public Dictionary> GetAllTimeframes(string isin) { var result = new Dictionary>(StringComparer.OrdinalIgnoreCase); if (string.IsNullOrWhiteSpace(isin)) return result; var cleanIsin = isin.Trim().ToUpperInvariant(); if (_buffers.TryGetValue(cleanIsin, out var isinBuffers)) { foreach (var kvp in isinBuffers) { result[kvp.Key] = kvp.Value.ToArray(); } } return result; } }