feat(engine): add FinlyticEngine microservice with trade lifecycle, AI reasoning gate, composite scoring, and unit tests
This commit is contained in:
@@ -0,0 +1,200 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using FinlyticCore.Dtos;
|
||||
using FinlyticCore.Dtos.Settings;
|
||||
using FinlyticCore.Dtos.Trading;
|
||||
using FinlyticCore.Models;
|
||||
using FinlyticCore.Services;
|
||||
using FinlyticCore.Util;
|
||||
using FinlyticEngine.Services.Mqtt;
|
||||
using FinlyticEngine.Services.Trading;
|
||||
using FinlyticEngine.Settings;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace FinlyticEngine.Util;
|
||||
|
||||
public class EngineMqttClient : ManagedMqttClient, IHostedService, IEngineRpcClient
|
||||
{
|
||||
private readonly IConfiguration _configuration;
|
||||
private readonly IServiceScopeFactory _scopeFactory;
|
||||
private readonly ILogger<EngineMqttClient> _logger;
|
||||
|
||||
public EngineMqttClient(
|
||||
ILogger<EngineMqttClient> logger,
|
||||
IConfiguration configuration,
|
||||
IServiceScopeFactory scopeFactory) : base(logger)
|
||||
{
|
||||
_logger = logger;
|
||||
_configuration = configuration;
|
||||
_scopeFactory = scopeFactory;
|
||||
}
|
||||
|
||||
public async Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
var config = MqttConfiguration.FromConfiguration(_configuration, "FinlyticEngine");
|
||||
|
||||
_logger.LogInformation("Starting FinlyticEngine MQTT client. Host: {Host}, ClientId: {ClientId}", config.Host, config.ClientId);
|
||||
await ConnectAsync(config);
|
||||
}
|
||||
|
||||
public async Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_logger.LogInformation("Stopping FinlyticEngine MQTT client.");
|
||||
await DisconnectAsync();
|
||||
}
|
||||
|
||||
protected override async Task OnConnectedAsync()
|
||||
{
|
||||
_logger.LogInformation("FinlyticEngine MQTT client connected. Registering RPC endpoints...");
|
||||
|
||||
await SubscribeAsync(MqttTopics.ResponseWildcard);
|
||||
await SubscribeRpcAsync<GetTradeProposalsRequest, List<TradeProposalDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineGetProposals), HandleGetProposalsRpcAsync);
|
||||
await SubscribeRpcAsync<GetActiveTradesRequest, List<ActiveTradeDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineGetTrades), HandleGetTradesRpcAsync);
|
||||
await SubscribeRpcAsync<EvaluateAssetRequest, AssetEvaluationResultDto>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineEvaluateIsin), HandleEvaluateIsinRpcAsync);
|
||||
await SubscribeRpcAsync<GetEvaluationHistoryRequest, GetEvaluationHistoryResponse>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineGetEvaluationHistory), HandleGetEvaluationHistoryRpcAsync);
|
||||
await SubscribeRpcAsync<AddTradeFillRequest, ActiveTradeDto>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineAddFill), HandleAddFillRpcAsync);
|
||||
await SubscribeRpcAsync<UpdateTradeStopLossRequest, ActiveTradeDto>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineUpdateStopLoss), HandleUpdateStopLossRpcAsync);
|
||||
await SubscribeRpcAsync<CloseEngineTradeRequest, ActiveTradeDto>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineCloseTrade), HandleCloseTradeRpcAsync);
|
||||
await SubscribeRpcAsync<AcceptTradeProposalRequest, ActiveTradeDto>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineAcceptProposal), HandleAcceptProposalRpcAsync);
|
||||
await SubscribeRpcAsync<CreateManualTradeRequest, ActiveTradeDto>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineCreateManualTrade), HandleCreateManualTradeRpcAsync);
|
||||
await SubscribeRpcAsync<object, List<DynamicSettingDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineSettingsGetAll), HandleSettingsGetAllRpcAsync);
|
||||
await SubscribeRpcAsync<Dictionary<string, object?>, List<DynamicSettingDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineSettingsUpdate), HandleSettingsUpdateRpcAsync);
|
||||
await SubscribeAsync<object>(MqttTopics.RequestFilter(MqttTopics.Channels.HealthPing), HandleHealthPingRpcAsync);
|
||||
|
||||
FinlyticLogBroadcaster.OnLogPublished = async (logDto) =>
|
||||
{
|
||||
if (IsConnected && string.Equals(logDto.ServiceName, "FinlyticEngine", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
await PublishAsync(MqttTopics.Logs("FinlyticEngine"), logDto);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
private async Task<List<TradeProposalDto>> HandleGetProposalsRpcAsync(GetTradeProposalsRequest? req, string correlationId)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var lifecycleService = scope.ServiceProvider.GetRequiredService<ITradeLifecycleService>();
|
||||
return await lifecycleService.GetProposalsAsync(req?.OnlyActive ?? true, req?.Limit ?? 50);
|
||||
}
|
||||
|
||||
private async Task<List<ActiveTradeDto>> HandleGetTradesRpcAsync(GetActiveTradesRequest? req, string correlationId)
|
||||
{
|
||||
if (req == null) throw new ArgumentNullException(nameof(req));
|
||||
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var lifecycleService = scope.ServiceProvider.GetRequiredService<ITradeLifecycleService>();
|
||||
return await lifecycleService.GetActiveTradesAsync(req.UserId, req.Mode);
|
||||
}
|
||||
|
||||
private async Task<AssetEvaluationResultDto> HandleEvaluateIsinRpcAsync(EvaluateAssetRequest? req, string correlationId)
|
||||
{
|
||||
// A blank/missing ISIN is no longer a special case here: EvaluateAssetAsync now always returns a
|
||||
// populated AssetEvaluationResultDto (never null), including for a blank ISIN, so it is safe to just
|
||||
// delegate straight through.
|
||||
//
|
||||
// This RPC channel is only ever reached from the manual, on-demand Web UI flows
|
||||
// (AnalyzeController.TriggerManualAnalysis / EngineController.EvaluateAsset) - the autonomous
|
||||
// OpportunityPollerBackgroundService calls ITradeLifecycleService.EvaluateAssetAsync directly
|
||||
// in-process and never goes through MQTT for it - so TriggerSource is always Manual here. UserId comes
|
||||
// from EvaluateAssetRequest.UserId, which FinlyticBackend always overwrites server-side with the JWT
|
||||
// identity before publishing the request (see EvaluateAssetRequest's doc comment); Guid.Empty (the
|
||||
// request's own default) is treated as "no identity available" rather than a real user ID.
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var lifecycleService = scope.ServiceProvider.GetRequiredService<ITradeLifecycleService>();
|
||||
Guid? triggeredByUserId = req != null && req.UserId != Guid.Empty ? req.UserId : null;
|
||||
return await lifecycleService.EvaluateAssetAsync(
|
||||
req?.Isin ?? string.Empty, req?.Ticker, req?.ForceAiEvaluation ?? false,
|
||||
triggerSource: TriggerSource.Manual, triggeredByUserId: triggeredByUserId);
|
||||
}
|
||||
|
||||
private async Task<GetEvaluationHistoryResponse> HandleGetEvaluationHistoryRpcAsync(GetEvaluationHistoryRequest? req, string correlationId)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var historyService = scope.ServiceProvider.GetRequiredService<IEvaluationHistoryService>();
|
||||
return await historyService.GetHistoryAsync(req ?? new GetEvaluationHistoryRequest());
|
||||
}
|
||||
|
||||
private async Task<ActiveTradeDto> HandleAddFillRpcAsync(AddTradeFillRequest? req, string correlationId)
|
||||
{
|
||||
if (req == null) throw new ArgumentNullException(nameof(req));
|
||||
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var lifecycleService = scope.ServiceProvider.GetRequiredService<ITradeLifecycleService>();
|
||||
return await lifecycleService.AddTradeFillAsync(req.UserId, req.TradeId, req.ExecutedPrice, req.Quantity, req.Fee, req.Note);
|
||||
}
|
||||
|
||||
private async Task<ActiveTradeDto> HandleUpdateStopLossRpcAsync(UpdateTradeStopLossRequest? req, string correlationId)
|
||||
{
|
||||
if (req == null) throw new ArgumentNullException(nameof(req));
|
||||
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var lifecycleService = scope.ServiceProvider.GetRequiredService<ITradeLifecycleService>();
|
||||
return await lifecycleService.UpdateStopLossAsync(req.UserId, req.TradeId, req.NewStopLoss, req.Reason);
|
||||
}
|
||||
|
||||
private async Task<ActiveTradeDto> HandleCloseTradeRpcAsync(CloseEngineTradeRequest? req, string correlationId)
|
||||
{
|
||||
if (req == null) throw new ArgumentNullException(nameof(req));
|
||||
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var lifecycleService = scope.ServiceProvider.GetRequiredService<ITradeLifecycleService>();
|
||||
return await lifecycleService.CloseTradeAsync(req.UserId, req.TradeId, req.ClosePrice, req.Reason);
|
||||
}
|
||||
|
||||
private async Task<ActiveTradeDto> HandleAcceptProposalRpcAsync(AcceptTradeProposalRequest? req, string correlationId)
|
||||
{
|
||||
if (req == null) throw new ArgumentNullException(nameof(req));
|
||||
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var lifecycleService = scope.ServiceProvider.GetRequiredService<ITradeLifecycleService>();
|
||||
return await lifecycleService.AcceptProposalAsync(req);
|
||||
}
|
||||
|
||||
private async Task<ActiveTradeDto> HandleCreateManualTradeRpcAsync(CreateManualTradeRequest? req, string correlationId)
|
||||
{
|
||||
if (req == null) throw new ArgumentNullException(nameof(req));
|
||||
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var lifecycleService = scope.ServiceProvider.GetRequiredService<ITradeLifecycleService>();
|
||||
return await lifecycleService.CreateManualTradeAsync(req);
|
||||
}
|
||||
|
||||
private async Task<List<DynamicSettingDto>> HandleSettingsGetAllRpcAsync(object? _, string correlationId)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
||||
return await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(EngineSettingKeys) });
|
||||
}
|
||||
|
||||
private async Task<List<DynamicSettingDto>> HandleSettingsUpdateRpcAsync(Dictionary<string, object?>? updates, string correlationId)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
||||
|
||||
if (updates != null && updates.Count > 0)
|
||||
{
|
||||
await settingsService.UpdateSettingsAsync(updates);
|
||||
}
|
||||
|
||||
return await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(EngineSettingKeys) });
|
||||
}
|
||||
|
||||
private async Task HandleHealthPingRpcAsync(object? _, string topic, string correlationId)
|
||||
{
|
||||
if (topic.Contains("FinlyticEngine", StringComparison.OrdinalIgnoreCase) || !topic.Contains("/", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
string respTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.HealthPing, correlationId);
|
||||
await PublishAsync(respTopic, new ServiceHealthResponse("FinlyticEngine", "Online", DateTime.UtcNow, "Connected"));
|
||||
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var logger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<EngineMqttClient>>();
|
||||
await logger.LogInfoAsync(EngineSettingKeys.HealthPingChannel,
|
||||
"[FinlyticEngine] Responded to health_Ping RPC [CorrelationId: {CorrelationId}]", correlationId);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user