Kleidukos/Finlytic
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sentiment): add persistent sentiment entities, summaries, SentimentDbService and MQTT RPC refactoring

Browse Source
This commit is contained in:
Kleidukos
2026-08-24 21:35:48 +02:00
parent 600ccf299e
commit 12e7b57b16
24 changed files with 1510 additions and 1300 deletions
Download Patch File Download Diff File Expand all files Collapse all files
FinlyticSentiment/Util/SentimentMqttClient.cs
+180 -361
View File
@@ -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;
/// <summary>
/// Managed MQTT client for requesting pending news articles and updating sentiment results.
/// Managed MQTT client for sentiment evaluations and RPC queries.
/// </summary>
public class SentimentMqttClient : ManagedMqttClient, IHostedService
{
@@ -43,12 +44,7 @@ public class SentimentMqttClient : ManagedMqttClient, IHostedService
/// </summary>
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
/// <inheritdoc />
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<GetSentimentByIsinRequest, IsinSentimentSummaryDto?>(MqttTopics.RequestFilter(MqttTopics.Channels.SentimentGetIsin), HandleSentimentGetIsinRpcAsync);
await SubscribeRpcAsync<GetSectorSentimentRequest, SectorSentimentSummaryDto?>(MqttTopics.RequestFilter(MqttTopics.Channels.SentimentGetSector), HandleSentimentGetSectorRpcAsync);
await SubscribeRpcAsync<ArticleRequest, IsinAnalysisEntry?>(MqttTopics.RequestFilter(MqttTopics.Channels.SentimentGetArticle), HandleSentimentGetArticleRpcAsync);
await SubscribeRpcAsync<PaginatedRequest, List<CompanySentimentSummaryEntity>>(MqttTopics.RequestFilter(MqttTopics.Channels.SentimentGetAll), HandleSentimentGetAllRpcAsync);
await SubscribeRpcAsync<JsonElement, IsinAnalysisEntry?>(MqttTopics.RequestFilter(MqttTopics.Channels.SentimentAnalyze), HandleSentimentAnalyzeRpcAsync);
await SubscribeRpcAsync<object, List<DynamicSettingDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.SentimentSettingsGetAll), HandleSettingsGetAllRpcAsync);
await SubscribeRpcAsync<Dictionary<string, object?>, List<DynamicSettingDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.SentimentSettingsUpdate), HandleSettingsUpdateRpcAsync);
await SubscribeAsync<object>(MqttTopics.RequestFilter(MqttTopics.Channels.HealthPing), HandleHealthPingRpcAsync);
await SubscribeAsync<NewsArticleDto>(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);
}
};
}
/// <inheritdoc />
protected override async Task OnMessageReceivedAsync(string topic, string payload)
/// <summary>
/// Requests pending news articles from FinlyticNews via MQTT RPC.
/// </summary>
public async Task<List<NewsArticleDto>> 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<List<NewsArticleDto>, 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)
/// <summary>
/// Updates the article processing status in FinlyticNews.
/// </summary>
public async Task UpdateArticleStatusAsync(Guid articleId, string status)
{
try
{
var req = new UpdateNewsStatusRequest(articleId, status);
await SendRpcRequestAsync<UpdateNewsStatusResponse, UpdateNewsStatusRequest>(
MqttTopics.Channels.NewsUpdateStatus,
req,
TimeSpan.FromSeconds(5)
);
}
catch (Exception ex)
{
_logger.LogError(ex, "Failed to update article status for {ArticleId} to '{Status}'", articleId, status);
}
}
/// <summary>
/// Broadcasts an updated sentiment result for an asset over MQTT.
/// </summary>
public async Task BroadcastSentimentResultAsync(string isin, IsinSentimentSummaryDto summary)
{
if (string.IsNullOrWhiteSpace(isin) || summary == null) return;
await PublishAsync(MqttTopics.SentimentStream(isin), summary);
}
private async Task<IsinSentimentSummaryDto?> 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<IFinlyticLogger<SentimentMqttClient>>();
var dbService = scope.ServiceProvider.GetRequiredService<ISentimentDbService>();
await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Processing RPC sentiment_GetIsin for {Isin} [CorrelationId: {CorrelationId}]", targetIsin, correlationId);
return await dbService.GetIsinSummaryDtoAsync(targetIsin);
}
private async Task<SectorSentimentSummaryDto?> 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<IFinlyticLogger<SentimentMqttClient>>();
var dbService = scope.ServiceProvider.GetRequiredService<ISentimentDbService>();
await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Processing RPC sentiment_GetSector for {Sector} [CorrelationId: {CorrelationId}]", sector, correlationId);
return await dbService.GetSectorSentimentAsync(sector);
}
private async Task<IsinAnalysisEntry?> 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<IFinlyticLogger<SentimentMqttClient>>();
var dbService = scope.ServiceProvider.GetRequiredService<ISentimentDbService>();
await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Processing RPC sentiment_GetArticle for {ArticleId} [CorrelationId: {CorrelationId}]", articleId, correlationId);
return await dbService.GetArticleSentimentEntryAsync(articleId);
}
private async Task<List<CompanySentimentSummaryEntity>> HandleSentimentGetAllRpcAsync(PaginatedRequest? req, string correlationId)
{
using var scope = _scopeFactory.CreateScope();
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<SentimentMqttClient>>();
var dbService = scope.ServiceProvider.GetRequiredService<ISentimentDbService>();
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<IsinAnalysisEntry?> 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<IFinlyticLogger<SentimentMqttClient>>();
var analyzer = scope.ServiceProvider.GetRequiredService<IFinBertAnalyzerService>();
var dbService = scope.ServiceProvider.GetRequiredService<ISentimentDbService>();
NewsArticleDto? article = null;
try
{
var rawText = rawPayload.GetRawText();
if (rawPayload.TryGetProperty("contentRaw", out _) || rawPayload.TryGetProperty("sourceUrl", out _))
{
article = JsonSerializer.Deserialize<NewsArticleDto>(rawText);
}
else if (rawPayload.TryGetProperty("articleId", out var articleIdProp))
{
var articleIdStr = articleIdProp.GetString();
if (Guid.TryParse(articleIdStr, out var parsedGuid))
{
article = await SendRpcRequestAsync<NewsArticleDto, ArticleRequest>(
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<List<DynamicSettingDto>> HandleSettingsGetAllRpcAsync(object? _, string correlationId)
{
using var scope = _scopeFactory.CreateScope();
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<SentimentMqttClient>>();
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
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<List<DynamicSettingDto>> HandleSettingsUpdateRpcAsync(Dictionary<string, object?>? updates, string correlationId)
{
if (string.IsNullOrWhiteSpace(payload)) return;
using var scope = _scopeFactory.CreateScope();
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<SentimentMqttClient>>();
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticSentiment] [Settings_Update] Processing settings update RPC [CorrelationId: {CorrelationId}]", correlationId);
try
if (updates != null && updates.Count > 0)
{
Dictionary<string, object?>? updates = null;
try
{
updates = JsonSerializer.Deserialize<Dictionary<string, object?>>(payload);
}
catch
{
var list = JsonSerializer.Deserialize<List<DynamicSettingDto>>(payload);
if (list != null)
{
updates = new Dictionary<string, object?>();
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<Dictionary<string, object?>>(settingsProp.GetRawText());
if (dict != null && dict.Count > 0)
{
using var scope = _scopeFactory.CreateScope();
var settings = scope.ServiceProvider.GetRequiredService<ISettingsService>();
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<IFinlyticLogger<SentimentMqttClient>>();
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<IFinlyticLogger<SentimentMqttClient>>();
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<IFinlyticLogger<SentimentMqttClient>>();
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<ISentimentStorageService>();
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<IFinlyticLogger<SentimentMqttClient>>();
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<ISentimentStorageService>();
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<IFinlyticLogger<SentimentMqttClient>>();
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<ISentimentStorageService>();
var analyzerService = scope.ServiceProvider.GetRequiredService<IFinBertAnalyzerService>();
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<NewsArticleDto, ArticleRequest>(
"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<List<NewsArticleDto>> GetPendingArticlesAsync(int limit = 10)
{
try
{
var payload = new LimitRequest(Math.Min(limit, 10));
var articles = await SendRpcRequestAsync<List<NewsArticleDto>, 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<bool> UpdateArticleStatusAsync(Guid id, string status = "Analyzed")
{
try
{
var request = new UpdateNewsStatusRequest(id, status);
var response = await SendRpcRequestAsync<UpdateNewsStatusResponse, UpdateNewsStatusRequest>("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;
}
}
FinlyticSentiment/Util/SettingKeys.cs
+2 -2
View File
@@ -13,8 +13,8 @@ public static class SettingKeys
public static readonly SettingKey<int> MaxBatchSize = new("FinBert.MaxBatchSize", 10);
public static readonly SettingKey<double> MinimumConfidenceThreshold = new("FinBert.MinConfidenceThreshold", 0.60);
public static readonly SettingKey<int> TimeoutSeconds = new("FinBert.TimeoutSeconds", 30);
public static readonly SettingKey<int> SentimentWindowDays = new("Sentiment.WindowDays", 14);
public static readonly SettingKey<double> DecayFactorPerDay = new("Sentiment.DecayFactorPerDay", 0.90);
public static readonly SettingKey<int> SentimentWindowDays = new("Sentiment.WindowDays", 30);
public static readonly SettingKey<double> TimeDecayHalfLifeDays = new("Sentiment.TimeDecayHalfLifeDays", 7.0);
public static readonly SettingKey<bool> EnableAutoSummarization = new("Feature.EnableAutoSummarization", true);
// --- N8N / Webhook-Konfiguration ---