관심종목의 RSI, Stochastic Slow 값을 텔레그램으로 받아보기 (feat. Python, AWS Lambda, Amazon EventBridge, Amazon S3)

 

주식을 매수할 때 RSI, Stochastic Slow 값을 참고하여 각각 30, 20 이하일 때는 더 떨어질 여력이 얼마 남아 있지 않은 단기 저점이므로 과감하게 매수한다. 반대로 각각 70, 80 이상일 때는 절대 매수하지 않는다.

 

10개가 넘는 관심종목의 차트를 일일이 열어보기 힘들어서, 점심시간에 텔레그램 채널로 받아볼 수 있게 만들었다.

 

파이썬, AWS에 대한 지식과 경험이 어느정도 있는 분들은 아래 자료를 참고하여 자신만의 채널을 만들 수 있을 것이다.

 

Amazon S3

미국 종목의 경우 외우기 쉬운 Ticker(e.g., AAPL)를 사용하면 되지만, 한국 종목의 경우 종목코드를 사용해야 하는데 외우고 다니는 사람은 없을 것이므로 관심종목을 입력 받을 때 종목명도 사용할 수 있게 만들었다.

 

네이버 금융에서 종목명, 종목코드 확인

tickers.json

0.09MB

 

종목명을 인식하기 위해 종목명과 종목코드를 매핑해 놓은 파일을 S3에 저장해놓고, 람다가 처음 실행될 때 가져와 활용하도록 했다. FinanceDataReader로 가져올 수도 있지만 시간이 꽤 오래 걸리는 작업이어서 S3에 캐싱하는 방법을 선택했다.

 

tickers.json을 다운로드 받아 S3에 업로드하고 버킷 이름을 기억해두자. 파이썬 코드에 입력해야 한다.

 

신규 상장종목의 경우 종목명을 입력으로 사용하려면 tickers.json에 추가하거나, 종목명 대신 종목코드를 사용해야 한다.

 

AWS Lambda

런타임: Python 3.8

메모리(MB): 512

제한시간: 1분

 

주식 데이터를 읽어오기 위해 필요한 라이브러리가 포함된 레이어를 람다에 추가해야 한다.

 

fdr_libs.zip

 

텔레그램으로 메시지를 보내기 위한 필요한 라이브러리가 포함된 레이어를 람다에 추가해야 한다.

 

telegram_libs.zip

파이썬 코드는 아래와 같다. 상단의 S3 Bucket 이름을 자신의 것으로 업데이트하고, 람다에서 접근할 수 있도록 읽기 권한을 부여해야 한다.

 

get_latest_data()에서 1주일 전부터 데이터를 가져오는 이유는 휴장일 고려한 것이다. get_price_df()에서 1달 전부터 데이터를 가져오는 이유는 RSI를 계산하기 위해선 최근 14 영업일의 데이터가 필요하기 때문이다. 여기에 휴장일까지 고려하여 1달의 여유를 두었다.

 

import json
import datetime
import math
import numpy as np
import pandas as pd
import FinanceDataReader as fdr
import concurrent.futures
import boto3
import telegram


s3 = boto3.client('s3')
obj = s3.get_object(Bucket='{{your-bucket}}', Key='tickers.json')
all_tickers = json.loads(obj['Body'].read().decode('utf-8'))


def _get_rsi(df, n=14):
    u = np.where(df.Close.diff(1) > 0, df.Close.diff(1), 0)
    d = np.where(df.Close.diff(1) < 0, df.Close.diff(1) * (-1), 0)
    au = pd.DataFrame(u).rolling(window=n, min_periods=n).mean()
    ad = pd.DataFrame(d).rolling(window=n, min_periods=n).mean()
    rsi = au.div(ad+au) * 100
    rsi.index = df.index
    df = df.assign(RSI=rsi).dropna()
    return df


def _get_stochastic(df, n=15, m=5, t=3):
    ndays_high = df.High.rolling(window=n, min_periods=1).max()
    ndays_low = df.Low.rolling(window=n, min_periods=1).min()

    fast_k = ((df.Close - ndays_low) / (ndays_high - ndays_low)) * 100
    slow_k = fast_k.ewm(span=m).mean()
    slow_d = slow_k.ewm(span=t).mean()

    df = df.assign(StoK=slow_k, StoD=slow_d).dropna()

    return df


def get_price_df(ticker, start_date, end_date=None):
    df = fdr.DataReader(ticker, pd.to_datetime(
        start_date) - pd.DateOffset(months=1), end_date)
    df = df[df.Volume > 0]
    df = _get_stochastic(df, 5, 5, 3)
    df = _get_rsi(df, 14)
    df = df[start_date:]
    return df


def get_latest_data(ticker):
    a_week_ago = pd.to_datetime("today") - pd.DateOffset(weeks=1)
    df = get_price_df(ticker, a_week_ago)
    return {
        'STC': math.floor(df.iloc[-1].StoK),
        'RSI': math.floor(df.iloc[-1].RSI),
        'Change': np.round(df.iloc[-1].Change * 100, 2),
        'Close': df.iloc[-1].Close,
    }


def convert_tickers(tickers):
    converted_tickers = []
    for ticker in tickers:
        converted_tickers.append(
            ticker if ticker not in all_tickers else all_tickers[ticker])
    return converted_tickers


def do(tickers, threads=8):
    result = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=threads) as executor:
        converted_tickers = convert_tickers(tickers)
        for ticker, data in zip(tickers, executor.map(get_latest_data, converted_tickers)):
            result.append((ticker, data))
    return sorted(result, key=lambda x: x[1]['RSI']*100 + x[1]['STC'])


def send_telegram_message(token, chat_id, result):
    bot = telegram.Bot(token=token)
    body = ""
    for item in result:
        ticker = item[0]
        data = item[1]
        rsi = data['RSI']
        stc = data['STC']
        change = data['Change']
        if change > 0:
            change = f'+{change}'
        if ticker.startswith(('TIGER', 'KINDEX', 'KODEX', 'KBSTAR', 'KOSEF', 'ARIRANG')):  # Korean ETF
            body += f'{ticker} {change}% {rsi} {stc}\n'
        else:
            close = data['Close']
            body += f'{ticker} {close:,g} {change}% {rsi} {stc}\n'
            
    bot.sendMessage(chat_id=chat_id, text=body)


def lambda_handler(event, context):
    result = do(event['tickers'], threads=8)
    send_telegram_message(event['telegram_token'],
                          event['telegram_chat_id'], result)
    print(result)
    return {}

 

Amazon EventBridge

 

규칙을 만들 때 위와 같이 Cron 식으로 일정을 입력하면 한국시간으로 월~금 오후 12시에 이벤트가 실행된다.

 

 

이벤트가 발생했을 때 호출할 대상으로 람다 함수를 선택하고 JSON 형식의 입력을 추가해야 한다. 포맷은 아래와 같다. 텔레그램 Chat ID와 Token, 관심종목 리스트를 자신의 것으로 업데이트해야 한다. 미국 종목의 경우 Ticker(e.g., TSLA)를 그대로 사용하면 된다.

 

{
  "telegram_chat_id": {{telegram_chat_id}},
  "telegram_token": "{{telegram_token}}",
  "tickers": [
    "LG우",
    "삼성전자우",
    "현대차3우B",
    "청담러닝",
    "카카오",
    "NAVER",
    "엑셈",
    "더존비즈온",
    "뷰웍스",
    "케이아이엔엑스",
    "JYP Ent."
  ]
}

 

참고자료

Simple Telegram Bot with Python and AWS Lambda