Add timer-based subscription restart mechanism for trader

- Add SubscriptionTimeoutInterval (5 minutes) to detect silent subscription failures
- Track LastTradesDataTicks and LastMarketDataTicks for both subscription streams
- Implement CheckTradesTimeout and CheckMarketDataTimeout methods to monitor data flow
- Update ReceiveTradesLoop and SendOrdersLoop to restart subscriptions when no data received
- Fix existing Task.Delay calls to properly handle cancellation tokens

Fixes issue where trades subscription was broken after 17 days of uptime.
The bot now proactively restarts subscriptions if no data is received within the timeout period.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <[email protected]>

Author: konard

csharp/TraderBot/TradingService.cs

@@ -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;
@@ -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;
@@ -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)
@@ -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)
             {
@@ -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);
                 }
             }
         }
@@ -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();
@@ -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)
             {