diff --git a/FinlyticSentiment/Database/SentimentDbContext.cs b/FinlyticSentiment/Database/SentimentDbContext.cs new file mode 100644 index 0000000..85b0240 --- /dev/null +++ b/FinlyticSentiment/Database/SentimentDbContext.cs @@ -0,0 +1,29 @@ +using FinlyticSentiment.Entities; +using Microsoft.EntityFrameworkCore; + +namespace FinlyticSentiment.Database; + +/// +/// EF Core DbContext for managing FinlyticSentiment settings in PostgreSQL. +/// +public class SentimentDbContext : DbContext +{ + public SentimentDbContext(DbContextOptions options) : base(options) + { + } + + public DbSet Settings => Set(); + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + base.OnModelCreating(modelBuilder); + + modelBuilder.Entity(entity => + { + entity.ToTable("sentiment_settings"); + entity.HasKey(e => e.Id); + entity.Property(e => e.GermanWebhookUrl).HasMaxLength(500); + entity.Property(e => e.EnglishWebhookUrl).HasMaxLength(500); + }); + } +} diff --git a/FinlyticSentiment/Dockerfile b/FinlyticSentiment/Dockerfile new file mode 100644 index 0000000..4a00792 --- /dev/null +++ b/FinlyticSentiment/Dockerfile @@ -0,0 +1,22 @@ +FROM mcr.microsoft.com/dotnet/runtime:10.0 AS base +USER app +WORKDIR /app + +FROM mcr.microsoft.com/dotnet/sdk:10.0 AS build +ARG BUILD_CONFIGURATION=Release +WORKDIR /src +COPY ["FinlyticSentiment/FinlyticSentiment.csproj", "FinlyticSentiment/"] +COPY ["FinlyticCore/FinlyticCore.csproj", "FinlyticCore/"] +RUN dotnet restore "FinlyticSentiment/FinlyticSentiment.csproj" +COPY . . +WORKDIR "/src/FinlyticSentiment" +RUN dotnet build "FinlyticSentiment.csproj" -c $BUILD_CONFIGURATION -o /app/build + +FROM build AS publish +ARG BUILD_CONFIGURATION=Release +RUN dotnet publish "FinlyticSentiment.csproj" -c $BUILD_CONFIGURATION -o /app/publish /p:UseAppHost=false + +FROM base AS final +WORKDIR /app +COPY --from=publish /app/publish . +ENTRYPOINT ["dotnet", "FinlyticSentiment.dll"] diff --git a/FinlyticSentiment/Entities/SentimentSettingsEntity.cs b/FinlyticSentiment/Entities/SentimentSettingsEntity.cs new file mode 100644 index 0000000..6377eb0 --- /dev/null +++ b/FinlyticSentiment/Entities/SentimentSettingsEntity.cs @@ -0,0 +1,23 @@ +using System; + +namespace FinlyticSentiment.Entities; + +/// +/// Entity representing runtime operational settings for FinlyticSentiment stored in PostgreSQL. +/// +public class SentimentSettingsEntity +{ + public Guid Id { get; set; } = Guid.NewGuid(); + + public double MinConfidenceScore { get; set; } = 0.70; + + public int MaxBatchSize { get; set; } = 10; + + public int SweepIntervalMinutes { get; set; } = 2; + + public string GermanWebhookUrl { get; set; } = "https://n8n.kleidukos.me/webhook/sentiment/de"; + + public string EnglishWebhookUrl { get; set; } = "https://n8n.kleidukos.me/webhook/sentiment/en"; + + public DateTime UpdatedAt { get; set; } = DateTime.UtcNow; +} diff --git a/FinlyticSentiment/FinlyticSentiment.csproj b/FinlyticSentiment/FinlyticSentiment.csproj new file mode 100644 index 0000000..7a02b00 --- /dev/null +++ b/FinlyticSentiment/FinlyticSentiment.csproj @@ -0,0 +1,26 @@ + + + + net10.0 + enable + enable + Linux + + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + + + + + + + diff --git a/FinlyticSentiment/Migrations/20260801090401_InitialSentimentSettings.Designer.cs b/FinlyticSentiment/Migrations/20260801090401_InitialSentimentSettings.Designer.cs new file mode 100644 index 0000000..febe352 --- /dev/null +++ b/FinlyticSentiment/Migrations/20260801090401_InitialSentimentSettings.Designer.cs @@ -0,0 +1,63 @@ +// +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("20260801090401_InitialSentimentSettings")] + partial class InitialSentimentSettings + { + /// + 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("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/20260801090401_InitialSentimentSettings.cs b/FinlyticSentiment/Migrations/20260801090401_InitialSentimentSettings.cs new file mode 100644 index 0000000..3663991 --- /dev/null +++ b/FinlyticSentiment/Migrations/20260801090401_InitialSentimentSettings.cs @@ -0,0 +1,39 @@ +using System; +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace FinlyticSentiment.Migrations +{ + /// + public partial class InitialSentimentSettings : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.CreateTable( + name: "sentiment_settings", + columns: table => new + { + Id = table.Column(type: "uuid", nullable: false), + MinConfidenceScore = table.Column(type: "double precision", nullable: false), + MaxBatchSize = table.Column(type: "integer", nullable: false), + SweepIntervalMinutes = table.Column(type: "integer", nullable: false), + GermanWebhookUrl = table.Column(type: "character varying(500)", maxLength: 500, nullable: false), + EnglishWebhookUrl = table.Column(type: "character varying(500)", maxLength: 500, nullable: false), + UpdatedAt = table.Column(type: "timestamp with time zone", nullable: false) + }, + constraints: table => + { + table.PrimaryKey("PK_sentiment_settings", x => x.Id); + }); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropTable( + name: "sentiment_settings"); + } + } +} diff --git a/FinlyticSentiment/Migrations/SentimentDbContextModelSnapshot.cs b/FinlyticSentiment/Migrations/SentimentDbContextModelSnapshot.cs new file mode 100644 index 0000000..5d0acd5 --- /dev/null +++ b/FinlyticSentiment/Migrations/SentimentDbContextModelSnapshot.cs @@ -0,0 +1,60 @@ +// +using System; +using FinlyticSentiment.Database; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; +using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata; + +#nullable disable + +namespace FinlyticSentiment.Migrations +{ + [DbContext(typeof(SentimentDbContext))] + partial class SentimentDbContextModelSnapshot : ModelSnapshot + { + protected override void BuildModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasAnnotation("ProductVersion", "10.0.9") + .HasAnnotation("Relational:MaxIdentifierLength", 63); + + NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder); + + 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/Program.cs b/FinlyticSentiment/Program.cs new file mode 100644 index 0000000..64c1a54 --- /dev/null +++ b/FinlyticSentiment/Program.cs @@ -0,0 +1,44 @@ +using FinlyticSentiment.Database; +using FinlyticSentiment.Services; +using FinlyticSentiment.Util; +using Microsoft.EntityFrameworkCore; + +var builder = Host.CreateApplicationBuilder(args); + +// Register PostgreSQL DbContext for settings persistence +builder.Services.AddDbContext(options => + options.UseNpgsql(builder.Configuration.GetConnectionString("DefaultConnection"))); + +// Register HttpClient +builder.Services.AddHttpClient(); + +// Register Service interfaces and co-located implementations +builder.Services.AddScoped(); +builder.Services.AddSingleton(); +builder.Services.AddSingleton(); + +// Register MQTT Client (as singleton hosted service) +builder.Services.AddSingleton(); +builder.Services.AddHostedService(sp => sp.GetRequiredService()); + +// Register Sentiment Background Worker +builder.Services.AddHostedService(); + +var host = builder.Build(); + +// Auto-migrate database on startup +using (var scope = host.Services.CreateScope()) +{ + try + { + var db = scope.ServiceProvider.GetRequiredService(); + await db.Database.MigrateAsync(); + } + catch (Exception ex) + { + var logger = scope.ServiceProvider.GetRequiredService>(); + logger.LogError(ex, "An error occurred during database migration for FinlyticSentiment on startup."); + } +} + +await host.RunAsync(); diff --git a/FinlyticSentiment/Project.md b/FinlyticSentiment/Project.md new file mode 100644 index 0000000..0458002 --- /dev/null +++ b/FinlyticSentiment/Project.md @@ -0,0 +1,34 @@ +# Finlytic Sentiment Service + +Finlytic Sentiment is a C# microservice dedicated to real-time AI sentiment analysis of financial news. It consumes pending news articles, runs FinBERT neural model evaluations, and aggregates sentiment scores at the asset (ISIN) and sector level. + +--- + +## Core Features & Architecture + +1. **FinBERT AI Integration**: + - Evaluates positive, negative, and neutral sentiment probabilities (`positiveProbability`, `negativeProbability`, `neutralProbability`, `compoundScore`). + +2. **ISIN & Sector Sentiment Aggregation**: + - Aggregates sentiment scores per financial asset (`IsinSentimentSummaryDto`) and sector (`SectorSentimentSummaryDto`). + +3. **Background Worker Engine**: + - Runs a 5-minute background loop (`SentimentBackgroundService`) polling pending articles, processing FinBERT evaluations, and publishing results over MQTT. + +4. **MQTT Event Channels**: + - Publishes updates to `finlytic/sentiment/result` and responds to RPC requests on `services/request/sentiment_GetArticle/#` and `services/request/sentiment_GetIsin/#`. + +--- + +## Feature Status + +### Implemented Features +- [x] FinBERT AI Sentiment Evaluation Service (`IFinBertAnalyzerService`). +- [x] Sentiment Storage & In-Memory Aggregation (`ISentimentStorageService`). +- [x] 5-minute Background Worker loop (`SentimentBackgroundService`). +- [x] Zero-Allocation MQTT serialization via `FinlyticJsonSerializerContext`. +- [x] Pure Worker Service architecture (`Host.CreateApplicationBuilder`, no Kestrel HTTP server). + +### Planned Features +- [ ] Historical sentiment trend charting (multi-month sentiment drift per asset). +- [ ] Financial entity sentiment impact correlation model. diff --git a/FinlyticSentiment/Services/FinBertAnalyzerService.cs b/FinlyticSentiment/Services/FinBertAnalyzerService.cs new file mode 100644 index 0000000..5504ede --- /dev/null +++ b/FinlyticSentiment/Services/FinBertAnalyzerService.cs @@ -0,0 +1,204 @@ +using System.Net.Http.Json; +using System.Text.Json; +using FinlyticCore.Dtos.News; +using FinlyticCore.Dtos.Sentiment; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.Logging; + +namespace FinlyticSentiment.Services; + +/// +/// Defines the sentiment analysis contract for evaluating news articles via FinBERT webhooks. +/// +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); +} + +/// +/// Implementation of the sentiment analysis contract for evaluating news articles via FinBERT webhooks. +/// +public class FinBertAnalyzerService : IFinBertAnalyzerService +{ + private readonly HttpClient _httpClient; + private readonly IServiceScopeFactory _scopeFactory; + private readonly ILogger _logger; + + /// + /// 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) + { + _httpClient = httpClient; + _scopeFactory = scopeFactory; + _logger = logger; + } + + /// + /// Analyzes a news article using language-targeted FinBERT webhooks and returns structured metrics. + /// + public async Task AnalyzeArticleAsync(NewsArticleDto article) + { + ArgumentNullException.ThrowIfNull(article); + + string targetUrl; + double minConfidence; + using (var scope = _scopeFactory.CreateScope()) + { + var settingsDb = scope.ServiceProvider.GetRequiredService(); + var settings = await settingsDb.GetSettingsAsync(); + targetUrl = string.Equals(article.Language, "en", StringComparison.OrdinalIgnoreCase) + ? settings.EnglishWebhookUrl + : settings.GermanWebhookUrl; + 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); + + var requestBody = new + { + article_id = article.Id.ToString(), + title = article.Title, + summary = article.Summary ?? string.Empty, + content = article.ContentRaw ?? string.Empty, + source = article.Author ?? "FinlyticNews", + source_url = article.SourceUrl, + published_at = article.PublishedAt.ToString("o") + }; + + try + { + using var response = await _httpClient.PostAsJsonAsync(targetUrl, requestBody); + if (response.IsSuccessStatusCode) + { + var content = await response.Content.ReadAsStringAsync(); + if (!string.IsNullOrWhiteSpace(content)) + { + 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) + root = jsonChild; + else if (root.TryGetProperty("output", out var outChild) && outChild.ValueKind == JsonValueKind.Object) + root = outChild; + else if (root.TryGetProperty("data", out var dataChild) && dataChild.ValueKind == JsonValueKind.Object) + root = dataChild; + else if (root.TryGetProperty("body", out var bodyChild) && bodyChild.ValueKind == JsonValueKind.Object) + root = bodyChild; + } + + if (root.ValueKind == JsonValueKind.Object) + { + string rawLabel = GetStringProp(root, "label") + ?? GetStringProp(root, "sentiment_label") + ?? GetStringProp(root, "sentiment") + ?? "NEUTRAL"; + double compoundScore = GetDoubleProp(root, "compound_score") + ?? GetDoubleProp(root, "compoundScore") + ?? GetDoubleProp(root, "score") + ?? 0.0; + double confidence = GetDoubleProp(root, "confidence") + ?? GetDoubleProp(root, "confidence_score") + ?? 0.5; + string snippet = GetStringProp(root, "summary_snippet") + ?? GetStringProp(root, "summary") + ?? GetStringProp(root, "text") + ?? article.Summary + ?? article.Title; + + double pos = 0.0, neg = 0.0, neu = 1.0; + if (root.TryGetProperty("probabilities", out var probsElem) && probsElem.ValueKind == JsonValueKind.Object) + { + pos = GetDoubleProp(probsElem, "positive") ?? pos; + neg = GetDoubleProp(probsElem, "negative") ?? neg; + neu = GetDoubleProp(probsElem, "neutral") ?? neu; + } + + // Normalize German vs English labels + string label = rawLabel.Trim().ToUpperInvariant() switch + { + "POSITIV" or "POSITIVE" => "POSITIVE", + "NEGATIV" or "NEGATIVE" => "NEGATIVE", + _ => "NEUTRAL" + }; + + // If compoundScore is 0 but probabilities or label indicate sentiment, compute compoundScore + if (Math.Abs(compoundScore) < 0.001) + { + if (pos > 0 || neg > 0) + { + compoundScore = pos - neg; + } + else if (label == "POSITIVE") + { + compoundScore = 0.8; + } + else if (label == "NEGATIVE") + { + compoundScore = -0.8; + } + } + + return new FinBertResultDto + { + Label = label, + CompoundScore = Math.Round(compoundScore, 4), + Confidence = Math.Round(confidence, 4), + Probabilities = new FinBertProbabilities + { + Positive = Math.Round(pos, 4), + Negative = Math.Round(neg, 4), + Neutral = Math.Round(neu, 4) + }, + SummarySnippet = snippet + }; + } + + } + } + + _logger.LogWarning("[{Channel}] n8n Webhook returned non-success status: {StatusCode}. Falling back to rule analyzer.", "SentimentChannel", response.StatusCode); + } + catch (Exception ex) + { + _logger.LogError(ex, "[{Channel}] Failed to call n8n sentiment webhook. Executing fallback sentiment analyzer.", "SentimentChannel"); + } + + return null; + } + + private static string? GetStringProp(JsonElement elem, string propName) + { + return elem.TryGetProperty(propName, out var prop) && prop.ValueKind == JsonValueKind.String ? prop.GetString() : null; + } + + private static double? GetDoubleProp(JsonElement elem, string propName) + { + if (elem.TryGetProperty(propName, out var prop)) + { + if (prop.ValueKind == JsonValueKind.Number && prop.TryGetDouble(out var d)) return d; + if (prop.ValueKind == JsonValueKind.String && double.TryParse(prop.GetString(), out var parsed)) return parsed; + } + return null; + } +} diff --git a/FinlyticSentiment/Services/SentimentBackgroundService.cs b/FinlyticSentiment/Services/SentimentBackgroundService.cs new file mode 100644 index 0000000..87a5286 --- /dev/null +++ b/FinlyticSentiment/Services/SentimentBackgroundService.cs @@ -0,0 +1,216 @@ +using System.Collections.Concurrent; +using FinlyticCore.Dtos.News; +using FinlyticSentiment.Util; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; + +namespace FinlyticSentiment.Services; + +/// +/// Background hosted worker executing periodic sentiment analysis sweeps on pending news articles +/// and processing real-time article broadcasts. +/// +public class SentimentBackgroundService : BackgroundService +{ + private readonly SentimentMqttClient _mqttClient; + private readonly IFinBertAnalyzerService _analyzer; + private readonly ISentimentStorageService _storage; + private readonly IServiceScopeFactory _scopeFactory; + private readonly ILogger _logger; + + // 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) + { + _mqttClient = mqttClient; + _analyzer = analyzer; + _storage = storage; + _scopeFactory = scopeFactory; + _logger = logger; + } + + /// + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + _logger.LogInformation("[{Channel}] FinlyticSentiment Background Service started.", "SentimentChannel"); + + // 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; + + using (var scope = _scopeFactory.CreateScope()) + { + var settingsDb = scope.ServiceProvider.GetRequiredService(); + var settings = await settingsDb.GetSettingsAsync(); + maxBatchSize = settings.MaxBatchSize; + sweepIntervalMinutes = settings.SweepIntervalMinutes; + } + + var interval = TimeSpan.FromMinutes(Math.Max(1, sweepIntervalMinutes)); + + try + { + await PerformSentimentSweepAsync(maxBatchSize, stoppingToken); + } + 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"); + } + + _logger.LogInformation("[{Channel}] Waiting {Minutes} minute(s) until next sentiment sweep...", + "SentimentChannel", interval.TotalMinutes); + + // Verwendet PeriodicTimer oder CancellationToken-resistenten Delay + using var timer = new PeriodicTimer(interval); + try + { + await timer.WaitForNextTickAsync(stoppingToken); + } + catch (OperationCanceledException) + { + break; + } + } + + _logger.LogInformation("[{Channel}] FinlyticSentiment Background Service is shutting down gracefully.", + "SentimentChannel"); + } + + /// + /// 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); + + List pendingArticles = await _mqttClient.GetPendingArticlesAsync(limit: maxBatchSize); + if (pendingArticles.Count == 0) + { + _logger.LogInformation("[{Channel}] No pending news articles found in FinlyticNews.", "SentimentChannel"); + return; + } + + _logger.LogInformation("[{Channel}] Retrieved {Count} pending article(s) for FinBERT sentiment evaluation.", + "SentimentChannel", pendingArticles.Count); + + foreach (var article in pendingArticles) + { + if (cancellationToken.IsCancellationRequested) break; + await ProcessSingleArticleAsync(article, cancellationToken); + } + } + + /// + /// 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); + return; + } + + try + { + _logger.LogInformation("[{Channel}] Evaluating FinBERT sentiment for article: '{Title}' (ID: {Id})", + "SentimentChannel", 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); + 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) + { + if (cancellationToken.IsCancellationRequested) break; + if (string.IsNullOrWhiteSpace(asset.Isin)) continue; + + await _storage.UpdateIsinSummaryAsync( + asset.Isin, + asset.Name, + "General", + article, + finbert); + + await _storage.UpdateSectorSummaryAsync( + "General", + asset.Isin, + article.Id.ToString(), + finbert); + } + } + + 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); + } + else + { + _logger.LogWarning( + "[{Channel}] Failed to confirm status update to 'Analyzed' in FinlyticNews for article: {Id}", + "SentimentChannel", article.Id); + } + } + catch (Exception ex) + { + _logger.LogError(ex, "[{Channel}] Error processing sentiment for article: {Id} ({Title})", + "SentimentChannel", article.Id, article.Title); + } + finally + { + // Lock nach der Verarbeitung immer freigeben + ProcessingArticles.TryRemove(article.Id, out _); + } + } +} \ No newline at end of file diff --git a/FinlyticSentiment/Services/SentimentStorageService.cs b/FinlyticSentiment/Services/SentimentStorageService.cs new file mode 100644 index 0000000..33ca733 --- /dev/null +++ b/FinlyticSentiment/Services/SentimentStorageService.cs @@ -0,0 +1,395 @@ +using System.Collections.Concurrent; +using System.Text.Encodings.Web; +using System.Text.Json; +using FinlyticCore.Dtos.News; +using FinlyticCore.Dtos.Sentiment; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.Logging; + +namespace FinlyticSentiment.Services; + +/// +/// Defines the persistence contract for maintaining two-stage ISIN and Sector JSON sentiment summaries in the file system. +/// +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); +} + +public class SentimentStorageService : ISentimentStorageService +{ + private static readonly ConcurrentDictionary FileLocks = new(); + + private readonly ILogger _logger; + private readonly string _basePath; + private readonly JsonSerializerOptions _jsonOptions; + + public SentimentStorageService(IConfiguration configuration, ILogger logger) + { + _logger = logger; + _basePath = configuration["Storage:SummariesPath"] ?? "data/summaries"; + + Directory.CreateDirectory(Path.Combine(_basePath, "isin")); + Directory.CreateDirectory(Path.Combine(_basePath, "sectors")); + Directory.CreateDirectory(Path.Combine(_basePath, "articles")); + + _jsonOptions = new JsonSerializerOptions + { + WriteIndented = true, + Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping + }; + } + + /// + public async Task SaveArticleSentimentAsync(NewsArticleDto article, FinBertResultDto finbert) + { + if (article == null || article.Id == Guid.Empty) return; + + string cleanId = article.Id.ToString(); + string filePath = Path.Combine(_basePath, "articles", $"{cleanId}.json"); + + var fileLock = FileLocks.GetOrAdd(filePath, _ => new SemaphoreSlim(1, 1)); + await fileLock.WaitAsync(); + + 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, + Timestamp = nowIso, + Article = new IsinAnalysisArticleRef + { + ArticleId = cleanId, + Title = article.Title ?? "", + Source = article.Author ?? "FinlyticNews", + PublishedAt = article.PublishedAt.ToString("yyyy-MM-ddTHH:mm:ssZ") + }, + FinbertResult = finbert, + SummarySnippet = finbert.SummarySnippet ?? article.Summary ?? article.Title ?? "" + }; + + string json = JsonSerializer.Serialize(entry, _jsonOptions); + await File.WriteAllTextAsync(filePath, json); + _logger.LogInformation("[{Channel}] Successfully saved article sentiment file: {Path}", "SentimentChannel", filePath); + } + catch (Exception ex) + { + _logger.LogError(ex, "[{Channel}] Failed to write article sentiment file: {Path}", "SentimentChannel", filePath); + } + finally + { + fileLock.Release(); + } + } + + /// + public async Task GetArticleSentimentAsync(string articleId) + { + if (string.IsNullOrWhiteSpace(articleId)) return null; + + var cleanId = articleId.Trim(); + var articleFilePath = Path.Combine(_basePath, "articles", $"{cleanId}.json"); + + // 1. Primärer Lookup + if (File.Exists(articleFilePath)) + { + var fileLock = FileLocks.GetOrAdd(articleFilePath, _ => new SemaphoreSlim(1, 1)); + await fileLock.WaitAsync(); + try + { + var json = await File.ReadAllTextAsync(articleFilePath); + return JsonSerializer.Deserialize(json, _jsonOptions); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "[{Channel}] Failed to read article sentiment file: {Path}", "SentimentChannel", articleFilePath); + } + finally + { + fileLock.Release(); + } + } + + // 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; + } + + /// + public async Task GetIsinSummaryAsync(string isin) + { + if (string.IsNullOrWhiteSpace(isin)) return null; + + var cleanIsin = isin.Trim(); + var filePath = Path.Combine(_basePath, "isin", $"{cleanIsin}.json"); + + if (!File.Exists(filePath)) return null; + + var fileLock = FileLocks.GetOrAdd(filePath, _ => new SemaphoreSlim(1, 1)); + await fileLock.WaitAsync(); + + try + { + var json = await File.ReadAllTextAsync(filePath); + return JsonSerializer.Deserialize(json, _jsonOptions); + } + catch (Exception ex) + { + _logger.LogError(ex, "[{Channel}] Error reading ISIN summary file: {Path}", "SentimentChannel", filePath); + return null; + } + finally + { + fileLock.Release(); + } + } + + /// + public async Task UpdateIsinSummaryAsync(string isin, string companyName, string sector, NewsArticleDto article, FinBertResultDto finbert) + { + if (string.IsNullOrWhiteSpace(isin) || article == null) return; + + string cleanIsin = isin.Trim(); + string filePath = Path.Combine(_basePath, "isin", $"{cleanIsin}.json"); + + var fileLock = FileLocks.GetOrAdd(filePath, _ => new SemaphoreSlim(1, 1)); + await fileLock.WaitAsync(); + + try + { + IsinSentimentSummaryDto isinDoc; + + if (File.Exists(filePath)) + { + try + { + string json = await File.ReadAllTextAsync(filePath); + isinDoc = JsonSerializer.Deserialize(json, _jsonOptions) ?? new IsinSentimentSummaryDto { Isin = cleanIsin }; + } + catch + { + isinDoc = new IsinSentimentSummaryDto { Isin = cleanIsin }; + } + } + else + { + isinDoc = new IsinSentimentSummaryDto { Isin = cleanIsin }; + } + + 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, + 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") + }, + FinbertResult = finbert, + SummarySnippet = finbert.SummarySnippet ?? article.Summary ?? article.Title ?? "" + }; + + // 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(); + + // Neuen Eintrag oben einfügen + updatedAnalyses.Insert(0, newEntry); + + // Capping: Maximal die letzten 100 Analysen aufheben (verhindert gigantische JSON-Dateien) + if (updatedAnalyses.Count > 100) + { + updatedAnalyses = updatedAnalyses.Take(100).ToList(); + } + + double avgCompound = updatedAnalyses.Average(a => a.FinbertResult.CompoundScore); + double avgConf = updatedAnalyses.Average(a => a.FinbertResult.Confidence); + string label = CalculateLabel(avgCompound); + + 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." + }; + + var updatedDoc = isinDoc with + { + Isin = cleanIsin, + CompanyName = !string.IsNullOrWhiteSpace(companyName) ? companyName : isinDoc.CompanyName, + Sector = !string.IsNullOrWhiteSpace(sector) ? sector : isinDoc.Sector, + LastUpdated = nowIso, + CurrentSummary = new IsinCurrentSummary + { + CompoundScore = Math.Round(avgCompound, 2), + SentimentLabel = label, + AvgConfidence = Math.Round(avgConf, 2), + TotalArticlesAnalyzed = updatedAnalyses.Count, + Text = textSummary + }, + 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); + } + finally + { + fileLock.Release(); + } + } + + /// + public async Task UpdateSectorSummaryAsync(string sector, string isin, string articleId, FinBertResultDto finbert) + { + if (string.IsNullOrWhiteSpace(sector)) return; + + string sanitizedSector = string.Concat(sector.Split(Path.GetInvalidFileNameChars())).Trim(); + string filePath = Path.Combine(_basePath, "sectors", $"{sanitizedSector}.json"); + + var fileLock = FileLocks.GetOrAdd(filePath, _ => new SemaphoreSlim(1, 1)); + await fileLock.WaitAsync(); + + try + { + SectorSentimentSummaryDto sectorDoc; + + if (File.Exists(filePath)) + { + try + { + string json = await File.ReadAllTextAsync(filePath); + sectorDoc = JsonSerializer.Deserialize(json, _jsonOptions) ?? new SectorSentimentSummaryDto { Sector = sector }; + } + catch + { + sectorDoc = new SectorSentimentSummaryDto { Sector = sector }; + } + } + else + { + sectorDoc = new SectorSentimentSummaryDto { Sector = sector }; + } + + 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 + { + AnalysisId = analysisId, + Timestamp = nowIso, + RelatedIsin = isin, + 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) + { + updatedAnalyses = updatedAnalyses.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); + + 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." + }; + + var updatedDoc = sectorDoc with + { + Sector = sector, + LastUpdated = nowIso, + CurrentSummary = new SectorCurrentSummary + { + CompoundScore = Math.Round(avgCompound, 2), + SentimentLabel = label, + ActiveIsins = activeIsins, + Text = overviewText + }, + 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); + } + 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/Services/SettingsDbService.cs b/FinlyticSentiment/Services/SettingsDbService.cs new file mode 100644 index 0000000..8d99d9d --- /dev/null +++ b/FinlyticSentiment/Services/SettingsDbService.cs @@ -0,0 +1,115 @@ +using System; +using System.Collections.Generic; +using System.Threading.Tasks; +using FinlyticSentiment.Database; +using FinlyticSentiment.Entities; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging; + +namespace FinlyticSentiment.Services; + +/// +/// Interface for reading and persisting FinlyticSentiment runtime configuration settings in PostgreSQL. +/// +public interface ISettingsDbService +{ + /// + /// Retrieves current sentiment settings from PostgreSQL database, seeding defaults if empty. + /// + Task GetSettingsAsync(); + + /// + /// Persists updated settings entity to PostgreSQL. + /// + Task SaveSettingsAsync(SentimentSettingsEntity settings); + + /// + /// Updates settings from a key-value dictionary received via Admin Panel MQTT events. + /// + Task UpdateSettingsFromDictionaryAsync(Dictionary dictionary); +} + +/// +/// EF Core PostgreSQL implementation of . +/// +public class SettingsDbService : ISettingsDbService +{ + private readonly SentimentDbContext _context; + private readonly ILogger _logger; + + /// + /// Initializes a new instance of the class. + /// + public SettingsDbService(SentimentDbContext context, ILogger logger) + { + _context = context; + _logger = logger; + } + + /// + /// Retrieves current sentiment settings from PostgreSQL database. + /// + public async Task GetSettingsAsync() + { + var settings = await _context.Settings.AsNoTracking().FirstOrDefaultAsync(); + if (settings == null) + { + settings = new SentimentSettingsEntity { Id = Guid.NewGuid() }; + _context.Settings.Add(settings); + await _context.SaveChangesAsync(); + _context.ChangeTracker.Clear(); + } + return settings; + } + + /// + /// Persists updated settings entity to PostgreSQL. + /// + public async Task SaveSettingsAsync(SentimentSettingsEntity settings) + { + var existing = await _context.Settings.FirstOrDefaultAsync(); + if (existing == null) + { + if (settings.Id == Guid.Empty) settings.Id = Guid.NewGuid(); + _context.Settings.Add(settings); + } + else + { + existing.MinConfidenceScore = settings.MinConfidenceScore; + existing.MaxBatchSize = settings.MaxBatchSize; + existing.SweepIntervalMinutes = settings.SweepIntervalMinutes; + existing.GermanWebhookUrl = settings.GermanWebhookUrl; + existing.EnglishWebhookUrl = settings.EnglishWebhookUrl; + existing.UpdatedAt = settings.UpdatedAt; + _context.Settings.Update(existing); + } + await _context.SaveChangesAsync(); + return settings; + } + + /// + /// Updates settings from a key-value dictionary. + /// + public async Task UpdateSettingsFromDictionaryAsync(Dictionary dictionary) + { + var settings = await GetSettingsAsync(); + + foreach (var (key, value) in dictionary) + { + if (string.Equals(key, "MinConfidenceScore", StringComparison.OrdinalIgnoreCase) && double.TryParse(value, out var mcs)) + settings.MinConfidenceScore = mcs; + else if (string.Equals(key, "MaxBatchSize", StringComparison.OrdinalIgnoreCase) && int.TryParse(value, out var mbs)) + settings.MaxBatchSize = mbs; + else if (string.Equals(key, "SweepIntervalMinutes", StringComparison.OrdinalIgnoreCase) && int.TryParse(value, out var sim)) + settings.SweepIntervalMinutes = sim; + else if (string.Equals(key, "GermanWebhookUrl", StringComparison.OrdinalIgnoreCase) && !string.IsNullOrWhiteSpace(value)) + settings.GermanWebhookUrl = value.Trim(); + else if (string.Equals(key, "EnglishWebhookUrl", StringComparison.OrdinalIgnoreCase) && !string.IsNullOrWhiteSpace(value)) + settings.EnglishWebhookUrl = value.Trim(); + } + + settings.UpdatedAt = DateTime.UtcNow; + await SaveSettingsAsync(settings); + _logger.LogInformation("[{Channel}] Successfully updated {Count} sentiment settings in PostgreSQL database.", "SentimentChannel", dictionary.Count); + } +} diff --git a/FinlyticSentiment/Util/SentimentMqttClient.cs b/FinlyticSentiment/Util/SentimentMqttClient.cs new file mode 100644 index 0000000..8c053c8 --- /dev/null +++ b/FinlyticSentiment/Util/SentimentMqttClient.cs @@ -0,0 +1,478 @@ +using System.Text.Json; +using FinlyticCore.Dtos; +using FinlyticCore.Dtos.News; +using FinlyticCore.Dtos.Sentiment; +using FinlyticCore.Models; +using FinlyticCore.Util; +using FinlyticSentiment.Services; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; + +namespace FinlyticSentiment.Util; + +/// +/// Managed MQTT client for requesting pending news articles and updating sentiment results. +/// +public class SentimentMqttClient : ManagedMqttClient, IHostedService +{ + private readonly ILogger _logger; + private readonly IConfiguration _configuration; + private readonly IServiceScopeFactory _scopeFactory; + + /// + /// Initializes a new instance of the class. + /// + public SentimentMqttClient( + ILogger logger, + IConfiguration configuration, + IServiceScopeFactory scopeFactory) : base(logger) + { + _logger = logger; + _configuration = configuration; + _scopeFactory = scopeFactory; + } + + /// + /// Starts the MQTT client and connects to the broker. + /// + public async Task StartAsync(CancellationToken cancellationToken) + { + var config = new MqttConfiguration + { + 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()}" + }; + + _logger.LogInformation("[{Channel}] Starting Sentiment MQTT client. Host: {Host}, ClientId: {ClientId}", + "SentimentChannel", config.Host, config.ClientId); + await ConnectAsync(config); + } + + /// + /// Stops the MQTT client and disconnects from the broker. + /// + public async Task StopAsync(CancellationToken cancellationToken) + { + _logger.LogInformation("[{Channel}] Stopping Sentiment MQTT client and disconnecting.", "SentimentChannel"); + await DisconnectAsync(); + } + + /// + /// Event triggered when a real-time article is broadcasted on services/news/completed. + /// + public event Func? OnArticleReceived; + + /// + protected override async Task OnConnectedAsync() + { + _logger.LogInformation( + "[{Channel}] Sentiment MQTT client connected. Subscribing to RPC response and broadcast topics...", + "SentimentChannel"); + 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/health_Ping/#"); + await SubscribeAsync("services/config/updated/#"); + } + + /// + protected override async Task OnMessageReceivedAsync(string topic, string payload) + { + 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; + } + + if (topic.Equals("services/news/completed", StringComparison.OrdinalIgnoreCase)) + { + await OnNewsCompletedAsync(payload); + 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.Contains("sentiment_GetArticle", StringComparison.OrdinalIgnoreCase)) + { + await OnSentimentGetArticleAsync(payload, correlationId); + } + else if (topic.Contains("sentiment_GetIsin", StringComparison.OrdinalIgnoreCase)) + { + await OnSentimentGetIsinAsync(payload, correlationId); + } + else if (topic.Contains("sentiment_Analyze", StringComparison.OrdinalIgnoreCase)) + { + await OnSentimentAnalyzeAsync(payload, correlationId); + } + else if (topic.Contains("health_Ping", StringComparison.OrdinalIgnoreCase)) + { + await OnHealthPingAsync(topic, correlationId); + } + } + + /// + /// Handles dynamic service config update events. + /// + 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; + 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); + } + } + } + catch (Exception ex) + { + _logger.LogError(ex, "[{Channel}] [SentimentMqttClient] Error processing MQTT config update event.", + "SentimentChannel"); + } + } + + /// + /// 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); + } + } + + /// + /// 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; + 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); + await OnArticleReceived.Invoke(article); + } + } + catch (Exception ex) + { + _logger.LogError(ex, "[{Channel}] Error parsing broadcasted article on services/news/completed.", + "SentimentChannel"); + } + } + + /// + /// 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; + + try + { + 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"); + 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 PublishAsync(responseTopic, sentimentEntry); + } + catch (Exception ex) + { + _logger.LogError(ex, + "[{Channel}] [SentimentMqttClient] Failed to process sentiment_GetArticle RPC request.", + "SentimentChannel"); + } + } + + /// + /// 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; + + try + { + 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"); + 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 PublishAsync(responseTopic, isinSummary); + } + catch (Exception ex) + { + _logger.LogError(ex, "[{Channel}] [SentimentMqttClient] Failed to process sentiment_GetIsin RPC request.", + "SentimentChannel"); + } + } + + /// + /// 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); + string responseTopic = $"services/response/sentiment_Analyze/{correlationId}"; + + if (string.IsNullOrWhiteSpace(payload)) + { + await PublishAsync(responseTopic, (object?)null); + return; + } + + try + { + var request = + JsonSerializer.Deserialize(payload, typeof(AnalyzeSentimentRequest), + FinlyticJsonSerializerContext.Default) as AnalyzeSentimentRequest; + + 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 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); + + IsinAnalysisEntry? existingEntry = null; + + if (!request.ForceReload) + { + existingEntry = await storageService.GetArticleSentimentAsync(cleanArticleId); + } + + if (existingEntry != null) + { + result = existingEntry; + } + else + { + // Artikel per RPC von FinlyticNews abfragen + if (Guid.TryParse(cleanArticleId, out var articleGuid)) + { + var article = await SendRpcRequestAsync( + "news_GetById", + new ArticleRequest(cleanArticleId, cleanArticleId), + TimeSpan.FromSeconds(8)); + + if (article != null) + { + 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 PublishAsync(responseTopic, (object?)null); + return; + } + + // Speichern & Summaries aktualisieren + await storageService.SaveArticleSentimentAsync(article, finbertResult); + + if (article.MatchedAssets != null && article.MatchedAssets.Count > 0) + { + foreach (var asset in article.MatchedAssets) + { + if (string.IsNullOrWhiteSpace(asset.Isin)) continue; + + await storageService.UpdateIsinSummaryAsync( + asset.Isin, + asset.Name, + "General", + article, + finbertResult); + + await storageService.UpdateSectorSummaryAsync( + "General", + asset.Isin, + article.Id.ToString(), + finbertResult); + } + } + + // 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); + } + } + } + } + // 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); + + result = await storageService.GetIsinSummaryAsync(cleanIsin); + } + + _logger.LogInformation( + "[{Channel}] [SentimentMqttClient] Publishing RPC sentiment_Analyze response to {ResponseTopic}", + "SentimentChannel", responseTopic); + await PublishAsync(responseTopic, result); + } + catch (Exception ex) + { + _logger.LogError(ex, "[{Channel}] [SentimentMqttClient] Failed to process sentiment_Analyze RPC request.", + "SentimentChannel"); + 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)); + return articles ?? []; + } + catch (Exception ex) + { + _logger.LogError(ex, "[{Channel}] Error executing MQTT RPC for news_GetPending.", "SentimentChannel"); + } + + 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)); + return response?.Success ?? false; + } + catch (Exception ex) + { + _logger.LogError(ex, "[{Channel}] Error executing MQTT RPC for news_UpdateStatus (ID: {Id}).", + "SentimentChannel", id); + } + + return false; + } +} \ No newline at end of file diff --git a/FinlyticSentiment/appsettings.json b/FinlyticSentiment/appsettings.json new file mode 100644 index 0000000..aa793e8 --- /dev/null +++ b/FinlyticSentiment/appsettings.json @@ -0,0 +1,24 @@ +{ + "Logging": { + "LogLevel": { + "Default": "Information", + "Microsoft.Hosting.Lifetime": "Information", + "Microsoft.EntityFrameworkCore.Database.Command": "Warning" + } + }, + "ConnectionStrings": { + "DefaultConnection": "Host=localhost;Database=finlytic_sentiment;Username=admin;Password=admin" + }, + "MQTT": { + "Host": "localhost", + "Port": 1883, + "ClientId": "FinlyticSentiment" + }, + "Storage": { + "SummariesPath": "data/summaries" + }, + "Webhooks": { + "German": "https://n8n.kleidukos.me/webhook/sentiment/de", + "English": "https://n8n.kleidukos.me/webhook/sentiment/en" + } +}