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

refactor(assets): update asset entity mappings, database migrations, and MQTT RPC handlers

Browse Source
This commit is contained in:
Kleidukos
2026-08-24 21:37:21 +02:00
parent 7060f0f7b1
commit 6974b2075b
26 changed files with 645 additions and 2237 deletions
Download Patch File Download Diff File Expand all files Collapse all files
FinlyticAssets/Util/AssetMapper.cs
+55 -57
View File
@@ -1,3 +1,6 @@
using System;
using System.Collections.Generic;
using System.Linq;
using FinlyticAssets.Entities;
using FinlyticCore.Dtos.Assets;
@@ -14,6 +17,8 @@ public static class AssetMapper
Type = t.Type
}).ToList();
var dynamicLogoUrl = $"/api/v1/logo/{entity.Isin}";
return entity switch
{
StockEntity stock => new StockDto
@@ -23,7 +28,7 @@ public static class AssetMapper
Type = stock.Type,
InstrumentCategory = stock.InstrumentCategory,
HasCfd = stock.HasCfd,
ImageId = stock.ImageId,
ImageId = dynamicLogoUrl,
LastUpdatedAt = stock.LastUpdatedAt,
Tags = dtoTags,
DerivativeProductCategories = stock.DerivativeProductCategories
@@ -35,7 +40,7 @@ public static class AssetMapper
Type = etf.Type,
InstrumentCategory = etf.InstrumentCategory,
HasCfd = etf.HasCfd,
ImageId = etf.ImageId,
ImageId = dynamicLogoUrl,
LastUpdatedAt = etf.LastUpdatedAt,
Tags = dtoTags,
DerivativeProductCategories = etf.DerivativeProductCategories,
@@ -44,59 +49,6 @@ public static class AssetMapper
Subtitle = etf.Subtitle,
SearchSubtitle = etf.SearchSubtitle
},
CryptoEntity crypto => new CryptoDto
{
Isin = crypto.Isin,
Name = crypto.Name,
Type = crypto.Type,
InstrumentCategory = crypto.InstrumentCategory,
HasCfd = crypto.HasCfd,
ImageId = crypto.ImageId,
LastUpdatedAt = crypto.LastUpdatedAt,
Tags = dtoTags,
Subtitle = crypto.Subtitle,
SearchSubtitle = crypto.SearchSubtitle
},
BondEntity bond => new BondDto
{
Isin = bond.Isin,
Name = bond.Name,
Type = bond.Type,
InstrumentCategory = bond.InstrumentCategory,
HasCfd = bond.HasCfd,
ImageId = bond.ImageId,
LastUpdatedAt = bond.LastUpdatedAt,
Tags = dtoTags,
BondIssuerName = bond.BondIssuerName,
SearchSubtitle = bond.SearchSubtitle
},
DerivativeEntity deriv => new DerivativeDto
{
Isin = deriv.Isin,
Name = deriv.Name,
Type = deriv.Type,
InstrumentCategory = deriv.InstrumentCategory,
HasCfd = deriv.HasCfd,
ImageId = deriv.ImageId,
LastUpdatedAt = deriv.LastUpdatedAt,
Tags = dtoTags,
DerivativeProductCategories = deriv.DerivativeProductCategories,
UnderlyingIsin = deriv.UnderlyingIsin,
OptionType = deriv.OptionType.ToString(),
ProductCategoryName = deriv.ProductCategoryName,
NextGenProductCategoryName = deriv.NextGenProductCategoryName,
Strike = deriv.Strike,
Barrier = deriv.Barrier,
Leverage = deriv.Leverage,
Size = deriv.Size,
Factor = deriv.Factor,
Delta = deriv.Delta,
Currency = deriv.Currency,
Expiry = deriv.Expiry,
Issuer = deriv.Issuer,
IssuerDisplayName = deriv.IssuerDisplayName,
IssuerImageId = deriv.IssuerImageId
},
SyntheticEntity synth => new SyntheticDto
{
Isin = synth.Isin,
@@ -104,12 +56,53 @@ public static class AssetMapper
Type = synth.Type,
InstrumentCategory = synth.InstrumentCategory,
HasCfd = synth.HasCfd,
ImageId = synth.ImageId,
ImageId = dynamicLogoUrl,
LastUpdatedAt = synth.LastUpdatedAt,
Tags = dtoTags,
DerivativeProductCategories = synth.DerivativeProductCategories
},
_ => throw new NotSupportedException($"Mapping for type {entity.GetType().Name} is not supported.")
_ => new StockDto
{
Isin = entity.Isin,
Name = entity.Name,
Type = entity.Type,
InstrumentCategory = entity.InstrumentCategory,
HasCfd = entity.HasCfd,
ImageId = dynamicLogoUrl,
LastUpdatedAt = entity.LastUpdatedAt,
Tags = dtoTags
}
};
}
public static DerivativeDto ToDto(this DerivativeEntity deriv)
{
return new DerivativeDto
{
Isin = deriv.Isin,
Name = deriv.Name,
Type = "derivative",
InstrumentCategory = "derivative",
HasCfd = false,
ImageId = $"/api/v1/logo/{deriv.UnderlyingIsin}",
LastUpdatedAt = deriv.LastUpdatedAt,
Tags = new List<TagDto>(),
DerivativeProductCategories = deriv.DerivativeProductCategories,
UnderlyingIsin = deriv.UnderlyingIsin,
OptionType = deriv.OptionType.ToString(),
ProductCategoryName = deriv.ProductCategoryName,
NextGenProductCategoryName = deriv.NextGenProductCategoryName,
Strike = deriv.Strike,
Barrier = deriv.Barrier,
Leverage = deriv.Leverage,
Size = deriv.Size,
Factor = deriv.Factor,
Delta = deriv.Delta,
Currency = deriv.Currency,
Expiry = deriv.Expiry,
Issuer = deriv.Issuer,
IssuerDisplayName = deriv.IssuerDisplayName,
IssuerImageId = null
};
}
@@ -117,4 +110,9 @@ public static class AssetMapper
{
return entities.Select(e => e.ToDto()).ToList();
}
public static List<DerivativeDto> ToDtoList(this IEnumerable<DerivativeEntity> derivatives)
{
return derivatives.Select(d => d.ToDto()).ToList();
}
}
FinlyticAssets/Util/AssetsMqttClient.cs
+113 -231
View File
@@ -1,4 +1,5 @@
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text.Json;
@@ -6,9 +7,14 @@ 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;
@@ -18,13 +24,14 @@ using Microsoft.Extensions.Logging;
namespace FinlyticAssets.Util;
/// <summary>
/// Represents a managed MQTT client acting as a server-side RPC provider within the asset microservice.
/// 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,
@@ -41,12 +48,7 @@ public class AssetsMqttClient : 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"] ?? "FinlyticAssets")}_{Guid.NewGuid()}"
};
var config = MqttConfiguration.FromConfiguration(_configuration, "FinlyticAssets");
_logger.LogInformation("Starting Assets MQTT client. Host: {Host}, ClientId: {ClientId}", config.Host, config.ClientId);
await ConnectAsync(config);
@@ -66,278 +68,158 @@ public class AssetsMqttClient : ManagedMqttClient, IHostedService
/// </summary>
protected override async Task OnConnectedAsync()
{
_logger.LogInformation("Assets MQTT Client connected. Subscribing to topics...");
await SubscribeAsync("services/request/assets_Get/#");
await SubscribeAsync("services/request/assets_Search/#");
await SubscribeAsync("services/request/assets_GetDiscovery/#");
await SubscribeAsync("services/request/assets_GetDerivatives/#");
await SubscribeAsync("services/request/assets_FetchLogo/#");
await SubscribeAsync("services/request/assets_settings_GetAll/#");
await SubscribeAsync("services/request/assets_settings_Update/#");
await SubscribeAsync("services/request/health_Ping/#");
await SubscribeAsync("services/config/updated/#");
_logger.LogInformation("Assets MQTT Client connected. Registering topic subscriptions...");
FinlyticCore.Services.FinlyticLogBroadcaster.OnLogPublished = async (logDto) =>
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("finlytic/logs/FinlyticAssets", logDto);
await PublishAsync(MqttTopics.Logs("FinlyticAssets"), logDto);
}
};
}
/// <summary>
/// Processes incoming messages on the subscribed topics.
/// </summary>
protected override async Task OnMessageReceivedAsync(string topic, string payload)
private async Task<List<AssetDto>> HandleAssetsGetRpcAsync(GetValidAssetRequest? req, string correlationId)
{
if (topic.StartsWith("services/config/updated", StringComparison.OrdinalIgnoreCase))
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)
{
await HandleConfigUpdatedAsync(topic, payload);
return;
return cached.Price;
}
var segments = topic.Split('/');
if (segments.Length < 4) return;
using var scope = _scopeFactory.CreateScope();
var trService = scope.ServiceProvider.GetRequiredService<ITradeRepublicService>();
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<AssetsMqttClient>>();
var channel = segments[2];
var correlationId = segments[segments.Length - 1];
if (topic.Contains("health_Ping", StringComparison.OrdinalIgnoreCase))
{
await HandleHealthPingAsync(topic, segments, correlationId);
return;
}
if (topic.StartsWith("services/request/assets_settings_GetAll", StringComparison.OrdinalIgnoreCase))
{
await HandleSettingsGetAllAsync(correlationId);
return;
}
if (topic.StartsWith("services/request/assets_settings_Update", StringComparison.OrdinalIgnoreCase))
{
await HandleSettingsUpdateAsync(payload, correlationId);
return;
}
var tcs = new TaskCompletionSource<LivePriceDto?>(TaskCreationOptions.RunContinuationsAsynchronously);
int? subId = null;
try
{
using var scope = _scopeFactory.CreateScope();
var dbService = scope.ServiceProvider.GetRequiredService<IAssetsDbService>();
var indexService = scope.ServiceProvider.GetRequiredService<IAssetsIndexService>();
if (channel == "assets_FetchLogo")
subId = await trService.SubscribeRealtimeTickerAsync(cleanIsin, tick =>
{
await HandleFetchLogoAsync(payload, correlationId, indexService);
return;
}
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));
List<AssetEntity> responseData = [];
decimal preClose = tick.Pre?.PriceValue > 0 ? tick.Pre.PriceValue : (tick.Open?.PriceValue ?? 0m);
decimal dailyChange = preClose > 0m ? ((currentPrice - preClose) / preClose) * 100m : 0m;
switch (channel)
{
case "assets_Get":
responseData = await HandleAssetsGetAsync(payload, dbService);
break;
case "assets_Search":
responseData = await HandleAssetsSearchAsync(payload, dbService);
break;
case "assets_GetDiscovery":
responseData = await HandleAssetsGetDiscoveryAsync(payload, dbService);
break;
case "assets_GetDerivatives":
responseData = (await HandleAssetsGetDerivativesAsync(payload, dbService)).Cast<AssetEntity>().ToList();
break;
}
var livePrice = new LivePriceDto(
Isin: cleanIsin,
CurrentPrice: currentPrice,
DailyChangePercent: dailyChange,
Bid: tick.Bid?.PriceValue,
Ask: tick.Ask?.PriceValue
);
string defaultResponseTopic = $"services/response/{channel}/{correlationId}";
await PublishAsync(defaultResponseTopic, responseData.ToDtoList());
_priceCache[cleanIsin] = (livePrice, DateTime.UtcNow);
tcs.TrySetResult(livePrice);
});
var result = await tcs.Task.WaitAsync(TimeSpan.FromSeconds(3));
return result;
}
catch (Exception ex)
{
OnError(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 HandleSettingsGetAllAsync(string correlationId)
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 all dynamic settings via reflection [CorrelationId: {CorrelationId}]", correlationId);
try
{
var settings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
var responseTopic = $"services/response/assets_settings_GetAll/{correlationId}";
await PublishAsync(responseTopic, settings);
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticAssets] [Settings_GetAll] Published {Count} settings to '{ResponseTopic}'", settings.Count, responseTopic);
}
catch (Exception ex)
{
await finlyticLogger.LogErrorAsync(SettingKeys.MqttChannel, ex, "[FinlyticAssets] [Settings_GetAll] Failed to retrieve settings.");
}
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 HandleSettingsUpdateAsync(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<AssetsMqttClient>>();
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticAssets] [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, "[FinlyticAssets] [Settings_Update] Successfully updated {Count} settings in database and cache.", updates.Count);
}
var currentSettings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
var responseTopic = $"services/response/assets_settings_Update/{correlationId}";
await PublishAsync(responseTopic, currentSettings);
}
catch (Exception ex)
{
await finlyticLogger.LogErrorAsync(SettingKeys.MqttChannel, ex, "[FinlyticAssets] [Settings_Update] Failed to update settings.");
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 HandleConfigUpdatedAsync(string topic, string payload)
private async Task HandleHealthPingRpcAsync(object? _, string topic, string correlationId)
{
if (!topic.EndsWith("FinlyticAssets", StringComparison.OrdinalIgnoreCase))
return;
try
if (topic.Contains("FinlyticAssets", StringComparison.OrdinalIgnoreCase))
{
var updatePayload = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.ServiceConfigUpdatePayload);
if (updatePayload?.Settings != null && updatePayload.Settings.Count > 0)
{
using var scope = _scopeFactory.CreateScope();
var settings = scope.ServiceProvider.GetRequiredService<ISettingsService>();
var dict = updatePayload.Settings.ToDictionary(k => k.Key, v => (object?)v.Value);
await settings.UpdateSettingsAsync(dict);
}
}
catch (Exception ex)
{
using var scope = _scopeFactory.CreateScope();
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<AssetsMqttClient>>();
await finlyticLogger.LogErrorAsync(SettingKeys.AssetsChannel, ex, "[AssetsMqttClient] Error processing MQTT config update event.");
}
}
private async Task HandleHealthPingAsync(string topic, string[] segments, string correlationId)
{
bool isForMe = segments.Length >= 5
? segments[3].Equals("FinlyticAssets", StringComparison.OrdinalIgnoreCase)
: topic.Contains("FinlyticAssets", StringComparison.OrdinalIgnoreCase);
if (isForMe)
{
string respTopic = $"services/response/health_Ping/{correlationId}";
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 live health_Ping RPC request [CorrelationId: {CorrelationId}].", correlationId);
await finlyticLogger.LogInfoAsync(SettingKeys.HealthPingChannel, "[AssetsMqttClient] Responded to health_Ping RPC request [CorrelationId: {CorrelationId}].", correlationId);
}
}
private async Task HandleFetchLogoAsync(string payload, string correlationId, IAssetsIndexService indexService)
{
var req = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.IsinRequest);
string? isin = req?.Isin;
string? savedPath = null;
if (!string.IsNullOrEmpty(isin))
{
savedPath = await indexService.DownloadAndSaveLogoAsync(isin);
}
string responseTopic = $"services/response/assets_FetchLogo/{correlationId}";
await PublishAsync(responseTopic, new FinlyticCore.Dtos.FetchLogoResponse(isin, savedPath, savedPath != null));
}
private async Task<List<AssetEntity>> HandleAssetsGetAsync(string payload, IAssetsDbService dbService)
{
var validReq = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.GetValidAssetRequest);
if (validReq != null)
{
return await dbService.GetValidAssetsByIsinAsync(validReq.Isin);
}
return [];
}
private async Task<List<AssetEntity>> HandleAssetsSearchAsync(string payload, IAssetsDbService dbService)
{
var searchReq = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.SearchAssetsRequest);
if (searchReq != null)
{
return await dbService.FindAffectedActiveAssetsAsync(searchReq.SearchQuery);
}
return [];
}
private async Task<List<AssetEntity>> HandleAssetsGetDiscoveryAsync(string payload, IAssetsDbService dbService)
{
int limit = 15;
if (!string.IsNullOrWhiteSpace(payload))
{
try
{
var discReq = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.GetDiscoveryAssetsRequest);
if (discReq != null && discReq.Limit > 0) limit = discReq.Limit;
}
catch { }
}
return await dbService.GetDiscoveryAssetsAsync(limit);
}
private async Task<List<DerivativeEntity>> HandleAssetsGetDerivativesAsync(string payload, IAssetsDbService dbService)
{
if (string.IsNullOrWhiteSpace(payload)) return [];
try
{
var req = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.GetDerivativesRequest);
if (req != null && !string.IsNullOrEmpty(req.UnderlyingIsin))
{
return await dbService.GetDerivativesByUnderlyingAsync(
req.UnderlyingIsin,
req.OptionType,
req.TargetLeverage,
req.After,
req.Page,
req.ShouldForceRefresh);
}
}
catch (Exception ex)
{
using var scope = _scopeFactory.CreateScope();
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<AssetsMqttClient>>();
await finlyticLogger.LogErrorAsync(SettingKeys.AssetsChannel, ex, "[AssetsMqttClient] Error parsing GetDerivativesRequest payload.");
}
return [];
}
}
FinlyticAssets/Util/SettingKeys.cs
+15 -9
View File
@@ -8,15 +8,21 @@ public static class SettingKeys
public static readonly SettingKey<bool> AssetsChannel = new("Logging.Channel.Assets", true);
public static readonly SettingKey<bool> MqttChannel = new("Logging.Channel.MQTT", true);
public static readonly SettingKey<bool> HealthPingChannel = new("Logging.Channel.Health", true);
public static readonly SettingKey<bool> TradeRepublicChannel = new("Logging.Channel.TradeRepublic", true);
// --- Asset Scanning ---
public static readonly SettingKey<bool> EnableAutoScan = new("Scanner.EnableAutoScan", true);
public static readonly SettingKey<int> ScanIntervalHours = new("Scanner.ScanIntervalHours", 12);
public static readonly SettingKey<int> MaxConcurrentScans = new("Scanner.MaxConcurrentScans", 5);
public static readonly SettingKey<bool> EnableDerivativeScanning = new("Scanner.EnableDerivativeScanning", true);
// --- Trade Republic Connection ---
public static readonly SettingKey<int> TradeRepublicWsReconnectInterval = new("TradeRepublic.WsReconnectIntervalSeconds", 5);
public static readonly SettingKey<int> TradeRepublicWsTimeout = new("TradeRepublic.WsTimeoutSeconds", 15);
// --- Logos & Media ---
public static readonly SettingKey<bool> AutoFetchLogos = new("Media.AutoFetchLogos", true);
public static readonly SettingKey<int> LogoFetchBatchSize = new("Media.LogoFetchBatchSize", 25);
public static readonly SettingKey<string> LogoStorageDirectory = new("Media.LogoStorageDirectory", "data/logos");
// --- Asset Scanning Config ---
public static readonly SettingKey<bool> ScannerEnableAutoScan = new("Scanner.EnableAutoScan", true);
public static readonly SettingKey<string> ScannerCurrentScanningType = new("Scanner.CurrentScanningType", "Stock");
public static readonly SettingKey<int> ScannerCurrentScanningPage = new("Scanner.CurrentScanningPage", 0);
public static readonly SettingKey<bool> ScannerFinishedInitialScan = new("Scanner.FinishedInitialScan", false);
public static readonly SettingKey<int> ScannerBatchDelay = new("Scanner.BatchAssetUpdateDelay", 0);
public static readonly SettingKey<int> ScannerTypeDelay = new("Scanner.AssetUpdateTypeDelay", 0);
public static readonly SettingKey<int> ScannerInitBatchDelay = new("Scanner.InitBatchAssetUpdateDelay", 0);
public static readonly SettingKey<int> ScannerInitTypeDelay = new("Scanner.InitAssetUpdateTypeDelay", 0);
public static readonly SettingKey<int> ScannerMaxPageSize = new("Scanner.TradeRepublicMaxRequestPageSize", 50);
public static readonly SettingKey<int> ScannerCycleDelayMinutes = new("Scanner.CycleDelayMinutes", 1440);
}
FinlyticAssets/Util/StringCodeGenerator.cs
-17
View File
@@ -1,17 +0,0 @@
namespace FinlyticAssets.Util;
public class StringCodeGenerator
{
/// <summary>
/// Generates a W3C traceparent string for telemetry tracking.
/// </summary>
public static string GenerateTraceparent()
{
var traceId = Guid.NewGuid().ToString("N");
var spanId = Guid.NewGuid().ToString("N").Substring(0, 16);
return $"00-{traceId}-{spanId}-01";
}
}