9. 투자지표를 활용한 매매 시점 모니터링

import pandas as pd
import numpy as np
import datetime as dt
import time 
import matplotlib.pyplot as plt
import FinanceDataReader as fdr
import copy
import pickle

이번 글은 다른 글의 코드 없이 위의 패키지부터 시작하면 구동이 가능하다.

df_krx = fdr.StockListing('KRX')

def rsi(ohlc: pd.DataFrame,y_label, period: int = 14): 
    delta = ohlc[y_label].diff() 
    ups, downs = delta.copy(), delta.copy() 
    ups[ups < 0] = 0 
    downs[downs > 0] = 0 
    AU = ups.ewm(com = period-1, min_periods = period).mean() 
    AD = downs.abs().ewm(com = period-1, min_periods = period).mean()
    RS = AU/AD 
    return pd.Series(100 - (100/(1 + RS)), name = "RSI") 

kospi_data = df_krx[(df_krx["Market"]=="KOSPI") & (~df_krx["Sector"].isna())]

stock_list = list(kospi_data.Name)

len(stock_list)

Kospi 종목 리스트를 부르고, RSI 계산 코드 함수를 작성해준다.

%%time
lower_cut = 30
upper_cut = 70
check_df = None
for i, kn in enumerate(stock_list):
    print(i,kn)
    ks = df_krx[df_krx["Name"]==kn].Symbol.values[0]
    price_data1 = fdr.DataReader(ks,"2020","2022-11-14")
    price_data1["stock_name"]= kn
    RSI_df = np.round(rsi(price_data1, "Close",14),2)
    price_data1["RSI"] = RSI_df
    price_data1["RSI_signal"] = np.round(price_data1["RSI"].rolling(9).mean(),2)
    price_data1["lower"] = None
    price_data1["upper"] = None
    price_data1["buy_tag"] = 0
    price_data1["sell_tag"] = 0
    
    lower_list = False
    upper_list = False
    now_index_list = 0

    for ti in price_data1.index:
        tmp = price_data1[price_data1.index == ti]
        now_index_list = tmp.RSI.values[0]

        if now_index_list <=lower_cut:
            lower_list = True
            
        elif now_index_list > lower_cut and lower_list == True:  
            lower_list = False
            
            price_data1.loc[price_data1.index == ti,"buy_tag"] = 1
            print("buy")
            
        elif now_index_list >upper_cut:
            upper_list = True
            
        elif now_index_list <= upper_cut and upper_list == True:     
            upper_list = False
            price_data1.loc[price_data1.index == ti,"sell_tag"] = 1
            print("sell")
            
        price_data1.loc[price_data1.index == ti,"lower"] = lower_list
        price_data1.loc[price_data1.index == ti,"upper"] = upper_list
        
    check_df = pd.concat([check_df,price_data1],axis = 0)
    
check_df.to_csv("stock_data_2022.csv")
# check_df = pd.read_csv("stock_data_2022.csv",index_col = 0)
# check_df.index = pd.to_datetime(check_df.index)

각각의 종목을 2020년부터 데이터를 불러와 RSI, RSI_signal, RSI 전략 조건에 의한 매매 시그널을 붙여준다.

14일까지 데이터를 만든 이유는 글 작성일기준이기도 하지만, 매일매일 모든 데이터를 불러와 계산하기는 비효율적이기 때문에 전날 데이터에 대한 정보를 세팅해놓는다.

즉, 15일에 계산하기 위해서 14일까지 계산된 데이터를 불러와 15일 데이터만 붙여 해당 일에 대한 매매 시그널 분석만 하면 될 것이다.

또한, 2020년부터 데이터를 가져온 이유는 RSI를 계산할 때 최근 일에 가중치를 더 주는 방식이 적용되서인데, 이에 대한 기준이 약 150일 정도 되어야 네이버 RSI값과 동일해서 초기 데이터를 여유롭게 수집했다.

14일까지 만들어진 데이터이다. RSI 값과 lower, upper 스위티, buy, sell tag가 만들어진 것을 볼 수 있다.

이제 15일 기준의 매매 시그널 정보를 계산해야 한다.

t_date = "2022-11-15"

처음에 기준일을 설정해준다. 그 이후는 자동으로 날짜가 바뀌는 로직을 넣어놨기 때문에 실행시키면 된다.

while True:
    set_rsi_report = True
    x = dt.datetime.now()
    time_log = str(x.month)+"월"+str(x.day)+"일 "+str(x.hour%24+9)+":"+str(x.minute)
    set_rsi_report = True
    if x.hour+9 == 15 and x.minute == 35 and set_rsi_report == True:
        print(x.month, x.day,x.hour+9,x.minute)
        price_data = fdr.DataReader("033780", t_date)
        date = str(price_data.tail(1).index)[16:26]
        if t_date != date:
            t_date = date
            print(time_log,"reporting")
            df_krx = fdr.StockListing('KRX')
            kospi_data = df_krx[(df_krx["Market"]=="KOSPI") & (~df_krx["Sector"].isna())]
            stock_list = list(kospi_data.Name)
            today_stock_data = None
            for i, kn in enumerate(stock_list):
                print(i,kn)
                ks = df_krx[df_krx["Name"]==kn].Symbol.values[0]
                price_data1 = fdr.DataReader(ks,t_date).tail(1)
                price_data1["stock_name"]= kn
                price_data1["RSI"] = 0
                price_data1["RSI_signal"] = 0
                price_data1["lower"] = None
                price_data1["upper"] = None
                price_data1["buy_tag"] = 0
                price_data1["sell_tag"] = 0

                today_stock_data = pd.concat([today_stock_data,price_data1],axis = 0)
            today_stock_data.to_csv("day_stock_tmp.csv")
            today_stock_data = pd.read_csv("day_stock_tmp.csv",index_col = 0)
            today_stock_data.index = pd.to_datetime(today_stock_data.index)
            today_stock_data["lower"] = None
            today_stock_data["upper"] = None
            
            check_df = pd.concat([check_df,today_stock_data],axis = 0)

            check_df2 = None
            for i, kn in enumerate(stock_list):
                price_target = check_df[check_df["stock_name"]==kn].tail(200)
                if price_target.shape[0]>20:
                    RSI_df = np.round(rsi(price_target, "Close",14),2)
                    price_target["RSI"] = RSI_df
                    price_target["RSI_signal"] = np.round(price_target["RSI"].rolling(9).mean(),2)

                    lower_list = price_target[:-1].tail(1).lower.values[0]
                    upper_list = price_target[:-1].tail(1).upper.values[0]

                    tmp = price_target[price_target.index == t_date]
                    now_index_list = tmp.RSI.values[0]

                    if now_index_list <=lower_cut:
                        lower_list = True

                    elif now_index_list > lower_cut and lower_list == True:  
                        lower_list = False
                        price_target.loc[price_target.index == t_date,"buy_tag"] = 1
                        print("buy")

                    elif now_index_list >upper_cut:
                        upper_list = True

                    elif now_index_list <= upper_cut and upper_list == True:     
                        upper_list = False
                        price_target.loc[price_target.index == t_date,"sell_tag"] = 1
                        print("sell")
                    price_target.loc[price_target.index == t_date,"lower"] = lower_list
                    price_target.loc[price_target.index == t_date,"upper"] = upper_list
                else:
                    price_target.loc[price_target.index == t_date,"lower"] = None
                    price_target.loc[price_target.index == t_date,"upper"] = None
                    
                check_df2 = pd.concat([check_df2,price_target],axis = 0)

            check_df2.to_csv("./RSI_monitoring_log/stock_rsi_data.csv")

            
            with open('RSI_monitoring_log/rsi_monitoring_log.txt', 'a') as f: f.write(time_log + ' --- rsi 모니터닝 완료.'  +"\n")   
            
            check_today = check_df2[check_df2.index == t_date]
            check_buy = check_today[check_today["buy_tag"]==1]
            check_sell = check_today[check_today["sell_tag"]==1]
            check_buy.to_csv("./RSI_monitoring_log/stock_rsi_buy"+str(x.month)+"월"+str(x.day)+"일.csv")
            check_sell.to_csv("./RSI_monitoring_log/stock_rsi_sell"+str(x.month)+"월"+str(x.day)+"일.csv")
            
            check_df = check_df2.copy()
            del check_df2, today_stock_data, price_target,check_buy,check_sell
            
        else:
            with open('RSI_monitoring_log/rsi_monitoring_log.txt', 'a') as f: f.write(time_log + ' --- rsi 거래일이 아닙니다.'  +"\n")   
            time.sleep(50)
        set_rsi_report = False
        
    else:
        if x_tmp != x.hour:
            x_tmp = x.hour
            with open('RSI_monitoring_log/rsi_monitoring_log.txt', 'a') as f: f.write(time_log + ' --- rsi 모니터링 대기중.'  +"\n")   
        time.sleep(50)

주요 로직을 요약하면 다음과 같다.

장이 끝나는 매일 15시 35분이 되면 해당 로직이 실행된다. 이때 휴장 하는 날이면 거래일이 else문으로 넘어가 '거래일이 아닙니다.'를 로그에 저장하고, 거래일이면 15일 종가 기준의 데이터를 수집한다. 수집을 완료하면 "모니터링 완료" 로그를 기록한다.
전날까지 수집한 데이터에 15일 데이터를 붙여서 RSI 및 매매 시그널을 계산한다.
당일 주가가 반영된 데이터를 저장하고 이중 매수 시그널이 난 종목, 매도 시그널이 난 종목을 따로 저장한다.
15시 35분이 아닌 경우 마지막의 else문으로 들어가며 50초 대기 후 다시 while문을 돌린다. 60초로 설정했었는데 로직이 돌며 날짜를 업데이트하는 과정에서 1분 단위를 건너뛰는 경우가 생겨 모니터링을 스킵하기에 여유롭게 50초로 설정했다.
1시간마다 프로그램이 잘 동작 중인지 로그를 기록한다 - "모니터링 대기 중"

중간에 삼성전자의 종목코드를 이용해 날짜를 업데이트하는 부분이 있다. 삼성전자는 망할 일이 없다고 생각해서 지정한 것이고 큰 의미는 없다. 단순히 날짜 업데이트 용이라 생각하면 된다.

set_rsi_report는 True와 False값을 가지는데, 위의 로직에서는 의미 없다. 나중에 카카오톡 API로 구동 중인 변동성 전략 모니터링 시스템에 심기 위해 구현해 놓은 것이다.

2022.10.28 - [개발일지/주식 단타 전략] - 7. 주식 단타 전략 모니터링 시스템 - 파이썬 자동 감지 프로그램(3)

위의 로직이 잘 실행되면 이런 파일들이 저장될 것이다.

우선 rsi_monitoring_log.txt를 보면

프로그램 상태를 볼 수 있다.

대기 중인 상태가 아니면, 로직에 오류가 있다는 것이므로 점검이 필요하다.

buy와 sell 파일을 보면

이런 결과가 나오는 것을 볼 수 있다.

15일은 매수 시그널이 1개 매도 시그널이 16개 나왔다. 시장이 전반적으로 하락 추세인가 보다.

프로그램이 while문으로 되어있기 때문에 계속 실행되어있으면서 매일 장이 마감되면 25분 정도 결과를 낸 후 파일이 저장된다. 그럼 시장의 흐름이나 매매 시그널이 뜬 종목을 보고 다음날 매수하던지, 아니면 시간외거래를 통해 매수하면 될 것 같다.

후에 장 중에 1시간 간격으로 혹은 고가를 기준으로 로직을 수정해 활용해 볼 수 있을 것 같다.

글 중간에서도 언급했지만, 변동성 돌파 전략으로 실행 중인 시스템에 RSI 혹은 새로운 투자 지표를 반영한 모니터링 시스템을 하나로 묶어서 내가 필요한 정보들을 실시간으로 얻을 수 있는 시스템으로 만들 생각이다.

시간이 걸리겠지만 차근차근 엮어나가면서 소개하겠다.

위의 프로그램은 RSI값을 이용한 종목 분석일 뿐이니 투자 실행은 종목 분석과 본인의 판단이 필요한 부분이기 때문에 참고하는 자료로만 사용하면 좋을 것 같다.

728x90