using Predictalytics.Domain.Entities; using Predictalytics.Domain.Interfaces; using Predictalytics.Infrastructure.Providers.Polymarket; using Predictalytics.Application.Interfaces; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; namespace Predictalytics.Worker.Services; /// /// Retroactively enriches trades of top/watchlisted traders with high-resolution /// 1-minute price contexts immediately before and after execution. /// Avoids burdening the live PollingWorker. /// public class TradeContextEnrichmentWorker : BackgroundService { private readonly IServiceProvider _services; private readonly ILogger _logger; public TradeContextEnrichmentWorker(IServiceProvider services, ILogger logger) { _services = services; _logger = logger; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { _logger.LogInformation("🧠 TradeContextEnrichmentWorker started"); await Task.Delay(10000, stoppingToken); // Wait for app startup while (!stoppingToken.IsCancellationRequested) { try { using var scope = _services.CreateScope(); var tradeRepo = scope.ServiceProvider.GetRequiredService(); var polymarketClient = scope.ServiceProvider.GetRequiredService(); var rateLimiter = scope.ServiceProvider.GetRequiredService(); var estimator = scope.ServiceProvider.GetRequiredService(); // Fetch a batch of unenriched trades var unenrichedTrades = await tradeRepo.GetTradesForContextEnrichmentAsync(50, stoppingToken); if (unenrichedTrades.Count == 0) { // No work to do, sleep longer await Task.Delay(TimeSpan.FromMinutes(5), stoppingToken); continue; } // Group by AssetId to minimize API calls for the history endpoint (if we still use it) var tradesByAsset = unenrichedTrades.GroupBy(t => t.AssetId).ToList(); _logger.LogInformation("Enriching {TradeCount} trades across {AssetCount} assets...", unenrichedTrades.Count, tradesByAsset.Count); int updatedCount = 0; foreach (var group in tradesByAsset) { if (stoppingToken.IsCancellationRequested) break; var assetId = group.Key; try { // Wait for rate limiter to respect global limits await rateLimiter.WaitAsync(Predictalytics.Domain.Enums.PlatformType.Polymarket, stoppingToken); // We still fetch history for PriceBefore1m var history = await polymarketClient.GetPricesHistoryAsync(assetId, "max", stoppingToken); var orderedHistory = history?.OrderBy(h => h.Timestamp).ToList() ?? new List(); foreach (var trade in group) { var tradeTimeUnix = ((DateTimeOffset)trade.ExecutedAt).ToUnixTimeSeconds(); // Find the closest point BEFORE the trade (approx 1 min before) var prePoint = orderedHistory .LastOrDefault(h => h.Timestamp < tradeTimeUnix); // Find the closest point AFTER the trade (approx 1 min after) - old logic var postPoint = orderedHistory .FirstOrDefault(h => h.Timestamp > tradeTimeUnix); var prePrice = prePoint != null ? (decimal?)prePoint.Price : null; trade.PreTradePrice1m = prePrice; trade.PostTradePrice1m = postPoint != null ? (decimal?)postPoint.Price : null; trade.IsContextEnriched = true; // NEW: Calculate exact follower fill prices from Trade Tape var followerFill10s = await estimator.EstimateFollowerFillPriceAsync(trade, 10, stoppingToken); var followerFill60s = await estimator.EstimateFollowerFillPriceAsync(trade, 60, stoppingToken); // Populate new high-res TradeContext trade.Context = new TradeContext { TradeId = trade.Id, PriceBefore1m = prePrice, PriceAfter1m = trade.PostTradePrice1m, FollowerFillPrice10s = followerFill10s, FollowerFillPrice60s = followerFill60s, EstimatedSlippage = prePrice.HasValue ? Math.Abs(trade.Price - prePrice.Value) : null, EstimatedOrderType = DetermineOrderType(trade, prePrice) }; await tradeRepo.UpdateAsync(trade, stoppingToken); updatedCount++; } } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { _logger.LogInformation("Cancellation requested during enrichment, stopping batch."); break; } catch (Exception ex) { _logger.LogError(ex, "Failed to enrich asset {AssetId}", assetId); // Do NOT mark as enriched on failure, try again later } } _logger.LogInformation("✅ Enriched {UpdatedCount} trades in this cycle.", updatedCount); } catch (Exception ex) { _logger.LogError(ex, "Error in TradeContextEnrichmentWorker loop"); } // Sleep briefly before next batch await Task.Delay(TimeSpan.FromSeconds(30), stoppingToken); } _logger.LogInformation("🧠 TradeContextEnrichmentWorker stopped"); } private static Predictalytics.Domain.Enums.OrderType DetermineOrderType(Trade trade, decimal? priceBefore) { if (priceBefore == null) return Predictalytics.Domain.Enums.OrderType.Unknown; if (trade.Side == Predictalytics.Domain.Enums.TradeSide.Buy) { return trade.Price <= priceBefore.Value ? Predictalytics.Domain.Enums.OrderType.Maker : Predictalytics.Domain.Enums.OrderType.Taker; } else if (trade.Side == Predictalytics.Domain.Enums.TradeSide.Sell) { return trade.Price >= priceBefore.Value ? Predictalytics.Domain.Enums.OrderType.Maker : Predictalytics.Domain.Enums.OrderType.Taker; } return Predictalytics.Domain.Enums.OrderType.Unknown; } }