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
{
///
/// Exports feedback data for closed trades.
///
Task ExportFeedbackDataAsync(CancellationToken cancellationToken = default);
}
public class FeedbackExporterEngine : BackgroundService, IFeedbackExporterEngine
{
private readonly IServiceScopeFactory _scopeFactory;
private readonly ILogger _logger;
private readonly string _feedbackDir;
public FeedbackExporterEngine(IServiceScopeFactory scopeFactory, ILogger 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");
}
///
/// 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.
///
public async Task ExportFeedbackDataAsync(CancellationToken cancellationToken = default)
{
using var scope = _scopeFactory.CreateScope();
var dbContext = scope.ServiceProvider.GetRequiredService();
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();
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;
}
}