파이썬 자동 감지 프로그램(3)
2022.10.25 - [개발일지/주식 단타 전략] - 주식 단타 전략 모니터링 시스템 - 5. 파이썬 자동 감지 프로그램(1)
2022.10.25 - [개발일지/주식 단타 전략] - 주식 단타 전략 모니터링 시스템 - 5. 파이썬 자동 감지 프로그램(2)
지난 글에 이어 카카오톡 내게 메세지 보내기를 구현하는 것으로 1차 개발을 마무리하고자 한다.
앞선 내용를 정리하면
전날 가격을 이용해 range를 계산하고, 이 값으로 매수 기준가를 설정 한 다음, 당일 주식 가격이 기준가를 넘는 순간 매수 시그널을 보내는 것까지 구현했다.
시그널은 현재 Jupyter notebook의 셀 출력 값과 로그파일을 통해서 기록되기 때문에 직접 모니터를 보고 있어야지만 시그널을 보낸 것을 알 수 있었다.
이 때문에 다른 일을 하고 있어도 매수 시그널을 알려줄 무언가가 필요했다.
이런 이유로 가장 먼저 생각난 것이 카카오톡이다.
매수 시그널이 뜰 때 마다 메세지를 보내 알려줄 수 있다면, 다른 일을 하면서도 체크를 할 수 있겠다 생각했다.
로직을 간단히 설명하면
매수 기준가를 이용해 한 종목씩 모니터링를 진행하다 매수 기준가를 넘으면 해당 종목에 대한 정보를 카카오톡 API를 이용해 메세지를 보낸다.
이때 담고 있는 정보는 종목, 가격, 매수기준가, Range 등의 정보와 해당 종목을 자세하기 확인하기 위해 링크를 네이버 금융의 해당 종목 페이지를 링크로 보낸다.
한 가지, 카카오톡 API의 토큰은 만료기간이 있으므로 메세지 보내기를 실패했을 때 자동 갱신 후 다시 보내는 로직을 심어야 했다.
그리고, 자동감지 프로그램은 24시간 내내 돌아가도록 설계하였기 때문에 중간에 오류가 나서 멈춘다면 이 역시 메세지를 보내 대응할 수 있도록 만들어야 했다. 여기서도 토큰이 만료되면 오류 메세지를 못 보내기 때문에 토큰을 갱신하는 코드를 심었다.
그전에, 파이썬 자동 감지 프로그램(2)의 코드를 돌려놓으면 아침 9시에 에러가 발생해 더 이상 작동하지 않는 상황이 가끔 발생했다.
원인을 찾아보니까 주식 데이터를 불러오는 과정에서 9시에 개장을 했지만, 아직 거래가 이뤄지지 않거나, 업데이트가 늦는 경우에 당일 가격을 불러오지 못해 빈 데이터 프레임이 생성이 되었다.
항상 가격이 조회될 것이란 생각으로 구현했더니, 빈 프레임에 대응이 안 되었던 것이다.
따라서 아직 가격이 조회되지 않은 종목은 넘기고 가격이 생성되어 조회되는 종목부터 모니터링하도록 코드를 수정했다.
아래의 코드가 모든 내용이 반영된 최종 버전의 코드이다.
우선 저번 코드에 이어 필요한 패키지와, 토큰을 불러와아한다.
import requests
import json
with open("kakao_code.json", "r") as fp:
tokens = json.load(fp)
url = "https://kapi.kakao.com/v2/api/talk/memo/default/send"
headers = {
"Authorization": "Bearer " + tokens["access_token"]
}
def refreshToken(refresh_token):
REST_API_KEY = "자신의 REST_API_KEY"
REDIRECT_URI = "자신의 REDIRECT_URI"
data = {
"grant_type": "refresh_token",
"client_id":f"{REST_API_KEY}",
"refresh_token": refresh_token
}
resp = requests.post(REDIRECT_URI , data=data)
print(resp)
new_token = resp.json()
return new_token['access_token']토큰을 재발행해주는 함수까지 넣어놓으면 준비가 끝난다.
try:
while True:
x = dt.datetime.now()
file_name = "./monitoring_log/"+str(x.month) + str(x.day) +"_kospi_log"
time_log = str(x.month)+"월"+str(x.day)+"일 "+str(x.hour%24+9)+":"+str(x.minute)
if x.hour+9 >=9 and x.hour+9 <16:
with open(file_name+'.txt', 'a') as f:
f.write(time_log + " --- Monitoring 종목 개수 : "+ str(kospi_data.shape[0])+"\n")
set_before = True
print(x.hour+9,x.minute)
for t_i in kospi_data["Symbol"]:
t_n = kospi_data[kospi_data["Symbol"]==t_i]["Name"].values[0]
price_data = fdr.DataReader(t_i, t_date).tail(1)
if price_data.shape[0]!= 0:
date = price_data.index
if t_date != date : t_date = str(price_data.index)[16:26]
price_data.insert(0,"Symbol",t_i)
price_data.insert(0,"Name",t_n)
target_band = before_price[before_price["Symbol"]==t_i]["range"].values[0]
price_data["Price_stand"] = price_data["Open"] + int(target_band*1.5)
base_price = price_data["Price_stand"].values[0]
current_price = price_data["Close"].values[0]
bot_log = (time_log + " ### 프로그램 log test ### 종목명 : " + t_n + " / 종목코드 : " + t_i +
" / 현재가격 : " + str(current_price) + " / 기준가격 : " + str(base_price) +
" / Range : " +str(target_band) +"\n")
#log test
with open(file_name+'_test.txt', 'a') as f:
f.write(bot_log)
if current_price >= base_price :
kospi_data = kospi_data[kospi_data["Symbol"]!=t_i]
data = {"template_object" : json.dumps({
"object_type" : "text",
"text" : "매수사인"+"\n" +
"종목 : "+t_n +"\n" + "코드 : "+t_i +"\n"+ "매수가격 : "+str(current_price) +"\n"+
"배수 : "+str(np.round((current_price/base_price-1)*100,2)) +"\n"+
"Stand : "+ str(base_price)+ "\n"+
"Range : "+ str(target_band),
"link" : {"web_url": "https://finance.naver.com/item/main.naver?code="+str(t_i),
"mobile_web_url":"https://finance.naver.com/item/main.naver?code="+str(t_i)}
})}
response = requests.post(url, headers=headers, data=data)
if response.json().get('result_code') == 0:
print('메시지를 성공적으로 보냈습니다.')
print(t_n, current_price, base_price,target_band)
with open(file_name+'.txt', 'a') as f:
f.write(time_log + ' --- 메시지를 성공적으로 보냈습니다.' +"\n")
f.write(bot_log)
else:
tokens["access_token"] = refreshToken(tokens["refresh_token"])
headers = {"Authorization": "Bearer " + tokens["access_token"]}
response = requests.post(url, headers=headers, data=data)
if response.json().get('result_code') == 0:
print("토큰 갱신에 성공했습니다.")
print('메시지를 성공적으로 보냈습니다.')
print(t_n, current_price, base_price, target_band)
with open(file_name+'.txt', 'a') as f:
f.write(time_log + ' --- 토큰을 갱신했습니다.' +"\n")
f.write(time_log + ' --- 메시지를 성공적으로 보냈습니다.' +"\n")
f.write(bot_log)
else:
print("토큰 갱신에 실패했습니다.")
with open(file_name+'.txt', 'a') as f: f.write(time_log + ' --- 토큰 갱신에 실패했습니다.' +"\n")
elif x.hour+9 == 16 and x.minute >= 10 and set_before == True:
print(x.month, x.day,x.hour+9,x.minute)
price_data = fdr.DataReader(t_i, t_date)
date = str(price_data.tail(1).index)[16:26]
t_date = date
kospi_data = df_krx[df_krx["Market"].str.contains("KOSPI")]
kospi_data = kospi_data[kospi_data["Name"].isin(kospi_200_list[kospi_200_list["Symbol"]==102110].STK_NM_KOR)]
before_price = None
for t_i2 in kospi_data["Symbol"]:
t_n = kospi_data[kospi_data["Symbol"]==t_i2]["Name"].values[0]
price_data = fdr.DataReader(t_i2, t_date)
price_data.insert(0,"Symbol",t_i2)
price_data.insert(0,"Name",t_n)
before_price = pd.concat([before_price,price_data.tail(1)],axis = 0)
before_price["band"] = before_price["High"] - before_price["Low"]
set_before = False
print(x.month,x.day,"completed")
before_price.to_csv("./before_data/before_data_kospi"+str(x.month)+str(x.day)+".csv")
with open(file_name+'.txt', 'a') as f: f.write(time_log + ' --- Before_data를 갱신했습니다.' +"\n")
else:
with open(file_name+'.txt', 'a') as f: f.write(time_log + ' --- 거래시간이 아닙니다. 프로그램 대기중' +"\n")
time.sleep(60)
except:
print("Error")
with open(file_name+'.txt', 'a') as f: f.write(time_log + ' --- 코드 에러.' +"\n")
data = {"template_object" : json.dumps({
"object_type" : "text",
"text" : "코드에러"+"\n" ,
"link" : None
})}
response = requests.post(url, headers=headers, data=data)
if response.json().get('result_code') == 0:
with open(file_name+'.txt', 'a') as f: f.write(time_log + ' --- 코드 에러 메시지를 성공적으로 보냈습니다.' +"\n")
else:
tokens["access_token"] = refreshToken(tokens["refresh_token"])
headers = {"Authorization": "Bearer " + tokens["access_token"]}
response = requests.post(url, headers=headers, data=data)
if response.json().get('result_code') == 0:
with open(file_name+'.txt', 'a') as f: f.write(time_log + ' --- 코드 에러 토큰 갱신에 성공했습니다.' +"\n")
else:
with open(file_name+'.txt', 'a') as f: f.write(time_log + ' --- 코드 에러 토큰 갱신에 실패했습니다.' +"\n")
전체 코드인데, 기능을 하나씩 추가한 것이라 조금 중복적인 내용이 많다.
일단은 잘 돌아가니까.. 사용해보고 하나씩 정리해보겠다.
가장 먼저 try, except문이 보인다.
종목 모니터링 중 오류가 나면 except문으로 빠져서 코드 에러 메세지를 보낸다.
만약 여기서 메세지 보내기를 실패했다면, 토큰이 만료된 것이기 때문에, 토큰 재발행을 한 후 다시 갱신되었다는 메세지를 보낸다.
따라서 코드 에러 메세지를 받으면 바로 대응할 수 있도록 했다.
종목 감지 로직은 저번 글과 동일함으로 메세지 보내기 코드 위주로 설명하자면,
data = {"template_object" : json.dumps({
"object_type" : "text",
"text" : "매수사인"+"\n" +
"종목 : "+t_n +"\n" + "코드 : "+t_i +"\n"+ "매수가격 : "+str(current_price) +"\n"+
"배수 : "+str(np.round((current_price/base_price-1)*100,2)) +"\n"+
"Stand : "+ str(base_price)+ "\n"+
"Range : "+ str(target_band),
"link" : {"web_url": "https://finance.naver.com/item/main.naver?code="+str(t_i),
"mobile_web_url":"https://finance.naver.com/item/main.naver?code="+str(t_i)}
})}
response = requests.post(url, headers=headers, data=data)
if response.json().get('result_code') == 0:
print('메시지를 성공적으로 보냈습니다.')
print(t_n, current_price, base_price,target_band)
else:
tokens["access_token"] = refreshToken(tokens["refresh_token"])
headers = {"Authorization": "Bearer " + tokens["access_token"]}
response = requests.post(url, headers=headers, data=data)
if response.json().get('result_code') == 0:
print("토큰 갱신에 성공했습니다.")
print('메시지를 성공적으로 보냈습니다.')
print(t_n, current_price, base_price, target_band)
else:
print("토큰 갱신에 실패했습니다.")우선 data부분이 카카오톡 API로 메세지를 보내기 위한 자료를 만는 것이다.
text 부분에 내가 필요한 정보인 종목, 코드, 가격 등을 저장한다.(필요한 정보에 따라 수정하면 된다.)
link 부분은 제외하려고 했다가 메세지 보내기에 기본적으로 들어가 있어서 어떻게 활용할까 고민하다가 종목별로 네이버금융 페이지를 연결했다.
현재 가격이 기준 가격을 넘어서 매수 시그널을 보내면, 바로 매수할 수도 있지만, 어떤 종목인지 상세히 보고 매수를 결정하기도 해서 종목코드를 이용해 링크를 만들어 보낸다.
따라서 카카오톡 메세지를 받고 링크를 누르면 바로 정보 페이지와 바로 연결되기 때문에 검색하고 찾아가는 시간을 아낄 수 있었다.
전송할 자료를 만들고 난 후는 앞서 정리했던 메세지 보내기 코드이다.
만약 실패했다면, 토큰이 만료된 것이기 때문에 갱신 후 메세지를 다시보낸다. 그런데 두 번째 메세지 보내기도 실패한 경우는 토큰 갱신에 실패했다는 것이다.
토큰은 유효기간뿐만 아니라 재발행할 수 있는 기간도 제한이 되어있다. 문서상 몇 개월 정도 사용할 수 있다고 한 것 같은데 이게 만료되면 토큰 자체를 새로 발행받아아 한다.
이런 경우 번거롭지만, 카카오톡 developers에서 새로 발행받아 REST API KEY를 변경해주어야 한다.
2022.10.24 - [개발일지/주식 단타 전략] - 주식 단타 전략 모니터링 시스템 - 4. 카카오톡 API 나에게 메세지 보내기
정상적으로 코드를 구현했다면, 카카오톡으로 이런 메세지가 올 것이다.
내가 저장한 정보들과 자세히 보기(네이버 링크)를 가진 메세지를 확인할 수 있다.
자세히 보기를 누르면
이렇게 링크를 타고 해당 종목의 상세정보를 확인 할 수 있다.
이것으로 감지 프로그램의 구현을 끝냈다.
초기에 생각했던 내용을 대부분 구현이 되었기 때문에, 단타 전략 프로그램 구현에 대한 글은 여기까지 쓰도록 하겠다.
이제 실제로 내가 사용해 볼 차례인데.. 어떤 종목을 대상으로 할지 정해야 할 것 같다.
코스피 200개 기업을 대상으로 하면 모든 종목을 한 번 다 보는데 약 100초가 걸린다.
빠르게 봐야 하는 경우 종목을 조절해야 할 것이다.
그리고 실제로 써먹기 위해선 몇 가지 수정이 필요하다 생각된다.
- 단타 전략 수정 - 전날 가격 기준 range를 이용해 매수기준을 잡는데 좀 더 디테일하게 설정이 필요할 것 같다.
- 매수 타이밍 선정 - 매수 시그널이 떴을 때 바로 사는 게 좋은지, 종가에 사는게 좋은지 테스트가 필요하다.
- 카카오톡 소리 알림 기능 없음 - 메세지가 와도 소리 알림이 없으니 pc카톡으로 한쪽 구석에 띄워 놓거나 해야 한다. 더 좋은 알림 방법을 찾아야 할 것 같다.
- 전략이 수정되면 증권 API와 연결해 자동 매매가 이뤄지게 바꿀 예정이다. 아무래도 사람의 손으로 하나하나 매수, 매도를 하는 것을 번거롭기 때문이다.
위의 내용은 당장 계획을 하고 있지 않기 때문에 테스트 혹은 구현할 때마다 추가 글로 소개를 하겠다.
그리고 장기적으론 단타 전략이 아닌 종목 분석 전략도 하나씩 구현 혹은 만들어 보려 한다.
딥러닝을 이용해서 주가를 예측하는 논문이 많이 나오고 있다.
시간이 허락한다면, 몇 가지를 소개하는 글을 써보도록 하고, 추후엔 직접 구현하는 것까지 해보겠다.
이상 주식 단타 전략 모니터링 시스템 개발 일지를 마친다.
'개발일지 > 주식 단타 전략' 카테고리의 다른 글
| 6. 주식 단타 전략 모니터링 시스템 - 파이썬 자동 감지 프로그램(2) (0) | 2022.10.25 |
|---|---|
| 5. 주식 단타 전략 모니터링 시스템 - 파이썬 자동 감지 프로그램(1) (0) | 2022.10.25 |
| 4. 주식 단타 전략 모니터링 시스템 - 카카오톡 API 나에게 메세지 보내기 (0) | 2022.10.24 |
| 3. 주식 단타 전략 모니터링 시스템 - 단기 기술적 지표 백테스팅 (0) | 2022.10.19 |
| 2. 주식 단타 전략 모니터링 시스템 - 매수 기준 감지 (2) | 2022.10.19 |
댓글