using System.Text.Json; using System.Text.Json.Serialization; using QuantEngine.Core.Interfaces; using QuantEngine.Application.Interfaces; using QuantEngine.Application.Services; namespace QuantEngine.Application.Services; public class KisDataCollectionOrchestrator : ICollectionOrchestrator { private readonly IKisApiClient _kisApiClient; private readonly ICollectionRepository _repository; private readonly PriceDataNormalizer _normalizer; private readonly SourcePriorityResolver _priorityResolver; // Logging removed for simplicity public KisDataCollectionOrchestrator( IKisApiClient kisApiClient, ICollectionRepository repository, PriceDataNormalizer normalizer, SourcePriorityResolver priorityResolver) { _kisApiClient = kisApiClient; _repository = repository; _normalizer = normalizer; _priorityResolver = priorityResolver; } public async Task RunCollectionAsync(string runId, string account, List tickers) { var startedAt = DataNormalizationHelper.KstNowIso(); var result = new CollectionRunResult { RunId = runId, Status = "RUNNING", StartedAt = startedAt, SuccessCount = 0, ErrorCount = 0 }; try { // Log: skipped var kisSource = new KisApiPriceSource(_kisApiClient); var rows = new List>(); var errors = new List>(); var sourceCounts = new Dictionary(); foreach (var ticker in tickers) { try { // Log: skipped var kisResult = await kisSource.GetPriceDataAsync(ticker, account); var seedRow = new Dictionary { { "Ticker", ticker } }; var (normalized, provenance) = _normalizer.NormalizeCollectionRow(seedRow, kisResult, null, false); // Save to DB await _repository.SaveSnapshotAsync(new CollectionSnapshotRecord( RunId: runId, DatasetName: "data_feed", Ticker: ticker, SourceName: (string)(provenance.GetValueOrDefault("source") ?? "kis_open_api"), PayloadJson: JsonSerializer.Serialize(normalized), CapturedAt: DataNormalizationHelper.KstNowIso() )); // Track source var source = (string)(provenance.GetValueOrDefault("source") ?? "kis_open_api"); if (!sourceCounts.ContainsKey(source)) sourceCounts[source] = 0; sourceCounts[source]++; rows.Add(normalized); result.SuccessCount++; } catch (Exception ex) { // Log: skipped result.ErrorCount++; errors.Add(new Dictionary { { "ticker", ticker }, { "error", ex.Message }, { "error_kind", ex.GetType().Name } }); await _repository.SaveErrorAsync(new CollectionErrorRecord( RunId: runId, SourceName: "kis_collector", ErrorKind: ex.GetType().Name, ErrorMessage: ex.Message, Ticker: ticker )); } } var finishedAt = DataNormalizationHelper.KstNowIso(); result.Status = result.ErrorCount == 0 ? "COMPLETED" : "COMPLETED_WITH_ERRORS"; result.FinishedAt = finishedAt; result.SourceCounts = sourceCounts; result.Rows = rows; result.Errors = errors; // Save run record await _repository.SaveRunAsync(new CollectionRunRecord( RunId: runId, Status: result.Status, StartedAt: startedAt, FinishedAt: finishedAt, TotalSnapshots: result.SuccessCount, TotalErrors: result.ErrorCount )); // Output JSON file var outputPath = Path.Combine(Path.GetTempPath(), "kis_data_collection_v1.json"); var outputData = new { formula_id = "KIS_DATA_COLLECTION_V1", run_id = runId, started_at = startedAt, finished_at = finishedAt, row_count = rows.Count, source_counts = sourceCounts, errors = errors, rows = rows }; File.WriteAllText(outputPath, JsonSerializer.Serialize(outputData, new JsonSerializerOptions { WriteIndented = true })); // Log: skipped return result; } catch (Exception ex) { // Log: skipped result.Status = "FAILED"; result.FinishedAt = DataNormalizationHelper.KstNowIso(); result.ErrorMessage = ex.Message; return result; } } }