ANTOOZA-BACKEND/Infrastructure/StockData/KrxEtpSyncService.cs

using Application.Abstractions.Data;
using Application.Helpers;
using Domain.Entities.Stocks;
using Domain.Entities.Stocks.ValueObject;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using SharedKernel;

namespace Infrastructure.StockData;

/// <summary>
/// 증권상품(ETF/ETN/ELW) 일별매매 수집 (KRX OpenAPI) — etf_bydd_trd + etn_bydd_trd + elw_bydd_trd.
/// 기본 18:20 KST 실행(장 마감 15:30 이후 확정, 주식 수집 18:10 뒤). 공용 KrxBackfill 로 최근 BackfillYears(기본 3)년치를
/// endDate(직전 영업일)부터 과거로 훑으며 미적재일만 채운다. quota 보호를 위해 1회 실행당 BackfillMaxPerRun(기본 60)일까지만 fetch →
/// 여러 날에 걸쳐 3년치를 메우고 이후엔 최신만 유지한다. 각 날짜는 3개 유형 엔드포인트를 모두 수집해 EtpDailyTrade upsert
/// (UQ = EtpType+Code+TradeDate). existsForDate 는 3개 유형이 유형별 데이터 시작일 기준으로 전부 적재됐을 때만 skip
/// (일부 유형만 성공한 부분 적재 날짜는 재수집 — 영구 결손 방지). ApiKey 미설정 시 로그만 남기고 skip (다른 KRX 배치와 동일 정책).
/// </summary>
internal sealed class KrxEtpSyncService(
    IServiceScopeFactory scopeFactory,
    IHttpClientFactory httpClientFactory,
    IOptions<AppSettings> settings,
    ILogger<KrxEtpSyncService> logger
) : DailyScheduledService(logger)
{
    // (유형, 엔드포인트 경로) — ETF/ETN/ELW
    private static readonly (EtpType Type, string Path)[] Endpoints =
    [
        (EtpType.ETF, "/svc/apis/etp/etf_bydd_trd"),
        (EtpType.ETN, "/svc/apis/etp/etn_bydd_trd"),
        (EtpType.ELW, "/svc/apis/etp/elw_bydd_trd")
    ];

    // 유형별 데이터 시작일 (docs/KRX/readme.md) — 시작일 이전 날짜는 해당 유형 미존재가 정상이라 완전성 판정에서 제외
    private static readonly (EtpType Type, DateOnly DataStartDate)[] TypeStartDates =
    [
        (EtpType.ETF, new DateOnly(2010, 1, 4)),
        (EtpType.ETN, new DateOnly(2014, 11, 17)),
        (EtpType.ELW, new DateOnly(2010, 1, 4))
    ];

    protected override string JobName => "KrxEtpSync";

    protected override TimeOnly TargetTime => ParseTime(settings.Value.KRXCoKr.EtpSyncTime, new TimeOnly(18, 20));

    protected override int MaxRetryCount => 2;

    protected override TimeSpan RetryDelay => TimeSpan.FromHours(2);

    protected override async Task<bool> RunOnceAsync(DateOnly todayKst, CancellationToken ct)
    {
        var cfg = settings.Value.KRXCoKr;

        using var scope = scopeFactory.CreateScope();
        var collectorSettings = scope.ServiceProvider.GetRequiredService<ICollectorSettingsProvider>();
        if (!await collectorSettings.IsEnabledAsync(CollectorFlag.KrxEtp, ct))
        {
            return true;
        }
        cfg = cfg with { ApiKey = await collectorSettings.GetKeyAsync(CollectorKey.Krx, ct) ?? cfg.ApiKey };
        if (string.IsNullOrWhiteSpace(cfg.ApiKey))
        {
            Logger.LogWarning("[{Job}] KRXCoKr:ApiKey 미설정 — 수집 skip", JobName);
            return true;
        }

        var db = scope.ServiceProvider.GetRequiredService<IAppDbContext>();
        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);

        // 백필 창 전체의 휴장일을 한 번에 로드 (KrxBackfill 은 주말은 자동 제외, 휴장일만 필요)
        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;

        // 유형(ETF/ETN/ELW)별 완전성 판정 — 일부 엔드포인트만 성공한 부분 적재 날짜는 skip 하지 않고 재수집한다
        async Task<bool> ExistsForDateAsync(DateOnly day, CancellationToken token)
        {
            var types = await db.EtpDailyTrade.AsNoTracking().Where(c => c.TradeDate == day).Select(c => c.EtpType).Distinct().ToListAsync(token);
            return BackfillCompleteness.IsComplete(types, TypeStartDates, day);
        }

        var fetched = await KrxBackfill.RunAsync(
            existsForDate: ExistsForDateAsync,
            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);

        // 최신 영업일(endDate) 데이터가 이번 실행에서도 미적재면 false → 베이스가 RetryDelay 후 재시도 (KRX T+0 마감 데이터 미반영 대비)
        if (!await ExistsForDateAsync(endDate, ct))
        {
            Logger.LogWarning("[{Job}] 최신 영업일 {End} 데이터 미적재 — {Delay} 후 재시도 (최대 {Max}회)", JobName, endDate, RetryDelay, MaxRetryCount);
            return false;
        }

        return true;
    }

    /// <summary>한 날짜에 대해 3개 유형 엔드포인트를 모두 수집하고 EtpDailyTrade upsert.</summary>
    private async Task FetchAndUpsertAsync(IAppDbContext db, HttpClient client, string baseUrl, string apiKey, DateOnly day, CancellationToken ct)
    {
        var rows = new List<KrxEtpParser.EtpRow>();

        foreach (var (type, path) in Endpoints)
        {
            var url = $"{baseUrl.TrimEnd('/')}{path}?basDd={day:yyyyMMdd}";
            var json = await KrxCoKrHttp.GetStringWithRetryAsync(client, url, apiKey, Logger, ct);
            var parsed = KrxEtpParser.ParseDaily(json, type);

            Logger.LogInformation("[{Job}] {Type} basDd={Day} rows={Rows}", JobName, type, day, parsed.Count);
            rows.AddRange(parsed);
        }

        if (rows.Count == 0)
        {
            Logger.LogInformation("[{Job}] basDd={Day} 증권상품 미반영 (0건)", JobName, day);
            return;
        }

        var existing = await db.EtpDailyTrade.Where(c => c.TradeDate == day).ToListAsync(ct);
        var existingByKey = existing.ToDictionary(c => (c.EtpType, c.Code));
        var inserted = 0;
        var updated = 0;

        foreach (var row in rows)
        {
            if (existingByKey.TryGetValue((row.EtpType, row.Code), out var trade))
            {
                trade.Update(row.Name, row.Close, row.Open, row.High, row.Low, row.ChangeAmount, row.ChangeRate, row.Volume, row.TradeValue, row.MarketCap, row.ListedShares, row.Nav, row.NetAssetTotal, row.BaseIndexName, row.BaseIndexClose, row.Underlying, row.UnderlyingClose);
                updated++;
            }
            else
            {
                var created = EtpDailyTrade.Create(row.EtpType, row.Code, row.Name, row.TradeDate, row.Close, row.Open, row.High, row.Low, row.ChangeAmount, row.ChangeRate, row.Volume, row.TradeValue, row.MarketCap, row.ListedShares, row.Nav, row.NetAssetTotal, row.BaseIndexName, row.BaseIndexClose, row.Underlying, row.UnderlyingClose);
                await db.EtpDailyTrade.AddAsync(created, ct);
                existingByKey[(row.EtpType, row.Code)] = created;
                inserted++;
            }
        }

        await db.SaveChangesAsync(ct);

        Logger.LogInformation("[{Job}] basDd={Day} 적재 — rows={Rows}, inserted={Inserted}, updated={Updated}",
            JobName, day, rows.Count, inserted, updated);
    }
}