feat(analyzer): dynamic settings, IFinlyticLogger, live log streaming, and EF migration
This commit is contained in:
@@ -8,9 +8,11 @@ using FinlyticAnalyzer.Database;
|
||||
using FinlyticAnalyzer.Entities;
|
||||
using FinlyticAnalyzer.Services;
|
||||
using FinlyticCore.Dtos;
|
||||
using FinlyticCore.Dtos.Settings;
|
||||
using FinlyticCore.Models;
|
||||
using FinlyticCore.Models.Analyzer;
|
||||
using FinlyticCore.Models.Trades;
|
||||
using FinlyticCore.Services;
|
||||
using FinlyticCore.Util;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
@@ -64,19 +66,19 @@ public class AnalyzerMqttClient : ManagedMqttClient, IHostedService
|
||||
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);
|
||||
_logger.LogInformation("Starting Unified Analyzer MQTT Client. Host: {Host}, ClientId: {ClientId}", config.Host, config.ClientId);
|
||||
await ConnectAsync(config);
|
||||
}
|
||||
|
||||
public async Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_logger.LogInformation("[{Channel}] Stopping Unified Analyzer MQTT Client.", "AnalyzerChannel");
|
||||
_logger.LogInformation("Stopping Unified Analyzer MQTT Client.");
|
||||
await DisconnectAsync();
|
||||
}
|
||||
|
||||
protected override async Task OnConnectedAsync()
|
||||
{
|
||||
_logger.LogInformation("[{Channel}] Analyzer MQTT Client connected. Subscribing to topics and RPC response channels...", "AnalyzerChannel");
|
||||
_logger.LogInformation("Analyzer MQTT Client connected. Subscribing to topics and RPC response channels...");
|
||||
|
||||
// Incoming Event Topics
|
||||
await SubscribeAsync("services/news/#");
|
||||
@@ -85,16 +87,29 @@ public class AnalyzerMqttClient : ManagedMqttClient, IHostedService
|
||||
await SubscribeAsync("services/config/updated/#");
|
||||
await SubscribeAsync("services/request/health_Ping/#");
|
||||
await SubscribeAsync("services/request/analyzer_TriggerManual/#");
|
||||
await SubscribeAsync("services/request/analyzer_settings_GetAll/#");
|
||||
await SubscribeAsync("services/request/analyzer_settings_Update/#");
|
||||
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/sentiment_Analyze/#");
|
||||
await SubscribeAsync("services/response/trades_Get/#");
|
||||
await SubscribeAsync("services/response/tr_GetLivePrice/#");
|
||||
await SubscribeAsync("services/response/events_GetByMonth/#");
|
||||
await SubscribeAsync("services/response/events_GetAll/#");
|
||||
|
||||
_logger.LogInformation("[{Channel}] Successfully subscribed to all event and RPC channels.", "AnalyzerChannel");
|
||||
FinlyticCore.Services.FinlyticLogBroadcaster.OnLogPublished = async (logDto) =>
|
||||
{
|
||||
if (IsConnected && string.Equals(logDto.ServiceName, "FinlyticAnalyzer", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
await PublishAsync("finlytic/logs/FinlyticAnalyzer", logDto);
|
||||
}
|
||||
};
|
||||
|
||||
_logger.LogInformation("Successfully subscribed to all event and RPC channels.");
|
||||
}
|
||||
|
||||
protected override async Task OnMessageReceivedAsync(string topic, string payloadStr)
|
||||
@@ -114,10 +129,9 @@ public class AnalyzerMqttClient : ManagedMqttClient, IHostedService
|
||||
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);
|
||||
}
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<AnalyzerMqttClient>>();
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.HealthPingChannel, "[AnalyzerMqttClient] Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", correlationId);
|
||||
}
|
||||
return;
|
||||
}
|
||||
@@ -126,21 +140,22 @@ public class AnalyzerMqttClient : ManagedMqttClient, IHostedService
|
||||
{
|
||||
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);
|
||||
var settings = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
||||
var dict = configUpdate.Settings.ToDictionary(k => k.Key, v => (object?)v.Value);
|
||||
await settings.UpdateSettingsAsync(dict);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "[{Channel}] [AnalyzerMqttClient] Error processing MQTT config update event.", "AnalyzerChannel");
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<AnalyzerMqttClient>>();
|
||||
await finlyticLogger.LogErrorAsync(SettingKeys.AnalyzerChannel, ex, "[AnalyzerMqttClient] Error processing MQTT config update event.");
|
||||
}
|
||||
}
|
||||
return;
|
||||
@@ -160,6 +175,16 @@ public class AnalyzerMqttClient : ManagedMqttClient, IHostedService
|
||||
var correlationId = topic.Split('/').Last();
|
||||
await HandleManualTriggerAsync(correlationId, payloadStr, CancellationToken.None);
|
||||
}
|
||||
else if (topic.StartsWith("services/request/analyzer_settings_GetAll", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
var correlationId = topic.Split('/').Last();
|
||||
await HandleSettingsGetAllAsync(correlationId);
|
||||
}
|
||||
else if (topic.StartsWith("services/request/analyzer_settings_Update", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
var correlationId = topic.Split('/').Last();
|
||||
await HandleSettingsUpdateAsync(payloadStr, correlationId);
|
||||
}
|
||||
else if (topic.StartsWith("finlytic/trades/closed/"))
|
||||
{
|
||||
await HandleClosedTradeFeedbackAsync(payloadStr);
|
||||
@@ -167,12 +192,80 @@ public class AnalyzerMqttClient : ManagedMqttClient, IHostedService
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "[{Channel}] Error processing incoming MQTT message on topic {Topic}", "AnalyzerChannel", topic);
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<AnalyzerMqttClient>>();
|
||||
await finlyticLogger.LogErrorAsync(SettingKeys.AnalyzerChannel, ex, "[AnalyzerMqttClient] Error processing incoming MQTT message on topic {Topic}", topic);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task HandleSettingsGetAllAsync(string correlationId)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<AnalyzerMqttClient>>();
|
||||
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
||||
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticAnalyzer] [Settings_GetAll] Retrieving all dynamic settings via reflection [CorrelationId: {CorrelationId}]", correlationId);
|
||||
try
|
||||
{
|
||||
var settings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
|
||||
var responseTopic = $"services/response/analyzer_settings_GetAll/{correlationId}";
|
||||
|
||||
await PublishAsync(responseTopic, settings);
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticAnalyzer] [Settings_GetAll] Published {Count} settings to '{ResponseTopic}'", settings.Count, responseTopic);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
await finlyticLogger.LogErrorAsync(SettingKeys.MqttChannel, ex, "[FinlyticAnalyzer] [Settings_GetAll] Failed to retrieve settings.");
|
||||
}
|
||||
}
|
||||
|
||||
private async Task HandleSettingsUpdateAsync(string payload, string correlationId)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(payload)) return;
|
||||
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<AnalyzerMqttClient>>();
|
||||
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
||||
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticAnalyzer] [Settings_Update] Processing settings update RPC [CorrelationId: {CorrelationId}]", correlationId);
|
||||
try
|
||||
{
|
||||
Dictionary<string, object?>? updates = null;
|
||||
try
|
||||
{
|
||||
updates = JsonSerializer.Deserialize<Dictionary<string, object?>>(payload);
|
||||
}
|
||||
catch
|
||||
{
|
||||
var list = JsonSerializer.Deserialize<List<DynamicSettingDto>>(payload);
|
||||
if (list != null)
|
||||
{
|
||||
updates = new Dictionary<string, object?>();
|
||||
foreach (var item in list) updates[item.Key] = item.Value;
|
||||
}
|
||||
}
|
||||
|
||||
if (updates != null && updates.Count > 0)
|
||||
{
|
||||
await settingsService.UpdateSettingsAsync(updates);
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticAnalyzer] [Settings_Update] Successfully updated {Count} settings in database and cache.", updates.Count);
|
||||
}
|
||||
|
||||
var currentSettings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
|
||||
var responseTopic = $"services/response/analyzer_settings_Update/{correlationId}";
|
||||
await PublishAsync(responseTopic, currentSettings);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
await finlyticLogger.LogErrorAsync(SettingKeys.MqttChannel, ex, "[FinlyticAnalyzer] [Settings_Update] Failed to update settings.");
|
||||
}
|
||||
}
|
||||
|
||||
private async Task HandleClosedTradeFeedbackAsync(string payloadStr)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<AnalyzerMqttClient>>();
|
||||
|
||||
try
|
||||
{
|
||||
var options = new JsonSerializerOptions { PropertyNameCaseInsensitive = true };
|
||||
@@ -209,32 +302,31 @@ public class AnalyzerMqttClient : ManagedMqttClient, IHostedService
|
||||
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);
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.AnalyzerChannel, "[AnalyzerMqttClient] Processed closed trade feedback for {TradeId}. Saved to {FilePath}", closedDto.TradeId, filePath);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "[{Channel}] Error processing closed trade feedback.", "AnalyzerChannel");
|
||||
await finlyticLogger.LogErrorAsync(SettingKeys.AnalyzerChannel, ex, "[AnalyzerMqttClient] Error processing closed trade feedback.");
|
||||
}
|
||||
}
|
||||
|
||||
private async Task HandleManualTriggerAsync(string correlationId, string payloadStr, CancellationToken cancellationToken)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<AnalyzerMqttClient>>();
|
||||
|
||||
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");
|
||||
await finlyticLogger.LogWarningAsync(SettingKeys.AnalyzerChannel, "[AnalyzerMqttClient] Manual trigger received without valid request or ISIN.");
|
||||
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);
|
||||
}
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.AnalyzerChannel, "[ManualAnalyzer] [TRIGGERED] Processing rich manual trigger for ISIN '{Isin}' (Symbol: {Symbol}). CorrelationId: {CorrelationId}", manualReq.Isin, manualReq.Symbol, correlationId);
|
||||
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<AnalyzerDbContext>();
|
||||
|
||||
var regime = _vixTracker.GetCurrentRegime();
|
||||
@@ -325,9 +417,8 @@ public class AnalyzerMqttClient : ManagedMqttClient, IHostedService
|
||||
|
||||
var n8nResponse = await _n8nService.EvaluateAssetAsync(n8nRequest, cancellationToken);
|
||||
|
||||
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsDbService>();
|
||||
var settings = await settingsService.GetSettingsAsync();
|
||||
double minSignalScore = settings.MinSignalScore;
|
||||
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
||||
double minSignalScore = await settingsService.GetSettingAsync(SettingKeys.MinWinRateThreshold, cancellationToken);
|
||||
|
||||
double dynamicWinRate = _winRateCalculator.CalculateDynamicWinRate(
|
||||
manualReq.Sector,
|
||||
@@ -422,12 +513,12 @@ public class AnalyzerMqttClient : ManagedMqttClient, IHostedService
|
||||
{
|
||||
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);
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.AnalyzerChannel, "[ManualAnalyzer] [DISPATCHED] Dispatched Manual Trade Proposal {AnalysisId} to topic {Topic}", analysisId, propTopic);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "[{Channel}] Failed to handle manual trigger for correlation {CorrelationId}.", "AnalyzerChannel", correlationId);
|
||||
await finlyticLogger.LogErrorAsync(SettingKeys.AnalyzerChannel, ex, "[AnalyzerMqttClient] Failed to handle manual trigger for correlation {CorrelationId}.", correlationId);
|
||||
try
|
||||
{
|
||||
var errorResponse = new ManualAnalysisResponseDto
|
||||
@@ -439,7 +530,7 @@ public class AnalyzerMqttClient : ManagedMqttClient, IHostedService
|
||||
}
|
||||
catch (Exception pubEx)
|
||||
{
|
||||
_logger.LogError(pubEx, "[{Channel}] Failed to publish error response for correlation {CorrelationId}.", "AnalyzerChannel", correlationId);
|
||||
await finlyticLogger.LogErrorAsync(SettingKeys.AnalyzerChannel, pubEx, "[AnalyzerMqttClient] Failed to publish error response for correlation {CorrelationId}.", correlationId);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -458,13 +549,16 @@ public class AnalyzerMqttClient : ManagedMqttClient, IHostedService
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "[{Channel}] Failed to parse VIX tick message.", "AnalyzerChannel");
|
||||
_logger.LogWarning(ex, "Failed to parse VIX tick message.");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ProcessNewsMessageAsync(string payloadStr, CancellationToken cancellationToken)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<AnalyzerMqttClient>>();
|
||||
|
||||
var newsArticle = JsonSerializer.Deserialize(payloadStr, FinlyticJsonSerializerContext.Default.NewsArticleDto);
|
||||
if (newsArticle == null) return;
|
||||
|
||||
@@ -474,17 +568,11 @@ public class AnalyzerMqttClient : ManagedMqttClient, IHostedService
|
||||
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);
|
||||
}
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.AnalyzerChannel, "[AutoScreener] [SKIPPED] News message skipped for ISIN '{Isin}'. Reason: {Reason}", filterResult.Isin, filterResult.RejectReason);
|
||||
return;
|
||||
}
|
||||
|
||||
if (LogCategoryFilter.IsEnabled(LogCategory.AnalyzerAuto))
|
||||
{
|
||||
_logger.LogInformation("[{Channel}] [AutoScreener] [TRIGGERED] Screening market opportunity for ISIN '{Isin}'...", "AnalyzerChannel", filterResult.Isin);
|
||||
}
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.AnalyzerChannel, "[AutoScreener] [TRIGGERED] Screening market opportunity for ISIN '{Isin}'...", filterResult.Isin);
|
||||
|
||||
string analysisId = Guid.NewGuid().ToString("N");
|
||||
string eventId = newsArticle.Id != Guid.Empty ? newsArticle.Id.ToString() : analysisId;
|
||||
@@ -543,7 +631,6 @@ public class AnalyzerMqttClient : ManagedMqttClient, IHostedService
|
||||
{
|
||||
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>(
|
||||
@@ -606,7 +693,6 @@ public class AnalyzerMqttClient : ManagedMqttClient, IHostedService
|
||||
if (sentResp != null)
|
||||
{
|
||||
double compound = sentResp.CurrentSummary?.CompoundScore ?? 0.0;
|
||||
// FinBERT compound score is in range [-1.0, +1.0]. Normalize to [0.0, 1.0] for AI prompt context
|
||||
double normalizedScore = Math.Clamp((compound + 1.0) / 2.0, 0.0, 1.0);
|
||||
|
||||
sentInfo = new SentimentContextInfo
|
||||
@@ -620,7 +706,7 @@ public class AnalyzerMqttClient : ManagedMqttClient, IHostedService
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "[{Channel}] Failed to fetch context data for auto screener analysis.", "AnalyzerChannel");
|
||||
await finlyticLogger.LogWarningAsync(SettingKeys.AnalyzerChannel, ex, "[AnalyzerMqttClient] Failed to fetch context data for auto screener analysis.");
|
||||
}
|
||||
|
||||
var n8nRequest = new N8nAnalysisRequestDto
|
||||
@@ -670,13 +756,8 @@ public class AnalyzerMqttClient : ManagedMqttClient, IHostedService
|
||||
|
||||
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;
|
||||
}
|
||||
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
||||
double minSignalScore = await settingsService.GetSettingAsync(SettingKeys.MinWinRateThreshold, cancellationToken);
|
||||
|
||||
double confidenceScore = n8nResponse?.EvalScore > 0 ? n8nResponse.EvalScore : 0.75;
|
||||
bool isHighConviction = n8nResponse != null &&
|
||||
@@ -752,47 +833,44 @@ public class AnalyzerMqttClient : ManagedMqttClient, IHostedService
|
||||
sentimentScore: sentResp?.CurrentSummary?.CompoundScore,
|
||||
signalType: n8nResponse?.SuggestedDirection ?? "BUY");
|
||||
|
||||
using (var scope = _scopeFactory.CreateScope())
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<AnalyzerDbContext>();
|
||||
|
||||
bool hasRecentProposal = await dbContext.Analyses.AnyAsync(a =>
|
||||
a.Isin == filterResult.Isin &&
|
||||
a.IsTradeProposed &&
|
||||
a.CreatedAt >= DateTime.UtcNow.AddHours(-4),
|
||||
cancellationToken);
|
||||
|
||||
if (hasRecentProposal && isHighConviction)
|
||||
{
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<AnalyzerDbContext>();
|
||||
|
||||
bool hasRecentProposal = await dbContext.Analyses.AnyAsync(a =>
|
||||
a.Isin == filterResult.Isin &&
|
||||
a.IsTradeProposed &&
|
||||
a.CreatedAt >= DateTime.UtcNow.AddHours(-4),
|
||||
cancellationToken);
|
||||
|
||||
if (hasRecentProposal && isHighConviction)
|
||||
{
|
||||
_logger.LogInformation("[{Channel}] [AutoScreener] Asset {Symbol} ({Isin}) already has an active trade proposal in the last 4 hours. Skipping duplicate trade proposal generation.",
|
||||
"AnalyzerChannel", finalSymbol, filterResult.Isin);
|
||||
isHighConviction = false;
|
||||
}
|
||||
|
||||
var analysisEntity = new AnalysisEntity
|
||||
{
|
||||
AnalysisId = analysisId,
|
||||
EventId = eventId,
|
||||
Sector = filterResult.Sector,
|
||||
Symbol = finalSymbol,
|
||||
Isin = filterResult.Isin,
|
||||
VixRegime = regime,
|
||||
VixValue = currentVix,
|
||||
ImpactScore = filterResult.ImpactScore,
|
||||
WinRate = dynamicWinRate,
|
||||
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);
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.AnalyzerChannel, "[AutoScreener] Asset {Symbol} ({Isin}) already has an active trade proposal in the last 4 hours. Skipping duplicate trade proposal generation.",
|
||||
finalSymbol, filterResult.Isin);
|
||||
isHighConviction = false;
|
||||
}
|
||||
|
||||
var analysisEntity = new AnalysisEntity
|
||||
{
|
||||
AnalysisId = analysisId,
|
||||
EventId = eventId,
|
||||
Sector = filterResult.Sector,
|
||||
Symbol = finalSymbol,
|
||||
Isin = filterResult.Isin,
|
||||
VixRegime = regime,
|
||||
VixValue = currentVix,
|
||||
ImpactScore = filterResult.ImpactScore,
|
||||
WinRate = dynamicWinRate,
|
||||
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
|
||||
@@ -830,7 +908,7 @@ public class AnalyzerMqttClient : ManagedMqttClient, IHostedService
|
||||
|
||||
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);
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.AnalyzerChannel, "[AutoScreener] Dispatched High-Conviction Proposal {TradeId} to topic {Topic}", autoProposalDto.TradeId, propTopic);
|
||||
}
|
||||
|
||||
if (isHighConviction)
|
||||
@@ -839,19 +917,13 @@ public class AnalyzerMqttClient : ManagedMqttClient, IHostedService
|
||||
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);
|
||||
}
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.AnalyzerChannel, "[AutoScreener] [RECOMMENDED] High-Conviction Opportunity found for {Symbol} (Bias: {Bias}, Confidence: {Score:F2}). Published to {Topic}",
|
||||
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);
|
||||
}
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.AnalyzerChannel, "[AutoScreener] [DROPPED] Low-conviction signal for {Symbol} dropped (Confidence: {Score:F2}, Action: NO_ACTION)",
|
||||
finalSymbol, recommendation.RecommendedAsset.ConfidenceScore);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user