주식 종목 추천 봇 만들기 [4] 텔레그램 봇으로 키움 조건검색식 API 결과 전달

이전 포스팅 읽기 👇

주식 종목 추천 봇 만들기 [3] 키움 증권 API Key 발급 받기

---

이제 본격적으로 키움에 API로 질의한 내용을 텔레그램 봇한테 전달하여 내가 확인할 수 있도록 하면 된다.

우선 내가 첫 번째로 알고 싶은 내용은 키움의 조건검색식 조회 결과이다.

(물론 조건검색식 결과는 MTS로 실시간 모니터링이 가능하지만,
1. 최종적으로는 MTS에 들어가지 않고 bot만으로 종목을 확인할 수 있어야 함 (이후 자동매매까지 고려)
2. MTS에 실시간 조회 화면에서 새로운 종목이 나타나더라도 회사에서 내가 바로 확인하기 어려움
위의 이유로 bot에 구현하였다.)

조건검색식 결과를 bot으로 확인하는 주요 흐름은 다음과 같다.

1. 키움으로 API를 통해 조건검색식 내용 질의
2. telegram bot에 데이터 전달
3. bot으로부터 주기적으로 메시지 수신 (또는 내가 질문하여 대답 확인)

우선, 앞선 포스팅에서 pykiwoom 모듈을 사용 예시 코드에선 Kiwoom 클래스를 사용했지만, 텔레그램 bot이 돌고 있는 thread와 kiwoom으로 질의하는 thread를 분리하고, 독립적으로 수행되도록 하기 위해 KiwoomManager 클래스를 사용하는 것으로 변경했다.

KiwoomManager를 사용하면 독립된 별도의 프로세스로 kiwoom 질의 작업을 처리할 수 있다.
이렇게 독립된 별도의 프로세스로 처리하지 않으면 telegram bot에서 처리하는 async 비동기 함수의 await 처리와 kiwoom api에서 처리하는 동기 작업이 맞물려 lock이 걸려 버린다. (<< 이 문제를 해결하느라 꽤 애를 먹었다)

아무튼 이를 염두에 두고, 조검검색식 값을 전달하는 과정은 다음과 같다.

1. 키움 조건검색식 내용 질의

1. 우선 GetConditionNameList 함수를 사용하여 나의 모든 조건검색식 리스트를 조회한다.
2. 조회 결과를 통해 내가 원하는 조건검색식의 index 번호를 확인하고,
3. getCondition 함수를 통해 해당 조건검색식의 결과를 확인한다.
4. 검색식의 결과를 예쁘게 정제하여 return한다.

내가 확인하고 싶은 조건검색식 이름은 "break_through" 라는 이름의 검색식이고, 해당 검색식의 검색 결과를 확인하는 코드는 다음과 같다.

class kiri:
    def __init__(self):
        self.kiwoom = KiwoomManager()
        self.last_req_time = datetime.datetime.now() - datetime.timedelta(minutes=5)

    def getStock(self):
        # 요청 수 제한 대응
        diff = datetime.datetime.now() - self.last_req_time
        if diff.seconds <= 60 :
            return "Please wait " + str(60 - diff.seconds) +"s..."

        # 나의 조건검색식 리스트 조회
        kiwoom = self.kiwoom
        cmd = {'func_name': 'GetConditionNameList'}
        kiwoom.put_cond(cmd)
        conditions = kiwoom.get_cond(method=True)
        self.last_req_time = datetime.datetime.now()

        result_msg = ""
        for condition in conditions:
            # 내가 원하는 조건검색식 결과 조회
            if condition[1] == 'break_through':
                condition_index = condition[0]
                condition_name = condition[1]
                result_msg += '\n*고점 돌파 종목*\n'
                # time.sleep(30)
                target_tickers = self.getCondition(condition_name, condition_index)
                if len(target_tickers) == 0:
                    result_msg += 'No Result\n'
                else:
                    result_msg = self.makeResultMsg(result_msg, target_tickers,0) # 결과 데이터를 예쁘게 정제
        return result_msg

 

2. Telegram Bot으로 조건검색식 결과 전달

텔레그램 봇으로부터 결과를 받기 위해 우선 내가 텔레그램 봇한테 말을 걸어 요청하는 방식으로 먼저 개발하였다.

1. CommandHandler를 통해 조건 검색식 결과를 요청할 command를 정의하고, 해당 command에 대해 수행할 함수를 정의한다.
2. 정의한 함수에서 조건 검색식 결과를 받아온다.
3. 받아온 결과를 sendMessage 함수를 통해 전송한다.

조건검색식 요청 command는 todayBreakStock이라고 정의하였고, 수행할 함수는 getTodayStock() 으로 정의하였다.
코드는 다음과 같다.

async def getTodayStock(update: Update, context: ContextTypes.DEFAULT_TYPE):
    result = stock.getStock() # 조건검색식 결과 조회 함수
    await context.bot.send_message(chat_id=update.effective_chat.id, text=result)

if __name__ == '__main__':
    stock = Stock.Stock()
    application = ApplicationBuilder().token(token).build()

    getTodayStock_handler = CommandHandler('todayBreakStock', getTodayStock)
    application.add_handler(getTodayStock_handler)

    application.run_polling()

그러면 내가 /getTodayStock 이라고 명령을 bot에게 전달하면, kiwoom 으로 내가 원하는 조건검색식 결과를 질의하여 그 결과를 대답한다.

 

3. bot으로부터 주기적으로 메시지 수신

내가 회사에서 하나하나 물어보고 있을 수 없으니 bot이 알아서 때되면 나에게 메시지를 보내줘야 한다.
주기적으로 나에게 조건검색식 결과를 보내주도록 python의 schedule 모듈을 사용하였다.
main 함수에 아래와 같이 9:00 ~ 9:30 까지 15분 간격으로 메시지를 보내도록 코드를 추가하였다.

news_callback 함수는 async 함수를 동기식으로 처리하기 위해 만든 중간 처리 함수이다.
텔레그램에서 메시지를 전송하는 함수는 전부 async로 구현이 되는데, 정상적으로 schedule 처리가 되게 하려면 sync로 처리해야 해서 asyncio 모듈을 사용하여 event loop를 하나 더 만들고, 텔레그램으로 메시지를 전송하는 함수가 완료처리 될 때 까지 기다리도록 구현하였다. (run_until_complete)

if __name__ == '__main__':
    stock = Stock.Stock()
    application = ApplicationBuilder().token(token).build()

    getTodayStock_handler = CommandHandler('todayBreakStock', getTodayStock)
    application.add_handler(getTodayStock_handler)

    schedule.every().day.at("09:00:00").do(news_callback)
    schedule.every().day.at("09:15:00").do(news_callback)
    schedule.every().day.at("09:30:00").do(news_callback)
    Thread(target=schedule_checker).start() 

    application.run_polling()

이렇게 하여 성공적으로 매일 9시, 9시 15분, 9시 30분 마다 조건 검색식 결과 종목들을 메시지로 받을 수 있었다.

다음은 내가 정말 구현하고 싶었던 내용인 테마주 알림 받기 를 구현한 내용을 포스팅 하겠다.

---

다음 포스팅 읽기 👇

주식 종목 추천 봇 만들기 [5] 오늘의 테마주 알림 받기