226 lines
11 KiB
C#
226 lines
11 KiB
C#
using System;
|
|
using System.Collections.Concurrent;
|
|
using System.Collections.Generic;
|
|
using System.Linq;
|
|
using System.Text.Json;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using FinlyticAssets.Entities;
|
|
using FinlyticAssets.Services;
|
|
using FinlyticCore.Dtos;
|
|
using FinlyticCore.Dtos.Assets;
|
|
using FinlyticCore.Dtos.Settings;
|
|
using FinlyticCore.Dtos.TechnicalAnalysis;
|
|
using FinlyticCore.Models;
|
|
using FinlyticCore.Models.Assets;
|
|
using FinlyticCore.Services;
|
|
using FinlyticCore.Services.TradeRepublic;
|
|
using FinlyticCore.Util;
|
|
using Microsoft.Extensions.Configuration;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Hosting;
|
|
using Microsoft.Extensions.Logging;
|
|
|
|
namespace FinlyticAssets.Util;
|
|
|
|
/// <summary>
|
|
/// Managed MQTT client acting as an RPC provider for asset lookups, discovery, and on-demand derivatives.
|
|
/// </summary>
|
|
public class AssetsMqttClient : ManagedMqttClient, IHostedService
|
|
{
|
|
private readonly ILogger<AssetsMqttClient> _logger;
|
|
private readonly IServiceScopeFactory _scopeFactory;
|
|
private readonly IConfiguration _configuration;
|
|
private readonly ConcurrentDictionary<string, (LivePriceDto Price, DateTime CachedAt)> _priceCache = new(StringComparer.OrdinalIgnoreCase);
|
|
|
|
public AssetsMqttClient(
|
|
ILogger<AssetsMqttClient> logger,
|
|
IServiceScopeFactory scopeFactory,
|
|
IConfiguration configuration) : base(logger)
|
|
{
|
|
_logger = logger;
|
|
_scopeFactory = scopeFactory;
|
|
_configuration = configuration;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Starts the MQTT client and connects to the configured broker.
|
|
/// </summary>
|
|
public async Task StartAsync(CancellationToken cancellationToken)
|
|
{
|
|
var config = MqttConfiguration.FromConfiguration(_configuration, "FinlyticAssets");
|
|
|
|
_logger.LogInformation("Starting Assets MQTT client. Host: {Host}, ClientId: {ClientId}", config.Host, config.ClientId);
|
|
await ConnectAsync(config);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Gracefully stops and disconnects the MQTT client.
|
|
/// </summary>
|
|
public async Task StopAsync(CancellationToken cancellationToken)
|
|
{
|
|
_logger.LogInformation("Stopping Assets MQTT client.");
|
|
await DisconnectAsync();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Invoked automatically once the connection to the MQTT broker is successfully established or restored.
|
|
/// </summary>
|
|
protected override async Task OnConnectedAsync()
|
|
{
|
|
_logger.LogInformation("Assets MQTT Client connected. Registering topic subscriptions...");
|
|
|
|
await SubscribeAsync(MqttTopics.ResponseWildcard);
|
|
await SubscribeRpcAsync<GetValidAssetRequest, List<AssetDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.AssetsGet), HandleAssetsGetRpcAsync);
|
|
await SubscribeRpcAsync<GetDiscoveryAssetsRequest, List<AssetDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.AssetsGetDiscovery), HandleAssetsGetDiscoveryRpcAsync);
|
|
await SubscribeRpcAsync<GetDerivativesRequest, List<DerivativeDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.AssetsGetDerivatives), HandleAssetsGetDerivativesRpcAsync);
|
|
await SubscribeRpcAsync<IsinRequest, LivePriceDto?>(MqttTopics.RequestFilter(MqttTopics.Channels.TrGetLivePrice), HandleGetLivePriceRpcAsync);
|
|
await SubscribeRpcAsync<object, List<DynamicSettingDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.AssetsSettingsGetAll), HandleSettingsGetAllRpcAsync);
|
|
await SubscribeRpcAsync<Dictionary<string, object?>, List<DynamicSettingDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.AssetsSettingsUpdate), HandleSettingsUpdateRpcAsync);
|
|
await SubscribeAsync<object>(MqttTopics.RequestFilter(MqttTopics.Channels.HealthPing), HandleHealthPingRpcAsync);
|
|
|
|
FinlyticLogBroadcaster.OnLogPublished = async (logDto) =>
|
|
{
|
|
if (IsConnected && string.Equals(logDto.ServiceName, "FinlyticAssets", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
await PublishAsync(MqttTopics.Logs("FinlyticAssets"), logDto);
|
|
}
|
|
};
|
|
}
|
|
|
|
private async Task<List<AssetDto>> HandleAssetsGetRpcAsync(GetValidAssetRequest? req, string correlationId)
|
|
{
|
|
if (req == null || string.IsNullOrWhiteSpace(req.Isin)) return [];
|
|
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var dbService = scope.ServiceProvider.GetRequiredService<IAssetsDbService>();
|
|
var assets = await dbService.GetValidAssetsByIsinAsync(req.Isin.Trim().ToUpperInvariant());
|
|
return assets.ToDtoList();
|
|
}
|
|
|
|
private async Task<List<AssetDto>> HandleAssetsGetDiscoveryRpcAsync(GetDiscoveryAssetsRequest? req, string correlationId)
|
|
{
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var dbService = scope.ServiceProvider.GetRequiredService<IAssetsDbService>();
|
|
int limit = req?.Limit > 0 ? req.Limit : 15;
|
|
var discoveryAssets = await dbService.GetDiscoveryAssetsAsync(limit);
|
|
return discoveryAssets.ToDtoList();
|
|
}
|
|
|
|
private async Task<List<DerivativeDto>> HandleAssetsGetDerivativesRpcAsync(GetDerivativesRequest? req, string correlationId)
|
|
{
|
|
if (req == null || string.IsNullOrWhiteSpace(req.UnderlyingIsin)) return [];
|
|
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var dbService = scope.ServiceProvider.GetRequiredService<IAssetsDbService>();
|
|
var derivatives = await dbService.GetDerivativesByUnderlyingAsync(
|
|
req.UnderlyingIsin.Trim().ToUpperInvariant(),
|
|
req.OptionType,
|
|
req.TargetLeverage,
|
|
req.After,
|
|
req.Page,
|
|
req.ShouldForceRefresh);
|
|
return derivatives.ToDtoList();
|
|
}
|
|
|
|
private async Task<LivePriceDto?> HandleGetLivePriceRpcAsync(IsinRequest? req, string correlationId)
|
|
{
|
|
if (req == null || string.IsNullOrWhiteSpace(req.Isin)) return null;
|
|
|
|
var cleanIsin = req.Isin.Trim().ToUpperInvariant();
|
|
|
|
// Return fresh price from cache if less than 3 seconds old
|
|
if (_priceCache.TryGetValue(cleanIsin, out var cached) && (DateTime.UtcNow - cached.CachedAt).TotalSeconds < 3)
|
|
{
|
|
return cached.Price;
|
|
}
|
|
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var trService = scope.ServiceProvider.GetRequiredService<ITradeRepublicService>();
|
|
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<AssetsMqttClient>>();
|
|
|
|
var tcs = new TaskCompletionSource<LivePriceDto?>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
int? subId = null;
|
|
|
|
try
|
|
{
|
|
subId = await trService.SubscribeRealtimeTickerAsync(cleanIsin, tick =>
|
|
{
|
|
decimal currentPrice = tick.Last?.PriceValue > 0 ? tick.Last.PriceValue :
|
|
(tick.Bid?.PriceValue > 0 && tick.Ask?.PriceValue > 0 ? (tick.Bid.PriceValue + tick.Ask.PriceValue) / 2m :
|
|
(tick.Ask?.PriceValue ?? tick.Bid?.PriceValue ?? 0m));
|
|
|
|
decimal preClose = tick.Pre?.PriceValue > 0 ? tick.Pre.PriceValue : (tick.Open?.PriceValue ?? 0m);
|
|
decimal dailyChange = preClose > 0m ? ((currentPrice - preClose) / preClose) * 100m : 0m;
|
|
|
|
var livePrice = new LivePriceDto(
|
|
Isin: cleanIsin,
|
|
CurrentPrice: currentPrice,
|
|
DailyChangePercent: dailyChange,
|
|
Bid: tick.Bid?.PriceValue,
|
|
Ask: tick.Ask?.PriceValue
|
|
);
|
|
|
|
_priceCache[cleanIsin] = (livePrice, DateTime.UtcNow);
|
|
tcs.TrySetResult(livePrice);
|
|
});
|
|
|
|
var result = await tcs.Task.WaitAsync(TimeSpan.FromSeconds(3));
|
|
return result;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
await finlyticLogger.LogWarningAsync(SettingKeys.AssetsChannel, "[AssetsMqttClient] Failed or timed out fetching live price for {Isin}: {Message}", cleanIsin, ex.Message);
|
|
if (_priceCache.TryGetValue(cleanIsin, out var stale))
|
|
{
|
|
return stale.Price;
|
|
}
|
|
return null;
|
|
}
|
|
finally
|
|
{
|
|
if (subId.HasValue)
|
|
{
|
|
_ = trService.UnsubscribeRealtimeTickerAsync(subId.Value);
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task<List<DynamicSettingDto>> HandleSettingsGetAllRpcAsync(object? _, string correlationId)
|
|
{
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<AssetsMqttClient>>();
|
|
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
|
|
|
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticAssets] [Settings_GetAll] Retrieving service dynamic settings [CorrelationId: {CorrelationId}]", correlationId);
|
|
return await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
|
|
}
|
|
|
|
private async Task<List<DynamicSettingDto>> HandleSettingsUpdateRpcAsync(Dictionary<string, object?>? updates, string correlationId)
|
|
{
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<AssetsMqttClient>>();
|
|
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
|
|
|
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticAssets] [Settings_Update] Processing settings update RPC [CorrelationId: {CorrelationId}]", correlationId);
|
|
if (updates != null && updates.Count > 0)
|
|
{
|
|
await settingsService.UpdateSettingsAsync(updates);
|
|
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticAssets] [Settings_Update] Successfully updated {Count} settings in database and cache.", updates.Count);
|
|
}
|
|
return await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
|
|
}
|
|
|
|
private async Task HandleHealthPingRpcAsync(object? _, string topic, string correlationId)
|
|
{
|
|
if (topic.Contains("FinlyticAssets", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
string respTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.HealthPing, correlationId);
|
|
await PublishAsync(respTopic, new FinlyticCore.Dtos.ServiceHealthResponse("FinlyticAssets", "Online", DateTime.UtcNow, "Connected"));
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<AssetsMqttClient>>();
|
|
await finlyticLogger.LogInfoAsync(SettingKeys.HealthPingChannel, "[AssetsMqttClient] Responded to health_Ping RPC request [CorrelationId: {CorrelationId}].", correlationId);
|
|
}
|
|
}
|
|
}
|