Finlytic/FinlyticTrades/Services/FeedbackExporterEngine.cs

using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text.Json;
using System.Text.RegularExpressions;
using System.Threading;
using System.Threading.Tasks;
using FinlyticCore.Models.Trades;
using FinlyticTrades.Database;
using FinlyticTrades.Entities;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Parquet.Serialization;

namespace FinlyticTrades.Services;

public interface IFeedbackExporterEngine
{
    /// <summary>
    /// Exports feedback data for closed trades.
    /// </summary>
    Task ExportFeedbackDataAsync(CancellationToken cancellationToken = default);
}


public class FeedbackExporterEngine : BackgroundService, IFeedbackExporterEngine
{
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly ILogger<FeedbackExporterEngine> _logger;
    private readonly string _feedbackDir;

    public FeedbackExporterEngine(IServiceScopeFactory scopeFactory, ILogger<FeedbackExporterEngine> logger)
    {
        _scopeFactory = scopeFactory;
        _logger = logger;
        _feedbackDir = Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "data", "feedback");
        
        if (!Directory.Exists(_feedbackDir))
        {
            Directory.CreateDirectory(_feedbackDir);
        }
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation("[{Channel}] Feedback Exporter Engine background service started.", "TradesChannel");

        try
        {
            await Task.Delay(TimeSpan.FromSeconds(30), stoppingToken);
        }
        catch (OperationCanceledException)
        {
            return;
        }

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                await ExportFeedbackDataAsync(stoppingToken);
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                break;
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "[{Channel}] Error executing feedback exporter job.", "TradesChannel");
            }

            try
            {
                await Task.Delay(TimeSpan.FromHours(6), stoppingToken);
            }
            catch (OperationCanceledException)
            {
                break;
            }
        }

        _logger.LogInformation("[{Channel}] Feedback Exporter Engine background service stopped.", "TradesChannel");
    }

    /// <summary>
    /// Exports feedback data for closed trades into sector-based JSON and Parquet formats.
    /// Uses atomic file-writes to avoid thread-lock conflicts with reader processes.
    /// </summary>
    public async Task ExportFeedbackDataAsync(CancellationToken cancellationToken = default)
    {
        using var scope = _scopeFactory.CreateScope();
        var dbContext = scope.ServiceProvider.GetRequiredService<TradesDbContext>();

        var closedTrades = await dbContext.Trades
            .AsNoTracking()
            .Where(t => t.Status == TradeStatus.Closed && t.UserExitPrice.HasValue)
            .ToListAsync(cancellationToken);

        if (closedTrades.Count == 0)
        {
            _logger.LogInformation("[{Channel}] No closed trades available for export.", "TradesChannel");
            return;
        }

        var groups = closedTrades.GroupBy(t => SanitizeSectorName(t.Sector));

        foreach (var group in groups)
        {
            if (cancellationToken.IsCancellationRequested) break;

            var sectorName = group.Key;
            var sectorDir = Path.Combine(_feedbackDir, sectorName);

            if (!Directory.Exists(sectorDir))
            {
                Directory.CreateDirectory(sectorDir);
            }

            var feedbackRecords = new List<TradeFeedbackRecord>();

            foreach (var t in group)
            {
                var startTime = t.ExecutionTimestamp ?? t.CreatedAt;
                var endTime = t.UserExitTimestamp ?? t.ClosedAt ?? DateTime.UtcNow;
                double reactionDelay = Math.Max(0, (endTime - startTime).TotalMinutes);

                decimal exitPrice = t.UserExitPrice ?? t.EntryPrice;

                decimal entryPrice = t.ActualEntryPrice.HasValue && t.ActualEntryPrice.Value > 0 
                    ? t.ActualEntryPrice.Value 
                    : t.EntryPrice;

                decimal slippagePct = t.EntryPrice > 0 
                    ? Math.Abs((entryPrice - t.EntryPrice) / t.EntryPrice) * 100.0m 
                    : 0m;

                var rec = new TradeFeedbackRecord
                {
                    TradeId = t.TradeId,
                    AnalysisId = t.AnalysisId,
                    Sector = t.Sector,
                    Symbol = t.Symbol,
                    Isin = t.Isin,
                    EntryPrice = entryPrice,
                    StopLoss = t.StopLoss,
                    TakeProfit = t.TakeProfit,
                    UserExitPrice = exitPrice,
                    PnlAbsolute = t.PnlAbsolute ?? 0m,
                    PnlPercent = t.PnlPercent ?? 0m,
                    IsWin = t.IsWin ?? false,
                    CloseReason = t.CloseReason ?? "Unknown",
                    VixRegime = t.VixRegime,
                    VixValue = t.VixValue,
                    ReactionDelayMinutes = Math.Round(reactionDelay, 2),
                    SlippagePercent = Math.Round(slippagePct, 2),
                    CreatedAt = t.CreatedAt,
                    ClosedAt = endTime
                };

                feedbackRecords.Add(rec);
            }

            // 1. Atomic JSON Export (.tmp -> move)
            string jsonPath = Path.Combine(sectorDir, $"{sectorName}_feedback.json");
            string jsonTmpPath = Path.Combine(sectorDir, $"{sectorName}_feedback.json.tmp");
            string jsonContent = JsonSerializer.Serialize(feedbackRecords, new JsonSerializerOptions { WriteIndented = true });
            
            await File.WriteAllTextAsync(jsonTmpPath, jsonContent, cancellationToken);
            File.Move(jsonTmpPath, jsonPath, overwrite: true);

            // 2. Atomic Parquet Export (.tmp -> move)
            try
            {
                string parquetPath = Path.Combine(sectorDir, $"{sectorName}_feedback.parquet");
                string parquetTmpPath = Path.Combine(sectorDir, $"{sectorName}_feedback.parquet.tmp");
                
                await using (var fileStream = new FileStream(parquetTmpPath, FileMode.Create, FileAccess.Write, FileShare.None, 4096, useAsync: true))
                {
                    await ParquetSerializer.SerializeAsync(feedbackRecords, fileStream, cancellationToken: cancellationToken);
                }

                File.Move(parquetTmpPath, parquetPath, overwrite: true);

                _logger.LogInformation("[{Channel}] Exported Parquet feedback file for sector '{Sector}' to {ParquetPath}", "TradesChannel", sectorName, parquetPath);
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex, "[{Channel}] Failed to write Parquet file for sector '{Sector}'. JSON file was written successfully.", "TradesChannel", sectorName);
            }
        }

        _logger.LogInformation("[{Channel}] Successfully exported feedback data for {Count} closed trades across {Sectors} sectors.",
            "TradesChannel", closedTrades.Count, groups.Count());
    }

    private static string SanitizeSectorName(string? sector)
    {
        if (string.IsNullOrWhiteSpace(sector)) return "general";
        
        var clean = Regex.Replace(sector.Trim().ToLowerInvariant(), @"[^a-z0-9_\-]", "_");
        return string.IsNullOrWhiteSpace(clean) ? "general" : clean;
    }
}