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

feat(backend): generic settings RPC bridge, LogStreamHub SignalR, and log ringbuffer

Browse Source
This commit is contained in:
Kleidukos
2026-08-15 21:30:56 +02:00
parent 1f9d66405a
commit f18f75c1ab
5 changed files with 207 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
FinlyticBackend/Controllers/AdminSettingsController.cs
+108 -64
View File
@@ -6,6 +6,7 @@ using System.Text.Json.Serialization;
using System.Threading.Tasks;
using FinlyticBackend.Util;
using FinlyticCore.Dtos;
using FinlyticCore.Dtos.Settings;
using FinlyticCore.Util;
using Microsoft.AspNetCore.Authorization;
using Microsoft.AspNetCore.Cors;
@@ -58,39 +59,19 @@ public class AdminSettingsController : ControllerBase
private readonly WebMqttClient _mqttClient;
private readonly ILogger<AdminSettingsController> _logger;
// In-memory static store for default UI configuration templates
private static readonly Dictionary<string, string> ServiceRpcPrefixes = new(StringComparer.OrdinalIgnoreCase)
{
["FinlyticFundamentals"] = "fundamentals",
["FinlyticNews"] = "news",
["FinlyticTechnicalAnalysis"] = "ta",
["FinlyticSentiment"] = "sentiment",
["FinlyticAnalyzer"] = "analyzer",
["FinlyticTrades"] = "trades",
["FinlyticAssets"] = "assets"
};
private static readonly ConcurrentDictionary<string, List<ServiceConfigItem>> _inMemorySettings = new(StringComparer.OrdinalIgnoreCase);
static AdminSettingsController()
{
// Default-Templates für Microservice-Konfigurationen initialisieren
_inMemorySettings["FinlyticAnalyzer"] = new List<ServiceConfigItem>
{
new("MinSignalScore", "75.0", "double", "Mindest-Score für KI-Trade-Proposals (0-100)", DateTime.UtcNow),
new("VixPanicThreshold", "30.0", "double", "VIX-Wert ab dem Panik-Modus aktiviert wird", DateTime.UtcNow),
new("ProposalTtlMinutes", "180", "int", "Gültigkeitsdauer von Trade-Proposals in Minuten", DateTime.UtcNow)
};
_inMemorySettings["FinlyticNews"] = new List<ServiceConfigItem>
{
new("ScrapeIntervalMinutes", "15", "int", "Intervall für das Scraping neuer Nachrichten", DateTime.UtcNow),
new("FinBertBatchSize", "8", "int", "Batch-Größe für die Sentiment-Analyse", DateTime.UtcNow)
};
_inMemorySettings["FinlyticTechnicalAnalysis"] = new List<ServiceConfigItem>
{
new("EmaShortPeriod", "20", "int", "Kurze Periode für EMA-Berechnungen", DateTime.UtcNow),
new("EmaLongPeriod", "50", "int", "Lange Periode für EMA-Berechnungen", DateTime.UtcNow),
new("RsiPeriod", "14", "int", "Standard-Periode für RSI-Berechnung", DateTime.UtcNow)
};
_inMemorySettings["FinlyticTrades"] = new List<ServiceConfigItem>
{
new("ExportFeedbackIntervalHours", "6", "int", "Intervall für den Parquet/JSON Feedback-Export", DateTime.UtcNow),
new("DefaultLeverageLimit", "10", "decimal", "Standardmäßiger Maximalhebel für Derivate", DateTime.UtcNow)
};
}
public AdminSettingsController(
WebMqttClient mqttClient,
ILogger<AdminSettingsController> logger)
@@ -100,11 +81,57 @@ public class AdminSettingsController : ControllerBase
}
/// <summary>
/// Retrieves all service configurations grouped by service name.
/// Retrieves recent buffered in-memory logs for a specific service.
/// </summary>
[HttpGet("logs/{serviceName}")]
public IActionResult GetServiceLogs(string serviceName)
{
if (BackendMqttBridge.ServiceLogsRingBuffer.TryGetValue(serviceName, out var queue))
{
return Ok(queue.ToList());
}
return Ok(new List<FinlyticCore.Dtos.Logging.LogMessageDto>());
}
/// <summary>
/// Retrieves all service configurations grouped by service name via live MQTT RPC queries.
/// </summary>
[HttpGet]
public IActionResult GetAllSettings()
public async Task<IActionResult> GetAllSettings()
{
if (_mqttClient.IsConnected)
{
var fetchTasks = ServiceRpcPrefixes.Select(async kvp =>
{
var serviceName = kvp.Key;
var prefix = kvp.Value;
try
{
var liveSettings = await _mqttClient.SendRpcRequestAsync<List<DynamicSettingDto>, string>(
$"{prefix}_settings_GetAll",
"",
TimeSpan.FromSeconds(2));
if (liveSettings != null && liveSettings.Count > 0)
{
_inMemorySettings[serviceName] = liveSettings.Select(d => new ServiceConfigItem(
Key: d.Key,
Value: d.Value?.ToString() ?? "",
DataType: d.Type,
Description: d.Description,
UpdatedAt: d.UpdatedAt ?? DateTime.UtcNow
)).ToList();
}
}
catch (Exception ex)
{
_logger.LogWarning(ex, "[AdminSettings] Could not fetch live settings from {ServiceName} via MQTT.", serviceName);
}
});
await Task.WhenAll(fetchTasks);
}
var grouped = _inMemorySettings.ToDictionary(
g => g.Key,
g => g.Value.Select(item => new ServiceConfigItemResponseDto(
@@ -198,11 +225,37 @@ public class AdminSettingsController : ControllerBase
}
/// <summary>
/// Retrieves settings for a specific service.
/// Retrieves settings for a specific service via live MQTT RPC.
/// </summary>
[HttpGet("{serviceName}")]
public IActionResult GetServiceSettings(string serviceName)
public async Task<IActionResult> GetServiceSettings(string serviceName)
{
if (ServiceRpcPrefixes.TryGetValue(serviceName, out var prefix) && _mqttClient.IsConnected)
{
try
{
var liveSettings = await _mqttClient.SendRpcRequestAsync<List<DynamicSettingDto>, string>(
$"{prefix}_settings_GetAll",
"",
TimeSpan.FromSeconds(2));
if (liveSettings != null && liveSettings.Count > 0)
{
_inMemorySettings[serviceName] = liveSettings.Select(d => new ServiceConfigItem(
Key: d.Key,
Value: d.Value?.ToString() ?? "",
DataType: d.Type,
Description: d.Description,
UpdatedAt: d.UpdatedAt ?? DateTime.UtcNow
)).ToList();
}
}
catch (Exception ex)
{
_logger.LogWarning(ex, "[AdminSettings] Could not fetch live settings for {ServiceName} via MQTT.", serviceName);
}
}
if (_inMemorySettings.TryGetValue(serviceName, out var list))
{
var dtos = list.Select(item => new ServiceConfigItemResponseDto(
@@ -220,11 +273,10 @@ public class AdminSettingsController : ControllerBase
}
/// <summary>
/// Broadcasts configuration settings to the targeted microservice via MQTT.
/// Does NOT write to Backend database. The microservice persists updated settings directly into its own database.
/// Broadcasts configuration settings to the targeted microservice via MQTT RPC and updates in-memory cache.
/// </summary>
[HttpPut("{serviceName}")]
public async Task<IActionResult> UpdateServiceSettings(string serviceName, [FromBody] Dictionary<string, string> updatedValues)
public async Task<IActionResult> UpdateServiceSettings(string serviceName, [FromBody] Dictionary<string, object?> updatedValues)
{
if (updatedValues == null || !updatedValues.Any())
{
@@ -233,40 +285,32 @@ public class AdminSettingsController : ControllerBase
_logger.LogInformation("[AdminSettings] Transmitting {Count} config settings to microservice '{ServiceName}' via MQTT", updatedValues.Count, serviceName);
// In-Memory Template-Store aktualisieren
if (_inMemorySettings.TryGetValue(serviceName, out var existingList))
{
foreach (var (key, value) in updatedValues)
{
var idx = existingList.FindIndex(item => item.Key.Equals(key, StringComparison.OrdinalIgnoreCase));
if (idx >= 0)
{
var old = existingList[idx];
existingList[idx] = old with { Value = value, UpdatedAt = DateTime.UtcNow };
}
else
{
existingList.Add(new ServiceConfigItem(key, value, "string", $"Setting for {serviceName}", DateTime.UtcNow));
}
}
}
else
{
var newList = updatedValues.Select(kv => new ServiceConfigItem(kv.Key, kv.Value, "string", $"Setting for {serviceName}", DateTime.UtcNow)).ToList();
_inMemorySettings[serviceName] = newList;
}
// MQTT Config-Update Event senden
bool mqttPublished = false;
try
{
if (_mqttClient.IsConnected)
if (_mqttClient.IsConnected && ServiceRpcPrefixes.TryGetValue(serviceName, out var prefix))
{
var updated = await _mqttClient.SendRpcRequestAsync<List<DynamicSettingDto>, Dictionary<string, object?>>(
$"{prefix}_settings_Update",
updatedValues,
TimeSpan.FromSeconds(3));
if (updated != null && updated.Count > 0)
{
_inMemorySettings[serviceName] = updated.Select(d => new ServiceConfigItem(
Key: d.Key,
Value: d.Value?.ToString() ?? "",
DataType: d.Type,
Description: d.Description,
UpdatedAt: d.UpdatedAt ?? DateTime.UtcNow
)).ToList();
}
string topic = $"services/config/updated/{serviceName}";
var payload = new ServiceConfigUpdatePayload(
ServiceName: serviceName,
Timestamp: DateTime.UtcNow,
Settings: updatedValues
Settings: updatedValues.ToDictionary(kv => kv.Key, kv => kv.Value?.ToString() ?? "")
);
await _mqttClient.PublishAsync(topic, payload);
FinlyticBackend/Hubs/LogStreamHub.cs
+49
View File
@@ -0,0 +1,49 @@
using System;
using System.Threading.Tasks;
using Microsoft.AspNetCore.SignalR;
using Microsoft.Extensions.Logging;
namespace FinlyticBackend.Hubs;
/// <summary>
/// SignalR Hub that streams live log messages from microservices to connected web UI clients.
/// </summary>
public class LogStreamHub : Hub
{
private readonly ILogger<LogStreamHub> _logger;
public LogStreamHub(ILogger<LogStreamHub> logger)
{
_logger = logger;
}
public async Task JoinServiceLogs(string serviceName)
{
if (!string.IsNullOrWhiteSpace(serviceName))
{
await Groups.AddToGroupAsync(Context.ConnectionId, serviceName);
_logger.LogInformation("[LogStreamHub] Client {ConnectionId} joined log stream for {ServiceName}", Context.ConnectionId, serviceName);
}
}
public async Task LeaveServiceLogs(string serviceName)
{
if (!string.IsNullOrWhiteSpace(serviceName))
{
await Groups.RemoveFromGroupAsync(Context.ConnectionId, serviceName);
_logger.LogInformation("[LogStreamHub] Client {ConnectionId} left log stream for {ServiceName}", Context.ConnectionId, serviceName);
}
}
public override async Task OnConnectedAsync()
{
_logger.LogInformation("[LogStreamHub] Client connected: {ConnectionId}", Context.ConnectionId);
await base.OnConnectedAsync();
}
public override async Task OnDisconnectedAsync(Exception? exception)
{
_logger.LogInformation("[LogStreamHub] Client disconnected: {ConnectionId}", Context.ConnectionId);
await base.OnDisconnectedAsync(exception);
}
}
FinlyticBackend/Program.cs
+4
View File
@@ -191,6 +191,10 @@ app.MapHub<FavoritesPriceHub>("/hubs/favorites-prices", options =>
{
options.Transports = HttpTransportType.WebSockets | HttpTransportType.ServerSentEvents;
});
app.MapHub<LogStreamHub>("/hubs/logs", options =>
{
options.Transports = HttpTransportType.WebSockets | HttpTransportType.ServerSentEvents;
});
app.MapGet("/health", () => Results.Ok(new { status = "Healthy", service = "FinlyticBackend", timestamp = DateTime.UtcNow }));
FinlyticBackend/Util/BackendMqttBridge.cs
+29 -1
View File
@@ -7,6 +7,7 @@ using System.Threading.Tasks;
using FinlyticBackend.Database;
using FinlyticBackend.Hubs;
using FinlyticBackend.Services;
using FinlyticCore.Dtos.Logging;
using FinlyticCore.Dtos.News;
using FinlyticCore.Models;
using FinlyticCore.Models.Auth;
@@ -28,12 +29,14 @@ public class BackendMqttBridge : ManagedMqttClient, IHostedService
{
public static readonly ConcurrentDictionary<string, JsonElement> FundamentalsCache = new(StringComparer.OrdinalIgnoreCase);
public static readonly ConcurrentDictionary<string, JsonElement> TechnicalsCache = new(StringComparer.OrdinalIgnoreCase);
public static readonly ConcurrentDictionary<string, ConcurrentQueue<LogMessageDto>> ServiceLogsRingBuffer = new(StringComparer.OrdinalIgnoreCase);
private readonly IConfiguration _configuration;
private readonly IServiceScopeFactory _scopeFactory;
private readonly IHubContext<TradeRealtimeHub, ITradeClient> _hubContext;
private readonly IHubContext<TradeHub> _tradeHubContext;
private readonly IHubContext<NewsHub> _newsHubContext;
private readonly IHubContext<LogStreamHub> _logHubContext;
private readonly IFirebaseNotificationService _firebaseService;
private readonly ILogger<BackendMqttBridge> _logger;
@@ -43,6 +46,7 @@ public class BackendMqttBridge : ManagedMqttClient, IHostedService
IHubContext<TradeRealtimeHub, ITradeClient> hubContext,
IHubContext<TradeHub> tradeHubContext,
IHubContext<NewsHub> newsHubContext,
IHubContext<LogStreamHub> logHubContext,
IFirebaseNotificationService firebaseService,
ILogger<BackendMqttBridge> logger) : base(logger)
{
@@ -51,6 +55,7 @@ public class BackendMqttBridge : ManagedMqttClient, IHostedService
_hubContext = hubContext;
_tradeHubContext = tradeHubContext;
_newsHubContext = newsHubContext;
_logHubContext = logHubContext;
_firebaseService = firebaseService;
_logger = logger;
}
@@ -95,6 +100,8 @@ public class BackendMqttBridge : ManagedMqttClient, IHostedService
await SubscribeAsync("finlytic/technicalanalysis/#");
await SubscribeAsync("finlytic/ta/#");
// Real-time Logs
await SubscribeAsync("finlytic/logs/#");
}
/// <inheritdoc />
@@ -104,7 +111,11 @@ public class BackendMqttBridge : ManagedMqttClient, IHostedService
try
{
if (topic.StartsWith("finlytic/trades/proposed/", StringComparison.OrdinalIgnoreCase) ||
if (topic.StartsWith("finlytic/logs/", StringComparison.OrdinalIgnoreCase))
{
await HandleLogMessageAsync(payloadStr);
}
else if (topic.StartsWith("finlytic/trades/proposed/", StringComparison.OrdinalIgnoreCase) ||
topic.StartsWith("finlytic/trades/update", StringComparison.OrdinalIgnoreCase))
{
await HandleTradeProposalAsync(payloadStr);
@@ -136,6 +147,23 @@ public class BackendMqttBridge : ManagedMqttClient, IHostedService
}
}
private async Task HandleLogMessageAsync(string payloadStr)
{
var logDto = JsonSerializer.Deserialize<LogMessageDto>(payloadStr, FinlyticJsonSerializerContext.Default.LogMessageDto);
if (logDto == null) return;
string serviceKey = logDto.ServiceName;
var queue = ServiceLogsRingBuffer.GetOrAdd(serviceKey, _ => new ConcurrentQueue<LogMessageDto>());
queue.Enqueue(logDto);
// Keep buffer capped at 250 entries
while (queue.Count > 250 && queue.TryDequeue(out _)) { }
// Broadcast to SignalR clients
await _logHubContext.Clients.Group(serviceKey).SendAsync("ReceiveLogMessage", logDto);
await _logHubContext.Clients.All.SendAsync("ReceiveLogMessage", logDto);
}
private async Task HandleTradeProposalAsync(string payloadStr)
{
var proposal = JsonSerializer.Deserialize<TradeProposalDto>(payloadStr);
FinlyticBackend/Util/WebMqttClient.cs
+18 -1
View File
@@ -61,10 +61,27 @@ public class WebMqttClient : ManagedMqttClient, IHostedService
await SubscribeAsync("services/response/assets_GetDiscovery/#");
await SubscribeAsync("services/response/assets_GetDerivatives/#");
await SubscribeAsync("services/response/trades_Get/#");
await SubscribeAsync("services/response/trades_Close/#");
await SubscribeAsync("services/response/trades_Reject/#");
await SubscribeAsync("services/response/trades_Accept/#");
await SubscribeAsync("services/response/analyzer_TriggerManual/#");
await SubscribeAsync("services/response/health_Ping/#");
// Settings RPC response channels for all microservices
await SubscribeAsync("services/response/fundamentals_settings_GetAll/#");
await SubscribeAsync("services/response/fundamentals_settings_Update/#");
await SubscribeAsync("services/response/news_settings_GetAll/#");
await SubscribeAsync("services/response/news_settings_Update/#");
await SubscribeAsync("services/response/ta_settings_GetAll/#");
await SubscribeAsync("services/response/ta_settings_Update/#");
await SubscribeAsync("services/response/sentiment_settings_GetAll/#");
await SubscribeAsync("services/response/sentiment_settings_Update/#");
await SubscribeAsync("services/response/analyzer_settings_GetAll/#");
await SubscribeAsync("services/response/analyzer_settings_Update/#");
await SubscribeAsync("services/response/trades_settings_GetAll/#");
await SubscribeAsync("services/response/trades_settings_Update/#");
await SubscribeAsync("services/response/assets_settings_GetAll/#");
await SubscribeAsync("services/response/assets_settings_Update/#");
}
protected override Task OnMessageReceivedAsync(string topic, string payload)