825 lines
41 KiB
C#
825 lines
41 KiB
C#
using System;
|
|
using System.Collections.Generic;
|
|
using System.Linq;
|
|
using System.Text.Json;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using FinlyticAnalyzer.Database;
|
|
using FinlyticAnalyzer.Entities;
|
|
using FinlyticAnalyzer.Services;
|
|
using FinlyticCore.Dtos;
|
|
using FinlyticCore.Models;
|
|
using FinlyticCore.Models.Analyzer;
|
|
using FinlyticCore.Models.Trades;
|
|
using FinlyticCore.Util;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using Microsoft.Extensions.Configuration;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Hosting;
|
|
using Microsoft.Extensions.Logging;
|
|
|
|
namespace FinlyticAnalyzer.Util;
|
|
|
|
/// <summary>
|
|
/// Unified Managed MQTT Client for FinlyticAnalyzer.
|
|
/// Handles event subscriptions, market screening, manual AI evaluation triggers,
|
|
/// and dispatches trade proposals via MQTT.
|
|
/// </summary>
|
|
public class AnalyzerMqttClient : ManagedMqttClient, IHostedService
|
|
{
|
|
private readonly IConfiguration _configuration;
|
|
private readonly IServiceScopeFactory _scopeFactory;
|
|
private readonly IVixTrackerService _vixTracker;
|
|
private readonly IThreeLayerFilterEngine _filterEngine;
|
|
private readonly IWinRateCalculator _winRateCalculator;
|
|
private readonly IN8nEvaluationService _n8nService;
|
|
private readonly ILogger<AnalyzerMqttClient> _logger;
|
|
|
|
public AnalyzerMqttClient(
|
|
IConfiguration configuration,
|
|
IServiceScopeFactory scopeFactory,
|
|
IVixTrackerService vixTracker,
|
|
IThreeLayerFilterEngine filterEngine,
|
|
IWinRateCalculator winRateCalculator,
|
|
IN8nEvaluationService n8nService,
|
|
ILogger<AnalyzerMqttClient> logger) : base(logger)
|
|
{
|
|
_configuration = configuration;
|
|
_scopeFactory = scopeFactory;
|
|
_vixTracker = vixTracker;
|
|
_filterEngine = filterEngine;
|
|
_winRateCalculator = winRateCalculator;
|
|
_n8nService = n8nService;
|
|
_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_analyzer")}_{Guid.NewGuid():N}"
|
|
};
|
|
|
|
_logger.LogInformation("[{Channel}] Starting Unified Analyzer MQTT Client. Host: {Host}, ClientId: {ClientId}", "AnalyzerChannel", config.Host, config.ClientId);
|
|
await ConnectAsync(config);
|
|
}
|
|
|
|
public async Task StopAsync(CancellationToken cancellationToken)
|
|
{
|
|
_logger.LogInformation("[{Channel}] Stopping Unified Analyzer MQTT Client.", "AnalyzerChannel");
|
|
await DisconnectAsync();
|
|
}
|
|
|
|
protected override async Task OnConnectedAsync()
|
|
{
|
|
_logger.LogInformation("[{Channel}] Analyzer MQTT Client connected. Subscribing to topics and RPC response channels...", "AnalyzerChannel");
|
|
|
|
// Incoming Event Topics
|
|
await SubscribeAsync("services/news/#");
|
|
await SubscribeAsync("finlytic/news/raw/#");
|
|
await SubscribeAsync("finlytic/market/ticks/#");
|
|
await SubscribeAsync("services/config/updated/#");
|
|
await SubscribeAsync("services/request/health_Ping/#");
|
|
await SubscribeAsync("services/request/analyzer_TriggerManual/#");
|
|
await SubscribeAsync("finlytic/trades/closed/#");
|
|
|
|
// RPC Response Channels
|
|
await SubscribeAsync("services/response/ta_GetAnalysis/#");
|
|
await SubscribeAsync("services/response/fundamentals_Get/#");
|
|
await SubscribeAsync("services/response/sentiment_GetIsin/#");
|
|
await SubscribeAsync("services/response/trades_Get/#");
|
|
await SubscribeAsync("services/response/tr_GetLivePrice/#");
|
|
|
|
_logger.LogInformation("[{Channel}] Successfully subscribed to all event and RPC channels.", "AnalyzerChannel");
|
|
}
|
|
|
|
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("FinlyticAnalyzer", StringComparison.OrdinalIgnoreCase)
|
|
: topic.Contains("FinlyticAnalyzer", StringComparison.OrdinalIgnoreCase);
|
|
|
|
if (isForMe)
|
|
{
|
|
var correlationId = segments[^1];
|
|
string respTopic = $"services/response/health_Ping/{correlationId}";
|
|
var healthResp = new ServiceHealthResponse("FinlyticAnalyzer", "Online", DateTime.UtcNow, "Connected");
|
|
await PublishAsync(respTopic, healthResp);
|
|
if (LogCategoryFilter.IsEnabled(LogCategory.MqttHealthPing))
|
|
{
|
|
_logger.LogInformation("[{Channel}] [AnalyzerMqttClient] Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", "AnalyzerChannel", correlationId);
|
|
}
|
|
}
|
|
return;
|
|
}
|
|
|
|
if (topic.StartsWith("services/config/updated", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
if (topic.EndsWith("FinlyticAnalyzer", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
_logger.LogInformation("[{Channel}] [AnalyzerMqttClient] Received config update event for FinlyticAnalyzer.", "AnalyzerChannel");
|
|
try
|
|
{
|
|
var configUpdate = JsonSerializer.Deserialize(payloadStr, FinlyticJsonSerializerContext.Default.ServiceConfigUpdatePayload);
|
|
if (configUpdate?.Settings != null && configUpdate.Settings.Count > 0)
|
|
{
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var settingsDb = scope.ServiceProvider.GetRequiredService<ISettingsDbService>();
|
|
await settingsDb.UpdateSettingsFromDictionaryAsync(configUpdate.Settings);
|
|
_logger.LogInformation("[{Channel}] [AnalyzerMqttClient] Persisted {Count} updated settings to FinlyticAnalyzer database.", "AnalyzerChannel", configUpdate.Settings.Count);
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "[{Channel}] [AnalyzerMqttClient] Error processing MQTT config update event.", "AnalyzerChannel");
|
|
}
|
|
}
|
|
return;
|
|
}
|
|
|
|
if (topic.StartsWith("finlytic/market/ticks/"))
|
|
{
|
|
ProcessTickMessage(topic, payloadStr);
|
|
}
|
|
else if (topic.StartsWith("finlytic/news/raw/", StringComparison.OrdinalIgnoreCase) ||
|
|
topic.StartsWith("services/news/", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
await ProcessNewsMessageAsync(payloadStr, CancellationToken.None);
|
|
}
|
|
else if (topic.StartsWith("services/request/analyzer_TriggerManual/"))
|
|
{
|
|
var correlationId = topic.Split('/').Last();
|
|
await HandleManualTriggerAsync(correlationId, payloadStr, CancellationToken.None);
|
|
}
|
|
else if (topic.StartsWith("finlytic/trades/closed/"))
|
|
{
|
|
await HandleClosedTradeFeedbackAsync(payloadStr);
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "[{Channel}] Error processing incoming MQTT message on topic {Topic}", "AnalyzerChannel", topic);
|
|
}
|
|
}
|
|
|
|
private async Task HandleClosedTradeFeedbackAsync(string payloadStr)
|
|
{
|
|
try
|
|
{
|
|
var options = new JsonSerializerOptions { PropertyNameCaseInsensitive = true };
|
|
var closedDto = JsonSerializer.Deserialize<TradeProposalDto>(payloadStr, options);
|
|
|
|
if (closedDto != null && !string.IsNullOrWhiteSpace(closedDto.TradeId))
|
|
{
|
|
bool isWin = closedDto.Status.Contains("Profit", StringComparison.OrdinalIgnoreCase) ||
|
|
closedDto.Status.Contains("Win", StringComparison.OrdinalIgnoreCase);
|
|
|
|
var feedback = new TradeFeedbackRecord
|
|
{
|
|
TradeId = closedDto.TradeId,
|
|
AnalysisId = closedDto.AnalysisId,
|
|
Sector = closedDto.Sector,
|
|
Symbol = closedDto.Symbol,
|
|
Isin = closedDto.Isin,
|
|
EntryPrice = closedDto.EntryPrice,
|
|
StopLoss = closedDto.StopLoss,
|
|
TakeProfit = closedDto.TakeProfit,
|
|
IsWin = isWin,
|
|
VixRegime = closedDto.VixRegime,
|
|
VixValue = closedDto.VixValue,
|
|
CreatedAt = closedDto.CreatedAt,
|
|
ClosedAt = DateTime.UtcNow
|
|
};
|
|
|
|
string feedbackDir = System.IO.Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "data", "feedback");
|
|
if (!System.IO.Directory.Exists(feedbackDir))
|
|
{
|
|
System.IO.Directory.CreateDirectory(feedbackDir);
|
|
}
|
|
|
|
string filePath = System.IO.Path.Combine(feedbackDir, $"{closedDto.TradeId}.json");
|
|
await System.IO.File.WriteAllTextAsync(filePath, JsonSerializer.Serialize(new[] { feedback }, options));
|
|
|
|
_logger.LogInformation("[{Channel}] Processed closed trade feedback for {TradeId}. Saved to {FilePath}", "AnalyzerChannel", closedDto.TradeId, filePath);
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "[{Channel}] Error processing closed trade feedback.", "AnalyzerChannel");
|
|
}
|
|
}
|
|
|
|
private async Task HandleManualTriggerAsync(string correlationId, string payloadStr, CancellationToken cancellationToken)
|
|
{
|
|
try
|
|
{
|
|
var manualReq = JsonSerializer.Deserialize(payloadStr, FinlyticJsonSerializerContext.Default.ManualAnalysisRpcRequest);
|
|
if (manualReq == null || string.IsNullOrWhiteSpace(manualReq.Isin))
|
|
{
|
|
_logger.LogWarning("[{Channel}] Manual trigger received without valid request or ISIN.", "AnalyzerChannel");
|
|
return;
|
|
}
|
|
|
|
if (LogCategoryFilter.IsEnabled(LogCategory.AnalyzerManual))
|
|
{
|
|
_logger.LogInformation("[{Channel}] [ManualAnalyzer] [TRIGGERED] Processing rich manual trigger for ISIN '{Isin}' (Symbol: {Symbol}). CorrelationId: {CorrelationId}", "AnalyzerChannel", manualReq.Isin, manualReq.Symbol, correlationId);
|
|
}
|
|
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var dbContext = scope.ServiceProvider.GetRequiredService<AnalyzerDbContext>();
|
|
|
|
var regime = _vixTracker.GetCurrentRegime();
|
|
var currentVix = _vixTracker.GetCurrentVix();
|
|
string analysisId = Guid.NewGuid().ToString("N");
|
|
double winRate = _winRateCalculator.CalculateWinRate(manualReq.Sector, manualReq.Symbol, regime);
|
|
|
|
string riskLabel = manualReq.RiskScore > 70 ? $"Aggressiv ({manualReq.RiskScore}/100)" : (manualReq.RiskScore > 30 ? $"Balanced ({manualReq.RiskScore}/100)" : $"Konservativ ({manualReq.RiskScore}/100)");
|
|
string timeframeFormatted = $"{manualReq.MinTimeframeValue}-{manualReq.MaxTimeframeValue} {manualReq.TimeframeUnit}";
|
|
|
|
var n8nRequest = new N8nAnalysisRequestDto
|
|
{
|
|
RequestId = analysisId,
|
|
Timestamp = DateTime.UtcNow,
|
|
TriggerType = "Manual",
|
|
TargetAsset = new TargetAssetInfo
|
|
{
|
|
Symbol = manualReq.FundamentalsData?.Ticker ?? manualReq.Symbol.ToUpperInvariant(),
|
|
Name = manualReq.FundamentalsData?.CompanyName ?? manualReq.Isin.ToUpperInvariant(),
|
|
Isin = manualReq.Isin.ToUpperInvariant(),
|
|
Sector = manualReq.Sector
|
|
},
|
|
MarketContext = new MarketContextInfo
|
|
{
|
|
Vix = currentVix,
|
|
MarketRegime = regime.ToString()
|
|
},
|
|
FilterContext = new FilterContextInfo
|
|
{
|
|
ImpactScore = 1.0,
|
|
RawNewsHeadline = string.IsNullOrWhiteSpace(manualReq.Headline) ? "Manual User Trigger" : manualReq.Headline
|
|
},
|
|
UserPreferences = new UserPreferencesInfo
|
|
{
|
|
RiskScore = manualReq.RiskScore,
|
|
RiskTolerance = riskLabel,
|
|
MinTimeframeValue = manualReq.MinTimeframeValue,
|
|
MaxTimeframeValue = manualReq.MaxTimeframeValue,
|
|
TimeframeUnit = manualReq.TimeframeUnit,
|
|
TimeframeFormatted = timeframeFormatted,
|
|
InstrumentType = manualReq.InstrumentType,
|
|
UserNotes = manualReq.UserNotes
|
|
},
|
|
TradeFeedback = new TradeFeedbackInfo
|
|
{
|
|
TotalAssetTrades = 0,
|
|
AssetWinRate = winRate,
|
|
AvgReturnPercent = 0.0,
|
|
LastTradeResult = "UNKNOWN"
|
|
},
|
|
TechnicalContext = new TechnicalContextInfo
|
|
{
|
|
Rsi = manualReq.TaData?.Indicators?.LastOrDefault()?.Rsi14?.ToString("F1") ?? "N/A",
|
|
SupertrendStatus = manualReq.TaData?.Indicators?.LastOrDefault()?.SupertrendDirection ?? "NEUTRAL",
|
|
Atr = manualReq.TaData?.Indicators?.LastOrDefault()?.Atr14?.ToString("F2") ?? "N/A",
|
|
Sma50 = (double?)manualReq.TaData?.Indicators?.LastOrDefault()?.Sma50,
|
|
Sma200 = (double?)manualReq.TaData?.Indicators?.LastOrDefault()?.Sma200,
|
|
DetectedPatterns = manualReq.TaData?.Patterns?.Select(p => new PatternContextInfo
|
|
{
|
|
PatternName = p.Type,
|
|
BreakoutDirection = p.BreakoutSignal?.Direction,
|
|
TargetPrice = (double?)p.BreakoutSignal?.TargetPrice,
|
|
PotentialPercent = (double?)p.BreakoutSignal?.PotentialPercent
|
|
}).ToList() ?? new List<PatternContextInfo>()
|
|
},
|
|
SentimentContext = new SentimentContextInfo
|
|
{
|
|
AssetSentimentScore = manualReq.SentimentData?.CurrentSummary?.CompoundScore ?? 0.0,
|
|
SectorSentimentScore = 0.0,
|
|
NewsSentimentSummary = manualReq.SentimentData?.CurrentSummary?.SentimentLabel ?? "Neutral"
|
|
},
|
|
FundamentalContext = new FundamentalContextInfo
|
|
{
|
|
PeRatio = (double?)manualReq.FundamentalsData?.PeRatioTrailing,
|
|
ForwardPeRatio = (double?)manualReq.FundamentalsData?.PeRatioForward,
|
|
PegRatio = (double?)manualReq.FundamentalsData?.PegRatio,
|
|
MarketCap = (double?)manualReq.FundamentalsData?.MarketCapitalization,
|
|
DebtToEquity = (double?)manualReq.FundamentalsData?.DebtToEquity,
|
|
GrossMargin = (double?)manualReq.FundamentalsData?.GrossMargin,
|
|
NetProfitMargin = (double?)manualReq.FundamentalsData?.NetProfitMargin,
|
|
ReturnOnEquity = (double?)manualReq.FundamentalsData?.ReturnOnEquity,
|
|
DividendYield = (double?)manualReq.FundamentalsData?.DividendYield,
|
|
ShortPercentOfFloat = (double?)manualReq.FundamentalsData?.ShortPercentOfFloat,
|
|
AnalystTargetMedian = (double?)manualReq.FundamentalsData?.PriceTargetMedian,
|
|
EvToEbitda = (double?)manualReq.FundamentalsData?.EvToEbitda
|
|
}
|
|
};
|
|
|
|
var n8nResponse = await _n8nService.EvaluateAssetAsync(n8nRequest, cancellationToken);
|
|
|
|
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsDbService>();
|
|
var settings = await settingsService.GetSettingsAsync();
|
|
double minSignalScore = settings.MinSignalScore;
|
|
|
|
double confidenceScore = n8nResponse?.EvalScore > 0 ? n8nResponse.EvalScore : 0.75;
|
|
bool shouldProceed = n8nResponse != null &&
|
|
string.Equals(n8nResponse.AiDecision, "Proceed", StringComparison.OrdinalIgnoreCase) &&
|
|
(confidenceScore * 100.0) >= minSignalScore &&
|
|
winRate >= minSignalScore;
|
|
|
|
TradeProposalDto? proposalDto = null;
|
|
if (n8nResponse != null)
|
|
{
|
|
proposalDto = new TradeProposalDto
|
|
{
|
|
TradeId = "PROP-" + Guid.NewGuid().ToString("N")[..10].ToUpperInvariant(),
|
|
AnalysisId = analysisId,
|
|
EventId = analysisId,
|
|
Sector = manualReq.Sector,
|
|
Symbol = manualReq.Symbol.ToUpperInvariant(),
|
|
Isin = manualReq.Isin.ToUpperInvariant(),
|
|
CompanyName = manualReq.FundamentalsData?.CompanyName ?? manualReq.Symbol,
|
|
EntryPrice = manualReq.CurrentPrice,
|
|
SignalType = string.Equals(n8nResponse.SuggestedDirection, "Short", StringComparison.OrdinalIgnoreCase) ? "SELL" : "BUY",
|
|
Status = shouldProceed ? "Proposed" : "Rejected",
|
|
RiskTolerance = n8nResponse.SuggestedRisk,
|
|
Timeframe = timeframeFormatted,
|
|
InstrumentType = manualReq.InstrumentType,
|
|
WinRate = winRate,
|
|
VixRegime = regime,
|
|
VixValue = currentVix,
|
|
TtlMinutes = 60,
|
|
Reasoning = $"Manual n8n Evaluation ({n8nResponse.AiDecision}): {n8nResponse.AiReasoning}",
|
|
|
|
StopLoss = n8nResponse.ExecutionPlan?.StopLoss ?? 0,
|
|
TakeProfit = n8nResponse.ExecutionPlan?.TakeProfitTargets != null && n8nResponse.ExecutionPlan.TakeProfitTargets.Count > 0 ? n8nResponse.ExecutionPlan.TakeProfitTargets[0] : 0,
|
|
EntryZoneMin = n8nResponse.ExecutionPlan?.EntryZone?.Min,
|
|
EntryZoneMax = n8nResponse.ExecutionPlan?.EntryZone?.Max,
|
|
TakeProfitTargets = n8nResponse.ExecutionPlan?.TakeProfitTargets,
|
|
RiskRewardRatio = n8nResponse.ExecutionPlan?.RiskRewardRatio,
|
|
MaxLeverage = n8nResponse.ExecutionPlan?.MaxLeverage,
|
|
TechnicalRationale = n8nResponse.DetailedAnalysis?.TechnicalRationale ?? string.Empty,
|
|
FundamentalRationale = n8nResponse.DetailedAnalysis?.FundamentalRationale ?? string.Empty,
|
|
RiskWarning = n8nResponse.DetailedAnalysis?.RiskWarning ?? string.Empty,
|
|
|
|
CreatedAt = DateTime.UtcNow
|
|
};
|
|
}
|
|
|
|
var analysisEntity = new AnalysisEntity
|
|
{
|
|
AnalysisId = analysisId,
|
|
EventId = analysisId,
|
|
Sector = manualReq.Sector,
|
|
Symbol = manualReq.Symbol.ToUpperInvariant(),
|
|
Isin = manualReq.Isin.ToUpperInvariant(),
|
|
VixRegime = regime,
|
|
VixValue = currentVix,
|
|
ImpactScore = 1.0,
|
|
WinRate = winRate,
|
|
RawDataJson = JsonSerializer.Serialize(manualReq),
|
|
AiOutputJson = proposalDto != null ? JsonSerializer.Serialize(proposalDto) : "{}",
|
|
N8nResponseJson = n8nResponse != null ? JsonSerializer.Serialize(n8nResponse) : "{}",
|
|
N8nEvalScore = n8nResponse?.EvalScore ?? 0,
|
|
N8nDecision = n8nResponse?.AiDecision ?? "Rejected",
|
|
IsTradeProposed = shouldProceed,
|
|
CreatedAt = DateTime.UtcNow
|
|
};
|
|
|
|
dbContext.Analyses.Add(analysisEntity);
|
|
await dbContext.SaveChangesAsync(cancellationToken);
|
|
|
|
var responseTopic = $"services/response/analyzer_TriggerManual/{correlationId}";
|
|
var responsePayload = new ManualAnalysisResponseDto
|
|
{
|
|
AnalysisId = analysisId,
|
|
IsTradeProposed = shouldProceed,
|
|
Status = shouldProceed ? "Success" : "Rejected",
|
|
Recommendation = shouldProceed ? "RECOMMENDED" : "NOT_RECOMMENDED",
|
|
N8nResponse = n8nResponse,
|
|
Proposal = proposalDto
|
|
};
|
|
|
|
await PublishAsync(responseTopic, responsePayload);
|
|
|
|
if (proposalDto != null && shouldProceed)
|
|
{
|
|
string propTopic = $"finlytic/trades/proposed/{(string.IsNullOrWhiteSpace(manualReq.Sector) ? "general" : manualReq.Sector.ToLowerInvariant())}/{manualReq.Symbol.ToLowerInvariant()}";
|
|
await PublishAsync(propTopic, proposalDto);
|
|
_logger.LogInformation("[{Channel}] [ManualAnalyzer] [DISPATCHED] Dispatched Manual Trade Proposal {AnalysisId} to topic {Topic}", "AnalyzerChannel", analysisId, propTopic);
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "[{Channel}] Failed to handle manual trigger for correlation {CorrelationId}.", "AnalyzerChannel", correlationId);
|
|
try
|
|
{
|
|
var errorResponse = new ManualAnalysisResponseDto
|
|
{
|
|
Status = "ERROR",
|
|
Message = $"Analysis failed: {ex.Message}"
|
|
};
|
|
await PublishAsync($"services/response/analyzer_TriggerManual/{correlationId}",
|
|
JsonSerializer.Serialize(errorResponse, FinlyticJsonSerializerContext.Default.ManualAnalysisResponseDto));
|
|
}
|
|
catch (Exception pubEx)
|
|
{
|
|
_logger.LogError(pubEx, "[{Channel}] Failed to publish error response for correlation {CorrelationId}.", "AnalyzerChannel", correlationId);
|
|
}
|
|
}
|
|
}
|
|
|
|
private void ProcessTickMessage(string topic, string payloadStr)
|
|
{
|
|
if (topic.EndsWith("VIX", StringComparison.OrdinalIgnoreCase) || topic.EndsWith("^VIX", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
try
|
|
{
|
|
var tick = JsonSerializer.Deserialize(payloadStr, FinlyticJsonSerializerContext.Default.TickMessageDto);
|
|
if (tick != null && tick.Price > 0)
|
|
{
|
|
_vixTracker.UpdateVixFromTick(tick.Price);
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "[{Channel}] Failed to parse VIX tick message.", "AnalyzerChannel");
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task ProcessNewsMessageAsync(string payloadStr, CancellationToken cancellationToken)
|
|
{
|
|
var newsArticle = JsonSerializer.Deserialize(payloadStr, FinlyticJsonSerializerContext.Default.NewsArticleDto);
|
|
if (newsArticle == null) return;
|
|
|
|
var regime = _vixTracker.GetCurrentRegime();
|
|
var currentVix = _vixTracker.GetCurrentVix();
|
|
|
|
var filterResult = _filterEngine.EvaluateNews(newsArticle, regime);
|
|
if (!filterResult.Passed)
|
|
{
|
|
if (LogCategoryFilter.IsEnabled(LogCategory.AnalyzerAuto))
|
|
{
|
|
_logger.LogInformation("[{Channel}] [AutoScreener] [SKIPPED] News message skipped for ISIN '{Isin}'. Reason: {Reason}", "AnalyzerChannel", filterResult.Isin, filterResult.RejectReason);
|
|
}
|
|
return;
|
|
}
|
|
|
|
if (LogCategoryFilter.IsEnabled(LogCategory.AnalyzerAuto))
|
|
{
|
|
_logger.LogInformation("[{Channel}] [AutoScreener] [TRIGGERED] Screening market opportunity for ISIN '{Isin}'...", "AnalyzerChannel", filterResult.Isin);
|
|
}
|
|
|
|
string analysisId = Guid.NewGuid().ToString("N");
|
|
string eventId = newsArticle.Id != Guid.Empty ? newsArticle.Id.ToString() : analysisId;
|
|
string rawHeadline = newsArticle.Title ?? string.Empty;
|
|
|
|
double winRate = _winRateCalculator.CalculateWinRate(filterResult.Sector, filterResult.Symbol, regime);
|
|
|
|
int riskScore = 50;
|
|
string riskTolerance = "Balanced (50/100)";
|
|
int minTf = 4;
|
|
int maxTf = 7;
|
|
|
|
if (winRate < 45.0)
|
|
{
|
|
riskScore = 30;
|
|
riskTolerance = "Konservativ (30/100)";
|
|
minTf = 7;
|
|
maxTf = 14;
|
|
}
|
|
else if (winRate >= 65.0)
|
|
{
|
|
riskScore = 75;
|
|
riskTolerance = "Aggressiv (75/100)";
|
|
minTf = 1;
|
|
maxTf = 4;
|
|
}
|
|
|
|
TechnicalContextInfo taInfo = new();
|
|
FundamentalContextInfo fundInfo = new();
|
|
SentimentContextInfo sentInfo = new();
|
|
|
|
string resolvedSymbol = filterResult.Symbol;
|
|
string resolvedName = filterResult.Symbol;
|
|
|
|
if (newsArticle.MatchedAssets != null && newsArticle.MatchedAssets.Count > 0)
|
|
{
|
|
var firstAsset = newsArticle.MatchedAssets[0];
|
|
if (!string.IsNullOrWhiteSpace(firstAsset.Name))
|
|
{
|
|
resolvedName = firstAsset.Name;
|
|
if (resolvedSymbol == "UNKNOWN" || resolvedSymbol == filterResult.Isin)
|
|
{
|
|
resolvedSymbol = resolvedName;
|
|
}
|
|
}
|
|
}
|
|
|
|
FinlyticCore.Dtos.TechnicalAnalysis.TechnicalAnalysisDto? taResp = null;
|
|
FinlyticCore.Dtos.Fundamentals.AssetFundamentalsDto? fundResp = null;
|
|
FinlyticCore.Dtos.TechnicalAnalysis.LivePriceDto? livePriceResp = null;
|
|
|
|
try
|
|
{
|
|
if (IsConnected)
|
|
{
|
|
var isinReq = new IsinRequest(filterResult.Isin);
|
|
|
|
// Parallel RPC calls (was sequential — up to 12s latency reduced to ~3s)
|
|
var livePriceTask = SendRpcRequestAsync<FinlyticCore.Dtos.TechnicalAnalysis.LivePriceDto, IsinRequest>(
|
|
"tr_GetLivePrice", isinReq, TimeSpan.FromSeconds(5));
|
|
var taTask = SendRpcRequestAsync<FinlyticCore.Dtos.TechnicalAnalysis.TechnicalAnalysisDto, IsinRequest>(
|
|
"ta_GetAnalysis", isinReq, TimeSpan.FromSeconds(5));
|
|
var fundTask = SendRpcRequestAsync<FinlyticCore.Dtos.Fundamentals.AssetFundamentalsDto, IsinRequest>(
|
|
"fundamentals_Get", isinReq, TimeSpan.FromSeconds(5));
|
|
var sentTask = SendRpcRequestAsync<FinlyticCore.Dtos.Sentiment.IsinSentimentSummaryDto, IsinRequest>(
|
|
"sentiment_GetIsin", isinReq, TimeSpan.FromSeconds(5));
|
|
|
|
await Task.WhenAll(livePriceTask, taTask, fundTask, sentTask);
|
|
|
|
livePriceResp = livePriceTask.Result;
|
|
taResp = taTask.Result;
|
|
fundResp = fundTask.Result;
|
|
var sentResp = sentTask.Result;
|
|
|
|
if (taResp?.Indicators != null)
|
|
{
|
|
var latestIndicator = taResp.Indicators.LastOrDefault();
|
|
taInfo = new TechnicalContextInfo
|
|
{
|
|
Rsi = latestIndicator?.Rsi14?.ToString("F1") ?? "50.0",
|
|
SupertrendStatus = latestIndicator?.SupertrendDirection ?? "NEUTRAL",
|
|
Atr = latestIndicator?.Atr14?.ToString("F2") ?? "0.0",
|
|
Sma50 = (double?)latestIndicator?.Sma50,
|
|
Sma200 = (double?)latestIndicator?.Sma200,
|
|
DetectedPatterns = taResp.Patterns?.Select(p => new PatternContextInfo
|
|
{
|
|
PatternName = p.Type,
|
|
BreakoutDirection = p.BreakoutSignal?.Direction,
|
|
TargetPrice = (double?)p.BreakoutSignal?.TargetPrice,
|
|
PotentialPercent = (double?)p.BreakoutSignal?.PotentialPercent
|
|
}).ToList() ?? new List<PatternContextInfo>()
|
|
};
|
|
}
|
|
|
|
if (fundResp != null)
|
|
{
|
|
resolvedSymbol = !string.IsNullOrWhiteSpace(fundResp.Ticker) ? fundResp.Ticker : resolvedSymbol;
|
|
resolvedName = !string.IsNullOrWhiteSpace(fundResp.CompanyName) ? fundResp.CompanyName : resolvedName;
|
|
|
|
fundInfo = new FundamentalContextInfo
|
|
{
|
|
PeRatio = (double?)fundResp.PeRatioTrailing,
|
|
ForwardPeRatio = (double?)fundResp.PeRatioForward,
|
|
PegRatio = (double?)fundResp.PegRatio,
|
|
MarketCap = (double?)fundResp.MarketCapitalization,
|
|
DebtToEquity = (double?)fundResp.DebtToEquity,
|
|
GrossMargin = (double?)fundResp.GrossMargin,
|
|
NetProfitMargin = (double?)fundResp.NetProfitMargin,
|
|
ReturnOnEquity = (double?)fundResp.ReturnOnEquity,
|
|
DividendYield = (double?)fundResp.DividendYield,
|
|
ShortPercentOfFloat = (double?)fundResp.ShortPercentOfFloat,
|
|
AnalystTargetMedian = (double?)fundResp.PriceTargetMedian,
|
|
EvToEbitda = (double?)fundResp.EvToEbitda
|
|
};
|
|
}
|
|
|
|
if (sentResp != null)
|
|
{
|
|
sentInfo = new SentimentContextInfo
|
|
{
|
|
AssetSentimentScore = sentResp.CurrentSummary?.CompoundScore ?? 0.0,
|
|
SectorSentimentScore = 0.5,
|
|
NewsSentimentSummary = sentResp.CurrentSummary?.SentimentLabel ?? "Neutral"
|
|
};
|
|
}
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "[{Channel}] Failed to fetch context data for auto screener analysis.", "AnalyzerChannel");
|
|
}
|
|
|
|
var n8nRequest = new N8nAnalysisRequestDto
|
|
{
|
|
RequestId = analysisId,
|
|
Timestamp = DateTime.UtcNow,
|
|
TriggerType = "AutoScreener",
|
|
TargetAsset = new TargetAssetInfo
|
|
{
|
|
Symbol = resolvedSymbol.ToUpperInvariant(),
|
|
Name = resolvedName,
|
|
Isin = filterResult.Isin.ToUpperInvariant(),
|
|
Sector = filterResult.Sector
|
|
},
|
|
MarketContext = new MarketContextInfo
|
|
{
|
|
Vix = currentVix,
|
|
MarketRegime = regime.ToString()
|
|
},
|
|
FilterContext = new FilterContextInfo
|
|
{
|
|
ImpactScore = filterResult.ImpactScore,
|
|
RawNewsHeadline = rawHeadline
|
|
},
|
|
UserPreferences = new UserPreferencesInfo
|
|
{
|
|
RiskScore = riskScore,
|
|
RiskTolerance = riskTolerance,
|
|
MinTimeframeValue = minTf,
|
|
MaxTimeframeValue = maxTf,
|
|
TimeframeUnit = "Tage",
|
|
TimeframeFormatted = $"{minTf}-{maxTf} Tage",
|
|
InstrumentType = "KnockOut",
|
|
UserNotes = "High-Conviction Screener Mode: Evaluate underlying data for strong reliable chart moves."
|
|
},
|
|
TradeFeedback = new TradeFeedbackInfo
|
|
{
|
|
TotalAssetTrades = 0,
|
|
AssetWinRate = winRate,
|
|
AvgReturnPercent = 0.0,
|
|
LastTradeResult = "UNKNOWN"
|
|
},
|
|
TechnicalContext = taInfo,
|
|
SentimentContext = sentInfo,
|
|
FundamentalContext = fundInfo
|
|
};
|
|
|
|
var n8nResponse = await _n8nService.EvaluateAssetAsync(n8nRequest, cancellationToken);
|
|
|
|
double minSignalScore = 75.0;
|
|
using (var scope = _scopeFactory.CreateScope())
|
|
{
|
|
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsDbService>();
|
|
var settings = await settingsService.GetSettingsAsync();
|
|
minSignalScore = settings.MinSignalScore;
|
|
}
|
|
|
|
double confidenceScore = n8nResponse?.EvalScore > 0 ? n8nResponse.EvalScore : 0.75;
|
|
bool isHighConviction = n8nResponse != null &&
|
|
string.Equals(n8nResponse.AiDecision, "Proceed", StringComparison.OrdinalIgnoreCase) &&
|
|
(confidenceScore * 100.0) >= minSignalScore &&
|
|
winRate >= minSignalScore;
|
|
|
|
string finalSymbol = !string.IsNullOrWhiteSpace(resolvedSymbol) && resolvedSymbol != "UNKNOWN"
|
|
? resolvedSymbol
|
|
: (!string.IsNullOrWhiteSpace(filterResult.Symbol) && filterResult.Symbol != "UNKNOWN" ? filterResult.Symbol : filterResult.Isin);
|
|
|
|
string finalName = !string.IsNullOrWhiteSpace(resolvedName) && resolvedName != "UNKNOWN"
|
|
? resolvedName
|
|
: finalSymbol;
|
|
|
|
string marketRegion = filterResult.Isin.StartsWith("DE", StringComparison.OrdinalIgnoreCase) ? "GERMAN_EQUITIES" : "US_EQUITIES";
|
|
|
|
var supportLevels = new List<double>();
|
|
var resistanceLevels = new List<double>();
|
|
|
|
double currentPrice = (double)(livePriceResp?.CurrentPrice > 0 ? livePriceResp.CurrentPrice : (fundResp?.CurrentPrice > 0 ? fundResp.CurrentPrice : 0.0m));
|
|
if (currentPrice > 0)
|
|
{
|
|
supportLevels.Add(Math.Round(currentPrice * 0.98, 2));
|
|
supportLevels.Add(Math.Round(currentPrice * 0.95, 2));
|
|
resistanceLevels.Add(Math.Round(currentPrice * 1.03, 2));
|
|
resistanceLevels.Add(Math.Round(currentPrice * 1.06, 2));
|
|
}
|
|
|
|
if (n8nResponse?.ExecutionPlan?.EntryZone != null)
|
|
{
|
|
if (n8nResponse.ExecutionPlan.EntryZone.Min > 0) supportLevels.Insert(0, (double)n8nResponse.ExecutionPlan.EntryZone.Min);
|
|
if (n8nResponse.ExecutionPlan.EntryZone.Max > 0) resistanceLevels.Insert(0, (double)n8nResponse.ExecutionPlan.EntryZone.Max);
|
|
}
|
|
|
|
var recommendation = new AssetRecommendationDto
|
|
{
|
|
Mode = "AUTO_SCREENER",
|
|
Timestamp = DateTime.UtcNow,
|
|
RecommendedAsset = new RecommendedAssetInfo
|
|
{
|
|
Symbol = finalSymbol,
|
|
CompanyName = finalName,
|
|
Isin = filterResult.Isin,
|
|
Market = marketRegion,
|
|
Bias = string.Equals(n8nResponse?.SuggestedDirection, "Short", StringComparison.OrdinalIgnoreCase) ? "BEARISH" : "BULLISH",
|
|
ConfidenceScore = Math.Round(confidenceScore, 2),
|
|
Timeframe = !string.IsNullOrWhiteSpace(n8nResponse?.SuggestedTimeframe) ? n8nResponse.SuggestedTimeframe : "1D"
|
|
},
|
|
Rationale = new RecommendationRationaleInfo
|
|
{
|
|
PatternDetected = taInfo.DetectedPatterns?.Count > 0
|
|
? string.Join(", ", taInfo.DetectedPatterns.Select(p => p.PatternName))
|
|
: (!string.IsNullOrWhiteSpace(n8nResponse?.DetailedAnalysis?.TechnicalRationale) ? n8nResponse.DetailedAnalysis.TechnicalRationale : "Multi-Timeframe Trend & Volume Confluence"),
|
|
VixContext = $"VIX at {currentVix:F1} ({regime} volatility environment)",
|
|
KeyTechnicalLevels = new KeyTechnicalLevelsInfo
|
|
{
|
|
Support = supportLevels.Distinct().ToList(),
|
|
Resistance = resistanceLevels.Distinct().ToList()
|
|
},
|
|
Summary = !string.IsNullOrWhiteSpace(n8nReasoning(n8nResponse))
|
|
? n8nResponse!.AiReasoning
|
|
: "High conviction setup based on multi-timeframe technical confluence, sentiment, and fundamental data."
|
|
},
|
|
ActionRequired = isHighConviction ? "PROMPT_USER_FOR_MANUAL_TRADE" : "NO_ACTION"
|
|
};
|
|
|
|
using (var scope = _scopeFactory.CreateScope())
|
|
{
|
|
var dbContext = scope.ServiceProvider.GetRequiredService<AnalyzerDbContext>();
|
|
|
|
var analysisEntity = new AnalysisEntity
|
|
{
|
|
AnalysisId = analysisId,
|
|
EventId = eventId,
|
|
Sector = filterResult.Sector,
|
|
Symbol = finalSymbol,
|
|
Isin = filterResult.Isin,
|
|
VixRegime = regime,
|
|
VixValue = currentVix,
|
|
ImpactScore = filterResult.ImpactScore,
|
|
WinRate = winRate,
|
|
RawDataJson = payloadStr,
|
|
AiOutputJson = JsonSerializer.Serialize(recommendation),
|
|
N8nResponseJson = n8nResponse != null ? JsonSerializer.Serialize(n8nResponse) : "{}",
|
|
N8nEvalScore = n8nResponse?.EvalScore ?? 0,
|
|
N8nDecision = n8nResponse?.AiDecision ?? "None",
|
|
IsTradeProposed = isHighConviction,
|
|
CreatedAt = DateTime.UtcNow
|
|
};
|
|
|
|
dbContext.Analyses.Add(analysisEntity);
|
|
await dbContext.SaveChangesAsync(cancellationToken);
|
|
}
|
|
|
|
if (isHighConviction && n8nResponse != null)
|
|
{
|
|
var autoProposalDto = new TradeProposalDto
|
|
{
|
|
TradeId = "PROP-" + Guid.NewGuid().ToString("N")[..10].ToUpperInvariant(),
|
|
AnalysisId = analysisId,
|
|
EventId = eventId,
|
|
Sector = filterResult.Sector,
|
|
Symbol = finalSymbol,
|
|
Isin = filterResult.Isin,
|
|
CompanyName = finalName,
|
|
EntryPrice = (decimal)currentPrice,
|
|
SignalType = string.Equals(n8nResponse.SuggestedDirection, "Short", StringComparison.OrdinalIgnoreCase) ? "SELL" : "BUY",
|
|
Status = "Proposed",
|
|
RiskTolerance = n8nResponse.SuggestedRisk ?? "Balanced",
|
|
Timeframe = $"{minTf}-{maxTf} Tage",
|
|
InstrumentType = "KnockOut",
|
|
WinRate = winRate,
|
|
VixRegime = regime,
|
|
VixValue = currentVix,
|
|
TtlMinutes = 180,
|
|
Reasoning = n8nResponse.AiReasoning ?? "Auto-Screener High Conviction Trade",
|
|
StopLoss = n8nResponse.ExecutionPlan?.StopLoss ?? 0,
|
|
TakeProfit = n8nResponse.ExecutionPlan?.TakeProfitTargets != null && n8nResponse.ExecutionPlan.TakeProfitTargets.Count > 0 ? n8nResponse.ExecutionPlan.TakeProfitTargets[0] : 0,
|
|
EntryZoneMin = n8nResponse.ExecutionPlan?.EntryZone?.Min,
|
|
EntryZoneMax = n8nResponse.ExecutionPlan?.EntryZone?.Max,
|
|
TakeProfitTargets = n8nResponse.ExecutionPlan?.TakeProfitTargets,
|
|
RiskRewardRatio = n8nResponse.ExecutionPlan?.RiskRewardRatio,
|
|
MaxLeverage = n8nResponse.ExecutionPlan?.MaxLeverage,
|
|
TechnicalRationale = n8nResponse.DetailedAnalysis?.TechnicalRationale ?? string.Empty,
|
|
FundamentalRationale = n8nResponse.DetailedAnalysis?.FundamentalRationale ?? string.Empty,
|
|
RiskWarning = n8nResponse.DetailedAnalysis?.RiskWarning ?? string.Empty,
|
|
CreatedAt = DateTime.UtcNow
|
|
};
|
|
|
|
string propTopic = $"finlytic/trades/proposed/{(string.IsNullOrWhiteSpace(filterResult.Sector) ? "general" : filterResult.Sector.ToLowerInvariant())}/{finalSymbol.ToLowerInvariant()}";
|
|
await PublishAsync(propTopic, autoProposalDto);
|
|
_logger.LogInformation("[{Channel}] [AutoScreener] Dispatched High-Conviction Proposal {TradeId} to topic {Topic}", "AnalyzerChannel", autoProposalDto.TradeId, propTopic);
|
|
}
|
|
|
|
if (isHighConviction)
|
|
{
|
|
string recTopic = $"finlytic/recommendations/auto/{(string.IsNullOrWhiteSpace(filterResult.Sector) ? "general" : filterResult.Sector.ToLowerInvariant())}/{finalSymbol.ToLowerInvariant()}";
|
|
await PublishAsync(recTopic, recommendation);
|
|
await PublishAsync("finlytic/recommendations/auto", recommendation);
|
|
|
|
if (LogCategoryFilter.IsEnabled(LogCategory.AnalyzerAuto))
|
|
{
|
|
_logger.LogInformation("[{Channel}] [AutoScreener] [RECOMMENDED] High-Conviction Opportunity found for {Symbol} (Bias: {Bias}, Confidence: {Score:F2}). Published to {Topic}",
|
|
"AnalyzerChannel", finalSymbol, recommendation.RecommendedAsset.Bias, recommendation.RecommendedAsset.ConfidenceScore, recTopic);
|
|
}
|
|
}
|
|
else
|
|
{
|
|
if (LogCategoryFilter.IsEnabled(LogCategory.AnalyzerAuto))
|
|
{
|
|
_logger.LogInformation("[{Channel}] [AutoScreener] [DROPPED] Low-conviction signal for {Symbol} dropped (Confidence: {Score:F2}, Action: NO_ACTION)",
|
|
"AnalyzerChannel", finalSymbol, recommendation.RecommendedAsset.ConfidenceScore);
|
|
}
|
|
}
|
|
}
|
|
|
|
private static string n8nReasoning(N8nAnalysisResponseDto? resp) => resp?.AiReasoning ?? string.Empty;
|
|
} |