[Python] 미국증시 시황 알림 봇 만들기

미국증시 시황 알림 봇 만들기

 

안녕하세요.

오늘은 지난 포스팅에서 작성했던 코드들을 활용하여 미국증시 시황 알림 봇을 만들어보려고 합니다.

 

먼저, 미국증시 시황 알림 봇에 대해 간략히 말씀드리면, 매일 아침 8시에 미국주식 3대 지수, FRED API를 통한 장단기 금리차 데이터와 네이버 뉴스를 전송해 주는 봇이며, 아래에 제가 작성한 파이썬 코드를 같이 보면서 설명드리도록 하겠습니다.

 

▼ 지난 포스팅 확인은 아래 링크를 이용하여 확인해 보시면 되겠습니다.

[Python] 네이버 뉴스기사를 텔레그램 채팅방에 전송하기

[Python] 네이버 뉴스기사를 텔레그램 채팅방에 전송하기

 

[Python] 텔레그램 채팅방 이미지 전송

[Python] 텔레그램 채팅방 이미지 전송

 

 

 

필요 라이브러리

 

먼저, 3대 지수 정보를 가져오는데 야후 파이낸스의 정보를 가져올 수 있도록 해주는 yahoo_fin 라이브러리를 이용하였습니다.

 

yahoo_fin 라이브러리의 사용방법은 아래 링크의 yahoo_fin 사용 방법이 적힌 문서 페이지에서 사용방법을 확인해 보시면 되겠습니다.

 

yahoo_fin 문서 페이지로 이동

Yahoo_fin Documentation - Open Source Automation

 

 

그리고 텔레그램 봇 라이브러리는 최근에 업데이트된 v20.0 버전 기준으로 코드를 작성하였습니다.

 

아래 pip 명령어를 이용하여 필요한 라이브러리를 다운로드하실 수 있습니다.

 

pip install requests
pip install beautifulsoup4
pip install python-telegram-bot
pip install fredapi
pip install pandas
pip install matplotlib
pip install datetime
pip install yahoo_fin

 

 

 

전체코드

 

아래 코드는 이번에 만들어본 미국증시 시황 알림 봇의 전체 코드입니다. 

 

from datetime import datetime, timedelta
import time
import asyncio
import pandas as pd
import matplotlib.pyplot as plt
import fredapi
import requests

from  bs4  import  BeautifulSoup
import telegram
import yahoo_fin.stock_info as si

# FRED
key = 'FRED API Key 입력'
fred = fredapi.Fred(api_key=key)

# 텔레그램 토큰 및 chat_id
s_token = '텔레그램 토큰 입력'
l_chat_id = 아이디값 입력

# 텔레그램 메세지 전송 함수
async def send_msg(p_text):
    bot = telegram.Bot(s_token)
    async with bot:
        await bot.send_message(text=p_text, chat_id=l_chat_id)

# 텔레그램 메세지 전송 함수
async def send_img(p_path):
    bot = telegram.Bot(s_token)
    async with bot:
        await bot.send_photo(chat_id=l_chat_id, photo=open(p_path, 'rb'))

def send_idx_info():
    """미국주식 3대 지수 전송"""
    try:
        # 1. 다우지수 구하기
        dow_data = si.get_quote_data('^DJI')
        dow_prev_price = round(dow_data['regularMarketPreviousClose'], 2)
        dow_current_price = round(dow_data['regularMarketPrice'], 2)
        dow_gap_price = round(dow_current_price - dow_prev_price, 2)
        dow_total_percent = round((dow_current_price - dow_prev_price) / dow_prev_price * 100, 2)
        
        total_msg = "[Dow Jones]\n"
        total_msg = total_msg + " 전일 종가 : " + str(dow_prev_price) + "\n"
        total_msg = total_msg + " 금일 종가 : " + str(dow_current_price) + " / " + str(dow_gap_price) + " (" + str(dow_total_percent) + "%)"

        # 2. S&P500 지수 구하기
        snp_data = si.get_quote_data('^GSPC')
        snp_prev_price = round(snp_data['regularMarketPreviousClose'], 2)
        snp_current_price = round(snp_data['regularMarketPrice'], 2)
        snp_gap_price = round(snp_current_price - snp_prev_price, 2)
        snp_total_percent = round((snp_current_price - snp_prev_price) / snp_prev_price * 100, 2)
        
        total_msg = total_msg + "\n\n[S&P 500]\n"
        total_msg = total_msg + " 전일 종가 : " + str(snp_prev_price) + "\n"
        total_msg = total_msg + " 금일 종가 : " + str(snp_current_price) + " / " + str(snp_gap_price) + " (" + str(snp_total_percent) + "%)"

        # 3. 나스닥 지수 구하기
        nas_data = si.get_quote_data('^IXIC')
        nas_prev_price = round(nas_data['regularMarketPreviousClose'], 2)
        nas_current_price = round(nas_data['regularMarketPrice'], 2)
        nas_gap_price = round(nas_current_price - nas_prev_price, 2)
        nas_total_percent = round((nas_current_price - nas_prev_price) / nas_prev_price * 100, 2)
        
        total_msg = total_msg + "\n\n[NASDAQ]\n"
        total_msg = total_msg + " 전일 종가 : " + str(nas_prev_price) + "\n"
        total_msg = total_msg + " 금일 종가 : " + str(nas_current_price) + " / " + str(nas_gap_price) + " (" + str(nas_total_percent) + "%)"

        # 텔레그램 채팅방으로 메세지 전송
        asyncio.run(send_msg(total_msg))

    except Exception as ex:
        print("send_idx_info() -> 함수 예외 발생! [내역 : " + str(ex) + "]")

def send_fred_data(p_type):
    """장단기 금리차 전송"""
    try:
        # 데이터 시작일 지정
        start_date = datetime.today() - timedelta(days=365*5)

        #  FRED 데이터 불러오기
        fred_data = pd.DataFrame(fred.get_series(p_type, observation_start=start_date), columns=[p_type])

        # 불필요 데이터 삭제
        fred_data = fred_data.dropna()

        # 데이터 출력
        print(fred_data)

        # 그래프 사이즈
        plt.figure(figsize=(10, 5))

        # 그래프 그리기
        plt.plot(fred_data)

        # 그래프 제목 작성
        plt.title(p_type + ' Yield Spread : {:.2f}'.format(fred_data.iloc[-1][0]))

        # 0일때 라인 그리기
        plt.axhline(0, 0, 1, color='black', linestyle='solid', linewidth=1)

        # 이미지 저장 경로지정
        img_path = './temp_img.png'

        # 이미지 저장
        plt.savefig(img_path, facecolor='#eeeeee')

        # 텔레그램 채팅방으로 텍스트 전송
        total_msg = '[' + p_type + ' 장단기 금리차 : {:.2f}]'.format(fred_data.iloc[-1][0])
        asyncio.run(send_msg(total_msg))

        # 텔레그램 채팅방으로 사진 전송
        asyncio.run(send_img(img_path))

    except Exception as ex:
        print("send_fred_data() -> 함수 예외 발생! [내역 : " + str(ex) + "]")

def send_news_links():
    """네이버 뉴스 링크를 텔레그램 채팅방으로 전송"""
    try:

        # 출력할 기사 갯수
        count = 3

        # 네이버 뉴스 URL
        url = f'https://search.naver.com/search.naver?where=news&query=%EA%B9%80%EC%98%81%ED%95%84%EC%9D%98%203%EB%B6%84%20%EC%9B%94%EC%8A%A4%ED%8A%B8%EB%A6%AC%ED%8A%B8&sort=1&sm=tab_smr&nso=so:dd,p:all,a:all'

        # html 문서 파싱
        response = requests.get(url)
        soup = BeautifulSoup(response.text , 'html.parser')

        # 뉴스 제목 추출
        news_titles  =  soup.select('a.news_tit')
        titles = []

        for i in range (0, count):
            titles.append(news_titles[i].get('title'))

        # 네이버 뉴스 링크 추출
        news_links = soup.select('a.info')
        links = []

        for i in range(0, count * 2):
            if i % 2 == 0:
                continue # 신문사 URL 제외
            links.append(news_links[i].get('href'))

        # 최종 메세지 작성
        total_msg = ""
        for i in range(0, count):
            total_msg = total_msg + titles[i] + "\n"
            total_msg = total_msg + links[i] + "\n"

        # 텔레그램 채팅방으로 메세지 전송
        asyncio.run(send_msg(total_msg))

    except Exception as ex:
        print("send_news_links() -> 함수 예외 발생! [내역 : " + str(ex) + "]")


# 메인 함수
if __name__ == '__main__': 
    try:
        while True:

            # 현재 시간
            t_now = datetime.now()

            # 요일 저장
            t_weekday = datetime.today().weekday()

            # 알림 시간
            t_alert_start = t_now.replace(hour=7, minute=55, second=0, microsecond=0)
            t_alert_end = t_now.replace(hour=8, minute=5, second=0, microsecond=0)

            if t_weekday != 5 and t_weekday != 6:  # 토요일, 일요일 제외

                # 알림 시간 체크
                if t_alert_start < t_now < t_alert_end:
                    
                    if (t_now.minute == 0 and 0 <= t_now.second <= 11):
                        # 지수 알림
                        send_idx_info()
                        time.sleep(1)
                    
                        # 장단기 금리차 전송
                        s_code = ['T10Y2Y', 'T10Y3M'] # 10Y-2Y, 10Y-3M 금리차
                        for code in s_code:
                            send_fred_data(code)
                            time.sleep(1)

                        # 뉴스 전송
                        send_news_links()
                        time.sleep(65) # 중복으로 처리되지 않도록 65초 정도 일시정지

    except Exception as ex:
        print("__main__ -> exception!  [내역 : " + str(ex) + "]")

 

서버 혹은 코드를 계속해서 실행하실 수 있는 환경에서 위 코드를 계속 실행해 놓으신다면, 매일 아침 8시에 시황 관련 데이터를 텔레그램으로 받아보실 수 있게 됩니다.

 

 

[코드 설명]

우선, 텔레그램 봇 라이브러리가 20.0 버전으로 업데이트되면서, 채팅을 보내는 명령어 사용방식이 조금 변화가 있었는데요. 아래 코드처럼 비동기 방식으로 사용방법이 바뀌었습니다.

 

# 텔레그램 메세지 전송 함수
async def send_msg(p_text):
    bot = telegram.Bot(s_token)
    async with bot:
        await bot.send_message(text=p_text, chat_id=l_chat_id)

# 텔레그램 메세지 전송 함수
async def send_img(p_path):
    bot = telegram.Bot(s_token)
    async with bot:
        await bot.send_photo(chat_id=l_chat_id, photo=open(p_path, 'rb'))

 

그렇기 때문에, 텍스트 메시지나 사진 파일을 전송할 때는 아래 코드처럼 실행해야 전송이 됩니다. 

 

# 텔레그램 채팅방으로 메세지 전송
asyncio.run(send_msg(total_msg))

# 텔레그램 채팅방으로 사진 전송
asyncio.run(send_img(img_path))

 

3대 지수를 전송하는 함수를 보시면 yahoo_fin을 사용하여 코드를 작성하였습니다. get_quote_data 함수에 지수 코드를 입력해서 지수 데이터를 가져왔고, 가져온 데이터중 전일 지수와 금일 지수를 비교하여 등락 퍼센트를 계산하였습니다.

(지수코드 : 다우지수(^DJI), S&P 500(^GSPC), 나스닥(^IXIC))

 

def send_idx_info():
    """미국주식 3대 지수 전송"""
    try:
        # 1. 다우지수 구하기
        dow_data = si.get_quote_data('^DJI')
        dow_prev_price = round(dow_data['regularMarketPreviousClose'], 2)
        dow_current_price = round(dow_data['regularMarketPrice'], 2)
        dow_gap_price = round(dow_current_price - dow_prev_price, 2)
        dow_total_percent = round((dow_current_price - dow_prev_price) / dow_prev_price * 100, 2)
        
        total_msg = "[Dow Jones]\n"
        total_msg = total_msg + " 전일 종가 : " + str(dow_prev_price) + "\n"
        total_msg = total_msg + " 금일 종가 : " + str(dow_current_price) + " / " + str(dow_gap_price) + " (" + str(dow_total_percent) + "%)"
        
        # 아래 코드는 생략.......

 

메인 함수 부분에는 반복문을 통하여 코드가 계속 실행되도록 했고, datetime 모듈을 이용하여 매일 아침 8시에 알림이 발송되도록 코드를 작성하였습니다. 그리고, 토요일과 일요일은 알람이 불필요하기 때문에 제외시켰습니다.

 

# 메인 함수
if __name__ == '__main__': 
    try:
        while True:

            # 현재 시간
            t_now = datetime.now()

            # 요일 저장
            t_weekday = datetime.today().weekday()

            # 알림 시간
            t_alert_start = t_now.replace(hour=7, minute=55, second=0, microsecond=0)
            t_alert_end = t_now.replace(hour=8, minute=5, second=0, microsecond=0)

            if t_weekday != 5 and t_weekday != 6:  # 토요일, 일요일 제외

                # 알림 시간 체크
                if t_alert_start < t_now < t_alert_end:
                    
                    if (t_now.minute == 0 and 0 <= t_now.second <= 11):
                        # 지수 알림
                        send_idx_info()
                        time.sleep(1)
                    
                        # 장단기 금리차 전송
                        s_code = ['T10Y2Y', 'T10Y3M'] # 10Y-2Y, 10Y-3M 금리차
                        for code in s_code:
                            send_fred_data(code)
                            time.sleep(1)

                        # 뉴스 전송
                        send_news_links()
                        time.sleep(65) # 중복으로 처리되지 않도록 65초 정도 일시정지

 

 

 

실행 결과

 

위의 코드를 테스트해 보니 정상적으로 채팅방에 알림이 오는 것을 확인할 수 있었습니다.

 

채팅방 알림 테스트

 

 

 

마무리...

 

오늘은 미국증시 시황을 알려주는 알림 봇을 만들어 보았습니다. 다음 포스팅에서도 유용한 정보를 소개해드리는 시간으로 찾아뵙겠습니다. 감사합니다!