시스템 트레이딩 :: (1)키움 API를 활용한 백테스팅(Back Testing)

개발환경 : window10 pycharm python3.8 32bit


1. 키움 API 사용

 https://www1.kiwoom.com/nkw.templateFrameSet.do?m=m1408000000

 1) 키움 API 사용을 위해 키움증권 가입 후 키움Open API신청
 2) Open API+ 모듈 다운로드 및 설치
 3) 필요에 따라 개발가이드, KOA Studio 등 다운로드
 4) 상시 모의투자 신청

 오류해결
     1-1 kiwoom' object has no attribute 'oneventconnect'
            Python 32BIT, 버전 3.6 이하인지 확인
            안된다면 키움API 삭제 -> 설치 -> KOA Studio 재다운로드

2. 가상환경 세팅

 1) anaconda 명령프롬트에서 명령어 입력
        set conda_force_32bit=1 (KIWOOM API는 32BIT에서 동작)
        conda create --name envname python=3.6 (PyQt5는 python 3.6버전에서 사용)

 2) 패키지 설치
    anaconda 명령프롬트에서 입력
        acitvate 가상환경이름
        pip install PyQt5 (키움 API사용시 필요)
        pip install pandas (데이터 관리)
        pip install numpy (데이터 관리)

3. KIWOOM API연동

 키움 API는 OCX(Object Linking and Embedding Custom Control)방식을 사용함.
 OCX 방식은 파이썬에서 사용하기 쉽지 않기때문에 PyQt5 패키지의 QAxContainer
 모듈을  통해 OCX를 사용.

 UI를 따로 구현하지 않으면 아래 코드와 같이 키움 API 연동 가능

class Kiwoom(QAxWidget):

    def __init__(self):
        super().__init__()
        self._create_kiwoom_instance()

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

if __name__ == '__main__':
    # 키움 api 연결
    app = QApplication(sys.argv)
    kiwoom = Kiwoom()

4. 분 단위 데이터 수집

 개발자 가이드 P.50/53
9.3 대량 DATA(차트 조회) 수신
기존 GetCommData()함수로 항목당 한 개의 데이터를 받아오는 처리
에서 배열로 조회된 전 항목의 데이터를 한꺼번에 받을 수 있게 함수입니다.
차트 조회처럼 반복데이터가 많은 데이터를 한꺼번에 받아서 처리하고 싶을 때 사용하시면 됩니다.

개발자 가이드는 C++언어로 설명돼있기 때문에 참고만,

사용법은 KOA Studio가 더 깔끔해서 보기 좋음

[ opt10080 : 주식분봉차트조회요청 ]

 1. Open API 조회 함수 입력값을 설정합니다.
종목코드 = 전문 조회할 종목코드
SetInputValue("종목코드" ,  "입력값 1");

틱범위 = 1:1분, 3:3분, 5:5분, 10:10분, 15:15분, 30:30분, 45:45분, 60:60분
SetInputValue("틱범위" ,  "입력값 2");

수정주가구분 = 0 or 1, 수신데이터 1:유상증자, 2:무상증자, 4:배당락, 8:액면분할, 16:액면병합, 32:기업합병, 64:감자, 256:권리락
SetInputValue("수정주가구분" ,  "입력값 3");


 2. Open API 조회 함수를 호출해서 전문을 서버로 전송합니다.
CommRqData( "RQName" ,  "opt10080" ,  "0" ,  "화면번호");


 데이터를 요청시 OnReceiveTrData 함수가 호출되며 아래 코드와 같이 요청한 데이터를
 받을 수 있음

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

    def receiveTrData(self, screen_no, rqname, trcode, record_name, next, unused1,                                 unused2, unused3, unused4):
        if trcode == 'opt10080':
            self.count = self.dynamicCall("GetRepeatCnt(QString, QString)", trcode, rqname)
            self.opt10080_key = ['현재가', '거래량', '체결시간', '시가', '고가', '저가']
            self.opt10080_data = []
            for i in range(self.count):
                ls = []
                for key in self.opt10080_key:
                    value = abs(int(self.dynamicCall("GetCommData(QString, QString, int,                                          QString)", trcode, rqname, i, key).strip()))
                    ls.append(value)
                self.opt10080_data.append(ls)
            self.tr_event_loop.exit()