diff --git a/src/KArtSell.Host/Features/MarketData/VS03_IngestionEndpoint.cs b/src/KArtSell.Host/Features/MarketData/VS03_IngestionEndpoint.cs new file mode 100644 index 00000000..3d08e1f1 --- /dev/null +++ b/src/KArtSell.Host/Features/MarketData/VS03_IngestionEndpoint.cs @@ -0,0 +1,281 @@ +using FastEndpoints; +using Hangfire; +using Npgsql; +using System.Text.Json; +using KArtSell.Modules.ModelOperations.Domain; + +namespace KArtSell.Host.Features.MarketData; + +/// +/// VS-03 BE: Market Data Ingestion Endpoints +/// POST /api/market/ingest - Trigger data ingestion +/// GET /api/market/ingest/{jobId} - Check job status +/// +/// Schedules market data collection from KRX/OpenDart +/// - Idempotent by date range + data source +/// - Returns 202 Accepted (async processing) +/// - Audit trail with correlation ID +/// + +public sealed class IngestionRequest +{ + public string DataSource { get; set; } = "KRX"; // "KRX", "OpenDart", "Stub" + public string FromDate { get; set; } = ""; // "2026-01-01" + public string ToDate { get; set; } = ""; // "2026-12-31" +} + +public sealed class IngestionResponse +{ + public Guid JobId { get; set; } + public string Status { get; set; } = "Queued"; + public int ExpectedRowCount { get; set; } + public DateTime QueuedAt { get; set; } +} + +public sealed class IngestionStatusResponse +{ + public Guid JobId { get; set; } + public string Status { get; set; } = "Running"; + public int RowsProcessed { get; set; } + public int RowsFailed { get; set; } + public int RowsSkipped { get; set; } + public DateTime? CompletedAt { get; set; } + public int? DurationSeconds { get; set; } + public string? ErrorMessage { get; set; } +} + +public sealed class TriggerIngestionEndpoint : Endpoint +{ + private readonly IMarketDataIngestionService _ingestionService; + + public TriggerIngestionEndpoint(IMarketDataIngestionService ingestionService) + { + _ingestionService = ingestionService; + } + + public override void Configure() + { + Post("/api/market/ingest"); + Roles("DataAdmin"); + AllowAnonymous(); + } + + public override async Task HandleAsync(IngestionRequest req, CancellationToken ct) + { + if (!DateOnly.TryParse(req.FromDate, out var fromDate) || + !DateOnly.TryParse(req.ToDate, out var toDate)) + { + ThrowError("Invalid date format. Use YYYY-MM-DD"); + } + + if (fromDate > toDate) + { + ThrowError("FromDate must be <= ToDate"); + } + + var correlationId = HttpContext.TraceIdentifier; + + var (jobId, expectedCount) = await _ingestionService.ScheduleIngestionAsync( + dataSource: req.DataSource, + fromDate: fromDate, + toDate: toDate, + correlationId: correlationId, + cancellationToken: ct); + + Response.StatusCode = StatusCodes.Status202Accepted; + Response.ContentType = "application/json"; + await Response.WriteAsync(JsonSerializer.Serialize(new IngestionResponse + { + JobId = jobId, + Status = "Queued", + ExpectedRowCount = expectedCount, + QueuedAt = DateTime.UtcNow, + }), ct); + } +} + +public sealed class GetIngestionStatusEndpoint : EndpointWithoutRequest +{ + private readonly IMarketDataIngestionService _ingestionService; + + public GetIngestionStatusEndpoint(IMarketDataIngestionService ingestionService) + { + _ingestionService = ingestionService; + } + + public override void Configure() + { + Get("/api/market/ingest/{jobId}"); + AllowAnonymous(); + } + + public override async Task HandleAsync(CancellationToken ct) + { + var jobIdStr = Route("jobId"); + if (!Guid.TryParse(jobIdStr, out var jobId)) + { + ThrowError("Invalid job ID format"); + } + + var status = await _ingestionService.GetIngestionStatusAsync(jobId, ct); + + if (status == null) + { + ThrowError("Job not found"); + } + + Response.StatusCode = StatusCodes.Status200OK; + Response.ContentType = "application/json"; + await Response.WriteAsync(JsonSerializer.Serialize(new IngestionStatusResponse + { + JobId = status.JobId, + Status = status.Status, + RowsProcessed = status.RowsProcessed, + RowsFailed = status.RowsFailed, + RowsSkipped = status.RowsSkipped, + CompletedAt = status.CompletedAt, + DurationSeconds = status.DurationSeconds, + ErrorMessage = status.ErrorMessage, + }), ct); + } +} + +/// +/// VS-03 Application Handler: Orchestrates ingestion +/// +/// Responsibilities: +/// - Schedule ingestion job (Hangfire) +/// - Validate date range +/// - Check idempotency (same date range = no re-run) +/// - Audit logging +/// + +public interface IMarketDataIngestionService +{ + Task<(Guid JobId, int ExpectedRowCount)> ScheduleIngestionAsync( + string dataSource, + DateOnly fromDate, + DateOnly toDate, + string correlationId, + CancellationToken cancellationToken); + + Task GetIngestionStatusAsync(Guid jobId, CancellationToken cancellationToken); +} + +public record IngestionJobStatus( + Guid JobId, + string Status, + int RowsProcessed, + int RowsFailed, + int RowsSkipped, + DateTime? CompletedAt, + int? DurationSeconds, + string? ErrorMessage); + +public class MarketDataIngestionService : IMarketDataIngestionService +{ + private readonly NpgsqlDataSource _dataSource; + private readonly IBackgroundJobClient _jobClient; + + public MarketDataIngestionService(NpgsqlDataSource dataSource, IBackgroundJobClient jobClient) + { + _dataSource = dataSource; + _jobClient = jobClient; + } + + public async Task<(Guid JobId, int ExpectedRowCount)> ScheduleIngestionAsync( + string dataSource, + DateOnly fromDate, + DateOnly toDate, + string correlationId, + CancellationToken cancellationToken) + { + var jobId = Guid.NewGuid(); + + // Check idempotency: Is there already a job for this date range? + const string checkSql = """ + SELECT job_id FROM market_data.ingestion_jobs + WHERE data_source = @source + AND from_date = @fromDate + AND to_date = @toDate + AND status IN ('Running', 'Completed') + LIMIT 1; + """; + + await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken); + await using var checkCmd = connection.CreateCommand(); + checkCmd.CommandText = checkSql; + checkCmd.Parameters.AddWithValue("@source", dataSource); + checkCmd.Parameters.AddWithValue("@fromDate", fromDate.ToDateTime(TimeOnly.MinValue)); + checkCmd.Parameters.AddWithValue("@toDate", toDate.ToDateTime(TimeOnly.MinValue)); + + var existingJobId = await checkCmd.ExecuteScalarAsync(cancellationToken); + if (existingJobId != null) + { + return ((Guid)existingJobId, 0); + } + + // Estimate row count (rough: days * ~2000 stocks) + var days = (toDate.DayNumber - fromDate.DayNumber) + 1; + var expectedCount = days * 2000; // Stub estimate + + // Insert job record + const string insertSql = """ + INSERT INTO market_data.ingestion_jobs (job_id, data_source, from_date, to_date, status, correlation_id, triggered_by) + VALUES (@jobId, @source, @fromDate, @toDate, 'Queued', @correlationId, 'API'); + """; + + await using var insertCmd = connection.CreateCommand(); + insertCmd.CommandText = insertSql; + insertCmd.Parameters.AddWithValue("@jobId", jobId); + insertCmd.Parameters.AddWithValue("@source", dataSource); + insertCmd.Parameters.AddWithValue("@fromDate", fromDate.ToDateTime(TimeOnly.MinValue)); + insertCmd.Parameters.AddWithValue("@toDate", toDate.ToDateTime(TimeOnly.MinValue)); + insertCmd.Parameters.AddWithValue("@correlationId", correlationId); + + await insertCmd.ExecuteNonQueryAsync(cancellationToken); + + // Schedule Hangfire job + _jobClient.Enqueue(j => + j.ExecuteAsync(jobId, dataSource, fromDate, toDate, correlationId, CancellationToken.None)); + + return (jobId, expectedCount); + } + + public async Task GetIngestionStatusAsync(Guid jobId, CancellationToken cancellationToken) + { + const string sql = """ + SELECT job_id, status, rows_processed, rows_failed, rows_skipped, completed_at, duration_seconds, last_error_message + FROM market_data.ingestion_jobs + WHERE job_id = @jobId; + """; + + await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken); + await using var cmd = connection.CreateCommand(); + cmd.CommandText = sql; + cmd.Parameters.AddWithValue("@jobId", jobId); + + await using var reader = await cmd.ExecuteReaderAsync(cancellationToken); + if (!await reader.ReadAsync(cancellationToken)) + return null; + + return new IngestionJobStatus( + JobId: reader.GetGuid(0), + Status: reader.GetString(1), + RowsProcessed: reader.IsDBNull(2) ? 0 : reader.GetInt32(2), + RowsFailed: reader.IsDBNull(3) ? 0 : reader.GetInt32(3), + RowsSkipped: reader.IsDBNull(4) ? 0 : reader.GetInt32(4), + CompletedAt: reader.IsDBNull(5) ? null : reader.GetDateTime(5), + DurationSeconds: reader.IsDBNull(6) ? null : reader.GetInt32(6), + ErrorMessage: reader.IsDBNull(7) ? null : reader.GetString(7)); + } +} + +/// +/// Abstraction: Market data ingestion job (Hangfire worker) +/// + +public interface IMarketDataIngestionJob +{ + Task ExecuteAsync(Guid jobId, string dataSource, DateOnly fromDate, DateOnly toDate, string correlationId, CancellationToken ct); +} diff --git a/src/KArtSell.Host/Features/MarketData/VS03_IngestionJobs.cs b/src/KArtSell.Host/Features/MarketData/VS03_IngestionJobs.cs new file mode 100644 index 00000000..8f6a98b3 --- /dev/null +++ b/src/KArtSell.Host/Features/MarketData/VS03_IngestionJobs.cs @@ -0,0 +1,257 @@ +using Hangfire; +using System.Text.Json; +using KArtSell.Modules.ModelOperations.Domain; + +namespace KArtSell.Host.Features.MarketData; + +/// +/// VS-03 ASYNC: Market Data Ingestion Job +/// +/// Scheduled: Daily 9:00 KST (before market open) +/// Responsibility: Fetch, validate, normalize, persist market data +/// Idempotency: By date range (same range = no re-run) +/// + +public class MarketDataSyncedEvent +{ + public Guid EventId { get; set; } = Guid.NewGuid(); + public string EventType { get; set; } = "MarketDataSynced"; + public DateOnly FromDate { get; set; } + public DateOnly ToDate { get; set; } + public int RowsProcessed { get; set; } + public int RowsFailed { get; set; } + public DateTime SyncedAt { get; set; } + public string CorrelationId { get; set; } = ""; +} + +public interface IMarketDataEventPublisher +{ + Task PublishSyncedAsync(MarketDataSyncedEvent evt, CancellationToken ct); +} + +public class MarketDataEventPublisher : IMarketDataEventPublisher +{ + private readonly Npgsql.NpgsqlDataSource _dataSource; + + public MarketDataEventPublisher(Npgsql.NpgsqlDataSource dataSource) + { + _dataSource = dataSource; + } + + public async Task PublishSyncedAsync(MarketDataSyncedEvent evt, CancellationToken ct) + { + await using var connection = await _dataSource.OpenConnectionAsync(ct); + + const string sql = """ + INSERT INTO shared.outbox (aggregate_id, event_type, payload, published_at, correlation_id) + VALUES (@aggregateId, @eventType, @payload, CURRENT_TIMESTAMP, @correlationId); + """; + + await using var cmd = connection.CreateCommand(); + cmd.CommandText = sql; + cmd.Parameters.AddWithValue("@aggregateId", Guid.NewGuid()); + cmd.Parameters.AddWithValue("@eventType", evt.EventType); + cmd.Parameters.AddWithValue("@payload", JsonSerializer.Serialize(evt)); + cmd.Parameters.AddWithValue("@correlationId", evt.CorrelationId); + + await cmd.ExecuteNonQueryAsync(ct); + } +} + +/// +/// VS-03 ASYNC: Daily ingestion job +/// +/// Runs at 9:00 KST daily +/// Flow: Fetch → Validate → Normalize → Persist → Event publish +/// + +public class MarketDataIngestionJobHandler : IMarketDataIngestionJob +{ + private readonly IMarketDataDataSourceClient _krxClient; + private readonly IMarketDataEventPublisher _eventPublisher; + private readonly Npgsql.NpgsqlDataSource _dataSource; + + public MarketDataIngestionJobHandler( + IMarketDataDataSourceClient krxClient, + IMarketDataEventPublisher eventPublisher, + Npgsql.NpgsqlDataSource dataSource) + { + _krxClient = krxClient; + _eventPublisher = eventPublisher; + _dataSource = dataSource; + } + + public async Task ExecuteAsync( + Guid jobId, + string dataSource, + DateOnly fromDate, + DateOnly toDate, + string correlationId, + CancellationToken ct) + { + var startTime = DateTime.UtcNow; + var rowsProcessed = 0; + var rowsFailed = 0; + + try + { + // Update job status + await UpdateJobStatusAsync(jobId, "Running", ct); + + // Fetch prices from data source + var prices = await _krxClient.FetchPricesAsync(dataSource, fromDate, toDate, ct); + + if (prices.Count == 0) + { + await UpdateJobStatusAsync(jobId, "Completed", 0, 0, ct); + return; + } + + // Validate & normalize + var validPrices = new List(); + foreach (var price in prices) + { + var result = MarketDataPolicy.ValidatePrice(price, toDate); + if (result.IsValid) + { + var normalized = MarketDataPolicy.NormalizePrice(price); + if (normalized != null) + { + validPrices.Add(normalized); + rowsProcessed++; + } + } + else + { + rowsFailed++; + } + } + + // Persist to database + await PersistPricesAsync(validPrices, ct); + + // Publish event + var evt = new MarketDataSyncedEvent + { + FromDate = fromDate, + ToDate = toDate, + RowsProcessed = rowsProcessed, + RowsFailed = rowsFailed, + SyncedAt = DateTime.UtcNow, + CorrelationId = correlationId, + }; + + await _eventPublisher.PublishSyncedAsync(evt, ct); + + // Mark complete + var duration = (int)(DateTime.UtcNow - startTime).TotalSeconds; + await UpdateJobStatusAsync(jobId, "Completed", rowsProcessed, rowsFailed, duration, ct); + } + catch (Exception ex) + { + await UpdateJobStatusAsync(jobId, "Failed", rowsProcessed, rowsFailed, null, ex.Message, ct); + throw; + } + } + + private async Task PersistPricesAsync(List prices, CancellationToken ct) + { + if (prices.Count == 0) + return; + + const string sql = """ + INSERT INTO market_data.daily_prices + (symbol, trading_date, open_price, high_price, low_price, close_price, volume, published_at, revision, data_source, correlation_id) + VALUES (@symbol, @date, @open, @high, @low, @close, @volume, CURRENT_TIMESTAMP, 1, @source, @corrId) + ON CONFLICT (symbol, trading_date, revision) DO UPDATE SET + open_price = EXCLUDED.open_price, + close_price = EXCLUDED.close_price, + volume = EXCLUDED.volume, + published_at = CURRENT_TIMESTAMP + WHERE EXCLUDED.published_at > market_data.daily_prices.published_at; + """; + + await using var connection = await _dataSource.OpenConnectionAsync(ct); + foreach (var price in prices) + { + await using var cmd = connection.CreateCommand(); + cmd.CommandText = sql; + cmd.Parameters.AddWithValue("@symbol", price.Symbol); + cmd.Parameters.AddWithValue("@date", price.TradingDate.ToDateTime(TimeOnly.MinValue)); + cmd.Parameters.AddWithValue("@open", price.OpenPrice); + cmd.Parameters.AddWithValue("@high", price.HighPrice); + cmd.Parameters.AddWithValue("@low", price.LowPrice); + cmd.Parameters.AddWithValue("@close", price.ClosePrice); + cmd.Parameters.AddWithValue("@volume", price.Volume); + cmd.Parameters.AddWithValue("@source", price.DataSource); + cmd.Parameters.AddWithValue("@corrId", price.CorrelationId); + + await cmd.ExecuteNonQueryAsync(ct); + } + } + + private async Task UpdateJobStatusAsync(Guid jobId, string status, CancellationToken ct) + => await UpdateJobStatusAsync(jobId, status, 0, 0, null, null, ct); + + private async Task UpdateJobStatusAsync( + Guid jobId, + string status, + int rowsProcessed, + int rowsFailed, + CancellationToken ct) + => await UpdateJobStatusAsync(jobId, status, rowsProcessed, rowsFailed, null, null, ct); + + private async Task UpdateJobStatusAsync( + Guid jobId, + string status, + int rowsProcessed, + int rowsFailed, + int? durationSeconds, + string? errorMessage, + CancellationToken ct) + { + const string sql = """ + UPDATE market_data.ingestion_jobs + SET status = @status, + rows_processed = @rows, + rows_failed = @failed, + duration_seconds = @duration, + last_error_message = @error, + completed_at = CASE WHEN @status IN ('Completed', 'Failed') THEN CURRENT_TIMESTAMP ELSE NULL END, + 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("@rows", rowsProcessed); + cmd.Parameters.AddWithValue("@failed", rowsFailed); + cmd.Parameters.AddWithValue("@duration", durationSeconds ?? (object)DBNull.Value); + cmd.Parameters.AddWithValue("@error", errorMessage ?? (object)DBNull.Value); + + await cmd.ExecuteNonQueryAsync(ct); + } +} + +/// +/// Abstraction: Market data source client (KRX, OpenDart, Stub) +/// + +public interface IMarketDataDataSourceClient +{ + Task> FetchPricesAsync(string dataSource, DateOnly fromDate, DateOnly toDate, CancellationToken ct); +} + +public class StubMarketDataClient : IMarketDataDataSourceClient +{ + public async Task> FetchPricesAsync(string dataSource, DateOnly fromDate, DateOnly toDate, CancellationToken ct) + { + await Task.Delay(100, ct); // Stub delay + + // Return empty for now (real implementation would call KRX/OpenDart) + return new(); + } +}