using Application.Abstractions.Data; using Application.Helpers; using Domain.Entities.Stocks; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using SharedKernel; namespace Infrastructure.StockData; /// /// ESG 데이터(ESG 증권상품 + ESG 지수 + 사회책임투자채권) 수집 — KRX OpenAPI (esg/esg_etp_info + esg_index_info + sri_bond_info). /// 세 엔드포인트 모두 basDd 파라미터로 그날 전량(스냅샷)을 반환하므로 공용 KrxBackfill 로 최근 BackfillYears(기본 3)년치를 endDate(직전 영업일)부터 /// 과거로 훑으며 미적재일만 채운다. quota 보호를 위해 1회 실행당 BackfillMaxPerRun(기본 60)일까지만 fetch → 여러 날에 걸쳐 3년치를 메우고 이후엔 최신만 유지. /// 각 날짜는 3개 엔드포인트를 모두 수집해 EsgSecurity(UQ Name+TradeDate) / EsgIndexDailyPrice(UQ IndexName+TradeDate) / SriBond(UQ Code+TradeDate) upsert. /// 세 테이블 모두 해당 날짜가 있어야 "적재됨"으로 간주(부분 적재 방어). 기본 18:55 KST 실행(일반상품 18:50 뒤). ApiKey 미설정 시 skip. /// internal sealed class KrxEsgSyncService( IServiceScopeFactory scopeFactory, IHttpClientFactory httpClientFactory, IOptions settings, ILogger logger ) : DailyScheduledService(logger) { private const string EtpPath = "/svc/apis/esg/esg_etp_info"; private const string IndexPath = "/svc/apis/esg/esg_index_info"; private const string SriBondPath = "/svc/apis/esg/sri_bond_info"; protected override string JobName => "KrxEsgSync"; protected override TimeOnly TargetTime => ParseTime(settings.Value.KRXCoKr.EsgSyncTime, new TimeOnly(18, 55)); protected override int MaxRetryCount => 2; protected override TimeSpan RetryDelay => TimeSpan.FromHours(2); protected override async Task RunOnceAsync(DateOnly todayKst, CancellationToken ct) { var cfg = settings.Value.KRXCoKr; if (string.IsNullOrWhiteSpace(cfg.ApiKey)) { Logger.LogWarning("[{Job}] KRXCoKr:ApiKey 미설정 — 수집 skip", JobName); return true; } using var scope = scopeFactory.CreateScope(); var db = scope.ServiceProvider.GetRequiredService(); var client = httpClientFactory.CreateClient(KrxCoKrHttp.ClientName); var endDate = await MarketCalendar.GetPreviousBusinessDayAsync(db, todayKst.AddDays(1), ct); var years = cfg.BackfillYears > 0 ? cfg.BackfillYears : 3; var startDate = todayKst.AddYears(-years); var holidays = (await db.MarketHoliday.AsNoTracking() .Where(c => c.Date >= startDate && c.Date <= endDate) .Select(c => c.Date) .ToListAsync(ct)).ToHashSet(); var maxPerRun = cfg.BackfillMaxPerRun > 0 ? cfg.BackfillMaxPerRun : 60; var fetched = await KrxBackfill.RunAsync( existsForDate: (day, token) => AllPresentForDateAsync(db, day, token), fetchAndUpsertForDate: (day, token) => FetchAndUpsertAsync(db, client, cfg.BaseUrl, cfg.ApiKey, day, token), startDate: startDate, endDate: endDate, holidays: holidays, maxPerRun: maxPerRun, delayMs: 300, ct: ct); Logger.LogInformation("[{Job}] 완료 — 창=[{Start}~{End}], 이번 실행 fetch={Fetched}일 (maxPerRun={Max})", JobName, startDate, endDate, fetched, maxPerRun); return true; } /// 세 테이블 모두 해당 날짜가 적재되어 있는지 (부분 적재 시 refetch 하도록 AND 조건). private static async Task AllPresentForDateAsync(IAppDbContext db, DateOnly day, CancellationToken ct) { var etp = await db.EsgSecurity.AsNoTracking().AnyAsync(c => c.TradeDate == day, ct); var index = await db.EsgIndexDailyPrice.AsNoTracking().AnyAsync(c => c.TradeDate == day, ct); var sri = await db.SriBond.AsNoTracking().AnyAsync(c => c.TradeDate == day, ct); return etp && index && sri; } /// 한 날짜에 대해 3개 ESG 엔드포인트를 모두 수집하고 각 테이블 upsert. private async Task FetchAndUpsertAsync(IAppDbContext db, HttpClient client, string baseUrl, string apiKey, DateOnly day, CancellationToken ct) { var trimmed = baseUrl.TrimEnd('/'); await UpsertEsgSecuritiesAsync(db, client, trimmed, apiKey, day, ct); await UpsertEsgIndicesAsync(db, client, trimmed, apiKey, day, ct); await UpsertSriBondsAsync(db, client, trimmed, apiKey, day, ct); } private async Task UpsertEsgSecuritiesAsync(IAppDbContext db, HttpClient client, string baseUrl, string apiKey, DateOnly day, CancellationToken ct) { var url = $"{baseUrl}{EtpPath}?basDd={day:yyyyMMdd}"; var json = await KrxCoKrHttp.GetStringWithRetryAsync(client, url, apiKey, Logger, ct); var rows = KrxEsgSecurityParser.ParseDaily(json); Logger.LogInformation("[{Job}] ESG증권상품 basDd={Day} rows={Rows}", JobName, day, rows.Count); if (rows.Count == 0) { return; } var existing = await db.EsgSecurity.Where(c => c.TradeDate == day).ToListAsync(ct); var existingByKey = existing.ToDictionary(c => c.Name); foreach (var row in rows) { if (existingByKey.TryGetValue(row.Name, out var entity)) { entity.Update(row.Close, row.ChangeAmount, row.ChangeRate, row.ListedShares, row.Volume, row.TradeValue); } else { var created = EsgSecurity.Create(row.Name, row.TradeDate, row.Close, row.ChangeAmount, row.ChangeRate, row.ListedShares, row.Volume, row.TradeValue); await db.EsgSecurity.AddAsync(created, ct); existingByKey[row.Name] = created; } } await db.SaveChangesAsync(ct); } private async Task UpsertEsgIndicesAsync(IAppDbContext db, HttpClient client, string baseUrl, string apiKey, DateOnly day, CancellationToken ct) { var url = $"{baseUrl}{IndexPath}?basDd={day:yyyyMMdd}"; var json = await KrxCoKrHttp.GetStringWithRetryAsync(client, url, apiKey, Logger, ct); var rows = KrxEsgIndexParser.ParseDaily(json); Logger.LogInformation("[{Job}] ESG지수 basDd={Day} rows={Rows}", JobName, day, rows.Count); if (rows.Count == 0) { return; } var existing = await db.EsgIndexDailyPrice.Where(c => c.TradeDate == day).ToListAsync(ct); var existingByKey = existing.ToDictionary(c => c.IndexName); foreach (var row in rows) { if (existingByKey.TryGetValue(row.IndexName, out var entity)) { entity.Update(row.Close, row.ChangeVal, row.FlucRateBp, row.ConstituentCount, row.Volume, row.Value); } else { var created = EsgIndexDailyPrice.Create(row.IndexName, row.TradeDate, row.Close, row.ChangeVal, row.FlucRateBp, row.ConstituentCount, row.Volume, row.Value); await db.EsgIndexDailyPrice.AddAsync(created, ct); existingByKey[row.IndexName] = created; } } await db.SaveChangesAsync(ct); } private async Task UpsertSriBondsAsync(IAppDbContext db, HttpClient client, string baseUrl, string apiKey, DateOnly day, CancellationToken ct) { var url = $"{baseUrl}{SriBondPath}?basDd={day:yyyyMMdd}"; var json = await KrxCoKrHttp.GetStringWithRetryAsync(client, url, apiKey, Logger, ct); var rows = KrxSriBondParser.ParseDaily(json); Logger.LogInformation("[{Job}] 사회책임투자채권 basDd={Day} rows={Rows}", JobName, day, rows.Count); if (rows.Count == 0) { return; } var existing = await db.SriBond.Where(c => c.TradeDate == day).ToListAsync(ct); var existingByKey = existing.ToDictionary(c => c.Code); foreach (var row in rows) { if (existingByKey.TryGetValue(row.Code, out var entity)) { entity.Update(row.IssuerName, row.SriBondType, row.Name, row.ListDate, row.IssueDate, row.RedemptionDate, row.CouponRate, row.IssueAmount, row.ListAmount, row.BondType); } else { var created = SriBond.Create(row.Code, row.TradeDate, row.IssuerName, row.SriBondType, row.Name, row.ListDate, row.IssueDate, row.RedemptionDate, row.CouponRate, row.IssueAmount, row.ListAmount, row.BondType); await db.SriBond.AddAsync(created, ct); existingByKey[row.Code] = created; } } await db.SaveChangesAsync(ct); } }