업비트 공지사항 크롤링하여 텔레그램으로 메세지 알림받기

예전에는 업비트에서 프로젝트 공시를 공지사항에 올리면 해당 종목이 급등하는 일명 공시펌핑이 있던 시절이 있었습니다. 하지만 여러가지 사유로 시장을 흐린다는 판단하에 현재는 업비트에서는 프로젝트 공시를 하고 있지는 않습니다.

 

하지만 일반 공지사항 중에서도 입출금 일시 중지와 같은 펌핑의 재료로 쓰일 수 있는 공지들이 있는데요. 이러한 공지들은 빠르게 확인하고 대처하는 것이 가장 중요합니다.

 

이번 시간에는 업비트 공지사항의 글을 5초 간격으로 크롤링하여 새로운 글이 올라오면 텔레그램 메시지로 알림을 받는 프로그램을 만들어 보겠습니다.

 

 

목차 - 클릭하면 이동합니다.

 

업비트 공지사항 크롤링

업비트 공지사항 크롤링 주소

업비트 웹사이트를 보면 고객센터 아래에 위와 같은 공지사항 메뉴가 있습니다. 여러가지 게시판이 있지만 현재 프로젝트 공시 메뉴는 더 이상 공시되고 있지 않아 그 중에서 첫 번째인 공지사항 부분을 크롤링하도록 하겠습니다. 

 

현재 페이지를 크롤링하면 안되고 오른쪽의 공지사항 데이터를 크롤링 해야 하는데 해당 URL은 아래와 같습니다.

 

https://api-manager.upbit.com/api/v1/notices?page=1&per_page=20&thread_name=general 

 

인터넷 브라우저에 주소를 쳐 보면 html 구조가 아닌 json 구조로 되어 있어 사용하기에도 아주 편리할 것 같습니다.

 

공통 모듈 - upbit.py

공통 모듈에 일부 수정이 있었으며 최신 파일은 아래 파일을 참고 하시면 됩니다. 기존에 포스팅한 로직이 작동하지 않을 수 있으므로 기존 파일을 백업 후 사용하시면 추후 오류를 찾는데 도움이 될 것 같습니다.

 

upbit.py

0.07MB

 

공지사항 크롤링 프로그램 - mon_notice.py

import os
import sys
import time
import logging
import traceback
import requests

# 공통 모듈 Import
sys.path.append(os.path.dirname(os.path.dirname(__file__)))
from module import upbit

# Program Name
pgm_name = 'mon_notice'
pgm_name_kr = '업비트 공지사항 크롤링'

# Headers & Page URLs
headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.110 Safari/537.36'}
page_url = 'https://api-manager.upbit.com/api/v1/notices?page=1&per_page=20&thread_name=general'
query_url = 'https://upbit.com/service_center/notice?id='

# -----------------------------------------------------------------------------
# - Name : mon_notice
# - Desc : 업비트 공지사항 크롤링
# -----------------------------------------------------------------------------
def mon_notice():
    try:

        while True:
            logging.info('업비트 공지사항 크롤링 중...')

            response = requests.get(page_url, headers=headers)
            upbit_notice = response.json()

            for notice in upbit_notice['data']['list']:

                # 제목 & 입력일시 & 글 번호 추출
                title = notice['title']
                dt = notice['created_at']
                article_no = notice['id']

                tmp_notice = {'title': title, 'dt': dt, 'article_no': article_no}

                # 기존 공지사항 목록 조회
                prev_notice_list = upbit.read_file(pgm_name)

                # 기존 공지내역이 있으면 비교하기
                if len(prev_notice_list) > 0:
                    # 기존 공지내역이 있으면 PASS
                    # and len(list(filter(lambda x: x['dt'] == dt, prev_notice_list))) > 0 \
                    if len(list(filter(lambda x: x['title'] == title, prev_notice_list))) > 0 \
                            and len(list(filter(lambda x: x['article_no'] == article_no, prev_notice_list))) > 0:
                        logging.info('기존 공지내역이 있음!')
                        logging.info(tmp_notice)

                    else:

                        logging.info('기존 공지내역이 없음! 메세지 발송!')

                        # 메세지 발송
                        msg_contents = '[' + pgm_name_kr + '] 신규 공지 등록'
                        msg_contents = msg_contents + '\n\n- 제목: ' + str(title)
                        msg_contents = msg_contents + '\n- 작성시간: ' + str(dt)
                        msg_contents = msg_contents + '\n\n- 글 확인하기: ' + str(query_url) + str(article_no)

                        # 메세지 발송
                        upbit.send_telegram_message(msg_contents)

                        # 공지내역에 추가
                        logging.info('신규 공지내역 추가!')
                        logging.info(tmp_notice)

                        # 신규 공지내역 추가
                        upbit.write_config_append(pgm_name, '\n' + str(tmp_notice))

                        # 공지 작성 후 1초 대기
                        time.sleep(1)

                # 기존 공지내역이 없으면 파일에 쓰기
                else:

                    logging.info('공지내역 최초 추가!')
                    logging.info(tmp_notice)

                    # 신규 공지내역 추가
                    upbit.write_config_append(pgm_name, tmp_notice)

                    # 공지 작성 후 1초 대기
                    time.sleep(1)

                # 5초 간격으로 조회
            time.sleep(5)

    # ---------------------------------------
    # 모든 함수의 공통 부분(Exception 처리)
    # ----------------------------------------
    except Exception:
        raise


# -----------------------------------------------------------------------------
# - Name : main
# - Desc : 메인
# -----------------------------------------------------------------------------
if __name__ == '__main__':

    # noinspection PyBroadException
    try:
        print("***** USAGE ******")
        print("[1] 로그레벨(D:DEBUG, E:ERROR, 그외:INFO)")

        if sys.platform.startswith('win32'):
            # 로그레벨(D:DEBUG, E:ERROR, 그외:INFO)
            log_level = 'I'
            upbit.set_loglevel(log_level)
        else:
            # 로그레벨(D:DEBUG, E:ERROR, 그외:INFO)
            log_level = sys.argv[1].upper()
            upbit.set_loglevel(log_level)

        if log_level == '':
            logging.error("입력값 오류!")
            sys.exit(-1)

        logging.info("***** INPUT ******")
        logging.info("[1] 로그레벨(D:DEBUG, E:ERROR, 그외:INFO):" + str(log_level))

        # ---------------------------------------------------------------------
        # Logic Start!
        # ---------------------------------------------------------------------
        mon_notice()

    except KeyboardInterrupt:
        logging.error("KeyboardInterrupt Exception 발생!")
        logging.error(traceback.format_exc())
        sys.exit(-100)

    except Exception:
        logging.error("Exception 발생!")
        logging.error(traceback.format_exc())
        sys.exit(-200)

[Line 17] headers : 웹사이트를 호출할 때 간혹 사용자가 아닌 봇이 접근하는 것을 차단하는 사이트들이 있어서 예방 차원에서 넣어둔 부분입니다. 위의 내용은 Chrome 브라우저에서 접근한 듯한 User-Agent를 가지고 웹사이트에 접속하도록 합니다.

 

[Line 18] page_url : 업비트 공지사항 리스트를 받아오는 URL입니다.

 

[Line 19] query_url : 텔레그램 메세지로 받은 URL을 클릭하여 개별 공지를 쉽게 확인할 수 있도록 개별 공지사항이 열리는 URL입니다. 뒤에 게시물 ID를 붙여주면 공지를 바로 확인할 수 있습니다.

 

같은 공지를 중복하여 메세지를 보내지 않도록 하기 위해 한번 알림을 보낸 공지는 파일로 관리하여 메세지 재발송을 방지합니다. 먼저 /home/python/trade_bot 폴더 아래에 conf 폴더를 생성해 줍니다.

 

mon_notice.txt 빈 파일을 먼저 하나 생성해 주어도 되고 생성하지 않고 프로그램을 수행하면 파일을 자동으로 생성합니다. 오류가 발생하지만 파일이 생성된 후 다시 수행하면 그 이후부터는 에러가 발생하지 않습니다.

 

반응형

cd /home/python/trade_bot
python mon_notice.py I

./env/env.txt 환경 파일을 읽어오기 위해 파이썬을 수행하는 폴더를 기존의 /home/python 이 아닌 /home/python/trade_bot에서 수행 합니다.

 

위와 같이 수행하면 기존 발송되지 않은 공지내역은 텔레그램으로 메세지를 발송합니다.

 

[Line 46~51] 한번 발송된 후 파일에 기록된 공지사항은 제목과 ID를 기준으로 기 발송 내역으로 판단하여 다시 메세지를 보내지 않습니다.

 

[Line 91] 너무 자주 크롤링을 하면 거부를 당할 수 있어 5초 간격으로 조회하도록 합니다.

 

마치며

오늘 살펴본 내용은 단순히 공지사항을 크롤링하여 텔레그램으로 알림을 받는 것이지만 알림을 받는 로직에 매수 로직을 추가할 수도 있습니다.

 

예를 들면 입출금 중지 공지가 새로 생성되면 1분간 해당 종목을 모니터링 하면서 10%이상 상승하는 경우 매수하는 로직으로 활용할 수도 있습니다.

 

하지만 모든 공지가 펌핑으로 이어지지는 않으며 이어지더라도 순간의 펌핑 후 다시 제자리로 돌아오는 경우가 많기 때문에 이런 부분을 주의할 필요가 있습니다.

 

항상 소액으로 프로그램을 검증 후 사용하시기를 부탁 드리며 여러분들의 성공적인 투자에 조금이나마 도움이 되기를 바랍니다. 블로그를 구독하면 소식을 조금 더 빨리 받아 보실 수 있습니다.

 

감사합니다.