Fix Estimator scoring and PnL engine cashflow

src/Predictalytics.Worker/Services/TradeRetentionWorker.cs

@@ -71,13 +71,47 @@ public class TradeRetentionWorker : BackgroundService
         var retentionCutoff = utcNow.Date.AddDays(-retentionDays);
         var compactionCutoff = utcNow.Date.AddDays(-compactionDays);
 
-        // 1. Prune Old Trades (C1 & C2)
         // Exclude trades if the trader is on any active Watchlist
         _logger.LogInformation("Pruning trades older than {Cutoff}...", retentionCutoff);
+        
+        var deletedTrades = await db.Trades
+            .Where(t => t.ExecutedAt < retentionCutoff && !t.Trader.WatchlistEntries.Any())
+            .Select(t => new { t.TraderId, t.MarketOutcomeId })
+            .Distinct()
+            .ToListAsync(ct);
+
         var deletedCount = await db.Trades
             .Where(t => t.ExecutedAt < retentionCutoff && !t.Trader.WatchlistEntries.Any())
             .ExecuteDeleteAsync(ct);
 
+        if (deletedCount > 0 && deletedTrades.Any())
+        {
+            var validDeletedTrades = deletedTrades.Where(d => d.MarketOutcomeId.HasValue).ToList();
+            if (validDeletedTrades.Any())
+            {
+                var posIdsToUpdate = await db.TraderPositions
+                    .Where(tp => validDeletedTrades.Select(d => d.TraderId).Contains(tp.TraderId) && 
+                                 validDeletedTrades.Select(d => d.MarketOutcomeId!.Value).Contains(tp.MarketOutcomeId))
+                    .Select(tp => tp.Id)
+                    .ToListAsync(ct);
+
+                // Filter on client side due to EF Core limitation with tuple Contains
+                var actualPosIdsToUpdate = (await db.TraderPositions
+                    .Where(tp => posIdsToUpdate.Contains(tp.Id))
+                    .ToListAsync(ct))
+                    .Where(tp => validDeletedTrades.Any(d => d.TraderId == tp.TraderId && d.MarketOutcomeId == tp.MarketOutcomeId))
+                    .Select(tp => tp.Id)
+                    .ToList();
+
+                if (actualPosIdsToUpdate.Any())
+                {
+                    await db.TraderPositions
+                        .Where(tp => actualPosIdsToUpdate.Contains(tp.Id))
+                        .ExecuteUpdateAsync(s => s.SetProperty(p => p.IsHistoryPruned, true), ct);
+                }
+            }
+        }
+
         _logger.LogInformation("Pruned {Count} old trades from the database.", deletedCount);
 
         // 2. Compact Bot Trades (C3)
@@ -100,14 +134,25 @@ public class TradeRetentionWorker : BackgroundService
         {
             if (ct.IsCancellationRequested) break;
 
+            // Fetch positions to ensure we only compact applied trades and can bump checkpoints
+            var positions = await db.TraderPositions
+                .Where(tp => tp.TraderId == traderId)
+                .ToDictionaryAsync(tp => tp.MarketOutcomeId, ct);
+
             // Load candidate trades to compact (older than compactionCutoff, newer than retentionCutoff)
             var tradesToCompact = await db.Trades
                 .Where(t => t.TraderId == traderId && 
                             t.ExecutedAt >= retentionCutoff && 
                             t.ExecutedAt < compactionCutoff && 
-                            !t.PlatformTradeId.StartsWith("COMPACT_"))
+                            !t.PlatformTradeId.StartsWith("COMPACT_") &&
+                            t.MarketOutcomeId != null)
                 .ToListAsync(ct);
 
+            // Filter strictly to trades that are already applied
+            tradesToCompact = tradesToCompact
+                .Where(t => positions.TryGetValue(t.MarketOutcomeId!.Value, out var pos) && t.Id <= pos.LastAppliedTradeId)
+                .ToList();
+
             if (tradesToCompact.Count == 0) continue;
 
             // Group trades by outcome, date, and side to aggregate
@@ -160,6 +205,19 @@ public class TradeRetentionWorker : BackgroundService
 
                 // Add the compacted trade
                 db.Trades.Add(compactedTrade);
+                
+                // Save immediately so compactedTrade gets an ID
+                await db.SaveChangesAsync(ct);
+
+                // Bump the position checkpoint so it doesn't get double counted
+                if (positions.TryGetValue(outcomeId, out var pos))
+                {
+                    pos.LastAppliedTradeId = Math.Max(pos.LastAppliedTradeId, compactedTrade.Id);
+                    // Mark as pruned so we don't accidentally reset and replay (which would lose the exact intraday timestamps)
+                    pos.IsHistoryPruned = true; 
+                    db.TraderPositions.Update(pos);
+                }
+
                 compactedTradeCount += list.Count - 1;
             }