한국투자증권 API, 3분봉, 5분봉, 10분봉… 원하는 분봉 자유롭게 만드는 방법

한국투자증권 API는 주식 자동매매를 위해 유용한 기능을 제공합니다. 리눅스 환경에서도 사용할 수 있어서, 내 컴퓨터가 아닌 아마존 서버를 활용할 수도 있습니다. 그런데 안타깝게도 한국투자증권 API는 당일에 한해 1분봉 데이터만 제공하며, 그것도 30개씩만 가져올 수 있습니다. 지난 포스팅에 이어 이번에는 1분봉 데이터로 다른 종류의 분봉 OHLCV 데이터를 만드는 방법에 대해 알아보겠습니다. 파이썬 판다스 패키지가 얼마나 유용한지 확인해보시기 바랍니다.

글의 순서

한국투자증권 API로 1분봉 데이터 가져오기
1분봉으로 다양한 분봉 OHLCV 데이터 만들기
파이썬 코드 : 1분봉 OHLCV 데이터로 3분봉, 5분봉, 10분봉… 원하는 분봉 데이터 만들기

한국투자증권 API로 1분봉 데이터 가져오기

한국투자증권 API는 일, 주, 월의 캔들을 가져올 수 있습니다. 그러나 분봉 데이터의 경우 당일(오늘)만 가능합니다. 그것도 1분봉만 가능하며, 한번에 30건만 조회할 수 있습니다. 그러니까 시간을 입력하면, 그 시간 기준으로 30분 전의 데이터부터 가져옵니다. 따라서 오늘 전체 시간인 09:00~15:30의 데이터를 가져오려면 15:30부터 30분씩 줄여가며 09:00시까지 거슬러 올라가야 합니다. 지난 포스팅, 한국투자증권 API로 1분봉 OHLCV 데이터 가져오기 실습코드의 decrease_time 함수에서 이를 구현하였습니다.

1분봉으로 3분봉, 5분봉, 10분봉… 원하는 분봉 OHLCV 데이터 만들기

1분봉으로 3분봉, 5분봉, 15분봉 등 다양한 분봉을 만드는 원리는 간단합니다. 1분봉 데이터를 묶어서 Open, High, Low, Close, Volume 데이터를 재계산하기만 하면 되니 말입니다. 파이썬의 경우, 판다스 패키지의 resample 함수를 사용하면 간단하게 해결됩니다. 예를 들어, resample(‘3T’)을 사용한다면, 3분 단위로 데이터를 그룹화하며, OHLCV 값을 다시 계산합니다.

resample_ohlcv(df, freq) 함수

1분봉 데이터를 가져온 후 이를 3분봉, 5분봉, 60분봉 등 다른 분봉 데이터로 바꾸는데 핵심적인 역할을 하는 함수가 resample_ohlcv(df, freq) 함수입니다. 이 함수는 1분 단위로 이미 모아둔 OHLCV 데이터(df)를 원하는 시간 간격(freq)으로 묶어서 새로운 OHLCV를 만들어주는 역할을 합니다.

resample() 함수의 freq 인자는 Pandas 내부에서 미리 정의해둔 시간 간격 문자열을 사용합니다. 이를 “오프셋(alias)”이라고 부르는데, 예를 들어 아래와 같은 문자열들이 있습니다.
▶T 또는 min : 분(minute)
▶H : 시간(hour)
▶D : 일(day)
▶W : 주(week)
▶M : 달(month)
▶Y 또는 A : 연(year)
▶1H, 5T 같은 식으로 숫자 + 단위 조합도 가능
‘5T’는 “5분 간격”, ‘1H’는 “1시간 간격” 등처럼 Pandas가 인식하는 형식대로 지정해주면 됩니다. 만약 더 복잡한 패턴이 필요하다면 Pandas의 DateOffset 객체를 직접 만들어서 사용할 수도 있어요. 즉, 기본적으로 freq=’3T’처럼 쓰면 3분 단위로 데이터가 묶이고, ’60T’나 ‘1H’처럼 쓰면 1시간 단위로 묶이는 식으로 동작합니다. 이런 문자열들은 Pandas가 미리 인식하도록 만들어놓은 것이라서 직접 함수를 구현할 필요 없이 편리하게 사용할 수 있습니다.

파이썬 코드 : 1분봉 OHLCV 데이터로 3분봉, 5분봉, 10분봉… 원하는 분봉 데이터 만들기

실습코드에 1분봉 데이터를 가져온 후 이를 3분봉, 5분봉 등으로 변환시킬 수 있는 방법을 나타내었습니다. 실습코드에서 time_end라는 변수에 시간을 입력하면, 그 시간부터 시간을 거꾸로 거슬러 올라가서 30개의 데이터를 받아옵니다. 한번에 30개만 받을 수 있으므로, 09:00부터 15:30까지의 데이터를 모두 받기 위해서는 time_end라는 변수를 자동으로 30분씩 당겨서 입력하게 하면 됩니다. decrease_time() 함수가 이를 담당해줍니다.

1분봉 데이터를 가져온 후 이를 3분봉, 5분봉, 60분봉 등 다른 분봉 데이터로 바꾸는데 핵심적인 역할을 하는 함수가 resample_ohlcv(df, freq) 함수입니다. 이 함수는 1분 단위로 이미 모아둔 OHLCV 데이터(df)를 원하는 시간 간격(freq)으로 묶어서 새로운 OHLCV를 만들어주는 역할을 합니다. 만약 resample_ohlcv() 함수의 파리미터 freq를 freq=’3T’와 같은 방식으로 넘겨주면 1분봉 데이터를 3분봉으로 파이썬 판다스 패키지가 바꿔줍니다.

1분봉을 3분봉으로 바꿔주는 핵심함수인 resample_ohlcv(df, freq)의 작동 절차를 정리하면 아래와 같습니다.

step1) df.set_index(‘DateTime’)
DataFrame의 DateTime 열을 인덱스로 설정해줍니다.
resample() 메서드를 사용하려면 DateTime 타입의 열을 인덱스로 지정해야하기 때문입니다.

step2) df.resample(freq).agg({…})
여기서 freq는 ‘3T'(3분), ‘5T'(5분), ’10T'(10분), ’60T'(60분)처럼 Pandas가 이해할 수 있는 시간주기를 의미합니다.
resample(‘3T’)면 3분 간격으로 데이터를 묶어주고, resample(’60T’)면 60분 간격(1시간)으로 묶어줍니다.

.agg({…}) 부분에서는 어떤 식으로 묶을지(집계) 방법을 정해줍니다.
▶Open(시가) : 구간 시작값 (첫 번째 값)
▶High(고가) : 구간 내 최고값
▶Low(저가) : 구간 내 최저값
▶Close(종가) : 구간 마지막값 (마지막 값)
▶Volume(거래량) : 구간 합계

step3) resampled.dropna(how=’any’)
묶으면서 생긴 결측치(NaN)를 제거합니다. 데이터가 없는 구간이 있다면 dropna를 이용하여 NaN을 제거할 수 있습니다.
필요에 따라 dropna() 대신에 다른 방법으로 결측치를 처리할 수도 있습니다.

step4) resampled.reset_index()
인덱스로 사용했던 DateTime을 다시 컬럼으로 만들어주기 위해 reset_index()를 사용합니다.

step1) ~ 4)의 절차를 통해 이 함수가 반환하는 값은 freq 단위로 묶인 새로운 OHLCV DataFrame입니다.

import os
import requests
import json
import datetime
import pandas as pd

# KIS API 설정
api_key = os.getenv('KIS_APP_KEY')
api_secret = os.getenv('KIS_APP_SECRET')

# KIS API 기본 설정
URL_BASE = "https://openapi.koreainvestment.com:9443"

# 액세스 토큰 발급 함수
def get_access_token():
    """KIS API를 위한 액세스 토큰 발급"""
    headers = {"content-type": "application/json"}
    body = {
        "grant_type": "client_credentials",
        "appkey": api_key,
        "appsecret": api_secret
    }
    PATH = "/oauth2/tokenP"
    URL = f"{URL_BASE}{PATH}"
   
    res = requests.post(URL, headers=headers, data=json.dumps(body))
   
    if res.status_code == 200:
        access_token = res.json()["access_token"]
        return access_token
    else:
        print(f"Failed to get access token: {res.json()}")
        return None

# time_end를 minutes만큼 줄여주는 함수
def decrease_time(time_str, minutes):
    """time_str을 분 단위로 감소시켜서 새로운 시간을 반환"""
    time_format = '%H%M%S'
    dt = datetime.datetime.strptime(time_str, time_format)
    dt -= datetime.timedelta(minutes=minutes)
    return dt.strftime(time_format)

# 당일 분봉 OHLCV 데이터 조회 함수
def get_minute_ohlcv_data(access_token, code, time_end="153000"):
    """당일 분봉 OHLCV 데이터를 조회하고 DataFrame으로 반환"""
    PATH = "/uapi/domestic-stock/v1/quotations/inquire-time-itemchartprice"
    URL = f"{URL_BASE}{PATH}"

    all_data = []
   
    # 반복적으로 데이터를 요청하면서 time_end 값을 줄여가며 데이터를 받아옴
    while True:
        # 현재 날짜 설정
        # today = datetime.datetime.now().strftime('%Y%m%d')

        headers = {
            "Content-Type": "application/json",
            "authorization": f"Bearer {access_token}",
            "appKey": api_key,
            "appSecret": api_secret,
            "tr_id": "FHKST03010200",  # 당일 분봉 데이터 조회용 거래 ID
        }

        params = {
            "FID_ETC_CLS_CODE": "",               # 기타 구분 코드
            "FID_COND_MRKT_DIV_CODE": "J",        # 시장 분류 코드
            "FID_INPUT_ISCD": code,               # 종목 코드
            "FID_INPUT_HOUR_1": time_end,         # 조회 종료 시간 (HHMMSS)
            "FID_PW_DATA_INCU_YN": "N"            # 과거 데이터 포함 여부 (N: 당일만 조회)
        }

        res = requests.get(URL, headers=headers, params=params)
       
        if res.status_code == 200:
            res_data = res.json()
            data = res_data.get('output2', [])
            if data:
                # OHLCV 데이터를 DataFrame으로 변환
                ohlcv_data = [{
                    "Date": item['stck_bsop_date'],  # 거래일자
                    "Time": item['stck_cntg_hour'],  # 체결 시간
                    "Open": item['stck_oprc'],       # 시가
                    "High": item['stck_hgpr'],       # 고가
                    "Low": item['stck_lwpr'],        # 저가
                    "Close": item['stck_prpr'],      # 종가
                    "Volume": item['cntg_vol']       # 체결 거래량
                } for item in data]

                all_data.extend(ohlcv_data)

                # 30개의 데이터를 받았으면 time_end를 줄여서 다음 데이터를 요청
                time_end = decrease_time(time_end, minutes=30)
                print(f"Next time_end: {time_end}")
            else:
                print("No more data available for the specified time.")
                break
        else:
            print(f"Failed to retrieve data: {res.status_code}, {res.text}")
            return None

    # DataFrame 생성 및 정렬
    df = pd.DataFrame(all_data)
    df['DateTime'] = pd.to_datetime(df['Date'] + df['Time'], format='%Y%m%d%H%M%S')  # 날짜와 시간 결합
    df = df.astype({'Open': 'float', 'High': 'float', 'Low': 'float', 'Close': 'float', 'Volume': 'float'})  # 데이터 타입 변환
    df = df[['DateTime', 'Open', 'High', 'Low', 'Close', 'Volume']]  # 열 정렬
    df = df.sort_values(by='DateTime', ascending=True)  # 시간 오름차순 정렬
    df = df.reset_index(drop=True)  # 인덱스를 다시 설정

    return df

def resample_ohlcv(df, freq='3T'):
    """
    분 단위 OHLCV 데이터(DataFrame)를 원하는 주기로 묶어서(resample) 새 OHLCV로 반환.
    
    freq 예시
    - '3T': 3분 간격
    - '5T': 5분 간격
    - '10T': 10분 간격
    - '60T' or '1H': 60분 간격
    """
    # 1) DateTime을 인덱스로 설정
    df = df.set_index('DateTime')

    # 2) resample()을 이용해 묶기
    #    Open: 구간의 첫 시가
    #    High: 구간 중 최고가
    #    Low:  구간 중 최저가
    #    Close: 구간의 마지막 종가
    #    Volume: 구간 거래량 합계
    resampled = df.resample(freq).agg({
        'Open': 'first',
        'High': 'max',
        'Low': 'min',
        'Close': 'last',
        'Volume': 'sum'
    })

    # 3) NaN 제거 혹은 처리
    #    시작 시점에 묶는 과정에서 빈 구간이 생길 수 있으니 필요시 dropna() 처리
    resampled = resampled.dropna(how='any')

    # 4) 인덱스를 다시 컬럼으로
    resampled = resampled.reset_index()

    return resampled


# 메인 실행 부분
if __name__ == "__main__":
    access_token = get_access_token()
    if access_token:

        # 종목, 조회 종료 시간 지정 (예: 15:30:00까지 데이터를 받아오기 시작)
        code_stock_krx = '005380' # 예) 현대차  
        time_end = "153000"       # 종료 시간

        df_1min = get_minute_ohlcv_data(access_token, code=code_stock_krx, time_end=time_end)
        if df_1min is not None:
            print("===== 1분봉 데이터 =====")
            print(df_1min.head())

            # 3분봉으로 변환
            df_3min = resample_ohlcv(df_1min, freq='3T')
            print("===== 3분봉 데이터 =====")
            print(df_3min.head())

    else:
        print("Failed to get access token.")

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

import os

import requests

import json

import datetime

import pandas as pd

# KIS API 설정

api_key = os.getenv('KIS_APP_KEY')

api_secret = os.getenv('KIS_APP_SECRET')

# KIS API 기본 설정

URL_BASE = "https://openapi.koreainvestment.com:9443"

# 액세스 토큰 발급 함수

def get_access_token():

"""KIS API를 위한 액세스 토큰 발급"""

headers = {"content-type": "application/json"}

body = {

"grant_type": "client_credentials",

"appkey": api_key,

"appsecret": api_secret

}

PATH = "/oauth2/tokenP"

URL = f"{URL_BASE}{PATH}"

res = requests.post(URL, headers=headers, data=json.dumps(body))

if res.status_code == 200:

access_token = res.json()["access_token"]

return access_token

else:

print(f"Failed to get access token: {res.json()}")

return None

# time_end를 minutes만큼 줄여주는 함수

def decrease_time(time_str, minutes):

"""time_str을 분 단위로 감소시켜서 새로운 시간을 반환"""

time_format = '%H%M%S'

dt = datetime.datetime.strptime(time_str, time_format)

dt -= datetime.timedelta(minutes=minutes)

return dt.strftime(time_format)

# 당일 분봉 OHLCV 데이터 조회 함수

def get_minute_ohlcv_data(access_token, code, time_end="153000"):

"""당일 분봉 OHLCV 데이터를 조회하고 DataFrame으로 반환"""

PATH = "/uapi/domestic-stock/v1/quotations/inquire-time-itemchartprice"

URL = f"{URL_BASE}{PATH}"

all_data = []

# 반복적으로 데이터를 요청하면서 time_end 값을 줄여가며 데이터를 받아옴

while True:

# 현재 날짜 설정

# today = datetime.datetime.now().strftime('%Y%m%d')

headers = {

"Content-Type": "application/json",

"authorization": f"Bearer {access_token}",

"appKey": api_key,

"appSecret": api_secret,

"tr_id": "FHKST03010200", # 당일 분봉 데이터 조회용 거래 ID

}

params = {

"FID_ETC_CLS_CODE": "", # 기타 구분 코드

"FID_COND_MRKT_DIV_CODE": "J", # 시장 분류 코드

"FID_INPUT_ISCD": code, # 종목 코드

"FID_INPUT_HOUR_1": time_end, # 조회 종료 시간 (HHMMSS)

"FID_PW_DATA_INCU_YN": "N" # 과거 데이터 포함 여부 (N: 당일만 조회)

}

res = requests.get(URL, headers=headers, params=params)

if res.status_code == 200:

res_data = res.json()

data = res_data.get('output2', [])

if data:

# OHLCV 데이터를 DataFrame으로 변환

ohlcv_data = [{

"Date": item['stck_bsop_date'], # 거래일자

"Time": item['stck_cntg_hour'], # 체결 시간

"Open": item['stck_oprc'], # 시가

"High": item['stck_hgpr'], # 고가

"Low": item['stck_lwpr'], # 저가

"Close": item['stck_prpr'], # 종가

"Volume": item['cntg_vol'] # 체결 거래량

} for item in data]

all_data.extend(ohlcv_data)

# 30개의 데이터를 받았으면 time_end를 줄여서 다음 데이터를 요청

time_end = decrease_time(time_end, minutes=30)

print(f"Next time_end: {time_end}")

else:

print("No more data available for the specified time.")

break

else:

print(f"Failed to retrieve data: {res.status_code}, {res.text}")

return None

# DataFrame 생성 및 정렬

df = pd.DataFrame(all_data)

df['DateTime'] = pd.to_datetime(df['Date'] + df['Time'], format='%Y%m%d%H%M%S') # 날짜와 시간 결합

df = df.astype({'Open': 'float', 'High': 'float', 'Low': 'float', 'Close': 'float', 'Volume': 'float'}) # 데이터 타입 변환

df = df[['DateTime', 'Open', 'High', 'Low', 'Close', 'Volume']] # 열 정렬

df = df.sort_values(by='DateTime', ascending=True) # 시간 오름차순 정렬

df = df.reset_index(drop=True) # 인덱스를 다시 설정

return df

def resample_ohlcv(df, freq='3T'):

"""

분 단위 OHLCV 데이터(DataFrame)를 원하는 주기로 묶어서(resample) 새 OHLCV로 반환.

freq 예시

- '3T': 3분 간격

- '5T': 5분 간격

- '10T': 10분 간격

- '60T' or '1H': 60분 간격

"""

# 1) DateTime을 인덱스로 설정

df = df.set_index('DateTime')

# 2) resample()을 이용해 묶기

# Open: 구간의 첫 시가

# High: 구간 중 최고가

# Low: 구간 중 최저가

# Close: 구간의 마지막 종가

# Volume: 구간 거래량 합계

resampled = df.resample(freq).agg({

'Open': 'first',

'High': 'max',

'Low': 'min',

'Close': 'last',

'Volume': 'sum'

})

# 3) NaN 제거 혹은 처리

# 시작 시점에 묶는 과정에서 빈 구간이 생길 수 있으니 필요시 dropna() 처리

resampled = resampled.dropna(how='any')

# 4) 인덱스를 다시 컬럼으로

resampled = resampled.reset_index()

return resampled

# 메인 실행 부분

if __name__ == "__main__":

access_token = get_access_token()

if access_token:

# 종목, 조회 종료 시간 지정 (예: 15:30:00까지 데이터를 받아오기 시작)

code_stock_krx = '005380' # 예) 현대차

time_end = "153000" # 종료 시간

df_1min = get_minute_ohlcv_data(access_token, code=code_stock_krx, time_end=time_end)

if df_1min is not None:

print("===== 1분봉 데이터 =====")

print(df_1min.head())

# 3분봉으로 변환

df_3min = resample_ohlcv(df_1min, freq='3T')

print("===== 3분봉 데이터 =====")

print(df_3min.head())

else:

print("Failed to get access token.")

Next time_end: 150000
Next time_end: 143000
Next time_end: 140000
Next time_end: 133000
Next time_end: 130000
Next time_end: 123000
Next time_end: 120000
Next time_end: 113000
Next time_end: 110000
Next time_end: 103000
Next time_end: 100000
Next time_end: 093000
Next time_end: 090000
Next time_end: 083000
No more data available for the specified time.
===== 1분봉 데이터 =====
             DateTime      Open      High       Low     Close   Volume
0 2025-01-24 09:00:00  208000.0  208500.0  207000.0  207500.0  17025.0
1 2025-01-24 09:01:00  207500.0  208000.0  207500.0  207500.0   4199.0
2 2025-01-24 09:02:00  207500.0  208000.0  207500.0  208000.0   2021.0
3 2025-01-24 09:03:00  207500.0  207500.0  207000.0  207000.0  13711.0
4 2025-01-24 09:04:00  207000.0  207500.0  207000.0  207000.0   1945.0
===== 3분봉 데이터 =====
             DateTime      Open      High       Low     Close   Volume
0 2025-01-24 09:00:00  208000.0  208500.0  207000.0  208000.0  23245.0
1 2025-01-24 09:03:00  207500.0  207500.0  207000.0  207000.0  16434.0
2 2025-01-24 09:06:00  207000.0  207500.0  206000.0  207000.0  30947.0
3 2025-01-24 09:09:00  207000.0  207500.0  206500.0  207000.0  10061.0
4 2025-01-24 09:12:00  207500.0  207500.0  206500.0  207000.0  17399.0

Next time_end: 150000

Next time_end: 143000

Next time_end: 140000

Next time_end: 133000

Next time_end: 130000

Next time_end: 123000

Next time_end: 120000

Next time_end: 113000

Next time_end: 110000

Next time_end: 103000

Next time_end: 100000

Next time_end: 093000

Next time_end: 090000

Next time_end: 083000

No more data available for the specified time.

===== 1분봉 데이터 =====

DateTime Open High Low Close Volume

0 2025-01-24 09:00:00 208000.0 208500.0 207000.0 207500.0 17025.0

1 2025-01-24 09:01:00 207500.0 208000.0 207500.0 207500.0 4199.0

2 2025-01-24 09:02:00 207500.0 208000.0 207500.0 208000.0 2021.0

3 2025-01-24 09:03:00 207500.0 207500.0 207000.0 207000.0 13711.0

4 2025-01-24 09:04:00 207000.0 207500.0 207000.0 207000.0 1945.0

===== 3분봉 데이터 =====

DateTime Open High Low Close Volume

0 2025-01-24 09:00:00 208000.0 208500.0 207000.0 208000.0 23245.0

1 2025-01-24 09:03:00 207500.0 207500.0 207000.0 207000.0 16434.0

2 2025-01-24 09:06:00 207000.0 207500.0 206000.0 207000.0 30947.0

3 2025-01-24 09:09:00 207000.0 207500.0 206500.0 207000.0 10061.0

4 2025-01-24 09:12:00 207500.0 207500.0 206500.0 207000.0 17399.0

마치며 …

한국투자증권 API를 사용하면 제한된 형태의 1분봉 데이터만 제공받을 수 있어 여러모로 아쉽습니다. 하지만 파이썬 판다스 패키지의 resample() 함수를 적절히 활용하면, 1분봉 데이터로부터 3분봉이나 5분봉, 60분봉 등 원하는 분봉 데이터를 손쉽게 생성할 수 있습니다.

이번 포스팅을 통해 1분봉 데이터 수집 방법과 다양한 분봉 변환 프로세스를 익히셨길 바랍니다. 이러한 아이디어를 확장해 백테스팅, 모델 검증, 실시간 자동매매 등 더 폭넓은 영역에서 활용해 보시기 바랍니다.

함께 참고하면 좋은 글
▶ 한국투자증권 API로 1분봉 OHLCV 데이터 가져오기
▶ 한국투자증권 REST API로 시계열 데이터를 가져오는 파이썬 코드
▶ 한국투자증권 API로 주가를 가져오는 파이썬 프로그래밍
▶ 한국투자증권 API 신청 방법
▶ RESTful API : 웹 애플리케이션 개발자를 위한 기초부터 실전까지
▶ 파이썬 데이터 분석을 위한 API 이해, REST API, WebSocket API

참고자료

한국투자증권 파이썬 API 문서

한국투자증권 API, 3분봉, 5분봉, 10분봉… 원하는 분봉 자유롭게 만드는 방법

한국투자증권 API, 3분봉, 5분봉, 10분봉… 원하는 분봉 자유롭게 만드는 방법

글의 순서

한국투자증권 API로 1분봉 데이터 가져오기

1분봉으로 3분봉, 5분봉, 10분봉… 원하는 분봉 OHLCV 데이터 만들기

resample_ohlcv(df, freq) 함수

파이썬 코드 : 1분봉 OHLCV 데이터로 3분봉, 5분봉, 10분봉… 원하는 분봉 데이터 만들기

마치며 …

참고자료

피보나치 볼린저밴드 파이썬 코딩

거래량 가중 평균값, VWAP 파이썬 코딩

Leave a Comment 응답 취소