feat(Trades): refactor trades MQTT client and DTOs
This commit is contained in:
@@ -0,0 +1,296 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Globalization;
|
||||
using System.Linq;
|
||||
using System.Text.Json;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using FinlyticCore.Dtos;
|
||||
using FinlyticCore.Models;
|
||||
using FinlyticCore.Models.Trades;
|
||||
using FinlyticCore.Util;
|
||||
using FinlyticTrades.Entities;
|
||||
using FinlyticTrades.Services;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace FinlyticTrades.Util;
|
||||
|
||||
public class TradesMqttClient : ManagedMqttClient, IHostedService
|
||||
{
|
||||
private readonly IConfiguration _configuration;
|
||||
private readonly ITradeLifecycleService _tradeLifecycleService;
|
||||
private readonly IServiceScopeFactory _scopeFactory;
|
||||
private readonly ILogger<TradesMqttClient> _logger;
|
||||
|
||||
public TradesMqttClient(
|
||||
IConfiguration configuration,
|
||||
ITradeLifecycleService tradeLifecycleService,
|
||||
IServiceScopeFactory scopeFactory,
|
||||
ILogger<TradesMqttClient> logger) : base(logger)
|
||||
{
|
||||
_configuration = configuration;
|
||||
_tradeLifecycleService = tradeLifecycleService;
|
||||
_scopeFactory = scopeFactory;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
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"),
|
||||
Username = _configuration["MQTT:Username"] ?? _configuration["MQTT__Username"],
|
||||
Password = _configuration["MQTT:Password"] ?? _configuration["MQTT__Password"],
|
||||
ClientId = $"{(_configuration["MQTT:ClientId"] ?? _configuration["MQTT__ClientId"] ?? "finlytic_trades")}_{Guid.NewGuid():N}"
|
||||
};
|
||||
|
||||
_logger.LogInformation("[{Channel}] Starting Unified Trades MQTT Client. Host: {Host}, ClientId: {ClientId}", "TradesChannel", config.Host, config.ClientId);
|
||||
await ConnectAsync(config);
|
||||
}
|
||||
|
||||
public async Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_logger.LogInformation("[{Channel}] Stopping Unified Trades MQTT Client.", "TradesChannel");
|
||||
await DisconnectAsync();
|
||||
}
|
||||
|
||||
protected override async Task OnConnectedAsync()
|
||||
{
|
||||
_logger.LogInformation("[{Channel}] Trades MQTT Client connected. Subscribing to topics...", "TradesChannel");
|
||||
|
||||
await SubscribeAsync("finlytic/trades/proposed/#");
|
||||
await SubscribeAsync("finlytic/trades/updates/#");
|
||||
await SubscribeAsync("finlytic/trades/accept/#");
|
||||
await SubscribeAsync("services/request/trades_Get/#");
|
||||
await SubscribeAsync("services/request/trades_Close/#");
|
||||
await SubscribeAsync("services/request/trades_Reject/#");
|
||||
await SubscribeAsync("services/request/trades_Accept/#");
|
||||
await SubscribeAsync("services/config/updated/#");
|
||||
await SubscribeAsync("services/request/health_Ping/#");
|
||||
|
||||
_logger.LogInformation("[{Channel}] Successfully subscribed to all event and RPC channels.", "TradesChannel");
|
||||
}
|
||||
|
||||
protected override async Task OnMessageReceivedAsync(string topic, string payloadStr)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (topic.Contains("health_Ping", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
var segments = topic.Split('/');
|
||||
bool isForMe = segments.Length >= 5
|
||||
? segments[3].Equals("FinlyticTrades", StringComparison.OrdinalIgnoreCase)
|
||||
: topic.Contains("FinlyticTrades", StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
if (isForMe)
|
||||
{
|
||||
var correlationId = segments[^1];
|
||||
string respTopic = $"services/response/health_Ping/{correlationId}";
|
||||
var healthResp = new ServiceHealthResponse("FinlyticTrades", "Online", DateTime.UtcNow, "Connected");
|
||||
await PublishAsync(respTopic, healthResp);
|
||||
_logger.LogInformation("[{Channel}] [TradesMqttClient] Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", "TradesChannel", correlationId);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if (topic.StartsWith("services/config/updated", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
if (topic.EndsWith("FinlyticTrades", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
_logger.LogInformation("[{Channel}] [TradesMqttClient] Received config update event for FinlyticTrades.", "TradesChannel");
|
||||
var payload = JsonSerializer.Deserialize(payloadStr, FinlyticJsonSerializerContext.Default.ServiceConfigUpdatePayload);
|
||||
if (payload?.Settings != null && payload.Settings.Count > 0)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var settingsDb = scope.ServiceProvider.GetRequiredService<ISettingsDbService>();
|
||||
await settingsDb.UpdateSettingsFromDictionaryAsync(payload.Settings);
|
||||
_logger.LogInformation("[{Channel}] [TradesMqttClient] Persisted {Count} updated settings to FinlyticTrades database.", "TradesChannel", payload.Settings.Count);
|
||||
}
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if (topic.StartsWith("finlytic/trades/proposed/"))
|
||||
{
|
||||
var proposal = JsonSerializer.Deserialize(payloadStr, FinlyticJsonSerializerContext.Default.TradeProposalDto);
|
||||
if (proposal != null && (!string.IsNullOrWhiteSpace(proposal.Symbol) || !string.IsNullOrWhiteSpace(proposal.Isin)))
|
||||
{
|
||||
await _tradeLifecycleService.ProcessProposedTradeAsync(proposal, CancellationToken.None);
|
||||
}
|
||||
else
|
||||
{
|
||||
_logger.LogWarning("[{Channel}] [TradesMqttClient] Received proposed trade payload but Symbol/ISIN is empty. Skipping ingestion.", "TradesChannel");
|
||||
}
|
||||
}
|
||||
else if (topic.StartsWith("finlytic/trades/accept/"))
|
||||
{
|
||||
var acceptDto = JsonSerializer.Deserialize(payloadStr, FinlyticJsonSerializerContext.Default.TradeAcceptanceDto);
|
||||
if (acceptDto != null)
|
||||
{
|
||||
var newTrade = await _tradeLifecycleService.AcceptTradeAsync(acceptDto, CancellationToken.None);
|
||||
if (newTrade != null)
|
||||
{
|
||||
var dto = MapToDto(newTrade);
|
||||
await PublishTradeUpdateAsync(dto);
|
||||
}
|
||||
}
|
||||
}
|
||||
else if (topic.StartsWith("services/request/trades_Accept/"))
|
||||
{
|
||||
var correlationId = topic.Split('/').Last();
|
||||
var acceptDto = JsonSerializer.Deserialize(payloadStr, FinlyticJsonSerializerContext.Default.TradeAcceptanceDto);
|
||||
if (acceptDto != null)
|
||||
{
|
||||
var acceptedTrade = await _tradeLifecycleService.AcceptTradeAsync(acceptDto, CancellationToken.None);
|
||||
if (acceptedTrade != null)
|
||||
{
|
||||
var acceptedDto = MapToDto(acceptedTrade);
|
||||
await PublishAsync($"services/response/trades_Accept/{correlationId}", acceptedDto);
|
||||
await PublishTradeUpdateAsync(acceptedDto);
|
||||
}
|
||||
}
|
||||
}
|
||||
else if (topic.StartsWith("finlytic/trades/updates/"))
|
||||
{
|
||||
var update = JsonSerializer.Deserialize(payloadStr, FinlyticJsonSerializerContext.Default.TradeHourlyUpdateDto);
|
||||
if (update != null)
|
||||
{
|
||||
await _tradeLifecycleService.AddHourlyUpdateAsync(update, CancellationToken.None);
|
||||
}
|
||||
}
|
||||
else if (topic.StartsWith("services/request/trades_Get/"))
|
||||
{
|
||||
var correlationId = topic.Split('/').Last();
|
||||
var request = JsonSerializer.Deserialize(payloadStr, FinlyticJsonSerializerContext.Default.GetTradesRequest);
|
||||
|
||||
string? isin = request?.Isin;
|
||||
string? status = request?.Status;
|
||||
string? userId = request?.UserId;
|
||||
|
||||
var trades = await _tradeLifecycleService.GetTradesAsync(isin, status, userId);
|
||||
var dtos = trades.Select(MapToDto).ToList();
|
||||
|
||||
await PublishAsync($"services/response/trades_Get/{correlationId}", dtos);
|
||||
}
|
||||
else if (topic.StartsWith("services/request/trades_Close/"))
|
||||
{
|
||||
var parts = topic.Split('/');
|
||||
var tradeId = parts.Length > 3 ? parts[3] : string.Empty;
|
||||
var correlationId = parts.Length > 4 ? parts[4] : string.Empty;
|
||||
|
||||
var request = JsonSerializer.Deserialize(payloadStr, FinlyticJsonSerializerContext.Default.CloseTradeRequest);
|
||||
|
||||
if (request != null && !string.IsNullOrEmpty(tradeId))
|
||||
{
|
||||
var closedTrade = await _tradeLifecycleService.CloseTradeAsync(tradeId, request);
|
||||
if (closedTrade != null)
|
||||
{
|
||||
var closedDto = MapToDto(closedTrade);
|
||||
await PublishAsync($"services/response/trades_Close/{correlationId}", closedDto);
|
||||
|
||||
// Send event stream update specifically for closed trades (used by Feedback Engine & Analytics)
|
||||
string sectorSafe = string.IsNullOrWhiteSpace(closedTrade.Sector) ? "general" : closedTrade.Sector.ToLowerInvariant();
|
||||
await PublishAsync($"finlytic/trades/closed/{sectorSafe}/{closedTrade.Symbol.ToLowerInvariant()}", closedDto);
|
||||
await PublishTradeUpdateAsync(closedDto);
|
||||
}
|
||||
}
|
||||
}
|
||||
else if (topic.StartsWith("services/request/trades_Reject/"))
|
||||
{
|
||||
var parts = topic.Split('/');
|
||||
var tradeId = parts.Length > 3 ? parts[3] : string.Empty;
|
||||
var correlationId = parts.Length > 4 ? parts[4] : string.Empty;
|
||||
|
||||
var request = JsonSerializer.Deserialize(payloadStr, FinlyticJsonSerializerContext.Default.CloseTradeRequest);
|
||||
|
||||
if (request != null && !string.IsNullOrEmpty(tradeId))
|
||||
{
|
||||
var rejectedTrade = await _tradeLifecycleService.RejectTradeAsync(tradeId, request);
|
||||
if (rejectedTrade != null)
|
||||
{
|
||||
var rejectedDto = MapToDto(rejectedTrade);
|
||||
await PublishAsync($"services/response/trades_Reject/{correlationId}", rejectedDto);
|
||||
await PublishTradeUpdateAsync(rejectedDto);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "[{Channel}] Error processing incoming MQTT message on topic {Topic}", "TradesChannel", topic);
|
||||
}
|
||||
}
|
||||
|
||||
public async Task PublishTradeUpdateAsync(TradeProposalDto trade)
|
||||
{
|
||||
await PublishAsync("finlytic/trades/update", trade);
|
||||
}
|
||||
|
||||
private static TradeProposalDto MapToDto(TradeEntity t)
|
||||
{
|
||||
List<decimal>? parseTakeProfitTargets()
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(t.TakeProfitTargets)) return null;
|
||||
|
||||
var list = new List<decimal>();
|
||||
var parts = t.TakeProfitTargets.Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries);
|
||||
foreach (var part in parts)
|
||||
{
|
||||
if (decimal.TryParse(part, NumberStyles.Number, CultureInfo.InvariantCulture, out var val))
|
||||
{
|
||||
list.Add(val);
|
||||
}
|
||||
}
|
||||
return list.Count > 0 ? list : null;
|
||||
}
|
||||
|
||||
return new TradeProposalDto
|
||||
{
|
||||
TradeId = t.TradeId,
|
||||
Status = t.Status.ToString(),
|
||||
AnalysisId = t.AnalysisId,
|
||||
EventId = t.EventId,
|
||||
Sector = t.Sector,
|
||||
Symbol = t.Symbol,
|
||||
Isin = t.Isin,
|
||||
CompanyName = t.CompanyName,
|
||||
EntryPrice = t.EntryPrice,
|
||||
StopLoss = t.StopLoss,
|
||||
TakeProfit = t.TakeProfit,
|
||||
SignalType = t.SignalType,
|
||||
RiskTolerance = t.RiskTolerance,
|
||||
Timeframe = t.Timeframe,
|
||||
InstrumentType = t.InstrumentType,
|
||||
WinRate = t.WinRate,
|
||||
VixRegime = t.VixRegime,
|
||||
VixValue = t.VixValue,
|
||||
TtlMinutes = t.TtlMinutes,
|
||||
Reasoning = t.Reasoning,
|
||||
EntryZoneMin = t.EntryZoneMin,
|
||||
EntryZoneMax = t.EntryZoneMax,
|
||||
TakeProfitTargets = parseTakeProfitTargets(),
|
||||
RiskRewardRatio = t.RiskRewardRatio,
|
||||
MaxLeverage = t.MaxLeverage,
|
||||
TechnicalRationale = t.TechnicalRationale,
|
||||
FundamentalRationale = t.FundamentalRationale,
|
||||
RiskWarning = t.RiskWarning,
|
||||
CreatedAt = t.CreatedAt,
|
||||
|
||||
UserId = t.UserId,
|
||||
IsGlobalProposal = t.IsGlobalProposal,
|
||||
ActualEntryPrice = t.ActualEntryPrice,
|
||||
PositionSize = t.PositionSize,
|
||||
LeverageUsed = t.LeverageUsed,
|
||||
EntryFee = t.EntryFee,
|
||||
ExitFee = t.ExitFee,
|
||||
ExecutionTimestamp = t.ExecutionTimestamp,
|
||||
Quantity = t.Quantity,
|
||||
KnockoutThreshold = t.KnockoutThreshold,
|
||||
IsRecurring = t.IsRecurring
|
||||
};
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user