Oda/managing-apps
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update pricing timing

Browse Source
This commit is contained in:
cryptooda
2025-09-14 22:27:54 +07:00
parent daeb26375b
commit 2847778c7c
10 changed files with 145 additions and 351 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
src/Managing.Application.Abstractions/Grains/IPriceFetcher15MinGrain.cs
View File

@@ -1,18 +0,0 @@
using Orleans;
namespace Managing.Application.Abstractions.Grains;
/// <summary>
/// Orleans grain interface for 5-minute price fetching operations.
/// This stateless worker grain handles fetching 5-minute price data from external APIs
/// and publishing to Orleans streams.
/// </summary>
public partial interface IPriceFetcher15MinGrain : IGrainWithIntegerKey
{
/// <summary>
/// Fetches 5-minute price data for all supported exchange/ticker combinations
/// and publishes new candles to their respective streams.
/// </summary>
/// <returns>True if the operation completed successfully, false otherwise</returns>
Task<bool> FetchAndPublishPricesAsync();
}

19
src/Managing.Application.Abstractions/Grains/IPriceFetcher1DayGrain.cs
View File

@@ -1,19 +0,0 @@
using Orleans;
namespace Managing.Application.Abstractions.Grains;
/// <summary>
/// Orleans grain interface for daily price fetching operations.
/// This stateless worker grain handles fetching daily price data from external APIs
/// and publishing to Orleans streams.
/// </summary>
public interface IPriceFetcher1DayGrain : IGrainWithIntegerKey
{
/// <summary>
/// Fetches daily price data for all supported exchange/ticker combinations
/// and publishes new candles to their respective streams.
/// </summary>
/// <returns>True if the operation completed successfully, false otherwise</returns>
Task<bool> FetchAndPublishPricesAsync();
}

19
src/Managing.Application.Abstractions/Grains/IPriceFetcher1HourGrain.cs
View File

@@ -1,19 +0,0 @@
using Orleans;
namespace Managing.Application.Abstractions.Grains;
/// <summary>
/// Orleans grain interface for 1-hour price fetching operations.
/// This stateless worker grain handles fetching 1-hour price data from external APIs
/// and publishing to Orleans streams.
/// </summary>
public interface IPriceFetcher1HourGrain : IGrainWithIntegerKey
{
/// <summary>
/// Fetches 1-hour price data for all supported exchange/ticker combinations
/// and publishes new candles to their respective streams.
/// </summary>
/// <returns>True if the operation completed successfully, false otherwise</returns>
Task<bool> FetchAndPublishPricesAsync();
}

19
src/Managing.Application.Abstractions/Grains/IPriceFetcher4HourGrain.cs
View File

@@ -1,19 +0,0 @@
using Orleans;
namespace Managing.Application.Abstractions.Grains;
/// <summary>
/// Orleans grain interface for 4-hour price fetching operations.
/// This stateless worker grain handles fetching 4-hour price data from external APIs
/// and publishing to Orleans streams.
/// </summary>
public interface IPriceFetcher4HourGrain : IGrainWithIntegerKey
{
/// <summary>
/// Fetches 4-hour price data for all supported exchange/ticker combinations
/// and publishes new candles to their respective streams.
/// </summary>
/// <returns>True if the operation completed successfully, false otherwise</returns>
Task<bool> FetchAndPublishPricesAsync();
}

26
src/Managing.Application.Tests/GrainHelpersTests.cs Normal file
View File

@@ -0,0 +1,26 @@
using Managing.Application.Shared;
using Xunit;
using static Managing.Common.Enums;
namespace Managing.Application.Tests;
public class GrainHelpersTests
{
[Fact]
public void GetIntervalMinutes_FifteenMinutes_ShouldReturn75()
{
var result = GrainHelpers.GetIntervalMinutes(Timeframe.FifteenMinutes);
Assert.Equal(7.5, result);
}
[Fact]
public void GetRandomizedTimerOptions_OneHour_ShouldReturn30()
{
var result = GrainHelpers.GetDynamicRandomizedTimerOptions(TimeSpan.FromMinutes(1), 200);
Assert.True(result.period.TotalSeconds <= 60.0);
Assert.True(result.dueTime.TotalSeconds <= 60.0);
Assert.True(result.period.TotalSeconds > 0.0);
Assert.True(result.dueTime.TotalSeconds > 0.0);
}
}

7
src/Managing.Application/Bots/Grains/AgentGrain.cs
View File

@@ -2,6 +2,7 @@ using Managing.Application.Abstractions;
using Managing.Application.Abstractions.Grains;
using Managing.Application.Abstractions.Services;
using Managing.Application.Bots.Models;
using Managing.Application.Shared;
using Managing.Domain.Statistics;
using Microsoft.Extensions.Logging;
using static Managing.Common.Enums;
@@ -60,9 +61,11 @@ public class AgentGrain : Grain, IAgentGrain, IRemindable
{
try
{
var options = GrainHelpers.GetDynamicRandomizedTimerOptions(TimeSpan.FromMinutes(2), 200);
// Register a reminder that fires every 5 minutes
await this.RegisterOrUpdateReminder(_updateSummaryReminderName, TimeSpan.FromMinutes(5),
TimeSpan.FromMinutes(1));
await this.RegisterOrUpdateReminder(_updateSummaryReminderName, options.dueTime,
options.period);
_logger.LogInformation("Reminder registered for agent {UserId} to update summary every 5 minutes",
this.GetPrimaryKeyLong());
}

50
src/Managing.Application/Bots/Grains/LiveTradingBotGrain.cs
View File

@@ -1,6 +1,7 @@
using Managing.Application.Abstractions;
using Managing.Application.Abstractions.Grains;
using Managing.Application.Abstractions.Services;
using Managing.Application.Shared;
using Managing.Common;
using Managing.Core;
using Managing.Domain.Accounts;
@@ -184,12 +185,14 @@ public class LiveTradingBotGrain : Grain, ILiveTradingBotGrain, IRemindable
if (_timer != null) return;
var botOptions = GrainHelpers.GetDynamicRandomizedTimerOptions(TimeSpan.FromMinutes(1), 20);
_timer = this.RegisterGrainTimer(
async _ => await ExecuteBotCycle(),
new GrainTimerCreationOptions
{
Period = TimeSpan.FromMinutes(1),
DueTime = TimeSpan.FromMinutes(1),
Period = botOptions.period,
DueTime = botOptions.dueTime,
KeepAlive = true
});
}
@@ -294,11 +297,8 @@ public class LiveTradingBotGrain : Grain, ILiveTradingBotGrain, IRemindable
}
// Check broker balance before running
var balances = await ServiceScopeHelpers.WithScopedService<IExchangeService, List<Balance>>(_scopeFactory, async exchangeService =>
{
return await exchangeService.GetBalances(_tradingBot.Account, false);
});
var balances = await ServiceScopeHelpers.WithScopedService<IExchangeService, List<Balance>>(_scopeFactory,
async exchangeService => { return await exchangeService.GetBalances(_tradingBot.Account, false); });
var usdcBalance = balances.FirstOrDefault(b => b.TokenName == Ticker.USDC.ToString());
var ethBalance = balances.FirstOrDefault(b => b.TokenName == Ticker.ETH.ToString());
@@ -325,10 +325,13 @@ public class LiveTradingBotGrain : Grain, ILiveTradingBotGrain, IRemindable
{
try
{
var swapInfo = await ServiceScopeHelpers.WithScopedService<IAccountService, SwapInfos>(_scopeFactory, async accountService =>
{
return await accountService.SwapGmxTokensAsync(_state.State.User, _tradingBot.Account.Name, Ticker.USDC, Ticker.ETH, 5);
});
var swapInfo = await ServiceScopeHelpers.WithScopedService<IAccountService, SwapInfos>(
_scopeFactory,
async accountService =>
{
return await accountService.SwapGmxTokensAsync(_state.State.User,
_tradingBot.Account.Name, Ticker.USDC, Ticker.ETH, 5);
});
if (swapInfo.Success)
{
@@ -507,6 +510,7 @@ public class LiveTradingBotGrain : Grain, ILiveTradingBotGrain, IRemindable
{
throw new InvalidOperationException("Bot is not running - cannot get account information");
}
return Task.FromResult(_tradingBot.Account);
}
@@ -617,19 +621,22 @@ public class LiveTradingBotGrain : Grain, ILiveTradingBotGrain, IRemindable
var actualStatus = await botRegistry.GetBotStatus(botId);
if (actualStatus == status)
{
_logger.LogDebug("Bot {BotId} status successfully updated to {Status} in BotRegistry (attempt {Attempt})",
_logger.LogDebug(
"Bot {BotId} status successfully updated to {Status} in BotRegistry (attempt {Attempt})",
botId, status, attempt);
return;
}
else
{
_logger.LogWarning("Bot {BotId} status update verification failed. Expected: {Expected}, Actual: {Actual} (attempt {Attempt})",
_logger.LogWarning(
"Bot {BotId} status update verification failed. Expected: {Expected}, Actual: {Actual} (attempt {Attempt})",
botId, status, actualStatus, attempt);
}
}
catch (Exception ex)
{
_logger.LogError(ex, "Failed to update bot {BotId} status to {Status} in BotRegistry (attempt {Attempt})",
_logger.LogError(ex,
"Failed to update bot {BotId} status to {Status} in BotRegistry (attempt {Attempt})",
botId, status, attempt);
if (attempt == maxRetries)
@@ -734,7 +741,8 @@ public class LiveTradingBotGrain : Grain, ILiveTradingBotGrain, IRemindable
{
_logger.LogDebug(
"Successfully saved bot statistics for bot {BotId}: Wins={Wins}, Losses={Losses}, PnL={PnL}, ROI={ROI}%, Volume={Volume}, Fees={Fees}, Long={LongPositions}, Short={ShortPositions}",
_state.State.Identifier, bot.TradeWins, bot.TradeLosses, bot.Pnl, bot.Roi, bot.Volume, bot.Fees, bot.LongPositionCount, bot.ShortPositionCount);
_state.State.Identifier, bot.TradeWins, bot.TradeLosses, bot.Pnl, bot.Roi, bot.Volume, bot.Fees,
bot.LongPositionCount, bot.ShortPositionCount);
}
else
{
@@ -750,7 +758,8 @@ public class LiveTradingBotGrain : Grain, ILiveTradingBotGrain, IRemindable
/// <summary>
/// Notifies the user about swap operations via webhook/telegram
/// </summary>
private async Task NotifyUserAboutSwap(bool isSuccess, decimal amount, string? transactionHash, string? errorMessage = null)
private async Task NotifyUserAboutSwap(bool isSuccess, decimal amount, string? transactionHash,
string? errorMessage = null)
{
try
{
@@ -769,10 +778,11 @@ public class LiveTradingBotGrain : Grain, ILiveTradingBotGrain, IRemindable
$"⏰ **Time:** {DateTime.UtcNow:yyyy-MM-dd HH:mm:ss} UTC";
// Send notification via webhook service
await ServiceScopeHelpers.WithScopedService<IWebhookService>(_scopeFactory, async webhookService =>
{
await webhookService.SendMessage(message, _state.State.User?.TelegramChannel);
});
await ServiceScopeHelpers.WithScopedService<IWebhookService>(_scopeFactory,
async webhookService =>
{
await webhookService.SendMessage(message, _state.State.User?.TelegramChannel);
});
}
catch (Exception ex)
{

229
src/Managing.Application/Grains/PriceFetcher15MinGrain.cs
View File

@@ -1,229 +0,0 @@
using Managing.Application.Abstractions.Grains;
using Managing.Application.Abstractions.Repositories;
using Managing.Application.Abstractions.Services;
using Managing.Common;
using Managing.Domain.Accounts;
using Managing.Domain.Candles;
using Microsoft.Extensions.Logging;
using Orleans.Concurrency;
using Orleans.Streams;
using static Managing.Common.Enums;
namespace Managing.Application.Grains;
/// <summary>
/// StatelessWorker grain for fetching 5-minute price data from external APIs and publishing to Orleans streams.
/// This grain runs every 5 minutes and processes all exchange/ticker combinations for the 5-minute timeframe.
/// </summary>
[StatelessWorker]
public class PriceFetcher15MinGrain : Grain, IPriceFetcher15MinGrain, IRemindable
{
private readonly ILogger<PriceFetcher15MinGrain> _logger;
private readonly IExchangeService _exchangeService;
private readonly ICandleRepository _candleRepository;
private readonly IGrainFactory _grainFactory;
private const string FetchPricesReminderName = "Fetch15minPricesReminder";
private IDisposable _timer;
// Predefined lists of trading parameters to fetch
private static readonly TradingExchanges[] SupportedExchanges =
{
TradingExchanges.Evm
};
private static readonly Ticker[] SupportedTickers = Constants.GMX.Config.SupportedTickers;
private static readonly Timeframe TargetTimeframe = Timeframe.FifteenMinutes;
public PriceFetcher15MinGrain(
ILogger<PriceFetcher15MinGrain> logger,
IExchangeService exchangeService,
ICandleRepository candleRepository,
IGrainFactory grainFactory)
{
_logger = logger;
_exchangeService = exchangeService;
_candleRepository = candleRepository;
_grainFactory = grainFactory;
}
public override async Task OnActivateAsync(CancellationToken cancellationToken)
{
_logger.LogInformation("{0} activated", nameof(PriceFetcher15MinGrain));
// Register a reminder to enable timer if not existing
await this.RegisterOrUpdateReminder(
FetchPricesReminderName,
TimeSpan.FromSeconds(5),
TimeSpan.FromMinutes(7.5));
await base.OnActivateAsync(cancellationToken);
}
public override async Task OnDeactivateAsync(DeactivationReason reason, CancellationToken cancellationToken)
{
_logger.LogInformation("{0} deactivating. Reason: {Reason}",
nameof(PriceFetcher15MinGrain), reason.Description);
StopAndDisposeTimer();
await base.OnDeactivateAsync(reason, cancellationToken);
}
public async Task<bool> FetchAndPublishPricesAsync()
{
try
{
_logger.LogInformation("Starting {timeframe} price fetch cycle", TargetTimeframe);
var fetchTasks = new List<Task>();
// Create fetch tasks for all exchange/ticker combinations for 15-minute timeframe
foreach (var exchange in SupportedExchanges)
{
foreach (var ticker in SupportedTickers)
{
fetchTasks.Add(FetchAndPublish(exchange, ticker, TargetTimeframe));
}
}
// Execute all fetch operations in parallel
await Task.WhenAll(fetchTasks);
_logger.LogInformation("{0} - Completed {1} price fetch cycle for {2} combinations",
nameof(PriceFetcher15MinGrain), TargetTimeframe, fetchTasks.Count);
return true;
}
catch (Exception ex)
{
_logger.LogError(ex, "Error during price fetch cycle");
return false;
}
}
private async Task FetchAndPublish(TradingExchanges exchange, Ticker ticker, Timeframe timeframe)
{
try
{
// Create a dummy account for API calls (this may need to be adjusted based on your implementation)
var account = new Account
{
Name = "PriceFetcher",
Exchange = exchange,
Type = AccountType.Watch
};
// Get the last candle date from database
var existingCandles = await _candleRepository.GetCandles(exchange, ticker, timeframe,
DateTime.UtcNow.AddDays(-30), DateTime.UtcNow.AddDays(1), 20);
var isFirstCall = !existingCandles.Any();
var startDate = !isFirstCall
? existingCandles.Last().Date
: new DateTime(2017, 1, 1);
// Fetch new candles from external API
var newCandles = await _exchangeService.GetCandles(account, ticker, startDate, timeframe, isFirstCall);
if (newCandles?.Any() == true)
{
var streamKey = CandleHelpers.GetCandleStoreGrainKey(exchange, ticker, timeframe);
_logger.LogDebug("Fetched {CandleCount} new candles for {StreamKey}",
newCandles.Count, streamKey);
// Process all new candles
var processedCandles = newCandles.OrderBy(c => c.Date)
.Where(c => c.Date <= DateTime.UtcNow) // Avoid duplicates
.ToList();
foreach (var candle in processedCandles)
{
// Ensure candle has correct metadata
candle.Exchange = exchange;
candle.Ticker = ticker;
candle.Timeframe = timeframe;
}
// Save all candles to database in a single batch
if (processedCandles.Any())
{
await _candleRepository.InsertCandles(processedCandles).ConfigureAwait(false);
_logger.LogDebug(
"[{Ticker}][{Exchange}][{Timeframe}] Inserted {CandleCount} candles for {StreamKey}",
ticker, exchange, timeframe,
processedCandles.Count, streamKey);
}
var streamProvider = this.GetStreamProvider("DefaultStreamProvider");
var stream = streamProvider.GetStream<Candle>(streamKey);
// Publish to stream (if needed)
foreach (var candle in processedCandles)
{
await stream.OnNextAsync(candle);
_logger.LogTrace("Published candle for {StreamKey} at {Date}",
streamKey, candle.Date);
}
}
}
catch (Exception ex)
{
_logger.LogError(ex, "Error fetching prices for {Exchange}-{Ticker}-{Timeframe}",
exchange, ticker, timeframe);
}
}
/// <summary>
/// Starts the Orleans timer for periodic price fetching
/// </summary>
private void RegisterAndStartTimer()
{
if (_timer != null) return;
// Calculate the next execution time aligned to X-minute boundaries
var now = DateTime.UtcNow;
var dueTime = CandleHelpers.GetDueTimeForTimeframe(TargetTimeframe, now);
var period = TimeSpan.FromSeconds(CandleHelpers.GetBaseIntervalInSeconds(TargetTimeframe));
_logger.LogInformation("{0} next execution scheduled in {1} seconds and at {2:} UTC every {3} seconds",
nameof(PriceFetcher15MinGrain), dueTime, now.Add(dueTime), period);
_timer = this.RegisterGrainTimer(
async _ => await FetchAndPublishPricesAsync(),
new GrainTimerCreationOptions
{
Period = period,
DueTime = dueTime,
KeepAlive = true
});
_logger.LogInformation("{0} timer registered and started", nameof(PriceFetcher15MinGrain));
}
private void StopAndDisposeTimer()
{
if (_timer != null)
{
_timer?.Dispose();
_timer = null;
_logger.LogInformation("{0} timer stopped and disposed", nameof(PriceFetcher15MinGrain));
}
}
public Task ReceiveReminder(string reminderName, TickStatus status)
{
if (reminderName == FetchPricesReminderName)
{
// Only enable timer if not existing anymore
if (_timer == null)
{
RegisterAndStartTimer();
}
}
return Task.CompletedTask;
}
}

25
src/Managing.Application/Grains/PriceFetcherInitializer.cs
View File

@@ -1,5 +1,6 @@
using Managing.Application.Abstractions.Grains;
using Microsoft.Extensions.Hosting;
using static Managing.Common.Enums;
namespace Managing.Application.Grains;
@@ -14,14 +15,26 @@ public class PriceFetcherInitializer : IHostedService
public async Task StartAsync(CancellationToken cancellationToken)
{
var fiveMinute = _grainFactory.GetGrain<IPriceFetcher15MinGrain>(0);
Console.WriteLine("GrainId : {0}", fiveMinute.GetGrainId());
// Initialize grains for different timeframes
var timeframes = new[]
{
Timeframe.FifteenMinutes,
Timeframe.OneHour,
Timeframe.FourHour,
Timeframe.OneDay
};
// Actually call a method on the grain to activate it
// This will trigger OnActivateAsync and register the reminder
await fiveMinute.FetchAndPublishPricesAsync();
foreach (var timeframe in timeframes)
{
var grain = _grainFactory.GetGrain<IPriceFetcherGrain>(timeframe.ToString());
Console.WriteLine("GrainId for {0}: {1}", timeframe, grain.GetGrainId());
Console.WriteLine("PriceFetcher5MinGrain activated and initial fetch completed");
// Actually call a method on the grain to activate it
// This will trigger OnActivateAsync and register the reminder
await grain.FetchAndPublishPricesAsync();
Console.WriteLine("PriceFetcherGrain for {0} activated and initial fetch completed", timeframe);
}
}
public Task StopAsync(CancellationToken cancellationToken)

46
src/Managing.Application/Shared/GrainHelpers.cs Normal file
View File

@@ -0,0 +1,46 @@
using Managing.Common;
namespace Managing.Application.Shared;
public static class GrainHelpers
{
/// <summary>
/// Gets the interval in minutes for the reminder based on the timeframe
/// </summary>
public static double GetIntervalMinutes(Enums.Timeframe timeframe)
{
return timeframe switch
{
Enums.Timeframe.OneMinute => 1.0,
Enums.Timeframe.FiveMinutes => 5.0,
Enums.Timeframe.FifteenMinutes => 7.5, // Half of the timeframe for more frequent checks
Enums.Timeframe.OneHour => 30.0, // Half of the timeframe for more frequent checks
Enums.Timeframe.FourHour => 120.0, // Half of the timeframe for more frequent checks
Enums.Timeframe.OneDay => 720.0, // Half of the timeframe for more frequent checks
_ => 5.0 // Default to 5 minutes
};
}
// Helper method to get a randomized due time and period
public static (TimeSpan dueTime, TimeSpan period) GetDynamicRandomizedTimerOptions(TimeSpan basePeriod,
int numberOfBots)
{
var random = new Random();
// Determine the jitter percentage based on bot count
// Example: 2% jitter for 100 bots, 10% for 1000 bots, etc.
// This formula can be fine-tuned based on your system's performance metrics.
double jitterPercentage = Math.Min(0.20, (double)numberOfBots / 10000); // Caps at 20% for 10,000 bots
// Calculate a random due time within the period
var dueTimeSeconds = random.Next(1, (int)basePeriod.TotalSeconds);
var dueTime = TimeSpan.FromSeconds(dueTimeSeconds);
// Add a random jitter to the period
var jitterFactor = (random.NextDouble() * jitterPercentage) - (jitterPercentage / 2);
var periodSeconds = (int)(basePeriod.TotalSeconds * (1 + jitterFactor));
var period = TimeSpan.FromSeconds(periodSeconds);
return (dueTime, period);
}
}