diff --git a/src/Predictalytics.Application.Tests/Services/TradeAggregationTests.cs b/src/Predictalytics.Application.Tests/Services/TradeAggregationTests.cs new file mode 100644 index 0000000..6824652 --- /dev/null +++ b/src/Predictalytics.Application.Tests/Services/TradeAggregationTests.cs @@ -0,0 +1,81 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using Predictalytics.Application.Services; +using Predictalytics.Domain.Entities; +using Predictalytics.Domain.Enums; +using Xunit; + +namespace Predictalytics.Application.Tests.Services; + +public class TradeAggregationTests +{ + private static Trade Raw(int outcomeId, TradeSide side, decimal price, decimal size, DateTime at) + => new() + { + MarketOutcomeId = outcomeId, DbMarketId = 10, MarketId = "cond", AssetId = "asset", + Outcome = "Yes", Side = side, Price = price, Size = size, Amount = price * size, ExecutedAt = at + }; + + [Fact] + public void AggregateCompletedHours_SumsCompletedHourWithVwap() + { + var now = new DateTime(2026, 07, 13, 15, 30, 0, DateTimeKind.Utc); + var completedHour = new DateTime(2026, 07, 13, 10, 0, 0, DateTimeKind.Utc); + + var trades = new List + { + Raw(100, TradeSide.Buy, 0.40m, 50m, completedHour.AddMinutes(5)), + Raw(100, TradeSide.Buy, 0.60m, 50m, completedHour.AddMinutes(45)), + }; + + var result = TradeAggregation.AggregateCompletedHours(trades, traderId: 7, nowUtc: now); + + var agg = Assert.Single(result); + Assert.Equal("AGG_7_100_Buy_2026071310", agg.PlatformTradeId); + Assert.Equal(100m, agg.Size); // 50 + 50 + Assert.Equal(50m, agg.Amount); // 0.40*50 + 0.60*50 = 20 + 30 + Assert.Equal(0.50m, agg.Price); // VWAP = 50/100 + Assert.Equal(2, agg.AggregatedCount); + } + + [Fact] + public void AggregateCompletedHours_DefersCurrentHour() + { + // The bug this guards against: the current, still-growing hour must NOT be aggregated, + // because its aggregate row is mutable-by-key while the PnL engine checkpoints by Id. + var now = new DateTime(2026, 07, 13, 15, 30, 0, DateTimeKind.Utc); + + var trades = new List + { + // completed hour -> aggregated + Raw(100, TradeSide.Buy, 0.50m, 40m, new DateTime(2026, 07, 13, 14, 10, 0, DateTimeKind.Utc)), + // current hour (15:xx) -> deferred + Raw(100, TradeSide.Buy, 0.50m, 99m, new DateTime(2026, 07, 13, 15, 05, 0, DateTimeKind.Utc)), + }; + + var result = TradeAggregation.AggregateCompletedHours(trades, traderId: 7, nowUtc: now); + + Assert.Single(result); // only the 14:00 bucket + Assert.EndsWith("_2026071314", result[0].PlatformTradeId); + Assert.Equal(40m, result[0].Size); // current-hour 99 shares NOT included + } + + [Fact] + public void AggregateCompletedHours_SeparatesOutcomeAndSide() + { + var now = new DateTime(2026, 07, 13, 15, 0, 0, DateTimeKind.Utc); + var h = new DateTime(2026, 07, 13, 12, 0, 0, DateTimeKind.Utc); + + var trades = new List + { + Raw(100, TradeSide.Buy, 0.50m, 10m, h), + Raw(100, TradeSide.Sell, 0.50m, 10m, h), + Raw(101, TradeSide.Buy, 0.50m, 10m, h), + }; + + var result = TradeAggregation.AggregateCompletedHours(trades, traderId: 7, nowUtc: now); + + Assert.Equal(3, result.Count); // (100,Buy), (100,Sell), (101,Buy) are distinct buckets + } +} diff --git a/src/Predictalytics.Application/Services/TradeAggregation.cs b/src/Predictalytics.Application/Services/TradeAggregation.cs new file mode 100644 index 0000000..740299f --- /dev/null +++ b/src/Predictalytics.Application/Services/TradeAggregation.cs @@ -0,0 +1,63 @@ +using Predictalytics.Domain.Entities; + +namespace Predictalytics.Application.Services; + +/// +/// Hourly bucketing for the Aggregated ingest tier. +/// +public static class TradeAggregation +{ + /// + /// Aggregates trades into one VWAP row per (MarketOutcomeId, Side, hour) — but ONLY for hours + /// that are already complete (strictly before the current UTC hour). + /// + /// The still-growing current hour is deliberately skipped: aggregate rows are keyed by + /// AGG_{trader}_{outcome}_{side}_{yyyyMMddHH} and upserted via ON DUPLICATE KEY UPDATE, + /// i.e. they are MUTABLE. The PnL engine, however, checkpoints positions by row Id + /// (LastAppliedTradeId) — a bucket that keeps growing after it was already applied would + /// have the same Id and its later growth would be silently skipped (Id ≤ checkpoint). Deferring + /// the current hour until it is complete makes every PERSISTED aggregate effectively immutable, + /// which is exactly what the checkpoint engine requires. + /// + /// The deferred current-hour trades are simply not returned this cycle; because raw aggregate + /// source trades are never stored, they are re-fetched next cycle and aggregated once their hour + /// has closed. + /// + public static List AggregateCompletedHours(IEnumerable trades, int traderId, DateTime nowUtc) + { + var currentHourKey = nowUtc.ToString("yyyyMMddHH"); + var result = new List(); + + foreach (var grp in trades + .Where(t => t.MarketOutcomeId.HasValue) + .GroupBy(t => new { t.MarketOutcomeId, t.Side, Hour = t.ExecutedAt.ToString("yyyyMMddHH") })) + { + if (grp.Key.Hour == currentHourKey) + continue; // defer the still-growing current hour + + var first = grp.First(); + var totalAmount = grp.Sum(t => t.Amount); + var totalSize = grp.Sum(t => t.Size); + var vwap = totalSize > 0 ? totalAmount / totalSize : first.Price; + + result.Add(new Trade + { + PlatformTradeId = $"AGG_{traderId}_{grp.Key.MarketOutcomeId}_{grp.Key.Side}_{grp.Key.Hour}", + TraderId = traderId, + MarketOutcomeId = first.MarketOutcomeId, + DbMarketId = first.DbMarketId, + MarketId = first.MarketId, + AssetId = first.AssetId, + Outcome = first.Outcome, + Side = first.Side, + Price = vwap, + Amount = totalAmount, + Size = totalSize, + ExecutedAt = first.ExecutedAt, + AggregatedCount = grp.Count() + }); + } + + return result; + } +} diff --git a/src/Predictalytics.Worker/Services/PollingWorker.cs b/src/Predictalytics.Worker/Services/PollingWorker.cs index 7462c4a..efa7b47 100644 --- a/src/Predictalytics.Worker/Services/PollingWorker.cs +++ b/src/Predictalytics.Worker/Services/PollingWorker.cs @@ -124,33 +124,11 @@ public class PollingWorker : BackgroundService if (trader.IngestMode == IngestMode.Aggregated && newTrades.Count > 0) { - var aggregated = new List(); - foreach (var grp in newTrades.GroupBy(t => new { t.MarketOutcomeId, t.Side, Hour = t.ExecutedAt.ToString("yyyyMMddHH") })) - { - var first = grp.First(); - var totalAmount = grp.Sum(t => t.Amount); - var totalSize = grp.Sum(t => t.Size); - var vwap = totalSize > 0 ? totalAmount / totalSize : first.Price; - - var aggTrade = new Domain.Entities.Trade - { - PlatformTradeId = $"AGG_{trader.Id}_{grp.Key.MarketOutcomeId}_{grp.Key.Side}_{grp.Key.Hour}", - TraderId = trader.Id, - MarketOutcomeId = first.MarketOutcomeId, - DbMarketId = first.DbMarketId, - MarketId = first.MarketId, - AssetId = first.AssetId, - Outcome = first.Outcome, - Side = first.Side, - Price = vwap, - Amount = totalAmount, - Size = totalSize, - ExecutedAt = first.ExecutedAt, - AggregatedCount = grp.Count() - }; - aggregated.Add(aggTrade); - } - newTrades = aggregated; + // Only completed hours are aggregated+persisted; the current (growing) + // hour is deferred so aggregate rows stay immutable for the checkpoint + // engine. See TradeAggregation.AggregateCompletedHours. + newTrades = Predictalytics.Application.Services.TradeAggregation + .AggregateCompletedHours(newTrades, trader.Id, DateTime.UtcNow); } // ── Persist new trades ── diff --git a/src/Predictalytics.Worker/Services/TradeHistoryWorker.cs b/src/Predictalytics.Worker/Services/TradeHistoryWorker.cs index 7490146..e5a6f86 100644 --- a/src/Predictalytics.Worker/Services/TradeHistoryWorker.cs +++ b/src/Predictalytics.Worker/Services/TradeHistoryWorker.cs @@ -258,34 +258,11 @@ public class TradeHistoryWorker : BackgroundService if (trader.IngestMode == IngestMode.Aggregated && newTrades.Count > 0) { - // Aggregate trades: Bucket (TraderId, MarketOutcomeId, Side, Stunde) - var aggregated = new List(); - foreach (var grp in newTrades.GroupBy(t => new { t.MarketOutcomeId, t.Side, Hour = t.ExecutedAt.ToString("yyyyMMddHH") })) - { - var first = grp.First(); - var totalAmount = grp.Sum(t => t.Amount); - var totalSize = grp.Sum(t => t.Size); - var vwap = totalSize > 0 ? totalAmount / totalSize : first.Price; - - var aggTrade = new Domain.Entities.Trade - { - PlatformTradeId = $"AGG_{trader.Id}_{grp.Key.MarketOutcomeId}_{grp.Key.Side}_{grp.Key.Hour}", - TraderId = trader.Id, - MarketOutcomeId = first.MarketOutcomeId, - DbMarketId = first.DbMarketId, - MarketId = first.MarketId, - AssetId = first.AssetId, - Outcome = first.Outcome, - Side = first.Side, - Price = vwap, - Amount = totalAmount, - Size = totalSize, - ExecutedAt = first.ExecutedAt, - AggregatedCount = grp.Count() - }; - aggregated.Add(aggTrade); - } - newTrades = aggregated; + // Only completed hours are aggregated+persisted; the current (growing) + // hour is deferred so aggregate rows stay immutable for the checkpoint + // engine. See TradeAggregation.AggregateCompletedHours. + newTrades = Predictalytics.Application.Services.TradeAggregation + .AggregateCompletedHours(newTrades, trader.Id, DateTime.UtcNow); } if (isWeeklyBiopsy && newTrades.Count > 0)