using FastEndpoints;
using Hangfire;
using Npgsql;
using System.Text.Json;
using KArtSell.Modules.ModelOperations.Domain;
namespace KArtSell.Host.Features.Portfolio;
///
/// VS-04 BE: Portfolio Rebalance Endpoint
/// POST /api/portfolio/{id}/rebalance - Trigger portfolio rebalancing
/// GET /api/portfolio/{id}/composition - Get current composition
///
/// Orchestrates portfolio aggregation, drift analysis, and Hangfire job scheduling
/// Idempotent by (portfolio_id, target_weights_hash, correlation_id)
///
public sealed class RebalanceRequest
{
public List TargetWeights { get; set; } = new();
public decimal DriftThreshold { get; set; } = 5;
}
public sealed class TargetWeightDto
{
public string Symbol { get; set; } = "";
public decimal TargetPercent { get; set; }
}
public sealed class RebalanceResponse
{
public Guid JobId { get; set; }
public string Status { get; set; } = "Queued";
public int EstimatedTradeCount { get; set; }
public decimal EstimatedCost { get; set; }
public string CorrelationId { get; set; } = "";
public DateTime QueuedAt { get; set; }
}
public sealed class PortfolioCompositionResponse
{
public Guid PortfolioId { get; set; }
public DateOnly SnapshotDate { get; set; }
public List Positions { get; set; } = new();
public decimal TotalValue { get; set; }
public DateTime LastUpdate { get; set; }
}
public sealed class PositionDto
{
public string Symbol { get; set; } = "";
public decimal Quantity { get; set; }
public decimal MarketPrice { get; set; }
public decimal MarketValue { get; set; }
public decimal WeightPercent { get; set; }
}
public sealed class TriggerRebalanceEndpoint : Endpoint
{
private readonly IPortfolioRebalanceService _rebalanceService;
public TriggerRebalanceEndpoint(IPortfolioRebalanceService rebalanceService)
{
_rebalanceService = rebalanceService;
}
public override void Configure()
{
Post("/api/portfolio/{portfolioId}/rebalance");
Roles("PortfolioManager");
AllowAnonymous();
}
public override async Task HandleAsync(RebalanceRequest req, CancellationToken ct)
{
var portfolioIdStr = Route("portfolioId");
if (!Guid.TryParse(portfolioIdStr, out var portfolioId))
{
ThrowError("Invalid portfolio ID");
return;
}
var correlationId = HttpContext.TraceIdentifier;
var (jobId, tradeCount, cost) = await _rebalanceService.ScheduleRebalanceAsync(
portfolioId: portfolioId,
targetWeights: req.TargetWeights.Select(w => new TargetWeight(w.Symbol, w.TargetPercent)).ToList(),
driftThreshold: req.DriftThreshold,
correlationId: correlationId,
cancellationToken: ct);
HttpContext.Response.StatusCode = StatusCodes.Status202Accepted;
HttpContext.Response.ContentType = "application/json";
await HttpContext.Response.WriteAsync(JsonSerializer.Serialize(new RebalanceResponse
{
JobId = jobId,
Status = "Queued",
EstimatedTradeCount = tradeCount,
EstimatedCost = cost,
CorrelationId = correlationId,
QueuedAt = DateTime.UtcNow,
}), ct);
}
}
public sealed class GetCompositionEndpoint : EndpointWithoutRequest
{
private readonly IPortfolioRebalanceService _rebalanceService;
public GetCompositionEndpoint(IPortfolioRebalanceService rebalanceService)
{
_rebalanceService = rebalanceService;
}
public override void Configure()
{
Get("/api/portfolio/{portfolioId}/composition");
AllowAnonymous();
}
public override async Task HandleAsync(CancellationToken ct)
{
var portfolioIdStr = Route("portfolioId");
if (!Guid.TryParse(portfolioIdStr, out var portfolioId))
{
ThrowError("Invalid portfolio ID");
return;
}
var composition = await _rebalanceService.GetCompositionAsync(portfolioId, ct);
if (composition == null)
{
ThrowError("Portfolio not found");
return;
}
HttpContext.Response.StatusCode = StatusCodes.Status200OK;
HttpContext.Response.ContentType = "application/json";
await HttpContext.Response.WriteAsync(JsonSerializer.Serialize(composition), ct);
}
}
///
/// VS-04 Application Handler: Orchestrates rebalance operations
///
public interface IPortfolioRebalanceService
{
Task<(Guid JobId, int TradeCount, decimal Cost)> ScheduleRebalanceAsync(
Guid portfolioId,
List targetWeights,
decimal driftThreshold,
string correlationId,
CancellationToken cancellationToken);
Task GetCompositionAsync(Guid portfolioId, CancellationToken cancellationToken);
}
public class PortfolioRebalanceService : IPortfolioRebalanceService
{
private readonly NpgsqlDataSource _dataSource;
private readonly IBackgroundJobClient _jobClient;
public PortfolioRebalanceService(NpgsqlDataSource dataSource, IBackgroundJobClient jobClient)
{
_dataSource = dataSource;
_jobClient = jobClient;
}
public async Task<(Guid JobId, int TradeCount, decimal Cost)> ScheduleRebalanceAsync(
Guid portfolioId,
List targetWeights,
decimal driftThreshold,
string correlationId,
CancellationToken cancellationToken)
{
var jobId = Guid.NewGuid();
// Fetch current portfolio composition
var positions = await FetchPositionsAsync(portfolioId, cancellationToken);
var portfolio = PortfolioPolicy.AggregatePortfolio(portfolioId, DateOnly.FromDateTime(DateTime.UtcNow), positions);
var currentWeights = PortfolioPolicy.CalculateCurrentWeights(portfolio);
// Analyze drift
var analysis = PortfolioPolicy.AnalyzeDrift(portfolio, targetWeights, driftThreshold);
var cost = PortfolioPolicy.EstimateRebalanceCost(analysis);
// Check idempotency
var existingJob = await CheckIdempotencyAsync(portfolioId, targetWeights, correlationId, cancellationToken);
if (existingJob.HasValue)
return (existingJob.Value, analysis.TradesRequired.Count, cost);
// Insert job record
await InsertJobRecordAsync(jobId, portfolioId, targetWeights, correlationId, cancellationToken);
// Schedule Hangfire job
_jobClient.Enqueue(j =>
j.ExecuteAsync(jobId, portfolioId, targetWeights, correlationId, CancellationToken.None));
return (jobId, analysis.TradesRequired.Count, cost);
}
public async Task GetCompositionAsync(Guid portfolioId, CancellationToken cancellationToken)
{
const string sql = """
SELECT symbol, quantity, market_price, market_value, weight_percent
FROM risk_management.portfolio_positions
WHERE portfolio_id = @portfolioId
AND published_at <= @cutoff
AND removed_at IS NULL
AND trading_date = CURRENT_DATE
ORDER BY weight_percent DESC;
""";
await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken);
await using var cmd = connection.CreateCommand();
cmd.CommandText = sql;
cmd.Parameters.AddWithValue("@portfolioId", portfolioId);
cmd.Parameters.AddWithValue("@cutoff", DateTime.UtcNow);
var positions = new List();
decimal totalValue = 0;
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken);
while (await reader.ReadAsync(cancellationToken))
{
var marketValue = reader.GetDecimal(3);
positions.Add(new PositionDto
{
Symbol = reader.GetString(0),
Quantity = reader.GetDecimal(1),
MarketPrice = reader.GetDecimal(2),
MarketValue = marketValue,
WeightPercent = reader.GetDecimal(4),
});
totalValue += marketValue;
}
if (positions.Count == 0)
return null;
return new PortfolioCompositionResponse
{
PortfolioId = portfolioId,
SnapshotDate = DateOnly.FromDateTime(DateTime.UtcNow),
Positions = positions,
TotalValue = totalValue,
LastUpdate = DateTime.UtcNow,
};
}
private async Task> FetchPositionsAsync(Guid portfolioId, CancellationToken cancellationToken)
{
const string sql = """
SELECT symbol, quantity, market_price, cost_basis_per_unit
FROM risk_management.portfolio_positions
WHERE portfolio_id = @portfolioId
AND published_at <= @cutoff
AND removed_at IS NULL
AND trading_date = CURRENT_DATE;
""";
await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken);
await using var cmd = connection.CreateCommand();
cmd.CommandText = sql;
cmd.Parameters.AddWithValue("@portfolioId", portfolioId);
cmd.Parameters.AddWithValue("@cutoff", DateTime.UtcNow);
var positions = new List();
await using var reader = await cmd.ExecuteReaderAsync(cancellationToken);
while (await reader.ReadAsync(cancellationToken))
{
positions.Add(new Position(
Symbol: reader.GetString(0),
Quantity: reader.GetDecimal(1),
MarketPrice: reader.GetDecimal(2),
CostBasisPerUnit: reader.IsDBNull(3) ? 0 : reader.GetDecimal(3)));
}
return positions;
}
private async Task CheckIdempotencyAsync(
Guid portfolioId,
List targetWeights,
string correlationId,
CancellationToken cancellationToken)
{
var weightsHash = HashTargetWeights(targetWeights);
const string sql = """
SELECT job_id FROM risk_management.rebalance_jobs
WHERE portfolio_id = @portfolioId
AND target_weights_hash = @hash
AND correlation_id = @correlationId
AND status IN ('Running', 'Completed')
LIMIT 1;
""";
await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken);
await using var cmd = connection.CreateCommand();
cmd.CommandText = sql;
cmd.Parameters.AddWithValue("@portfolioId", portfolioId);
cmd.Parameters.AddWithValue("@hash", weightsHash);
cmd.Parameters.AddWithValue("@correlationId", correlationId);
var result = await cmd.ExecuteScalarAsync(cancellationToken);
return result is Guid jobId ? jobId : null;
}
private async Task InsertJobRecordAsync(
Guid jobId,
Guid portfolioId,
List targetWeights,
string correlationId,
CancellationToken cancellationToken)
{
var weightsHash = HashTargetWeights(targetWeights);
const string sql = """
INSERT INTO risk_management.rebalance_jobs
(job_id, portfolio_id, target_weights_hash, correlation_id, status, requested_at, requested_by)
VALUES (@jobId, @portfolioId, @hash, @correlationId, 'Queued', CURRENT_TIMESTAMP, 'API');
""";
await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken);
await using var cmd = connection.CreateCommand();
cmd.CommandText = sql;
cmd.Parameters.AddWithValue("@jobId", jobId);
cmd.Parameters.AddWithValue("@portfolioId", portfolioId);
cmd.Parameters.AddWithValue("@hash", weightsHash);
cmd.Parameters.AddWithValue("@correlationId", correlationId);
await cmd.ExecuteNonQueryAsync(cancellationToken);
}
private static string HashTargetWeights(List weights)
{
var sorted = weights.OrderBy(w => w.Symbol).Select(w => $"{w.Symbol}:{w.TargetPercent}");
var hash = string.Join("|", sorted);
return Convert.ToHexString(System.Text.Encoding.UTF8.GetBytes(hash));
}
}
///
/// VS-04 ASYNC: Rebalance Job Handler (Hangfire)
///
public interface IPortfolioRebalanceJob
{
Task ExecuteAsync(Guid jobId, Guid portfolioId, List targetWeights, string correlationId, CancellationToken ct);
}
public class PortfolioRebalanceJobHandler : IPortfolioRebalanceJob
{
private readonly NpgsqlDataSource _dataSource;
public PortfolioRebalanceJobHandler(NpgsqlDataSource dataSource)
{
_dataSource = dataSource;
}
public async Task ExecuteAsync(Guid jobId, Guid portfolioId, List targetWeights, string correlationId, CancellationToken ct)
{
var startTime = DateTime.UtcNow;
try
{
await UpdateJobStatusAsync(jobId, "Running", null, null, ct);
// Simulate rebalance execution (real implementation: call trading API)
await Task.Delay(1000, ct);
// Mark complete
var duration = (int)(DateTime.UtcNow - startTime).TotalSeconds;
await UpdateJobStatusAsync(jobId, "Completed", duration, null, ct);
// Publish event
await PublishRebalancedEventAsync(jobId, portfolioId, correlationId, ct);
}
catch (Exception ex)
{
await UpdateJobStatusAsync(jobId, "Failed", null, ex.Message, ct);
throw;
}
}
private async Task UpdateJobStatusAsync(Guid jobId, string status, int? durationSeconds = null, string? errorMessage = null, CancellationToken ct = default)
{
const string sql = """
UPDATE risk_management.rebalance_jobs
SET status = @status, completed_at = CASE WHEN @status IN ('Completed', 'Failed') THEN CURRENT_TIMESTAMP ELSE NULL END,
duration_seconds = @duration, last_error_message = @error, updated_at = CURRENT_TIMESTAMP
WHERE job_id = @jobId;
""";
await using var connection = await _dataSource.OpenConnectionAsync(ct);
await using var cmd = connection.CreateCommand();
cmd.CommandText = sql;
cmd.Parameters.AddWithValue("@jobId", jobId);
cmd.Parameters.AddWithValue("@status", status);
cmd.Parameters.AddWithValue("@duration", durationSeconds ?? (object)DBNull.Value);
cmd.Parameters.AddWithValue("@error", errorMessage ?? (object)DBNull.Value);
await cmd.ExecuteNonQueryAsync(ct);
}
private async Task PublishRebalancedEventAsync(Guid jobId, Guid portfolioId, string correlationId, CancellationToken ct)
{
const string sql = """
INSERT INTO shared.outbox (aggregate_id, event_type, payload, published_at, correlation_id)
VALUES (@aggregateId, 'PortfolioRebalanced', @payload, CURRENT_TIMESTAMP, @correlationId);
""";
var payload = JsonSerializer.Serialize(new
{
eventType = "PortfolioRebalanced",
portfolioId,
jobId,
rebalancedAt = DateTime.UtcNow,
});
await using var connection = await _dataSource.OpenConnectionAsync(ct);
await using var cmd = connection.CreateCommand();
cmd.CommandText = sql;
cmd.Parameters.AddWithValue("@aggregateId", portfolioId);
cmd.Parameters.AddWithValue("@payload", payload);
cmd.Parameters.AddWithValue("@correlationId", correlationId);
await cmd.ExecuteNonQueryAsync(ct);
}
}