[주식] 키움 영웅문 통한 주식 자료다운로드

영웅문 가입은 https://wikidocs.net/2872를 참고하여 진행하였다.

 

목차

1. python 과 키움 API를 활용한 자료 다운로드

2. R을 활용한 자동화

 

아래는 python 을 통해 주식 분봉 자료를 다운로드 받는 코드이다.

 

import sys
from PyQt5.QtWidgets import *
from PyQt5.QAxContainer import *
from PyQt5.QtCore import *
import time
import os

import pandas as pd


class Kiwoom(QAxWidget):
    def __init__(self):
        super().__init__()
        self._create_kiwoom_instance()
        self._set_signal_slots()

    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, err_code):
        if err_code == 0:
            print("connected")
        else:
            print("disconnected")

        self.login_event_loop.exit()

    def get_code_list_by_market(self, market):
        code_list = self.dynamicCall("GetCodeListByMarket(QString)", market)
        code_list = code_list.split(';')
        return code_list[:-1]

    def get_master_code_name(self, code):
        code_name = self.dynamicCall("GetMasterCodeName(QString)", code)
        return code_name

    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, screen_no, rqname, trcode, record_name, next, unused1, unused2, unused3, unused4):
        if next == '2':
            self.remained_data = True
        else:
            self.remained_data = False

        if rqname == "opt10081_req":
            self._opt10081(rqname, trcode)
        elif rqname =='opt10080_req':
            self._opt10080(rqname, trcode)
        try:
            self.tr_event_loop.exit()
        except AttributeError:
            pass

    def _opt10081(self, rqname, trcode):
        data_cnt = self._get_repeat_cnt(trcode, rqname)
        
        for i in range(data_cnt):
            code= self._comm_get_data(trcode, "", rqname, i, "종목코드")
            date = self._comm_get_data(trcode, "", rqname, i, "일자")
            open = self._comm_get_data(trcode, "", rqname, i, "시가")
            high = self._comm_get_data(trcode, "", rqname, i, "고가")
            low = self._comm_get_data(trcode, "", rqname, i, "저가")
            close = self._comm_get_data(trcode, "", rqname, i, "현재가")
            volume = self._comm_get_data(trcode, "", rqname, i, "거래량")
            
            self.ohlcv['code'].append(code)
            self.ohlcv['date'].append(date)
            self.ohlcv['open'].append(abs(int(open)))
            self.ohlcv['high'].append(abs(int(high)))
            self.ohlcv['low'].append(abs(int(low)))
            self.ohlcv['close'].append(abs(int(close)))
            self.ohlcv['volume'].append(abs(int(volume)))

    def _opt10080(self, rqname, trcode):
        data_cnt = self._get_repeat_cnt(trcode, rqname)

        for i in range(data_cnt):
            code= self._comm_get_data(trcode, "", rqname, i, "종목코드")
            date = self._comm_get_data(trcode, "", rqname, i, "체결시간")
            open = self._comm_get_data(trcode, "", rqname, i, "시가")
            high = self._comm_get_data(trcode, "", rqname, i, "고가")
            low = self._comm_get_data(trcode, "", rqname, i, "저가")
            close = self._comm_get_data(trcode, "", rqname, i, "현재가")
            volume = self._comm_get_data(trcode, "", rqname, i, "거래량")
            
            self.ohlcv['code'].append(code)
            self.ohlcv['date'].append(date)
            self.ohlcv['open'].append(abs(int(open)))
            self.ohlcv['high'].append(abs(int(high)))
            self.ohlcv['low'].append(abs(int(low)))
            self.ohlcv['close'].append(abs(int(close)))
            self.ohlcv['volume'].append(abs(int(volume)))

#################

for switch in [0,1]:
    if switch==0:
        app = QApplication(sys.argv)
        kiwoom = Kiwoom()
        kiwoom.comm_connect()
    code_list = kiwoom.get_code_list_by_market('10')
    dic=[]
    for x in code_list:
        dic.append(kiwoom.get_master_code_name(x))
    dic2=dict(zip(code_list,dic))
    dic2
    for x in code_list:
        try:
            if not(os.path.isdir('D:/주식/data/일봉/'+dic2[x])):
                os.makedirs(os.path.join('D:/주식/data/일봉/'+dic2[x]))
        except OSError as e:
            if e.errno != errno.EEXIST:
                print("Failed to create directory!!!!!")
                raise
    for x in code_list:
        try:
            if not(os.path.isdir('D:/주식/data/분봉/'+dic2[x])):
                os.makedirs(os.path.join('D:/주식/data/분봉/'+dic2[x]))
        except OSError as e:
            if e.errno != errno.EEXIST:
                print("Failed to create directory!!!!!")
                raise

###########
    if switch==1:
        kiwoom.ohlcv = {'code':[],'date': [], 'open': [], 'high': [], 'low': [], 'close': [], 'volume': []}
        a=0
        t=0
        aa=0
        TR_REQ_TIME_INTERVAL = 0.35
        for code in code_list:
            if aa<1000:
                t=t+1
                filename=time.strftime('%Y%m%d', time.localtime(time.time()))+'.csv'
                if (filename in os.listdir('D:/주식/data/분봉/'+dic2[code]))==False:
                    aa=aa+1
                    if aa==999:
                        break
                    time.sleep(1.5)
                    kiwoom.ohlcv = {'code':[],'date': [], 'open': [], 'high': [], 'low': [], 'close': [], 'volume': []}
                    kiwoom.set_input_value("종목코드", code)
                    kiwoom.set_input_value("기준일자", time.strftime('%Y%m%d', time.localtime(time.time())))
                    kiwoom.set_input_value("수정주가구분", 1)
                    kiwoom.comm_rq_data("opt10080_req", "opt10080", 0, "0101")
                    sys.stdout.write(str(dic2[code])+str(aa)+'\n')
                    sys.stdout.flush()
    
                    if aa<1000:
                        while kiwoom.remained_data == True:
                            aa=aa+1
                            if aa==999:
                                break
                            sys.stdout.write(str(dic2[code])+str(aa)+'\n')
                            sys.stdout.flush()
                            time.sleep(TR_REQ_TIME_INTERVAL)
                            kiwoom.set_input_value("종목코드", code)
                            kiwoom.set_input_value("기준일자", time.strftime('%Y%m%d', time.localtime(time.time())))
                            kiwoom.set_input_value("수정주가구분", 1)
                            kiwoom.comm_rq_data("opt10080_req", "opt10080", 2, "0101")
                        
                        df = pd.DataFrame(kiwoom.ohlcv, columns=['date','code','open', 'high', 'low',
                        'close', 'volume'])
                        if len(df)!=0:
                            
                            if a%5==0:
                                a=a+1
                                
                                time.sleep(3)
                        df.to_csv('D:/주식/data/분봉/'+dic2[code]+'/'+filename,mode='w',index=False)



위의 파이썬 분봉자료의 경우 횟수 재한이 있어 자동화 하려면 추가 코드가 필요하다.

따라서 R에서 반복문을 활용하여 다운로드 받게 설정하였다.

 

library(rstudioapi) 
myTerm <- rstudioapi::terminalCreate() 
rstudioapi::terminalSend(myTerm, 'conda activate py36_32\n') 
system('conda activate py36_32') 
getwd() 
while(T){ 
  rstudioapi::terminalSend(myTerm, 'python C:/Users/cj/Desktop/code/miniteCode.py\n') 
  rstudioapi::terminalSend(myTerm, 'python C:/Users/cj/Desktop/code/miniteCode1.py\n') 
  rstudioapi::terminalSend(myTerm, 'python C:/Users/cj/Desktop/code/miniteCode_today.py\n') 
}