using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using Microsoft.Extensions.DependencyInjection; using Predictalytics.Infrastructure.Data; using Predictalytics.Application.Interfaces; using Predictalytics.Domain.Interfaces; namespace Predictalytics.Worker.Services; public class TraderAnalyticsWorker : BackgroundService { private readonly IServiceProvider _services; private readonly ILogger _logger; public TraderAnalyticsWorker(IServiceProvider services, ILogger _logger) { _services = services; this._logger = _logger; } protected override async Task ExecuteAsync(CancellationToken ct) { _logger.LogInformation("TraderAnalyticsWorker starting..."); while (!ct.IsCancellationRequested) { try { var processedCount = await RunAnalyticsAsync(ct); if (processedCount == 0) { _logger.LogInformation("TraderAnalyticsWorker sleeping for 1 minute..."); await Task.Delay(TimeSpan.FromMinutes(1), ct); } } catch (Exception ex) { _logger.LogError(ex, "Error in TraderAnalyticsWorker"); await Task.Delay(TimeSpan.FromMinutes(1), ct); } } } private async Task RunAnalyticsAsync(CancellationToken ct) { List traderIds = new List(); Domain.Entities.BackgroundJob? activeJob = null; using (var scope = _services.CreateScope()) { var jobRepo = scope.ServiceProvider.GetRequiredService(); activeJob = await jobRepo.GetNextPendingJobAsync(Predictalytics.Domain.Enums.JobType.TraderAnalysis, ct); if (activeJob != null && activeJob.TraderId.HasValue) { traderIds.Add(activeJob.TraderId.Value); activeJob.Status = Predictalytics.Domain.Enums.JobStatus.InProgress; activeJob.StartedAt = DateTime.UtcNow; await jobRepo.UpdateAsync(activeJob, ct); } else { var db = scope.ServiceProvider.GetRequiredService(); // Bug 5 Fix: Add 30-minute cooldown to prevent CPU looping. var cooldown = DateTime.UtcNow.AddMinutes(-30); // Find traders who have never been analyzed, or whose last analysis was before their latest trade. // Include Resolution-Trigger: Traders with open positions in resolved markets. traderIds = await db.Traders .Where(t => t.LastAnalyzedAt == null || (t.LastAnalyzedAt < cooldown && t.LastTradesUpdatedAt != null && t.LastTradesUpdatedAt > t.LastAnalyzedAt) || (t.LastAnalyzedAt < cooldown && t.Positions.Any(p => p.SharesHeld > 0 && p.MarketOutcome != null && p.MarketOutcome.Market != null && p.MarketOutcome.Market.IsResolved)) ) .OrderBy(t => t.LastAnalyzedAt == null ? 0 : 1) .ThenBy(t => t.LastAnalyzedAt) .Select(t => t.Id) .Take(500) // Limit batch size to prevent long-running loops without save .ToListAsync(ct); } } if (traderIds.Count > 0) { _logger.LogInformation("Found {Count} active or unanalyzed traders to update", traderIds.Count); } foreach (var id in traderIds) { try { using var traderScope = _services.CreateScope(); var pnlEngine = traderScope.ServiceProvider.GetRequiredService(); await pnlEngine.RecalculateTraderPositionsAsync(id, ct); // Run CopytradingEstimator var traderRepo = traderScope.ServiceProvider.GetRequiredService(); var db = traderScope.ServiceProvider.GetRequiredService(); var estimator = traderScope.ServiceProvider.GetRequiredService(); var trader = await traderRepo.GetByIdAsync(id, ct); if (trader != null) { var trades = await db.Trades .Include(t => t.MarketOutcome) .Include(t => t.Context) .Where(t => t.TraderId == id && t.DbMarketId != null) .OrderByDescending(t => t.ExecutedAt) .Take(1000) .ToListAsync(ct); if (trades.Count > 0) { var estScores = await estimator.CalculateScoresAsync(trader, trades, ct); var analyticsObj = trader.Analytics ?? new Predictalytics.Domain.Entities.TraderAnalytics { TraderId = trader.Id }; // Persist advanced copyability and quality scores derived from tape replay analyticsObj.CopytradingScore = estScores.CombinedScore; analyticsObj.CopytradingQualityScore = estScores.QualityScore; analyticsObj.CopytradingCopyabilityScore = estScores.CopyabilityScore; trader.Analytics = analyticsObj; // Only stamp if there were actually trades to analyze trader.LastAnalyzedAt = DateTime.UtcNow; await traderRepo.UpdateAsync(trader, ct); } else if (trader.LastAnalyzedAt == null) { // If it's a completely new trader with no trades, we still don't stamp LastAnalyzedAt // so it remains null until the TradeHistoryWorker pulls trades. } var statsService = traderScope.ServiceProvider.GetService(); if (statsService != null) { statsService.TrackTradersAnalyzed(trader.Platform, 1); } } if (activeJob != null && activeJob.TraderId == id) { using var jobScope = _services.CreateScope(); var updateJobRepo = jobScope.ServiceProvider.GetRequiredService(); activeJob.Status = Predictalytics.Domain.Enums.JobStatus.Completed; activeJob.CompletedAt = DateTime.UtcNow; await updateJobRepo.UpdateAsync(activeJob, ct); } } catch (Exception ex) { _logger.LogError(ex, "Error recalculating positions/PnL for trader {TraderId}", id); if (activeJob != null && activeJob.TraderId == id) { try { using var jobScope = _services.CreateScope(); var updateJobRepo = jobScope.ServiceProvider.GetRequiredService(); activeJob.Status = Predictalytics.Domain.Enums.JobStatus.Failed; activeJob.CompletedAt = DateTime.UtcNow; activeJob.ErrorMessage = ex.Message; await updateJobRepo.UpdateAsync(activeJob, CancellationToken.None); } catch { /* Ignore secondary errors */ } } } } if (traderIds.Count > 0) { _logger.LogInformation("Trader analytics update complete."); } return traderIds.Count; } }