AI 자동매매 시스템 만들기 #3 - 키움증권 API 연동과 실전 거래

API 연동 삽질기 - 토큰 관리의 중요성

키움 REST API는 OAuth 2.0 방식을 씁니다. access_token과 refresh_token을 발급받아서 쓰는 구조입니다.

access_token은 24시간 유효, refresh_token은 30일 유효

 

문제는 토큰이 만료되면 자동매매가 멈춘다는 거죠. 그래서 토큰 갱신 로직을 반드시 구현해야 합니다.

@Service
class KiwoomAuthService(
    private val kiwoomProperties: KiwoomProperties,
    private val webClient: WebClient,
) {
    @Volatile
    private var accessToken: String? = null
    @Volatile
    private var refreshToken: String? = null
    @Volatile
    private var tokenExpiry: LocalDateTime? = null

    fun getAccessToken(): String {
        // 토큰이 없거나 만료 임박(10분 이내)이면 갱신
        if (accessToken == null || isTokenExpiringSoon()) {
            refreshAccessToken()
        }
        return accessToken ?: throw IllegalStateException("토큰 없음")
    }

    private fun isTokenExpiringSoon(): Boolean {
        val expiry = tokenExpiry ?: return true
        return LocalDateTime.now().plusMinutes(10).isAfter(expiry)
    }

    private fun refreshAccessToken() {
        val response = webClient.post()
            .uri("/oauth/token")
            .bodyValue(mapOf(
                "grant_type" to "refresh_token",
                "refresh_token" to refreshToken,
                "client_id" to kiwoomProperties.appKey,
                "client_secret" to kiwoomProperties.appSecret
            ))
            .retrieve()
            .bodyToMono(TokenResponse::class.java)
            .block()

        accessToken = response?.accessToken
        tokenExpiry = LocalDateTime.now().plusSeconds(response?.expiresIn ?: 86400)
    }
}

 

토큰 만료 10분 전에 미리 갱신하도록 했습니다. 장 중에 토큰이 만료되면 거래를 못하니까요!

API 호출 제한 - Rate Limiter 구현

키움 API에는 호출 제한이 있습니다:

• 초당 최대 20회
• 분당 최대 200회
• 시간당 최대 1,000회

이 제한을 넘으면 일시적으로 API 사용이 차단됩니다. 그래서 Rate Limiter를 만들었습니다:

@Component
class KiwoomRateLimiter {
    private val buckets = ConcurrentHashMap<String, RateLimitBucket>()

    fun acquire(apiId: String) {
        val bucket = buckets.computeIfAbsent(apiId) {
            RateLimitBucket(
                perSecond = 20,
                perMinute = 200,
                perHour = 1000
            )
        }

        while (!bucket.tryAcquire()) {
            Thread.sleep(50)  // 50ms 대기 후 재시도
        }
    }

    private class RateLimitBucket(
        private val perSecond: Int,
        private val perMinute: Int,
        private val perHour: Int,
    ) {
        private val secondWindow = LinkedList<Long>()
        private val minuteWindow = LinkedList<Long>()
        private val hourWindow = LinkedList<Long>()

        @Synchronized
        fun tryAcquire(): Boolean {
            val now = System.currentTimeMillis()

            // 오래된 기록 정리
            cleanup(secondWindow, now - 1000)
            cleanup(minuteWindow, now - 60000)
            cleanup(hourWindow, now - 3600000)

            // 제한 확인
            if (secondWindow.size >= perSecond) return false
            if (minuteWindow.size >= perMinute) return false
            if (hourWindow.size >= perHour) return false

            // 기록 추가
            secondWindow.add(now)
            minuteWindow.add(now)
            hourWindow.add(now)
            return true
        }

        private fun cleanup(window: LinkedList<Long>, threshold: Long) {
            while (window.isNotEmpty() && window.first < threshold) {
                window.removeFirst()
            }
        }
    }
}

 

슬라이딩 윈도우 방식으로 구현했습니다. API 호출 전에 acquire()를 호출하면 자동으로 제한을 지켜줍니다.

WebSocket으로 장 상태 실시간 감지

REST API만으로는 현재 장이 열려있는지 닫혀있는지 알 수 없습니다. 매번 API를 호출해서 확인하면 비효율적이고요.

그래서 WebSocket을 사용합니다. 키움은 실시간 시세와 장 상태를 WebSocket으로 제공합니다:

@Service
class KiwoomMarketStateService(
    private val kiwoomWebSocketClient: WebSocketClient,
    private val kiwoomWebSocketUrl: String,
    private val authService: KiwoomAuthService,
) {
    private val connected = AtomicBoolean(false)
    private val activePhases = ConcurrentHashMap.newKeySet<MarketPhase>()

    @PostConstruct
    fun init() {
        activePhases.add(MarketPhase.CLOSED)
        connect()
    }

    private fun connect() {
        kiwoomWebSocketClient.execute(URI.create(kiwoomWebSocketUrl)) { session ->
            handleSession(session)
        }
            .retryWhen(
                Retry.backoff(100, Duration.ofSeconds(5))
                    .maxBackoff(Duration.ofSeconds(60))
            )
            .subscribe()
    }

    private fun handleSession(session: WebSocketSession): Mono<Void> {
        connected.set(true)

        // LOGIN 전문 전송
        val loginMessage = buildLoginMessage()
        val sendMono = session.send(Mono.just(session.textMessage(loginMessage)))

        // 실시간 메시지 수신
        val receiveMono = session.receive()
            .map(WebSocketMessage::getPayloadAsText)
            .flatMap { payload -> processMessage(payload, session) }
            .then()

        return sendMono.thenMany(receiveMono).then()
    }
}

 

WebSocket이 끊기면 자동으로 재연결합니다. 최대 100회까지 5초 간격으로 재시도하고, 그래도 실패하면 텔레그램으로 알림을 보냅니다.

장 상태 코드 해석

키움 API는 장 상태를 숫자 코드로 보내줍니다:

• 215 = 장운영구분
  • 01: 장 시작 전
  • 02: 정규장 시작
  • 03: 정규장 마감
  • 04: 시간외 종가
  • 등등...

문제는 KRX(일반 주식), NXT(야간 거래), 시간외 거래가 각각 독립적으로 움직인다는 겁니다.

enum class MarketStateCode(val code: String, val description: String) {
    PRE_MARKET_ALERT("01", "장 시작 전"),
    MARKET_OPEN("02", "정규장 시작"),
    MARKET_CLOSE("03", "정규장 마감"),
    REGULAR_CLOSE("08", "정규장 마감 확정"),

    AFTER_HOURS_CLOSING_START("04", "시간외 종가 시작"),
    AFTER_HOURS_CLOSING_END("05", "시간외 종가 종료"),
    AFTER_HOURS_SINGLE_START("06", "시간외 단일가 시작"),
    AFTER_HOURS_SINGLE_END("07", "시간외 단일가 종료"),

    NXT_PRE_MARKET_START("11", "NXT 프리마켓 시작"),
    NXT_PRE_MARKET_END("12", "NXT 프리마켓 종료"),
    NXT_MAIN_MARKET_START("21", "NXT 메인마켓 시작"),
    NXT_MAIN_MARKET_END("22", "NXT 메인마켓 종료"),
    NXT_AFTER_MARKET_START("31", "NXT 에프터마켓 시작"),
    NXT_AFTER_MARKET_END("32", "NXT 에프터마켓 종료"),

    ALL_CLOSE("99", "전체 마감"),
    UNKNOWN("00", "알 수 없음");
}

여러 장이 동시에 열릴 수 있어서, Set으로 관리합니다:

private val activePhases = ConcurrentHashMap.newKeySet<MarketPhase>()

fun isKrxOpen(): Boolean = MarketPhase.KRX_OPEN in activePhases

fun isNxtOpen(): Boolean =
    MarketPhase.NXT_PRE_MARKET in activePhases ||
    MarketPhase.NXT_MAIN_MARKET in activePhases ||
    MarketPhase.NXT_AFTER_MARKET in activePhases

fun isAnyMarketOpen(): Boolean = isKrxOpen() || isNxtOpen()

이제 다른 서비스에서 간단하게 확인할 수 있습니다:

if (!marketStateService.isKrxOpen()) {
    logger.info { "장 마감 중 — 주문 스킵" }
    return
}

 

실제 주문 실행

드디어 실제로 주문을 넣는 코드를 만들 차례입니다.

@Service
class KiwoomOrderService(
    private val apiClient: KiwoomApiClient,
    private val marketStateService: KiwoomMarketStateService,
) {
    fun buyStock(
        accountNumber: String,
        stockCode: String,
        quantity: Int,
        price: Long,
        exchangeType: ExchangeType = ExchangeType.KRX,
    ): String? {
        // 1. 장 상태 확인
        if (!marketStateService.canTrade(exchangeType)) {
            logger.warn { "[$stockCode] 현재 거래 불가 상태 (exchange=$exchangeType)" }
            return null
        }

        // 2. 주문 가능 금액 확인
        val deposit = accountService.getDeposit()
        val orderableAmount = KiwoomNumberParser.parseLong(deposit?.output?.orderableAmount)
        val totalCost = price * quantity

        if (totalCost > orderableAmount) {
            logger.warn { "[$stockCode] 주문 가능 금액 부족: $totalCost > $orderableAmount" }
            return null
        }

        // 3. 주문 실행
        val response = apiClient.callApi<KiwoomOrderResponse>(
            apiId = "ka10007",
            uri = "/api/dotr/order",
            body = mapOf(
                "acno" to accountNumber,
                "stk_cd" to stockCode,
                "buy_sell_tp" to "1",  // 1=매수
                "ord_qty" to quantity.toString(),
                "ord_pric" to price.toString(),
                "ord_tp" to "00",  // 00=지정가
                "excg_tp" to exchangeType.code,
            ),
            label = "매수 주문 [$stockCode]",
        )

        return response?.orderNumber
    }
}

주문 전에 반드시 확인하는 것들:

1. 장이 열려있는가?
2. 주문 가능 금액이 충분한가?
3. 입력값이 올바른가? (음수, 0 등 체크)

모의투자 vs 실전 - Profile로 분리

처음부터 실전 계좌로 테스트할 수는 없습니다. 그래서 Spring Profile로 환경을 분리했습니다:

spring:
  profiles:
    active: mock  # 또는 real

kiwoom:
  mock:
    url: https://mock.kiwoom.com
    app-key: ${MOCK_APP_KEY}
  real:
    url: https://api.kiwoom.com
    app-key: ${REAL_APP_KEY}

 

애플리케이션 시작 시 Profile을 검증합니다:

@Component
class ProfileValidator(
    private val environment: Environment
) : ApplicationRunner {
    override fun run(args: ApplicationArguments) {
        val profiles = environment.activeProfiles
        val validProfiles = setOf("mock", "real")

        if (profiles.none { it in validProfiles }) {
            logger.error { "잘못된 프로파일: $profiles" }
            logger.error { "반드시 'mock' 또는 'real' 프로파일을 지정하세요" }
            exitProcess(1)
        }

        logger.info { "실행 환경: ${profiles.joinToString()}" }
    }
}

 

프로파일을 안 주거나 잘못 주면 아예 실행을 안 합니다.

주문 체결 추적 - WebSocket 활용

주문을 넣으면 바로 체결되는 게 아닙니다. 시장 상황에 따라 몇 초 ~ 몇 분 걸릴 수 있죠.

WebSocket으로 주문 체결 상태를 실시간으로 받을 수 있습니다:

@Service
class KiwoomOrderTrackingService(
    private val marketStateService: KiwoomMarketStateService,
) {
    private val pendingOrders = ConcurrentHashMap<String, OrderStatus>()

    @PostConstruct
    fun init() {
        // 주문체결 실시간 리스너 등록
        marketStateService.addOrderExecutionListener { data ->
            handleExecutionUpdate(data)
        }
    }

    fun trackOrder(orderNumber: String, stockCode: String, orderType: String) {
        pendingOrders[orderNumber] = OrderStatus(
            stockCode = stockCode,
            orderType = orderType,
            status = "대기중",
            createdAt = LocalDateTime.now()
        )
    }

    private fun handleExecutionUpdate(data: JsonNode) {
        val orderNumber = data.get("values")?.get("주문번호")?.asText() ?: return
        val status = data.get("values")?.get("체결여부")?.asText() ?: return

        val order = pendingOrders[orderNumber] ?: return

        when (status) {
            "체결" -> {
                logger.info { "[${order.stockCode}] 주문 체결 완료: $orderNumber" }
                pendingOrders.remove(orderNumber)
                // DB에 거래 이력 저장
                saveTradeHistory(order, data)
            }
            "취소" -> {
                logger.warn { "[${order.stockCode}] 주문 취소됨: $orderNumber" }
                pendingOrders.remove(orderNumber)
            }
        }
    }
}

 

주문을 넣으면 추적 목록에 등록하고, WebSocket으로 체결 통지가 오면 DB에 저장합니다.

에러 처리 - 실패에 대비하기

API 호출은 언제든 실패할 수 있습니다:

• 네트워크 오류
• API 서버 장애
• 토큰 만료
• 일시적인 과부하

그래서 모든 API 호출을 try-catch로 감싸고, 실패 시 로그를 남깁니다:

inline fun <reified T> callApi(
    apiId: String,
    uri: String,
    body: Map<String, Any>,
    label: String,
): T? {
    return try {
        rateLimiter.acquire(apiId)

        val response = webClient.post()
            .uri(uri)
            .header("Authorization", "Bearer ${authService.getAccessToken()}")
            .bodyValue(body)
            .retrieve()
            .bodyToMono(T::class.java)
            .timeout(Duration.ofSeconds(10))
            .block()

        logger.debug { "$label 성공" }
        response
    } catch (e: WebClientResponseException) {
        logger.error { "$label 실패: HTTP ${e.statusCode} - ${e.responseBodyAsString}" }
        null
    } catch (e: TimeoutException) {
        logger.error { "$label 타임아웃 (10초 초과)" }
        null
    } catch (e: Exception) {
        logger.error(e) { "$label 오류" }
        null
    }
}

 

10초 타임아웃을 걸어서 무한 대기를 방지하고, 실패하면 null을 반환합니다.

실제 커밋 히스토리

이 기능들을 만들면서 남긴 커밋들:

b578174 - Add improved error handling and thread-safe token management
  └─ 토큰 갱신 로직 개선
     스레드 안전성 확보
     에러 처리 강화

2a4a94f - Add Kiwoom WebSocket for real-time market state updates
  └─ WebSocket 연결 구현
     장 상태 실시간 감지
     KRX/NXT/SOR 거래소 로직

 

WebSocket 구현이 생각보다 까다로웠습니다. 연결이 끊겼다 재연결할 때 LOGIN부터 다시 해야 하는 등의 디테일이 많았거든요.

배운 점들

1. 금융 API는 엄격하다

토큰 만료, 호출 제한, 장 시간 제약 등 지켜야 할 규칙이 많습니다. 하나라도 놓치면 거래가 안 되거나 계정이 일시 정지될 수 있어요.

 

2. 실시간 연결은 불안정하다

WebSocket 연결은 언제든 끊길 수 있습니다. 자동 재연결 로직은 필수입니다.

 

3. Profile 분리로 실수 방지

코드 한 줄만 바꿔서 mock↔real을 전환하면 위험합니다. 명시적인 Profile 설정으로 실수를 방지해야 합니다.

 

4. 로그가 생명이다

거래 로직에서 뭔가 잘못되면 로그를 보고 추적해야 합니다. 모든 주요 동작을 로그로 남기세요.

다음 글 예고

오늘은 키움증권 API 연동과 실제 거래 구현에 대해 이야기했습니다. 다음 글에서는:

• AI가 실제로 어떻게 매매 판단을 내리는지
• 기술적 지표 계산 로직
• 백테스팅으로 전략 검증하기
• 실전 투자 결과와 개선 사항

이런 내용들을 다뤄볼 예정입니다. 실제로 AI가 어떤 판단을 내렸고 결과는 어땠는지 공유하겠습니다!