feat: ingest canonical JSON seeds into PostgreSQL
This commit is contained in:
@@ -0,0 +1,69 @@
|
|||||||
|
using System.Text.Json;
|
||||||
|
using QuantEngine.Core.Interfaces;
|
||||||
|
|
||||||
|
namespace QuantEngine.Application.Services;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// JSON-first seed ingestion path. XLSX conversion remains an external
|
||||||
|
/// preparation step; the runtime application only reads canonical JSON and
|
||||||
|
/// persists normalized snapshots to PostgreSQL.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class JsonSeedIngestionService
|
||||||
|
{
|
||||||
|
private readonly GatherTradingDataParser _parser;
|
||||||
|
private readonly ICollectionRepository _repository;
|
||||||
|
private readonly ILogger<JsonSeedIngestionService> _logger;
|
||||||
|
|
||||||
|
public JsonSeedIngestionService(
|
||||||
|
GatherTradingDataParser parser,
|
||||||
|
ICollectionRepository repository,
|
||||||
|
ILogger<JsonSeedIngestionService> logger)
|
||||||
|
{
|
||||||
|
_parser = parser;
|
||||||
|
_repository = repository;
|
||||||
|
_logger = logger;
|
||||||
|
}
|
||||||
|
|
||||||
|
public async Task<CollectionRunResult> IngestAsync(string jsonPath, string runId)
|
||||||
|
{
|
||||||
|
var startedAt = DateTimeOffset.UtcNow;
|
||||||
|
var rows = _parser.ParseGatherTradingData(jsonPath);
|
||||||
|
await _repository.SaveRunAsync(new CollectionRunRecord(
|
||||||
|
runId, "RUNNING", startedAt.ToString("O"), null, rows.Count, 0));
|
||||||
|
|
||||||
|
var errors = 0;
|
||||||
|
foreach (var row in rows)
|
||||||
|
{
|
||||||
|
if (!row.TryGetValue("Ticker", out var tickerValue) || string.IsNullOrWhiteSpace(tickerValue?.ToString()))
|
||||||
|
{
|
||||||
|
errors++;
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
var ticker = tickerValue.ToString()!;
|
||||||
|
await _repository.SaveSnapshotAsync(new CollectionSnapshotRecord(
|
||||||
|
runId,
|
||||||
|
"data_feed",
|
||||||
|
ticker,
|
||||||
|
"json_seed",
|
||||||
|
JsonSerializer.Serialize(row),
|
||||||
|
startedAt.ToString("O")));
|
||||||
|
}
|
||||||
|
|
||||||
|
var finishedAt = DateTimeOffset.UtcNow;
|
||||||
|
var status = rows.Count > 0 && errors == 0 ? "COMPLETED" : "COMPLETED_WITH_ERRORS";
|
||||||
|
await _repository.UpdateRunStatusAsync(runId, status, finishedAt.ToString("O"), rows.Count - errors, errors);
|
||||||
|
_logger.LogInformation("JSON seed ingestion {RunId} completed: {Snapshots} snapshots, {Errors} errors", runId, rows.Count - errors, errors);
|
||||||
|
|
||||||
|
return new CollectionRunResult
|
||||||
|
{
|
||||||
|
RunId = runId,
|
||||||
|
Status = status,
|
||||||
|
StartedAt = startedAt.ToString("O"),
|
||||||
|
FinishedAt = finishedAt.ToString("O"),
|
||||||
|
SuccessCount = rows.Count - errors,
|
||||||
|
ErrorCount = errors,
|
||||||
|
Rows = rows
|
||||||
|
};
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -100,6 +100,8 @@ try
|
|||||||
builder.Services.AddScoped<ILearningDatasetReader, LearningDatasetReader>();
|
builder.Services.AddScoped<ILearningDatasetReader, LearningDatasetReader>();
|
||||||
builder.Services.AddScoped<DecisionLearningService>();
|
builder.Services.AddScoped<DecisionLearningService>();
|
||||||
builder.Services.AddScoped<LearningDatasetService>();
|
builder.Services.AddScoped<LearningDatasetService>();
|
||||||
|
builder.Services.AddSingleton<GatherTradingDataParser>();
|
||||||
|
builder.Services.AddScoped<JsonSeedIngestionService>();
|
||||||
builder.Services.AddScoped<IPostgresqlHistorySnapshotReader, PostgresqlHistorySnapshotReader>();
|
builder.Services.AddScoped<IPostgresqlHistorySnapshotReader, PostgresqlHistorySnapshotReader>();
|
||||||
builder.Services.AddScoped<HistoryIngestionService>();
|
builder.Services.AddScoped<HistoryIngestionService>();
|
||||||
builder.Services.AddScoped<ICollectionRepository, CollectionRepository>();
|
builder.Services.AddScoped<ICollectionRepository, CollectionRepository>();
|
||||||
|
|||||||
Reference in New Issue
Block a user