refactor(bot): update paper trading models, broker integration, background services, and test project
This commit is contained in:
+301
-106
@@ -1,15 +1,26 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Text.Json;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using FinlyticBot.Services;
|
||||
using FinlyticCore.Dtos;
|
||||
using FinlyticCore.Dtos.Bot;
|
||||
using FinlyticCore.Dtos.Settings;
|
||||
using FinlyticCore.Dtos.TechnicalAnalysis;
|
||||
using FinlyticCore.Dtos.Trading;
|
||||
using FinlyticCore.Models;
|
||||
using FinlyticCore.Models.Trades;
|
||||
using FinlyticCore.Services;
|
||||
using FinlyticCore.Util;
|
||||
using FinlyticBot.Database;
|
||||
using FinlyticBot.Database.Entities;
|
||||
using FinlyticBot.Services.Alpaca;
|
||||
using FinlyticBot.Services.Consumers;
|
||||
using FinlyticBot.Services.Execution;
|
||||
using FinlyticBot.Services.Ledger;
|
||||
using FinlyticBot.Services.Mqtt;
|
||||
using FinlyticBot.Settings;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
@@ -17,176 +28,360 @@ using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace FinlyticBot.Util;
|
||||
|
||||
public class BotMqttClient : ManagedMqttClient, IHostedService
|
||||
public class BotMqttClient : ManagedMqttClient, IHostedService, IBotRpcClient
|
||||
{
|
||||
private readonly IConfiguration _configuration;
|
||||
private readonly IServiceScopeFactory _scopeFactory;
|
||||
private readonly ILogger<BotMqttClient> _logger;
|
||||
private readonly IFinlyticLogger<BotMqttClient> _finlyticLogger;
|
||||
|
||||
public BotMqttClient(
|
||||
ILogger<BotMqttClient> logger,
|
||||
IConfiguration configuration,
|
||||
IServiceScopeFactory scopeFactory,
|
||||
ILogger<BotMqttClient> logger) : base(logger)
|
||||
IFinlyticLogger<BotMqttClient> finlyticLogger) : base(logger)
|
||||
{
|
||||
_logger = logger;
|
||||
_configuration = configuration;
|
||||
_scopeFactory = scopeFactory;
|
||||
_logger = logger;
|
||||
_finlyticLogger = finlyticLogger;
|
||||
}
|
||||
|
||||
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_bot")}_{Guid.NewGuid():N}"
|
||||
};
|
||||
var config = MqttConfiguration.FromConfiguration(_configuration, "FinlyticBot");
|
||||
|
||||
_logger.LogInformation("Starting FinlyticBot MQTT Client. Host: {Host}, ClientId: {ClientId}", config.Host, config.ClientId);
|
||||
_logger.LogInformation("Starting FinlyticBot MQTT client. Host: {Host}, ClientId: {ClientId}", config.Host, config.ClientId);
|
||||
await _finlyticLogger.LogInfoAsync(BotSettingKeys.MqttChannel, "[BotMqttClient] Starting FinlyticBot MQTT client. Host: {Host}, ClientId: {ClientId}", config.Host, config.ClientId);
|
||||
await ConnectAsync(config);
|
||||
}
|
||||
|
||||
public async Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_logger.LogInformation("Stopping FinlyticBot MQTT Client.");
|
||||
_logger.LogInformation("Stopping FinlyticBot MQTT client.");
|
||||
// Broadcast via IFinlyticLogger too (see OnConnectedAsync's doc comment) so the live console shows a
|
||||
// clean "stopped" line instead of just silently going quiet.
|
||||
await _finlyticLogger.LogInfoAsync(BotSettingKeys.MqttChannel, "[BotMqttClient] Stopping FinlyticBot MQTT client.");
|
||||
await DisconnectAsync();
|
||||
}
|
||||
|
||||
protected override async Task OnConnectedAsync()
|
||||
{
|
||||
_logger.LogInformation("FinlyticBot MQTT Client connected. Subscribing to topics...");
|
||||
_logger.LogInformation("FinlyticBot MQTT client connected. Registering RPC endpoints...");
|
||||
|
||||
await SubscribeAsync("services/events/trades/proposal");
|
||||
await SubscribeAsync("finlytic/trades/proposed/#");
|
||||
await SubscribeAsync("services/events/analyzer/trade_proposed");
|
||||
await SubscribeAsync("services/request/bot_Settings_GetAll/#");
|
||||
await SubscribeAsync("services/request/bot_Settings_Update/#");
|
||||
await SubscribeAsync("services/request/health_Ping/#");
|
||||
await SubscribeAsync(MqttTopics.ResponseWildcard);
|
||||
await SubscribeRpcAsync<object, BotStatusDto>(MqttTopics.RequestFilter(MqttTopics.Channels.BotGetStatus), HandleGetStatusRpcAsync);
|
||||
await SubscribeRpcAsync<object, List<BotTradeOrderDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.BotGetPositions), HandleGetPositionsRpcAsync);
|
||||
await SubscribeRpcAsync<object, AccountSummaryDto>(MqttTopics.RequestFilter(MqttTopics.Channels.BotGetSummary), HandleGetSummaryRpcAsync);
|
||||
await SubscribeRpcAsync<ExecuteProposalRequest, BotTradeOrderDto?>(MqttTopics.RequestFilter(MqttTopics.Channels.BotExecuteProposal), HandleExecuteProposalRpcAsync);
|
||||
await SubscribeRpcAsync<object, PanicCloseResultDto>(MqttTopics.RequestFilter(MqttTopics.Channels.BotPanicClose), HandlePanicCloseRpcAsync);
|
||||
await SubscribeRpcAsync<object, List<DynamicSettingDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.BotSettingsGetAll), HandleSettingsGetAllRpcAsync);
|
||||
await SubscribeRpcAsync<Dictionary<string, object?>, List<DynamicSettingDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.BotSettingsUpdate), HandleSettingsUpdateRpcAsync);
|
||||
await SubscribeAsync<object>(MqttTopics.RequestFilter(MqttTopics.Channels.HealthPing), HandleHealthPingRpcAsync);
|
||||
|
||||
FinlyticCore.Services.FinlyticLogBroadcaster.OnLogPublished = async (logDto) =>
|
||||
// Subscribe to Engine Proposals
|
||||
await SubscribeAsync(MqttTopics.EngineProposalsCreated);
|
||||
|
||||
FinlyticLogBroadcaster.OnLogPublished = async (logDto) =>
|
||||
{
|
||||
if (IsConnected && string.Equals(logDto.ServiceName, "FinlyticBot", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
await PublishAsync("finlytic/logs/FinlyticBot", logDto);
|
||||
await PublishAsync(MqttTopics.Logs("FinlyticBot"), logDto);
|
||||
}
|
||||
};
|
||||
|
||||
_logger.LogInformation("Successfully subscribed to FinlyticBot event and RPC channels.");
|
||||
await _finlyticLogger.LogInfoAsync(BotSettingKeys.MqttChannel, "[BotMqttClient] FinlyticBot MQTT client connected. Registering RPC endpoints...");
|
||||
}
|
||||
|
||||
protected override async Task OnMessageReceivedAsync(string topic, string payloadStr)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(topic) || string.IsNullOrWhiteSpace(payloadStr)) return;
|
||||
|
||||
try
|
||||
{
|
||||
if (topic.Contains("health_Ping", StringComparison.OrdinalIgnoreCase))
|
||||
if (topic.Equals(MqttTopics.EngineProposalsCreated, StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
var segments = topic.Split('/');
|
||||
bool isForMe = segments.Length >= 5
|
||||
? segments[3].Equals("FinlyticBot", StringComparison.OrdinalIgnoreCase)
|
||||
: topic.Contains("FinlyticBot", StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
if (isForMe)
|
||||
var proposal = JsonSerializer.Deserialize<TradeProposalDto>(payloadStr, DefaultJsonOptions);
|
||||
if (proposal != null && EngineProposalConsumerBackgroundService.OnProposalReceived != null)
|
||||
{
|
||||
string correlationId = segments[^1];
|
||||
string respTopic = $"services/response/health_Ping/{correlationId}";
|
||||
await PublishAsync(respTopic, new ServiceHealthResponse("FinlyticBot", "Online", DateTime.UtcNow, "Connected"));
|
||||
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<BotMqttClient>>();
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.HealthPingChannel,
|
||||
"[FinlyticBot] Responded to live health_Ping RPC [CorrelationId: {CorrelationId}].", correlationId);
|
||||
EngineProposalConsumerBackgroundService.OnProposalReceived(proposal);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if (topic.StartsWith("services/request/bot_Settings_GetAll/", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
var correlationId = topic.Split('/')[^1];
|
||||
await HandleSettingsGetAllAsync(correlationId);
|
||||
return;
|
||||
}
|
||||
|
||||
if (topic.StartsWith("services/request/bot_Settings_Update/", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
var correlationId = topic.Split('/')[^1];
|
||||
await HandleSettingsUpdateAsync(correlationId, payloadStr);
|
||||
return;
|
||||
}
|
||||
|
||||
if (topic.Equals("services/events/trades/proposal", StringComparison.OrdinalIgnoreCase) ||
|
||||
topic.StartsWith("finlytic/trades/proposed/", StringComparison.OrdinalIgnoreCase) ||
|
||||
topic.Equals("services/events/analyzer/trade_proposed", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
await HandleTradeProposalEventAsync(payloadStr);
|
||||
return;
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Error processing incoming MQTT message on topic {Topic}", topic);
|
||||
_logger.LogError(ex, "[BotMqttClient] Error handling incoming proposal on topic {Topic}", topic);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task HandleTradeProposalEventAsync(string payloadStr)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(payloadStr)) return;
|
||||
|
||||
TradeProposalDto? proposal = null;
|
||||
try
|
||||
{
|
||||
proposal = JsonSerializer.Deserialize<TradeProposalDto>(payloadStr, FinlyticJsonSerializerContext.Default.TradeProposalDto);
|
||||
}
|
||||
catch
|
||||
{
|
||||
proposal = JsonSerializer.Deserialize<TradeProposalDto>(payloadStr);
|
||||
}
|
||||
|
||||
if (proposal == null) return;
|
||||
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var executionService = scope.ServiceProvider.GetRequiredService<IBotOrderExecutionService>();
|
||||
await executionService.ProcessTradeProposalAsync(proposal);
|
||||
}
|
||||
|
||||
private async Task HandleSettingsGetAllAsync(string correlationId)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
||||
var settings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
|
||||
|
||||
var responseTopic = $"services/response/bot_Settings_GetAll/{correlationId}";
|
||||
await PublishAsync(responseTopic, settings);
|
||||
}
|
||||
|
||||
private async Task HandleSettingsUpdateAsync(string correlationId, string payload)
|
||||
private async Task<BotStatusDto> HandleGetStatusRpcAsync(object? _, string correlationId)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<BotDbContext>();
|
||||
var alpacaService = scope.ServiceProvider.GetRequiredService<IAlpacaTradingService>();
|
||||
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
||||
|
||||
Dictionary<string, object?>? updates = null;
|
||||
int active = await db.Positions.CountAsync(p => p.Status == BotPositionStatus.Active || p.Status == BotPositionStatus.BreakEvenTriggered);
|
||||
bool autoExec = await settingsService.GetSettingAsync(BotSettingKeys.EnableAutoExecution);
|
||||
int maxPositions = await settingsService.GetSettingAsync(BotSettingKeys.MaxConcurrentPositions);
|
||||
decimal riskPerTradePercent = await settingsService.GetSettingAsync(BotSettingKeys.RiskPerTradePercent);
|
||||
|
||||
// MinCompositeScore is owned by FinlyticEngine (EngineSettingKeys.MinCompositeScore ==
|
||||
// "Engine.MinCompositeScore", read as `decimal` there — see TradeLifecycleService). FinlyticBot has no
|
||||
// project reference to FinlyticEngine (and is out of scope for adding one here), but both services share
|
||||
// the same dynamic settings store, so the raw string key is read directly instead of duplicating/inventing
|
||||
// a Bot-local setting. Default mirrors EngineSettingKeys' own documented default.
|
||||
decimal minCompositeScoreRaw = await settingsService.GetSettingAsync("Engine.MinCompositeScore", 75.0m);
|
||||
|
||||
// The bot worker process itself is always running once started; "IsRunning" from the client's
|
||||
// perspective means "is the bot actively acting on proposals", which is exactly what
|
||||
// EngineProposalConsumerBackgroundService gates on before calling IBotOrderExecutor. So IsRunning
|
||||
// is deliberately the same flag as AutoExecutionEnabled rather than a separate process-alive flag.
|
||||
bool isRunning = autoExec;
|
||||
|
||||
// Synthetic broker is always available (in-process ledger); Alpaca is only listed once real
|
||||
// credentials are configured (IAlpacaTradingService.IsConfigured), mirroring the venue selection
|
||||
// logic in BotOrderExecutor.
|
||||
string venuesActive = alpacaService.IsConfigured
|
||||
? $"{BotExecutionVenue.SyntheticPaperBroker}, {BotExecutionVenue.AlpacaPaperTrading}"
|
||||
: BotExecutionVenue.SyntheticPaperBroker.ToString();
|
||||
|
||||
return new BotStatusDto(
|
||||
IsRunning: isRunning,
|
||||
AutoExecutionEnabled: autoExec,
|
||||
ActivePositionsCount: active,
|
||||
MaxPositions: maxPositions,
|
||||
RiskPerTradePercent: riskPerTradePercent,
|
||||
MinCompositeScore: (int)Math.Round(minCompositeScoreRaw, MidpointRounding.AwayFromZero),
|
||||
VenuesActive: venuesActive
|
||||
);
|
||||
}
|
||||
|
||||
private async Task<List<BotTradeOrderDto>> HandleGetPositionsRpcAsync(object? _, string correlationId)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<BotDbContext>();
|
||||
|
||||
var positions = await db.Positions
|
||||
.AsNoTracking()
|
||||
.Where(p => p.Status == BotPositionStatus.Active || p.Status == BotPositionStatus.BreakEvenTriggered || p.Status == BotPositionStatus.Tp1Hit)
|
||||
.OrderByDescending(p => p.OpenedAtUtc)
|
||||
.ToListAsync();
|
||||
|
||||
return positions.Select(BotOrderExecutor.MapEntityToDto).ToList();
|
||||
}
|
||||
|
||||
private async Task<AccountSummaryDto> HandleGetSummaryRpcAsync(object? _, string correlationId)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var syntheticBroker = scope.ServiceProvider.GetRequiredService<ISyntheticPaperBroker>();
|
||||
return await syntheticBroker.GetSummaryAsync();
|
||||
}
|
||||
|
||||
private async Task<BotTradeOrderDto?> HandleExecuteProposalRpcAsync(ExecuteProposalRequest? req, string correlationId)
|
||||
{
|
||||
if (req == null) return null;
|
||||
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var executor = scope.ServiceProvider.GetRequiredService<IBotOrderExecutor>();
|
||||
var logger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<BotMqttClient>>();
|
||||
|
||||
// Look up the existing proposal by its actual GUID via the engine_GetProposals RPC channel — the
|
||||
// same channel EngineController/UserTradesController already use for proposal listings.
|
||||
// (req.ProposalId is a proposal GUID, not an ISIN; a previous version of this method sent it to
|
||||
// engine_EvaluateIsin as if it were one, which started a full re-evaluation of a nonsensical "ISIN"
|
||||
// and could never resolve a proposal — see Bug A of the security review.)
|
||||
// A generous limit is used because the proposal the caller wants to execute may not be among the
|
||||
// most recently created handful if several proposals were generated in quick succession.
|
||||
List<TradeProposalDto>? proposals;
|
||||
try
|
||||
{
|
||||
updates = JsonSerializer.Deserialize<Dictionary<string, object?>>(payload);
|
||||
proposals = await SendRpcRequestAsync<List<TradeProposalDto>, GetTradeProposalsRequest>(
|
||||
MqttTopics.Channels.EngineGetProposals,
|
||||
new GetTradeProposalsRequest(OnlyActive: true, Limit: 200),
|
||||
TimeSpan.FromSeconds(5)
|
||||
);
|
||||
}
|
||||
catch
|
||||
catch (Exception ex)
|
||||
{
|
||||
var list = JsonSerializer.Deserialize<List<DynamicSettingDto>>(payload);
|
||||
if (list != null)
|
||||
await logger.LogWarningAsync(BotSettingKeys.BotChannel, ex,
|
||||
"[BotMqttClient] RPC failure while fetching proposals from FinlyticEngine to resolve proposal {ProposalId} for execution [CorrelationId: {CorrelationId}]",
|
||||
req.ProposalId, correlationId);
|
||||
return null;
|
||||
}
|
||||
|
||||
if (proposals == null)
|
||||
{
|
||||
await logger.LogWarningAsync(BotSettingKeys.BotChannel,
|
||||
"[BotMqttClient] FinlyticEngine did not respond to engine_GetProposals in time while resolving proposal {ProposalId} for execution [CorrelationId: {CorrelationId}]",
|
||||
req.ProposalId, correlationId);
|
||||
return null;
|
||||
}
|
||||
|
||||
var proposal = proposals.FirstOrDefault(p => p.ProposalId == req.ProposalId);
|
||||
if (proposal == null)
|
||||
{
|
||||
await logger.LogWarningAsync(BotSettingKeys.BotChannel,
|
||||
"[BotMqttClient] Proposal {ProposalId} was not found among FinlyticEngine's active proposals (expired or invalid) [CorrelationId: {CorrelationId}]",
|
||||
req.ProposalId, correlationId);
|
||||
return null;
|
||||
}
|
||||
|
||||
await logger.LogInfoAsync(BotSettingKeys.BotChannel,
|
||||
"[BotMqttClient] Executing manual paper trade for proposal {Id} [CorrelationId: {CorrelationId}]", req.ProposalId, correlationId);
|
||||
return await executor.ExecuteProposalAsync(proposal, req.PreferredVenue, req.CustomQuantity);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Emergency-closes every currently open paper-trading position (Status in Active, BreakEvenTriggered,
|
||||
/// Tp1Hit, Tp2Hit — the terminal statuses Closed/StoppedOut/KnockedOut/Canceled are, by definition,
|
||||
/// already not open and Pending is not currently assigned by any code path).
|
||||
/// <para>
|
||||
/// Synthetic ledger positions are closed unconditionally via <see cref="ISyntheticPaperBroker"/> — there
|
||||
/// is no external broker to confirm with, so the internal ledger IS the authority.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Alpaca positions are the safety-critical case: this handler NEVER marks an Alpaca position as closed
|
||||
/// unless <see cref="IAlpacaTradingService.ClosePositionAsync"/> returns successfully (i.e. Alpaca's REST
|
||||
/// API confirmed it accepted the liquidation order). If Alpaca is not configured, or the broker call
|
||||
/// throws, the position is left completely untouched in the database and is counted in
|
||||
/// <see cref="PanicCloseResultDto.SkippedCount"/> rather than silently reported as closed
|
||||
/// (Rules.md §4 — a false-positive "closed" on a real broker position would be fatal).
|
||||
/// </para>
|
||||
/// </summary>
|
||||
private async Task<PanicCloseResultDto> HandlePanicCloseRpcAsync(object? _, string correlationId)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<BotDbContext>();
|
||||
var syntheticBroker = scope.ServiceProvider.GetRequiredService<ISyntheticPaperBroker>();
|
||||
var alpacaService = scope.ServiceProvider.GetRequiredService<IAlpacaTradingService>();
|
||||
var logger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<BotMqttClient>>();
|
||||
|
||||
var openStatuses = new[]
|
||||
{
|
||||
BotPositionStatus.Active,
|
||||
BotPositionStatus.BreakEvenTriggered,
|
||||
BotPositionStatus.Tp1Hit,
|
||||
BotPositionStatus.Tp2Hit
|
||||
};
|
||||
|
||||
var positions = await db.Positions
|
||||
.Where(p => openStatuses.Contains(p.Status))
|
||||
.OrderBy(p => p.OpenedAtUtc)
|
||||
.ToListAsync();
|
||||
|
||||
await logger.LogWarningAsync(BotSettingKeys.BotChannel,
|
||||
"[BotMqttClient] PANIC CLOSE triggered: attempting to close {Count} open position(s) [CorrelationId: {CorrelationId}]",
|
||||
positions.Count, correlationId);
|
||||
|
||||
var closedOrders = new List<BotTradeOrderDto>();
|
||||
int skippedCount = 0;
|
||||
|
||||
foreach (var pos in positions)
|
||||
{
|
||||
if (pos.Venue == BotExecutionVenue.SyntheticPaperBroker)
|
||||
{
|
||||
updates = new Dictionary<string, object?>();
|
||||
foreach (var item in list) updates[item.Key] = item.Value;
|
||||
// No external broker to confirm with — the internal ledger is the authority for its own
|
||||
// positions, so this closes unconditionally at the last synced price (same direction-aware
|
||||
// realized P&L formula ISyntheticPaperBroker already uses elsewhere for consistency).
|
||||
var closed = await syntheticBroker.ClosePositionAsync(pos.Id, pos.CurrentPrice, BotPositionStatus.Closed);
|
||||
closedOrders.Add(BotOrderExecutor.MapEntityToDto(closed));
|
||||
continue;
|
||||
}
|
||||
|
||||
// Alpaca venue: only ever mark closed after the broker confirms the liquidation.
|
||||
if (!alpacaService.IsConfigured)
|
||||
{
|
||||
await logger.LogWarningAsync(BotSettingKeys.BotChannel,
|
||||
"[BotMqttClient] PANIC CLOSE SKIPPED Alpaca position {PositionId} ({Symbol}): Alpaca is not configured. Position left OPEN in the ledger — manual intervention required.",
|
||||
pos.Id, pos.Symbol);
|
||||
skippedCount++;
|
||||
continue;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
var closeResult = await alpacaService.ClosePositionAsync(pos.Symbol);
|
||||
|
||||
// Alpaca confirmed acceptance of the liquidation order — only now is it safe to persist
|
||||
// the position as closed. AverageFillPrice can still be null immediately after submission
|
||||
// (e.g. outside market hours); fall back to the last synced price rather than fabricating one.
|
||||
decimal exitPrice = closeResult.AverageFillPrice ?? pos.CurrentPrice;
|
||||
pos.Status = BotPositionStatus.Closed;
|
||||
pos.ClosedAtUtc = DateTime.UtcNow;
|
||||
pos.CurrentPrice = exitPrice;
|
||||
pos.LastSyncAtUtc = DateTime.UtcNow;
|
||||
pos.RealizedPnlEur = CalculateRealizedPnl(pos, exitPrice);
|
||||
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
await logger.LogInfoAsync(BotSettingKeys.BotChannel,
|
||||
"[BotMqttClient] PANIC CLOSE confirmed by Alpaca for position {PositionId} ({Symbol}): Order {OrderId} (Status: {Status}), exit {Exit:F2} €.",
|
||||
pos.Id, pos.Symbol, closeResult.OrderId, closeResult.OrderStatus, exitPrice);
|
||||
|
||||
closedOrders.Add(BotOrderExecutor.MapEntityToDto(pos));
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// The broker call itself failed (not configured mid-flight, network/HTTP error, or Alpaca
|
||||
// rejected the request) — the position is left completely untouched, exactly as it was
|
||||
// before this handler ran. It must NOT be counted as closed.
|
||||
await logger.LogWarningAsync(BotSettingKeys.BotChannel, ex,
|
||||
"[BotMqttClient] PANIC CLOSE FAILED for Alpaca position {PositionId} ({Symbol}): broker call did not confirm liquidation. Position left OPEN in the ledger — manual intervention required.",
|
||||
pos.Id, pos.Symbol);
|
||||
skippedCount++;
|
||||
}
|
||||
}
|
||||
|
||||
return new PanicCloseResultDto(closedOrders.Count, skippedCount, closedOrders);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Direction-aware realized P&L for a position closed at <paramref name="exitPrice"/>, mirroring the
|
||||
/// formula <see cref="ISyntheticPaperBroker"/> already uses for its own (non-knock-out) closes, so both
|
||||
/// venues report P&L consistently. Only used for the Alpaca panic-close path here — the synthetic path
|
||||
/// delegates to <see cref="ISyntheticPaperBroker.ClosePositionAsync"/> directly, which computes its own.
|
||||
/// </summary>
|
||||
private static decimal CalculateRealizedPnl(BotPositionEntity pos, decimal exitPrice)
|
||||
{
|
||||
decimal pnl = pos.Direction == SignalDirection.Buy
|
||||
? ((exitPrice - pos.AverageBuyIn) * pos.Quantity) - pos.TotalFeesEur
|
||||
: ((pos.AverageBuyIn - exitPrice) * pos.Quantity) - pos.TotalFeesEur;
|
||||
|
||||
return Math.Round(pnl, 2);
|
||||
}
|
||||
|
||||
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(BotSettingKeys) });
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
var currentSettings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
|
||||
var responseTopic = $"services/response/bot_Settings_Update/{correlationId}";
|
||||
await PublishAsync(responseTopic, currentSettings);
|
||||
return await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(BotSettingKeys) });
|
||||
}
|
||||
|
||||
private async Task HandleHealthPingRpcAsync(object? _, string topic, string correlationId)
|
||||
{
|
||||
if (topic.Contains("FinlyticBot", StringComparison.OrdinalIgnoreCase) || !topic.Contains("/", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
string respTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.HealthPing, correlationId);
|
||||
await PublishAsync(respTopic, new ServiceHealthResponse("FinlyticBot", "Online", DateTime.UtcNow, "Connected"));
|
||||
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var logger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<BotMqttClient>>();
|
||||
await logger.LogInfoAsync(BotSettingKeys.HealthPingChannel,
|
||||
"[FinlyticBot] Responded to health_Ping RPC [CorrelationId: {CorrelationId}]", correlationId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,36 +0,0 @@
|
||||
using FinlyticCore.Models.Settings;
|
||||
|
||||
namespace FinlyticBot.Util;
|
||||
|
||||
public static class SettingKeys
|
||||
{
|
||||
// --- Logging-Kanäle ---
|
||||
public static readonly SettingKey<bool> BotChannel = new("Logging.Channel.Bot", true);
|
||||
public static readonly SettingKey<bool> MqttChannel = new("Logging.Channel.MQTT", true);
|
||||
public static readonly SettingKey<bool> HealthPingChannel = new("Logging.Channel.Health", true);
|
||||
|
||||
// --- Master Bot Control ---
|
||||
public static readonly SettingKey<bool> IsEnabled = new("Bot.IsEnabled", false);
|
||||
|
||||
// --- Filter & Zulassungskriterien ---
|
||||
public static readonly SettingKey<double> MinCrv = new("Bot.MinCrv", 1.50);
|
||||
public static readonly SettingKey<double> MinWinRate = new("Bot.MinWinRate", 65.0);
|
||||
public static readonly SettingKey<double> MaxVixThreshold = new("Bot.MaxVixThreshold", 25.0);
|
||||
public static readonly SettingKey<double> MaxEntryDeviationPercent = new("Bot.MaxEntryDeviationPercent", 0.75);
|
||||
|
||||
// --- Risiko-Management & Positionsgrößen ---
|
||||
public static readonly SettingKey<double> RiskPerTradePercent = new("Bot.RiskPerTradePercent", 1.0);
|
||||
public static readonly SettingKey<double> MaxSinglePositionCap = new("Bot.MaxSinglePositionCap", 5000.0);
|
||||
public static readonly SettingKey<int> MaxOpenTrades = new("Bot.MaxOpenTrades", 5);
|
||||
public static readonly SettingKey<double> DailyLossLimitPercent = new("Bot.DailyLossLimitPercent", 3.0);
|
||||
public static readonly SettingKey<int> MaxConsecutiveLosses = new("Bot.MaxConsecutiveLosses", 3);
|
||||
|
||||
// --- Order Execution Strategie ---
|
||||
public static readonly SettingKey<string> TakeProfitMode = new("Bot.TakeProfitMode", "Split50_50"); // "TP1_Only", "TP2_Only", "Split50_50"
|
||||
public static readonly SettingKey<string> ExecutionOrderType = new("Bot.ExecutionOrderType", "Limit"); // "Limit", "Market"
|
||||
|
||||
// --- Alpaca API Konfiguration (Optional live im UI überschreibbar) ---
|
||||
public static readonly SettingKey<string> AlpacaKeyId = new("Alpaca.KeyId", "");
|
||||
public static readonly SettingKey<string> AlpacaSecretKey = new("Alpaca.SecretKey", "");
|
||||
public static readonly SettingKey<bool> AlpacaIsPaper = new("Alpaca.IsPaper", true);
|
||||
}
|
||||
Reference in New Issue
Block a user