feat(bot): add FinlyticBot autonomous paper trading microservice with Alpaca Markets API integration
This commit is contained in:
@@ -0,0 +1,208 @@
|
||||
using System;
|
||||
using System.Linq;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Alpaca.Markets;
|
||||
using FinlyticBot.Database;
|
||||
using FinlyticBot.Util;
|
||||
using FinlyticCore.Services;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
|
||||
namespace FinlyticBot.Services;
|
||||
|
||||
public class AlpacaWebSocketMonitorWorker : BackgroundService
|
||||
{
|
||||
private readonly IServiceScopeFactory _scopeFactory;
|
||||
private readonly ISettingsService _settingsService;
|
||||
private readonly BotMqttClient _mqttClient;
|
||||
private readonly IConfiguration _configuration;
|
||||
private readonly IFinlyticLogger<AlpacaWebSocketMonitorWorker> _finlyticLogger;
|
||||
|
||||
private IAlpacaStreamingClient? _streamingClient;
|
||||
|
||||
public AlpacaWebSocketMonitorWorker(
|
||||
IServiceScopeFactory scopeFactory,
|
||||
ISettingsService settingsService,
|
||||
BotMqttClient mqttClient,
|
||||
IConfiguration configuration,
|
||||
IFinlyticLogger<AlpacaWebSocketMonitorWorker> finlyticLogger)
|
||||
{
|
||||
_scopeFactory = scopeFactory;
|
||||
_settingsService = settingsService;
|
||||
_mqttClient = mqttClient;
|
||||
_configuration = configuration;
|
||||
_finlyticLogger = finlyticLogger;
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
await _finlyticLogger.LogInfoAsync(SettingKeys.BotChannel,
|
||||
"[AlpacaWebSocketMonitor] Starting Alpaca Trade Update Stream Monitor...");
|
||||
|
||||
while (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
try
|
||||
{
|
||||
string keyId = await _settingsService.GetSettingAsync(SettingKeys.AlpacaKeyId, stoppingToken);
|
||||
if (string.IsNullOrWhiteSpace(keyId))
|
||||
{
|
||||
keyId = _configuration["Alpaca:KeyId"] ?? _configuration["Alpaca__KeyId"] ?? string.Empty;
|
||||
}
|
||||
|
||||
string secretKey = await _settingsService.GetSettingAsync(SettingKeys.AlpacaSecretKey, stoppingToken);
|
||||
if (string.IsNullOrWhiteSpace(secretKey))
|
||||
{
|
||||
secretKey = _configuration["Alpaca:SecretKey"] ?? _configuration["Alpaca__SecretKey"] ?? string.Empty;
|
||||
}
|
||||
|
||||
bool isPaper = await _settingsService.GetSettingAsync(SettingKeys.AlpacaIsPaper, stoppingToken);
|
||||
|
||||
if (string.IsNullOrWhiteSpace(keyId) || string.IsNullOrWhiteSpace(secretKey))
|
||||
{
|
||||
await Task.Delay(TimeSpan.FromSeconds(15), stoppingToken);
|
||||
continue;
|
||||
}
|
||||
|
||||
var environment = isPaper ? Alpaca.Markets.Environments.Paper : Alpaca.Markets.Environments.Live;
|
||||
_streamingClient = environment.GetAlpacaStreamingClient(new SecretKey(keyId, secretKey));
|
||||
|
||||
_streamingClient.OnTradeUpdate += HandleTradeUpdate;
|
||||
|
||||
var authStatus = await _streamingClient.ConnectAndAuthenticateAsync(stoppingToken);
|
||||
await _finlyticLogger.LogInfoAsync(SettingKeys.BotChannel,
|
||||
"[AlpacaWebSocketMonitor] Connected & Authenticated to Alpaca Streaming WS. Status: {Status}", authStatus.ToString());
|
||||
|
||||
// Keep connection alive until cancellation
|
||||
var tcs = new TaskCompletionSource<bool>();
|
||||
using (stoppingToken.Register(() => tcs.TrySetResult(true)))
|
||||
{
|
||||
await tcs.Task;
|
||||
}
|
||||
|
||||
await _streamingClient.DisconnectAsync(CancellationToken.None);
|
||||
}
|
||||
catch (Exception ex) when (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
await _finlyticLogger.LogWarningAsync(SettingKeys.BotChannel, ex,
|
||||
"[AlpacaWebSocketMonitor] Streaming WebSocket disconnected. Retrying in 10s...");
|
||||
await Task.Delay(TimeSpan.FromSeconds(10), stoppingToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void HandleTradeUpdate(ITradeUpdate update)
|
||||
{
|
||||
_ = Task.Run(async () =>
|
||||
{
|
||||
try
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<BotDbContext>();
|
||||
|
||||
var order = update.Order;
|
||||
if (order == null) return;
|
||||
|
||||
var trade = await dbContext.ExecutedPaperTrades
|
||||
.FirstOrDefaultAsync(t => t.AlpacaOrderId == order.OrderId);
|
||||
|
||||
if (trade == null)
|
||||
{
|
||||
// Check if it's a child order (SL / TP) of an existing trade
|
||||
trade = await dbContext.ExecutedPaperTrades
|
||||
.Where(t => t.Symbol == order.Symbol && t.Status == "Filled")
|
||||
.OrderByDescending(t => t.PlacedAt)
|
||||
.FirstOrDefaultAsync();
|
||||
}
|
||||
|
||||
if (trade == null) return;
|
||||
|
||||
if (update.Event == TradeEvent.Fill)
|
||||
{
|
||||
decimal fillPrice = update.Price ?? order.AverageFillPrice ?? trade.SignalEntryPrice;
|
||||
trade.ActualFillPrice = fillPrice;
|
||||
trade.Status = "Filled";
|
||||
trade.FilledAt = DateTime.UtcNow;
|
||||
|
||||
if (trade.SignalEntryPrice > 0)
|
||||
{
|
||||
trade.SlippagePercent = Math.Round(((fillPrice - trade.SignalEntryPrice) / trade.SignalEntryPrice) * 100m, 3);
|
||||
}
|
||||
|
||||
await _finlyticLogger.LogInfoAsync(SettingKeys.BotChannel,
|
||||
"[AlpacaTradeUpdate] Order FILLED for {Symbol}: FillPrice=${Price:F2} (Signal: ${SigPrice:F2}, Slippage: {Slip:F3}%)",
|
||||
trade.Symbol, fillPrice, trade.SignalEntryPrice, trade.SlippagePercent ?? 0m);
|
||||
}
|
||||
else if (update.Event == TradeEvent.PartialFill)
|
||||
{
|
||||
trade.Status = "PartiallyFilled";
|
||||
}
|
||||
else if (update.Event == TradeEvent.Canceled || update.Event == TradeEvent.Expired || update.Event == TradeEvent.Rejected)
|
||||
{
|
||||
trade.Status = update.Event.ToString();
|
||||
trade.ClosedAt = DateTime.UtcNow;
|
||||
}
|
||||
else if (update.Event == TradeEvent.Stopped || update.Event == TradeEvent.Calculated)
|
||||
{
|
||||
// Position closed by Stop Loss or Take Profit
|
||||
trade.Status = "Closed";
|
||||
trade.ClosedAt = DateTime.UtcNow;
|
||||
|
||||
decimal exitPrice = update.Price ?? trade.ActualFillPrice ?? trade.SignalEntryPrice;
|
||||
if (trade.ActualFillPrice.HasValue && update.Price.HasValue)
|
||||
{
|
||||
exitPrice = update.Price.Value;
|
||||
decimal diff = trade.Side == "BUY" ? (exitPrice - trade.ActualFillPrice.Value) : (trade.ActualFillPrice.Value - exitPrice);
|
||||
trade.RealizedPnl = diff * trade.Quantity;
|
||||
if (trade.ActualFillPrice.Value > 0)
|
||||
{
|
||||
trade.RealizedPnlPercent = Math.Round((diff / trade.ActualFillPrice.Value) * 100m, 2);
|
||||
}
|
||||
}
|
||||
|
||||
await _finlyticLogger.LogInfoAsync(SettingKeys.BotChannel,
|
||||
"[AlpacaTradeUpdate] Position CLOSED for {Symbol}: Realized PnL: ${Pnl:F2} ({Pct:F2}%)",
|
||||
trade.Symbol, trade.RealizedPnl, trade.RealizedPnlPercent ?? 0m);
|
||||
|
||||
// Publish Closed Trade to MQTT for WinRate calibration & AI feedback loop
|
||||
bool isWin = trade.RealizedPnl > 0;
|
||||
var feedbackDto = new FinlyticCore.Models.Trades.TradeProposalDto
|
||||
{
|
||||
TradeId = trade.TradeId,
|
||||
Symbol = trade.Symbol,
|
||||
Isin = trade.Isin,
|
||||
CompanyName = trade.CompanyName,
|
||||
EntryPrice = trade.SignalEntryPrice,
|
||||
ActualEntryPrice = trade.ActualFillPrice,
|
||||
CurrentPrice = exitPrice,
|
||||
StopLoss = trade.StopLossPrice,
|
||||
TakeProfit = trade.TakeProfitPrice1,
|
||||
Status = isWin ? "Closed_Profit" : "Closed_Loss",
|
||||
SignalType = trade.Side,
|
||||
WinRate = trade.WinRate,
|
||||
PnlAbsolute = trade.RealizedPnl,
|
||||
PnlPercent = trade.RealizedPnlPercent,
|
||||
CloseReason = isWin ? "TakeProfit_Hit" : "StopLoss_Hit",
|
||||
UserExitTimestamp = trade.ClosedAt,
|
||||
CreatedAt = trade.PlacedAt
|
||||
};
|
||||
|
||||
await _mqttClient.PublishAsync($"finlytic/trades/closed/{trade.TradeId}", feedbackDto);
|
||||
await _finlyticLogger.LogInfoAsync(SettingKeys.BotChannel,
|
||||
"[AlpacaTradeUpdate] Dispatched closed trade feedback event to MQTT for {TradeId} (Win: {IsWin})",
|
||||
trade.TradeId, isWin);
|
||||
}
|
||||
|
||||
trade.UpdatedAt = DateTime.UtcNow;
|
||||
await dbContext.SaveChangesAsync();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_ = _finlyticLogger.LogErrorAsync(SettingKeys.BotChannel, ex,
|
||||
"[AlpacaWebSocketMonitor] Error processing TradeUpdate event.");
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user