diff --git a/FinlyticSentiment/Database/SentimentDbContext.cs b/FinlyticSentiment/Database/SentimentDbContext.cs index 85b0240..d380cbb 100644 --- a/FinlyticSentiment/Database/SentimentDbContext.cs +++ b/FinlyticSentiment/Database/SentimentDbContext.cs @@ -1,23 +1,33 @@ +using FinlyticCore.Database; +using FinlyticCore.Entities.Settings; using FinlyticSentiment.Entities; using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Design; namespace FinlyticSentiment.Database; /// /// EF Core DbContext for managing FinlyticSentiment settings in PostgreSQL. /// -public class SentimentDbContext : DbContext +public class SentimentDbContext : DbContext, ISettingsDbContext { public SentimentDbContext(DbContextOptions options) : base(options) { } + public DbSet DynamicSettings => Set(); public DbSet Settings => Set(); protected override void OnModelCreating(ModelBuilder modelBuilder) { base.OnModelCreating(modelBuilder); + modelBuilder.Entity(entity => + { + entity.HasKey(e => e.Id); + entity.HasIndex(e => e.Key).IsUnique(); + }); + modelBuilder.Entity(entity => { entity.ToTable("sentiment_settings"); @@ -27,3 +37,13 @@ public class SentimentDbContext : DbContext }); } } + +public class SentimentDbContextFactory : IDesignTimeDbContextFactory +{ + public SentimentDbContext CreateDbContext(string[] args) + { + var optionsBuilder = new DbContextOptionsBuilder(); + optionsBuilder.UseNpgsql("Host=localhost;Database=sentiment;Username=postgres;Password=postgres"); + return new SentimentDbContext(optionsBuilder.Options); + } +} diff --git a/FinlyticSentiment/Migrations/20260815184006_AddDynamicSettings.Designer.cs b/FinlyticSentiment/Migrations/20260815184006_AddDynamicSettings.Designer.cs new file mode 100644 index 0000000..fb14de4 --- /dev/null +++ b/FinlyticSentiment/Migrations/20260815184006_AddDynamicSettings.Designer.cs @@ -0,0 +1,94 @@ +// +using System; +using FinlyticSentiment.Database; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; +using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata; + +#nullable disable + +namespace FinlyticSentiment.Migrations +{ + [DbContext(typeof(SentimentDbContext))] + [Migration("20260815184006_AddDynamicSettings")] + partial class AddDynamicSettings + { + /// + protected override void BuildTargetModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasAnnotation("ProductVersion", "10.0.9") + .HasAnnotation("Relational:MaxIdentifierLength", 63); + + NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder); + + modelBuilder.Entity("FinlyticCore.Entities.Settings.SettingEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("Key") + .IsRequired() + .HasMaxLength(150) + .HasColumnType("character varying(150)"); + + b.Property("LastUpdatedUtc") + .HasColumnType("timestamp with time zone"); + + b.Property("ServiceIdentifier") + .IsRequired() + .HasMaxLength(100) + .HasColumnType("character varying(100)"); + + b.Property("ValueJson") + .IsRequired() + .HasColumnType("text"); + + b.HasKey("Id"); + + b.HasIndex("Key") + .IsUnique(); + + b.ToTable("DynamicSettings"); + }); + + modelBuilder.Entity("FinlyticSentiment.Entities.SentimentSettingsEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("EnglishWebhookUrl") + .IsRequired() + .HasMaxLength(500) + .HasColumnType("character varying(500)"); + + b.Property("GermanWebhookUrl") + .IsRequired() + .HasMaxLength(500) + .HasColumnType("character varying(500)"); + + b.Property("MaxBatchSize") + .HasColumnType("integer"); + + b.Property("MinConfidenceScore") + .HasColumnType("double precision"); + + b.Property("SweepIntervalMinutes") + .HasColumnType("integer"); + + b.Property("UpdatedAt") + .HasColumnType("timestamp with time zone"); + + b.HasKey("Id"); + + b.ToTable("sentiment_settings", (string)null); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/FinlyticSentiment/Migrations/20260815184006_AddDynamicSettings.cs b/FinlyticSentiment/Migrations/20260815184006_AddDynamicSettings.cs new file mode 100644 index 0000000..fbd967a --- /dev/null +++ b/FinlyticSentiment/Migrations/20260815184006_AddDynamicSettings.cs @@ -0,0 +1,43 @@ +using System; +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace FinlyticSentiment.Migrations +{ + /// + public partial class AddDynamicSettings : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.CreateTable( + name: "DynamicSettings", + columns: table => new + { + Id = table.Column(type: "uuid", nullable: false), + Key = table.Column(type: "character varying(150)", maxLength: 150, nullable: false), + ValueJson = table.Column(type: "text", nullable: false), + ServiceIdentifier = table.Column(type: "character varying(100)", maxLength: 100, nullable: false), + LastUpdatedUtc = table.Column(type: "timestamp with time zone", nullable: false) + }, + constraints: table => + { + table.PrimaryKey("PK_DynamicSettings", x => x.Id); + }); + + migrationBuilder.CreateIndex( + name: "IX_DynamicSettings_Key", + table: "DynamicSettings", + column: "Key", + unique: true); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropTable( + name: "DynamicSettings"); + } + } +} diff --git a/FinlyticSentiment/Migrations/SentimentDbContextModelSnapshot.cs b/FinlyticSentiment/Migrations/SentimentDbContextModelSnapshot.cs index 5d0acd5..9f13197 100644 --- a/FinlyticSentiment/Migrations/SentimentDbContextModelSnapshot.cs +++ b/FinlyticSentiment/Migrations/SentimentDbContextModelSnapshot.cs @@ -22,6 +22,37 @@ namespace FinlyticSentiment.Migrations NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder); + modelBuilder.Entity("FinlyticCore.Entities.Settings.SettingEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("Key") + .IsRequired() + .HasMaxLength(150) + .HasColumnType("character varying(150)"); + + b.Property("LastUpdatedUtc") + .HasColumnType("timestamp with time zone"); + + b.Property("ServiceIdentifier") + .IsRequired() + .HasMaxLength(100) + .HasColumnType("character varying(100)"); + + b.Property("ValueJson") + .IsRequired() + .HasColumnType("text"); + + b.HasKey("Id"); + + b.HasIndex("Key") + .IsUnique(); + + b.ToTable("DynamicSettings"); + }); + modelBuilder.Entity("FinlyticSentiment.Entities.SentimentSettingsEntity", b => { b.Property("Id") diff --git a/FinlyticSentiment/Program.cs b/FinlyticSentiment/Program.cs index 64c1a54..f4ee768 100644 --- a/FinlyticSentiment/Program.cs +++ b/FinlyticSentiment/Program.cs @@ -1,13 +1,24 @@ +using System; +using FinlyticCore.Database; +using FinlyticCore.Services; using FinlyticSentiment.Database; using FinlyticSentiment.Services; using FinlyticSentiment.Util; using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; var builder = Host.CreateApplicationBuilder(args); // Register PostgreSQL DbContext for settings persistence builder.Services.AddDbContext(options => options.UseNpgsql(builder.Configuration.GetConnectionString("DefaultConnection"))); +builder.Services.AddScoped(sp => sp.GetRequiredService()); + +// Register Core Services +builder.Services.AddSingleton(); +builder.Services.AddSingleton(typeof(IFinlyticLogger<>), typeof(FinlyticLogger<>)); // Register HttpClient builder.Services.AddHttpClient(); @@ -36,8 +47,7 @@ using (var scope = host.Services.CreateScope()) } catch (Exception ex) { - var logger = scope.ServiceProvider.GetRequiredService>(); - logger.LogError(ex, "An error occurred during database migration for FinlyticSentiment on startup."); + Console.WriteLine($"Critical error during database migration for FinlyticSentiment: {ex.Message}"); } } diff --git a/FinlyticSentiment/Services/FinBertAnalyzerService.cs b/FinlyticSentiment/Services/FinBertAnalyzerService.cs index 5504ede..ad83f38 100644 --- a/FinlyticSentiment/Services/FinBertAnalyzerService.cs +++ b/FinlyticSentiment/Services/FinBertAnalyzerService.cs @@ -1,9 +1,13 @@ +using System; +using System.Net.Http; using System.Net.Http.Json; using System.Text.Json; +using System.Threading.Tasks; using FinlyticCore.Dtos.News; using FinlyticCore.Dtos.Sentiment; -using Microsoft.Extensions.Configuration; -using Microsoft.Extensions.Logging; +using FinlyticCore.Services; +using FinlyticSentiment.Util; +using Microsoft.Extensions.DependencyInjection; namespace FinlyticSentiment.Services; @@ -15,8 +19,6 @@ public interface IFinBertAnalyzerService /// /// Analyzes a news article using language-targeted FinBERT webhooks and returns structured metrics. /// - /// The news article DTO to evaluate. - /// A task returning the FinBERT sentiment analysis result. Task AnalyzeArticleAsync(NewsArticleDto article); } @@ -27,22 +29,16 @@ public class FinBertAnalyzerService : IFinBertAnalyzerService { private readonly HttpClient _httpClient; private readonly IServiceScopeFactory _scopeFactory; - private readonly ILogger _logger; + private readonly IFinlyticLogger _finlyticLogger; - /// - /// Initializes a new instance of the class. - /// - /// The HTTP client instance. - /// The service scope factory for DB access. - /// The logging channel. public FinBertAnalyzerService( HttpClient httpClient, IServiceScopeFactory scopeFactory, - ILogger logger) + IFinlyticLogger finlyticLogger) { _httpClient = httpClient; _scopeFactory = scopeFactory; - _logger = logger; + _finlyticLogger = finlyticLogger; } /// @@ -64,7 +60,7 @@ public class FinBertAnalyzerService : IFinBertAnalyzerService minConfidence = settings.MinConfidenceScore; } - _logger.LogInformation("[{Channel}] Analyzing article (ID: {Id}, Lang: {Lang}) via webhook: {Url} (MinConf: {Conf})", "SentimentChannel", article.Id, article.Language ?? "de", targetUrl, minConfidence); + await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[FinBertAnalyzerService] Analyzing article (ID: {Id}, Lang: {Lang}) via webhook: {Url} (MinConf: {Conf})", article.Id, article.Language ?? "de", targetUrl, minConfidence); var requestBody = new { @@ -88,13 +84,11 @@ public class FinBertAnalyzerService : IFinBertAnalyzerService using var doc = JsonDocument.Parse(content); var root = doc.RootElement; - // Handle array response if n8n returns an array of items (e.g. [{ "json": { ... } }]) if (root.ValueKind == JsonValueKind.Array && root.GetArrayLength() > 0) { root = root[0]; } - // Unwrap n8n wrapper objects: "json", "output", "data", "result", "body" if (root.ValueKind == JsonValueKind.Object) { if (root.TryGetProperty("json", out var jsonChild) && jsonChild.ValueKind == JsonValueKind.Object) @@ -134,7 +128,6 @@ public class FinBertAnalyzerService : IFinBertAnalyzerService neu = GetDoubleProp(probsElem, "neutral") ?? neu; } - // Normalize German vs English labels string label = rawLabel.Trim().ToUpperInvariant() switch { "POSITIV" or "POSITIVE" => "POSITIVE", @@ -142,7 +135,6 @@ public class FinBertAnalyzerService : IFinBertAnalyzerService _ => "NEUTRAL" }; - // If compoundScore is 0 but probabilities or label indicate sentiment, compute compoundScore if (Math.Abs(compoundScore) < 0.001) { if (pos > 0 || neg > 0) @@ -173,15 +165,14 @@ public class FinBertAnalyzerService : IFinBertAnalyzerService SummarySnippet = snippet }; } - } } - _logger.LogWarning("[{Channel}] n8n Webhook returned non-success status: {StatusCode}. Falling back to rule analyzer.", "SentimentChannel", response.StatusCode); + await _finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, "[FinBertAnalyzerService] n8n Webhook returned non-success status: {StatusCode}.", response.StatusCode); } catch (Exception ex) { - _logger.LogError(ex, "[{Channel}] Failed to call n8n sentiment webhook. Executing fallback sentiment analyzer.", "SentimentChannel"); + await _finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[FinBertAnalyzerService] Failed to call n8n sentiment webhook."); } return null; diff --git a/FinlyticSentiment/Services/SentimentBackgroundService.cs b/FinlyticSentiment/Services/SentimentBackgroundService.cs index 87a5286..b6f45aa 100644 --- a/FinlyticSentiment/Services/SentimentBackgroundService.cs +++ b/FinlyticSentiment/Services/SentimentBackgroundService.cs @@ -1,9 +1,13 @@ +using System; using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; using FinlyticCore.Dtos.News; +using FinlyticCore.Services; using FinlyticSentiment.Util; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; -using Microsoft.Extensions.Logging; namespace FinlyticSentiment.Services; @@ -17,53 +21,45 @@ public class SentimentBackgroundService : BackgroundService private readonly IFinBertAnalyzerService _analyzer; private readonly ISentimentStorageService _storage; private readonly IServiceScopeFactory _scopeFactory; - private readonly ILogger _logger; + private readonly IFinlyticLogger _finlyticLogger; - // In-Memory Mutex / Cache zur Vermeidung doppelter Verarbeitung (Race Conditions zwischen Broadcast & Sweep) private static readonly ConcurrentDictionary ProcessingArticles = new(); - /// - /// Initializes a new instance of the class. - /// public SentimentBackgroundService( SentimentMqttClient mqttClient, IFinBertAnalyzerService analyzer, ISentimentStorageService storage, IServiceScopeFactory scopeFactory, - ILogger logger) + IFinlyticLogger finlyticLogger) { _mqttClient = mqttClient; _analyzer = analyzer; _storage = storage; _scopeFactory = scopeFactory; - _logger = logger; + _finlyticLogger = finlyticLogger; } /// protected override async Task ExecuteAsync(CancellationToken stoppingToken) { - _logger.LogInformation("[{Channel}] FinlyticSentiment Background Service started.", "SentimentChannel"); + await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] FinlyticSentiment Background Service started."); - // Realtime Broadcast Event registrieren (Echtzeit-Artikel) _mqttClient.OnArticleReceived += async (article) => { await ProcessSingleArticleAsync(article, stoppingToken); }; - // Kurze Initialisierungs-Verzögerung für die MQTT-Verbindung await Task.Delay(3000, stoppingToken); while (!stoppingToken.IsCancellationRequested) { - int maxBatchSize; - int sweepIntervalMinutes; + int maxBatchSize = 10; + int sweepIntervalMinutes = 5; using (var scope = _scopeFactory.CreateScope()) { - var settingsDb = scope.ServiceProvider.GetRequiredService(); - var settings = await settingsDb.GetSettingsAsync(); - maxBatchSize = settings.MaxBatchSize; - sweepIntervalMinutes = settings.SweepIntervalMinutes; + var settings = scope.ServiceProvider.GetRequiredService(); + maxBatchSize = await settings.GetSettingAsync(SettingKeys.MaxBatchSize, stoppingToken); } var interval = TimeSpan.FromMinutes(Math.Max(1, sweepIntervalMinutes)); @@ -74,19 +70,15 @@ public class SentimentBackgroundService : BackgroundService } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { - // Normales Beenden beim Stoppen des Hosts break; } catch (Exception ex) { - _logger.LogError(ex, "[{Channel}] Unhandled exception encountered during sentiment sweep cycle.", - "SentimentChannel"); + await _finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[SentimentBackgroundService] Unhandled exception encountered during sentiment sweep cycle."); } - _logger.LogInformation("[{Channel}] Waiting {Minutes} minute(s) until next sentiment sweep...", - "SentimentChannel", interval.TotalMinutes); + await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] Waiting {Minutes} minute(s) until next sentiment sweep...", interval.TotalMinutes); - // Verwendet PeriodicTimer oder CancellationToken-resistenten Delay using var timer = new PeriodicTimer(interval); try { @@ -98,27 +90,21 @@ public class SentimentBackgroundService : BackgroundService } } - _logger.LogInformation("[{Channel}] FinlyticSentiment Background Service is shutting down gracefully.", - "SentimentChannel"); + await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] FinlyticSentiment Background Service is shutting down gracefully."); } - /// - /// Performs a single batch sweep of pending news articles. - /// private async Task PerformSentimentSweepAsync(int maxBatchSize, CancellationToken cancellationToken) { - _logger.LogInformation("[{Channel}] Starting sentiment sweep for pending news articles (Limit: {Limit})...", - "SentimentChannel", maxBatchSize); + await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] Starting sentiment sweep for pending news articles (Limit: {Limit})...", maxBatchSize); List pendingArticles = await _mqttClient.GetPendingArticlesAsync(limit: maxBatchSize); if (pendingArticles.Count == 0) { - _logger.LogInformation("[{Channel}] No pending news articles found in FinlyticNews.", "SentimentChannel"); + await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] No pending news articles found in FinlyticNews."); return; } - _logger.LogInformation("[{Channel}] Retrieved {Count} pending article(s) for FinBERT sentiment evaluation.", - "SentimentChannel", pendingArticles.Count); + await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] Retrieved {Count} pending article(s) for FinBERT sentiment evaluation.", pendingArticles.Count); foreach (var article in pendingArticles) { @@ -127,42 +113,32 @@ public class SentimentBackgroundService : BackgroundService } } - /// - /// Processes FinBERT sentiment analysis for a single article, updates two-stage JSON summaries, and notifies FinlyticNews of status update. - /// private async Task ProcessSingleArticleAsync(NewsArticleDto article, CancellationToken cancellationToken = default) { if (article == null || article.Id == Guid.Empty) return; - // Deduplizierung: Verhindert, dass derselbe Artikel zeitgleich im Sweep & im Broadcast verarbeitet wird if (!ProcessingArticles.TryAdd(article.Id, 0)) { - _logger.LogDebug("[{Channel}] Article {Id} is already being processed. Skipping duplicate run.", - "SentimentChannel", article.Id); + await _finlyticLogger.LogDebugAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] Article {Id} is already being processed. Skipping duplicate run.", article.Id); return; } try { - _logger.LogInformation("[{Channel}] Evaluating FinBERT sentiment for article: '{Title}' (ID: {Id})", - "SentimentChannel", article.Title, article.Id); + await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] Evaluating FinBERT sentiment for article: '{Title}' (ID: {Id})", article.Title, article.Id); var finbert = await _analyzer.AnalyzeArticleAsync(article); if (finbert == null) { - _logger.LogWarning( - "[{Channel}] FinBERT analysis returned NULL for article {Id} ('{Title}'). Aborting processing for this run.", - "SentimentChannel", article.Id, article.Title); + await _finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] FinBERT analysis returned NULL for article {Id} ('{Title}'). Aborting processing for this run.", article.Id, article.Title); return; } if (cancellationToken.IsCancellationRequested) return; - // 1. Artikel-Level Sentiment speichern await _storage.SaveArticleSentimentAsync(article, finbert); - // 2. ISIN- & Sektor-Summaries aktualisieren if (article.MatchedAssets != null && article.MatchedAssets.Count > 0) { foreach (var asset in article.MatchedAssets) @@ -187,29 +163,22 @@ public class SentimentBackgroundService : BackgroundService if (cancellationToken.IsCancellationRequested) return; - // 3. FinlyticNews über Erfolg informieren bool updated = await _mqttClient.UpdateArticleStatusAsync(article.Id, "Analyzed"); if (updated) { - _logger.LogInformation( - "[{Channel}] Article sentiment processed and status set to 'Analyzed' in FinlyticNews: {Title} (ID: {Id}) -> {Label}", - "SentimentChannel", article.Title, article.Id, finbert.Label); + await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] Article sentiment processed and status set to 'Analyzed' in FinlyticNews: {Title} (ID: {Id}) -> {Label}", article.Title, article.Id, finbert.Label); } else { - _logger.LogWarning( - "[{Channel}] Failed to confirm status update to 'Analyzed' in FinlyticNews for article: {Id}", - "SentimentChannel", article.Id); + await _finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] Failed to confirm status update to 'Analyzed' in FinlyticNews for article: {Id}", article.Id); } } catch (Exception ex) { - _logger.LogError(ex, "[{Channel}] Error processing sentiment for article: {Id} ({Title})", - "SentimentChannel", article.Id, article.Title); + await _finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[SentimentBackgroundService] Error processing sentiment for article: {Id} ({Title})", article.Id, article.Title); } finally { - // Lock nach der Verarbeitung immer freigeben ProcessingArticles.TryRemove(article.Id, out _); } } diff --git a/FinlyticSentiment/Services/SentimentStorageService.cs b/FinlyticSentiment/Services/SentimentStorageService.cs index 33ca733..3466a29 100644 --- a/FinlyticSentiment/Services/SentimentStorageService.cs +++ b/FinlyticSentiment/Services/SentimentStorageService.cs @@ -1,10 +1,17 @@ +using System; using System.Collections.Concurrent; +using System.Collections.Generic; +using System.IO; +using System.Linq; using System.Text.Encodings.Web; using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; using FinlyticCore.Dtos.News; using FinlyticCore.Dtos.Sentiment; +using FinlyticCore.Services; +using FinlyticSentiment.Util; using Microsoft.Extensions.Configuration; -using Microsoft.Extensions.Logging; namespace FinlyticSentiment.Services; @@ -13,40 +20,10 @@ namespace FinlyticSentiment.Services; /// public interface ISentimentStorageService { - /// - /// Appends a new FinBERT analysis event to the ISIN summary file and recalculates the current aggregate metrics. - /// - /// The asset ISIN code. - /// The name of the asset company. - /// The sector associated with the asset. - /// The analyzed article DTO. - /// The FinBERT sentiment result. - /// A task representing the file update operation. Task UpdateIsinSummaryAsync(string isin, string companyName, string sector, NewsArticleDto article, FinBertResultDto finbert); - - /// - /// Appends a new FinBERT analysis event to the Sector summary file and recalculates the sector aggregate metrics. - /// - /// The target sector name. - /// The asset ISIN code triggering the sector update. - /// The unique news article identifier. - /// The FinBERT sentiment result. - /// A task representing the file update operation. Task UpdateSectorSummaryAsync(string sector, string isin, string articleId, FinBertResultDto finbert); - - /// - /// Persists an article's sentiment analysis directly in data/summaries/articles/{articleId}.json. - /// Task SaveArticleSentimentAsync(NewsArticleDto article, FinBertResultDto finbert); - - /// - /// Retrieves the sentiment analysis entry for a specific news article. - /// Task GetArticleSentimentAsync(string articleId); - - /// - /// Retrieves the aggregate sentiment summary for a specific asset ISIN. - /// Task GetIsinSummaryAsync(string isin); } @@ -54,13 +31,13 @@ public class SentimentStorageService : ISentimentStorageService { private static readonly ConcurrentDictionary FileLocks = new(); - private readonly ILogger _logger; + private readonly IFinlyticLogger _finlyticLogger; private readonly string _basePath; private readonly JsonSerializerOptions _jsonOptions; - public SentimentStorageService(IConfiguration configuration, ILogger logger) + public SentimentStorageService(IConfiguration configuration, IFinlyticLogger finlyticLogger) { - _logger = logger; + _finlyticLogger = finlyticLogger; _basePath = configuration["Storage:SummariesPath"] ?? "data/summaries"; Directory.CreateDirectory(Path.Combine(_basePath, "isin")); @@ -88,30 +65,28 @@ public class SentimentStorageService : ISentimentStorageService try { string nowIso = DateTime.UtcNow.ToString("yyyy-MM-ddTHH:mm:ssZ"); - string analysisId = $"sent_{DateTime.UtcNow:yyyyMMdd}_{Random.Shared.Next(100, 999)}"; - var entry = new IsinAnalysisEntry { - AnalysisId = analysisId, + AnalysisId = $"sent_{Guid.NewGuid():N}", Timestamp = nowIso, Article = new IsinAnalysisArticleRef { ArticleId = cleanId, - Title = article.Title ?? "", - Source = article.Author ?? "FinlyticNews", - PublishedAt = article.PublishedAt.ToString("yyyy-MM-ddTHH:mm:ssZ") + Title = article.Title, + PublishedAt = article.PublishedAt.ToString("yyyy-MM-ddTHH:mm:ssZ"), + Source = article.Author ?? "FinlyticNews" }, FinbertResult = finbert, - SummarySnippet = finbert.SummarySnippet ?? article.Summary ?? article.Title ?? "" + SummarySnippet = finbert.SummarySnippet ?? article.Summary ?? string.Empty }; - string json = JsonSerializer.Serialize(entry, _jsonOptions); + var json = JsonSerializer.Serialize(entry, _jsonOptions); await File.WriteAllTextAsync(filePath, json); - _logger.LogInformation("[{Channel}] Successfully saved article sentiment file: {Path}", "SentimentChannel", filePath); + await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentStorageService] Successfully saved article sentiment file: {Path}", filePath); } catch (Exception ex) { - _logger.LogError(ex, "[{Channel}] Failed to write article sentiment file: {Path}", "SentimentChannel", filePath); + await _finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[SentimentStorageService] Failed to write article sentiment file: {Path}", filePath); } finally { @@ -124,10 +99,9 @@ public class SentimentStorageService : ISentimentStorageService { if (string.IsNullOrWhiteSpace(articleId)) return null; - var cleanId = articleId.Trim(); - var articleFilePath = Path.Combine(_basePath, "articles", $"{cleanId}.json"); + string cleanId = articleId.Trim(); + string articleFilePath = Path.Combine(_basePath, "articles", $"{cleanId}.json"); - // 1. Primärer Lookup if (File.Exists(articleFilePath)) { var fileLock = FileLocks.GetOrAdd(articleFilePath, _ => new SemaphoreSlim(1, 1)); @@ -139,7 +113,7 @@ public class SentimentStorageService : ISentimentStorageService } catch (Exception ex) { - _logger.LogWarning(ex, "[{Channel}] Failed to read article sentiment file: {Path}", "SentimentChannel", articleFilePath); + await _finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, ex, "[SentimentStorageService] Failed to read article sentiment file: {Path}", articleFilePath); } finally { @@ -147,22 +121,6 @@ public class SentimentStorageService : ISentimentStorageService } } - // 2. Fallback in ISIN-Dateien - var dirPath = Path.Combine(_basePath, "isin"); - if (!Directory.Exists(dirPath)) return null; - - foreach (var file in Directory.GetFiles(dirPath, "*.json")) - { - try - { - var json = await File.ReadAllTextAsync(file); - var doc = JsonSerializer.Deserialize(json, _jsonOptions); - var match = doc?.Analyses?.FirstOrDefault(a => string.Equals(a.Article?.ArticleId?.Trim(), cleanId, StringComparison.OrdinalIgnoreCase)); - if (match != null) return match; - } - catch { } - } - return null; } @@ -171,8 +129,8 @@ public class SentimentStorageService : ISentimentStorageService { if (string.IsNullOrWhiteSpace(isin)) return null; - var cleanIsin = isin.Trim(); - var filePath = Path.Combine(_basePath, "isin", $"{cleanIsin}.json"); + string cleanIsin = isin.Trim().ToUpperInvariant(); + string filePath = Path.Combine(_basePath, "isin", $"{cleanIsin}.json"); if (!File.Exists(filePath)) return null; @@ -186,7 +144,7 @@ public class SentimentStorageService : ISentimentStorageService } catch (Exception ex) { - _logger.LogError(ex, "[{Channel}] Error reading ISIN summary file: {Path}", "SentimentChannel", filePath); + await _finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[SentimentStorageService] Error reading ISIN summary file: {Path}", filePath); return null; } finally @@ -198,9 +156,9 @@ public class SentimentStorageService : ISentimentStorageService /// public async Task UpdateIsinSummaryAsync(string isin, string companyName, string sector, NewsArticleDto article, FinBertResultDto finbert) { - if (string.IsNullOrWhiteSpace(isin) || article == null) return; + if (string.IsNullOrWhiteSpace(isin)) return; - string cleanIsin = isin.Trim(); + string cleanIsin = isin.Trim().ToUpperInvariant(); string filePath = Path.Combine(_basePath, "isin", $"{cleanIsin}.json"); var fileLock = FileLocks.GetOrAdd(filePath, _ => new SemaphoreSlim(1, 1)); @@ -208,87 +166,92 @@ public class SentimentStorageService : ISentimentStorageService try { - IsinSentimentSummaryDto isinDoc; - + var analyses = new List(); if (File.Exists(filePath)) { - try + var existingJson = await File.ReadAllTextAsync(filePath); + var existing = JsonSerializer.Deserialize(existingJson, _jsonOptions); + if (existing?.Analyses != null) { - string json = await File.ReadAllTextAsync(filePath); - isinDoc = JsonSerializer.Deserialize(json, _jsonOptions) ?? new IsinSentimentSummaryDto { Isin = cleanIsin }; - } - catch - { - isinDoc = new IsinSentimentSummaryDto { Isin = cleanIsin }; + analyses.AddRange(existing.Analyses); } } - else - { - isinDoc = new IsinSentimentSummaryDto { Isin = cleanIsin }; - } + + string cleanArticleId = article.Id.ToString(); + analyses.RemoveAll(a => string.Equals(a.Article?.ArticleId, cleanArticleId, StringComparison.OrdinalIgnoreCase)); string nowIso = DateTime.UtcNow.ToString("yyyy-MM-ddTHH:mm:ssZ"); - string analysisId = $"sent_{DateTime.UtcNow:yyyyMMdd}_{Random.Shared.Next(100, 999)}"; - var newEntry = new IsinAnalysisEntry { - AnalysisId = analysisId, + AnalysisId = $"sent_{Guid.NewGuid():N}", Timestamp = nowIso, Article = new IsinAnalysisArticleRef { - ArticleId = article.Id.ToString(), - Title = article.Title ?? "", - Source = article.Author ?? "FinlyticNews", - PublishedAt = article.PublishedAt.ToString("yyyy-MM-ddTHH:mm:ssZ") + ArticleId = cleanArticleId, + Title = article.Title, + PublishedAt = article.PublishedAt.ToString("yyyy-MM-ddTHH:mm:ssZ"), + Source = article.Author ?? "FinlyticNews" }, FinbertResult = finbert, - SummarySnippet = finbert.SummarySnippet ?? article.Summary ?? article.Title ?? "" + SummarySnippet = finbert.SummarySnippet ?? article.Summary ?? string.Empty }; - // Duplikat-Bereinigung: Falls Artikel bereits existiert, alten Eintrag entfernen! - var updatedAnalyses = isinDoc.Analyses? - .Where(a => !string.Equals(a.Article?.ArticleId, article.Id.ToString(), StringComparison.OrdinalIgnoreCase)) - .ToList() ?? new List(); + analyses.Add(newEntry); - // Neuen Eintrag oben einfügen - updatedAnalyses.Insert(0, newEntry); - - // Capping: Maximal die letzten 100 Analysen aufheben (verhindert gigantische JSON-Dateien) - if (updatedAnalyses.Count > 100) + var cutoff = DateTime.UtcNow.AddDays(-14); + var validAnalyses = analyses.Where(a => { - updatedAnalyses = updatedAnalyses.Take(100).ToList(); + if (DateTime.TryParse(a.Article?.PublishedAt ?? a.Timestamp, out var pubDate)) + { + return pubDate >= cutoff; + } + return true; + }).ToList(); + + var updatedAnalyses = validAnalyses.OrderByDescending(a => a.Article?.PublishedAt ?? a.Timestamp).Take(50).ToList(); + + double totalCompound = 0.0; + double totalConf = 0.0; + + foreach (var item in updatedAnalyses) + { + if (item.FinbertResult == null) continue; + totalCompound += item.FinbertResult.CompoundScore; + totalConf += item.FinbertResult.Confidence; } - double avgCompound = updatedAnalyses.Average(a => a.FinbertResult.CompoundScore); - double avgConf = updatedAnalyses.Average(a => a.FinbertResult.Confidence); - string label = CalculateLabel(avgCompound); + int total = updatedAnalyses.Count; + double avgCompound = total > 0 ? totalCompound / total : 0.0; + double avgConf = total > 0 ? totalConf / total : 0.0; - string textSummary = label switch - { - "POSITIVE" => $"Die Stimmungsanalyse zeigt einen weiterhin positiven Trend ({avgCompound:F2}). Hauptursache sind positive Berichte und starke Markt-Signale.", - "NEGATIVE" => $"Die Stimmungsanalyse deutet auf einen verhaltenen bis negativen Trend hin ({avgCompound:F2}). Auf kritische Markt-Berichte sollte geachtet werden.", - _ => $"Das Gesamtsentiment ist neutral ({avgCompound:F2}). Ausgewogene Signale aus der aktuellen Berichterstattung." - }; + string overallLabel = "NEUTRAL"; + if (avgCompound >= 0.15) overallLabel = "POSITIVE"; + else if (avgCompound <= -0.15) overallLabel = "NEGATIVE"; - var updatedDoc = isinDoc with + var summary = new IsinSentimentSummaryDto { Isin = cleanIsin, - CompanyName = !string.IsNullOrWhiteSpace(companyName) ? companyName : isinDoc.CompanyName, - Sector = !string.IsNullOrWhiteSpace(sector) ? sector : isinDoc.Sector, + CompanyName = companyName, + Sector = sector, LastUpdated = nowIso, CurrentSummary = new IsinCurrentSummary { - CompoundScore = Math.Round(avgCompound, 2), - SentimentLabel = label, - AvgConfidence = Math.Round(avgConf, 2), - TotalArticlesAnalyzed = updatedAnalyses.Count, - Text = textSummary + CompoundScore = Math.Round(avgCompound, 4), + SentimentLabel = overallLabel, + AvgConfidence = Math.Round(avgConf, 4), + TotalArticlesAnalyzed = total, + Text = $"Synthesized sentiment across {total} articles is {overallLabel}." }, Analyses = updatedAnalyses }; - await File.WriteAllTextAsync(filePath, JsonSerializer.Serialize(updatedDoc, _jsonOptions)); - _logger.LogInformation("[{Channel}] Updated ISIN summary file: {Path} (Total: {Count}, Score: {Score:F2})", "SentimentChannel", filePath, updatedAnalyses.Count, avgCompound); + var outJson = JsonSerializer.Serialize(summary, _jsonOptions); + await File.WriteAllTextAsync(filePath, outJson); + await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentStorageService] Updated ISIN summary file: {Path} (Total: {Count}, Score: {Score:F2})", filePath, updatedAnalyses.Count, avgCompound); + } + catch (Exception ex) + { + await _finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[SentimentStorageService] Failed to update ISIN summary for {Isin}", cleanIsin); } finally { @@ -301,95 +264,82 @@ public class SentimentStorageService : ISentimentStorageService { if (string.IsNullOrWhiteSpace(sector)) return; - string sanitizedSector = string.Concat(sector.Split(Path.GetInvalidFileNameChars())).Trim(); - string filePath = Path.Combine(_basePath, "sectors", $"{sanitizedSector}.json"); + string cleanSector = sector.Trim().ToLowerInvariant(); + string filePath = Path.Combine(_basePath, "sectors", $"{cleanSector}.json"); var fileLock = FileLocks.GetOrAdd(filePath, _ => new SemaphoreSlim(1, 1)); await fileLock.WaitAsync(); try { - SectorSentimentSummaryDto sectorDoc; - + var analyses = new List(); if (File.Exists(filePath)) { - try + var existingJson = await File.ReadAllTextAsync(filePath); + var existing = JsonSerializer.Deserialize(existingJson, _jsonOptions); + if (existing?.Analyses != null) { - string json = await File.ReadAllTextAsync(filePath); - sectorDoc = JsonSerializer.Deserialize(json, _jsonOptions) ?? new SectorSentimentSummaryDto { Sector = sector }; + analyses.AddRange(existing.Analyses); } - catch - { - sectorDoc = new SectorSentimentSummaryDto { Sector = sector }; - } - } - else - { - sectorDoc = new SectorSentimentSummaryDto { Sector = sector }; } + string cleanIsin = isin.Trim().ToUpperInvariant(); string nowIso = DateTime.UtcNow.ToString("yyyy-MM-ddTHH:mm:ssZ"); - string analysisId = $"sec_sent_{DateTime.UtcNow:yyyyMMdd}_{Random.Shared.Next(100, 999)}"; - var newEntry = new SectorAnalysisEntry + analyses.RemoveAll(a => string.Equals(a.ArticleId, articleId, StringComparison.OrdinalIgnoreCase) && string.Equals(a.RelatedIsin, cleanIsin, StringComparison.OrdinalIgnoreCase)); + + analyses.Add(new SectorAnalysisEntry { - AnalysisId = analysisId, + AnalysisId = $"sec_{Guid.NewGuid():N}", Timestamp = nowIso, - RelatedIsin = isin, + RelatedIsin = cleanIsin, ArticleId = articleId, FinbertResult = finbert - }; + }); - // Duplikate bereinigen (selber Artikel für denselben Sektor) - var updatedAnalyses = sectorDoc.Analyses? - .Where(a => !string.Equals(a.ArticleId, articleId, StringComparison.OrdinalIgnoreCase)) - .ToList() ?? new List(); - - updatedAnalyses.Insert(0, newEntry); - - if (updatedAnalyses.Count > 100) + var cutoff = DateTime.UtcNow.AddDays(-14); + var updatedAnalyses = analyses.Where(a => { - updatedAnalyses = updatedAnalyses.Take(100).ToList(); - } + if (DateTime.TryParse(a.Timestamp, out var ts)) + { + return ts >= cutoff; + } + return true; + }).OrderByDescending(a => a.Timestamp).Take(100).ToList(); - var activeIsins = updatedAnalyses.Select(a => a.RelatedIsin).Where(i => !string.IsNullOrWhiteSpace(i)).Distinct().ToList(); - double avgCompound = updatedAnalyses.Average(a => a.FinbertResult.CompoundScore); - string label = CalculateLabel(avgCompound); + var activeIsins = updatedAnalyses.Select(a => a.RelatedIsin).Where(i => !string.IsNullOrEmpty(i)).Distinct().ToList(); + double totalSectorCompound = updatedAnalyses.Sum(s => s.FinbertResult.CompoundScore); + double avgSectorCompound = updatedAnalyses.Count > 0 ? totalSectorCompound / updatedAnalyses.Count : 0.0; - string overviewText = label switch - { - "POSITIVE" => $"Der Sektor {sector} tendiert insgesamt positiv. Starke Einzelergebnisse stützen den Trend.", - "NEGATIVE" => $"Der Sektor {sector} verzeichnet dämpfende Sentiment-Signale.", - _ => $"Der Sektor {sector} zeigt ein ausgewogenes neutrales Gesamtbild." - }; + string sectorLabel = "NEUTRAL"; + if (avgSectorCompound >= 0.15) sectorLabel = "POSITIVE"; + else if (avgSectorCompound <= -0.15) sectorLabel = "NEGATIVE"; - var updatedDoc = sectorDoc with + var summary = new SectorSentimentSummaryDto { Sector = sector, LastUpdated = nowIso, CurrentSummary = new SectorCurrentSummary { - CompoundScore = Math.Round(avgCompound, 2), - SentimentLabel = label, + CompoundScore = Math.Round(avgSectorCompound, 4), + SentimentLabel = sectorLabel, ActiveIsins = activeIsins, - Text = overviewText + Text = $"Sector {sector} aggregate sentiment: {sectorLabel}." }, Analyses = updatedAnalyses }; - await File.WriteAllTextAsync(filePath, JsonSerializer.Serialize(updatedDoc, _jsonOptions)); - _logger.LogInformation("[{Channel}] Updated Sector summary file: {Path} (Active ISINs: {Count})", "SentimentChannel", filePath, activeIsins.Count); + var outJson = JsonSerializer.Serialize(summary, _jsonOptions); + await File.WriteAllTextAsync(filePath, outJson); + await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentStorageService] Updated Sector summary file: {Path} (Active ISINs: {Count})", filePath, activeIsins.Count); + } + catch (Exception ex) + { + await _finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[SentimentStorageService] Failed to update Sector summary for {Sector}", sector); } finally { fileLock.Release(); } } - - private static string CalculateLabel(double score) => score switch - { - >= 0.15 => "POSITIVE", - <= -0.15 => "NEGATIVE", - _ => "NEUTRAL" - }; } \ No newline at end of file diff --git a/FinlyticSentiment/Util/SentimentMqttClient.cs b/FinlyticSentiment/Util/SentimentMqttClient.cs index 46b6b82..091a54a 100644 --- a/FinlyticSentiment/Util/SentimentMqttClient.cs +++ b/FinlyticSentiment/Util/SentimentMqttClient.cs @@ -1,10 +1,17 @@ +using System; +using System.Collections.Generic; using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; using FinlyticCore.Dtos; using FinlyticCore.Dtos.News; using FinlyticCore.Dtos.Sentiment; +using FinlyticCore.Dtos.Settings; using FinlyticCore.Models; +using FinlyticCore.Services; using FinlyticCore.Util; using FinlyticSentiment.Services; +using FinlyticSentiment.Util; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; @@ -21,9 +28,6 @@ public class SentimentMqttClient : ManagedMqttClient, IHostedService private readonly IConfiguration _configuration; private readonly IServiceScopeFactory _scopeFactory; - /// - /// Initializes a new instance of the class. - /// public SentimentMqttClient( ILogger logger, IConfiguration configuration, @@ -43,12 +47,10 @@ public class SentimentMqttClient : ManagedMqttClient, IHostedService { Host = _configuration["MQTT:Host"] ?? _configuration["MQTT__Host"] ?? "localhost", Port = Convert.ToInt32(_configuration["MQTT:Port"] ?? _configuration["MQTT__Port"] ?? "1883"), - ClientId = - $"{(_configuration["MQTT:ClientId"] ?? _configuration["MQTT__ClientId"] ?? "FinlyticSentiment")}_{Guid.NewGuid()}" + ClientId = $"{(_configuration["MQTT:ClientId"] ?? _configuration["MQTT__ClientId"] ?? "FinlyticSentiment")}_{Guid.NewGuid()}" }; - _logger.LogInformation("[{Channel}] Starting Sentiment MQTT client. Host: {Host}, ClientId: {ClientId}", - "SentimentChannel", config.Host, config.ClientId); + _logger.LogInformation("Starting Sentiment MQTT client. Host: {Host}, ClientId: {ClientId}", config.Host, config.ClientId); await ConnectAsync(config); } @@ -57,7 +59,7 @@ public class SentimentMqttClient : ManagedMqttClient, IHostedService /// public async Task StopAsync(CancellationToken cancellationToken) { - _logger.LogInformation("[{Channel}] Stopping Sentiment MQTT client and disconnecting.", "SentimentChannel"); + _logger.LogInformation("Stopping Sentiment MQTT client and disconnecting."); await DisconnectAsync(); } @@ -69,16 +71,24 @@ public class SentimentMqttClient : ManagedMqttClient, IHostedService /// protected override async Task OnConnectedAsync() { - _logger.LogInformation( - "[{Channel}] Sentiment MQTT client connected. Subscribing to RPC response and broadcast topics...", - "SentimentChannel"); + _logger.LogInformation("Sentiment MQTT client connected. Subscribing to RPC response and broadcast topics..."); await SubscribeAsync("services/response/#"); await SubscribeAsync("services/news/completed"); await SubscribeAsync("services/request/sentiment_GetArticle/#"); await SubscribeAsync("services/request/sentiment_GetIsin/#"); await SubscribeAsync("services/request/sentiment_Analyze/#"); + await SubscribeAsync("services/request/sentiment_settings_GetAll/#"); + await SubscribeAsync("services/request/sentiment_settings_Update/#"); await SubscribeAsync("services/request/health_Ping/#"); await SubscribeAsync("services/config/updated/#"); + + FinlyticCore.Services.FinlyticLogBroadcaster.OnLogPublished = async (logDto) => + { + if (IsConnected && string.Equals(logDto.ServiceName, "FinlyticSentiment", StringComparison.OrdinalIgnoreCase)) + { + await PublishAsync("finlytic/logs/FinlyticSentiment", logDto); + } + }; } /// @@ -86,14 +96,12 @@ public class SentimentMqttClient : ManagedMqttClient, IHostedService { if (string.IsNullOrWhiteSpace(topic)) return; - // 1. Config update events if (topic.StartsWith("services/config/updated", StringComparison.OrdinalIgnoreCase)) { if (topic.EndsWith("FinlyticSentiment", StringComparison.OrdinalIgnoreCase)) { await OnConfigUpdatedAsync(payload); } - return; } @@ -103,13 +111,11 @@ public class SentimentMqttClient : ManagedMqttClient, IHostedService return; } - // Extract correlationId from topic suffix (e.g. services/request/sentiment_GetArticle/{correlationId}) var lastSlash = topic.LastIndexOf('/'); if (lastSlash < 0 || lastSlash >= topic.Length - 1) return; var correlationId = topic.Substring(lastSlash + 1); - // 2. Dispatch to specific channel handlers if (topic.StartsWith("services/request/sentiment_GetArticle", StringComparison.OrdinalIgnoreCase)) { await OnSentimentGetArticleAsync(payload, correlationId); @@ -122,178 +128,205 @@ public class SentimentMqttClient : ManagedMqttClient, IHostedService { await OnSentimentAnalyzeAsync(payload, correlationId); } + else if (topic.StartsWith("services/request/sentiment_settings_GetAll", StringComparison.OrdinalIgnoreCase)) + { + await OnSettingsGetAllAsync(correlationId); + } + else if (topic.StartsWith("services/request/sentiment_settings_Update", StringComparison.OrdinalIgnoreCase)) + { + await OnSettingsUpdateAsync(payload, correlationId); + } else if (topic.StartsWith("services/request/health_Ping", StringComparison.OrdinalIgnoreCase)) { await OnHealthPingAsync(topic, correlationId); } } - /// - /// Handles dynamic service config update events. - /// + private async Task OnSettingsGetAllAsync(string correlationId) + { + using var scope = _scopeFactory.CreateScope(); + var finlyticLogger = scope.ServiceProvider.GetRequiredService>(); + var settingsService = scope.ServiceProvider.GetRequiredService(); + + await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticSentiment] [Settings_GetAll] Retrieving all dynamic settings via reflection [CorrelationId: {CorrelationId}]", correlationId); + try + { + var settings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) }); + var responseTopic = $"services/response/sentiment_settings_GetAll/{correlationId}"; + + await PublishAsync(responseTopic, settings); + await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticSentiment] [Settings_GetAll] Published {Count} settings to '{ResponseTopic}'", settings.Count, responseTopic); + } + catch (Exception ex) + { + await finlyticLogger.LogErrorAsync(SettingKeys.MqttChannel, ex, "[FinlyticSentiment] [Settings_GetAll] Failed to retrieve settings."); + } + } + + private async Task OnSettingsUpdateAsync(string payload, string correlationId) + { + if (string.IsNullOrWhiteSpace(payload)) return; + + using var scope = _scopeFactory.CreateScope(); + var finlyticLogger = scope.ServiceProvider.GetRequiredService>(); + var settingsService = scope.ServiceProvider.GetRequiredService(); + + await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticSentiment] [Settings_Update] Processing settings update RPC [CorrelationId: {CorrelationId}]", correlationId); + try + { + Dictionary? updates = null; + try + { + updates = JsonSerializer.Deserialize>(payload); + } + catch + { + var list = JsonSerializer.Deserialize>(payload); + if (list != null) + { + updates = new Dictionary(); + foreach (var item in list) updates[item.Key] = item.Value; + } + } + + if (updates != null && updates.Count > 0) + { + await settingsService.UpdateSettingsAsync(updates); + await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticSentiment] [Settings_Update] Successfully updated {Count} settings in database and cache.", updates.Count); + } + + var currentSettings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) }); + var responseTopic = $"services/response/sentiment_settings_Update/{correlationId}"; + await PublishAsync(responseTopic, currentSettings); + } + catch (Exception ex) + { + await finlyticLogger.LogErrorAsync(SettingKeys.MqttChannel, ex, "[FinlyticSentiment] [Settings_Update] Failed to update settings."); + } + } + private async Task OnConfigUpdatedAsync(string payload) { - _logger.LogInformation("[{Channel}] [SentimentMqttClient] Received config update event for FinlyticSentiment.", - "SentimentChannel"); try { using var doc = JsonDocument.Parse(payload); if (doc.RootElement.TryGetProperty("settings", out var settingsProp)) { - var dict = JsonSerializer.Deserialize(settingsProp.GetRawText(), typeof(Dictionary), - FinlyticJsonSerializerContext.Default) as Dictionary; + var dict = JsonSerializer.Deserialize>(settingsProp.GetRawText()); if (dict != null && dict.Count > 0) { using var scope = _scopeFactory.CreateScope(); - var settingsDb = scope.ServiceProvider.GetRequiredService(); - await settingsDb.UpdateSettingsFromDictionaryAsync(dict); - _logger.LogInformation( - "[{Channel}] [SentimentMqttClient] Successfully persisted {Count} updated settings for FinlyticSentiment.", - "SentimentChannel", dict.Count); + var settings = scope.ServiceProvider.GetRequiredService(); + await settings.UpdateSettingsAsync(dict); } } } - catch (Exception ex) - { - _logger.LogError(ex, "[{Channel}] [SentimentMqttClient] Error processing MQTT config update event.", - "SentimentChannel"); - } + catch { } } - /// - /// Handles health_Ping RPC requests. - /// private async Task OnHealthPingAsync(string topic, string correlationId) { if (topic.Contains("FinlyticSentiment", StringComparison.OrdinalIgnoreCase) || !topic.Contains("/", StringComparison.OrdinalIgnoreCase)) { string respTopic = $"services/response/health_Ping/{correlationId}"; - await PublishAsync(respTopic, - new ServiceHealthResponse("FinlyticSentiment", "Online", DateTime.UtcNow, "Connected")); - _logger.LogInformation( - "[{Channel}] [SentimentMqttClient] Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", - "SentimentChannel", correlationId); + await PublishAsync(respTopic, new ServiceHealthResponse("FinlyticSentiment", "Online", DateTime.UtcNow, "Connected")); + + using var scope = _scopeFactory.CreateScope(); + var finlyticLogger = scope.ServiceProvider.GetRequiredService>(); + await finlyticLogger.LogInfoAsync(SettingKeys.HealthPingChannel, "[FinlyticSentiment] Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", correlationId); } } - /// - /// Handles broadcasted articles on services/news/completed. - /// private async Task OnNewsCompletedAsync(string payload) { if (string.IsNullOrWhiteSpace(payload)) return; try { - var article = - JsonSerializer.Deserialize(payload, typeof(NewsArticleDto), FinlyticJsonSerializerContext.Default) as - NewsArticleDto; + var article = JsonSerializer.Deserialize(payload, typeof(NewsArticleDto), FinlyticJsonSerializerContext.Default) as NewsArticleDto; if (article != null && article.Id != Guid.Empty && OnArticleReceived != null) { - _logger.LogInformation( - "[{Channel}] Received real-time article broadcast on services/news/completed: {Title} (ID: {Id})", - "SentimentChannel", article.Title, article.Id); + using var scope = _scopeFactory.CreateScope(); + var finlyticLogger = scope.ServiceProvider.GetRequiredService>(); + await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "Received real-time article broadcast on services/news/completed: {Title} (ID: {Id})", article.Title, article.Id); await OnArticleReceived.Invoke(article); } } - catch (Exception ex) - { - _logger.LogError(ex, "[{Channel}] Error parsing broadcasted article on services/news/completed.", - "SentimentChannel"); - } + catch { } } - /// - /// Handles sentiment_GetArticle RPC requests using source-generated DTO deserialization. - /// private async Task OnSentimentGetArticleAsync(string payload, string correlationId) { - _logger.LogInformation( - "[{Channel}] [SentimentMqttClient] Processing RPC sentiment_GetArticle request [CorrelationId: {CorrelationId}]", - "SentimentChannel", correlationId); if (string.IsNullOrWhiteSpace(payload)) return; + using var scope = _scopeFactory.CreateScope(); + var finlyticLogger = scope.ServiceProvider.GetRequiredService>(); + + await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Processing RPC sentiment_GetArticle request [CorrelationId: {CorrelationId}]", correlationId); + try { - var request = - JsonSerializer.Deserialize(payload, typeof(ArticleRequest), FinlyticJsonSerializerContext.Default) as - ArticleRequest; + var request = JsonSerializer.Deserialize(payload, typeof(ArticleRequest), FinlyticJsonSerializerContext.Default) as ArticleRequest; var articleId = request?.ArticleId ?? request?.Id; if (string.IsNullOrWhiteSpace(articleId)) { - _logger.LogWarning("[{Channel}] [SentimentMqttClient] Missing articleId in request payload.", - "SentimentChannel"); + await finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Missing articleId in request payload."); return; } - using var scope = _scopeFactory.CreateScope(); var storageService = scope.ServiceProvider.GetRequiredService(); var sentimentEntry = await storageService.GetArticleSentimentAsync(articleId); string responseTopic = $"services/response/sentiment_GetArticle/{correlationId}"; - _logger.LogInformation( - "[{Channel}] [SentimentMqttClient] Publishing RPC sentiment_GetArticle response for article {ArticleId} to {ResponseTopic}", - "SentimentChannel", articleId, responseTopic); + await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Publishing RPC sentiment_GetArticle response for article {ArticleId} to {ResponseTopic}", articleId, responseTopic); await PublishAsync(responseTopic, sentimentEntry); } catch (Exception ex) { - _logger.LogError(ex, - "[{Channel}] [SentimentMqttClient] Failed to process sentiment_GetArticle RPC request.", - "SentimentChannel"); + await finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[SentimentMqttClient] Failed to process sentiment_GetArticle RPC request."); } } - /// - /// Handles sentiment_GetIsin RPC requests using source-generated DTO deserialization. - /// private async Task OnSentimentGetIsinAsync(string payload, string correlationId) { - _logger.LogInformation( - "[{Channel}] [SentimentMqttClient] Processing RPC sentiment_GetIsin request [CorrelationId: {CorrelationId}]", - "SentimentChannel", correlationId); if (string.IsNullOrWhiteSpace(payload)) return; + using var scope = _scopeFactory.CreateScope(); + var finlyticLogger = scope.ServiceProvider.GetRequiredService>(); + + await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Processing RPC sentiment_GetIsin request [CorrelationId: {CorrelationId}]", correlationId); + try { - var request = - JsonSerializer.Deserialize(payload, typeof(IsinRequest), FinlyticJsonSerializerContext.Default) as - IsinRequest; + var request = JsonSerializer.Deserialize(payload, typeof(IsinRequest), FinlyticJsonSerializerContext.Default) as IsinRequest; var isin = request?.Isin; if (string.IsNullOrWhiteSpace(isin)) { - _logger.LogWarning("[{Channel}] [SentimentMqttClient] Missing ISIN in request payload.", - "SentimentChannel"); + await finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Missing ISIN in request payload."); return; } - using var scope = _scopeFactory.CreateScope(); var storageService = scope.ServiceProvider.GetRequiredService(); var isinSummary = await storageService.GetIsinSummaryAsync(isin); string responseTopic = $"services/response/sentiment_GetIsin/{correlationId}"; - _logger.LogInformation( - "[{Channel}] [SentimentMqttClient] Publishing RPC sentiment_GetIsin response for ISIN {Isin} to {ResponseTopic}", - "SentimentChannel", isin, responseTopic); + await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Publishing RPC sentiment_GetIsin response for ISIN {Isin} to {ResponseTopic}", isin, responseTopic); await PublishAsync(responseTopic, isinSummary); } catch (Exception ex) { - _logger.LogError(ex, "[{Channel}] [SentimentMqttClient] Failed to process sentiment_GetIsin RPC request.", - "SentimentChannel"); + await finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[SentimentMqttClient] Failed to process sentiment_GetIsin RPC request."); } } - /// - /// Handles manual/forced sentiment_Analyze RPC requests using existing analyzer and storage services. - /// private async Task OnSentimentAnalyzeAsync(string payload, string correlationId) { - _logger.LogInformation( - "[{Channel}] [SentimentMqttClient] Processing RPC sentiment_Analyze request [CorrelationId: {CorrelationId}]", - "SentimentChannel", correlationId); + using var scope = _scopeFactory.CreateScope(); + var finlyticLogger = scope.ServiceProvider.GetRequiredService>(); + + await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Processing RPC sentiment_Analyze request [CorrelationId: {CorrelationId}]", correlationId); string responseTopic = $"services/response/sentiment_Analyze/{correlationId}"; if (string.IsNullOrWhiteSpace(payload)) @@ -304,33 +337,24 @@ public class SentimentMqttClient : ManagedMqttClient, IHostedService try { - var request = - JsonSerializer.Deserialize(payload, typeof(AnalyzeSentimentRequest), - FinlyticJsonSerializerContext.Default) as AnalyzeSentimentRequest; + var request = JsonSerializer.Deserialize(payload, typeof(AnalyzeSentimentRequest), FinlyticJsonSerializerContext.Default) as AnalyzeSentimentRequest; - if (request == null || - (string.IsNullOrWhiteSpace(request.ArticleId) && string.IsNullOrWhiteSpace(request.Isin))) + if (request == null || (string.IsNullOrWhiteSpace(request.ArticleId) && string.IsNullOrWhiteSpace(request.Isin))) { - _logger.LogWarning( - "[{Channel}] [SentimentMqttClient] Missing ArticleId or Isin in sentiment_Analyze request payload.", - "SentimentChannel"); + await finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Missing ArticleId or Isin in sentiment_Analyze request payload."); await PublishAsync(responseTopic, (object?)null); return; } - using var scope = _scopeFactory.CreateScope(); var storageService = scope.ServiceProvider.GetRequiredService(); var analyzerService = scope.ServiceProvider.GetRequiredService(); object? result = null; - // Fall 1: Manuelle Analyse für einen einzelnen Artikel if (!string.IsNullOrWhiteSpace(request.ArticleId)) { var cleanArticleId = request.ArticleId.Trim(); - _logger.LogInformation( - "[{Channel}] [SentimentMqttClient] Processing article analysis for ArticleId: {ArticleId} (ForceReload: {ForceReload})", - "SentimentChannel", cleanArticleId, request.ForceReload); + await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Processing article analysis for ArticleId: {ArticleId} (ForceReload: {ForceReload})", cleanArticleId, request.ForceReload); IsinAnalysisEntry? existingEntry = null; @@ -345,7 +369,6 @@ public class SentimentMqttClient : ManagedMqttClient, IHostedService } else { - // Artikel per RPC von FinlyticNews abfragen if (Guid.TryParse(cleanArticleId, out var articleGuid)) { var article = await SendRpcRequestAsync( @@ -357,17 +380,13 @@ public class SentimentMqttClient : ManagedMqttClient, IHostedService { var finbertResult = await analyzerService.AnalyzeArticleAsync(article); - // 🎯 NULL-CHECK: Falls Analyse fehlschlägt/null liefert -> abbrechen if (finbertResult == null) { - _logger.LogWarning( - "[{Channel}] [SentimentMqttClient] FinBERT analysis returned NULL for article {ArticleId}. Aborting manual analysis.", - "SentimentChannel", cleanArticleId); + await finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] FinBERT analysis returned NULL for article {ArticleId}. Aborting manual analysis.", cleanArticleId); await PublishAsync(responseTopic, (object?)null); return; } - // Speichern & Summaries aktualisieren await storageService.SaveArticleSentimentAsync(article, finbertResult); if (article.MatchedAssets != null && article.MatchedAssets.Count > 0) @@ -391,86 +410,62 @@ public class SentimentMqttClient : ManagedMqttClient, IHostedService } } - // FinlyticNews über Re-Analyse informieren await UpdateArticleStatusAsync(article.Id, "Analyzed"); result = await storageService.GetArticleSentimentAsync(cleanArticleId); } else { - _logger.LogWarning( - "[{Channel}] [SentimentMqttClient] Could not retrieve article {ArticleId} from FinlyticNews for re-analysis.", - "SentimentChannel", cleanArticleId); + await finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Could not retrieve article {ArticleId} from FinlyticNews for re-analysis.", cleanArticleId); } } } } - // Fall 2: ISIN Gesamtsummary anfordern else if (!string.IsNullOrWhiteSpace(request.Isin)) { var cleanIsin = request.Isin.Trim(); - _logger.LogInformation("[{Channel}] [SentimentMqttClient] Fetching sentiment summary for ISIN: {Isin}", - "SentimentChannel", cleanIsin); + await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Fetching sentiment summary for ISIN: {Isin}", cleanIsin); result = await storageService.GetIsinSummaryAsync(cleanIsin); } - _logger.LogInformation( - "[{Channel}] [SentimentMqttClient] Publishing RPC sentiment_Analyze response to {ResponseTopic}", - "SentimentChannel", responseTopic); + await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Publishing RPC sentiment_Analyze response to {ResponseTopic}", responseTopic); await PublishAsync(responseTopic, result); } catch (Exception ex) { - _logger.LogError(ex, "[{Channel}] [SentimentMqttClient] Failed to process sentiment_Analyze RPC request.", - "SentimentChannel"); + await finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[SentimentMqttClient] Failed to process sentiment_Analyze RPC request."); await PublishAsync(responseTopic, (object?)null); } } - /// - /// Requests pending news articles from FinlyticNews via MQTT RPC. - /// - /// The maximum number of articles to request (capped at 10). - /// A list of pending news article DTOs. public async Task> GetPendingArticlesAsync(int limit = 10) { try { var payload = new LimitRequest(Math.Min(limit, 10)); - var articles = - await SendRpcRequestAsync, LimitRequest>("news_GetPending", payload, - TimeSpan.FromSeconds(10)); + var articles = await SendRpcRequestAsync, LimitRequest>("news_GetPending", payload, TimeSpan.FromSeconds(10)); return articles ?? []; } catch (Exception ex) { - _logger.LogError(ex, "[{Channel}] Error executing MQTT RPC for news_GetPending.", "SentimentChannel"); + _logger.LogError(ex, "Error executing MQTT RPC for news_GetPending."); } return []; } - /// - /// Dispatches an RPC request to update the article status in FinlyticNews (e.g. to "Analyzed"). - /// - /// The article identifier. - /// The target status string (default "Analyzed"). - /// True if the status update succeeded; otherwise, false. public async Task UpdateArticleStatusAsync(Guid id, string status = "Analyzed") { try { var request = new UpdateNewsStatusRequest(id, status); - var response = - await SendRpcRequestAsync("news_UpdateStatus", - request, TimeSpan.FromSeconds(8)); + var response = await SendRpcRequestAsync("news_UpdateStatus", request, TimeSpan.FromSeconds(8)); return response?.Success ?? false; } catch (Exception ex) { - _logger.LogError(ex, "[{Channel}] Error executing MQTT RPC for news_UpdateStatus (ID: {Id}).", - "SentimentChannel", id); + _logger.LogError(ex, "Error executing MQTT RPC for news_UpdateStatus (ID: {Id}).", id); } return false; diff --git a/FinlyticSentiment/Util/SettingKeys.cs b/FinlyticSentiment/Util/SettingKeys.cs new file mode 100644 index 0000000..0a35dc3 --- /dev/null +++ b/FinlyticSentiment/Util/SettingKeys.cs @@ -0,0 +1,19 @@ +using FinlyticCore.Models.Settings; + +namespace FinlyticSentiment.Util; + +public static class SettingKeys +{ + // --- Logging-Kanäle --- + public static readonly SettingKey SentimentChannel = new("Logging.Channel.Sentiment", true); + public static readonly SettingKey MqttChannel = new("Logging.Channel.MQTT", true); + public static readonly SettingKey HealthPingChannel = new("Logging.Channel.Health", true); + + // --- FinBERT & Modell-Parameter --- + public static readonly SettingKey MaxBatchSize = new("FinBert.MaxBatchSize", 10); + public static readonly SettingKey MinimumConfidenceThreshold = new("FinBert.MinConfidenceThreshold", 0.60); + public static readonly SettingKey TimeoutSeconds = new("FinBert.TimeoutSeconds", 30); + public static readonly SettingKey SentimentWindowDays = new("Sentiment.WindowDays", 14); + public static readonly SettingKey DecayFactorPerDay = new("Sentiment.DecayFactorPerDay", 0.90); + public static readonly SettingKey EnableAutoSummarization = new("Feature.EnableAutoSummarization", true); +}