372 lines
15 KiB
C#
372 lines
15 KiB
C#
|
|
using System.Text.Json;
|
|
using System.Text.Json.Serialization;
|
|
using System.IO;
|
|
using Microsoft.Extensions.Logging;
|
|
using QuantEngine.Core.Interfaces;
|
|
using QuantEngine.Application.Interfaces;
|
|
using QuantEngine.Application.Services;
|
|
using QuantEngine.Application.Models;
|
|
|
|
namespace QuantEngine.Application.Services;
|
|
|
|
public class KisDataCollectionOrchestrator : ICollectionOrchestrator
|
|
{
|
|
private readonly IKisApiClient _kisApiClient;
|
|
private readonly ICollectionWriteRepository _writeRepository;
|
|
private readonly ICollectionReadRepository _readRepository;
|
|
private readonly PriceDataNormalizer _normalizer;
|
|
private readonly SourcePriorityResolver _priorityResolver;
|
|
private readonly ILogger<KisDataCollectionOrchestrator> _logger;
|
|
private readonly IRuntimeAuditTrailService _auditTrail;
|
|
|
|
public Func<DateTime> UtcNowProvider { get; set; } = () => DateTime.UtcNow;
|
|
|
|
public KisDataCollectionOrchestrator(
|
|
IKisApiClient kisApiClient,
|
|
ICollectionWriteRepository repository,
|
|
ICollectionReadRepository readRepository,
|
|
PriceDataNormalizer normalizer,
|
|
SourcePriorityResolver priorityResolver,
|
|
ILogger<KisDataCollectionOrchestrator> logger,
|
|
IRuntimeAuditTrailService auditTrail)
|
|
{
|
|
_kisApiClient = kisApiClient;
|
|
_writeRepository = repository;
|
|
_readRepository = readRepository;
|
|
_normalizer = normalizer;
|
|
_priorityResolver = priorityResolver;
|
|
_logger = logger;
|
|
_auditTrail = auditTrail;
|
|
}
|
|
|
|
public async Task<CollectionRunResult> RunCollectionAsync(string runId, string account, List<string> tickers)
|
|
{
|
|
var startedAt = DataNormalizationHelper.KstNowIso();
|
|
var result = new CollectionRunResult
|
|
{
|
|
RunId = runId,
|
|
Status = "RUNNING",
|
|
StartedAt = startedAt,
|
|
SuccessCount = 0,
|
|
ErrorCount = 0
|
|
};
|
|
|
|
try
|
|
{
|
|
_logger.LogInformation("Starting collection run {RunId}", runId);
|
|
_auditTrail.Append("collection_audit", runId, new CollectionExecutionAudit(runId, "RUNNING", DateTimeOffset.UtcNow, null, 0, 0, "started"));
|
|
|
|
var kisSource = new KisApiPriceSource(_kisApiClient);
|
|
var rows = new List<Dictionary<string, object>>();
|
|
var errors = new List<Dictionary<string, object>>();
|
|
var sourceCounts = new Dictionary<string, int>();
|
|
|
|
foreach (var ticker in tickers)
|
|
{
|
|
try
|
|
{
|
|
_logger.LogInformation("Collecting ticker {Ticker} (run {RunId})", ticker, runId);
|
|
|
|
CollectionSnapshotRecord? cachedSnapshot = null;
|
|
if (IsMarketClosed())
|
|
{
|
|
var latest = await _readRepository.GetLatestSnapshotsForTickerAsync(ticker, 1);
|
|
var todayPrefix = UtcNowProvider().AddHours(9).ToString("yyyy-MM-dd");
|
|
if (latest.Count > 0 && latest[0].CapturedAt.StartsWith(todayPrefix))
|
|
{
|
|
cachedSnapshot = latest[0];
|
|
}
|
|
}
|
|
|
|
Dictionary<string, object> normalized;
|
|
string sourceName;
|
|
|
|
if (cachedSnapshot != null)
|
|
{
|
|
_logger.LogInformation("Cache hit for ticker {Ticker} (run {RunId})", ticker, runId);
|
|
normalized = JsonSerializer.Deserialize<Dictionary<string, object>>(cachedSnapshot.PayloadJson)
|
|
?? new Dictionary<string, object>();
|
|
sourceName = cachedSnapshot.SourceName.EndsWith(" (Cached)")
|
|
? cachedSnapshot.SourceName
|
|
: cachedSnapshot.SourceName + " (Cached)";
|
|
}
|
|
else
|
|
{
|
|
var kisResult = await kisSource.GetPriceDataAsync(ticker, account);
|
|
var seedRow = new Dictionary<string, object> { { "Ticker", ticker } };
|
|
var (norm, provenance) = _normalizer.NormalizeCollectionRow(seedRow, kisResult, null, false);
|
|
normalized = norm;
|
|
sourceName = (string)(provenance.GetValueOrDefault("source") ?? "kis_open_api");
|
|
}
|
|
|
|
// Save to DB
|
|
await _writeRepository.SaveSnapshotAsync(new CollectionSnapshotRecord(
|
|
RunId: runId,
|
|
DatasetName: "data_feed",
|
|
Ticker: ticker,
|
|
SourceName: sourceName,
|
|
PayloadJson: JsonSerializer.Serialize(normalized),
|
|
CapturedAt: DataNormalizationHelper.KstNowIso()
|
|
));
|
|
|
|
// Persist daily OHLCV bars
|
|
try
|
|
{
|
|
var today = UtcNowProvider().AddHours(9).ToString("yyyyMMdd");
|
|
var chartResult = await _kisApiClient.GetDailyItemChartPriceAsync(ticker, today, today, "D", account);
|
|
if (chartResult.TryGetValue("output2", out var output2Obj) && output2Obj is JsonElement output2Elem && output2Elem.ValueKind == JsonValueKind.Array)
|
|
{
|
|
foreach (var barElement in output2Elem.EnumerateArray())
|
|
{
|
|
if (!TryParseOhlcvBar(barElement, ticker, out var priceRecord))
|
|
{
|
|
_logger.LogWarning("Skipped invalid OHLCV bar for {Ticker}: constraints not satisfied", ticker);
|
|
continue;
|
|
}
|
|
await _writeRepository.SavePriceHistoryDailyAsync(priceRecord);
|
|
}
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Failed to persist price history for {Ticker} (run {RunId})", ticker, runId);
|
|
}
|
|
|
|
// Track source
|
|
if (!sourceCounts.ContainsKey(sourceName))
|
|
sourceCounts[sourceName] = 0;
|
|
sourceCounts[sourceName]++;
|
|
|
|
rows.Add(normalized);
|
|
result.SuccessCount++;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Collection failed for {Ticker} (run {RunId})", ticker, runId);
|
|
result.ErrorCount++;
|
|
errors.Add(new Dictionary<string, object>
|
|
{
|
|
{ "ticker", ticker },
|
|
{ "error", ex.Message },
|
|
{ "error_kind", ex.GetType().Name }
|
|
});
|
|
|
|
await _writeRepository.SaveErrorAsync(new CollectionErrorRecord(
|
|
RunId: runId,
|
|
SourceName: "kis_collector",
|
|
ErrorKind: ex.GetType().Name,
|
|
ErrorMessage: ex.Message,
|
|
Ticker: ticker
|
|
));
|
|
}
|
|
}
|
|
|
|
var finishedAt = DataNormalizationHelper.KstNowIso();
|
|
result.Status = result.ErrorCount == 0 ? "COMPLETED" : "COMPLETED_WITH_ERRORS";
|
|
result.FinishedAt = finishedAt;
|
|
result.SourceCounts = sourceCounts;
|
|
result.Rows = rows;
|
|
result.Errors = errors;
|
|
_auditTrail.Append("collection_audit", runId, new CollectionExecutionAudit(runId, result.Status, DateTimeOffset.Parse(startedAt), DateTimeOffset.Parse(finishedAt), result.SuccessCount, result.ErrorCount, "finished"));
|
|
|
|
// Save run record
|
|
await _writeRepository.SaveRunAsync(new CollectionRunRecord(
|
|
RunId: runId,
|
|
Status: result.Status,
|
|
StartedAt: startedAt,
|
|
FinishedAt: finishedAt,
|
|
TotalSnapshots: result.SuccessCount,
|
|
TotalErrors: result.ErrorCount
|
|
));
|
|
|
|
// Determine gate status
|
|
var gate = result.SuccessCount == 0 ? "FAIL"
|
|
: result.ErrorCount == 0 ? "PASS"
|
|
: result.ErrorCount < result.SuccessCount * 0.1 ? "PASS"
|
|
: "PASS_WITH_WARNINGS";
|
|
|
|
// Output JSON file to <repo>/Temp/kis_dotnet_collection_v1.json
|
|
var outputPath = GetOutputPath();
|
|
var outputData = new
|
|
{
|
|
formula_id = "KIS_DOTNET_COLLECTION_V1",
|
|
gate = gate,
|
|
run_id = runId,
|
|
started_at = startedAt,
|
|
finished_at = finishedAt,
|
|
summary = new
|
|
{
|
|
success_count = result.SuccessCount,
|
|
error_count = result.ErrorCount,
|
|
source_counts = sourceCounts
|
|
}
|
|
};
|
|
File.WriteAllText(outputPath, JsonSerializer.Serialize(outputData, new JsonSerializerOptions { WriteIndented = true }));
|
|
LogLineageEvent(runId, result.Status, result.SuccessCount, result.ErrorCount);
|
|
|
|
_logger.LogInformation("Collection run {RunId} finished with status {Status}: {Success} ok, {Errors} errors",
|
|
runId, result.Status, result.SuccessCount, result.ErrorCount);
|
|
|
|
return result;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "Collection run {RunId} failed with exception", runId);
|
|
result.Status = "FAILED";
|
|
result.FinishedAt = DataNormalizationHelper.KstNowIso();
|
|
result.ErrorMessage = ex.Message;
|
|
_auditTrail.Append("collection_audit", runId, new CollectionExecutionAudit(runId, result.Status, DateTimeOffset.Parse(startedAt), DateTimeOffset.Parse(result.FinishedAt), result.SuccessCount, result.ErrorCount, ex.Message));
|
|
return result;
|
|
}
|
|
}
|
|
|
|
private static bool TryParseOhlcvBar(JsonElement barElement, string ticker, out PriceHistoryDailyRecord priceRecord)
|
|
{
|
|
priceRecord = null!;
|
|
|
|
try
|
|
{
|
|
if (barElement.ValueKind != JsonValueKind.Object)
|
|
return false;
|
|
|
|
var dateStr = GetJsonElementProperty(barElement, "stck_bsop_date");
|
|
var openStr = GetJsonElementProperty(barElement, "stck_oprc");
|
|
var highStr = GetJsonElementProperty(barElement, "stck_hgpr");
|
|
var lowStr = GetJsonElementProperty(barElement, "stck_lwpr");
|
|
var closeStr = GetJsonElementProperty(barElement, "stck_clpr");
|
|
var volumeStr = GetJsonElementProperty(barElement, "acml_vol");
|
|
|
|
if (string.IsNullOrEmpty(dateStr) || string.IsNullOrEmpty(openStr) ||
|
|
string.IsNullOrEmpty(highStr) || string.IsNullOrEmpty(lowStr) ||
|
|
string.IsNullOrEmpty(closeStr) || string.IsNullOrEmpty(volumeStr))
|
|
return false;
|
|
|
|
if (!DateOnly.TryParseExact(dateStr, "yyyyMMdd", null, System.Globalization.DateTimeStyles.None, out var tradeDate))
|
|
return false;
|
|
|
|
if (!decimal.TryParse(openStr.Replace(",", ""), out var open) ||
|
|
!decimal.TryParse(highStr.Replace(",", ""), out var high) ||
|
|
!decimal.TryParse(lowStr.Replace(",", ""), out var low) ||
|
|
!decimal.TryParse(closeStr.Replace(",", ""), out var close) ||
|
|
!long.TryParse(volumeStr.Replace(",", ""), out var volume))
|
|
return false;
|
|
|
|
if (volume < 0)
|
|
return false;
|
|
|
|
if (high < low || high < open || high < close || low > open || low > close)
|
|
return false;
|
|
|
|
priceRecord = new PriceHistoryDailyRecord(
|
|
Ticker: ticker,
|
|
TradeDate: tradeDate,
|
|
Open: open,
|
|
High: high,
|
|
Low: low,
|
|
Close: close,
|
|
Volume: volume,
|
|
Source: "kis_open_api"
|
|
);
|
|
|
|
return true;
|
|
}
|
|
catch
|
|
{
|
|
return false;
|
|
}
|
|
}
|
|
|
|
private static string? GetJsonElementProperty(JsonElement element, string propertyName)
|
|
{
|
|
if (element.TryGetProperty(propertyName, out var prop))
|
|
{
|
|
if (prop.ValueKind == JsonValueKind.String)
|
|
return prop.GetString();
|
|
else if (prop.ValueKind == JsonValueKind.Number)
|
|
return prop.GetRawText();
|
|
}
|
|
return null;
|
|
}
|
|
|
|
private static string GetOutputPath()
|
|
{
|
|
var baseDir = AppContext.BaseDirectory;
|
|
var current = new DirectoryInfo(baseDir);
|
|
|
|
while (current != null)
|
|
{
|
|
if (Directory.Exists(Path.Combine(current.FullName, ".git"))
|
|
|| File.Exists(Path.Combine(current.FullName, "GatherTradingData.json")))
|
|
{
|
|
var tempDir = Path.Combine(current.FullName, "Temp");
|
|
Directory.CreateDirectory(tempDir);
|
|
return Path.Combine(tempDir, "kis_dotnet_collection_v1.json");
|
|
}
|
|
current = current.Parent;
|
|
}
|
|
|
|
return Path.Combine(Path.GetTempPath(), "kis_dotnet_collection_v1.json");
|
|
}
|
|
|
|
private bool IsMarketClosed()
|
|
{
|
|
// KST Time conversion (UTC+9)
|
|
var kst = UtcNowProvider().AddHours(9);
|
|
|
|
// Weekend check
|
|
if (kst.DayOfWeek == DayOfWeek.Saturday || kst.DayOfWeek == DayOfWeek.Sunday)
|
|
return true;
|
|
|
|
// Market hours check (09:00 - 15:30)
|
|
var time = kst.TimeOfDay;
|
|
if (time < new TimeSpan(9, 0, 0) || time > new TimeSpan(15, 30, 0))
|
|
return true;
|
|
|
|
return false;
|
|
}
|
|
|
|
private void LogLineageEvent(string runId, string status, int successCount, int errorCount)
|
|
{
|
|
try
|
|
{
|
|
var baseDir = AppContext.BaseDirectory;
|
|
var current = new DirectoryInfo(baseDir);
|
|
string? repoRoot = null;
|
|
|
|
while (current != null)
|
|
{
|
|
if (Directory.Exists(Path.Combine(current.FullName, ".git")))
|
|
{
|
|
repoRoot = current.FullName;
|
|
break;
|
|
}
|
|
current = current.Parent;
|
|
}
|
|
|
|
if (repoRoot != null)
|
|
{
|
|
var runtimeDir = Path.Combine(repoRoot, "runtime");
|
|
Directory.CreateDirectory(runtimeDir);
|
|
var lineagePath = Path.Combine(runtimeDir, "lineage_events.jsonl");
|
|
|
|
var ev = new
|
|
{
|
|
@event = "collection_run_completed",
|
|
run_id = runId,
|
|
status = status,
|
|
success_count = successCount,
|
|
error_count = errorCount,
|
|
timestamp = DataNormalizationHelper.KstNowIso()
|
|
};
|
|
|
|
File.AppendAllText(lineagePath, JsonSerializer.Serialize(ev) + "\n");
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Failed to write lineage event for run {RunId}", runId);
|
|
}
|
|
}
|
|
}
|