[Python] 키움API 조건검색 텔레그램 알림 만들기 (1)
- 자아실현/파이썬
- 2020. 11. 2.
조대표님의 위키독스 '파이썬으로 배우는 알고리즘 트레이딩' 프로그램에 조건검색 기능을 추가하여, 조건검색 해당 종목이 뜨면 휴대폰 텔레그램으로 알림을 받는 프로그램을 만들고자 합니다. 증권사는 키움입니다.
따라서 키움API와 PyQt에 대한 기본적인 이해를 필요로 합니다. 필요할 경우 위키독스(wikidocs.net/book/110)에서 기본을 익히시면 됩니다.
전체 코드 링크는 맨 아래에 있습니다.
일부러 실시간으로 종목의 편입/이탈을 반복하는 조건식을 선택하여 테스트했습니다.
1. 개요
키움증권의 조건검색 기능을 이용하여, 해당 조건을 만족하는 종목이 포착되었을 경우 텔레그램 메시지를 발송토록 하고자 합니다.
키움증권 OpenAPI에서 제공하는 조건검색 기능은 영웅문(HTS)에서 작성한 조건식을 불러서 사용하는 방식이며 OpenAPI에서 조건검색 수식작성이나 수식편집은 지원하지 않습니다. 따라서 조건식을 사용하기 위해서는 먼저 HTS 조건식을 작성하여야 합니다. 모바일 영웅문에서도 작성 불가합니다.
아래는 키움 영웅문에서의 조건검색식 화면입니다 (화면번호 0150)
아래는 키움 API에서 콘솔로 저의 조건식 목록을 받고, 그 중에서 '단타2'를 조건 검색한 결과입니다.
위의 작동 영상과 같이 아래의 GUI 프로그램으로 조건 검색식을 실행 및 텔레그램 알람 발송이 목표입니다.
(참고) 조건 검색 관련 함수 목록
SetRealReg/SetRealRemove (실시간 종목 등록)은 여기서 다루지 않겠습니다.
작동 순서
(1) GetConditionLoad 실행 (조건검색 목록 요청)
GetConditionLoad 실행 -> 서버에서 응답이 오면 OnReceiveConditionVer 실행-> OnReceiveConditionVer 안에서 GetConditionNameList 실행되며 등록된 조건검색식 목록 출력
(2) SendCondition 실행 (조건검색 요청 - 위에서 찾은 조건 검색식 번호 및 이름 전달)
SendCondition 실행 -> 서버에서 응답이 오면 OnReceiveTrCondition 실행되며 해당 조건 검색 결과 표시(종목 코드)
-> (SendCondition 실시간 요청시) 해당 조건 검색 결과에 변동이 있을 경우 OnReceiveRealCondition 실행되어 실시간으로 종목 편입/이탈 결과 출력
메소드 설명
(1) GetConditionLoad
: 사용자 조건검색 목록을 서버에 요청합니다. 조건검색 목록을 모두 수신하면 OnReceiveConditionVer()이벤트가 호출됩니다. 조건검색 목록 요청을 성공하면 1, 아니면 0을 리턴합니다.
(2) getConditionNameList
서버에서 수신한 사용자 조건식을 조건명 인덱스와 조건식 이름을 한 쌍으로 하는 문자열들로 전달합니다. 조건식 하나는 조건명 인덱스와 조건식 이름은 '^'로 나뉘어져 있으며 각 조건식은 ';'로 나뉘어져 있습니다. 이 함수는 반드시 OnReceiveConditionVer()이벤트에서 사용해야 합니다.
(3) sendCondition
서버에 조건검색을 요청하는 함수로 맨 마지막 인자값으로 조건검색만 할것인지 실시간 조건검색도 할 것인지를 지정할 수 있습니다. 여기서 조건식 인덱스는 GetConditionNameList()함수가 조건식 이름과 함께 전달한 조건명 인덱스를 그대로 사용해야 합니다. 리턴값 1이면 성공이며, 0이면 실패입니다. 요청한 조건식이 없거나 조건명 인덱스와 조건식이 서로 안맞거나 조회횟수를 초과하는 경우 실패하게 됩니다.
(4) sendConditionStop
조건검색을 중지할 때 사용하는 함수입니다. 조건식 조회할때 얻는 조건식 이름과 조건명 인덱스 쌍을 맞춰서 사용해야 합니다.
아래 세개는 위의 함수 실행 시 서버에서 오는 응답을 받기 위해 자동으로 실행되는 함수입니다.
(5) OnReceiveConditionVer: 사용자 조건식요청에 대한 응답을 서버에서 수신하면 호출되는 이벤트입니다.
(6) OnReceiveTrCondition: 조건검색 요청으로 검색된 종목코드 리스트를 전달하는 이벤트입니다.
(7) OnReceiveRealCondition: 실시간 조건검색 요청으로 신규종목이 편입되거나 기존 종목이 이탈될때 마다 호출됩니다.
2. 조건검색 관련 코드 추가하기
- 편의를 위하여 18장 개발 2일차 코드를 기본으로, Kiwoom.py 파일에 조건검색 관련 코드를 추가합니다.
시작이 되는 코드 : github.com/pystockhub/book/tree/master/ch18/day02
- 각 함수가 실행되는 순서 및 흐름을 파악하기 위해, 함수가 실행되면 맨 처음에 [함수명]이 출력되도록 했습니다.
1) 조건 검색식 조회 관련 함수 추가
getConditionLoad 와 getConditionNameList 함수를 추가합니다.
getConditionLoad 실행 결과를 받을 OnReceiveConditionVer 안에서 getConditionNameList 가 실행되도록 할 예정입니다.
다시 말해서, getConditionLoad() 가 실행되면 OnReceiveConditionVer로 결과를 받고, 그 안에서 getConditionNameList가 실행되며 조건검색식 리스트가 출력됩니다.
위에서 언급하였듯 키움API의 getConditionNameList 원래 리턴 값은 문자열입니다. 함수 내에서 딕셔너리로 변환하여 리턴합니다.
class Kiwoom(QAxWidget):
# 중략
def getConditionLoad(self):
print("[getConditionLoad]")
""" 조건식 목록 요청 메서드 """
isLoad = self.dynamicCall("GetConditionLoad()")
# 요청 실패시
if not isLoad:
print("getConditionLoad(): 조건식 요청 실패")
# receiveConditionVer() 이벤트 메서드에서 루프 종료
self.conditionLoop = QEventLoop()
self.conditionLoop.exec_()
def getConditionNameList(self):
print("[getConditionNameList]")
"""
조건식 획득 메서드
조건식을 딕셔너리 형태로 반환합니다.
이 메서드는 반드시 receiveConditionVer() 이벤트 메서드안에서 사용해야 합니다.
:return: dict - {인덱스:조건명, 인덱스:조건명, ...}
"""
data = self.dynamicCall("GetConditionNameList()")
if data == "":
print("getConditionNameList(): 사용자 조건식이 없습니다.")
conditionList = data.split(';')
del conditionList[-1]
conditionDictionary = {}
for condition in conditionList:
key, value = condition.split('^')
conditionDictionary[int(key)] = value
return conditionDictionary
2) 조건 검색 실행 관련 함수 추가
sendCondition와 sendConditionStop 함수를 추가합니다.
sendCondition 함수가 실행되면 OnReceiveTrCondition 함수를 통해 응답이 옵니다.
sendCondition 함수를 만약 실시간으로 보내면(4번째 인자를 1로 설정), 이후 해당 조건검색에 종목 편입, 이탈이 발생할 경우 OnReceiveRealCondition 를 통해 해당 정보가 들어옵니다.
조건 검색은 최대 10개까지 동시에 등록할 수 있지만, 여기서는 하나만 설정하는 것으로 하겠습니다.
sendCondition 함수의 매개변수
- param screenNo: string
- param conditionName: string - 조건식 이름
- param conditionIndex: int - 조건식 인덱스
- param isRealTime: int - 조건검색 조회구분(0: 1회성 조회, 1: 실시간 조회)
class Kiwoom(QAxWidget):
# 중략
def sendCondition(self, screenNo, conditionName, conditionIndex, isRealTime):
print("[sendCondition]")
"""
종목 조건검색 요청 메서드
이 메서드로 얻고자 하는 것은 해당 조건에 맞는 종목코드이다.
해당 종목에 대한 상세정보는 setRealReg() 메서드로 요청할 수 있다.
요청이 실패하는 경우는, 해당 조건식이 없거나, 조건명과 인덱스가 맞지 않거나, 조회 횟수를 초과하는 경우 발생한다.
조건검색에 대한 결과는
1회성 조회의 경우, receiveTrCondition() 이벤트로 결과값이 전달되며
실시간 조회의 경우, receiveTrCondition()과 receiveRealCondition() 이벤트로 결과값이 전달된다.
:param screenNo: string
:param conditionName: string - 조건식 이름
:param conditionIndex: int - 조건식 인덱스
:param isRealTime: int - 조건검색 조회구분(0: 1회성 조회, 1: 실시간 조회)
"""
isRequest = self.dynamicCall("SendCondition(QString, QString, int, int",
screenNo, conditionName, conditionIndex, isRealTime)
if not isRequest:
print("sendCondition(): 조건검색 요청 실패")
# receiveTrCondition() 이벤트 메서드에서 루프 종료
self.conditionLoop = QEventLoop()
self.conditionLoop.exec_()
def sendConditionStop(self, screenNo, conditionName, conditionIndex):
print("[sendConditionStop]")
""" 종목 조건검색 중지 메서드 """
self.dynamicCall("SendConditionStop(QString, QString, int)", screenNo, conditionName, conditionIndex)
3) 서버로부터 응답을 받을 함수 추가
서버로부터 응답(시그널)이 오면 이를 받아 처리할 함수(슬롯)를 연결하기 위하여 아래와 같이 _set_signal_slots 메소드에 조건검색 관련 함수를 연결합니다.
(1) receiveConditionVer -> 조건 검색식 목록 받을 때
(2) receiveTrCondition -> 처음 조건 검색 할 때 결과
(3) receiveRealCondition -> 조건 검색 실시간 결과
class Kiwoom(QAxWidget):
# 중략
def _set_signal_slots(self):
#중략
## 조건검색식 관련 추가
self.OnReceiveConditionVer.connect(self.receiveConditionVer)
self.OnReceiveTrCondition.connect(self.receiveTrCondition)
self.OnReceiveRealCondition.connect(self.receiveRealCondition)
(1) receiveConditionVer
getConditionLoad를 실행하여 조건식 목록을 조회할 때 받는 OnReceiveConditionVer 시그널을 처리하기 위한 함수입니다.
정상적으로 처리될 경우 getConditionNameList()를 실행하고, 조건식 목록 딕셔너리를 self.condition 변수에 할당하고 보기 쉽게 출력합니다. 이를 위해 클래스 초기화 단계에서 self.condition 를 추가합니다.
class Kiwoom(QAxWidget):
def __init__(self):
# 중략
self.condition = {} #조건검색식 목록 받아올 변수
# 중략
def receiveConditionVer(self, receive, msg):
"""
getConditionLoad() 메서드의 조건식 목록 요청에 대한 응답 이벤트
:param receive: int - 응답결과(1: 성공, 나머지 실패)
:param msg: string - 메세지
"""
print("[receiveConditionVer]")
try:
if not receive:
return
self.condition = self.getConditionNameList()
print("조건식 개수: ", len(self.condition))
for key in self.condition.keys():
print("조건식: ", key, ": ", self.condition[key])
# print("key type: ", type(key))
except Exception as e:
print(e)
finally:
self.conditionLoop.exit()
(2) receiveTrCondition
sendCondition으로 조건 검색을 보내면 받는 OnReceiveTrCondition 시그널을 처리하기 위한 함수입니다.
class Kiwoom(QAxWidget):
# 중략
def receiveTrCondition(self, screenNo, codes, conditionName, conditionIndex, inquiry):
"""
(1회성, 실시간) 종목 조건검색 요청시 발생되는 이벤트
:param screenNo: string
:param codes: string - 종목코드 목록(각 종목은 세미콜론으로 구분됨)
:param conditionName: string - 조건식 이름
:param conditionIndex: int - 조건식 인덱스
:param inquiry: int - 조회구분(0: 남은데이터 없음, 2: 남은데이터 있음)
"""
print("[receiveTrCondition]")
try:
if codes == "":
return
codeList = codes.split(';')
del codeList[-1]
print(codeList)
print("종목개수: ", len(codeList))
finally:
self.conditionLoop.exit()
(3) receiveRealCondition
sendCondition으로 조건검색을 보낼 때 실시간으로 보내면, 해당 조건 검색 결과에 변동이 발생할 경우 받는 OnReceiveRealCondition 시그널을 처리하기 위한 함수입니다.
class Kiwoom(QAxWidget):
# 중략
def receiveRealCondition(self, code, event, conditionName, conditionIndex):
print("[receiveRealCondition]")
"""
실시간 종목 조건검색 요청시 발생되는 이벤트
:param code: string - 종목코드
:param event: string - 이벤트종류("I": 종목편입, "D": 종목이탈)
:param conditionName: string - 조건식 이름
:param conditionIndex: string - 조건식 인덱스(여기서만 인덱스가 string 타입으로 전달됨)
"""
print("종목코드: {}, 종목명: {}".format(code, self.get_master_code_name(code)))
print("이벤트: ", "종목편입" if event == "I" else "종목이탈")
4) 한번 실행해보기
Kiwoom.py
if __name__ == "__main__":
app = QApplication(sys.argv)
kiwoom = Kiwoom()
kiwoom.comm_connect()
kiwoom.getConditionLoad()Kiwoom.py 파일을 실행하면, 아래와 같이 등록된 조건검색식 목록이 뜹니다.
조건식 이름 앞의 숫자가 index이며, sendCondition 할 때 해당 번호가 필요합니다.
index : 2, 이름 : "단타2"인 두번 째 조건검색식을 '실시간 조회 적용하여' 실행해보겠습니다.
-> kiwoom.sendCondition("0","단타2",2,1)
Kiwoom.py
if __name__ == "__main__":
app = QApplication(sys.argv)
kiwoom = Kiwoom()
kiwoom.comm_connect()
kiwoom.getConditionLoad()
"""
:param screenNo: string
:param conditionName: string - 조건식 이름
:param conditionIndex: int - 조건식 인덱스
:param isRealTime: int - 조건검색 조회구분(0: 1회성 조회, 1: 실시간 조회)
"""
kiwoom.sendCondition("0","단타2",2,1)
이 글을 작성중인 지금이 주말이라 장 마감으로 실시간으로 편입되는 종목은 확인할 수 없지만, 위와 같은 실행 결과를 볼 수 있습니다.
3. UI 구성하기
기존 2일차 강의 UI에서 아래와 같이 조건검색용 UI를 추가하였습니다.
(로그창은 오브젝트 네임이 textEdit_cond 입니다. i는 오타)
(1) 조건 검색식 목록 (comboBox)
(2) 조건 검색 적용/해제 버튼 (pushButton)
(3) 텔레그램 알림 체크박스(checkBox)
(4) 로그 화면(textEdit)
pytrader.py를 실행하면 아래와 같은 화면입니다.
조건 검색 관련 버튼은 아직 동작하지 않습니다.
다음 게시물에서 pytrader.py 파일을 통하여 UI와 조건 검색 기능들을 연결시키도록 하겠습니다.
이 게시물의 전체 소스코드는 이곳에 있습니다.
이 게시물 코드 : github.com/kminito/pytrader-condition/tree/main/day1
참고한 Github Repository : github.com/kdseo/PyTrader