전 게시글에서는 1분단위 거래금액을 조회했는데, 이번엔 가격이다.
1분단위로 가격이 확확 변할 일이 많지는 않겠지만, 특정 시간대에 종종 생긴다.
그 시간대에 올랐던 친구들은 대략 20분정도는 유지했던 것 같아서 오로지 기억에 의지해서 프로그램을 만들어보자.
사실 이게 몇분정도 유지되는지도 프로그램으로 확인해보고싶은데, 글로 먼저 적어서 방향을 잡고 하는게 좋을 것 같아서 뒤로 미뤘다.
대충 방향을 정리해보자면 전체 종목에 대해서 거래금액을 합산해서 1분전에 비해 거래량이 급증하는 시간대를 찾는다. (개인적으로 오전9시를 생각중이며, 프로그램을 통해 거래금액이 급증하는 시간대를 조회했을 때 9시가 나오면 성공이라고 생각한다.)
그 시간대를 찾았으면, 약 1주일정도의 과거데이터를 기반으로 해당 시간대(편의상 9시로 기술)부터 9시~9시2분 사이에 가격이 5%이상 올랐으며, 그 상승세가 몇 분이나 가는지, 오르기 전 가격으로 다시 돌아가는지, 10분, 20분, 30분정도 지켜봤을 때 분봉의 2/3 이상이 상향세를 띄고 있는지를 조회한다.
이정도로 해서 어떤 공식을 찾고, 최소 10분 유지된다고 하면 9시 2~4분에 들어가서 8~9분에 나와도 야금야금 수익을 실현할 수 있을거라 생각한다.(어디까지나 믿음이다.)
이건 다음에 구현하기로 하고, 우선 지금 눈 앞의 과제에 집중하자.
우선, 1분단위로 가격변화를 불러오기 위해 get_minute_price_difference 함수를 만들었다.
이 함수는 get_minute_volume_difference 함수에서 volume을 price로 바꾼 함수다.
또한, get_minute_candles 함수의 파라미터 중 unit을 지우고, 1분봉으로 고정하였다.
def get_minute_candles(self, symbol, count):
# 종목의 분단위 캔들 데이터 반환
candles = pyupbit.get_ohlcv(symbol, interval="minute1", count=count)
return candles대략적인 수정을 마치고, 1분단위로 가격을 비교해 result.xlsx 파일에 저장하는 get_minute_price_difference 함수는 아래와 같다.
def get_minute_price_difference(self):
# 종목의 분단위 캔들 데이터에서 거래금액 추출
krw_markets = self.get_all_krw_markets()
# 10개씩 조회
batch_size = 10
for i in range(0, len(krw_markets), batch_size):
batch_symbols = krw_markets[i:i + batch_size]
for symbol in batch_symbols:
try:
# 2분 전 데이터 조회
candles_2min = self.get_minute_candles(symbol, count=2)
if len(candles_2min) >= 2:
close_2min = candles_2min.iloc[0]['close']
# 1분 전 데이터 조회
candles_1min = self.get_minute_candles(symbol, count=1)
if len(candles_1min) == 1:
close_1min = candles_1min.iloc[0]['close']
# 현재 날짜와 시간
now = datetime.now()
date_time = now.strftime("%Y-%m-%d %H:%M:%S")
diff_percent = (-1+(close_1min/close_2min))
# 데이터 구성
data = {
'Date': [now.strftime("%Y-%m-%d")],
'Time': [now.strftime("%H:%M:%S")],
'Symbol': [symbol],
'close_2min': [close_2min],
'close_1min': [close_1min],
'diff_percent' : [diff_percent]
}
# 결과 파일에 추가
self.append_to_excel(data)
print(
f"{symbol}: 2min: {close_2min:.2f}, 1min: {close_1min:.2f} - Data added to result.xlsx")
else:
print(f"No data available for {symbol} (1min data).")
else:
print(f"No data available for {symbol} (2min data).")
except Exception as e:
print(f"Error for {symbol}: {e}")
# 1초 간격으로 조회
time.sleep(1)
이렇게 만들어놓으니 새로운 욕심이 생겼다. 업비트가 허락하는 선에서 (초당 10회의 캔들 조회) 분봉을 계속 새로고침해 갑자기 1%(임의) 이상 변화가 생기면 슬랙으로 보낸다든지,,,,
근데 이게 아무리 초당 10회라지만, 분당 600회니까 요청수 제한에 걸리지 않을까 싶어서 업비트 고객센터에서 찾아봤다. 제한은 분당 900회로 제한 안에 들어가긴 하는데 너무 많이 요청하는게 아닐까 싶다... 아무튼 제한 안에 들어가니까 일단 코드를 만들어보자.
이제 코드를 수정해야 할 부분은 무한루프를 만들고, 엑셀에 굳이 출력하지 않고, 가격비교만 남겨두고 1% 이상인지 아닌지를 비교하고, 슬랙으로 전송할 수 있는 코드를 만들어야 한다.
우선 슬랙의 경우 저번 주식 프로그램에서 사용하던 툴을 그대로 사용할 것이기 때문에 슬랙전송을 제외하고 챗지피티에게 물어보도록 하자.
내 질문 목록은 아래와 같다.
| 순서 | 질문 |
| 1 | pyupbit 라이브러리의 ohlcv 함수를 이용해서 2분전과 1분전의 가격을 비교하는 코드를 만들어줘 |
| 2 | 이 데이터들에서 2분전 가격과 1분전 가격의 변화를 보여주는 변수를 만들어줘 |
| 3 | 변화를 백분율로 표시해줘 |
| 4 | 이 결과를 슬랙으로 보내고싶어 |
| 5 | float 변수를 string으로 변환하는 코드를 알려줘 |
1~4번 질문들로 백분율로 표시된 가격의 변화율이 1 이상이면 (1%) 슬랙으로 보내는 코드까지 알아냈다.
이제 이 코드들을 내 코드들에 맞게 적당히 수정해줘야 한다.
하지만 슬랙을 안다룬지 어언 3~4년... 나는 그새 다 까먹고 열심히 내 블로그 글들을 다시 읽었으며,, 기존에 사용하려던 코드들이 아닌 챗지피티가 알려준 새로운 코드를 사용하게 되었다. 슬랙으로 보내는 과정에서 오랜만에 슬랙에 접속해서 옛날에 사용하던 채팅방의 주소와 어쩌구저쩌구를 알아봐야했고,,,,
https://api.slack.com/apps/ 이 주소에 접속해서 내 앱 목록을 확인하고 (없으면 만들고)
앱을 눌러 설정에 들어간 후 Add features and functionality 를 눌러 Incoming Webhooks를 들어간다.
아래로 스크롤을 내려보면 아래 사진과 같이 채널별로 URL이 나오는데 사용할 채널의 URL을 복사해서 코드에 붙여넣으면 된다.
나는 이미 만들어놔서 있는건지 모르겠다. 없으면 추가해서 사용하면 될듯?
아무튼 슬랙의 경우 나는 키움프로젝트에서 하던대로 config 폴더를 만들어 slack.py를 두었으나 코드를 보니 그럴 필요 없이 main.py에서 모두 해결해도 될 것 같다. (챗지피티에게 물어보세요!) 내 코드는 물론 아래에 공유하겠음..
테스트하는 과정에서 1분간 가격이 1%이상 상승한 종목이 없어서 테스트에 어려웠다,, 0.1%로 변경했더니 수많은 알림이,,
아무튼 여기까지 만들었으니 얘를 반복할 함수를 만들면 오늘 목표는 마무리다.
마지막 질문으로 무한루프 안에 넣어달라고 요청했고 내 코드에 맞게 수정하였다.
챗지피티의 조언에 맞게 time.sleep(1)은 time.sleep(60)으로 조정하였고 오늘 추가한 get_minute_price_difference 함수는 뒤에 _to_slack 을 붙여 함수이름을 수정하였고 최종 코드는 아래와 같다.
* time.sleep(1) 조정하지 마세요 10개 조회하고 60초 기다리기 힘듦...
def get_minute_price_difference_to_slack(self):
# 종목의 분단위 캔들 데이터에서 거래금액 추출
krw_markets = self.get_all_krw_markets()
while True:
# 10개씩 조회
batch_size = 10
for i in range(0, len(krw_markets), batch_size):
batch_symbols = krw_markets[i:i + batch_size]
for symbol in batch_symbols:
try:
# 2분 전 데이터 조회
candles_2min = self.get_minute_candles(symbol, count=3)
if len(candles_2min) >= 3:
close_2min = candles_2min.iloc[-3]['close']
# 2분 전 가격
close_1min = candles_2min.iloc[-2]['close']
# 1분 전 가격
# 2분 전과 1분 전 가격의 변화
diff_percent = (-1+(close_1min/close_2min)) * 100
# # 현재 날짜와 시간
# now = datetime.now()
# date_time = now.strftime("%Y-%m-%d %H:%M:%S")
#
# # 데이터 구성
# data = {
# 'Date': [now.strftime("%Y-%m-%d")],
# 'Time': [now.strftime("%H:%M:%S")],
# 'Symbol': [symbol],
# 'close_2min': [close_2min],
# 'close_1min': [close_1min],
# 'diff_percent' : [diff_percent]
# }
#
# # 결과 파일에 추가
# self.append_to_excel(data)
# print(f"{symbol}: 2min: {close_2min:.2f}, 1min: {close_1min:.2f} - Data added to result.xlsx")
#
# 데이터를 엑셀에 저장하지 않고 1% 이상 상승시 슬랙으로 전송하도록 코드 변경
slack_message = str(symbol) + ", " + str(diff_percent)
# print(slack_message)
if diff_percent >= 1:
# 슬랙
self.slack.send_slack_message(slack_message)
else:
print(f"No data available for {symbol} (min data).")
except Exception as e:
print(f"Error for {symbol}: {e}")
# 60초 간격으로 조회
time.sleep(60)
중간 주석은 기존에 있던 코드들을 나중에 재활용 할 수 있을 것 같아서 주석으로만 남겨두었다.
slack.py의 경우 아래 사진과 같이 config 폴더 안에 추가하였고 코드는 아래와 같다.
import requests
class Slack():
def send_slack_message(self, message):
self.webhook_url = 'WEBHOOK_URL'
data = {'text': message}
response = requests.post(self.webhook_url, json=data)
if response.status_code != 200:
print('Failed to send message to Slack')
else:
print('Message sent to Slack successfully')
마지막으로 에러가 조금 발생할텐데 main.py에 아래와 같이 임포트를 진행하고, 초기 설정부분에 slack을 선언해주면 된다.
from coinStudy.config.slack import *
이상 스불재 좌충우돌 코인프로그램 만들기 3일차 종료..
결론을 짓자면 오늘 진행한 부분은
1. 2분전, 1분전 종가를 비교해서
2. 1% 이상 상승한 종목들에 대해서
3. 슬랙으로 전송
오늘 이거 하나 하는데 2시간 걸렸다................ 다들 화이팅...
'프로그램' 카테고리의 다른 글
| 비트코인 프로젝트_타점찾기1 (2) | 2024.03.24 |
|---|---|
| 비트코인 프로젝트_1분단위 거래량 조회 (0) | 2024.03.03 |
| 새로운 프로젝트. 비트코인. (0) | 2024.03.03 |
