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;
///
/// 일반상품(석유·금·배출권) 일별매매 수집 (KRX OpenAPI) — oil_bydd_trd + gold_bydd_trd + ets_bydd_trd.
/// 기본 19:15 KST 실행(장 마감 15:30 이후 확정, 파생상품 수집 19:00 이 6ep×백필로 무거워 여유). 공용 KrxBackfill 로 최근 BackfillYears(기본 3)년치를
/// endDate(직전 영업일)부터 과거로 훑으며 미적재일만 채운다. quota 보호를 위해 1회 실행당 BackfillMaxPerRun(기본 60)일까지만 fetch →
/// 여러 날에 걸쳐 3년치를 메우고 이후엔 최신만 유지한다. 각 날짜는 3개 시장 엔드포인트를 모두 수집해 CommodityDailyTrade upsert
/// (UQ = CommodityMarket+Code+TradeDate). existsForDate 는 3개 시장이 시장별 데이터 시작일(석유 2012~/금 2014~/배출권 2015~) 기준으로
/// 전부 적재됐을 때만 skip (일부 시장만 성공한 부분 적재 날짜는 재수집 — 영구 결손 방지). ApiKey 미설정 시 로그만 남기고 skip.
///
internal sealed class KrxCommoditySyncService(
IServiceScopeFactory scopeFactory,
IHttpClientFactory httpClientFactory,
IOptions settings,
ILogger logger
) : DailyScheduledService(logger)
{
// (시장, 엔드포인트 경로) — 석유(oil)/금(gold)/배출권(ets)
private static readonly (CommodityMarket Market, string Path)[] Endpoints =
[
(CommodityMarket.Oil, "/svc/apis/gen/oil_bydd_trd"),
(CommodityMarket.Gold, "/svc/apis/gen/gold_bydd_trd"),
(CommodityMarket.Emission, "/svc/apis/gen/ets_bydd_trd")
];
// 시장별 데이터 시작일 (docs/KRX/readme.md — 개장일이 서로 달라 시작일 이전 날짜는 해당 시장 미존재가 정상, 과잉 재fetch 방지)
private static readonly (CommodityMarket Market, DateOnly DataStartDate)[] MarketStartDates =
[
(CommodityMarket.Oil, new DateOnly(2012, 3, 30)),
(CommodityMarket.Gold, new DateOnly(2014, 3, 24)),
(CommodityMarket.Emission, new DateOnly(2015, 1, 12))
];
protected override string JobName => "KrxCommoditySync";
protected override TimeOnly TargetTime => ParseTime(settings.Value.KRXCoKr.CommoditySyncTime, new TimeOnly(19, 15));
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;
using var scope = scopeFactory.CreateScope();
var collectorSettings = scope.ServiceProvider.GetRequiredService();
if (!await collectorSettings.IsEnabledAsync(CollectorFlag.KrxCommodity, 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();
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;
// 시장(oil/gold/ets)별 완전성 판정 — 일부 엔드포인트만 성공한 부분 적재 날짜는 skip 하지 않고 재수집한다
async Task ExistsForDateAsync(DateOnly day, CancellationToken token)
{
var markets = await db.CommodityDailyTrade.AsNoTracking().Where(c => c.TradeDate == day).Select(c => c.CommodityMarket).Distinct().ToListAsync(token);
return BackfillCompleteness.IsComplete(markets, MarketStartDates, 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;
}
/// 한 날짜에 대해 3개 시장 엔드포인트를 모두 수집하고 CommodityDailyTrade upsert.
private async Task FetchAndUpsertAsync(IAppDbContext db, HttpClient client, string baseUrl, string apiKey, DateOnly day, CancellationToken ct)
{
var rows = new List();
foreach (var (market, path) in Endpoints)
{
var url = $"{baseUrl.TrimEnd('/')}{path}?basDd={day:yyyyMMdd}";
var json = await KrxCoKrHttp.GetStringWithRetryAsync(client, url, apiKey, Logger, ct);
var parsed = KrxCommodityParser.ParseDaily(json, market);
Logger.LogInformation("[{Job}] {Market} basDd={Day} rows={Rows}", JobName, market, day, parsed.Count);
rows.AddRange(parsed);
}
if (rows.Count == 0)
{
Logger.LogInformation("[{Job}] basDd={Day} 일반상품 미반영 (0건)", JobName, day);
return;
}
var existing = await db.CommodityDailyTrade.Where(c => c.TradeDate == day).ToListAsync(ct);
var existingByKey = existing.ToDictionary(c => (c.CommodityMarket, c.Code));
var inserted = 0;
var updated = 0;
foreach (var row in rows)
{
if (existingByKey.TryGetValue((row.CommodityMarket, row.Code), out var trade))
{
trade.Update(row.Name, row.Close, row.Open, row.High, row.Low, row.ChangeAmount, row.ChangeRate, row.WeightedDiscussionPrice, row.Volume, row.TradeValue);
updated++;
}
else
{
var created = CommodityDailyTrade.Create(row.CommodityMarket, row.Code, row.Name, row.TradeDate, row.Close, row.Open, row.High, row.Low, row.ChangeAmount, row.ChangeRate, row.WeightedDiscussionPrice, row.Volume, row.TradeValue);
await db.CommodityDailyTrade.AddAsync(created, ct);
existingByKey[(row.CommodityMarket, 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);
}
}