using System.Net; using System.Text.Json; using Dapper; using Microsoft.Extensions.Caching.Memory; using Microsoft.Extensions.Logging; using Npgsql; namespace KArtSell.Modules.ModelOperations.ShadowRun.Services; /// /// Fetches historical OHLCV and fee schedule data from Korea Exchange (KRX) API. /// Implements caching, retry logic, and PIT-safe lookups (no forward bias). /// Rate limiting: Client-side throttling via exponential backoff on 429 responses. /// public sealed class KrxDataService : IKrxDataService { private readonly HttpClient _httpClient; private readonly IMemoryCache _cache; private readonly ILogger _logger; private readonly NpgsqlDataSource _dataSource; private const int CacheDurationMinutes = 1440; // 24 hours private const int RecentDaysWindow = 7; // Last 7 days: always refresh (mutable data) private const int MaxRetries = 3; private const int InitialBackoffMs = 100; private const int MaxBackoffMs = 30000; private const string KrxApiBaseUrl = "https://data-dbg.krx.co.kr"; // Stock price API (HTTPS, from pykrx-openapi) private const string KrxApiEndpoint = "/svc/apis/sto/stk_bydd_trd"; // KOSPI daily trading endpoint private static readonly Action LogFetchingOhlcv = LoggerMessage.Define( LogLevel.Information, new EventId(1, nameof(LogFetchingOhlcv)), "Fetching OHLCV: {Ticker} ({Start:yyyy-MM-dd} to {End:yyyy-MM-dd})"); private static readonly Action LogFetchedOhlcv = LoggerMessage.Define( LogLevel.Information, new EventId(2, nameof(LogFetchedOhlcv)), "Fetched {BarCount} OHLCV bars for {Ticker}"); private static readonly Action LogCacheHit = LoggerMessage.Define( LogLevel.Debug, new EventId(3, nameof(LogCacheHit)), "Cache hit for {CacheKey}"); private static readonly Action LogRetryError = LoggerMessage.Define( LogLevel.Warning, new EventId(4, nameof(LogRetryError)), "Retryable error: {ErrorMessage}"); public KrxDataService( HttpClient httpClient, IMemoryCache cache, ILogger logger, NpgsqlDataSource? dataSource = null) { _httpClient = httpClient; _cache = cache; _logger = logger; _dataSource = dataSource!; } /// /// Get the last date that was successfully imported from KRX API. /// Returns null if no successful import exists or DB not available. /// Used for incremental fetching (avoid re-fetching old, immutable data). /// private async Task GetLastSuccessfulImportDateAsync(CancellationToken cancellationToken) { // In test environments or when DB is not available, skip incremental optimization if (_dataSource == null) { return null; } try { const string sql = """ SELECT MAX(DATE(import_at)) as last_date FROM market_data.krx_imports WHERE status = 'SUCCESS' """; await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken); var result = await connection.QuerySingleOrDefaultAsync(sql); if (result == null) { _logger.LogInformation("No previous successful KRX import found, will fetch full range"); return null; } return DateOnly.FromDateTime(result.Value); } catch (Exception ex) { _logger.LogWarning(ex, "Failed to get last successful import date, fetching full range"); return null; } } /// /// Fetch daily OHLCV bars for ticker within date range. /// Implements caching (24h), incremental fetching (skip old data), and retry logic. /// PIT-safe: Returns only requested date range (no lookback). /// Strategy: Last 7 days always fresh (mutable), older data fetched only once. /// public async Task> GetDailyOhlcvAsync( string ticker, DateOnly startDate, DateOnly endDate, CancellationToken cancellationToken) { LogFetchingOhlcv(_logger, ticker, startDate, endDate, null); // Incremental fetching: Skip old, immutable data that was already collected var today = DateOnly.FromDateTime(DateTime.UtcNow.Date); var lastSuccessfulImport = await GetLastSuccessfulImportDateAsync(cancellationToken); // Strategy: Only fetch data from 7 days ago onwards (last 7 days always fresh) // Skip anything older that was already imported successfully var effectiveStartDate = startDate; if (lastSuccessfulImport.HasValue) { var oldDataCutoff = today.AddDays(-RecentDaysWindow); var latestOldData = lastSuccessfulImport.Value; if (latestOldData >= oldDataCutoff) { // Already have recent data, skip to day after last import effectiveStartDate = latestOldData.AddDays(1); } } // If effective range is empty, return empty if (effectiveStartDate > endDate) { _logger.LogInformation("Incremental fetch: {Ticker} data already up-to-date, no fetch needed", ticker); return Array.Empty(); } var skippedDays = effectiveStartDate.DayNumber - startDate.DayNumber; if (skippedDays > 0) { _logger.LogInformation( "Incremental fetch: {Ticker} skipping {SkippedDays} immutable days (already imported), starting from {EffectiveStart}", ticker, skippedDays, effectiveStartDate); } var cacheKey = $"ohlcv:{ticker}:{effectiveStartDate:yyyyMMdd}:{endDate:yyyyMMdd}"; // Check cache first if (_cache.TryGetValue(cacheKey, out IReadOnlyList? cached)) { LogCacheHit(_logger, cacheKey, null); return cached!; } // Fetch with exponential backoff retry var bars = new List(); int attempt = 0; int backoffMs = InitialBackoffMs; while (attempt < MaxRetries) { try { var response = await FetchOhlcvFromApiAsync(ticker, startDate, endDate, cancellationToken); bars = ParseOhlcvResponse(ticker, response); break; } catch (HttpRequestException ex) when (ex.StatusCode == HttpStatusCode.TooManyRequests && attempt < MaxRetries - 1) { // 429: Rate limit hit → exponential backoff backoffMs = Math.Min(backoffMs * 2, MaxBackoffMs); LogRetryError(_logger, $"Rate limited (429), backoff {backoffMs}ms (attempt {attempt + 1}/{MaxRetries})", ex); await Task.Delay(backoffMs, cancellationToken); attempt++; } catch (HttpRequestException ex) when (IsTransientError(ex) && attempt < MaxRetries - 1) { // Other transient errors → fixed 1s delay LogRetryError(_logger, $"{ex.Message} (attempt {attempt + 1}/{MaxRetries})", ex); await Task.Delay(1000, cancellationToken); attempt++; } catch (HttpRequestException ex) when (!IsTransientError(ex)) { _logger.LogError(ex, "Permanent HTTP error fetching {Ticker}", ticker); throw; } } // Cache result var cacheOptions = new MemoryCacheEntryOptions { AbsoluteExpirationRelativeToNow = TimeSpan.FromMinutes(CacheDurationMinutes) }; _cache.Set(cacheKey, (IReadOnlyList)bars.AsReadOnly(), cacheOptions); LogFetchedOhlcv(_logger, ticker, bars.Count, null); return bars; } /// /// Fetch fee schedule (transaction costs) for date range. /// Returns piecewise-constant fee entries. /// public async Task> GetFeeScheduleAsync( DateOnly startDate, DateOnly endDate, CancellationToken cancellationToken) { var cacheKey = $"fees:{startDate:yyyyMMdd}:{endDate:yyyyMMdd}"; if (_cache.TryGetValue(cacheKey, out IReadOnlyList? cached)) { return cached!; } // Simplified: stub implementation (hardcoded fees for now) // In production: fetch from KRX fee schedule API var fees = new List { new(startDate, 0.00015m, 0.0005m), // Transaction: 0.015%, Slippage: 0.05% }; var cacheOptions = new MemoryCacheEntryOptions { AbsoluteExpirationRelativeToNow = TimeSpan.FromMinutes(CacheDurationMinutes) }; _cache.Set(cacheKey, (IReadOnlyList)fees.AsReadOnly(), cacheOptions); return fees; } private async Task FetchOhlcvFromApiAsync( string ticker, DateOnly startDate, DateOnly endDate, CancellationToken cancellationToken) { // For now: return stub data (KRX API not available in this environment) // In production: use real API with apiKey = Environment.GetEnvironmentVariable("KRX_OPENAPI") _logger.LogInformation("Using stub OHLCV data for {Ticker} ({Start:yyyy-MM-dd} to {End:yyyy-MM-dd})", ticker, startDate, endDate); await Task.Delay(100, cancellationToken); // Simulate API latency // Generate stub data: 2 rows per trading day (simplified) var bars = new List(); for (var date = startDate; date <= endDate; date = date.AddDays(1)) { var openPrice = 100.0m + (date.DayNumber % 10); bars.Add(new { BasDt = date.ToString("yyyyMMdd"), Mkp = openPrice, Hipr = openPrice + 5, Lopr = openPrice - 2, Clpr = openPrice + 2, Trqu = 1000000L + (date.DayNumber * 10000) }); } return System.Text.Json.JsonSerializer.Serialize(bars); } private IEnumerable GenerateDateRange(DateOnly startDate, DateOnly endDate) { for (var date = startDate; date <= endDate; date = date.AddDays(1)) { yield return date; } } private string ExtractPriceItems(string krxResponse) { try { using var doc = JsonDocument.Parse(krxResponse); var root = doc.RootElement; if (root.TryGetProperty("OutBlock_1", out var outBlock)) { return outBlock.GetRawText(); } return "[]"; } catch { return "[]"; } } private List ParseOhlcvResponse(string ticker, string jsonResponse) { var bars = new List(); try { using var doc = JsonDocument.Parse(jsonResponse); var root = doc.RootElement; if (root.ValueKind != JsonValueKind.Array) { _logger.LogWarning("Unexpected response format for {Ticker}: expected array", ticker); return bars; } // Convert to list first (JsonDocument can't be enumerated in parallel) var elements = root.EnumerateArray().ToList(); // Parse in parallel (4 threads) for 504K rows var parsedBars = new DataBackfiller.OhlcvBar[elements.Count]; var lockObj = new object(); Parallel.For(0, elements.Count, new ParallelOptions { MaxDegreeOfParallelism = 4 }, i => { var element = elements[i]; try { // Parse KRX PriceItem format if (!element.TryGetProperty("BasDt", out var basDto)) return; var date = DateOnly.ParseExact(basDto.GetString()!, "yyyyMMdd"); parsedBars[i] = new DataBackfiller.OhlcvBar( Date: date, Ticker: ticker, Open: element.GetProperty("Mkp").GetDecimal(), // 시가 High: element.GetProperty("Hipr").GetDecimal(), // 고가 Low: element.GetProperty("Lopr").GetDecimal(), // 저가 Close: element.GetProperty("Clpr").GetDecimal(), // 종가 Volume: element.GetProperty("Trqu").GetInt64()); // 거래량 } catch (Exception ex) { _logger.LogWarning(ex, "Failed to parse OHLCV element {Index} for {Ticker}", i, ticker); } }); // Add non-null bars to result bars.AddRange(parsedBars.Where(b => b != null)); } catch (JsonException ex) { _logger.LogWarning(ex, "Failed to deserialize OHLCV response for {Ticker}", ticker); } return bars; } private static bool IsTransientError(HttpRequestException ex) { // 429: Too Many Requests (rate limit) // 503: Service Unavailable // 504: Gateway Timeout // 408: Request Timeout return ex.StatusCode == HttpStatusCode.TooManyRequests || ex.StatusCode == HttpStatusCode.ServiceUnavailable || ex.StatusCode == HttpStatusCode.GatewayTimeout || ex.StatusCode == HttpStatusCode.RequestTimeout || (ex.InnerException is TimeoutException); } }