This commit is contained in:
2026-02-06 17:53:17 +09:00
parent 0b855188a9
commit aa1f3817bb
9 changed files with 274 additions and 316 deletions
+177 -253
View File
@@ -4,12 +4,17 @@ import TradingDecision
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.TimeoutCancellationException
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
import kotlinx.coroutines.withTimeout
import model.CandleData
import model.RankingStock
import model.RankingType
import network.DartCodeManager
import network.KisTradeService
@@ -17,296 +22,215 @@ import java.time.LocalDateTime
import java.time.LocalTime
import java.time.ZoneId
import java.time.format.DateTimeFormatter
import java.util.concurrent.atomic.AtomicLong
import kotlin.collections.List
import kotlin.math.*
// service/AutoTradingManager.kt
typealias TradingDecisionCallback = (TradingDecision?, Boolean)->Unit
object AutoTradingManager {
private val scope = CoroutineScope(Dispatchers.Default)
val targetStocks = mutableListOf<Pair<String, String>>()
// 자동 발굴 루프 제어용 Job
private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
private var discoveryJob: Job? = null
// 모니터링을 위한 상태 변수
private val lastTickTime = AtomicLong(System.currentTimeMillis())
private var watchdogJob: Job? = null
// 설정 상수
private const val MIN_RISE_RATE = 0.1
private const val MAX_RISE_RATE = 15.0
private const val CYCLE_TIMEOUT = 10 * 60 * 1000L // 한 사이클 최대 10분
private const val WATCHDOG_CHECK_INTERVAL = 30 * 1000L // 30초마다 생존 확인
private const val STUCK_THRESHOLD = 5 * 60 * 1000L // 5분간 반응 없으면 'Stuck'으로 판단
fun isRunning(): Boolean = discoveryJob?.isActive == true
/**
* 자동 발굴 루프 시작 및 Watchdog 실행
*/
fun startAutoDiscoveryLoop(tradeService: KisTradeService, callback: TradingDecisionCallback) {
if (isRunning()) return
// 1. 기존 Watchdog이 있다면 제거 후 새로 시작
watchdogJob?.cancel()
watchdogJob = scope.launch {
while (isActive) {
delay(WATCHDOG_CHECK_INTERVAL)
val now = System.currentTimeMillis()
if (isRunning() && (now - lastTickTime.get() > STUCK_THRESHOLD)) {
println("🚨 [Watchdog] 루프 멈춤 감지 (5분간 응답 없음). 강제 재시작합니다.")
restartLoop(tradeService, callback)
}
}
}
// 2. 메인 루프 실행
runDiscoveryLoop(tradeService, callback)
}
private fun runDiscoveryLoop(tradeService: KisTradeService, callback: TradingDecisionCallback) {
discoveryJob = scope.launch {
println("🚀 [AutoTrading] 발굴 루프 시작: ${LocalDateTime.now()}")
while (isActive) {
try {
lastTickTime.set(System.currentTimeMillis()) // 생존 신고
withTimeout(CYCLE_TIMEOUT) {
println("⏱️ [Cycle Start] ${LocalTime.now()}")
// [프로세스 1] 장 마감 및 잔고 체크
val now = LocalTime.now(ZoneId.of("Asia/Seoul"))
if (now.isAfter(LocalTime.of(15, 30)) && now.isBefore(LocalTime.of(15, 30))) {
executeClosingLiquidation(tradeService)
return@withTimeout
}
val balance = tradeService.fetchIntegratedBalance().getOrNull()
val myCash = balance?.deposit?.replace(",", "")?.toLongOrNull() ?: 0L
val myHoldings = balance?.holdings?.filter { it.quantity.toInt() > 0 }?.map { it.code }?.toSet() ?: emptySet()
val pendingStocks = DatabaseFactory.findAllMonitoringTrades().map { it.code }
// [프로세스 2] 후보군 수집
val candidates = fetchCandidates(tradeService)
.filter { (it.prdy_ctrt.toDoubleOrNull() ?: 0.0) in MIN_RISE_RATE..MAX_RISE_RATE }
.filter { it.code !in myHoldings && it.code !in pendingStocks }
.distinctBy { it.code }
// [프로세스 3] 종목별 순회 분석
candidates.forEach { stock ->
lastTickTime.set(System.currentTimeMillis()) // 종목별로도 생존 신고
processSingleStock(stock, myCash, tradeService, callback)
delay(300)
}
println("⏱️ [Cycle End] ${LocalTime.now()}")
}
} catch (e: TimeoutCancellationException) {
println("⏳ [Cycle Timeout] 사이클이 너무 길어져 초기화 후 재시작합니다.")
} catch (e: Exception) {
println("⚠️ [Loop Error] ${e.message}")
delay(10000)
}
waitForNextCycle(3)
}
}
}
private suspend fun processSingleStock(stock: RankingStock, myCash: Long, tradeService: KisTradeService, callback: TradingDecisionCallback) {
try {
// 개별 종목 분석은 최대 2분으로 제한
withTimeout(120000L) {
val corpInfo = DartCodeManager.getCorpCode(stock.code)
if (corpInfo?.cName.isNullOrEmpty()) {
// println("⏭️ [제외] ${stock.name}: 법인명 정보를 찾을 수 없음")
return@withTimeout
}
val dailyData = tradeService.fetchPeriodChartData(stock.code, "D", true).getOrNull() ?: return@withTimeout
val today = dailyData.lastOrNull() ?: return@withTimeout
val currentPrice = today.stck_prpr.toDouble()
if (currentPrice > myCash || currentPrice > 15000 || currentPrice < 900) return@withTimeout
println("🔍 [분석 진입] ${stock.name} (${LocalTime.now()})")
callback(TradingDecision().apply {
this.stockCode = stock.code
this.confidence = -1.0
this.stockName = stock.name
}, false)
val analyzer = coroutineScope {
val min30 = async { tradeService.fetchChartData(stock.code, true).getOrDefault(emptyList()) }
val weekly = async { tradeService.fetchPeriodChartData(stock.code, "W", true).getOrDefault(emptyList()) }
val monthly = async { tradeService.fetchPeriodChartData(stock.code, "M", true).getOrDefault(emptyList()) }
TechnicalAnalyzer().apply {
this.daily = dailyData
this.min30 = min30.await()
this.weekly = weekly.await()
this.monthly = monthly.await()
}
}
RagService.processStock(analyzer, stock.name, stock.code) { decision, isSuccess ->
callback(decision?.apply { this.currentPrice = currentPrice }, isSuccess)
}
println("✅ [분석 종료] ${stock.name} (${LocalTime.now()})")
}
} catch (e: Exception) {
println("❌ [Stock Error] ${stock.name}: ${e.message}")
}
}
private suspend fun fetchCandidates(tradeService: KisTradeService): List<RankingStock> = coroutineScope {
listOf(
async { tradeService.fetchMarketRanking(RankingType.VOLUME, true).getOrDefault(emptyList()) },
async { tradeService.fetchMarketRanking(RankingType.RISE, true).getOrDefault(emptyList()) },
async { tradeService.fetchMarketRanking(RankingType.FALL, true).getOrDefault(emptyList()) },
async { tradeService.fetchMarketRanking(RankingType.VALUE, true).getOrDefault(emptyList()) },
async { tradeService.fetchMarketRanking(RankingType.VOLUME_POWER, true).getOrDefault(emptyList()) },
// async { tradeService.fetchMarketRanking(RankingType.EXPECTED_RISE, true).getOrDefault(emptyList()) },
// async { tradeService.fetchMarketRanking(RankingType.COMPANY_TRADE, true).getOrDefault(emptyList()) }
).awaitAll().flatten()
}
private fun restartLoop(tradeService: KisTradeService, callback: TradingDecisionCallback) {
discoveryJob?.cancel()
startAutoDiscoveryLoop(tradeService, callback)
}
private suspend fun waitForNextCycle(minutes: Int) {
println("💤 대기 모드 진입...")
val endWait = System.currentTimeMillis() + (minutes * 60 * 1000L)
while (System.currentTimeMillis() < endWait && isRunning()) {
lastTickTime.set(System.currentTimeMillis()) // 대기 중에도 Watchdog에 생존 신고
delay(10000)
}
}
private suspend fun executeClosingLiquidation(tradeService: KisTradeService) {
// 1. DB에서 현재 감시 중인(보유 중인) 모든 종목 가져오기
val activeTrades = DatabaseFactory.findAllMonitoringTrades()
// 2. [추가] 실시간 증권사 잔고 조회 (실제 보유 주식인지 확인용)
val balanceResult = tradeService.fetchIntegratedBalance().getOrNull()
val realHoldings = balanceResult?.holdings
?.filter {
println("[${it.name}(${it.code})]: evalAmount ${it.evalAmount} , currentPrice : ${it.currentPrice} , ${it.quantity}")
it.quantity.toInt() > 0 && it.evalAmount.toDouble() > (it.currentPrice.toDouble() * it.quantity.toDouble()) } // 수량이 0보다 큰 것만
?.associateBy { it.code } ?: emptyMap()
val realHoldings = balanceResult?.holdings?.associateBy { it.code } ?: emptyMap()
activeTrades.forEach { trade ->
try {
// [검증] DB에는 MONITORING이지만 실제 잔고에는 없는 경우 처리
if (!realHoldings.containsKey(trade.code)) {
println("️ [제외] ${trade.name}: DB에는 감시 중이나 실제 잔고에 수량이 없어 스킵합니다.")
// 필요시 DB 상태를 COMPLETED 등으로 동기화
DatabaseFactory.updateStatusAndOrderNo(trade.id!!, TradeStatus.COMPLETED)
DatabaseFactory.updateStatusAndOrderNo(trade.id!!, TradeStatus.COMPLETED)
return@forEach
}
// 2. 수익 상태 먼저 체크 (현재가 조회)
val currentResult = tradeService.fetchChartData(trade.code, true).getOrNull()
val currentPrice = currentResult?.lastOrNull()?.stck_prpr?.toDouble() ?: 0.0
if (currentPrice > 0) {
// 매수가 역산 (목표가와 설정 수익률 기반)
val buyPrice = trade.targetPrice / (1 + trade.profitRate / 100.0)
val netProfitRate = ((currentPrice - buyPrice) / buyPrice * 100) - 0.3 // 수수료/세금 0.3% 차감
// 3. 매도 조건 판단 (최소 수익 0.1% 이상 확보 여부)
val isMinimumProfitSecured = netProfitRate >= 0.1
val isUrgent = LocalTime.now(ZoneId.of("Asia/Seoul")).isAfter(LocalTime.of(15, 20))
println("orderedPrice ${trade.orderedPrice}, currentPrice ${currentPrice} ")
// 수익이 났거나, 15:20분 이후 긴급 상황인 경우에만 진행
if (isMinimumProfitSecured || isUrgent) {
val reason = if (isUrgent) "시간 임박(탈출)" else "수익 확보(${String.format("%.2f", netProfitRate)}%)"
println("📢 [마감 정리 대상 포착] ${trade.name} | 사유: $reason")
// 4. [순서 변경] 수익 확인 후 기존 익절/손절 주문 취소 실행
// if (!trade.orderNo.isNullOrBlank()) {
// tradeService.cancelOrder(trade.orderNo, trade.code).onSuccess {
// println("✅ [취소 완료] ${trade.name} 기존 주문 취소됨")
// }.onFailure {
// println("️ [취소 건너뜀] ${trade.name}: ${it.message}")
// }
// }
//
// // 5. 즉시 시장가 매도 실행 (price를 "0"으로 전달)
// tradeService.postOrder(
// stockCode = trade.code,
// qty = trade.quantity.toString(),
// price = "0", // 시장가 주문
// isBuy = false
// ).onSuccess {
// DatabaseFactory.updateStatusAndOrderNo(trade.id!!, TradeStatus.COMPLETED)
// println("✨ [정리 완료] ${trade.name} 시장가 매도 성공")
// }.onFailure {
// println("❌ [매도 실패] ${trade.name}: ${it.message}")
// }
} else {
// 수익권이 아니면 그대로 유지 (기존 지정가 익절/손절 주문 유지)
// println("⏭️ [유지] ${trade.name}: 현재 수익권 아님 (${String.format("%.2f", netProfitRate)}%)")
}
}
// 마감 정리 로직 (필요 시 주석 해제하여 사용)
println("📢 [마감 정리 체크] ${trade.name}")
} catch (e: Exception) {
println("⚠️ [마감 정리 중 에러] ${trade.name}: ${e.message}")
println("⚠️ [마감 에러] ${trade.name}: ${e.message}")
}
delay(200) // API 호출 간격 조절
delay(200)
}
}
val MIN = 0.1
val MAX = 15.0
fun startAutoDiscoveryLoop(
tradeService: KisTradeService,
callback: TradingDecisionCallback
) {
if (discoveryJob?.isActive == true) return
discoveryJob = scope.launch {
println("🚀 [AutoTrading] 5분 주기 자동 발굴 시작")
while (discoveryJob?.isActive == true) {
try {
val now = LocalTime.now(ZoneId.of("Asia/Seoul"))
val isClosingTime = now.isAfter(LocalTime.of(15, 0)) && now.isBefore(LocalTime.of(15, 30))
if (isClosingTime) {
println("🕒 [장 마감 모드] 추가 매수를 중단하고 보유 종목 정리를 시작합니다.")
executeClosingLiquidation(tradeService) // 마감 정리 함수 호출
// 마감 중에는 1분 단위로 짧게 체크하며 대기
delay(60 * 1000)
continue
}
// 1. [체크] 현재 잔고 및 보유 종목 조회
val balanceResult = tradeService.fetchIntegratedBalance().getOrNull()
val myHoldings = balanceResult?.holdings?.filter { it.quantity.toInt() > 0 }?.map { it.code }?.toSet() ?: emptySet()
val pendingStocks = DatabaseFactory.findAllPendingBuyCodes()
val myCash = balanceResult?.deposit?.replace(",", "")?.toLongOrNull() ?: 0L
println("💰 보유 현금: ${String.format("%,d", myCash)}원 | 보유 종목 수: ${myHoldings.size}")
// 2. 랭킹 데이터 가져오기
// 1. 랭킹 데이터 가져오기 (비동기)
val volRankDeferred = async { tradeService.fetchMarketRanking(RankingType.VOLUME, true).getOrDefault(emptyList()) }
val riseRankDeferred = async { tradeService.fetchMarketRanking(RankingType.RISE, true).getOrDefault(emptyList()) }
// 거래대금(Amount) 상위 추가
val amountRankDeferred = async { tradeService.fetchMarketRanking(RankingType.VALUE, true).getOrDefault(emptyList()) }
val volumePowerDeferred = async { tradeService.fetchMarketRanking(RankingType.VOLUME_POWER, true).getOrDefault(emptyList()) }
val volList = volRankDeferred.await()
val riseList = riseRankDeferred.await()
val amountList = amountRankDeferred.await()
val volumeList = volumePowerDeferred.await()
// 3. 리스트 합치기 (중복 제거)
val candidates = (volList + riseList + amountList + volumeList + async { tradeService.fetchMarketRanking(RankingType.FOREIGNER_BUY, true).getOrDefault(emptyList()) + async { tradeService.fetchMarketRanking(RankingType.INSTITUTION_BUY, true).getOrDefault(emptyList()) }.await()}.await()).filter {stock ->
val rate = stock.prdy_ctrt.toDoubleOrNull() ?: 0.0
rate in MIN..MAX // 너무 과열되지 않은 주도주
}.filter { myHoldings.contains(it.code) == false && pendingStocks.contains(it.code) == false}.distinctBy { it.code }
println("🔎 1차 필터링 후보 ${candidates.size}개 (급등주 제외) 검증 시작...")
candidates.forEach { stock ->
try {
// [조건 1] 이미 보유한 종목 제외
var corpInfo = DartCodeManager.getCorpCode(stock.code)
if (corpInfo?.cName?.isNullOrEmpty() ?: true) {
println("⏭️ [제외] ${stock.name}: 법인명이 없음")
return@forEach
}
val currentPrice = stock.stck_prpr.replace(",", "").toDoubleOrNull() ?: 0.0
// [조건 2] 최소 1주 매수 가능 여부
if (currentPrice > myCash || currentPrice > 10000) {
println("⏭️ [제외] ${stock.name}: 주가($currentPrice)가 예산 초과")
return@forEach
}
callback(TradingDecision().apply {
this.stockCode = stock.code
this.confidence = -1.0
this.stockName = stock.name
}, false)
// 3. 일봉 데이터 조회 (필터링 용도 + TechnicalAnalyzer 입력용)
val dailyResult = tradeService.fetchPeriodChartData(stock.code, "D", true)
val dailyData = dailyResult.getOrNull()
val todayCandle = dailyData?.lastOrNull()
if (dailyData != null && todayCandle != null) {
val open = todayCandle.stck_oprc.toDoubleOrNull() ?: 0.0
val current = todayCandle.stck_prpr.toDoubleOrNull() ?: 0.0
if (open > 0) {
val riseRate = (current - open) / open * 100
// [조건 3] 상승 중(양봉)이면서 20% 이하 상승
if (riseRate > 0 && riseRate <= 20.0) {
println(
"✨ [발굴] ${stock.name} (+${
String.format(
"%.1f",
riseRate
)
}%) -> 데이터 수집 분석"
)
// [핵심 수정] AI 분석 전 필요한 차트 데이터(30분, 주봉, 월봉)를 모두 가져와 TechnicalAnalyzer에 주입
// 비동기로 동시에 요청하여 속도 향상
val min30Def = async {
tradeService.fetchChartData(stock.code, true).getOrDefault(emptyList())
}
val weekDef = async {
tradeService.fetchPeriodChartData(stock.code, "W", true)
.getOrDefault(emptyList())
}
val monthDef = async {
tradeService.fetchPeriodChartData(stock.code, "M", true)
.getOrDefault(emptyList())
}
val min30Data = min30Def.await()
val weeklyData = weekDef.await()
val monthlyData = monthDef.await()
// TechnicalAnalyzer 상태 업데이트 (싱글톤이므로 순차 처리 필수)
val t = TechnicalAnalyzer()
t.daily = dailyData
t.weekly = weeklyData
t.monthly = monthlyData
t.min30 = min30Data
try {
withTimeout(60000L) { // 60초 타임아웃 설정
RagService.processStock(t, stock.name, stock.code) { decision, isSuccess ->
if (decision != null) {
decision.stockName = stock.name
decision.currentPrice = current
}
callback(decision, isSuccess)
}
}
} catch (e: TimeoutCancellationException) {
println("⏳ [Timeout] ${stock.name} AI 분석 시간 초과로 스킵합니다.")
} catch (e: Exception) {
println("⚠️ [Error] AI 분석 중 오류: ${e.message}")
}
// 데이터 준비 완료 후 AI 분석 요청 (suspend 함수이므로 완료될 때까지 대기 -> 데이터 섞임 방지)
// 분석 후 잠시 대기 (서버 부하 조절)
delay(2000)
}
}
}
delay(300) // 종목 간 API 호출 간격
} catch (e: Exception) {
println("⚠️ [오류] ${stock.name} 분석 중 예외 발생: ${e.message}")
}
}
// --- 10초 주기 로그 대기 로직 시작 ---
val waitMinutes = 3
val totalWaitMillis = waitMinutes * 60 * 1000L
val tickMillis = 10 * 1000L
var currentWait = 0L
println("💤 사이클 종료. ${waitMinutes}분 대기...")
println("✅ 이번 사이클 분석 완료.")
while (currentWait < totalWaitMillis && discoveryJob?.isActive == true) {
delay(tickMillis)
currentWait += tickMillis
val leftSec = (totalWaitMillis - currentWait) / 1000
if (leftSec % 60 == 0L) {
val runtime = Runtime.getRuntime()
val usedMem = (runtime.totalMemory() - runtime.freeMemory()) / 1024 / 1024
println("📡 [System] 정상 작동 중... (남은 시간: ${leftSec}초 | 메모리 사용: ${usedMem}MB)")
}
}
} catch (e: Exception) {
println("⚠️ 루프 오류: ${e.message}")
delay(10000) // 오류 발생 시 10초 후 재시도
}
}
}
}
// 루프 중단 함수
fun stopDiscovery() {
discoveryJob?.cancel()
discoveryJob = null
println("🛑 [AutoTrading] 자동 발굴 중단됨")
}
// 기존 단일 종목 추가 로직 (유지)
fun addStock(technicalAnalyzer : TechnicalAnalyzer,stockName: String, stockCode: String, result: TradingDecisionCallback) {
scope.launch {
RagService.processStock(technicalAnalyzer,stockName, stockCode, result)
}
}
fun checkAndRestart(tradeService: KisTradeService, callback: TradingDecisionCallback) {
if (!isRunning()) {
println("⚠️ [Watchdog] 자동 발굴 루프가 중단된 것을 감지했습니다. 재시작을 시도합니다...")
startAutoDiscoveryLoop(tradeService, callback)
} else {
private fun executeOrder(code: String, type: String) {
// 실제 증권사 API 호출 로직 (한국투자증권, 키움 등)
println("🔥 [주문 집행] $code $type 완료")
}
}
}
data class InvestmentScores(
val ultraShort: Int, // 초단기 (분봉/에너지)