Finlytic/FinlyticBot/Services/AlpacaWebSocketMonitorWorker.cs

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<AlpacaWebSocketMonitorWorker> _finlyticLogger;

    private IAlpacaStreamingClient? _streamingClient;

    public AlpacaWebSocketMonitorWorker(
        IServiceScopeFactory scopeFactory,
        ISettingsService settingsService,
        BotMqttClient mqttClient,
        IConfiguration configuration,
        IFinlyticLogger<AlpacaWebSocketMonitorWorker> 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<bool>();
                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<BotDbContext>();

                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.");
            }
        });
    }
}