From 600ccf299ee168b0e05706b48ff248aa8720bf3b Mon Sep 17 00:00:00 2001 From: Kleidukos Date: Mon, 24 Aug 2026 21:35:43 +0200 Subject: [PATCH] feat(fundamentals): add KeyedLockPool for concurrent scraping synchronization and update MQTT RPC handlers --- FinlyticFundamentals/Program.cs | 7 +- FinlyticFundamentals/Project.md | 31 -------- .../Services/FundamentalsDbService.cs | 53 +++++++++++--- .../Services/KeyedLockPool.cs | 66 ++++++++++++++++++ .../Util/FundamentalsMqttClient.cs | 48 ++++++------- FinlyticFundamentals/yahoo.html | Bin 2070 -> 0 bytes 6 files changed, 136 insertions(+), 69 deletions(-) delete mode 100644 FinlyticFundamentals/Project.md create mode 100644 FinlyticFundamentals/Services/KeyedLockPool.cs delete mode 100644 FinlyticFundamentals/yahoo.html diff --git a/FinlyticFundamentals/Program.cs b/FinlyticFundamentals/Program.cs index 1e6eff8..0f89c07 100644 --- a/FinlyticFundamentals/Program.cs +++ b/FinlyticFundamentals/Program.cs @@ -18,8 +18,7 @@ var builder = Host.CreateApplicationBuilder(args); // Register DB Context & ISettingsDbContext builder.Services.AddDbContext(options => - options.UseNpgsql(builder.Configuration.GetConnectionString("DefaultConnection")) - .ConfigureWarnings(w => w.Ignore(RelationalEventId.PendingModelChangesWarning))); + options.UseNpgsql(builder.Configuration.GetConnectionString("DefaultConnection"))); builder.Services.AddScoped(sp => sp.GetRequiredService()); // Register HTTP Clients @@ -57,8 +56,10 @@ using (var scope = host.Services.CreateScope()) try { var context = scope.ServiceProvider.GetRequiredService(); - await context.Database.MigrateAsync(); + var connStr = builder.Configuration.GetConnectionString("DefaultConnection") ?? ""; + await context.MigrateWithBootstrapAsync(connStr); Console.WriteLine("Database migrations successfully executed for FinlyticFundamentals."); + } catch (Exception ex) { diff --git a/FinlyticFundamentals/Project.md b/FinlyticFundamentals/Project.md deleted file mode 100644 index ba18a19..0000000 --- a/FinlyticFundamentals/Project.md +++ /dev/null @@ -1,31 +0,0 @@ -# Finlytic Fundamentals Service - -Finlytic Fundamentals is a C# background worker microservice responsible for fetching, caching, and serving financial fundamental data (P/E ratios, market cap, dividend yield, revenue growth, corporate calendar events) across global equities. - ---- - -## Core Features & Architecture - -1. **Fundamental Data Ingestion**: - - Scrapes and ingests company fundamentals (`AssetFundamentalsDto`) including P/E, EPS, Market Cap, Dividend Yield, Revenue, Profit Margins, and Debt-to-Equity ratios. - -2. **Corporate Event Calendar**: - - Tracks earnings release dates, ex-dividend dates, payout dates, and shareholder meetings (`CorporateEventDto`). - -3. **MQTT Distribution Channels**: - - Publishes fundamental updates to `finlytic/fundamentals/{isin}` and `finlytic/assets/fundamentals/{isin}`. - - Responds to RPC requests on `services/request/fundamentals_Get/#` and `services/request/events_GetAll/#`. - ---- - -## Feature Status - -### Implemented Features -- [x] Fundamentals Database Persistence & Caching (`FundamentalsDbContext`). -- [x] Corporate Event Calendar storage & query handlers. -- [x] Zero-Allocation MQTT serialization via `FinlyticJsonSerializerContext`. -- [x] Pure Worker Service architecture (`Host.CreateApplicationBuilder`, no Kestrel HTTP server). - -### Planned Features -- [ ] Financial Modeling Prep / SEC EDGAR API automated quarterly filing sync. -- [ ] Automated Dividend Growth Rate & Dividend Safety Rating calculation engine. diff --git a/FinlyticFundamentals/Services/FundamentalsDbService.cs b/FinlyticFundamentals/Services/FundamentalsDbService.cs index f695bbe..09080dd 100644 --- a/FinlyticFundamentals/Services/FundamentalsDbService.cs +++ b/FinlyticFundamentals/Services/FundamentalsDbService.cs @@ -35,7 +35,7 @@ public interface IFundamentalsDbService public class FundamentalsDbService : IFundamentalsDbService { - private static readonly ConcurrentDictionary IsinLocks = new(); + private static readonly KeyedLockPool LockPool = new(); private readonly IServiceScopeFactory _scopeFactory; private readonly IYahooFinanceScraper _scraper; @@ -65,15 +65,13 @@ public class FundamentalsDbService : IFundamentalsDbService var cleanIsin = isin.Trim().ToUpperInvariant(); var requestedTicker = ticker?.Trim().ToUpperInvariant(); - var isinLock = IsinLocks.GetOrAdd(cleanIsin, _ => new SemaphoreSlim(1, 1)); - await isinLock.WaitAsync(cancellationToken); - - try + using (await LockPool.LockAsync(cleanIsin, cancellationToken)) { using var scope = _scopeFactory.CreateScope(); var context = scope.ServiceProvider.GetRequiredService(); var settingsService = scope.ServiceProvider.GetRequiredService(); + // 1. Dynamic Settings lesen bool allowForceRefresh = await settingsService.GetSettingAsync(SettingKeys.AllowForceRefresh, cancellationToken); @@ -380,6 +378,46 @@ public class FundamentalsDbService : IFundamentalsDbService assetData.AssetEvents.Add(newEvent); } } + + // Structured Trade Republic dividend data (ExpectedDividend + historical Dividends) carries + // a real ExDate per entry - a far more reliable "this is a dividend" signal than matching + // the generic Events/PastEvents feed's free-text Type/Title strings above, whose exact + // wording for dividend entries is not guaranteed. The canonical "Dividend" Type here is + // fully controlled by this codebase (not guessed from TR's free text), so + // AssetFundamentalsDto.DaysToNextExDividend can match on it reliably (Rules.md ยง4). + var trDividendList = new List(); + if (trDetails.ExpectedDividend != null) trDividendList.Add(trDetails.ExpectedDividend); + if (trDetails.Dividends != null) trDividendList.AddRange(trDetails.Dividends); + + foreach (var div in trDividendList) + { + if (string.IsNullOrWhiteSpace(div.ExDate) || + !DateTime.TryParse(div.ExDate, System.Globalization.CultureInfo.InvariantCulture, + System.Globalization.DateTimeStyles.None, out var exDate)) + { + continue; + } + + bool isDuplicateDividend = assetData.AssetEvents.Any(e => + e.Date.Date == exDate.Date && string.Equals(e.Type, "Dividend", StringComparison.OrdinalIgnoreCase)); + + if (!isDuplicateDividend) + { + var newDividendEvent = new AssetEventEntity + { + AssetDataIsin = cleanIsin, + Ticker = new TickerEntity + { + Ticker = yahooPrimaryTicker.Ticker, + Exchange = yahooPrimaryTicker.Exchange ?? "Unknown" + }, + Type = "Dividend", + Date = exDate.Date + }; + context.AssetEvents.Add(newDividendEvent); + assetData.AssetEvents.Add(newDividendEvent); + } + } } // --- Process Modules DTO (Executives & Fundamental Data) --- @@ -579,12 +617,9 @@ public class FundamentalsDbService : IFundamentalsDbService return MapToDto(assetData, fundamentalData, executivesList, eventsList); } - finally - { - isinLock.Release(); - } } + /// public async Task> GetAllEventsAsync(CancellationToken cancellationToken = default) { diff --git a/FinlyticFundamentals/Services/KeyedLockPool.cs b/FinlyticFundamentals/Services/KeyedLockPool.cs new file mode 100644 index 0000000..254fa12 --- /dev/null +++ b/FinlyticFundamentals/Services/KeyedLockPool.cs @@ -0,0 +1,66 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace FinlyticFundamentals.Services; + +public class KeyedLockPool +{ + private readonly ConcurrentDictionary _semaphores = new(StringComparer.OrdinalIgnoreCase); + + public async Task LockAsync(string key, CancellationToken cancellationToken = default) + { + var item = _semaphores.AddOrUpdate( + key, + _ => new RefCountedSemaphore(), + (_, existing) => + { + Interlocked.Increment(ref existing.RefCount); + return existing; + }); + + await item.Semaphore.WaitAsync(cancellationToken).ConfigureAwait(false); + return new Releaser(this, key, item); + } + + private void Release(string key, RefCountedSemaphore item) + { + item.Semaphore.Release(); + if (Interlocked.Decrement(ref item.RefCount) <= 0) + { + _semaphores.TryRemove(new KeyValuePair(key, item)); + } + } + + private sealed class RefCountedSemaphore + { + public int RefCount = 1; + public readonly SemaphoreSlim Semaphore = new(1, 1); + } + + private sealed class Releaser : IDisposable + { + private readonly KeyedLockPool _pool; + private readonly string _key; + private readonly RefCountedSemaphore _item; + private bool _disposed; + + public Releaser(KeyedLockPool pool, string key, RefCountedSemaphore item) + { + _pool = pool; + _key = key; + _item = item; + } + + public void Dispose() + { + if (!_disposed) + { + _disposed = true; + _pool.Release(_key, _item); + } + } + } +} diff --git a/FinlyticFundamentals/Util/FundamentalsMqttClient.cs b/FinlyticFundamentals/Util/FundamentalsMqttClient.cs index 5de7336..fa3e7c8 100644 --- a/FinlyticFundamentals/Util/FundamentalsMqttClient.cs +++ b/FinlyticFundamentals/Util/FundamentalsMqttClient.cs @@ -35,15 +35,10 @@ public class FundamentalsMqttClient : ManagedMqttClient, IHostedService /// 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"] ?? "finlytic_fundamentals_" + Guid.NewGuid().ToString("N") - }; + var config = MqttConfiguration.FromConfiguration(_configuration, "FinlyticFundamentals"); _logger.LogInformation("[{Channel}] [MQTT_Client] Starting Fundamentals MQTT client. Host: {Host}, ClientId: {ClientId}", "MqttChannel", config.Host, config.ClientId); - + await ConnectAsync(config); } @@ -58,18 +53,19 @@ public class FundamentalsMqttClient : ManagedMqttClient, IHostedService protected override async Task OnConnectedAsync() { _logger.LogInformation("[{Channel}] [MQTT_Client] Connected. Subscribing to RPC request topics...", "MqttChannel"); - await SubscribeAsync("services/request/fundamentals_Get/#"); - await SubscribeAsync("services/request/events_GetAll/#"); - await SubscribeAsync("services/request/events_GetByMonth/#"); - await SubscribeAsync("services/request/fundamentals_settings_GetAll/#"); - await SubscribeAsync("services/request/fundamentals_settings_Update/#"); - await SubscribeAsync("services/request/health_Ping/#"); + await SubscribeAsync(MqttTopics.ResponseWildcard); + await SubscribeAsync(MqttTopics.RequestFilter(MqttTopics.Channels.FundamentalsGet)); + await SubscribeAsync(MqttTopics.RequestFilter(MqttTopics.Channels.EventsGetAll)); + await SubscribeAsync(MqttTopics.RequestFilter(MqttTopics.Channels.EventsGetByMonth)); + await SubscribeAsync(MqttTopics.RequestFilter(MqttTopics.Channels.FundamentalsSettingsGetAll)); + await SubscribeAsync(MqttTopics.RequestFilter(MqttTopics.Channels.FundamentalsSettingsUpdate)); + await SubscribeAsync(MqttTopics.RequestFilter(MqttTopics.Channels.HealthPing)); FinlyticLogBroadcaster.OnLogPublished = async (logDto) => { if (IsConnected && string.Equals(logDto.ServiceName, "FinlyticFundamentals", StringComparison.OrdinalIgnoreCase)) { - await PublishAsync("finlytic/logs/FinlyticFundamentals", logDto); + await PublishAsync(MqttTopics.Logs("FinlyticFundamentals"), logDto); } }; } @@ -84,27 +80,27 @@ public class FundamentalsMqttClient : ManagedMqttClient, IHostedService var correlationId = topic.Substring(lastSlash + 1); - if (topic.StartsWith("services/request/fundamentals_Get", StringComparison.OrdinalIgnoreCase)) + if (topic.StartsWith(MqttTopics.RequestPrefix + MqttTopics.Channels.FundamentalsGet, StringComparison.OrdinalIgnoreCase)) { await OnFundamentalsGetAsync(payload, correlationId); } - else if (topic.StartsWith("services/request/events_GetAll", StringComparison.OrdinalIgnoreCase)) + else if (topic.StartsWith(MqttTopics.RequestPrefix + MqttTopics.Channels.EventsGetAll, StringComparison.OrdinalIgnoreCase)) { await OnEventsGetAllAsync(correlationId); } - else if (topic.StartsWith("services/request/events_GetByMonth", StringComparison.OrdinalIgnoreCase)) + else if (topic.StartsWith(MqttTopics.RequestPrefix + MqttTopics.Channels.EventsGetByMonth, StringComparison.OrdinalIgnoreCase)) { await OnEventsGetByMonthAsync(payload, correlationId); } - else if (topic.StartsWith("services/request/fundamentals_settings_GetAll", StringComparison.OrdinalIgnoreCase)) + else if (topic.StartsWith(MqttTopics.RequestPrefix + MqttTopics.Channels.FundamentalsSettingsGetAll, StringComparison.OrdinalIgnoreCase)) { await OnSettingsGetAllAsync(correlationId); } - else if (topic.StartsWith("services/request/fundamentals_settings_Update", StringComparison.OrdinalIgnoreCase)) + else if (topic.StartsWith(MqttTopics.RequestPrefix + MqttTopics.Channels.FundamentalsSettingsUpdate, StringComparison.OrdinalIgnoreCase)) { await OnSettingsUpdateAsync(payload, correlationId); } - else if (topic.StartsWith("services/request/health_Ping", StringComparison.OrdinalIgnoreCase)) + else if (topic.StartsWith(MqttTopics.RequestPrefix + MqttTopics.Channels.HealthPing, StringComparison.OrdinalIgnoreCase)) { await OnHealthPingAsync(topic, correlationId); } @@ -135,7 +131,7 @@ public class FundamentalsMqttClient : ManagedMqttClient, IHostedService request.Isin, request.ForceRefresh.ToString(), correlationId); var fundamentals = await dbService.GetFundamentalsAsync(request.Isin, request.Ticker, request.ForceRefresh); - var responseTopic = $"services/response/fundamentals_Get/{correlationId}"; + var responseTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.FundamentalsGet, correlationId); await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [MQTT_Client] Publishing RPC fundamentals response to '{ResponseTopic}'", responseTopic); await PublishAsync(responseTopic, fundamentals); @@ -156,7 +152,7 @@ public class FundamentalsMqttClient : ManagedMqttClient, IHostedService try { var events = await dbService.GetAllEventsAsync(); - var responseTopic = $"services/response/events_GetAll/{correlationId}"; + var responseTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.EventsGetAll, correlationId); await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [MQTT_Client] Publishing events RPC response to '{ResponseTopic}'", responseTopic); await PublishAsync(responseTopic, events); @@ -183,7 +179,7 @@ public class FundamentalsMqttClient : ManagedMqttClient, IHostedService await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [MQTT_Client] Processing RPC events_GetByMonth request for {Year}/{Month} [CorrelationId: {CorrelationId}]", request.Year, request.Month, correlationId); var events = await dbService.GetEventsByMonthAsync(request.Year, request.Month); - var responseTopic = $"services/response/events_GetByMonth/{correlationId}"; + var responseTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.EventsGetByMonth, correlationId); await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [MQTT_Client] Publishing monthly events RPC response to '{ResponseTopic}'", responseTopic); await PublishAsync(responseTopic, events); @@ -204,7 +200,7 @@ public class FundamentalsMqttClient : ManagedMqttClient, IHostedService try { var settings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) }); - var responseTopic = $"services/response/fundamentals_settings_GetAll/{correlationId}"; + var responseTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.FundamentalsSettingsGetAll, correlationId); await PublishAsync(responseTopic, settings); await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [Settings_GetAll] Published {Count} settings to '{ResponseTopic}'", settings.Count, responseTopic); @@ -251,7 +247,7 @@ public class FundamentalsMqttClient : ManagedMqttClient, IHostedService } var currentSettings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) }); - var responseTopic = $"services/response/fundamentals_settings_Update/{correlationId}"; + var responseTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.FundamentalsSettingsUpdate, correlationId); await PublishAsync(responseTopic, currentSettings); } catch (Exception ex) @@ -267,7 +263,7 @@ public class FundamentalsMqttClient : ManagedMqttClient, IHostedService await using var scope = _scopeFactory.CreateAsyncScope(); var finlyticLogger = scope.ServiceProvider.GetRequiredService>(); - var respTopic = $"services/response/health_Ping/{correlationId}"; + var respTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.HealthPing, correlationId); await PublishAsync(respTopic, new ServiceHealthResponse("FinlyticFundamentals", "Online", DateTime.UtcNow, "Connected")); await finlyticLogger.LogInfoAsync(SettingKeys.HealthPingChannel, "[FinlyticFundamentals] [Health_Ping] Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", correlationId); } diff --git a/FinlyticFundamentals/yahoo.html b/FinlyticFundamentals/yahoo.html deleted file mode 100644 index e15fe93bd32c75cb09d86722c08953b420d479ce..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 2070 zcmeH|Ur!oQ5XH~4NxuX7(1jRSO);&yU>fa%iM}-T*{lo5Cj2W4txe-sul{B(TLek` z0FuovcjultGjnG4^7JoJUo#CgP+JMVUEWevlwnUbRpFMFBF)+9sje@q+xnm#@CCT3 za!uF^wj4}Vzp;(f(U?_5yK3=#X={KV<5kmNXEJ0)1dBgnuL0V4XS`R5mw;{S9F*+_ zOojczGo6E-F02Fn!8Qk%>KAuTWvT9Xc418vJ5z#vsC$qFo*DV=u@htI5~Be}O8sMB zXFP}QcLutpL#|_e<~rg&&?nwLl9g#>s#F%&KEEUQOo0pdLM=7eEy_$kJO`_<2Ex!0 zJh%G3k;!VdOgi}9fVp%vhE!vk1VyT@_^OVkOOHD!G$q!0jutiSx74Jb=7u^V^a&m8 zka^$rIw1Cg+xz(ERArN7N|rk@J&TBV>Mm2$7B;T@fL_cDBJ z3XFe7)IulTt!Sa!TTIr;0?$(ASZ6SrBFl=GS%%4u@mQbj1f`AXwf$P(=;v1=lsVRk z@RK>rnL^Vf^ueA57*o@_xt!*p+o|{V3XhVfthEP|&9qH{P4cD*U9W2I|92X1$lSgK ai&B33FW1|1V0#X1&jI^~dvE7}oqo@$PexV%