364 lines
17 KiB
C#
364 lines
17 KiB
C#
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.Dtos.TechnicalAnalysis;
|
|
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 IServiceScopeFactory _scopeFactory;
|
|
private readonly ILogger<TradesMqttClient> _logger;
|
|
|
|
public TradesMqttClient(
|
|
IConfiguration configuration,
|
|
IServiceScopeFactory scopeFactory,
|
|
ILogger<TradesMqttClient> logger) : base(logger)
|
|
{
|
|
_configuration = configuration;
|
|
_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/#");
|
|
await SubscribeAsync("services/response/tr_GetLivePrice/#");
|
|
|
|
_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;
|
|
}
|
|
|
|
// Für Scoped-Services erzeugen wir pro eingehender Nachricht einen eigenen Scope
|
|
using var msgScope = _scopeFactory.CreateScope();
|
|
var tradeLifecycleService = msgScope.ServiceProvider.GetRequiredService<ITradeLifecycleService>();
|
|
|
|
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 activeTrades = trades.Where(t => t.Status == TradeStatus.Active && !string.IsNullOrWhiteSpace(t.Isin)).ToList();
|
|
if (activeTrades.Count > 0)
|
|
{
|
|
try
|
|
{
|
|
var priceTasks = activeTrades.Select(t => FetchLivePriceAsync(t.Isin)).ToList();
|
|
var livePricesTask = Task.WhenAll(priceTasks);
|
|
if (await Task.WhenAny(livePricesTask, Task.Delay(1500)) == livePricesTask)
|
|
{
|
|
var livePrices = await livePricesTask;
|
|
for (int i = 0; i < activeTrades.Count; i++)
|
|
{
|
|
var lp = livePrices[i];
|
|
if (lp != null && lp.CurrentPrice > 0m)
|
|
{
|
|
var trade = activeTrades[i];
|
|
tradeLifecycleService.CalculatePnL(trade, lp.CurrentPrice);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogDebug(ex, "[{Channel}] Live price fetch skipped or timed out during trades_Get", "TradesChannel");
|
|
}
|
|
}
|
|
|
|
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);
|
|
|
|
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/user/{trade.UserId ?? "all"}", trade);
|
|
await PublishAsync("finlytic/trades/update", trade);
|
|
}
|
|
|
|
private async Task<LivePriceDto?> FetchLivePriceAsync(string isin)
|
|
{
|
|
if (string.IsNullOrWhiteSpace(isin)) return null;
|
|
try
|
|
{
|
|
return await SendRpcRequestAsync<LivePriceDto, IsinRequest>(
|
|
"tr_GetLivePrice",
|
|
new IsinRequest(isin),
|
|
TimeSpan.FromMilliseconds(1200));
|
|
}
|
|
catch
|
|
{
|
|
return null;
|
|
}
|
|
}
|
|
|
|
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,
|
|
AssetType = t.AssetType,
|
|
HasCfd = t.HasCfd,
|
|
DerivativeProductCategories = t.DerivativeProductCategories ?? new List<string>(),
|
|
DerivativeIsin = t.DerivativeIsin,
|
|
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,
|
|
PnlAbsolute = t.PnlAbsolute,
|
|
PnlPercent = t.PnlPercent,
|
|
CurrentPrice = t.UserExitPrice ?? t.HourlyUpdates?.LastOrDefault()?.CurrentPrice,
|
|
CloseReason = t.CloseReason,
|
|
UserExitTimestamp = t.UserExitTimestamp,
|
|
HasPendingExitAlert = t.Status == TradeStatus.Active && t.HourlyUpdates != null && t.HourlyUpdates.Any(u => string.Equals(u.Recommendation, "Close", StringComparison.OrdinalIgnoreCase)),
|
|
PendingExitReason = t.Status == TradeStatus.Active ? t.HourlyUpdates?.LastOrDefault(u => string.Equals(u.Recommendation, "Close", StringComparison.OrdinalIgnoreCase))?.Reasoning : null,
|
|
HourlyUpdates = t.HourlyUpdates?.OrderBy(u => u.Timestamp).Select(u => new TradeHourlyUpdateDto
|
|
{
|
|
TradeId = t.TradeId,
|
|
Recommendation = u.Recommendation,
|
|
CurrentPrice = u.CurrentPrice,
|
|
SuggestedStopLoss = u.SuggestedStopLoss,
|
|
SuggestedTakeProfit = u.SuggestedTakeProfit,
|
|
VixValue = u.VixValue,
|
|
Reasoning = u.Reasoning,
|
|
Timestamp = u.Timestamp
|
|
}).ToList()
|
|
};
|
|
}
|
|
} |