키움API로 분봉 데이터 가져오기 (주식 데이터 수집 및 분석 , python 소스 코드 포함

안녕하세요. 주식 투자 관련 프로그램을 공유하는 인천고래입니다.

금일은 일봉 차트가 아닌 분봉 차트 데이터를 키움 API를 이용해서 가져오는 소스 코드를 공유하고자 합니다.

 

퀀트 투자를 하기 위해서는 우선 데이터를 취합을 하는게 우선입니다.

일봉 데이터는 쉽게 구할 수 있는 반면 분봉 데이터는 분 단위로 데이터가 발생되기 때문에 취합하는데 많은 시간을 할애를 해야 합니다. 일봉 하루치가 1분 봉 하나와 동일한 데이터이니 하루(6시간 30분간)의 1분 봉 데이터를 계산하면... 일봉 1년 치보다 많은 것 같네요. 

 

그리고 퀀트 투자를 하고는 싶어도 데이터가 저장이 되는 데이터베이스를 잘 모르는 사람들이 대부분인데 이를 쉽게 활용할 수 있도록 많은 사람들이 사용하는 엑셀 파일로 저장하여 언제든지 쉽게 접근가능하도록 소스 코드를 구성해 보았습니다.

종목별 분봉 데이터

 

이번 글에서는 Python을 이용해 주식 데이터를 수집하고, 이동평균선, 엔벨로프, ATR 등의 기술적 지표를 계산하는 코드가 존재하니. 이 소스를 통해 초보자도 쉽게 따라할 수 있도록 전체 코드를 넣어두었습니다.

 

키움 분봉 데이터.py

0.01MB

 

 

키움 API로 분봉 데이터 가져오기 목차

• 주식 데이터 수집: get_API_minute_data()
  키움 API를 통해 주식의 분봉 데이터를 수집합니다.
  
  
• 이동평균선 계산: calculate_moving_averages(df, periods)
  다양한 기간의 이동평균선을 계산합니다.
• 엔벨로프 계산: calculate_envelopes(df, period=20, deviation=0.02)
  특정 기간과 이격 비율을 기반으로 엔벨로프를 계산합니다.
• ATR 계산: calculate_atr(df, period=14)
  True Range와 ATR 백분율을 계산합니다.
• 데이터 저장: 수집한 데이터를 엑셀 파일로 저장합니다.

※ get_API_minute_data()함수 및 분봉 데이터를 요청하는 block_request() 메서드를 설명하는 글은 아래에 링크 글에 있습니다.

키움 API를 통해 분봉 데이터 가져오기 - kiwoom.block_request

 

 

그리고 위의 목차에 나와있지만 주식 데이터를 수집한 뒤 이동평균선, 엔벨로프, ATR 등을 어떻게 추가하는지 알려드리기 위해 넣어두었고요. 필요 없으시면 삭제하셔도 됩니다.

 

데이터에는 주가 데이터만 존재하고 불러서 사용할 때 필요한 보조지표를 계산해서 사용하는 것이 더 좋은 것 같더라고요.

 

아래는 전체 코드입니다.

사용하시다가 궁금한 점 있으시면 댓글 달아주세요. 성심성의껏 설명 드리겠습니다.

import time
from pykiwoom.kiwoom import Kiwoom
import pandas as pd
import numpy as np

def calculate_moving_averages(df, periods):
    """
    여러 기간에 대한 이동평균선을 계산하는 함수
    :param df: 입력 데이터프레임 (종가 데이터 포함)
    :param periods: 이동평균선을 계산할 기간 리스트 (예: [5, 10, 20, 60, 120])
    :return: 여러 기간의 이동평균선이 포함된 데이터프레임
    """
    for period in periods:
        df[f'MA_{period}'] = df['Close'].rolling(window=period).mean()
    return df

def calculate_envelopes(df, period=20, deviation=0.02):
    """
    엔벨로프를 계산하는 함수
    :param df: 입력 데이터프레임 (종가 데이터 포함)
    :param period: 이동평균선 기간 (기본값: 20일)
    :param deviation: 이격 비율 (기본값: 2%)
    :return: 엔벨로프가 포함된 데이터프레임
    """
    df['MA'] = df['Close'].rolling(window=period).mean()
    df['Upper Envelope'] = df['MA'] * (1 + deviation)
    df['Lower Envelope'] = df['MA'] * (1 - deviation)
    return df

def calculate_atr(df, period=14):
    """
    ATR(Average True Range)과 True Range 백분율을 계산하는 함수
    :param df: 입력 데이터프레임 (고가, 저가, 종가 데이터 포함)
    :param period: ATR을 계산할 기간 (기본값: 14일)
    :return: ATR과 True Range 백분율이 포함된 데이터프레임
    """
    df['High-Low'] = df['High'] - df['Low']
    df['High-PrevClose'] = np.abs(df['High'] - df['Close'].shift(1))
    df['Low-PrevClose'] = np.abs(df['Low'] - df['Close'].shift(1))
    df['True Range'] = df[['High-Low', 'High-PrevClose', 'Low-PrevClose']].max(axis=1)

    # 현재가 대비 True Range 백분율 계산
    df['True Range %'] = (df['True Range'] / df['Close'].shift(1)) * 100

    # ATR 계산
    df['ATR %'] = df['True Range %'].rolling(window=period).mean()

    return df

def get_API_minute_data(stock_code, start_date, end_date, tick_range=1):
    kiwoom = Kiwoom()
    kiwoom.CommConnect(block=True)

    tr = "opt10080"
    code = stock_code
    df_list = []

    # First block request
    df_firstblock = kiwoom.block_request(tr,
                                         종목코드=code,
                                         수정주가구분=1,
                                         틱범위=tick_range,
                                         output="주식분봉차트조회",
                                         next=0)
    df_list.append(df_firstblock)

    # Continue requesting while there is more data
    while kiwoom.tr_remained:
        df_remainblock = kiwoom.block_request(tr,
                                              종목코드=code,
                                              수정주가구분=1,
                                              틱범위=tick_range,
                                              output="주식분봉차트조회",
                                              next=2)

        df_list.append(df_remainblock)
        oldest_date_in_block = pd.to_datetime(df_remainblock['체결시간'].min(), format='%Y%m%d%H%M%S')
        if oldest_date_in_block < pd.to_datetime(start_date):
            break
        time.sleep(1)  # To avoid request limit error

    # Concatenate all data
    df = pd.concat(df_list)
    df.reset_index(drop=True, inplace=True)

    # 한글 컬럼 이름을 영어로 변경
    df.rename(columns={
        '현재가': 'Close',
        '거래량': 'Volume',
        '시가': 'Open',
        '고가': 'High',
        '저가': 'Low',
        '체결시간': 'DateTime'
    }, inplace=True)

    # Clean and preprocess the data
    # 방법 1: str.replace 사용, str.replace 변환 시간: 0.024933 초
    # 방법 2: apply + lambda 사용, apply + lambda 변환 시간: 0.012962 초
    # df['Close_replace'] = df['Close'].str.replace('-', '').astype('float')

    df['Close'] = df['Close'].apply(lambda x: x.lstrip('-')).astype('float')
    df['Open'] = df['Open'].apply(lambda x: x.lstrip('-')).astype('float')
    df['High'] = df['High'].apply(lambda x: x.lstrip('-')).astype('float')
    df['Low'] = df['Low'].apply(lambda x: x.lstrip('-')).astype('float')
    df['DateTime'] = pd.to_datetime(df['DateTime'], format='%Y%m%d%H%M%S', errors='raise')

    df_all = df.drop(['수정주가구분', '수정비율', '대업종구분', '소업종구분', '종목정보', '수정주가이벤트', '전일종가'], axis=1)


    # Filter by date range
    df_all = df_all[(df_all['DateTime'] >= pd.to_datetime(start_date + ' 09:00:00')) & (
            df_all['DateTime'] <= pd.to_datetime(end_date + ' 15:30:00'))]
    df_all = df_all.sort_values(by='DateTime')

    return df_all


# Example execution
if __name__ == "__main__":
    start_time = time.time()

    stock_code = "051380"  # Example stock code
    start_date = "20240526"
    end_date = "20240626"
    periods = [5, 10, 20, 60, 120]

    min_df = get_API_minute_data(stock_code, start_date, end_date, tick_range=1)

    atr_df = calculate_atr(min_df)

    env_df = calculate_envelopes(atr_df, 60, 7)

    env_df = calculate_moving_averages(atr_df, periods)

    min_df.to_excel(f'min_data/{stock_code}_분봉데이터.xlsx')

    end_time = time.time()
    elapsed_time = end_time - start_time
    msg = f"데이터 수집 시간(분): {elapsed_time / 60:.1f}\n------------------------------"

    print(msg)
    print(min_df.info())
    print(min_df.head())
    print(min_df.tail())