Fix Aggregated-tier mutable-bucket vs checkpoint bug

The Aggregated ingest tier bucketed trades by hour incl. the current, still-growing
hour, then upserted via ON DUPLICATE KEY UPDATE (mutable rows). The PnL engine
checkpoints positions by row Id, so a bucket that keeps growing after being applied
had its later growth silently skipped (Id <= LastAppliedTradeId).

Extract the duplicated aggregation logic from PollingWorker + TradeHistoryWorker into
TradeAggregation.AggregateCompletedHours, which only aggregates COMPLETED hours; the
current hour is deferred (re-fetched next cycle) so every persisted aggregate is
immutable. +3 unit tests.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

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<Trade>
+        {
+            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<Trade>
+        {
+            // 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<Trade>
+        {
+            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
+    }
+}

src/Predictalytics.Application/Services/TradeAggregation.cs

@@ -0,0 +1,63 @@
+using Predictalytics.Domain.Entities;
+
+namespace Predictalytics.Application.Services;
+
+/// <summary>
+/// Hourly bucketing for the <c>Aggregated</c> ingest tier.
+/// </summary>
+public static class TradeAggregation
+{
+    /// <summary>
+    /// 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
+    /// <c>AGG_{trader}_{outcome}_{side}_{yyyyMMddHH}</c> and upserted via ON DUPLICATE KEY UPDATE,
+    /// i.e. they are MUTABLE. The PnL engine, however, checkpoints positions by row Id
+    /// (<c>LastAppliedTradeId</c>) — 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.
+    /// </summary>
+    public static List<Trade> AggregateCompletedHours(IEnumerable<Trade> trades, int traderId, DateTime nowUtc)
+    {
+        var currentHourKey = nowUtc.ToString("yyyyMMddHH");
+        var result = new List<Trade>();
+
+        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;
+    }
+}

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<Domain.Entities.Trade>();
+                            // Only completed hours are aggregated+persisted; the current (growing)
-                            foreach (var grp in newTrades.GroupBy(t => new { t.MarketOutcomeId, t.Side, Hour = t.ExecutedAt.ToString("yyyyMMddHH") }))
+                            // hour is deferred so aggregate rows stay immutable for the checkpoint
-                            {
+                            // engine. See TradeAggregation.AggregateCompletedHours.
-                                var first = grp.First();
+                            newTrades = Predictalytics.Application.Services.TradeAggregation
-                                var totalAmount = grp.Sum(t => t.Amount);
+                                .AggregateCompletedHours(newTrades, trader.Id, DateTime.UtcNow);
-                                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;
                         }
 
                         // ── Persist new trades ──

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)
+                            // Only completed hours are aggregated+persisted; the current (growing)
-                            var aggregated = new List<Domain.Entities.Trade>();
+                            // hour is deferred so aggregate rows stay immutable for the checkpoint
-                            foreach (var grp in newTrades.GroupBy(t => new { t.MarketOutcomeId, t.Side, Hour = t.ExecutedAt.ToString("yyyyMMddHH") }))
+                            // engine. See TradeAggregation.AggregateCompletedHours.
-                            {
+                            newTrades = Predictalytics.Application.Services.TradeAggregation
-                                var first = grp.First();
+                                .AggregateCompletedHours(newTrades, trader.Id, DateTime.UtcNow);
-                                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;
                         }
 
                         if (isWeeklyBiopsy && newTrades.Count > 0)