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);
}
}