using System; using System.Linq; using System.Threading; using System.Threading.Tasks; using Alpaca.Markets; using FinlyticBot.Database; using FinlyticBot.Util; using FinlyticCore.Services; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; namespace FinlyticBot.Services; public class AlpacaWebSocketMonitorWorker : BackgroundService { private readonly IServiceScopeFactory _scopeFactory; private readonly ISettingsService _settingsService; private readonly BotMqttClient _mqttClient; private readonly IConfiguration _configuration; private readonly IFinlyticLogger _finlyticLogger; private IAlpacaStreamingClient? _streamingClient; public AlpacaWebSocketMonitorWorker( IServiceScopeFactory scopeFactory, ISettingsService settingsService, BotMqttClient mqttClient, IConfiguration configuration, IFinlyticLogger finlyticLogger) { _scopeFactory = scopeFactory; _settingsService = settingsService; _mqttClient = mqttClient; _configuration = configuration; _finlyticLogger = finlyticLogger; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { await _finlyticLogger.LogInfoAsync(SettingKeys.BotChannel, "[AlpacaWebSocketMonitor] Starting Alpaca Trade Update Stream Monitor..."); while (!stoppingToken.IsCancellationRequested) { try { string keyId = await _settingsService.GetSettingAsync(SettingKeys.AlpacaKeyId, stoppingToken); if (string.IsNullOrWhiteSpace(keyId)) { keyId = _configuration["Alpaca:KeyId"] ?? _configuration["Alpaca__KeyId"] ?? string.Empty; } string secretKey = await _settingsService.GetSettingAsync(SettingKeys.AlpacaSecretKey, stoppingToken); if (string.IsNullOrWhiteSpace(secretKey)) { secretKey = _configuration["Alpaca:SecretKey"] ?? _configuration["Alpaca__SecretKey"] ?? string.Empty; } bool isPaper = await _settingsService.GetSettingAsync(SettingKeys.AlpacaIsPaper, stoppingToken); if (string.IsNullOrWhiteSpace(keyId) || string.IsNullOrWhiteSpace(secretKey)) { await Task.Delay(TimeSpan.FromSeconds(15), stoppingToken); continue; } var environment = isPaper ? Alpaca.Markets.Environments.Paper : Alpaca.Markets.Environments.Live; _streamingClient = environment.GetAlpacaStreamingClient(new SecretKey(keyId, secretKey)); _streamingClient.OnTradeUpdate += HandleTradeUpdate; var authStatus = await _streamingClient.ConnectAndAuthenticateAsync(stoppingToken); await _finlyticLogger.LogInfoAsync(SettingKeys.BotChannel, "[AlpacaWebSocketMonitor] Connected & Authenticated to Alpaca Streaming WS. Status: {Status}", authStatus.ToString()); // Keep connection alive until cancellation var tcs = new TaskCompletionSource(); using (stoppingToken.Register(() => tcs.TrySetResult(true))) { await tcs.Task; } await _streamingClient.DisconnectAsync(CancellationToken.None); } catch (Exception ex) when (!stoppingToken.IsCancellationRequested) { await _finlyticLogger.LogWarningAsync(SettingKeys.BotChannel, ex, "[AlpacaWebSocketMonitor] Streaming WebSocket disconnected. Retrying in 10s..."); await Task.Delay(TimeSpan.FromSeconds(10), stoppingToken); } } } private void HandleTradeUpdate(ITradeUpdate update) { _ = Task.Run(async () => { try { using var scope = _scopeFactory.CreateScope(); var dbContext = scope.ServiceProvider.GetRequiredService(); var order = update.Order; if (order == null) return; var trade = await dbContext.ExecutedPaperTrades .FirstOrDefaultAsync(t => t.AlpacaOrderId == order.OrderId); if (trade == null) { // Check if it's a child order (SL / TP) of an existing trade trade = await dbContext.ExecutedPaperTrades .Where(t => t.Symbol == order.Symbol && t.Status == "Filled") .OrderByDescending(t => t.PlacedAt) .FirstOrDefaultAsync(); } if (trade == null) return; if (update.Event == TradeEvent.Fill) { decimal fillPrice = update.Price ?? order.AverageFillPrice ?? trade.SignalEntryPrice; trade.ActualFillPrice = fillPrice; trade.Status = "Filled"; trade.FilledAt = DateTime.UtcNow; if (trade.SignalEntryPrice > 0) { trade.SlippagePercent = Math.Round(((fillPrice - trade.SignalEntryPrice) / trade.SignalEntryPrice) * 100m, 3); } await _finlyticLogger.LogInfoAsync(SettingKeys.BotChannel, "[AlpacaTradeUpdate] Order FILLED for {Symbol}: FillPrice=${Price:F2} (Signal: ${SigPrice:F2}, Slippage: {Slip:F3}%)", trade.Symbol, fillPrice, trade.SignalEntryPrice, trade.SlippagePercent ?? 0m); } else if (update.Event == TradeEvent.PartialFill) { trade.Status = "PartiallyFilled"; } else if (update.Event == TradeEvent.Canceled || update.Event == TradeEvent.Expired || update.Event == TradeEvent.Rejected) { trade.Status = update.Event.ToString(); trade.ClosedAt = DateTime.UtcNow; } else if (update.Event == TradeEvent.Stopped || update.Event == TradeEvent.Calculated) { // Position closed by Stop Loss or Take Profit trade.Status = "Closed"; trade.ClosedAt = DateTime.UtcNow; decimal exitPrice = update.Price ?? trade.ActualFillPrice ?? trade.SignalEntryPrice; if (trade.ActualFillPrice.HasValue && update.Price.HasValue) { exitPrice = update.Price.Value; decimal diff = trade.Side == "BUY" ? (exitPrice - trade.ActualFillPrice.Value) : (trade.ActualFillPrice.Value - exitPrice); trade.RealizedPnl = diff * trade.Quantity; if (trade.ActualFillPrice.Value > 0) { trade.RealizedPnlPercent = Math.Round((diff / trade.ActualFillPrice.Value) * 100m, 2); } } await _finlyticLogger.LogInfoAsync(SettingKeys.BotChannel, "[AlpacaTradeUpdate] Position CLOSED for {Symbol}: Realized PnL: ${Pnl:F2} ({Pct:F2}%)", trade.Symbol, trade.RealizedPnl, trade.RealizedPnlPercent ?? 0m); // Publish Closed Trade to MQTT for WinRate calibration & AI feedback loop bool isWin = trade.RealizedPnl > 0; var feedbackDto = new FinlyticCore.Models.Trades.TradeProposalDto { TradeId = trade.TradeId, Symbol = trade.Symbol, Isin = trade.Isin, CompanyName = trade.CompanyName, EntryPrice = trade.SignalEntryPrice, ActualEntryPrice = trade.ActualFillPrice, CurrentPrice = exitPrice, StopLoss = trade.StopLossPrice, TakeProfit = trade.TakeProfitPrice1, Status = isWin ? "Closed_Profit" : "Closed_Loss", SignalType = trade.Side, WinRate = trade.WinRate, PnlAbsolute = trade.RealizedPnl, PnlPercent = trade.RealizedPnlPercent, CloseReason = isWin ? "TakeProfit_Hit" : "StopLoss_Hit", UserExitTimestamp = trade.ClosedAt, CreatedAt = trade.PlacedAt }; await _mqttClient.PublishAsync($"finlytic/trades/closed/{trade.TradeId}", feedbackDto); await _finlyticLogger.LogInfoAsync(SettingKeys.BotChannel, "[AlpacaTradeUpdate] Dispatched closed trade feedback event to MQTT for {TradeId} (Win: {IsWin})", trade.TradeId, isWin); } trade.UpdatedAt = DateTime.UtcNow; await dbContext.SaveChangesAsync(); } catch (Exception ex) { _ = _finlyticLogger.LogErrorAsync(SettingKeys.BotChannel, ex, "[AlpacaWebSocketMonitor] Error processing TradeUpdate event."); } }); } }