Bundle from worker to grain

This commit is contained in:
2025-09-15 12:56:59 +07:00
parent 77e6ce0789
commit 63bc7bbe59
19 changed files with 2112 additions and 79 deletions

View File

@@ -95,7 +95,7 @@ namespace Managing.Application.Backtests
try
{
var candles = GetCandles(config.Ticker, config.Timeframe, startDate, endDate);
var candles = await GetCandles(config.Ticker, config.Timeframe, startDate, endDate);
return await RunBacktestWithCandles(config, candles, user, save, withCandles, requestId, metadata);
}
catch (Exception ex)
@@ -197,11 +197,11 @@ namespace Managing.Application.Backtests
return await _accountService.GetAccountByAccountName(config.AccountName, false, false);
}
private HashSet<Candle> GetCandles(Ticker ticker, Timeframe timeframe,
private async Task<HashSet<Candle>> GetCandles(Ticker ticker, Timeframe timeframe,
DateTime startDate, DateTime endDate)
{
var candles = _exchangeService.GetCandlesInflux(TradingExchanges.Evm, ticker,
startDate, timeframe, endDate).Result;
var candles = await _exchangeService.GetCandlesInflux(TradingExchanges.Evm, ticker,
startDate, timeframe, endDate);
if (candles == null || candles.Count == 0)
throw new Exception($"No candles for {ticker} on {timeframe} timeframe");
@@ -277,19 +277,19 @@ namespace Managing.Application.Backtests
return backtests;
}
public IEnumerable<Backtest> GetBacktestsByRequestId(string requestId)
public IEnumerable<Backtest> GetBacktestsByRequestId(Guid requestId)
{
var backtests = _backtestRepository.GetBacktestsByRequestId(requestId).ToList();
return backtests;
}
public async Task<IEnumerable<Backtest>> GetBacktestsByRequestIdAsync(string requestId)
public async Task<IEnumerable<Backtest>> GetBacktestsByRequestIdAsync(Guid requestId)
{
var backtests = await _backtestRepository.GetBacktestsByRequestIdAsync(requestId);
return backtests;
}
public (IEnumerable<LightBacktest> Backtests, int TotalCount) GetBacktestsByRequestIdPaginated(string requestId,
public (IEnumerable<LightBacktest> Backtests, int TotalCount) GetBacktestsByRequestIdPaginated(Guid requestId,
int page, int pageSize, string sortBy = "score", string sortOrder = "desc")
{
var (backtests, totalCount) =
@@ -298,7 +298,7 @@ namespace Managing.Application.Backtests
}
public async Task<(IEnumerable<LightBacktest> Backtests, int TotalCount)> GetBacktestsByRequestIdPaginatedAsync(
string requestId, int page, int pageSize, string sortBy = "score", string sortOrder = "desc")
Guid requestId, int page, int pageSize, string sortBy = "score", string sortOrder = "desc")
{
var (backtests, totalCount) =
await _backtestRepository.GetBacktestsByRequestIdPaginatedAsync(requestId, page, pageSize, sortBy,
@@ -384,7 +384,7 @@ namespace Managing.Application.Backtests
}
}
public async Task<bool> DeleteBacktestsByRequestIdAsync(string requestId)
public async Task<bool> DeleteBacktestsByRequestIdAsync(Guid requestId)
{
try
{
@@ -418,6 +418,17 @@ namespace Managing.Application.Backtests
public void InsertBundleBacktestRequestForUser(User user, BundleBacktestRequest bundleRequest)
{
_backtestRepository.InsertBundleBacktestRequestForUser(user, bundleRequest);
// Trigger the BundleBacktestGrain to process this request
TriggerBundleBacktestGrain(bundleRequest.RequestId);
}
public async Task InsertBundleBacktestRequestForUserAsync(User user, BundleBacktestRequest bundleRequest)
{
await _backtestRepository.InsertBundleBacktestRequestForUserAsync(user, bundleRequest);
// Trigger the BundleBacktestGrain to process this request
await TriggerBundleBacktestGrainAsync(bundleRequest.RequestId);
}
public IEnumerable<BundleBacktestRequest> GetBundleBacktestRequestsByUser(User user)
@@ -430,12 +441,12 @@ namespace Managing.Application.Backtests
return await _backtestRepository.GetBundleBacktestRequestsByUserAsync(user);
}
public BundleBacktestRequest? GetBundleBacktestRequestByIdForUser(User user, string id)
public BundleBacktestRequest? GetBundleBacktestRequestByIdForUser(User user, Guid id)
{
return _backtestRepository.GetBundleBacktestRequestByIdForUser(user, id);
}
public async Task<BundleBacktestRequest?> GetBundleBacktestRequestByIdForUserAsync(User user, string id)
public async Task<BundleBacktestRequest?> GetBundleBacktestRequestByIdForUserAsync(User user, Guid id)
{
return await _backtestRepository.GetBundleBacktestRequestByIdForUserAsync(user, id);
}
@@ -450,12 +461,12 @@ namespace Managing.Application.Backtests
await _backtestRepository.UpdateBundleBacktestRequestAsync(bundleRequest);
}
public void DeleteBundleBacktestRequestByIdForUser(User user, string id)
public void DeleteBundleBacktestRequestByIdForUser(User user, Guid id)
{
_backtestRepository.DeleteBundleBacktestRequestByIdForUser(user, id);
}
public async Task DeleteBundleBacktestRequestByIdForUserAsync(User user, string id)
public async Task DeleteBundleBacktestRequestByIdForUserAsync(User user, Guid id)
{
await _backtestRepository.DeleteBundleBacktestRequestByIdForUserAsync(user, id);
}
@@ -480,5 +491,65 @@ namespace Managing.Application.Backtests
if (string.IsNullOrWhiteSpace(requestId) || response == null) return;
await _hubContext.Clients.Group($"bundle-{requestId}").SendAsync("BundleBacktestUpdate", response);
}
/// <summary>
/// Triggers the BundleBacktestGrain to process a bundle request synchronously (fire and forget)
/// </summary>
private void TriggerBundleBacktestGrain(Guid bundleRequestId)
{
try
{
var bundleBacktestGrain = _grainFactory.GetGrain<IBundleBacktestGrain>(bundleRequestId);
// Fire and forget - don't await
_ = Task.Run(async () =>
{
try
{
await bundleBacktestGrain.ProcessBundleRequestAsync();
}
catch (Exception ex)
{
_logger.LogError(ex, "Error triggering BundleBacktestGrain for request {RequestId}",
bundleRequestId);
}
});
}
catch (Exception ex)
{
_logger.LogError(ex, "Error in TriggerBundleBacktestGrain for request {RequestId}", bundleRequestId);
}
}
/// <summary>
/// Triggers the BundleBacktestGrain to process a bundle request asynchronously
/// </summary>
private Task TriggerBundleBacktestGrainAsync(Guid bundleRequestId)
{
try
{
var bundleBacktestGrain = _grainFactory.GetGrain<IBundleBacktestGrain>(bundleRequestId);
// Fire and forget - don't await the actual processing
return Task.Run(async () =>
{
try
{
await bundleBacktestGrain.ProcessBundleRequestAsync();
}
catch (Exception ex)
{
_logger.LogError(ex, "Error triggering BundleBacktestGrain for request {RequestId}",
bundleRequestId);
}
});
}
catch (Exception ex)
{
_logger.LogError(ex, "Error in TriggerBundleBacktestGrainAsync for request {RequestId}",
bundleRequestId);
return Task.CompletedTask;
}
}
}
}

View File

@@ -131,7 +131,7 @@ public class BacktestTradingBotGrain : Grain, IBacktestTradingBotGrain
var scoringResult = BacktestScorer.CalculateDetailedScore(scoringParams);
// Generate requestId if not provided
var finalRequestId = requestId ?? Guid.NewGuid().ToString();
var finalRequestId = requestId != null ? Guid.Parse(requestId) : Guid.NewGuid();
// Create backtest result with conditional candles and indicators values
var result = new Backtest(config, tradingBot.Positions, tradingBot.Signals,

View File

@@ -0,0 +1,411 @@
using System.Text.Json;
using Managing.Application.Abstractions.Grains;
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 Orleans.Concurrency;
namespace Managing.Application.Grains;
/// <summary>
/// Stateless worker grain for processing bundle backtest requests
/// Uses the bundle request ID as the primary key (Guid)
/// Implements IRemindable for automatic retry of failed bundles
/// </summary>
[StatelessWorker]
public class BundleBacktestGrain : Grain, IBundleBacktestGrain, IRemindable
{
private readonly ILogger<BundleBacktestGrain> _logger;
private readonly IServiceScopeFactory _scopeFactory;
// Reminder configuration
private const string RETRY_REMINDER_NAME = "BundleBacktestRetry";
private static readonly TimeSpan RETRY_INTERVAL = TimeSpan.FromMinutes(30);
public BundleBacktestGrain(
ILogger<BundleBacktestGrain> logger,
IServiceScopeFactory scopeFactory)
{
_logger = logger;
_scopeFactory = scopeFactory;
}
public async Task ProcessBundleRequestAsync()
{
// Get the RequestId from the grain's primary key
var bundleRequestId = this.GetPrimaryKey();
try
{
// Create a new service scope to get fresh instances of services with scoped DbContext
using var scope = _scopeFactory.CreateScope();
var backtester = scope.ServiceProvider.GetRequiredService<IBacktester>();
var messengerService = scope.ServiceProvider.GetRequiredService<IMessengerService>();
// Get the specific bundle request by ID
var bundleRequest = await GetBundleRequestById(backtester, bundleRequestId);
if (bundleRequest == null)
{
_logger.LogError("Bundle request {RequestId} not found", bundleRequestId);
return;
}
// Process only this specific bundle request
await ProcessBundleRequest(bundleRequest, backtester, messengerService);
}
catch (Exception ex)
{
_logger.LogError(ex, "Error in BundleBacktestGrain for request {RequestId}", bundleRequestId);
throw;
}
}
private async Task<BundleBacktestRequest> GetBundleRequestById(IBacktester backtester, Guid bundleRequestId)
{
try
{
// Get pending and failed bundle backtest requests for retry capability
var pendingRequests = await backtester.GetBundleBacktestRequestsByStatusAsync(BundleBacktestRequestStatus.Pending);
var failedRequests = await backtester.GetBundleBacktestRequestsByStatusAsync(BundleBacktestRequestStatus.Failed);
var allRequests = pendingRequests.Concat(failedRequests);
return allRequests.FirstOrDefault(r => r.RequestId == bundleRequestId);
}
catch (Exception ex)
{
_logger.LogError(ex, "Failed to get bundle request {RequestId}", bundleRequestId);
return null;
}
}
private async Task ProcessBundleRequest(
BundleBacktestRequest bundleRequest,
IBacktester backtester,
IMessengerService messengerService)
{
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<List<RunBacktestRequest>>(bundleRequest.BacktestRequestsJson);
if (backtestRequests == null)
{
throw new InvalidOperationException("Failed to deserialize backtest requests");
}
// Process each backtest request sequentially
for (int i = 0; i < backtestRequests.Count; i++)
{
await ProcessSingleBacktest(backtester, backtestRequests[i], bundleRequest, i);
}
// Update final status and send notifications
await UpdateFinalStatus(bundleRequest, backtester, messengerService);
_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);
await HandleBundleRequestError(bundleRequest, backtester, ex);
}
}
private async Task ProcessSingleBacktest(
IBacktester backtester,
RunBacktestRequest runBacktestRequest,
BundleBacktestRequest bundleRequest,
int index)
{
try
{
// Get total count from deserialized requests instead of string splitting
var backtestRequests = JsonSerializer.Deserialize<List<RunBacktestRequest>>(bundleRequest.BacktestRequestsJson);
var totalCount = backtestRequests?.Count ?? 0;
// Update current backtest being processed
bundleRequest.CurrentBacktest = $"Backtest {index + 1} of {totalCount}";
await backtester.UpdateBundleBacktestRequestAsync(bundleRequest);
// Run the backtest directly with the strongly-typed request
var backtestId = await RunSingleBacktest(backtester, runBacktestRequest, bundleRequest, index);
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}",
index + 1, bundleRequest.RequestId);
}
catch (Exception ex)
{
_logger.LogError(ex, "Error processing backtest {Index} for bundle request {RequestId}",
index + 1, bundleRequest.RequestId);
bundleRequest.FailedBacktests++;
await backtester.UpdateBundleBacktestRequestAsync(bundleRequest);
}
}
private async Task<string> RunSingleBacktest(
IBacktester backtester,
RunBacktestRequest runBacktestRequest,
BundleBacktestRequest bundleRequest,
int index)
{
if (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))
{
_logger.LogWarning("MoneyManagementName provided but cannot resolve in grain context: {Name}",
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
LightScenario scenario = null;
if (runBacktestRequest.Config.Scenario != null)
{
var sReq = runBacktestRequest.Config.Scenario;
scenario = new LightScenario(sReq.Name, sReq.LoopbackPeriod)
{
Indicators = sReq.Indicators?.Select(i => new LightIndicator(i.Name, i.Type)
{
SignalType = i.SignalType,
MinimumHistory = i.MinimumHistory,
Period = i.Period,
FastPeriods = i.FastPeriods,
SlowPeriods = i.SlowPeriods,
SignalPeriods = i.SignalPeriods,
Multiplier = i.Multiplier,
SmoothPeriods = i.SmoothPeriods,
StochPeriods = i.StochPeriods,
CyclePeriods = i.CyclePeriods
}).ToList() ?? new List<LightIndicator>()
};
}
// 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 ?? 1,
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
var result = await backtester.RunTradingBotBacktest(
backtestConfig,
runBacktestRequest.StartDate,
runBacktestRequest.EndDate,
bundleRequest.User,
true,
runBacktestRequest.WithCandles,
bundleRequest.RequestId.ToString()
);
_logger.LogInformation("Processed backtest for bundle request {RequestId}", bundleRequest.RequestId);
return result.Id;
}
private async Task UpdateFinalStatus(
BundleBacktestRequest bundleRequest,
IBacktester backtester,
IMessengerService messengerService)
{
if (bundleRequest.FailedBacktests == 0)
{
bundleRequest.Status = BundleBacktestRequestStatus.Completed;
await NotifyUser(bundleRequest, messengerService);
}
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";
await NotifyUser(bundleRequest, messengerService);
}
bundleRequest.CompletedAt = DateTime.UtcNow;
bundleRequest.CurrentBacktest = null;
await backtester.UpdateBundleBacktestRequestAsync(bundleRequest);
// Unregister retry reminder since bundle completed
await UnregisterRetryReminder();
}
private async Task HandleBundleRequestError(
BundleBacktestRequest bundleRequest,
IBacktester backtester,
Exception ex)
{
bundleRequest.Status = BundleBacktestRequestStatus.Failed;
bundleRequest.ErrorMessage = ex.Message;
bundleRequest.CompletedAt = DateTime.UtcNow;
await backtester.UpdateBundleBacktestRequestAsync(bundleRequest);
// Register retry reminder for failed bundle
await RegisterRetryReminder();
}
private async Task NotifyUser(BundleBacktestRequest bundleRequest, IMessengerService messengerService)
{
if (bundleRequest.User?.TelegramChannel != null)
{
var message = bundleRequest.FailedBacktests == 0
? $"✅ Bundle backtest '{bundleRequest.Name}' (ID: {bundleRequest.RequestId}) completed successfully."
: $"⚠️ Bundle backtest '{bundleRequest.Name}' (ID: {bundleRequest.RequestId}) completed with {bundleRequest.FailedBacktests} failed backtests.";
await messengerService.SendMessage(message, bundleRequest.User.TelegramChannel);
}
}
#region IRemindable Implementation
/// <summary>
/// Handles reminder callbacks for automatic retry of failed bundle backtests
/// </summary>
public async Task ReceiveReminder(string reminderName, TickStatus status)
{
if (reminderName != RETRY_REMINDER_NAME)
{
_logger.LogWarning("Unknown reminder {ReminderName} received", reminderName);
return;
}
var bundleRequestId = this.GetPrimaryKey();
_logger.LogInformation("Retry reminder triggered for bundle request {RequestId}", bundleRequestId);
try
{
using var scope = _scopeFactory.CreateScope();
var backtester = scope.ServiceProvider.GetRequiredService<IBacktester>();
// Get the bundle request
var bundleRequest = await GetBundleRequestById(backtester, bundleRequestId);
if (bundleRequest == null)
{
_logger.LogWarning("Bundle request {RequestId} not found during retry", bundleRequestId);
await UnregisterRetryReminder();
return;
}
// Check if bundle is still failed
if (bundleRequest.Status != BundleBacktestRequestStatus.Failed)
{
_logger.LogInformation("Bundle request {RequestId} is no longer failed (status: {Status}), unregistering reminder",
bundleRequestId, bundleRequest.Status);
await UnregisterRetryReminder();
return;
}
// Retry the bundle processing
_logger.LogInformation("Retrying failed bundle request {RequestId}", bundleRequestId);
// Reset status to pending for retry
bundleRequest.Status = BundleBacktestRequestStatus.Pending;
bundleRequest.ErrorMessage = null;
bundleRequest.CurrentBacktest = null;
await backtester.UpdateBundleBacktestRequestAsync(bundleRequest);
// Process the bundle again
await ProcessBundleRequestAsync();
}
catch (Exception ex)
{
_logger.LogError(ex, "Error during bundle backtest retry for request {RequestId}", bundleRequestId);
}
}
/// <summary>
/// Registers a retry reminder for this bundle request
/// </summary>
private async Task RegisterRetryReminder()
{
try
{
await this.RegisterOrUpdateReminder(RETRY_REMINDER_NAME, RETRY_INTERVAL, RETRY_INTERVAL);
_logger.LogInformation("Registered retry reminder for bundle request {RequestId}", this.GetPrimaryKey());
}
catch (Exception ex)
{
_logger.LogError(ex, "Failed to register retry reminder for bundle request {RequestId}", this.GetPrimaryKey());
}
}
/// <summary>
/// Unregisters the retry reminder for this bundle request
/// </summary>
private async Task UnregisterRetryReminder()
{
try
{
var reminder = await this.GetReminder(RETRY_REMINDER_NAME);
if (reminder != null)
{
await this.UnregisterReminder(reminder);
_logger.LogInformation("Unregistered retry reminder for bundle request {RequestId}", this.GetPrimaryKey());
}
}
catch (Exception ex)
{
_logger.LogError(ex, "Failed to unregister retry reminder for bundle request {RequestId}", this.GetPrimaryKey());
}
}
#endregion
}

View File

@@ -277,7 +277,7 @@ public class BundleBacktestWorker : BaseWorker<BundleBacktestWorker>
bundleRequest.User, // No user context in worker
true,
runBacktestRequest.WithCandles,
bundleRequest.RequestId // Use bundleRequestId as requestId for traceability
bundleRequest.RequestId.ToString() // Use bundleRequestId as requestId for traceability
);
_logger.LogInformation("Processed backtest for bundle request {RequestId}", bundleRequest.RequestId);

View File

@@ -13,7 +13,7 @@ public class NotifyBundleBacktestWorker : BaseWorker<NotifyBundleBacktestWorker>
{
private readonly IServiceProvider _serviceProvider;
private readonly IHubContext<BacktestHub> _hubContext;
private readonly ConcurrentDictionary<string, HashSet<string>> _sentBacktestIds = new();
private readonly ConcurrentDictionary<Guid, HashSet<string>> _sentBacktestIds = new();
public NotifyBundleBacktestWorker(
IServiceProvider serviceProvider,
@@ -39,7 +39,6 @@ public class NotifyBundleBacktestWorker : BaseWorker<NotifyBundleBacktestWorker>
foreach (var bundle in runningBundles)
{
var requestId = bundle.RequestId;
if (string.IsNullOrEmpty(requestId)) continue;
// Fetch all backtests for this bundle
var (backtests, _) = backtester.GetBacktestsByRequestIdPaginated(requestId, 1, 100);