using Predictalytics.Application.Interfaces; using Predictalytics.Domain.Enums; using Predictalytics.Domain.Interfaces; using Predictalytics.Infrastructure.Logging; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; namespace Predictalytics.Worker.Services; /// /// Background service that periodically fetches recent trades for active markets /// to discover new unknown traders. Updates markets starting with the oldest refreshed. /// public class MarketHistoryWorker : BackgroundService { private readonly IServiceProvider _services; private readonly ILogger _logger; private const int CooldownHours = 6; private const int MarketsPerCycle = 10; private const int TradesPerFetch = 100; public MarketHistoryWorker(IServiceProvider services, ILogger logger) { _services = services; _logger = logger; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { _logger.LogInformation("🕵️ MarketHistoryWorker started (cooldown: {Hours}h)", CooldownHours); await Task.Delay(25000, stoppingToken); while (!stoppingToken.IsCancellationRequested) { try { using var scope = _services.CreateScope(); var marketRepo = scope.ServiceProvider.GetRequiredService(); var traderRepo = scope.ServiceProvider.GetRequiredService(); var providers = scope.ServiceProvider.GetRequiredService>(); var rateLimiter = scope.ServiceProvider.GetRequiredService(); var markets = await marketRepo.GetMarketsDueForTradeUpdateAsync(CooldownHours, MarketsPerCycle, stoppingToken); if (markets.Count == 0) { _logger.LogDebug("🕵️ No markets due for trade history update. Sleeping."); } else { _logger.LogInformation("🕵️ Processing {Count} markets for trade history sync", markets.Count); } int totalDiscovered = 0; foreach (var market in markets) { if (stoppingToken.IsCancellationRequested) break; var provider = providers.FirstOrDefault(p => p.Platform == market.Platform && p.IsImplemented); if (provider == null) continue; try { using var platformCtx = PlatformLogContext.Push(provider.PlatformName); await rateLimiter.WaitAsync(market.Platform, stoppingToken); var trades = await provider.GetMarketTradesAsync(market.PlatformMarketId, TradesPerFetch, stoppingToken); // Extract unique wallet addresses from TransientWallet [NotMapped]. // Providers set this during in-memory mapping; it is NOT stored in the DB. var wallets = trades .Select(t => t.TransientWallet) .Where(w => !string.IsNullOrWhiteSpace(w)) .Select(w => w!) .Distinct() .ToList(); foreach (var wallet in wallets) { var existing = await traderRepo.GetByPlatformIdAsync( market.Platform, wallet, stoppingToken); if (existing == null) { var trader = new Domain.Entities.Trader { Platform = market.Platform, PlatformUserId = wallet, DisplayName = wallet.Length > 8 ? wallet[..8] + "..." : wallet, // Will be updated later IsAutoDiscovered = true, CreatedAt = DateTime.UtcNow }; await traderRepo.AddAsync(trader, stoppingToken); totalDiscovered++; _logger.LogInformation("[{Platform}] Discovered trader ({Wallet}) from market trades: {Market}", provider.PlatformName, wallet[..10] + "...", market.Question.Length > 60 ? market.Question[..60] + "..." : market.Question); } } // Update timestamps market.LastTradesUpdatedAt = DateTime.UtcNow; await marketRepo.UpdateAsync(market, stoppingToken); } catch (Exception ex) { _logger.LogError(ex, "Error syncing trade history for market {Market} on {Platform}", market.PlatformMarketId, market.Platform); } } if (totalDiscovered > 0) { _logger.LogInformation("✅ Market history cycle complete. {Count} new traders discovered.", totalDiscovered); } } catch (OperationCanceledException) { break; } catch (Exception ex) { _logger.LogError(ex, "MarketHistoryWorker error"); } // Run every 2 minutes await Task.Delay(TimeSpan.FromMinutes(2), stoppingToken); } _logger.LogInformation("🕵️ MarketHistoryWorker stopped"); } }