feat(Sentiment): update sentiment service
This commit is contained in:
@@ -0,0 +1,216 @@
|
||||
using System.Collections.Concurrent;
|
||||
using FinlyticCore.Dtos.News;
|
||||
using FinlyticSentiment.Util;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace FinlyticSentiment.Services;
|
||||
|
||||
/// <summary>
|
||||
/// Background hosted worker executing periodic sentiment analysis sweeps on pending news articles
|
||||
/// and processing real-time article broadcasts.
|
||||
/// </summary>
|
||||
public class SentimentBackgroundService : BackgroundService
|
||||
{
|
||||
private readonly SentimentMqttClient _mqttClient;
|
||||
private readonly IFinBertAnalyzerService _analyzer;
|
||||
private readonly ISentimentStorageService _storage;
|
||||
private readonly IServiceScopeFactory _scopeFactory;
|
||||
private readonly ILogger<SentimentBackgroundService> _logger;
|
||||
|
||||
// In-Memory Mutex / Cache zur Vermeidung doppelter Verarbeitung (Race Conditions zwischen Broadcast & Sweep)
|
||||
private static readonly ConcurrentDictionary<Guid, byte> ProcessingArticles = new();
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="SentimentBackgroundService"/> class.
|
||||
/// </summary>
|
||||
public SentimentBackgroundService(
|
||||
SentimentMqttClient mqttClient,
|
||||
IFinBertAnalyzerService analyzer,
|
||||
ISentimentStorageService storage,
|
||||
IServiceScopeFactory scopeFactory,
|
||||
ILogger<SentimentBackgroundService> logger)
|
||||
{
|
||||
_mqttClient = mqttClient;
|
||||
_analyzer = analyzer;
|
||||
_storage = storage;
|
||||
_scopeFactory = scopeFactory;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
_logger.LogInformation("[{Channel}] FinlyticSentiment Background Service started.", "SentimentChannel");
|
||||
|
||||
// Realtime Broadcast Event registrieren (Echtzeit-Artikel)
|
||||
_mqttClient.OnArticleReceived += async (article) =>
|
||||
{
|
||||
await ProcessSingleArticleAsync(article, stoppingToken);
|
||||
};
|
||||
|
||||
// Kurze Initialisierungs-Verzögerung für die MQTT-Verbindung
|
||||
await Task.Delay(3000, stoppingToken);
|
||||
|
||||
while (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
int maxBatchSize;
|
||||
int sweepIntervalMinutes;
|
||||
|
||||
using (var scope = _scopeFactory.CreateScope())
|
||||
{
|
||||
var settingsDb = scope.ServiceProvider.GetRequiredService<ISettingsDbService>();
|
||||
var settings = await settingsDb.GetSettingsAsync();
|
||||
maxBatchSize = settings.MaxBatchSize;
|
||||
sweepIntervalMinutes = settings.SweepIntervalMinutes;
|
||||
}
|
||||
|
||||
var interval = TimeSpan.FromMinutes(Math.Max(1, sweepIntervalMinutes));
|
||||
|
||||
try
|
||||
{
|
||||
await PerformSentimentSweepAsync(maxBatchSize, stoppingToken);
|
||||
}
|
||||
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
// Normales Beenden beim Stoppen des Hosts
|
||||
break;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "[{Channel}] Unhandled exception encountered during sentiment sweep cycle.",
|
||||
"SentimentChannel");
|
||||
}
|
||||
|
||||
_logger.LogInformation("[{Channel}] Waiting {Minutes} minute(s) until next sentiment sweep...",
|
||||
"SentimentChannel", interval.TotalMinutes);
|
||||
|
||||
// Verwendet PeriodicTimer oder CancellationToken-resistenten Delay
|
||||
using var timer = new PeriodicTimer(interval);
|
||||
try
|
||||
{
|
||||
await timer.WaitForNextTickAsync(stoppingToken);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
_logger.LogInformation("[{Channel}] FinlyticSentiment Background Service is shutting down gracefully.",
|
||||
"SentimentChannel");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Performs a single batch sweep of pending news articles.
|
||||
/// </summary>
|
||||
private async Task PerformSentimentSweepAsync(int maxBatchSize, CancellationToken cancellationToken)
|
||||
{
|
||||
_logger.LogInformation("[{Channel}] Starting sentiment sweep for pending news articles (Limit: {Limit})...",
|
||||
"SentimentChannel", maxBatchSize);
|
||||
|
||||
List<NewsArticleDto> pendingArticles = await _mqttClient.GetPendingArticlesAsync(limit: maxBatchSize);
|
||||
if (pendingArticles.Count == 0)
|
||||
{
|
||||
_logger.LogInformation("[{Channel}] No pending news articles found in FinlyticNews.", "SentimentChannel");
|
||||
return;
|
||||
}
|
||||
|
||||
_logger.LogInformation("[{Channel}] Retrieved {Count} pending article(s) for FinBERT sentiment evaluation.",
|
||||
"SentimentChannel", pendingArticles.Count);
|
||||
|
||||
foreach (var article in pendingArticles)
|
||||
{
|
||||
if (cancellationToken.IsCancellationRequested) break;
|
||||
await ProcessSingleArticleAsync(article, cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Processes FinBERT sentiment analysis for a single article, updates two-stage JSON summaries, and notifies FinlyticNews of status update.
|
||||
/// </summary>
|
||||
private async Task ProcessSingleArticleAsync(NewsArticleDto article, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (article == null || article.Id == Guid.Empty) return;
|
||||
|
||||
// Deduplizierung: Verhindert, dass derselbe Artikel zeitgleich im Sweep & im Broadcast verarbeitet wird
|
||||
if (!ProcessingArticles.TryAdd(article.Id, 0))
|
||||
{
|
||||
_logger.LogDebug("[{Channel}] Article {Id} is already being processed. Skipping duplicate run.",
|
||||
"SentimentChannel", article.Id);
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
_logger.LogInformation("[{Channel}] Evaluating FinBERT sentiment for article: '{Title}' (ID: {Id})",
|
||||
"SentimentChannel", article.Title, article.Id);
|
||||
|
||||
var finbert = await _analyzer.AnalyzeArticleAsync(article);
|
||||
|
||||
if (finbert == null)
|
||||
{
|
||||
_logger.LogWarning(
|
||||
"[{Channel}] FinBERT analysis returned NULL for article {Id} ('{Title}'). Aborting processing for this run.",
|
||||
"SentimentChannel", article.Id, article.Title);
|
||||
return;
|
||||
}
|
||||
|
||||
if (cancellationToken.IsCancellationRequested) return;
|
||||
|
||||
// 1. Artikel-Level Sentiment speichern
|
||||
await _storage.SaveArticleSentimentAsync(article, finbert);
|
||||
|
||||
// 2. ISIN- & Sektor-Summaries aktualisieren
|
||||
if (article.MatchedAssets != null && article.MatchedAssets.Count > 0)
|
||||
{
|
||||
foreach (var asset in article.MatchedAssets)
|
||||
{
|
||||
if (cancellationToken.IsCancellationRequested) break;
|
||||
if (string.IsNullOrWhiteSpace(asset.Isin)) continue;
|
||||
|
||||
await _storage.UpdateIsinSummaryAsync(
|
||||
asset.Isin,
|
||||
asset.Name,
|
||||
"General",
|
||||
article,
|
||||
finbert);
|
||||
|
||||
await _storage.UpdateSectorSummaryAsync(
|
||||
"General",
|
||||
asset.Isin,
|
||||
article.Id.ToString(),
|
||||
finbert);
|
||||
}
|
||||
}
|
||||
|
||||
if (cancellationToken.IsCancellationRequested) return;
|
||||
|
||||
// 3. FinlyticNews über Erfolg informieren
|
||||
bool updated = await _mqttClient.UpdateArticleStatusAsync(article.Id, "Analyzed");
|
||||
if (updated)
|
||||
{
|
||||
_logger.LogInformation(
|
||||
"[{Channel}] Article sentiment processed and status set to 'Analyzed' in FinlyticNews: {Title} (ID: {Id}) -> {Label}",
|
||||
"SentimentChannel", article.Title, article.Id, finbert.Label);
|
||||
}
|
||||
else
|
||||
{
|
||||
_logger.LogWarning(
|
||||
"[{Channel}] Failed to confirm status update to 'Analyzed' in FinlyticNews for article: {Id}",
|
||||
"SentimentChannel", article.Id);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "[{Channel}] Error processing sentiment for article: {Id} ({Title})",
|
||||
"SentimentChannel", article.Id, article.Title);
|
||||
}
|
||||
finally
|
||||
{
|
||||
// Lock nach der Verarbeitung immer freigeben
|
||||
ProcessingArticles.TryRemove(article.Id, out _);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user