diff --git a/FinlyticSentiment/Database/SentimentDbContext.cs b/FinlyticSentiment/Database/SentimentDbContext.cs index d380cbb..9a3fcf3 100644 --- a/FinlyticSentiment/Database/SentimentDbContext.cs +++ b/FinlyticSentiment/Database/SentimentDbContext.cs @@ -7,7 +7,7 @@ using Microsoft.EntityFrameworkCore.Design; namespace FinlyticSentiment.Database; /// -/// EF Core DbContext for managing FinlyticSentiment settings in PostgreSQL. +/// EF Core DbContext for managing FinlyticSentiment persistent entities and dynamic settings. /// public class SentimentDbContext : DbContext, ISettingsDbContext { @@ -16,24 +16,50 @@ public class SentimentDbContext : DbContext, ISettingsDbContext } public DbSet DynamicSettings => Set(); - public DbSet Settings => Set(); + public DbSet ArticleSentiments => Set(); + public DbSet CompanySentiments => Set(); + public DbSet SectorSentiments => Set(); protected override void OnModelCreating(ModelBuilder modelBuilder) { base.OnModelCreating(modelBuilder); + // Dynamic Settings modelBuilder.Entity(entity => { entity.HasKey(e => e.Id); entity.HasIndex(e => e.Key).IsUnique(); }); - modelBuilder.Entity(entity => + // Individual Article Sentiments + 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); + + // Prevent duplicate analysis of the same article for the same asset + entity.HasIndex(e => new { e.ArticleId, e.Isin }).IsUnique(); + + // High-performance time-series index for timeline charts & time-decay scans + entity.HasIndex(e => new { e.Isin, e.PublishedAtUtc }); + + entity.HasIndex(e => e.AnalyzedAtUtc); + entity.HasIndex(e => e.Sector); + }); + + // Pre-Aggregated Company Sentiment Summaries + modelBuilder.Entity(entity => + { + entity.HasKey(e => e.Isin); + entity.HasIndex(e => e.Sector); + entity.HasIndex(e => e.WeightedScore); + entity.HasIndex(e => e.LastUpdatedUtc); + }); + + // Pre-Aggregated Sector Sentiment Summaries + modelBuilder.Entity(entity => + { + entity.HasKey(e => e.Sector); + entity.HasIndex(e => e.LastUpdatedUtc); }); } } diff --git a/FinlyticSentiment/Dockerfile b/FinlyticSentiment/Dockerfile index 4a00792..f8be037 100644 --- a/FinlyticSentiment/Dockerfile +++ b/FinlyticSentiment/Dockerfile @@ -1,5 +1,5 @@ FROM mcr.microsoft.com/dotnet/runtime:10.0 AS base -USER app +USER $APP_UID WORKDIR /app FROM mcr.microsoft.com/dotnet/sdk:10.0 AS build diff --git a/FinlyticSentiment/Entities/ArticleSentimentEntity.cs b/FinlyticSentiment/Entities/ArticleSentimentEntity.cs new file mode 100644 index 0000000..7c0c16a --- /dev/null +++ b/FinlyticSentiment/Entities/ArticleSentimentEntity.cs @@ -0,0 +1,53 @@ +using System; +using System.ComponentModel.DataAnnotations; +using System.ComponentModel.DataAnnotations.Schema; + +namespace FinlyticSentiment.Entities; + +/// +/// Entity representing a single sentiment evaluation of an asset mentioned in a news article. +/// +[Table("article_sentiments")] +public class ArticleSentimentEntity +{ + [Key] + public Guid Id { get; set; } = Guid.NewGuid(); + + [Required] + public Guid ArticleId { get; set; } + + [Required] + [MaxLength(12)] + public string Isin { get; set; } = string.Empty; + + [Required] + [MaxLength(256)] + public string Name { get; set; } = string.Empty; + + [MaxLength(128)] + public string? Sector { get; set; } + + [Required] + [MaxLength(32)] + public string Label { get; set; } = "NEUTRAL"; + + public double CompoundScore { get; set; } + + public double Confidence { get; set; } + + [MaxLength(32)] + public string? Impact { get; set; } + + public double PositiveProbability { get; set; } + + public double NegativeProbability { get; set; } + + public double NeutralProbability { get; set; } + + [MaxLength(1024)] + public string? KeyHighlight { get; set; } + + public DateTime PublishedAtUtc { get; set; } + + public DateTime AnalyzedAtUtc { get; set; } = DateTime.UtcNow; +} diff --git a/FinlyticSentiment/Entities/CompanySentimentSummaryEntity.cs b/FinlyticSentiment/Entities/CompanySentimentSummaryEntity.cs new file mode 100644 index 0000000..c4c6bf3 --- /dev/null +++ b/FinlyticSentiment/Entities/CompanySentimentSummaryEntity.cs @@ -0,0 +1,52 @@ +using System; +using System.ComponentModel.DataAnnotations; +using System.ComponentModel.DataAnnotations.Schema; + +namespace FinlyticSentiment.Entities; + +/// +/// Pre-aggregated, real-time sentiment summary for a company/asset, enabling sub-millisecond lookups without recalculation. +/// +[Table("company_sentiment_summaries")] +public class CompanySentimentSummaryEntity +{ + [Key] + [MaxLength(12)] + public string Isin { get; set; } = string.Empty; + + [Required] + [MaxLength(256)] + public string Name { get; set; } = string.Empty; + + [MaxLength(128)] + public string? Sector { get; set; } + + [Required] + [MaxLength(32)] + public string CurrentLabel { get; set; } = "NEUTRAL"; + + public double AverageScore { get; set; } + + public double WeightedScore { get; set; } + + public double AverageConfidence { get; set; } + + public int TotalAnalysesCount { get; set; } + + public int PositiveCount { get; set; } + + public int NegativeCount { get; set; } + + public int NeutralCount { get; set; } + + [MaxLength(1024)] + public string? LatestKeyHighlight { get; set; } + + [MaxLength(32)] + public string? Trend { get; set; } = "STABLE"; + + public DateTime LastUpdatedUtc { get; set; } = DateTime.UtcNow; + + [ConcurrencyCheck] + public uint Version { get; set; } +} diff --git a/FinlyticSentiment/Entities/SectorSentimentSummaryEntity.cs b/FinlyticSentiment/Entities/SectorSentimentSummaryEntity.cs new file mode 100644 index 0000000..0453b9b --- /dev/null +++ b/FinlyticSentiment/Entities/SectorSentimentSummaryEntity.cs @@ -0,0 +1,28 @@ +using System; +using System.ComponentModel.DataAnnotations; +using System.ComponentModel.DataAnnotations.Schema; + +namespace FinlyticSentiment.Entities; + +/// +/// Pre-aggregated real-time sentiment summary for a market sector (e.g. Technology, Healthcare). +/// +[Table("sector_sentiment_summaries")] +public class SectorSentimentSummaryEntity +{ + [Key] + [MaxLength(128)] + public string Sector { get; set; } = string.Empty; + + [Required] + [MaxLength(32)] + public string CurrentLabel { get; set; } = "NEUTRAL"; + + public double AverageScore { get; set; } + + public int TotalArticlesCount { get; set; } + + public int TotalCompaniesCount { get; set; } + + public DateTime LastUpdatedUtc { get; set; } = DateTime.UtcNow; +} diff --git a/FinlyticSentiment/Entities/SentimentSettingsEntity.cs b/FinlyticSentiment/Entities/SentimentSettingsEntity.cs deleted file mode 100644 index 6377eb0..0000000 --- a/FinlyticSentiment/Entities/SentimentSettingsEntity.cs +++ /dev/null @@ -1,23 +0,0 @@ -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/Migrations/20260801090401_InitialSentimentSettings.Designer.cs b/FinlyticSentiment/Migrations/20260801090401_InitialSentimentSettings.Designer.cs deleted file mode 100644 index febe352..0000000 --- a/FinlyticSentiment/Migrations/20260801090401_InitialSentimentSettings.Designer.cs +++ /dev/null @@ -1,63 +0,0 @@ -// -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 deleted file mode 100644 index 3663991..0000000 --- a/FinlyticSentiment/Migrations/20260801090401_InitialSentimentSettings.cs +++ /dev/null @@ -1,39 +0,0 @@ -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/20260813202634_CheckPendingSentiment.Designer.cs b/FinlyticSentiment/Migrations/20260813202634_CheckPendingSentiment.Designer.cs deleted file mode 100644 index a43b810..0000000 --- a/FinlyticSentiment/Migrations/20260813202634_CheckPendingSentiment.Designer.cs +++ /dev/null @@ -1,63 +0,0 @@ -// -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("20260813202634_CheckPendingSentiment")] - partial class CheckPendingSentiment - { - /// - 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/20260813202634_CheckPendingSentiment.cs b/FinlyticSentiment/Migrations/20260813202634_CheckPendingSentiment.cs deleted file mode 100644 index 3271db9..0000000 --- a/FinlyticSentiment/Migrations/20260813202634_CheckPendingSentiment.cs +++ /dev/null @@ -1,22 +0,0 @@ -using Microsoft.EntityFrameworkCore.Migrations; - -#nullable disable - -namespace FinlyticSentiment.Migrations -{ - /// - public partial class CheckPendingSentiment : Migration - { - /// - protected override void Up(MigrationBuilder migrationBuilder) - { - - } - - /// - protected override void Down(MigrationBuilder migrationBuilder) - { - - } - } -} diff --git a/FinlyticSentiment/Migrations/20260815184006_AddDynamicSettings.Designer.cs b/FinlyticSentiment/Migrations/20260815184006_AddDynamicSettings.Designer.cs deleted file mode 100644 index fb14de4..0000000 --- a/FinlyticSentiment/Migrations/20260815184006_AddDynamicSettings.Designer.cs +++ /dev/null @@ -1,94 +0,0 @@ -// -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 deleted file mode 100644 index fbd967a..0000000 --- a/FinlyticSentiment/Migrations/20260815184006_AddDynamicSettings.cs +++ /dev/null @@ -1,43 +0,0 @@ -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/20260818202351_Init.Designer.cs b/FinlyticSentiment/Migrations/20260818202351_Init.Designer.cs new file mode 100644 index 0000000..fb18a76 --- /dev/null +++ b/FinlyticSentiment/Migrations/20260818202351_Init.Designer.cs @@ -0,0 +1,229 @@ +// +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("20260818202351_Init")] + partial class Init + { + /// + 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.ArticleSentimentEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("AnalyzedAtUtc") + .HasColumnType("timestamp with time zone"); + + b.Property("ArticleId") + .HasColumnType("uuid"); + + b.Property("CompoundScore") + .HasColumnType("double precision"); + + b.Property("Confidence") + .HasColumnType("double precision"); + + b.Property("Impact") + .HasMaxLength(32) + .HasColumnType("character varying(32)"); + + b.Property("Isin") + .IsRequired() + .HasMaxLength(12) + .HasColumnType("character varying(12)"); + + b.Property("KeyHighlight") + .HasMaxLength(1024) + .HasColumnType("character varying(1024)"); + + b.Property("Label") + .IsRequired() + .HasMaxLength(32) + .HasColumnType("character varying(32)"); + + b.Property("Name") + .IsRequired() + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.Property("NegativeProbability") + .HasColumnType("double precision"); + + b.Property("NeutralProbability") + .HasColumnType("double precision"); + + b.Property("PositiveProbability") + .HasColumnType("double precision"); + + b.Property("PublishedAtUtc") + .HasColumnType("timestamp with time zone"); + + b.Property("Sector") + .HasMaxLength(128) + .HasColumnType("character varying(128)"); + + b.HasKey("Id"); + + b.HasIndex("AnalyzedAtUtc"); + + b.HasIndex("Sector"); + + b.HasIndex("ArticleId", "Isin") + .IsUnique(); + + b.HasIndex("Isin", "PublishedAtUtc"); + + b.ToTable("article_sentiments"); + }); + + modelBuilder.Entity("FinlyticSentiment.Entities.CompanySentimentSummaryEntity", b => + { + b.Property("Isin") + .HasMaxLength(12) + .HasColumnType("character varying(12)"); + + b.Property("AverageConfidence") + .HasColumnType("double precision"); + + b.Property("AverageScore") + .HasColumnType("double precision"); + + b.Property("CurrentLabel") + .IsRequired() + .HasMaxLength(32) + .HasColumnType("character varying(32)"); + + b.Property("LastUpdatedUtc") + .HasColumnType("timestamp with time zone"); + + b.Property("LatestKeyHighlight") + .HasMaxLength(1024) + .HasColumnType("character varying(1024)"); + + b.Property("Name") + .IsRequired() + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.Property("NegativeCount") + .HasColumnType("integer"); + + b.Property("NeutralCount") + .HasColumnType("integer"); + + b.Property("PositiveCount") + .HasColumnType("integer"); + + b.Property("Sector") + .HasMaxLength(128) + .HasColumnType("character varying(128)"); + + b.Property("TotalAnalysesCount") + .HasColumnType("integer"); + + b.Property("Trend") + .HasMaxLength(32) + .HasColumnType("character varying(32)"); + + b.Property("Version") + .IsConcurrencyToken() + .HasColumnType("bigint"); + + b.Property("WeightedScore") + .HasColumnType("double precision"); + + b.HasKey("Isin"); + + b.HasIndex("LastUpdatedUtc"); + + b.HasIndex("Sector"); + + b.HasIndex("WeightedScore"); + + b.ToTable("company_sentiment_summaries"); + }); + + modelBuilder.Entity("FinlyticSentiment.Entities.SectorSentimentSummaryEntity", b => + { + b.Property("Sector") + .HasMaxLength(128) + .HasColumnType("character varying(128)"); + + b.Property("AverageScore") + .HasColumnType("double precision"); + + b.Property("CurrentLabel") + .IsRequired() + .HasMaxLength(32) + .HasColumnType("character varying(32)"); + + b.Property("LastUpdatedUtc") + .HasColumnType("timestamp with time zone"); + + b.Property("TotalArticlesCount") + .HasColumnType("integer"); + + b.Property("TotalCompaniesCount") + .HasColumnType("integer"); + + b.HasKey("Sector"); + + b.HasIndex("LastUpdatedUtc"); + + b.ToTable("sector_sentiment_summaries"); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/FinlyticSentiment/Migrations/20260818202351_Init.cs b/FinlyticSentiment/Migrations/20260818202351_Init.cs new file mode 100644 index 0000000..7c13628 --- /dev/null +++ b/FinlyticSentiment/Migrations/20260818202351_Init.cs @@ -0,0 +1,159 @@ +using System; +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace FinlyticSentiment.Migrations +{ + /// + public partial class Init : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.CreateTable( + name: "article_sentiments", + columns: table => new + { + Id = table.Column(type: "uuid", nullable: false), + ArticleId = table.Column(type: "uuid", nullable: false), + Isin = table.Column(type: "character varying(12)", maxLength: 12, nullable: false), + Name = table.Column(type: "character varying(256)", maxLength: 256, nullable: false), + Sector = table.Column(type: "character varying(128)", maxLength: 128, nullable: true), + Label = table.Column(type: "character varying(32)", maxLength: 32, nullable: false), + CompoundScore = table.Column(type: "double precision", nullable: false), + Confidence = table.Column(type: "double precision", nullable: false), + Impact = table.Column(type: "character varying(32)", maxLength: 32, nullable: true), + PositiveProbability = table.Column(type: "double precision", nullable: false), + NegativeProbability = table.Column(type: "double precision", nullable: false), + NeutralProbability = table.Column(type: "double precision", nullable: false), + KeyHighlight = table.Column(type: "character varying(1024)", maxLength: 1024, nullable: true), + PublishedAtUtc = table.Column(type: "timestamp with time zone", nullable: false), + AnalyzedAtUtc = table.Column(type: "timestamp with time zone", nullable: false) + }, + constraints: table => + { + table.PrimaryKey("PK_article_sentiments", x => x.Id); + }); + + migrationBuilder.CreateTable( + name: "company_sentiment_summaries", + columns: table => new + { + Isin = table.Column(type: "character varying(12)", maxLength: 12, nullable: false), + Name = table.Column(type: "character varying(256)", maxLength: 256, nullable: false), + Sector = table.Column(type: "character varying(128)", maxLength: 128, nullable: true), + CurrentLabel = table.Column(type: "character varying(32)", maxLength: 32, nullable: false), + AverageScore = table.Column(type: "double precision", nullable: false), + WeightedScore = table.Column(type: "double precision", nullable: false), + AverageConfidence = table.Column(type: "double precision", nullable: false), + TotalAnalysesCount = table.Column(type: "integer", nullable: false), + PositiveCount = table.Column(type: "integer", nullable: false), + NegativeCount = table.Column(type: "integer", nullable: false), + NeutralCount = table.Column(type: "integer", nullable: false), + LatestKeyHighlight = table.Column(type: "character varying(1024)", maxLength: 1024, nullable: true), + Trend = table.Column(type: "character varying(32)", maxLength: 32, nullable: true), + LastUpdatedUtc = table.Column(type: "timestamp with time zone", nullable: false), + Version = table.Column(type: "bigint", nullable: false) + }, + constraints: table => + { + table.PrimaryKey("PK_company_sentiment_summaries", x => x.Isin); + }); + + 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.CreateTable( + name: "sector_sentiment_summaries", + columns: table => new + { + Sector = table.Column(type: "character varying(128)", maxLength: 128, nullable: false), + CurrentLabel = table.Column(type: "character varying(32)", maxLength: 32, nullable: false), + AverageScore = table.Column(type: "double precision", nullable: false), + TotalArticlesCount = table.Column(type: "integer", nullable: false), + TotalCompaniesCount = table.Column(type: "integer", nullable: false), + LastUpdatedUtc = table.Column(type: "timestamp with time zone", nullable: false) + }, + constraints: table => + { + table.PrimaryKey("PK_sector_sentiment_summaries", x => x.Sector); + }); + + migrationBuilder.CreateIndex( + name: "IX_article_sentiments_AnalyzedAtUtc", + table: "article_sentiments", + column: "AnalyzedAtUtc"); + + migrationBuilder.CreateIndex( + name: "IX_article_sentiments_ArticleId_Isin", + table: "article_sentiments", + columns: new[] { "ArticleId", "Isin" }, + unique: true); + + migrationBuilder.CreateIndex( + name: "IX_article_sentiments_Isin_PublishedAtUtc", + table: "article_sentiments", + columns: new[] { "Isin", "PublishedAtUtc" }); + + migrationBuilder.CreateIndex( + name: "IX_article_sentiments_Sector", + table: "article_sentiments", + column: "Sector"); + + migrationBuilder.CreateIndex( + name: "IX_company_sentiment_summaries_LastUpdatedUtc", + table: "company_sentiment_summaries", + column: "LastUpdatedUtc"); + + migrationBuilder.CreateIndex( + name: "IX_company_sentiment_summaries_Sector", + table: "company_sentiment_summaries", + column: "Sector"); + + migrationBuilder.CreateIndex( + name: "IX_company_sentiment_summaries_WeightedScore", + table: "company_sentiment_summaries", + column: "WeightedScore"); + + migrationBuilder.CreateIndex( + name: "IX_DynamicSettings_Key", + table: "DynamicSettings", + column: "Key", + unique: true); + + migrationBuilder.CreateIndex( + name: "IX_sector_sentiment_summaries_LastUpdatedUtc", + table: "sector_sentiment_summaries", + column: "LastUpdatedUtc"); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropTable( + name: "article_sentiments"); + + migrationBuilder.DropTable( + name: "company_sentiment_summaries"); + + migrationBuilder.DropTable( + name: "DynamicSettings"); + + migrationBuilder.DropTable( + name: "sector_sentiment_summaries"); + } + } +} diff --git a/FinlyticSentiment/Migrations/SentimentDbContextModelSnapshot.cs b/FinlyticSentiment/Migrations/SentimentDbContextModelSnapshot.cs index 9f13197..3c8db27 100644 --- a/FinlyticSentiment/Migrations/SentimentDbContextModelSnapshot.cs +++ b/FinlyticSentiment/Migrations/SentimentDbContextModelSnapshot.cs @@ -53,37 +53,172 @@ namespace FinlyticSentiment.Migrations b.ToTable("DynamicSettings"); }); - modelBuilder.Entity("FinlyticSentiment.Entities.SentimentSettingsEntity", b => + modelBuilder.Entity("FinlyticSentiment.Entities.ArticleSentimentEntity", b => { b.Property("Id") .ValueGeneratedOnAdd() .HasColumnType("uuid"); - b.Property("EnglishWebhookUrl") - .IsRequired() - .HasMaxLength(500) - .HasColumnType("character varying(500)"); + b.Property("AnalyzedAtUtc") + .HasColumnType("timestamp with time zone"); - b.Property("GermanWebhookUrl") - .IsRequired() - .HasMaxLength(500) - .HasColumnType("character varying(500)"); + b.Property("ArticleId") + .HasColumnType("uuid"); - b.Property("MaxBatchSize") - .HasColumnType("integer"); - - b.Property("MinConfidenceScore") + b.Property("CompoundScore") .HasColumnType("double precision"); - b.Property("SweepIntervalMinutes") - .HasColumnType("integer"); + b.Property("Confidence") + .HasColumnType("double precision"); - b.Property("UpdatedAt") + b.Property("Impact") + .HasMaxLength(32) + .HasColumnType("character varying(32)"); + + b.Property("Isin") + .IsRequired() + .HasMaxLength(12) + .HasColumnType("character varying(12)"); + + b.Property("KeyHighlight") + .HasMaxLength(1024) + .HasColumnType("character varying(1024)"); + + b.Property("Label") + .IsRequired() + .HasMaxLength(32) + .HasColumnType("character varying(32)"); + + b.Property("Name") + .IsRequired() + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.Property("NegativeProbability") + .HasColumnType("double precision"); + + b.Property("NeutralProbability") + .HasColumnType("double precision"); + + b.Property("PositiveProbability") + .HasColumnType("double precision"); + + b.Property("PublishedAtUtc") .HasColumnType("timestamp with time zone"); + b.Property("Sector") + .HasMaxLength(128) + .HasColumnType("character varying(128)"); + b.HasKey("Id"); - b.ToTable("sentiment_settings", (string)null); + b.HasIndex("AnalyzedAtUtc"); + + b.HasIndex("Sector"); + + b.HasIndex("ArticleId", "Isin") + .IsUnique(); + + b.HasIndex("Isin", "PublishedAtUtc"); + + b.ToTable("article_sentiments"); + }); + + modelBuilder.Entity("FinlyticSentiment.Entities.CompanySentimentSummaryEntity", b => + { + b.Property("Isin") + .HasMaxLength(12) + .HasColumnType("character varying(12)"); + + b.Property("AverageConfidence") + .HasColumnType("double precision"); + + b.Property("AverageScore") + .HasColumnType("double precision"); + + b.Property("CurrentLabel") + .IsRequired() + .HasMaxLength(32) + .HasColumnType("character varying(32)"); + + b.Property("LastUpdatedUtc") + .HasColumnType("timestamp with time zone"); + + b.Property("LatestKeyHighlight") + .HasMaxLength(1024) + .HasColumnType("character varying(1024)"); + + b.Property("Name") + .IsRequired() + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.Property("NegativeCount") + .HasColumnType("integer"); + + b.Property("NeutralCount") + .HasColumnType("integer"); + + b.Property("PositiveCount") + .HasColumnType("integer"); + + b.Property("Sector") + .HasMaxLength(128) + .HasColumnType("character varying(128)"); + + b.Property("TotalAnalysesCount") + .HasColumnType("integer"); + + b.Property("Trend") + .HasMaxLength(32) + .HasColumnType("character varying(32)"); + + b.Property("Version") + .IsConcurrencyToken() + .HasColumnType("bigint"); + + b.Property("WeightedScore") + .HasColumnType("double precision"); + + b.HasKey("Isin"); + + b.HasIndex("LastUpdatedUtc"); + + b.HasIndex("Sector"); + + b.HasIndex("WeightedScore"); + + b.ToTable("company_sentiment_summaries"); + }); + + modelBuilder.Entity("FinlyticSentiment.Entities.SectorSentimentSummaryEntity", b => + { + b.Property("Sector") + .HasMaxLength(128) + .HasColumnType("character varying(128)"); + + b.Property("AverageScore") + .HasColumnType("double precision"); + + b.Property("CurrentLabel") + .IsRequired() + .HasMaxLength(32) + .HasColumnType("character varying(32)"); + + b.Property("LastUpdatedUtc") + .HasColumnType("timestamp with time zone"); + + b.Property("TotalArticlesCount") + .HasColumnType("integer"); + + b.Property("TotalCompaniesCount") + .HasColumnType("integer"); + + b.HasKey("Sector"); + + b.HasIndex("LastUpdatedUtc"); + + b.ToTable("sector_sentiment_summaries"); }); #pragma warning restore 612, 618 } diff --git a/FinlyticSentiment/Program.cs b/FinlyticSentiment/Program.cs index f4ee768..43608a8 100644 --- a/FinlyticSentiment/Program.cs +++ b/FinlyticSentiment/Program.cs @@ -11,7 +11,7 @@ using Microsoft.Extensions.Hosting; var builder = Host.CreateApplicationBuilder(args); -// Register PostgreSQL DbContext for settings persistence +// Register PostgreSQL DbContext for sentiment data & settings builder.Services.AddDbContext(options => options.UseNpgsql(builder.Configuration.GetConnectionString("DefaultConnection"))); builder.Services.AddScoped(sp => sp.GetRequiredService()); @@ -23,10 +23,9 @@ builder.Services.AddSingleton(typeof(IFinlyticLogger<>), typeof(FinlyticLogger<> // Register HttpClient builder.Services.AddHttpClient(); -// Register Service interfaces and co-located implementations -builder.Services.AddScoped(); +// Register Sentiment Services +builder.Services.AddScoped(); builder.Services.AddSingleton(); -builder.Services.AddSingleton(); // Register MQTT Client (as singleton hosted service) builder.Services.AddSingleton(); @@ -43,12 +42,14 @@ using (var scope = host.Services.CreateScope()) try { var db = scope.ServiceProvider.GetRequiredService(); - await db.Database.MigrateAsync(); + var connStr = builder.Configuration.GetConnectionString("DefaultConnection") ?? ""; + await db.MigrateWithBootstrapAsync(connStr); } catch (Exception ex) { Console.WriteLine($"Critical error during database migration for FinlyticSentiment: {ex.Message}"); } + } await host.RunAsync(); diff --git a/FinlyticSentiment/Project.md b/FinlyticSentiment/Project.md deleted file mode 100644 index 0458002..0000000 --- a/FinlyticSentiment/Project.md +++ /dev/null @@ -1,34 +0,0 @@ -# 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 index 5942829..ee9e227 100644 --- a/FinlyticSentiment/Services/FinBertAnalyzerService.cs +++ b/FinlyticSentiment/Services/FinBertAnalyzerService.cs @@ -52,26 +52,13 @@ public class FinBertAnalyzerService : IFinBertAnalyzerService double minConfidence = 0.60; using (var scope = _scopeFactory.CreateScope()) { - var settingsService = scope.ServiceProvider.GetService(); + var settingsService = scope.ServiceProvider.GetRequiredService(); bool isEnglish = string.Equals(article.Language, "en", StringComparison.OrdinalIgnoreCase); - if (settingsService != null) - { - targetUrl = isEnglish - ? await settingsService.GetSettingAsync(SettingKeys.EnglishWebhookUrl) - : await settingsService.GetSettingAsync(SettingKeys.GermanWebhookUrl); - minConfidence = await settingsService.GetSettingAsync(SettingKeys.MinimumConfidenceThreshold); - } - - if (string.IsNullOrWhiteSpace(targetUrl)) - { - var settingsDb = scope.ServiceProvider.GetRequiredService(); - var settings = await settingsDb.GetSettingsAsync(); - targetUrl = isEnglish - ? settings.EnglishWebhookUrl - : settings.GermanWebhookUrl; - minConfidence = settings.MinConfidenceScore; - } + targetUrl = isEnglish + ? await settingsService.GetSettingAsync(SettingKeys.EnglishWebhookUrl) + : await settingsService.GetSettingAsync(SettingKeys.GermanWebhookUrl); + minConfidence = await settingsService.GetSettingAsync(SettingKeys.MinimumConfidenceThreshold); } await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[FinBertAnalyzerService] Analyzing article (ID: {Id}, Lang: {Lang}) via webhook: {Url} (MinConf: {Conf})", article.Id, article.Language ?? "de", targetUrl, minConfidence); @@ -128,11 +115,11 @@ public class FinBertAnalyzerService : IFinBertAnalyzerService double confidence = GetDoubleProp(root, "confidence") ?? GetDoubleProp(root, "confidence_score") ?? 0.5; - string snippet = GetStringProp(root, "summary_snippet") + string? impact = GetStringProp(root, "impact") ?? "MEDIUM"; + string? keyHighlight = GetStringProp(root, "key_highlight") + ?? GetStringProp(root, "summary_snippet") ?? GetStringProp(root, "summary") - ?? GetStringProp(root, "text") - ?? article.Summary - ?? article.Title; + ?? article.Summary; double pos = 0.0, neg = 0.0, neu = 1.0; if (root.TryGetProperty("probabilities", out var probsElem) && probsElem.ValueKind == JsonValueKind.Object) @@ -165,44 +152,57 @@ public class FinBertAnalyzerService : IFinBertAnalyzerService } } - return new FinBertResultDto + var result = new FinBertResultDto { Label = label, CompoundScore = Math.Round(compoundScore, 4), Confidence = Math.Round(confidence, 4), + Impact = impact.ToUpperInvariant(), + KeyHighlight = keyHighlight, Probabilities = new FinBertProbabilities { Positive = Math.Round(pos, 4), Negative = Math.Round(neg, 4), Neutral = Math.Round(neu, 4) - }, - SummarySnippet = snippet + } }; + + await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[FinBertAnalyzerService] Successfully analyzed article {Id}: {Label} (Compound: {Score}, Conf: {Conf}, Impact: {Impact})", article.Id, result.Label, result.CompoundScore, result.Confidence, result.Impact); + + return result; } } } - - await _finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, "[FinBertAnalyzerService] n8n Webhook returned non-success status: {StatusCode}.", response.StatusCode); + else + { + await _finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, "[FinBertAnalyzerService] Webhook call failed with status: {Status} (Article ID: {Id})", response.StatusCode, article.Id); + } } catch (Exception ex) { - await _finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[FinBertAnalyzerService] Failed to call n8n sentiment webhook."); + await _finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[FinBertAnalyzerService] Exception during webhook request for article: {Id}", article.Id); } return null; } - private static string? GetStringProp(JsonElement elem, string propName) + private static string? GetStringProp(JsonElement element, string propName) { - return elem.TryGetProperty(propName, out var prop) && prop.ValueKind == JsonValueKind.String ? prop.GetString() : null; + if (element.TryGetProperty(propName, out var prop) && prop.ValueKind == JsonValueKind.String) + { + return prop.GetString(); + } + return null; } - private static double? GetDoubleProp(JsonElement elem, string propName) + private static double? GetDoubleProp(JsonElement element, string propName) { - if (elem.TryGetProperty(propName, out var prop)) + if (element.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; + if (prop.ValueKind == JsonValueKind.Number && prop.TryGetDouble(out var d)) + return d; + if (prop.ValueKind == JsonValueKind.String && double.TryParse(prop.GetString(), System.Globalization.NumberStyles.Any, System.Globalization.CultureInfo.InvariantCulture, out var parsed)) + return parsed; } return null; } diff --git a/FinlyticSentiment/Services/SentimentBackgroundService.cs b/FinlyticSentiment/Services/SentimentBackgroundService.cs index b6f45aa..db65c7f 100644 --- a/FinlyticSentiment/Services/SentimentBackgroundService.cs +++ b/FinlyticSentiment/Services/SentimentBackgroundService.cs @@ -13,13 +13,12 @@ namespace FinlyticSentiment.Services; /// /// Background hosted worker executing periodic sentiment analysis sweeps on pending news articles -/// and processing real-time article broadcasts. +/// and processing real-time article broadcasts from FinlyticNews. /// public class SentimentBackgroundService : BackgroundService { private readonly SentimentMqttClient _mqttClient; private readonly IFinBertAnalyzerService _analyzer; - private readonly ISentimentStorageService _storage; private readonly IServiceScopeFactory _scopeFactory; private readonly IFinlyticLogger _finlyticLogger; @@ -28,13 +27,11 @@ public class SentimentBackgroundService : BackgroundService public SentimentBackgroundService( SentimentMqttClient mqttClient, IFinBertAnalyzerService analyzer, - ISentimentStorageService storage, IServiceScopeFactory scopeFactory, IFinlyticLogger finlyticLogger) { _mqttClient = mqttClient; _analyzer = analyzer; - _storage = storage; _scopeFactory = scopeFactory; _finlyticLogger = finlyticLogger; } @@ -56,11 +53,13 @@ public class SentimentBackgroundService : BackgroundService int maxBatchSize = 10; int sweepIntervalMinutes = 5; - using (var scope = _scopeFactory.CreateScope()) + try { + using var scope = _scopeFactory.CreateScope(); var settings = scope.ServiceProvider.GetRequiredService(); maxBatchSize = await settings.GetSettingAsync(SettingKeys.MaxBatchSize, stoppingToken); } + catch { } var interval = TimeSpan.FromMinutes(Math.Max(1, sweepIntervalMinutes)); @@ -79,10 +78,9 @@ public class SentimentBackgroundService : BackgroundService await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] Waiting {Minutes} minute(s) until next sentiment sweep...", interval.TotalMinutes); - using var timer = new PeriodicTimer(interval); try { - await timer.WaitForNextTickAsync(stoppingToken); + await Task.Delay(interval, stoppingToken); } catch (OperationCanceledException) { @@ -90,7 +88,7 @@ public class SentimentBackgroundService : BackgroundService } } - await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] FinlyticSentiment Background Service is shutting down gracefully."); + await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] FinlyticSentiment Background Service shutting down."); } private async Task PerformSentimentSweepAsync(int maxBatchSize, CancellationToken cancellationToken) @@ -137,45 +135,40 @@ public class SentimentBackgroundService : BackgroundService if (cancellationToken.IsCancellationRequested) return; - await _storage.SaveArticleSentimentAsync(article, finbert); + using var scope = _scopeFactory.CreateScope(); + var dbService = scope.ServiceProvider.GetRequiredService(); 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 dbService.SaveArticleSentimentAsync( + articleId: article.Id, + isin: asset.Isin, + companyName: asset.Name, + sector: null, + publishedAt: article.PublishedAt, + finbert: finbert, + ct: cancellationToken + ); - await _storage.UpdateSectorSummaryAsync( - "General", - asset.Isin, - article.Id.ToString(), - finbert); + // Broadcast real-time updated summary for this asset + var summaryDto = await dbService.GetIsinSummaryDtoAsync(asset.Isin, cancellationToken); + if (summaryDto != null) + { + await _mqttClient.BroadcastSentimentResultAsync(asset.Isin, summaryDto); + } } } - if (cancellationToken.IsCancellationRequested) return; - - bool updated = await _mqttClient.UpdateArticleStatusAsync(article.Id, "Analyzed"); - if (updated) - { - 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 - { - await _finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] Failed to confirm status update to 'Analyzed' in FinlyticNews for article: {Id}", article.Id); - } + await _mqttClient.UpdateArticleStatusAsync(article.Id, "Analyzed"); + await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] Completed sentiment persistence and status transition for article {Id}.", article.Id); } catch (Exception ex) { - await _finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[SentimentBackgroundService] Error processing sentiment for article: {Id} ({Title})", article.Id, article.Title); + await _finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[SentimentBackgroundService] Failed to process sentiment for article: {Id}", article.Id); } finally { diff --git a/FinlyticSentiment/Services/SentimentDbService.cs b/FinlyticSentiment/Services/SentimentDbService.cs new file mode 100644 index 0000000..3553711 --- /dev/null +++ b/FinlyticSentiment/Services/SentimentDbService.cs @@ -0,0 +1,556 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using FinlyticCore.Dtos.News; +using FinlyticCore.Dtos.Sentiment; +using FinlyticCore.Services; +using FinlyticSentiment.Database; +using FinlyticSentiment.Entities; +using FinlyticSentiment.Util; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; + +namespace FinlyticSentiment.Services; + +/// +/// Service contract for persisting and retrieving asset, article, and sector sentiments. +/// +public interface ISentimentDbService +{ + Task SaveArticleSentimentAsync( + Guid articleId, + string isin, + string companyName, + string? sector, + DateTime publishedAt, + FinBertResultDto finbert, + CancellationToken ct = default); + + Task GetCompanySentimentAsync(string isin, CancellationToken ct = default); + + Task GetIsinSummaryDtoAsync(string isin, CancellationToken ct = default); + + Task> GetAllCompanySentimentsAsync(int limit, int offset, string? sector = null, CancellationToken ct = default); + + Task GetSectorSentimentAsync(string sector, CancellationToken ct = default); + + Task> GetArticleSentimentsAsync(Guid articleId, CancellationToken ct = default); + + Task GetArticleSentimentEntryAsync(Guid articleId, CancellationToken ct = default); + + Task> GetSentimentTimelineAsync(string isin, int days, CancellationToken ct = default); +} + +/// +/// Database persistence service implementing exponential half-life time-decay scoring, +/// optimistic concurrency protection, and sub-millisecond pre-aggregated lookups. +/// +public class SentimentDbService : ISentimentDbService +{ + private readonly SentimentDbContext _context; + private readonly IServiceScopeFactory _scopeFactory; + private readonly IFinlyticLogger _finlyticLogger; + + public SentimentDbService( + SentimentDbContext context, + IServiceScopeFactory scopeFactory, + IFinlyticLogger finlyticLogger) + { + _context = context; + _scopeFactory = scopeFactory; + _finlyticLogger = finlyticLogger; + } + + /// + public async Task SaveArticleSentimentAsync( + Guid articleId, + string isin, + string companyName, + string? sector, + DateTime publishedAt, + FinBertResultDto finbert, + CancellationToken ct = default) + { + if (articleId == Guid.Empty || string.IsNullOrWhiteSpace(isin) || finbert == null) + return null; + + var cleanIsin = isin.Trim().ToUpperInvariant(); + var cleanName = !string.IsNullOrWhiteSpace(companyName) ? companyName.Trim() : cleanIsin; + var cleanSector = !string.IsNullOrWhiteSpace(sector) ? sector.Trim() : null; + + var publishedUtc = publishedAt.Kind == DateTimeKind.Unspecified + ? DateTime.SpecifyKind(publishedAt, DateTimeKind.Utc) + : publishedAt.ToUniversalTime(); + + var articleSentiment = new ArticleSentimentEntity + { + Id = Guid.NewGuid(), + ArticleId = articleId, + Isin = cleanIsin, + Name = cleanName, + Sector = cleanSector, + Label = finbert.Label, + CompoundScore = finbert.CompoundScore, + Confidence = finbert.Confidence, + Impact = finbert.Impact, + PositiveProbability = finbert.Probabilities.Positive, + NegativeProbability = finbert.Probabilities.Negative, + NeutralProbability = finbert.Probabilities.Neutral, + KeyHighlight = finbert.KeyHighlight, + PublishedAtUtc = publishedUtc, + AnalyzedAtUtc = DateTime.UtcNow + }; + + // 1. Insert Article Sentiment with Idempotency Protection + try + { + var existing = await _context.ArticleSentiments + .AsNoTracking() + .FirstOrDefaultAsync(a => a.ArticleId == articleId && a.Isin == cleanIsin, ct); + + if (existing != null) + { + await _finlyticLogger.LogDebugAsync(SettingKeys.SentimentChannel, "[SentimentDbService] Article {ArticleId} already has sentiment for ISIN {Isin}.", articleId, cleanIsin); + return existing; + } + + _context.ArticleSentiments.Add(articleSentiment); + await _context.SaveChangesAsync(ct); + } + catch (DbUpdateException ex) + { + _context.ChangeTracker.Clear(); + await _finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, "[SentimentDbService] Unique constraint or concurrency hit on insert for ISIN {Isin}. Message: {Msg}", cleanIsin, ex.Message); + } + + // 2. Concurrency-Safe Recalculation of Company Summary with Exponential Time-Decay + await UpdateCompanySummaryWithRetryAsync(cleanIsin, cleanName, cleanSector, ct); + + // 3. Update Sector Summary if sector is present + if (!string.IsNullOrWhiteSpace(cleanSector)) + { + await UpdateSectorSummaryAsync(cleanSector, ct); + } + + return articleSentiment; + } + + private async Task UpdateCompanySummaryWithRetryAsync( + string isin, + string companyName, + string? sector, + CancellationToken ct) + { + const int maxRetries = 5; + for (int attempt = 1; attempt <= maxRetries; attempt++) + { + try + { + double halfLifeDays = 7.0; + int windowDays = 30; + + try + { + using var scope = _scopeFactory.CreateScope(); + var settings = scope.ServiceProvider.GetRequiredService(); + halfLifeDays = await settings.GetSettingAsync(SettingKeys.TimeDecayHalfLifeDays, ct); + windowDays = await settings.GetSettingAsync(SettingKeys.SentimentWindowDays, ct); + } + catch { } + + var cutoffDate = DateTime.UtcNow.AddDays(-windowDays); + + // Query all recent sentiment analyses for this asset + var recentAnalyses = await _context.ArticleSentiments + .AsNoTracking() + .Where(a => a.Isin == isin && a.PublishedAtUtc >= cutoffDate) + .OrderByDescending(a => a.PublishedAtUtc) + .ToListAsync(ct); + + if (recentAnalyses.Count == 0) + { + // Fallback to latest available analysis if none within window + var latest = await _context.ArticleSentiments + .AsNoTracking() + .Where(a => a.Isin == isin) + .OrderByDescending(a => a.PublishedAtUtc) + .Take(1) + .ToListAsync(ct); + recentAnalyses = latest; + } + + if (recentAnalyses.Count == 0) return; + + // Mathematical Half-Life Time-Decay Calculation: + // lambda = ln(2) / HalfLifeDays + // Weight_i = Confidence_i * exp(-lambda * deltaDays_i) + // WeightedScore = sum(Weight_i * CompoundScore_i) / sum(Weight_i) + double lambda = Math.Log(2.0) / Math.Max(0.1, halfLifeDays); + var now = DateTime.UtcNow; + + double totalWeightedScore = 0.0; + double totalWeights = 0.0; + double sumRawScore = 0.0; + double sumConfidence = 0.0; + int positiveCount = 0; + int negativeCount = 0; + int neutralCount = 0; + + foreach (var a in recentAnalyses) + { + double deltaDays = Math.Max(0.0, (now - a.PublishedAtUtc).TotalDays); + double timeDecay = Math.Exp(-lambda * deltaDays); + double weight = Math.Max(0.01, a.Confidence) * timeDecay; + + totalWeightedScore += weight * a.CompoundScore; + totalWeights += weight; + sumRawScore += a.CompoundScore; + sumConfidence += a.Confidence; + + if (a.CompoundScore >= 0.15 || a.Label == "POSITIVE") positiveCount++; + else if (a.CompoundScore <= -0.15 || a.Label == "NEGATIVE") negativeCount++; + else neutralCount++; + } + + double finalWeightedScore = totalWeights > 0 ? Math.Round(totalWeightedScore / totalWeights, 4) : 0.0; + double finalAvgScore = Math.Round(sumRawScore / recentAnalyses.Count, 4); + double finalAvgConfidence = Math.Round(sumConfidence / recentAnalyses.Count, 4); + + string currentLabel = finalWeightedScore switch + { + > 0.50 => "VERY_BULLISH", + > 0.15 => "BULLISH", + < -0.50 => "VERY_BEARISH", + < -0.15 => "BEARISH", + _ => "NEUTRAL" + }; + + // Trend detection: Compare recent (last 7 days) vs older weighted scores + string trend = "STABLE"; + var last7DaysCutoff = now.AddDays(-7); + var veryRecent = recentAnalyses.Where(a => a.PublishedAtUtc >= last7DaysCutoff).ToList(); + var older = recentAnalyses.Where(a => a.PublishedAtUtc < last7DaysCutoff).ToList(); + + if (veryRecent.Count > 0 && older.Count > 0) + { + double recentScore = veryRecent.Average(a => a.CompoundScore); + double olderScore = older.Average(a => a.CompoundScore); + double diff = recentScore - olderScore; + if (diff > 0.20) trend = "IMPROVING"; + else if (diff < -0.20) trend = "DETERIORATING"; + } + + var latestHighlight = recentAnalyses.FirstOrDefault(a => !string.IsNullOrWhiteSpace(a.KeyHighlight))?.KeyHighlight; + + var existingSummary = await _context.CompanySentiments.FirstOrDefaultAsync(c => c.Isin == isin, ct); + if (existingSummary == null) + { + var newSummary = new CompanySentimentSummaryEntity + { + Isin = isin, + Name = companyName, + Sector = sector, + CurrentLabel = currentLabel, + AverageScore = finalAvgScore, + WeightedScore = finalWeightedScore, + AverageConfidence = finalAvgConfidence, + TotalAnalysesCount = recentAnalyses.Count, + PositiveCount = positiveCount, + NegativeCount = negativeCount, + NeutralCount = neutralCount, + LatestKeyHighlight = latestHighlight, + Trend = trend, + LastUpdatedUtc = DateTime.UtcNow, + Version = 1 + }; + _context.CompanySentiments.Add(newSummary); + } + else + { + existingSummary.Name = !string.IsNullOrWhiteSpace(companyName) ? companyName : existingSummary.Name; + existingSummary.Sector = sector ?? existingSummary.Sector; + existingSummary.CurrentLabel = currentLabel; + existingSummary.AverageScore = finalAvgScore; + existingSummary.WeightedScore = finalWeightedScore; + existingSummary.AverageConfidence = finalAvgConfidence; + existingSummary.TotalAnalysesCount = recentAnalyses.Count; + existingSummary.PositiveCount = positiveCount; + existingSummary.NegativeCount = negativeCount; + existingSummary.NeutralCount = neutralCount; + existingSummary.LatestKeyHighlight = latestHighlight ?? existingSummary.LatestKeyHighlight; + existingSummary.Trend = trend; + existingSummary.LastUpdatedUtc = DateTime.UtcNow; + existingSummary.Version++; + } + + await _context.SaveChangesAsync(ct); + await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentDbService] Updated company summary for {Isin} ({Name}): {Label} (Weighted: {Score}, Trend: {Trend})", isin, companyName, currentLabel, finalWeightedScore, trend); + return; + } + catch (DbUpdateConcurrencyException) + { + _context.ChangeTracker.Clear(); + if (attempt == maxRetries) throw; + await Task.Delay(Random.Shared.Next(50, 150) * attempt, ct); + } + catch (Exception ex) + { + _context.ChangeTracker.Clear(); + await _finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[SentimentDbService] Error updating company summary for {Isin} (Attempt {Attempt})", isin, attempt); + if (attempt == maxRetries) throw; + await Task.Delay(100, ct); + } + } + } + + private async Task UpdateSectorSummaryAsync(string sector, CancellationToken ct) + { + try + { + var companySummaries = await _context.CompanySentiments + .AsNoTracking() + .Where(c => c.Sector == sector) + .ToListAsync(ct); + + if (companySummaries.Count == 0) return; + + double avgScore = Math.Round(companySummaries.Average(c => c.WeightedScore), 4); + int totalArticles = companySummaries.Sum(c => c.TotalAnalysesCount); + int totalCompanies = companySummaries.Count; + + string currentLabel = avgScore switch + { + > 0.15 => "POSITIVE", + < -0.15 => "NEGATIVE", + _ => "NEUTRAL" + }; + + var existing = await _context.SectorSentiments.FirstOrDefaultAsync(s => s.Sector == sector, ct); + if (existing == null) + { + _context.SectorSentiments.Add(new SectorSentimentSummaryEntity + { + Sector = sector, + CurrentLabel = currentLabel, + AverageScore = avgScore, + TotalArticlesCount = totalArticles, + TotalCompaniesCount = totalCompanies, + LastUpdatedUtc = DateTime.UtcNow + }); + } + else + { + existing.CurrentLabel = currentLabel; + existing.AverageScore = avgScore; + existing.TotalArticlesCount = totalArticles; + existing.TotalCompaniesCount = totalCompanies; + existing.LastUpdatedUtc = DateTime.UtcNow; + } + + await _context.SaveChangesAsync(ct); + } + catch (Exception ex) + { + _context.ChangeTracker.Clear(); + await _finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, ex, "[SentimentDbService] Failed to update sector summary for {Sector}", sector); + } + } + + /// + public async Task GetCompanySentimentAsync(string isin, CancellationToken ct = default) + { + if (string.IsNullOrWhiteSpace(isin)) return null; + var cleanIsin = isin.Trim().ToUpperInvariant(); + return await _context.CompanySentiments + .AsNoTracking() + .FirstOrDefaultAsync(c => c.Isin == cleanIsin, ct); + } + + /// + public async Task GetIsinSummaryDtoAsync(string isin, CancellationToken ct = default) + { + if (string.IsNullOrWhiteSpace(isin)) return null; + var cleanIsin = isin.Trim().ToUpperInvariant(); + + var summary = await _context.CompanySentiments + .AsNoTracking() + .FirstOrDefaultAsync(c => c.Isin == cleanIsin, ct); + + if (summary == null) return null; + + var recentAnalyses = await _context.ArticleSentiments + .AsNoTracking() + .Where(a => a.Isin == cleanIsin) + .OrderByDescending(a => a.PublishedAtUtc) + .Take(15) + .ToListAsync(ct); + + return new IsinSentimentSummaryDto + { + Isin = summary.Isin, + CompanyName = summary.Name, + Sector = summary.Sector ?? "General", + LastUpdated = summary.LastUpdatedUtc.ToString("o"), + CurrentSummary = new IsinCurrentSummary + { + CompoundScore = summary.WeightedScore, + SentimentLabel = summary.CurrentLabel, + AvgConfidence = summary.AverageConfidence, + TotalArticlesAnalyzed = summary.TotalAnalysesCount, + PositiveArticles = summary.PositiveCount, + NegativeArticles = summary.NegativeCount, + NeutralArticles = summary.NeutralCount, + Trend = summary.Trend, + KeyHighlight = summary.LatestKeyHighlight + }, + Analyses = recentAnalyses.Select(a => new IsinAnalysisEntry + { + AnalysisId = a.Id.ToString(), + Timestamp = a.AnalyzedAtUtc.ToString("o"), + Article = new IsinAnalysisArticleRef + { + ArticleId = a.ArticleId.ToString(), + Title = a.Name, + Source = "FinlyticNews", + PublishedAt = a.PublishedAtUtc.ToString("o") + }, + FinbertResult = new FinBertResultDto + { + Label = a.Label, + CompoundScore = a.CompoundScore, + Confidence = a.Confidence, + Impact = a.Impact, + KeyHighlight = a.KeyHighlight, + Probabilities = new FinBertProbabilities + { + Positive = a.PositiveProbability, + Negative = a.NegativeProbability, + Neutral = a.NeutralProbability + } + }, + SummarySnippet = a.KeyHighlight ?? string.Empty + }).ToList() + }; + } + + /// + public async Task> GetAllCompanySentimentsAsync( + int limit, + int offset, + string? sector = null, + CancellationToken ct = default) + { + var query = _context.CompanySentiments.AsNoTracking().AsQueryable(); + + if (!string.IsNullOrWhiteSpace(sector)) + { + var cleanSector = sector.Trim(); + query = query.Where(c => c.Sector == cleanSector); + } + + return await query + .OrderByDescending(c => c.LastUpdatedUtc) + .Skip(Math.Max(0, offset)) + .Take(limit > 0 ? Math.Min(limit, 100) : 50) + .ToListAsync(ct); + } + + /// + public async Task GetSectorSentimentAsync(string sector, CancellationToken ct = default) + { + if (string.IsNullOrWhiteSpace(sector)) return null; + var cleanSector = sector.Trim(); + + var entity = await _context.SectorSentiments + .AsNoTracking() + .FirstOrDefaultAsync(s => s.Sector == cleanSector, ct); + + if (entity == null) return null; + + var activeIsins = await _context.CompanySentiments + .AsNoTracking() + .Where(c => c.Sector == cleanSector) + .Select(c => c.Isin) + .Take(20) + .ToListAsync(ct); + + return new SectorSentimentSummaryDto + { + Sector = entity.Sector, + LastUpdated = entity.LastUpdatedUtc.ToString("o"), + CurrentSummary = new SectorCurrentSummary + { + CompoundScore = entity.AverageScore, + SentimentLabel = entity.CurrentLabel, + ActiveIsins = activeIsins, + TotalArticlesAnalyzed = entity.TotalArticlesCount + } + }; + } + + /// + public async Task> GetArticleSentimentsAsync(Guid articleId, CancellationToken ct = default) + { + if (articleId == Guid.Empty) return []; + return await _context.ArticleSentiments + .AsNoTracking() + .Where(a => a.ArticleId == articleId) + .ToListAsync(ct); + } + + /// + public async Task GetArticleSentimentEntryAsync(Guid articleId, CancellationToken ct = default) + { + if (articleId == Guid.Empty) return null; + + var a = await _context.ArticleSentiments + .AsNoTracking() + .FirstOrDefaultAsync(x => x.ArticleId == articleId, ct); + + if (a == null) return null; + + return new IsinAnalysisEntry + { + AnalysisId = a.Id.ToString(), + Timestamp = a.AnalyzedAtUtc.ToString("o"), + Article = new IsinAnalysisArticleRef + { + ArticleId = a.ArticleId.ToString(), + Title = a.Name, + Source = "FinlyticNews", + PublishedAt = a.PublishedAtUtc.ToString("o") + }, + FinbertResult = new FinBertResultDto + { + Label = a.Label, + CompoundScore = a.CompoundScore, + Confidence = a.Confidence, + Impact = a.Impact, + KeyHighlight = a.KeyHighlight, + Probabilities = new FinBertProbabilities + { + Positive = a.PositiveProbability, + Negative = a.NegativeProbability, + Neutral = a.NeutralProbability + } + }, + SummarySnippet = a.KeyHighlight ?? string.Empty + }; + } + + /// + public async Task> GetSentimentTimelineAsync(string isin, int days, CancellationToken ct = default) + { + if (string.IsNullOrWhiteSpace(isin)) return []; + var cleanIsin = isin.Trim().ToUpperInvariant(); + var cutoff = DateTime.UtcNow.AddDays(-Math.Max(1, days)); + + return await _context.ArticleSentiments + .AsNoTracking() + .Where(a => a.Isin == cleanIsin && a.PublishedAtUtc >= cutoff) + .OrderByDescending(a => a.PublishedAtUtc) + .ToListAsync(ct); + } +} diff --git a/FinlyticSentiment/Services/SentimentStorageService.cs b/FinlyticSentiment/Services/SentimentStorageService.cs deleted file mode 100644 index 3466a29..0000000 --- a/FinlyticSentiment/Services/SentimentStorageService.cs +++ /dev/null @@ -1,345 +0,0 @@ -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; - -namespace FinlyticSentiment.Services; - -/// -/// Defines the persistence contract for maintaining two-stage ISIN and Sector JSON sentiment summaries in the file system. -/// -public interface ISentimentStorageService -{ - Task UpdateIsinSummaryAsync(string isin, string companyName, string sector, NewsArticleDto article, FinBertResultDto finbert); - Task UpdateSectorSummaryAsync(string sector, string isin, string articleId, FinBertResultDto finbert); - Task SaveArticleSentimentAsync(NewsArticleDto article, FinBertResultDto finbert); - Task GetArticleSentimentAsync(string articleId); - Task GetIsinSummaryAsync(string isin); -} - -public class SentimentStorageService : ISentimentStorageService -{ - private static readonly ConcurrentDictionary FileLocks = new(); - - private readonly IFinlyticLogger _finlyticLogger; - private readonly string _basePath; - private readonly JsonSerializerOptions _jsonOptions; - - public SentimentStorageService(IConfiguration configuration, IFinlyticLogger finlyticLogger) - { - _finlyticLogger = finlyticLogger; - _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"); - var entry = new IsinAnalysisEntry - { - AnalysisId = $"sent_{Guid.NewGuid():N}", - Timestamp = nowIso, - Article = new IsinAnalysisArticleRef - { - ArticleId = cleanId, - Title = article.Title, - PublishedAt = article.PublishedAt.ToString("yyyy-MM-ddTHH:mm:ssZ"), - Source = article.Author ?? "FinlyticNews" - }, - FinbertResult = finbert, - SummarySnippet = finbert.SummarySnippet ?? article.Summary ?? string.Empty - }; - - var json = JsonSerializer.Serialize(entry, _jsonOptions); - await File.WriteAllTextAsync(filePath, json); - await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentStorageService] Successfully saved article sentiment file: {Path}", filePath); - } - catch (Exception ex) - { - await _finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[SentimentStorageService] Failed to write article sentiment file: {Path}", filePath); - } - finally - { - fileLock.Release(); - } - } - - /// - public async Task GetArticleSentimentAsync(string articleId) - { - if (string.IsNullOrWhiteSpace(articleId)) return null; - - string cleanId = articleId.Trim(); - string articleFilePath = Path.Combine(_basePath, "articles", $"{cleanId}.json"); - - 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) - { - await _finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, ex, "[SentimentStorageService] Failed to read article sentiment file: {Path}", articleFilePath); - } - finally - { - fileLock.Release(); - } - } - - return null; - } - - /// - public async Task GetIsinSummaryAsync(string isin) - { - if (string.IsNullOrWhiteSpace(isin)) return null; - - string cleanIsin = isin.Trim().ToUpperInvariant(); - string 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) - { - await _finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[SentimentStorageService] Error reading ISIN summary file: {Path}", filePath); - return null; - } - finally - { - fileLock.Release(); - } - } - - /// - public async Task UpdateIsinSummaryAsync(string isin, string companyName, string sector, NewsArticleDto article, FinBertResultDto finbert) - { - if (string.IsNullOrWhiteSpace(isin)) return; - - string cleanIsin = isin.Trim().ToUpperInvariant(); - string filePath = Path.Combine(_basePath, "isin", $"{cleanIsin}.json"); - - var fileLock = FileLocks.GetOrAdd(filePath, _ => new SemaphoreSlim(1, 1)); - await fileLock.WaitAsync(); - - try - { - var analyses = new List(); - if (File.Exists(filePath)) - { - var existingJson = await File.ReadAllTextAsync(filePath); - var existing = JsonSerializer.Deserialize(existingJson, _jsonOptions); - if (existing?.Analyses != null) - { - analyses.AddRange(existing.Analyses); - } - } - - 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"); - var newEntry = new IsinAnalysisEntry - { - AnalysisId = $"sent_{Guid.NewGuid():N}", - Timestamp = nowIso, - Article = new IsinAnalysisArticleRef - { - 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 ?? string.Empty - }; - - analyses.Add(newEntry); - - var cutoff = DateTime.UtcNow.AddDays(-14); - var validAnalyses = analyses.Where(a => - { - 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; - } - - int total = updatedAnalyses.Count; - double avgCompound = total > 0 ? totalCompound / total : 0.0; - double avgConf = total > 0 ? totalConf / total : 0.0; - - string overallLabel = "NEUTRAL"; - if (avgCompound >= 0.15) overallLabel = "POSITIVE"; - else if (avgCompound <= -0.15) overallLabel = "NEGATIVE"; - - var summary = new IsinSentimentSummaryDto - { - Isin = cleanIsin, - CompanyName = companyName, - Sector = sector, - LastUpdated = nowIso, - CurrentSummary = new IsinCurrentSummary - { - CompoundScore = Math.Round(avgCompound, 4), - SentimentLabel = overallLabel, - AvgConfidence = Math.Round(avgConf, 4), - TotalArticlesAnalyzed = total, - Text = $"Synthesized sentiment across {total} articles is {overallLabel}." - }, - Analyses = updatedAnalyses - }; - - 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 - { - fileLock.Release(); - } - } - - /// - public async Task UpdateSectorSummaryAsync(string sector, string isin, string articleId, FinBertResultDto finbert) - { - if (string.IsNullOrWhiteSpace(sector)) return; - - 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 - { - var analyses = new List(); - if (File.Exists(filePath)) - { - var existingJson = await File.ReadAllTextAsync(filePath); - var existing = JsonSerializer.Deserialize(existingJson, _jsonOptions); - if (existing?.Analyses != null) - { - analyses.AddRange(existing.Analyses); - } - } - - string cleanIsin = isin.Trim().ToUpperInvariant(); - string nowIso = DateTime.UtcNow.ToString("yyyy-MM-ddTHH:mm:ssZ"); - - analyses.RemoveAll(a => string.Equals(a.ArticleId, articleId, StringComparison.OrdinalIgnoreCase) && string.Equals(a.RelatedIsin, cleanIsin, StringComparison.OrdinalIgnoreCase)); - - analyses.Add(new SectorAnalysisEntry - { - AnalysisId = $"sec_{Guid.NewGuid():N}", - Timestamp = nowIso, - RelatedIsin = cleanIsin, - ArticleId = articleId, - FinbertResult = finbert - }); - - var cutoff = DateTime.UtcNow.AddDays(-14); - var updatedAnalyses = analyses.Where(a => - { - 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.IsNullOrEmpty(i)).Distinct().ToList(); - double totalSectorCompound = updatedAnalyses.Sum(s => s.FinbertResult.CompoundScore); - double avgSectorCompound = updatedAnalyses.Count > 0 ? totalSectorCompound / updatedAnalyses.Count : 0.0; - - string sectorLabel = "NEUTRAL"; - if (avgSectorCompound >= 0.15) sectorLabel = "POSITIVE"; - else if (avgSectorCompound <= -0.15) sectorLabel = "NEGATIVE"; - - var summary = new SectorSentimentSummaryDto - { - Sector = sector, - LastUpdated = nowIso, - CurrentSummary = new SectorCurrentSummary - { - CompoundScore = Math.Round(avgSectorCompound, 4), - SentimentLabel = sectorLabel, - ActiveIsins = activeIsins, - Text = $"Sector {sector} aggregate sentiment: {sectorLabel}." - }, - Analyses = updatedAnalyses - }; - - 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(); - } - } -} \ No newline at end of file diff --git a/FinlyticSentiment/Services/SettingsDbService.cs b/FinlyticSentiment/Services/SettingsDbService.cs deleted file mode 100644 index 8d99d9d..0000000 --- a/FinlyticSentiment/Services/SettingsDbService.cs +++ /dev/null @@ -1,115 +0,0 @@ -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 index 091a54a..ca6b13a 100644 --- a/FinlyticSentiment/Util/SentimentMqttClient.cs +++ b/FinlyticSentiment/Util/SentimentMqttClient.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Linq; using System.Text.Json; using System.Threading; using System.Threading.Tasks; @@ -10,8 +11,8 @@ using FinlyticCore.Dtos.Settings; using FinlyticCore.Models; using FinlyticCore.Services; using FinlyticCore.Util; +using FinlyticSentiment.Entities; using FinlyticSentiment.Services; -using FinlyticSentiment.Util; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; @@ -20,7 +21,7 @@ using Microsoft.Extensions.Logging; namespace FinlyticSentiment.Util; /// -/// Managed MQTT client for requesting pending news articles and updating sentiment results. +/// Managed MQTT client for sentiment evaluations and RPC queries. /// public class SentimentMqttClient : ManagedMqttClient, IHostedService { @@ -43,12 +44,7 @@ public class SentimentMqttClient : ManagedMqttClient, IHostedService /// 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()}" - }; + var config = MqttConfiguration.FromConfiguration(_configuration, "FinlyticSentiment"); _logger.LogInformation("Starting Sentiment MQTT client. Host: {Host}, ClientId: {ClientId}", config.Host, config.ClientId); await ConnectAsync(config); @@ -71,403 +67,226 @@ public class SentimentMqttClient : ManagedMqttClient, IHostedService /// protected override async Task OnConnectedAsync() { - _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/#"); + _logger.LogInformation("Sentiment MQTT client connected. Registering RPC topic subscriptions..."); + + await SubscribeAsync(MqttTopics.ResponseWildcard); + await SubscribeRpcAsync(MqttTopics.RequestFilter(MqttTopics.Channels.SentimentGetIsin), HandleSentimentGetIsinRpcAsync); + await SubscribeRpcAsync(MqttTopics.RequestFilter(MqttTopics.Channels.SentimentGetSector), HandleSentimentGetSectorRpcAsync); + await SubscribeRpcAsync(MqttTopics.RequestFilter(MqttTopics.Channels.SentimentGetArticle), HandleSentimentGetArticleRpcAsync); + await SubscribeRpcAsync>(MqttTopics.RequestFilter(MqttTopics.Channels.SentimentGetAll), HandleSentimentGetAllRpcAsync); + await SubscribeRpcAsync(MqttTopics.RequestFilter(MqttTopics.Channels.SentimentAnalyze), HandleSentimentAnalyzeRpcAsync); + await SubscribeRpcAsync>(MqttTopics.RequestFilter(MqttTopics.Channels.SentimentSettingsGetAll), HandleSettingsGetAllRpcAsync); + await SubscribeRpcAsync, List>(MqttTopics.RequestFilter(MqttTopics.Channels.SentimentSettingsUpdate), HandleSettingsUpdateRpcAsync); + await SubscribeAsync(MqttTopics.RequestFilter(MqttTopics.Channels.HealthPing), HandleHealthPingRpcAsync); + await SubscribeAsync(MqttTopics.NewsCompleted, HandleNewsCompletedBroadcastAsync); FinlyticCore.Services.FinlyticLogBroadcaster.OnLogPublished = async (logDto) => { if (IsConnected && string.Equals(logDto.ServiceName, "FinlyticSentiment", StringComparison.OrdinalIgnoreCase)) { - await PublishAsync("finlytic/logs/FinlyticSentiment", logDto); + await PublishAsync(MqttTopics.Logs("FinlyticSentiment"), logDto); } }; } - /// - protected override async Task OnMessageReceivedAsync(string topic, string payload) + /// + /// Requests pending news articles from FinlyticNews via MQTT RPC. + /// + public async Task> GetPendingArticlesAsync(int limit = 10) { - if (string.IsNullOrWhiteSpace(topic)) return; - - if (topic.StartsWith("services/config/updated", StringComparison.OrdinalIgnoreCase)) + try { - if (topic.EndsWith("FinlyticSentiment", StringComparison.OrdinalIgnoreCase)) - { - await OnConfigUpdatedAsync(payload); - } - return; + var req = new LimitRequest(limit); + var response = await SendRpcRequestAsync, LimitRequest>( + MqttTopics.Channels.NewsGetPending, + req, + TimeSpan.FromSeconds(5) + ); + return response ?? []; } - - if (topic.Equals("services/news/completed", StringComparison.OrdinalIgnoreCase)) + catch (Exception ex) { - await OnNewsCompletedAsync(payload); - return; - } - - var lastSlash = topic.LastIndexOf('/'); - if (lastSlash < 0 || lastSlash >= topic.Length - 1) return; - - var correlationId = topic.Substring(lastSlash + 1); - - if (topic.StartsWith("services/request/sentiment_GetArticle", StringComparison.OrdinalIgnoreCase)) - { - await OnSentimentGetArticleAsync(payload, correlationId); - } - else if (topic.StartsWith("services/request/sentiment_GetIsin", StringComparison.OrdinalIgnoreCase)) - { - await OnSentimentGetIsinAsync(payload, correlationId); - } - else if (topic.StartsWith("services/request/sentiment_Analyze", StringComparison.OrdinalIgnoreCase)) - { - 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); + _logger.LogError(ex, "Failed to retrieve pending articles from FinlyticNews via news_GetPending"); + return []; } } - private async Task OnSettingsGetAllAsync(string correlationId) + /// + /// Updates the article processing status in FinlyticNews. + /// + public async Task UpdateArticleStatusAsync(Guid articleId, string status) + { + try + { + var req = new UpdateNewsStatusRequest(articleId, status); + await SendRpcRequestAsync( + MqttTopics.Channels.NewsUpdateStatus, + req, + TimeSpan.FromSeconds(5) + ); + } + catch (Exception ex) + { + _logger.LogError(ex, "Failed to update article status for {ArticleId} to '{Status}'", articleId, status); + } + } + + /// + /// Broadcasts an updated sentiment result for an asset over MQTT. + /// + public async Task BroadcastSentimentResultAsync(string isin, IsinSentimentSummaryDto summary) + { + if (string.IsNullOrWhiteSpace(isin) || summary == null) return; + await PublishAsync(MqttTopics.SentimentStream(isin), summary); + } + + private async Task HandleSentimentGetIsinRpcAsync(GetSentimentByIsinRequest? req, string correlationId) + { + var targetIsin = req?.Isin; + if (string.IsNullOrWhiteSpace(targetIsin)) return null; + + using var scope = _scopeFactory.CreateScope(); + var finlyticLogger = scope.ServiceProvider.GetRequiredService>(); + var dbService = scope.ServiceProvider.GetRequiredService(); + + await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Processing RPC sentiment_GetIsin for {Isin} [CorrelationId: {CorrelationId}]", targetIsin, correlationId); + return await dbService.GetIsinSummaryDtoAsync(targetIsin); + } + + private async Task HandleSentimentGetSectorRpcAsync(GetSectorSentimentRequest? req, string correlationId) + { + var sector = req?.Sector; + if (string.IsNullOrWhiteSpace(sector)) return null; + + using var scope = _scopeFactory.CreateScope(); + var finlyticLogger = scope.ServiceProvider.GetRequiredService>(); + var dbService = scope.ServiceProvider.GetRequiredService(); + + await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Processing RPC sentiment_GetSector for {Sector} [CorrelationId: {CorrelationId}]", sector, correlationId); + return await dbService.GetSectorSentimentAsync(sector); + } + + private async Task HandleSentimentGetArticleRpcAsync(ArticleRequest? req, string correlationId) + { + var targetId = req?.ArticleId ?? req?.Id; + if (string.IsNullOrWhiteSpace(targetId) || !Guid.TryParse(targetId, out var articleId)) + return null; + + using var scope = _scopeFactory.CreateScope(); + var finlyticLogger = scope.ServiceProvider.GetRequiredService>(); + var dbService = scope.ServiceProvider.GetRequiredService(); + + await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Processing RPC sentiment_GetArticle for {ArticleId} [CorrelationId: {CorrelationId}]", articleId, correlationId); + return await dbService.GetArticleSentimentEntryAsync(articleId); + } + + private async Task> HandleSentimentGetAllRpcAsync(PaginatedRequest? req, string correlationId) + { + using var scope = _scopeFactory.CreateScope(); + var finlyticLogger = scope.ServiceProvider.GetRequiredService>(); + var dbService = scope.ServiceProvider.GetRequiredService(); + + await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Processing RPC sentiment_GetAll [CorrelationId: {CorrelationId}]", correlationId); + return await dbService.GetAllCompanySentimentsAsync(req?.Limit ?? 50, req?.Offset ?? 0); + } + + private async Task HandleSentimentAnalyzeRpcAsync(JsonElement rawPayload, string correlationId) + { + if (rawPayload.ValueKind == JsonValueKind.Undefined || rawPayload.ValueKind == JsonValueKind.Null) return null; + + using var scope = _scopeFactory.CreateScope(); + var finlyticLogger = scope.ServiceProvider.GetRequiredService>(); + var analyzer = scope.ServiceProvider.GetRequiredService(); + var dbService = scope.ServiceProvider.GetRequiredService(); + + NewsArticleDto? article = null; + try + { + var rawText = rawPayload.GetRawText(); + if (rawPayload.TryGetProperty("contentRaw", out _) || rawPayload.TryGetProperty("sourceUrl", out _)) + { + article = JsonSerializer.Deserialize(rawText); + } + else if (rawPayload.TryGetProperty("articleId", out var articleIdProp)) + { + var articleIdStr = articleIdProp.GetString(); + if (Guid.TryParse(articleIdStr, out var parsedGuid)) + { + article = await SendRpcRequestAsync( + MqttTopics.Channels.NewsGetById, + new ArticleRequest(articleIdStr, articleIdStr), + TimeSpan.FromSeconds(5) + ); + } + } + } + catch (Exception ex) + { + await finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, ex, "[SentimentMqttClient] Failed to deserialize payload in sentiment_Analyze"); + } + + if (article == null || article.Id == Guid.Empty) return null; + + await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Processing RPC sentiment_Analyze for {ArticleId} [CorrelationId: {CorrelationId}]", article.Id, correlationId); + var result = await analyzer.AnalyzeArticleAsync(article); + + if (result != null && article.MatchedAssets != null) + { + foreach (var asset in article.MatchedAssets) + { + if (string.IsNullOrWhiteSpace(asset.Isin)) continue; + await dbService.SaveArticleSentimentAsync(article.Id, asset.Isin, asset.Name, null, article.PublishedAt, result); + } + + return await dbService.GetArticleSentimentEntryAsync(article.Id); + } + + return null; + } + + private async Task> HandleSettingsGetAllRpcAsync(object? _, 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."); - } + await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticSentiment] [Settings_GetAll] Retrieving service dynamic settings [CorrelationId: {CorrelationId}]", correlationId); + return await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) }); } - private async Task OnSettingsUpdateAsync(string payload, string correlationId) + private async Task> HandleSettingsUpdateRpcAsync(Dictionary? updates, 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 + if (updates != null && updates.Count > 0) { - 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."); + await settingsService.UpdateSettingsAsync(updates); + await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticSentiment] [Settings_Update] Successfully updated {Count} settings in database and cache.", updates.Count); } + return await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) }); } - private async Task OnConfigUpdatedAsync(string payload) + private async Task HandleHealthPingRpcAsync(object? _, string topic, string correlationId) { - try + if (topic.Contains("FinlyticSentiment", StringComparison.OrdinalIgnoreCase)) { - using var doc = JsonDocument.Parse(payload); - if (doc.RootElement.TryGetProperty("settings", out var settingsProp)) - { - var dict = JsonSerializer.Deserialize>(settingsProp.GetRawText()); - if (dict != null && dict.Count > 0) - { - using var scope = _scopeFactory.CreateScope(); - var settings = scope.ServiceProvider.GetRequiredService(); - await settings.UpdateSettingsAsync(dict); - } - } - } - catch { } - } - - 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}"; + string respTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.HealthPing, 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); + await finlyticLogger.LogInfoAsync(SettingKeys.HealthPingChannel, "Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", correlationId); } } - private async Task OnNewsCompletedAsync(string payload) + private async Task HandleNewsCompletedBroadcastAsync(NewsArticleDto? article, string topic, string correlationId) { - if (string.IsNullOrWhiteSpace(payload)) return; - try + if (article != null && OnArticleReceived != null) { - var article = JsonSerializer.Deserialize(payload, typeof(NewsArticleDto), FinlyticJsonSerializerContext.Default) as NewsArticleDto; - if (article != null && article.Id != Guid.Empty && OnArticleReceived != null) - { - 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); - } + await OnArticleReceived.Invoke(article); } - catch { } - } - - private async Task OnSentimentGetArticleAsync(string payload, string 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 articleId = request?.ArticleId ?? request?.Id; - - if (string.IsNullOrWhiteSpace(articleId)) - { - await finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Missing articleId in request payload."); - return; - } - - var storageService = scope.ServiceProvider.GetRequiredService(); - var sentimentEntry = await storageService.GetArticleSentimentAsync(articleId); - - string responseTopic = $"services/response/sentiment_GetArticle/{correlationId}"; - 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) - { - await finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[SentimentMqttClient] Failed to process sentiment_GetArticle RPC request."); - } - } - - private async Task OnSentimentGetIsinAsync(string payload, string 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 isin = request?.Isin; - - if (string.IsNullOrWhiteSpace(isin)) - { - await finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Missing ISIN in request payload."); - return; - } - - var storageService = scope.ServiceProvider.GetRequiredService(); - var isinSummary = await storageService.GetIsinSummaryAsync(isin); - - string responseTopic = $"services/response/sentiment_GetIsin/{correlationId}"; - 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) - { - await finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[SentimentMqttClient] Failed to process sentiment_GetIsin RPC request."); - } - } - - private async Task OnSentimentAnalyzeAsync(string payload, string 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)) - { - 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))) - { - await finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Missing ArticleId or Isin in sentiment_Analyze request payload."); - await PublishAsync(responseTopic, (object?)null); - return; - } - - var storageService = scope.ServiceProvider.GetRequiredService(); - var analyzerService = scope.ServiceProvider.GetRequiredService(); - - object? result = null; - - if (!string.IsNullOrWhiteSpace(request.ArticleId)) - { - var cleanArticleId = request.ArticleId.Trim(); - await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Processing article analysis for ArticleId: {ArticleId} (ForceReload: {ForceReload})", cleanArticleId, request.ForceReload); - - IsinAnalysisEntry? existingEntry = null; - - if (!request.ForceReload) - { - existingEntry = await storageService.GetArticleSentimentAsync(cleanArticleId); - } - - if (existingEntry != null) - { - result = existingEntry; - } - else - { - 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); - - if (finbertResult == null) - { - await finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] FinBERT analysis returned NULL for article {ArticleId}. Aborting manual analysis.", cleanArticleId); - await PublishAsync(responseTopic, (object?)null); - return; - } - - 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); - } - } - - await UpdateArticleStatusAsync(article.Id, "Analyzed"); - - result = await storageService.GetArticleSentimentAsync(cleanArticleId); - } - else - { - await finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Could not retrieve article {ArticleId} from FinlyticNews for re-analysis.", cleanArticleId); - } - } - } - } - else if (!string.IsNullOrWhiteSpace(request.Isin)) - { - var cleanIsin = request.Isin.Trim(); - await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Fetching sentiment summary for ISIN: {Isin}", cleanIsin); - - result = await storageService.GetIsinSummaryAsync(cleanIsin); - } - - await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Publishing RPC sentiment_Analyze response to {ResponseTopic}", responseTopic); - await PublishAsync(responseTopic, result); - } - catch (Exception ex) - { - await finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[SentimentMqttClient] Failed to process sentiment_Analyze RPC request."); - await PublishAsync(responseTopic, (object?)null); - } - } - - 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, "Error executing MQTT RPC for news_GetPending."); - } - - return []; - } - - 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, "Error executing MQTT RPC for news_UpdateStatus (ID: {Id}).", id); - } - - return false; } } \ No newline at end of file diff --git a/FinlyticSentiment/Util/SettingKeys.cs b/FinlyticSentiment/Util/SettingKeys.cs index 83e39fb..782582d 100644 --- a/FinlyticSentiment/Util/SettingKeys.cs +++ b/FinlyticSentiment/Util/SettingKeys.cs @@ -13,8 +13,8 @@ public static class SettingKeys 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 SentimentWindowDays = new("Sentiment.WindowDays", 30); + public static readonly SettingKey TimeDecayHalfLifeDays = new("Sentiment.TimeDecayHalfLifeDays", 7.0); public static readonly SettingKey EnableAutoSummarization = new("Feature.EnableAutoSummarization", true); // --- N8N / Webhook-Konfiguration ---