feat(ta): dynamic settings, IFinlyticLogger, live log streaming, and EF migration
This commit is contained in:
@@ -0,0 +1,26 @@
|
||||
using FinlyticCore.Models.Settings;
|
||||
|
||||
namespace FinlyticTechnicalAnalysis.Util;
|
||||
|
||||
public static class SettingKeys
|
||||
{
|
||||
// --- Logging-Kanäle ---
|
||||
public static readonly SettingKey<bool> TechnicalAnalysisChannel = new("Logging.Channel.TechnicalAnalysis", true);
|
||||
public static readonly SettingKey<bool> MqttChannel = new("Logging.Channel.MQTT", true);
|
||||
public static readonly SettingKey<bool> HealthPingChannel = new("Logging.Channel.Health", true);
|
||||
|
||||
// --- Indikator-Konfiguration ---
|
||||
public static readonly SettingKey<int> RsiPeriod = new("Indicators.RsiPeriod", 14);
|
||||
public static readonly SettingKey<int> MacdFastPeriod = new("Indicators.MacdFastPeriod", 12);
|
||||
public static readonly SettingKey<int> MacdSlowPeriod = new("Indicators.MacdSlowPeriod", 26);
|
||||
public static readonly SettingKey<int> MacdSignalPeriod = new("Indicators.MacdSignalPeriod", 9);
|
||||
public static readonly SettingKey<int> EmaShortPeriod = new("Indicators.EmaShortPeriod", 50);
|
||||
public static readonly SettingKey<int> EmaLongPeriod = new("Indicators.EmaLongPeriod", 200);
|
||||
public static readonly SettingKey<int> BollingerBandsPeriod = new("Indicators.BollingerBandsPeriod", 20);
|
||||
public static readonly SettingKey<double> BollingerBandsStdDev = new("Indicators.BollingerBandsStdDev", 2.0);
|
||||
public static readonly SettingKey<int> AtrPeriod = new("Indicators.AtrPeriod", 14);
|
||||
|
||||
// --- Cache & Performance ---
|
||||
public static readonly SettingKey<int> CacheDurationMinutes = new("Cache.DurationMinutes", 60);
|
||||
public static readonly SettingKey<bool> EnableAutoCache = new("Feature.EnableAutoCache", true);
|
||||
}
|
||||
@@ -1,10 +1,15 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Text.Json;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using FinlyticCore.Dtos;
|
||||
using FinlyticCore.Dtos.Settings;
|
||||
using FinlyticCore.Models;
|
||||
using FinlyticCore.Services;
|
||||
using FinlyticCore.Util;
|
||||
using FinlyticTechnicalAnalysis.Services;
|
||||
using FinlyticTechnicalAnalysis.Util;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
@@ -12,19 +17,30 @@ using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace FinlyticTechnicalAnalysis.Util;
|
||||
|
||||
public class TAMqttClient(
|
||||
ILogger<TAMqttClient> logger,
|
||||
IConfiguration configuration,
|
||||
IServiceScopeFactory scopeFactory) : ManagedMqttClient(logger), IHostedService
|
||||
public class TAMqttClient : ManagedMqttClient, IHostedService
|
||||
{
|
||||
private readonly IConfiguration _configuration;
|
||||
private readonly IServiceScopeFactory _scopeFactory;
|
||||
private readonly ILogger<TAMqttClient> _logger;
|
||||
|
||||
public TAMqttClient(
|
||||
ILogger<TAMqttClient> logger,
|
||||
IConfiguration configuration,
|
||||
IServiceScopeFactory scopeFactory) : base(logger)
|
||||
{
|
||||
_logger = logger;
|
||||
_configuration = configuration;
|
||||
_scopeFactory = scopeFactory;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Starts the MQTT client.
|
||||
/// </summary>
|
||||
public async Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
var host = configuration["MQTT:Host"] ?? configuration["MQTT__Host"] ?? "localhost";
|
||||
var portStr = configuration["MQTT:Port"] ?? configuration["MQTT__Port"] ?? "1883";
|
||||
var clientId = configuration["MQTT:ClientId"] ?? "finlytic_ta_" + Guid.NewGuid().ToString("N");
|
||||
var host = _configuration["MQTT:Host"] ?? _configuration["MQTT__Host"] ?? "localhost";
|
||||
var portStr = _configuration["MQTT:Port"] ?? _configuration["MQTT__Port"] ?? "1883";
|
||||
var clientId = _configuration["MQTT:ClientId"] ?? "finlytic_ta_" + Guid.NewGuid().ToString("N");
|
||||
|
||||
var config = new MqttConfiguration
|
||||
{
|
||||
@@ -33,7 +49,7 @@ public class TAMqttClient(
|
||||
ClientId = clientId
|
||||
};
|
||||
|
||||
logger.LogInformation("[{Channel}] Starting Technical Analysis MQTT client. Host: {Host}, ClientId: {ClientId}", "TechnicalAnalysisChannel", config.Host, config.ClientId);
|
||||
_logger.LogInformation("Starting Technical Analysis MQTT client. Host: {Host}, ClientId: {ClientId}", config.Host, config.ClientId);
|
||||
await ConnectAsync(config);
|
||||
}
|
||||
|
||||
@@ -42,17 +58,27 @@ public class TAMqttClient(
|
||||
/// </summary>
|
||||
public async Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
logger.LogInformation("[{Channel}] Stopping Technical Analysis MQTT client.", "TechnicalAnalysisChannel");
|
||||
_logger.LogInformation("Stopping Technical Analysis MQTT client.");
|
||||
await DisconnectAsync();
|
||||
}
|
||||
|
||||
protected override async Task OnConnectedAsync()
|
||||
{
|
||||
logger.LogInformation("[{Channel}] Technical Analysis MQTT client connected. Subscribing to RPC topic...", "TechnicalAnalysisChannel");
|
||||
_logger.LogInformation("Technical Analysis MQTT client connected. Subscribing to RPC topics...");
|
||||
await SubscribeAsync("services/request/ta_GetAnalysis/#");
|
||||
await SubscribeAsync("services/request/tr_GetLivePrice/#");
|
||||
await SubscribeAsync("services/request/ta_settings_GetAll/#");
|
||||
await SubscribeAsync("services/request/ta_settings_Update/#");
|
||||
await SubscribeAsync("services/request/health_Ping/#");
|
||||
await SubscribeAsync("services/config/updated/#");
|
||||
|
||||
FinlyticCore.Services.FinlyticLogBroadcaster.OnLogPublished = async (logDto) =>
|
||||
{
|
||||
if (IsConnected && string.Equals(logDto.ServiceName, "FinlyticTechnicalAnalysis", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
await PublishAsync("finlytic/logs/FinlyticTechnicalAnalysis", logDto);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
protected override async Task OnMessageReceivedAsync(string topic, string payload)
|
||||
@@ -69,21 +95,96 @@ public class TAMqttClient(
|
||||
if (segments.Length < 4) return;
|
||||
|
||||
var channel = segments[2];
|
||||
var correlationId = segments[segments.Length - 1];
|
||||
var correlationId = segments[^1];
|
||||
|
||||
if (topic.Contains("health_Ping", StringComparison.OrdinalIgnoreCase))
|
||||
switch (channel)
|
||||
{
|
||||
await HandleHealthPingAsync(topic, segments, correlationId);
|
||||
return;
|
||||
}
|
||||
case "ta_GetAnalysis":
|
||||
await HandleGetAnalysisAsync(payload, correlationId);
|
||||
break;
|
||||
|
||||
if (channel == "ta_GetAnalysis")
|
||||
{
|
||||
await HandleGetAnalysisAsync(payload, correlationId);
|
||||
case "tr_GetLivePrice":
|
||||
await HandleGetLivePriceAsync(payload, correlationId);
|
||||
break;
|
||||
|
||||
case "ta_settings_GetAll":
|
||||
await HandleSettingsGetAllAsync(correlationId);
|
||||
break;
|
||||
|
||||
case "ta_settings_Update":
|
||||
await HandleSettingsUpdateAsync(payload, correlationId);
|
||||
break;
|
||||
|
||||
case "health_Ping":
|
||||
await HandleHealthPingAsync(topic, segments, correlationId);
|
||||
break;
|
||||
|
||||
default:
|
||||
_logger.LogDebug("Received unhandled RPC channel: {Channel}", channel);
|
||||
break;
|
||||
}
|
||||
else if (channel == "tr_GetLivePrice")
|
||||
}
|
||||
|
||||
private async Task HandleSettingsGetAllAsync(string correlationId)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<TAMqttClient>>();
|
||||
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
||||
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticTechnicalAnalysis] [Settings_GetAll] Retrieving all dynamic settings via reflection [CorrelationId: {CorrelationId}]", correlationId);
|
||||
try
|
||||
{
|
||||
await HandleGetLivePriceAsync(payload, correlationId);
|
||||
var settings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
|
||||
var responseTopic = $"services/response/ta_settings_GetAll/{correlationId}";
|
||||
|
||||
await PublishAsync(responseTopic, settings);
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticTechnicalAnalysis] [Settings_GetAll] Published {Count} settings to '{ResponseTopic}'", settings.Count, responseTopic);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
await finlyticLogger.LogErrorAsync(SettingKeys.MqttChannel, ex, "[FinlyticTechnicalAnalysis] [Settings_GetAll] Failed to retrieve settings.");
|
||||
}
|
||||
}
|
||||
|
||||
private async Task HandleSettingsUpdateAsync(string payload, string correlationId)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(payload)) return;
|
||||
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<TAMqttClient>>();
|
||||
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
||||
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticTechnicalAnalysis] [Settings_Update] Processing settings update RPC [CorrelationId: {CorrelationId}]", correlationId);
|
||||
try
|
||||
{
|
||||
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, "[FinlyticTechnicalAnalysis] [Settings_Update] Successfully updated {Count} settings in database and cache.", updates.Count);
|
||||
}
|
||||
|
||||
var currentSettings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
|
||||
var responseTopic = $"services/response/ta_settings_Update/{correlationId}";
|
||||
await PublishAsync(responseTopic, currentSettings);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
await finlyticLogger.LogErrorAsync(SettingKeys.MqttChannel, ex, "[FinlyticTechnicalAnalysis] [Settings_Update] Failed to update settings.");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -92,22 +193,21 @@ public class TAMqttClient(
|
||||
if (!topic.EndsWith("FinlyticTechnicalAnalysis", StringComparison.OrdinalIgnoreCase))
|
||||
return;
|
||||
|
||||
logger.LogInformation("[{Channel}] [TAMqttClient] Received config update event for FinlyticTechnicalAnalysis.", "TechnicalAnalysisChannel");
|
||||
try
|
||||
{
|
||||
var updatePayload = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.ServiceConfigUpdatePayload);
|
||||
if (updatePayload?.Settings != null && updatePayload.Settings.Count > 0)
|
||||
using var doc = JsonDocument.Parse(payload);
|
||||
if (doc.RootElement.TryGetProperty("settings", out var settingsProp))
|
||||
{
|
||||
using var scope = scopeFactory.CreateScope();
|
||||
var settingsDb = scope.ServiceProvider.GetRequiredService<ISettingsDbService>();
|
||||
await settingsDb.UpdateSettingsFromDictionaryAsync(updatePayload.Settings);
|
||||
logger.LogInformation("[{Channel}] [TAMqttClient] Persisted {Count} updated settings to FinlyticTechnicalAnalysis database.", "TechnicalAnalysisChannel", updatePayload.Settings.Count);
|
||||
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 (Exception ex)
|
||||
{
|
||||
logger.LogError(ex, "[{Channel}] [TAMqttClient] Error processing MQTT config update event.", "TechnicalAnalysisChannel");
|
||||
}
|
||||
catch { }
|
||||
}
|
||||
|
||||
private async Task HandleHealthPingAsync(string topic, string[] segments, string correlationId)
|
||||
@@ -119,40 +219,42 @@ public class TAMqttClient(
|
||||
if (isForMe)
|
||||
{
|
||||
string respTopic = $"services/response/health_Ping/{correlationId}";
|
||||
await PublishAsync(respTopic, new FinlyticCore.Dtos.ServiceHealthResponse("FinlyticTechnicalAnalysis", "Online", DateTime.UtcNow, "Connected"));
|
||||
logger.LogInformation("[{Channel}] [TAMqttClient] Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", "TechnicalAnalysisChannel", correlationId);
|
||||
await PublishAsync(respTopic, new ServiceHealthResponse("FinlyticTechnicalAnalysis", "Online", DateTime.UtcNow, "Connected"));
|
||||
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<TAMqttClient>>();
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.HealthPingChannel, "[FinlyticTechnicalAnalysis] Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", correlationId);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task HandleGetAnalysisAsync(string payload, string correlationId)
|
||||
{
|
||||
logger.LogInformation("[{Channel}] Received RPC ta_GetAnalysis request. CorrelationId: {CorrelationId}", "TechnicalAnalysisChannel", correlationId);
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<TAMqttClient>>();
|
||||
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.TechnicalAnalysisChannel, "[FinlyticTechnicalAnalysis] Received RPC ta_GetAnalysis request. CorrelationId: {CorrelationId}", correlationId);
|
||||
|
||||
var req = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.IsinRequest);
|
||||
string responseTopic = $"services/response/ta_GetAnalysis/{correlationId}";
|
||||
|
||||
if (string.IsNullOrWhiteSpace(req?.Isin))
|
||||
{
|
||||
logger.LogWarning("[{Channel}] Request missing mandatory ISIN parameter.", "TechnicalAnalysisChannel");
|
||||
await finlyticLogger.LogWarningAsync(SettingKeys.TechnicalAnalysisChannel, "[FinlyticTechnicalAnalysis] Request missing mandatory ISIN parameter.");
|
||||
await PublishAsync<object?>(responseTopic, null);
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
using var scope = scopeFactory.CreateScope();
|
||||
var taDbService = scope.ServiceProvider.GetRequiredService<ITechnicalAnalysisDbService>();
|
||||
|
||||
var analysis = await taDbService.GetAnalysisAsync(req.Isin, req.ForceRefresh, req.Ticker);
|
||||
|
||||
logger.LogInformation("[{Channel}] Publishing RPC response to {ResponseTopic}", "TechnicalAnalysisChannel", responseTopic);
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.TechnicalAnalysisChannel, "[FinlyticTechnicalAnalysis] Publishing RPC response to {ResponseTopic}", responseTopic);
|
||||
await PublishAsync(responseTopic, analysis);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogError(ex, "[{Channel}] Failed to fetch technical analysis and publish RPC response for ISIN {Isin}", "TechnicalAnalysisChannel", req.Isin);
|
||||
|
||||
// Antworte mit null, damit der Aufrufer nicht im RPC-Timeout verharrt
|
||||
await finlyticLogger.LogErrorAsync(SettingKeys.TechnicalAnalysisChannel, ex, "[FinlyticTechnicalAnalysis] Failed to fetch technical analysis for ISIN {Isin}", req.Isin);
|
||||
try
|
||||
{
|
||||
await PublishAsync<object?>(responseTopic, null);
|
||||
@@ -163,32 +265,32 @@ public class TAMqttClient(
|
||||
|
||||
private async Task HandleGetLivePriceAsync(string payload, string correlationId)
|
||||
{
|
||||
logger.LogInformation("[{Channel}] Received RPC tr_GetLivePrice request. CorrelationId: {CorrelationId}", "TechnicalAnalysisChannel", correlationId);
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<TAMqttClient>>();
|
||||
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.TechnicalAnalysisChannel, "[FinlyticTechnicalAnalysis] Received RPC tr_GetLivePrice request. CorrelationId: {CorrelationId}", correlationId);
|
||||
|
||||
var req = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.IsinRequest);
|
||||
string responseTopic = $"services/response/tr_GetLivePrice/{correlationId}";
|
||||
|
||||
if (string.IsNullOrWhiteSpace(req?.Isin))
|
||||
{
|
||||
logger.LogWarning("[{Channel}] tr_GetLivePrice request missing mandatory ISIN parameter.", "TechnicalAnalysisChannel");
|
||||
await finlyticLogger.LogWarningAsync(SettingKeys.TechnicalAnalysisChannel, "[FinlyticTechnicalAnalysis] tr_GetLivePrice request missing mandatory ISIN parameter.");
|
||||
await PublishAsync<object?>(responseTopic, null);
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
using var scope = scopeFactory.CreateScope();
|
||||
var taDbService = scope.ServiceProvider.GetRequiredService<ITechnicalAnalysisDbService>();
|
||||
|
||||
var livePrice = await taDbService.GetLivePriceAsync(req.Isin);
|
||||
|
||||
logger.LogInformation("[{Channel}] Publishing RPC response to {ResponseTopic} for ISIN {Isin}", "TechnicalAnalysisChannel", responseTopic, req.Isin);
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.TechnicalAnalysisChannel, "[FinlyticTechnicalAnalysis] Publishing RPC response to {ResponseTopic} for ISIN {Isin}", responseTopic, req.Isin);
|
||||
await PublishAsync(responseTopic, livePrice);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogError(ex, "[{Channel}] Failed to fetch live price and publish RPC response for ISIN {Isin}", "TechnicalAnalysisChannel", req.Isin);
|
||||
|
||||
await finlyticLogger.LogErrorAsync(SettingKeys.TechnicalAnalysisChannel, ex, "[FinlyticTechnicalAnalysis] Failed to fetch live price for ISIN {Isin}", req.Isin);
|
||||
try
|
||||
{
|
||||
await PublishAsync<object?>(responseTopic, null);
|
||||
|
||||
Reference in New Issue
Block a user