using System.Text.Json; using Managing.Application.Abstractions.Services; using Managing.Domain.Backtests; using Managing.Domain.Bots; using Managing.Domain.MoneyManagements; using Managing.Domain.Scenarios; using Managing.Domain.Strategies; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using static Managing.Common.Enums; namespace Managing.Application.Workers; /// /// Worker for processing bundle backtest requests /// public class BundleBacktestWorker : BaseWorker { private readonly IServiceProvider _serviceProvider; private readonly IMessengerService _messengerService; private static readonly WorkerType _workerType = WorkerType.BundleBacktest; public BundleBacktestWorker( IServiceProvider serviceProvider, IMessengerService messengerService, ILogger logger) : base( _workerType, logger, TimeSpan.FromMinutes(1), serviceProvider) { _serviceProvider = serviceProvider; _messengerService = messengerService; } protected override async Task Run(CancellationToken cancellationToken) { var maxDegreeOfParallelism = 3; using var semaphore = new SemaphoreSlim(maxDegreeOfParallelism); var processingTasks = new List(); try { // Create a new service scope to get fresh instances of services with scoped DbContext using var scope = _serviceProvider.CreateScope(); var backtester = scope.ServiceProvider.GetRequiredService(); // Get pending bundle backtest requests var pendingRequests = await backtester.GetBundleBacktestRequestsByStatusAsync(BundleBacktestRequestStatus.Pending); foreach (var bundleRequest in pendingRequests) { if (cancellationToken.IsCancellationRequested) break; await semaphore.WaitAsync(cancellationToken); var task = Task.Run(async () => { try { await ProcessBundleRequest(bundleRequest, cancellationToken); } finally { semaphore.Release(); } }, cancellationToken); processingTasks.Add(task); } await Task.WhenAll(processingTasks); await RetryUnfinishedBacktestsInFailedBundles(backtester, cancellationToken); } catch (Exception ex) { _logger.LogError(ex, "Error in BundleBacktestWorker"); throw; } } private async Task ProcessBundleRequest(BundleBacktestRequest bundleRequest, CancellationToken cancellationToken) { // Create a new service scope for this task to avoid DbContext concurrency issues using var scope = _serviceProvider.CreateScope(); var backtester = scope.ServiceProvider.GetRequiredService(); try { _logger.LogInformation("Starting to process bundle backtest request {RequestId}", bundleRequest.RequestId); // Update status to running bundleRequest.Status = BundleBacktestRequestStatus.Running; await backtester.UpdateBundleBacktestRequestAsync(bundleRequest); // Deserialize the backtest requests as strongly-typed objects var backtestRequests = JsonSerializer.Deserialize>( bundleRequest.BacktestRequestsJson); if (backtestRequests == null) { throw new InvalidOperationException("Failed to deserialize backtest requests"); } // Process each backtest request for (int i = 0; i < backtestRequests.Count; i++) { if (cancellationToken.IsCancellationRequested) break; try { var runBacktestRequest = backtestRequests[i]; // Update current backtest being processed bundleRequest.CurrentBacktest = $"Backtest {i + 1} of {backtestRequests.Count}"; await backtester.UpdateBundleBacktestRequestAsync(bundleRequest); // Run the backtest directly with the strongly-typed request var backtestId = await RunSingleBacktest(backtester, runBacktestRequest, bundleRequest, i, cancellationToken); if (!string.IsNullOrEmpty(backtestId)) { bundleRequest.Results.Add(backtestId); } // Update progress bundleRequest.CompletedBacktests++; await backtester.UpdateBundleBacktestRequestAsync(bundleRequest); _logger.LogInformation("Completed backtest {Index} for bundle request {RequestId}", i + 1, bundleRequest.RequestId); } catch (Exception ex) { _logger.LogError(ex, "Error processing backtest {Index} for bundle request {RequestId}", i + 1, bundleRequest.RequestId); bundleRequest.FailedBacktests++; await backtester.UpdateBundleBacktestRequestAsync(bundleRequest); } } // Update final status and send notifications if (bundleRequest.FailedBacktests == 0) { bundleRequest.Status = BundleBacktestRequestStatus.Completed; // Send Telegram message to the user's channelId await NotifyUser(bundleRequest); } else if (bundleRequest.CompletedBacktests == 0) { bundleRequest.Status = BundleBacktestRequestStatus.Failed; bundleRequest.ErrorMessage = "All backtests failed"; } else { bundleRequest.Status = BundleBacktestRequestStatus.Completed; bundleRequest.ErrorMessage = $"{bundleRequest.FailedBacktests} backtests failed"; // Send Telegram message to the user's channelId even with partial failures await NotifyUser(bundleRequest); } bundleRequest.CompletedAt = DateTime.UtcNow; bundleRequest.CurrentBacktest = null; await backtester.UpdateBundleBacktestRequestAsync(bundleRequest); _logger.LogInformation("Completed processing bundle backtest request {RequestId} with status {Status}", bundleRequest.RequestId, bundleRequest.Status); } catch (Exception ex) { _logger.LogError(ex, "Error processing bundle backtest request {RequestId}", bundleRequest.RequestId); bundleRequest.Status = BundleBacktestRequestStatus.Failed; bundleRequest.ErrorMessage = ex.Message; bundleRequest.CompletedAt = DateTime.UtcNow; await backtester.UpdateBundleBacktestRequestAsync(bundleRequest); } } private async Task NotifyUser(BundleBacktestRequest bundleRequest) { if (bundleRequest.User?.TelegramChannel != null) { var message = $"⚠️ Bundle backtest '{bundleRequest.Name}' (ID: {bundleRequest.RequestId}) completed with {bundleRequest.FailedBacktests} failed backtests."; await _messengerService.SendMessage(message, bundleRequest.User.TelegramChannel); } } // Change RunSingleBacktest to accept RunBacktestRequest directly private async Task RunSingleBacktest(IBacktester backtester, RunBacktestRequest runBacktestRequest, BundleBacktestRequest bundleRequest, int index, CancellationToken cancellationToken) { if (runBacktestRequest == null || runBacktestRequest.Config == null) { _logger.LogError("Invalid RunBacktestRequest in bundle (null config)"); return string.Empty; } // Map MoneyManagement MoneyManagement moneyManagement = null; if (!string.IsNullOrEmpty(runBacktestRequest.Config.MoneyManagementName)) { // In worker context, we cannot resolve by name (no user/db), so skip or set null // Optionally, log a warning _logger.LogWarning("MoneyManagementName provided but cannot resolve in worker context: {Name}", (string)runBacktestRequest.Config.MoneyManagementName); } else if (runBacktestRequest.Config.MoneyManagement != null) { var mmReq = runBacktestRequest.Config.MoneyManagement; moneyManagement = new MoneyManagement { Name = mmReq.Name, Timeframe = mmReq.Timeframe, StopLoss = mmReq.StopLoss, TakeProfit = mmReq.TakeProfit, Leverage = mmReq.Leverage }; moneyManagement.FormatPercentage(); } // Map Scenario Scenario scenario = null; if (runBacktestRequest.Config.Scenario != null) { var sReq = runBacktestRequest.Config.Scenario; scenario = new Scenario(sReq.Name, sReq.LoopbackPeriod) { User = null // No user context in worker }; foreach (var indicatorRequest in sReq.Indicators) { var indicator = new Indicator(indicatorRequest.Name, indicatorRequest.Type) { SignalType = indicatorRequest.SignalType, MinimumHistory = indicatorRequest.MinimumHistory, Period = indicatorRequest.Period, FastPeriods = indicatorRequest.FastPeriods, SlowPeriods = indicatorRequest.SlowPeriods, SignalPeriods = indicatorRequest.SignalPeriods, Multiplier = indicatorRequest.Multiplier, SmoothPeriods = indicatorRequest.SmoothPeriods, StochPeriods = indicatorRequest.StochPeriods, CyclePeriods = indicatorRequest.CyclePeriods, User = null // No user context in worker }; scenario.AddIndicator(indicator); } } // Map TradingBotConfig var backtestConfig = new TradingBotConfig { AccountName = runBacktestRequest.Config.AccountName, MoneyManagement = moneyManagement, Ticker = runBacktestRequest.Config.Ticker, ScenarioName = runBacktestRequest.Config.ScenarioName, Scenario = scenario, Timeframe = runBacktestRequest.Config.Timeframe, IsForWatchingOnly = runBacktestRequest.Config.IsForWatchingOnly, BotTradingBalance = runBacktestRequest.Config.BotTradingBalance, IsForBacktest = true, CooldownPeriod = runBacktestRequest.Config.CooldownPeriod, MaxLossStreak = runBacktestRequest.Config.MaxLossStreak, MaxPositionTimeHours = runBacktestRequest.Config.MaxPositionTimeHours, FlipOnlyWhenInProfit = runBacktestRequest.Config.FlipOnlyWhenInProfit, FlipPosition = runBacktestRequest.Config.FlipPosition, Name = $"{bundleRequest.Name} #{index + 1}", CloseEarlyWhenProfitable = runBacktestRequest.Config.CloseEarlyWhenProfitable, UseSynthApi = runBacktestRequest.Config.UseSynthApi, UseForPositionSizing = runBacktestRequest.Config.UseForPositionSizing, UseForSignalFiltering = runBacktestRequest.Config.UseForSignalFiltering, UseForDynamicStopLoss = runBacktestRequest.Config.UseForDynamicStopLoss }; // Run the backtest (no user context) var result = await backtester.RunTradingBotBacktest( backtestConfig, runBacktestRequest.StartDate, runBacktestRequest.EndDate, bundleRequest.User, // No user context in worker true, runBacktestRequest.WithCandles, bundleRequest.RequestId // Use bundleRequestId as requestId for traceability ); _logger.LogInformation("Processed backtest for bundle request {RequestId}", bundleRequest.RequestId); // Assume the backtest is created and you have its ID (e.g., backtest.Id) // Return the backtest ID return result.Id; } private async Task RetryUnfinishedBacktestsInFailedBundles(IBacktester backtester, CancellationToken cancellationToken) { var failedBundles = await backtester.GetBundleBacktestRequestsByStatusAsync(BundleBacktestRequestStatus.Failed); foreach (var failedBundle in failedBundles) { if (cancellationToken.IsCancellationRequested) break; // Use Results property to determine which backtests need to be retried var succeededIds = new HashSet(failedBundle.Results ?? new List()); // Deserialize the original requests var originalRequests = JsonSerializer .Deserialize>(failedBundle.BacktestRequestsJson); if (originalRequests == null) continue; for (int i = failedBundle.CompletedBacktests; i < originalRequests.Count; i++) { var expectedId = /* logic to compute expected backtest id for this request */ string.Empty; // If this backtest was not run or did not succeed, re-run it if (!succeededIds.Contains(expectedId)) { var backtestId = await RunSingleBacktest(backtester, originalRequests[i], failedBundle, i, cancellationToken); if (!string.IsNullOrEmpty(backtestId)) { failedBundle.Results?.Add(backtestId); failedBundle.CompletedBacktests++; await backtester.UpdateBundleBacktestRequestAsync(failedBundle); } } } // If all backtests succeeded, update the bundle status if (failedBundle.CompletedBacktests == originalRequests.Count) { failedBundle.Status = BundleBacktestRequestStatus.Completed; failedBundle.ErrorMessage = null; // Clear any previous error failedBundle.CompletedAt = DateTime.UtcNow; await backtester.UpdateBundleBacktestRequestAsync(failedBundle); // Notify user about successful retry await NotifyUser(failedBundle); } else { _logger.LogWarning("Bundle {RequestId} still has unfinished backtests after retry", failedBundle.RequestId); } } } }