[Python/키움API] 주식 틱차트 조회요청 (OPT10079) 및 저장

 

아래는 ETF의 틱데이터를 키움 API로 내려받을 때 사용한 Python 코드를 정리한 것이다. '틱범위'를 변경하면 다른 단위의 틱차트도 조회가 가능하다. (1:1틱, 3:3틱, 5:5틱, 10:10틱, 30:30틱)

 

 

2023.04.24 - [시스템트레이딩] - ETF의 체결오차(슬리피지)는 얼마나 될까?

ETF의 체결오차(슬리피지)는 얼마나 될까?

 

 

 

import sys
from PyQt5.QtWidgets import *
from PyQt5.QAxContainer import *
from PyQt5.QtCore import *
import pandas as pd
from datetime import *
import sqlite3
import time

# 초당 조회 횟수를 회피하기 위한 대기시간 지정
TIME_TERM = 1


class Main(QAxWidget):
    def __init__(self):
        super().__init__()
        self.scrno = '1000'
        self._create_kiwoom_instance()
        self._set_signal_slots()
        self.login_event_loop = None
        self.tr_event_loop = None
        self.sPrevNext = None
        self.end_date = None

    def gen_scrno(self):
        self.scrno = str(int(self.scrno) + 1)
        return self.scrno

    def _create_kiwoom_instance(self):
        self.setControl("KHOPENAPI.KHOpenAPICtrl.1")

    def _set_signal_slots(self):
        self.OnEventConnect.connect(self._event_connect)
        self.OnReceiveTrData.connect(self._receive_tr_data)

    def comm_connect(self):
        self.dynamicCall("CommConnect()")
        self.login_event_loop = QEventLoop()
        self.login_event_loop.exec_()

    def _event_connect(self, nErrCode):
        if nErrCode == 0:
            print('로그인완료')
        self.login_event_loop.exit()

    def set_input_value(self, id, value):
        self.dynamicCall("SetInputValue(QString, QString)", id, value)

    def comm_rq_data(self, rqname, trcode, next, screen_no):
        self.dynamicCall("CommRqData(QString, QString, int, QString", rqname, trcode, next, screen_no)
        self.tr_event_loop = QEventLoop()
        self.tr_event_loop.exec_()

    def _comm_get_data(self, code, real_type, field_name, index, item_name):
        ret = self.dynamicCall("CommGetData(QString, QString, QString, int, QString", code,
                               real_type, field_name, index, item_name)
        return ret.strip()

    def _get_repeat_cnt(self, trcode, rqname):
        ret = self.dynamicCall("GetRepeatCnt(QString, QString)", trcode, rqname)
        return ret

    def _receive_tr_data(self, sScrNo, sRQName, sTrCode, sRecordName, sPrevNext, nDataLength, sErrorCode, sMessage, sSplmMsg):
        print('TR_Message:', sScrNo, sRQName, sTrCode, sRecordName, sPrevNext, nDataLength, sErrorCode, sMessage, sSplmMsg)
        self.sPrevNext = sPrevNext

        if sRQName == "opt10079_req":
            self._opt10080(sRQName, sTrCode)

        try:
            self.tr_event_loop.exit()
        except AttributeError:
            pass

    def req_tick_chart(self, code):
        # 최초조회
        time.sleep(TIME_TERM)
        self.set_input_value("종목코드", code)
        self.set_input_value("틱범위", "1")
        self.set_input_value("수정주가구분", "0")
        self.comm_rq_data(f'opt10079_req', "opt10079", '0', self.gen_scrno())

        # 연속조회 (sPrevNext 변수가 None이 아니라 2일 경우)
        while self.sPrevNext is not None:
            time.sleep(TIME_TERM)
            self.set_input_value("종목코드", code)
            self.comm_rq_data(f'opt10079_req', "opt10079", '2', self.gen_scrno())

    def _opt10080(self, rqname, trcode):
        # 조회된 데이터에서 종목코드를 가져옴 (싱글데이터)
        code = self._comm_get_data(trcode, "", rqname, 0, "종목코드")

        # 전체 데이터 개수 조회
        data_cnt = self._get_repeat_cnt(trcode, rqname)

        # 조회된 데이터 갯수 만큼 반복해서 데이터를 가져온 후 딕셔너리에 저장 (멀티데이터)
        total_ret = {}
        for i in range(data_cnt):
            ret = {key: self._comm_get_data(trcode, "", rqname, i, key) for key in ['현재가', '거래량']}
            index = self._comm_get_data(trcode, "", rqname, i, "체결시간")
            total_ret[index] = ret

        # 딕셔너리를 DataFrame으로 변환 / 컬럼명 변경 / datetime 컬럼 생성
        df = pd.DataFrame.from_dict(total_ret, orient='index')
        df.columns = ['close', 'volume']
        df['datetime'] = pd.to_datetime(df.index, format="%Y%m%d%H%M%S")

        # 최종 조회일자 : 해당 일자이전 기간이 조회데이터에 포함되면 조회를 멈춤
        if df['datetime'].iat[-1] < self.end_date:
            df = df[df['datetime'] >= self.end_date]
            self.sPrevNext = None

        # index (YYYYMMDDHHMMSS 형태로 반환된 인덱스)를 datetime 컬럼지정 / 코드컬럼 생성
        df['datetime'] = df.index
        df['code'] = code
        df = df.reset_index(drop=True)
        print(df)

        # DataFrame을 DB에 저장
        self.append_data_to_db(code, df)

    def append_data_to_db(self, code, df):
        # 지정된 Database의 테이블에 저장하되, 기존 자료가 존재할 경우 해당 테이블 하단에 신규 자료를 추가함
        con = sqlite3.connect('database.db')
        df.to_sql(code, con, if_exists='append', index=False)

        print('DB 저장완료', code, df['datetime'].iat[0], df['datetime'].iat[-1], len(df))


if __name__ == "__main__":
    app = QApplication(sys.argv)

    main = Main()
    main.comm_connect()
    main.end_date = datetime(2023, 4, 20)  # 현재부터 해당 날짜까지 조회

    code_list = ['233740', '069500']  # 조회대상 종목코드 지정
    for code in code_list:
        main.req_tick_chart(code)