Skip to content

Trader: add timer to restart subscriptions if no data received #278

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
konard wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Trader: add timer to restart subscriptions if no data received #278

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3a48a7f
Initial commit with task details for issue #183
konard Sep 11, 2025
aa1323f
Remove CLAUDE.md - PR created successfully
konard Sep 11, 2025
b52de54
Add timer-based subscription restart mechanism for trader
konard Sep 11, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
102 changes: 98 additions & 4 deletions csharp/TraderBot/TradingService.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ public class TradingService : BackgroundService
protected static readonly TimeSpan RefreshInterval = TimeSpan.FromSeconds(10);
protected static readonly TimeSpan SyncInterval = TimeSpan.FromSeconds(20);
protected static readonly TimeSpan WaitOutputInterval = TimeSpan.FromSeconds(20);
protected static readonly TimeSpan SubscriptionTimeoutInterval = TimeSpan.FromMinutes(5);
protected readonly InvestApiClient InvestApi;
protected readonly ILogger<TradingService> Logger;
protected readonly IHostApplicationLifetime Lifetime;
Expand All @@ -33,6 +34,8 @@ public class TradingService : BackgroundService
protected long LastRefreshTicks;
protected long LastSyncTicks;
protected long LastWaitOutputTicks;
protected long LastTradesDataTicks;
protected long LastMarketDataTicks;
protected TimeSpan MinimumTimeToBuy;
protected TimeSpan MaximumTimeToBuy;
protected readonly ConcurrentDictionary<string, OrderState> ActiveBuyOrders;
Expand Down Expand Up @@ -112,6 +115,9 @@ public TradingService(ILogger<TradingService> logger, InvestApiClient investApi,
LotsSets = new ConcurrentDictionary<decimal, long>();
ActiveSellOrderSourcePrice = new ConcurrentDictionary<string, decimal>();
LastOperationsCheckpoint = settings.LoadOperationsFrom;
var nowTicks = DateTime.UtcNow.Ticks;
LastTradesDataTicks = nowTicks;
LastMarketDataTicks = nowTicks;
}

protected async Task ReceiveTrades(CancellationToken cancellationToken)
Expand All @@ -122,6 +128,7 @@ protected async Task ReceiveTrades(CancellationToken cancellationToken)
});
await foreach (var data in tradesStream.ResponseStream.ReadAllAsync(cancellationToken))
{
Interlocked.Exchange(ref LastTradesDataTicks, DateTime.UtcNow.Ticks);
Logger.LogInformation($"Trade: {data}");
if (data.PayloadCase == TradesStreamResponse.PayloadOneofCase.OrderTrades)
{
Expand Down Expand Up @@ -349,14 +356,40 @@ protected async Task SendOrdersLoop(CancellationToken cancellationToken)
try
{
await Refresh(forceReset: true);
await SendOrders(cancellationToken);

using var timeoutCancellationTokenSource = new CancellationTokenSource();
using var combinedCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeoutCancellationTokenSource.Token);

var sendOrdersTask = SendOrders(combinedCancellationTokenSource.Token);
var timeoutTask = CheckMarketDataTimeout(timeoutCancellationTokenSource, cancellationToken);

await Task.WhenAny(sendOrdersTask, timeoutTask);

if (timeoutTask.IsCompleted && !timeoutTask.IsCanceled)
{
Logger.LogWarning("Market data subscription timeout detected, restarting subscription.");
timeoutCancellationTokenSource.Cancel();
}

try
{
await sendOrdersTask;
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
throw;
}
catch (OperationCanceledException)
{
Logger.LogInformation("Market data subscription cancelled due to timeout, will restart.");
}
}
catch (Exception ex)
{
if (!cancellationToken.IsCancellationRequested)
{
Logger.LogError(ex, "SendOrders exception.");
await Task.Delay(RecoveryInterval);
await Task.Delay(RecoveryInterval, cancellationToken);
}
}
}
Expand All @@ -369,19 +402,79 @@ protected async Task ReceiveTradesLoop(CancellationToken cancellationToken)
try
{
await Refresh(forceReset: true);
await ReceiveTrades(cancellationToken);

using var timeoutCancellationTokenSource = new CancellationTokenSource();
using var combinedCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeoutCancellationTokenSource.Token);

var receiveTradesTask = ReceiveTrades(combinedCancellationTokenSource.Token);
var timeoutTask = CheckTradesTimeout(timeoutCancellationTokenSource, cancellationToken);

await Task.WhenAny(receiveTradesTask, timeoutTask);

if (timeoutTask.IsCompleted && !timeoutTask.IsCanceled)
{
Logger.LogWarning("Trades subscription timeout detected, restarting subscription.");
timeoutCancellationTokenSource.Cancel();
}

try
{
await receiveTradesTask;
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
throw;
}
catch (OperationCanceledException)
{
Logger.LogInformation("Trades subscription cancelled due to timeout, will restart.");
}
}
catch (Exception ex)
{
if (!cancellationToken.IsCancellationRequested)
{
Logger.LogError(ex, "ReceiveTrades exception.");
await Task.Delay(RecoveryInterval);
await Task.Delay(RecoveryInterval, cancellationToken);
}
}
}
}

protected async Task CheckTradesTimeout(CancellationTokenSource timeoutCancellationTokenSource, CancellationToken cancellationToken)
{
while (!cancellationToken.IsCancellationRequested && !timeoutCancellationTokenSource.Token.IsCancellationRequested)
{
var nowTicks = DateTime.UtcNow.Ticks;
var lastDataTicks = Interlocked.Read(ref LastTradesDataTicks);

if (nowTicks - lastDataTicks > SubscriptionTimeoutInterval.Ticks)
{
Logger.LogWarning($"No trades data received for {SubscriptionTimeoutInterval.TotalMinutes} minutes, triggering restart.");
return;
}

await Task.Delay(TimeSpan.FromSeconds(30), cancellationToken);
}
}

protected async Task CheckMarketDataTimeout(CancellationTokenSource timeoutCancellationTokenSource, CancellationToken cancellationToken)
{
while (!cancellationToken.IsCancellationRequested && !timeoutCancellationTokenSource.Token.IsCancellationRequested)
{
var nowTicks = DateTime.UtcNow.Ticks;
var lastDataTicks = Interlocked.Read(ref LastMarketDataTicks);

if (nowTicks - lastDataTicks > SubscriptionTimeoutInterval.Ticks)
{
Logger.LogWarning($"No market data received for {SubscriptionTimeoutInterval.TotalMinutes} minutes, triggering restart.");
return;
}

await Task.Delay(TimeSpan.FromSeconds(30), cancellationToken);
}
}

protected async Task SendOrders(CancellationToken cancellationToken)
{
var marketDataStream = InvestApi.MarketDataStream.MarketDataStream();
Expand All @@ -402,6 +495,7 @@ await marketDataStream.RequestStream.WriteAsync(new MarketDataRequest
}, cancellationToken);
await foreach (var data in marketDataStream.ResponseStream.ReadAllAsync(cancellationToken))
{
Interlocked.Exchange(ref LastMarketDataTicks, DateTime.UtcNow.Ticks);
// Logger.LogInformation($"data.PayloadCase: {data.PayloadCase}");
if (data.PayloadCase == MarketDataResponse.PayloadOneofCase.SubscribeOrderBookResponse)
{
Expand Down
Loading