반응형
이번 시간부터 본격적으로 코인 자동매매 프로그램을 하나씩 만들어 볼 예정인데요. 오늘은 먼저 보조지표들을 활용한 자동매수 프로그램을 만들어 보도록 하겠습니다.
목차 - 클릭하면 이동합니다.
안내사항
시작하기에 앞서 몇 가지 드리고 싶은 이야기가 있습니다. 예전 부터 드리고 싶었던 이야기들인데 앞으로 본격적으로 프로그램을 작성하여 포스팅 할 예정이라 포스팅을 읽기전에 참고를 부탁 드리고자 말씀 드립니다.
① Tech&Fin 블로그에서 다루는 내용은 파이썬 전문가 혹은 업비트 API 전문가가 알려드리는 내용이 아닙니다. 그렇기 때문에 해당 부분의 전문가 분들에게는 내용이 적절하지 않거나 코딩의 간결도가 깔끔하지 않을 수 있고 로직이 원하는 방향으로 동작하지 않을 수도 있습니다.
② 백테스팅은 과거의 자료를 보관하여야 가능하기 때문에 Tech&Fin에서 다루는 실시간 데이터를 이용한 프로그램들은 충분한 실제 테스트 및 검증이 이루어지지 않은 프로그램 입니다. 그렇기 때문에 프로그램이 동작하다가 오류가 발생하거나 의도치 않은 결과를 초래할 수 있으니 반드시 실제 매매에 적용하기 전에는 적은 금액으로 충분한 테스트를 진행 후 실제 매매에 적용하시는 것을 추천 드립니다.
③ 수익을 보장하는 프로그램이 아닙니다. 절대 무심코 복사하여 실행하는 일이 없도록 주의하셔야 합니다.
④ 매수를 결정하는 로직은 상당히 주관적으로 작성된 부분이오니 핵심 매수 조건 로직을 스스로 이해하여 변경 후 사용하셔야 하며 포스팅 된 프로그램은 전반적인 작동 부분을 참고하는데 활용하시는 것을 추천 드립니다.
마지막으로, 본 프로그램을 사용하여 발생하는 모든 책임은 본인 스스로에게 있음을 인지하시고 항상 실제 매매에 적용하기 전에는 다시 한번 점검하시고 로직을 이해하신 후 참고하여 사용하시기를 부탁 드리겠습니다.
Tech&Fin 블로그를 처음 개설할 때 배우며 소통하고 의견을 수렴하면서 더 안전한 수익을 낼 수 있는 프로그램을 만들어 가고자 마음 먹었습니다. 궁금하신 점이나 의견이 있으시면 댓글로 알려주세요. 최대한 의견을 반영하여 포스팅에 반영될 수 있도록 노력해 보겠습니다.
보조지표를 활용한 자동 매수 프로그램
이번 시간에 다루어 볼 내용은 그 동안 알아보았던 여러가지 보조지표들을 활용해 자동으로 조건에 맞는 코인을 매수하는 프로그램입니다.
활용 보조지표 및 매수 로직
① RSI 보조지표
먼저 RSI 보조지표를 활용해 보도록 하겠습니다. RSI 보조지표에 대해서 과거에 다루었던 포스팅은 아래를 참고 부탁 드립니다.
2021.07.27 - [프로젝트/비트코인 자동매매] - RSI 상대강도지수 구하기 - 파이썬 업비트 비트코인 자동매매
RSI 상대강도지수 구하기 - 파이썬 업비트 비트코인 자동매매
주식과는 마찬가지로 코인 시장에서도 차트를 이용한 매매 기법을 사용하실 수 있습니다. 그 중에서도 RSI(상대강도지수)는 상당히 많이 사용되는 보조지표중에 하나 입니다. 앞으로 업비트 API
technfin.tistory.com
RSI는 일반적으로 30 아래로 떨어지는 경우 과매도 구간으로 판단하여 매수 타이밍으로 판단하고 있습니다.
하지만 모든 지표는 후행 지표로 앞으로를 예상하는 선행 지표가 아니기 때문에 RSI 지표가 30 아래로 떨어졌다고 해서 꼭 반등하리라는 법은 없습니다.
그래서 이번 프로그램에서는 RSI 지표가 30 아래로 떨어진 후 2일 연속 반등하는 경우를 매수 타이밍으로 잡는 로직을 적용해 보려고 합니다.
캔들은 일캔들을 사용하여 판단하고 아래 조건을 만족하는 경우 반등 추세에 접어 들었다고 판단하여 매수 타이밍으로 보는 로직 입니다.
① 2일전 RSI 지표가 30 미만
② 3일전 RSI 지표가 2일전 RSI 지표보다 높은 경우
③ 1일전 RSI 지표가 2일전 RSI 지표보다 높은 경우
④ 현재 RSI 지표가 1일전 RSI 지표보다 높은 경우
즉 RSI 지표가 지속적인 하락으로 2일전에 30 아래로 떨어졌지만 1일 전에 상승으로 전환했고 현재 지표가 1일전보다 상승하는 경우에 RSI 지표가 상승 전환을 했다고 판단하는 로직 입니다.
후행 지표를 감안한 로직이기 때문에 다소 늦게 상승 전환을 감지하는 경우가 많지만 그래도 조금 더 확실한 상승 전환을 판단하기 위해 생각해 본 로직 입니다.
② MFI 보조지표
두 번째로 함께 활용해 볼 지표는 MFI 지표 입니다. MFI 지표는 아래 포스팅을 참고하시면 됩니다.
2021.07.29 - [프로젝트/비트코인 자동매매] - MFI 자금흐름지수 구하기 - 파이썬 업비트 비트코인 자동매매
MFI 자금흐름지수 구하기 - 파이썬 업비트 비트코인 자동매매
지난 시간에 업비트API와 파이썬을 사용하여 과매수/과매도 구간을 확인하는데 유용한 지표인 RSI 보조지표를 구하는 방법에 대해서 살펴 보았습니다. 오늘은 가격 상승과 하락 변화량에 대한 상
technfin.tistory.com
MFI 지표 역시 RSI 지표와 거의 유사한 로직이지만 RSI에서 사용한 30 대신 20을 과매도 구간으로 판단하게 됩니다. RSI만으로 판단했을 때 보다 MFI 까지 더하여 같이 체크하면 조금 더 확실한 과매수 구간을 판단할 수 있을 것 같아 적용하였습니다.
매수 타이밍을 판단하는 로직은 아래와 같으며 RSI와 거의 흡사합니다.
① 2일전 MFI 지표가 20 미만
② 3일전 MFI 지표가 2일전 MFI 지표보다 높은 경우
③ 1일전 MFI 지표가 2일전 MFI 지표보다 높은 경우
④ 현재 MFI 지표가 1일전 MFI 지표보다 높은 경우
③ MACD 오실레이터 보조지표
마지막으로 MACD의 오실레이터를 활용해 보겠습니다. MACD의 오실레이터는 음수 구간에서 양수 구간으로 전환될 때를 골든 크로스로 보지만 해당 구간은 너무 늦은 판단이 될 수 있습니다.
이미 RSI 와 MFI 지표를 이용해 과매도 구간을 판단했기 때문에 MACD의 오실레이터는 아래 로직을 이용해 매수 타이밍을 잡아 보도록 하겠습니다.
① 3일전 오실레이터 < 0
② 2일전 오실레이터 < 0
③ 1일전 오실레이터 < 0
④ 3일전 오실레이터 값이 2일전 오실레이터 값보다 큰 경우
⑤ 1일전 오실레이터 값이 2일전 오실레이터 값보다 큰 경우
⑥ 현재 오실레이터 값이 1일전 오실레이터 값보다 큰 경우
1일, 2일, 3일전 오실레이터의 값이 음의 구간에 있어서 하락하던 중 2일전부터 연속 상승하여 상승 추세로 전환하였다고 판단하는 로직 입니다.
로직 구현
프로그램 구조 만들기
먼저 아래 프로그램 구조 만들기 포스팅을 참고하여 buy_bot 및 module 폴더 아래에 upbit.py의 공통 모듈을 만들어 줍니다.
과거 테스트 모듈들을 작성할 때 폴더를 module이 아닌 lib로 만들어서 헷갈리셨던 분들이 계셔서 다시 원래 계획했던 바와 같이 module 폴더 아래에 공통 모듈을 작성하였습니다.
2021.06.06 - [프로젝트/비트코인 자동매매] - 비트코인 자동매매 - 프로젝트 구조 만들기
비트코인 자동매매 - 프로젝트 구조 만들기
이번 시간에는 본격적으로 로직을 만들고 살을 붙이기 전에 프로젝트 구조를 만들어 보도록 하겠습니다. 비트코인 자동매매 프로그램을 만드는 것이 얼마나 간단한지는 아래 포
technfin.tistory.com
공통모듈 upbit.py 작성하기
그 동안 작성했던 공통 모듈을 모두 합쳤습니다. 중간 중간 수정한 내용들이 있어 이전 포스팅과 조금 맞지 않는 부분들이 있을 수 있으니 참고 부탁 드립니다. 또한 당장 쓰이지 않는 모듈들도 들어 있지만 추후 활용할 수 있는 부분이오니 이 부분도 참고 하시면 좋을 것 같습니다.
import time
import logging
import requests
import jwt
import uuid
import hashlib
import math
import os
import pandas as pd
import numpy
from urllib.parse import urlencode
from decimal import Decimal
# Keys
access_key = '업비트에서 발급받은 Access Key'
secret_key = '업비트에서 발급받은 Secret Key'
server_url = 'https://api.upbit.com'
# 상수 설정
min_order_amt = 5000
# -----------------------------------------------------------------------------
# - Name : set_loglevel
# - Desc : 로그레벨 설정
# - Input
# 1) level : 로그레벨
# 1. D(d) : DEBUG
# 2. E(e) : ERROR
# 3. 그외(기본) : INFO
# - Output
# -----------------------------------------------------------------------------
def set_loglevel(level):
try:
# ---------------------------------------------------------------------
# 로그레벨 : DEBUG
# ---------------------------------------------------------------------
if level.upper() == "D":
logging.basicConfig(
format='[%(asctime)s][%(levelname)s][%(filename)s:%(lineno)d]:%(message)s',
datefmt='%Y/%m/%d %I:%M:%S %p',
level=logging.DEBUG
)
# ---------------------------------------------------------------------
# 로그레벨 : ERROR
# ---------------------------------------------------------------------
elif level.upper() == "E":
logging.basicConfig(
format='[%(asctime)s][%(levelname)s][%(filename)s:%(lineno)d]:%(message)s',
datefmt='%Y/%m/%d %I:%M:%S %p',
level=logging.ERROR
)
# ---------------------------------------------------------------------
# 로그레벨 : INFO
# ---------------------------------------------------------------------
else:
# -----------------------------------------------------------------------------
# 로깅 설정
# 로그레벨(DEBUG, INFO, WARNING, ERROR, CRITICAL)
# -----------------------------------------------------------------------------
logging.basicConfig(
format='[%(asctime)s][%(levelname)s][%(filename)s:%(lineno)d]:%(message)s',
datefmt='%Y/%m/%d %I:%M:%S %p',
level=logging.INFO
)
# ----------------------------------------
# Exception Raise
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : send_request
# - Desc : 리퀘스트 처리
# - Input
# 1) reqType : 요청 타입
# 2) reqUrl : 요청 URL
# 3) reqParam : 요청 파라메타
# 4) reqHeader : 요청 헤더
# - Output
# 4) reponse : 응답 데이터
# -----------------------------------------------------------------------------
def send_request(reqType, reqUrl, reqParam, reqHeader):
try:
# 요청 가능회수 확보를 위해 기다리는 시간(초)
err_sleep_time = 0.3
# 요청에 대한 응답을 받을 때까지 반복 수행
while True:
# 요청 처리
response = requests.request(reqType, reqUrl, params=reqParam, headers=reqHeader)
# 요청 가능회수 추출
if 'Remaining-Req' in response.headers:
hearder_info = response.headers['Remaining-Req']
start_idx = hearder_info.find("sec=")
end_idx = len(hearder_info)
remain_sec = hearder_info[int(start_idx):int(end_idx)].replace('sec=', '')
else:
logging.error("헤더 정보 이상")
logging.error(response.headers)
break
# 요청 가능회수가 3개 미만이면 요청 가능회수 확보를 위해 일정시간 대기
if int(remain_sec) < 3:
logging.debug("요청 가능회수 한도 도달! 남은횟수:" + str(remain_sec))
time.sleep(err_sleep_time)
# 정상 응답
if response.status_code == 200 or response.status_code == 201:
break
# 요청 가능회수 초과인 경우
elif response.status_code == 429:
logging.error("요청 가능회수 초과!:" + str(response.status_code))
time.sleep(err_sleep_time)
# 그 외 오류
else:
logging.error("기타 에러:" + str(response.status_code))
logging.error(response.status_code)
break
# 요청 가능회수 초과 에러 발생시에는 다시 요청
logging.info("[restRequest] 요청 재처리중...")
return response
# ----------------------------------------
# Exception Raise
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : get_items
# - Desc : 전체 종목 리스트 조회
# - Input
# 1) market : 대상 마켓(콤마 구분자:KRW,BTC,USDT)
# 2) except_item : 제외 종목(콤마 구분자:BTC,ETH)
# - Output
# 1) 전체 리스트 : 리스트
# -----------------------------------------------------------------------------
def get_items(market, except_item):
try:
# 조회결과 리턴용
rtn_list = []
# 마켓 데이터
markets = market.split(',')
# 제외 데이터
except_items = except_item.split(',')
url = "https://api.upbit.com/v1/market/all"
querystring = {"isDetails": "false"}
response = send_request("GET", url, querystring, "")
data = response.json()
# 조회 마켓만 추출
for data_for in data:
for market_for in markets:
if data_for['market'].split('-')[0] == market_for:
rtn_list.append(data_for)
# 제외 종목 제거
for rtnlist_for in rtn_list[:]:
for exceptItemFor in except_items:
for marketFor in markets:
if rtnlist_for['market'] == marketFor + '-' + exceptItemFor:
rtn_list.remove(rtnlist_for)
return rtn_list
# ----------------------------------------
# Exception Raise
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : buycoin_mp
# - Desc : 시장가 매수
# - Input
# 1) target_item : 대상종목
# 2) buy_amount : 매수금액
# - Output
# 1) rtn_data : 매수결과
# -----------------------------------------------------------------------------
def buycoin_mp(target_item, buy_amount):
try:
query = {
'market': target_item,
'side': 'bid',
'price': buy_amount,
'ord_type': 'price',
}
query_string = urlencode(query).encode()
m = hashlib.sha512()
m.update(query_string)
query_hash = m.hexdigest()
payload = {
'access_key': access_key,
'nonce': str(uuid.uuid4()),
'query_hash': query_hash,
'query_hash_alg': 'SHA512',
}
jwt_token = jwt.encode(payload, secret_key)
authorize_token = 'Bearer {}'.format(jwt_token)
headers = {"Authorization": authorize_token}
res = send_request("POST", server_url + "/v1/orders", query, headers)
rtn_data = res.json()
logging.info("")
logging.info("----------------------------------------------")
logging.info("시장가 매수 완료!")
logging.info(rtn_data)
logging.info("----------------------------------------------")
return rtn_data
# ----------------------------------------
# Exception Raise
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : buycoin_tg
# - Desc : 지정가 매수
# - Input
# 1) target_item : 대상종목
# 2) buy_amount : 매수금액
# 3) buy_price : 매수가격
# - Output
# 1) rtn_data : 매수요청결과
# -----------------------------------------------------------------------------
def buycoin_tg(target_item, buy_amount, buy_price):
try:
# 매수수량 계산
vol = Decimal(str(buy_amount)) / Decimal(str(buy_price))
query = {
'market': target_item,
'side': 'bid',
'volume': vol,
'price': buy_price,
'ord_type': 'limit',
}
query_string = urlencode(query).encode()
m = hashlib.sha512()
m.update(query_string)
query_hash = m.hexdigest()
payload = {
'access_key': access_key,
'nonce': str(uuid.uuid4()),
'query_hash': query_hash,
'query_hash_alg': 'SHA512',
}
jwt_token = jwt.encode(payload, secret_key)
authorize_token = 'Bearer {}'.format(jwt_token)
headers = {"Authorization": authorize_token}
res = send_request("POST", server_url + "/v1/orders", query, headers)
rtn_data = res.json()
logging.info("")
logging.info("----------------------------------------------")
logging.info("지정가 매수요청 완료!")
logging.info(rtn_data)
logging.info("----------------------------------------------")
return rtn_data
# ----------------------------------------
# Exception Raise
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : sellcoin_mp
# - Desc : 시장가 매도
# - Input
# 1) target_item : 대상종목
# 2) cancel_yn : 기존 주문 취소 여부
# - Output
# 1) rtn_data : 매도결과
# -----------------------------------------------------------------------------
# 시장가 매도
def sellcoin_mp(target_item, cancel_yn):
try:
if cancel_yn == 'Y':
# 기존 주문이 있으면 취소
cancel_order(target_item, "SELL")
# 잔고 조회
cur_balance = get_balance(target_item)
query = {
'market': target_item,
'side': 'ask',
'volume': cur_balance,
'ord_type': 'market',
}
query_string = urlencode(query).encode()
m = hashlib.sha512()
m.update(query_string)
query_hash = m.hexdigest()
payload = {
'access_key': access_key,
'nonce': str(uuid.uuid4()),
'query_hash': query_hash,
'query_hash_alg': 'SHA512',
}
jwt_token = jwt.encode(payload, secret_key)
authorize_token = 'Bearer {}'.format(jwt_token)
headers = {"Authorization": authorize_token}
res = send_request("POST", server_url + "/v1/orders", query, headers)
rtn_data = res.json()
logging.info("")
logging.info("----------------------------------------------")
logging.info("시장가 매도 완료!")
logging.info(rtn_data)
logging.info("----------------------------------------------")
return rtn_data
# ----------------------------------------
# Exception Raise
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : sellcoin_tg
# - Desc : 지정가 매도
# - Input
# 1) target_item : 대상종목
# 2) sell_price : 매도희망금액
# - Output
# 1) rtn_data : 매도결과
# -----------------------------------------------------------------------------
def sellcoin_tg(target_item, sell_price):
try:
# 잔고 조회
cur_balance = get_balance(target_item)
query = {
'market': target_item,
'side': 'ask',
'volume': cur_balance,
'price': sell_price,
'ord_type': 'limit',
}
query_string = urlencode(query).encode()
m = hashlib.sha512()
m.update(query_string)
query_hash = m.hexdigest()
payload = {
'access_key': access_key,
'nonce': str(uuid.uuid4()),
'query_hash': query_hash,
'query_hash_alg': 'SHA512',
}
jwt_token = jwt.encode(payload, secret_key)
authorize_token = 'Bearer {}'.format(jwt_token)
headers = {"Authorization": authorize_token}
res = send_request("POST", server_url + "/v1/orders", query, headers)
rtn_data = res.json()
logging.info("")
logging.info("----------------------------------------------")
logging.info("지정가 매도 설정 완료!")
logging.info(rtn_data)
logging.info("----------------------------------------------")
return rtn_data
# ----------------------------------------
# Exception Raise
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : get_balance
# - Desc : 주문가능 잔고 조회
# - Input
# 1) target_item : 대상 종목
# - Output
# 2) rtn_balance : 주문가능 잔고
# -----------------------------------------------------------------------------
def get_balance(target_item):
try:
# 주문가능 잔고 리턴용
rtn_balance = 0
# 최대 재시도 횟수
max_cnt = 0
# 잔고가 조회 될 때까지 반복
while True:
# 조회 회수 증가
max_cnt = max_cnt + 1
payload = {
'access_key': access_key,
'nonce': str(uuid.uuid4()),
}
jwt_token = jwt.encode(payload, secret_key)
authorize_token = 'Bearer {}'.format(jwt_token)
headers = {"Authorization": authorize_token}
res = send_request("GET", server_url + "/v1/accounts", "", headers)
my_asset = res.json()
# 해당 종목에 대한 잔고 조회
# 잔고는 마켓에 상관없이 전체 잔고가 조회됨
for myasset_for in my_asset:
if myasset_for['currency'] == target_item.split('-')[1]:
rtn_balance = myasset_for['balance']
# 잔고가 0 이상일때까지 반복
if Decimal(str(rtn_balance)) > Decimal(str(0)):
break
# 최대 100회 수행
if max_cnt > 100:
break
logging.info("[주문가능 잔고 리턴용] 요청 재처리중...")
return rtn_balance
# ----------------------------------------
# Exception Raise
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : get_candle
# - Desc : 캔들 조회
# - Input
# 1) target_item : 대상 종목
# 2) tick_kind : 캔들 종류 (1, 3, 5, 10, 15, 30, 60, 240 - 분, D-일, W-주, M-월)
# 3) inq_range : 조회 범위
# - Output
# 1) 캔들 정보 배열
# -----------------------------------------------------------------------------
def get_candle(target_item, tick_kind, inq_range):
try:
# ----------------------------------------
# Tick 별 호출 URL 설정
# ----------------------------------------
# 분붕
if tick_kind == "1" or tick_kind == "3" or tick_kind == "5" or tick_kind == "10" or tick_kind == "15" or tick_kind == "30" or tick_kind == "60" or tick_kind == "240":
target_url = "minutes/" + tick_kind
# 일봉
elif tick_kind == "D":
target_url = "days"
# 주봉
elif tick_kind == "W":
target_url = "weeks"
# 월봉
elif tick_kind == "M":
target_url = "months"
# 잘못된 입력
else:
raise Exception("잘못된 틱 종류:" + str(tick_kind))
logging.debug(target_url)
# ----------------------------------------
# Tick 조회
# ----------------------------------------
querystring = {"market": target_item, "count": inq_range}
res = send_request("GET", server_url + "/v1/candles/" + target_url, querystring, "")
candle_data = res.json()
logging.debug(candle_data)
return candle_data
# ----------------------------------------
# Exception Raise
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : get_targetprice
# - Desc : 호가단위 금액 계산
# - Input
# 1) cal_type : H:호가로, R:비율로
# 2) st_price : 기준가격
# 3) chg_val : 변화단위
# - Output
# 1) rtn_price : 계산된 금액
# -----------------------------------------------------------------------------
def get_targetprice(cal_type, st_price, chg_val):
try:
# 계산된 가격
rtn_price = st_price
# 호가단위로 계산
if cal_type.upper() == "H":
for i in range(0, abs(int(chg_val))):
hoga_val = get_hoga(rtn_price)
if Decimal(str(chg_val)) > 0:
rtn_price = Decimal(str(rtn_price)) + Decimal(str(hoga_val))
elif Decimal(str(chg_val)) < 0:
rtn_price = Decimal(str(rtn_price)) - Decimal(str(hoga_val))
else:
break
# 비율로 계산
elif cal_type.upper() == "R":
while True:
# 호가단위 추출
hoga_val = get_hoga(st_price)
if Decimal(str(chg_val)) > 0:
rtn_price = Decimal(str(rtn_price)) + Decimal(str(hoga_val))
elif Decimal(str(chg_val)) < 0:
rtn_price = Decimal(str(rtn_price)) - Decimal(str(hoga_val))
else:
break
if Decimal(str(chg_val)) > 0:
if Decimal(str(rtn_price)) >= Decimal(str(st_price)) * (
Decimal(str(1)) + (Decimal(str(chg_val))) / Decimal(str(100))):
break
elif Decimal(str(chg_val)) < 0:
if Decimal(str(rtn_price)) <= Decimal(str(st_price)) * (
Decimal(str(1)) + (Decimal(str(chg_val))) / Decimal(str(100))):
break
return rtn_price
# ----------------------------------------
# Exception Raise
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : get_hoga
# - Desc : 호가 금액 계산
# - Input
# 1) cur_price : 현재가격
# - Output
# 1) hoga_val : 호가단위
# -----------------------------------------------------------------------------
def get_hoga(cur_price):
try:
# 호가 단위
if Decimal(str(cur_price)) < 10:
hoga_val = 0.01
elif Decimal(str(cur_price)) < 100:
hoga_val = 0.1
elif Decimal(str(cur_price)) < 1000:
hoga_val = 1
elif Decimal(str(cur_price)) < 10000:
hoga_val = 5
elif Decimal(str(cur_price)) < 100000:
hoga_val = 10
elif Decimal(str(cur_price)) < 500000:
hoga_val = 50
elif Decimal(str(cur_price)) < 1000000:
hoga_val = 100
elif Decimal(str(cur_price)) < 2000000:
hoga_val = 500
else:
hoga_val = 1000
return hoga_val
# ----------------------------------------
# Exception Raise
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : get_krwbal
# - Desc : KRW 잔고 조회
# - Input
# - Output
# 1) KRW 잔고 Dictionary
# 1. krw_balance : KRW 잔고
# 2. fee : 수수료
# 3. available_krw : 매수가능 KRW잔고(수수료를 고려한 금액)
# -----------------------------------------------------------------------------
def get_krwbal():
try:
# 잔고 리턴용
rtn_balance = {}
# 수수료 0.05%(업비트 기준)
fee_rate = 0.05
payload = {
'access_key': access_key,
'nonce': str(uuid.uuid4()),
}
jwt_token = jwt.encode(payload, secret_key)
authorize_token = 'Bearer {}'.format(jwt_token)
headers = {"Authorization": authorize_token}
res = send_request("GET", server_url + "/v1/accounts", "", headers)
data = res.json()
for dataFor in data:
if (dataFor['currency']) == "KRW":
krw_balance = math.floor(Decimal(str(dataFor['balance'])))
# 잔고가 있는 경우만
if Decimal(str(krw_balance)) > Decimal(str(0)):
# 수수료
fee = math.ceil(Decimal(str(krw_balance)) * (Decimal(str(fee_rate)) / Decimal(str(100))))
# 매수가능금액
available_krw = math.floor(Decimal(str(krw_balance)) - Decimal(str(fee)))
else:
# 수수료
fee = 0
# 매수가능금액
available_krw = 0
# 결과 조립
rtn_balance['krw_balance'] = krw_balance
rtn_balance['fee'] = fee
rtn_balance['available_krw'] = available_krw
return rtn_balance
# ----------------------------------------
# Exception Raise
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : get_accounts
# - Desc : 잔고정보 조회
# - Input
# 1) except_yn : KRW 및 소액 제외
# 2) market_code : 마켓코드 추가(매도시 필요)
# - Output
# 1) 잔고 정보
# -----------------------------------------------------------------------------
# 계좌 조회
def get_accounts(except_yn, market_code):
try:
rtn_data = []
# 소액 제외 기준
min_price = 5000
payload = {
'access_key': access_key,
'nonce': str(uuid.uuid4()),
}
jwt_token = jwt.encode(payload, secret_key)
authorize_token = 'Bearer {}'.format(jwt_token)
headers = {"Authorization": authorize_token}
res = send_request("GET", server_url + "/v1/accounts", "", headers)
account_data = res.json()
for account_data_for in account_data:
# KRW 및 소액 제외
if except_yn == "Y" or except_yn == "y":
if account_data_for['currency'] != "KRW" and Decimal(str(account_data_for['avg_buy_price'])) * (
Decimal(str(account_data_for['balance'])) + Decimal(
str(account_data_for['locked']))) >= Decimal(str(min_price)):
rtn_data.append(
{'market': market_code + '-' + account_data_for['currency'],
'balance': account_data_for['balance'],
'locked': account_data_for['locked'],
'avg_buy_price': account_data_for['avg_buy_price'],
'avg_buy_price_modified': account_data_for['avg_buy_price_modified']})
else:
rtn_data.append(
{'market': market_code + '-' + account_data_for['currency'], 'balance': account_data_for['balance'],
'locked': account_data_for['locked'],
'avg_buy_price': account_data_for['avg_buy_price'],
'avg_buy_price_modified': account_data_for['avg_buy_price_modified']})
return rtn_data
# ----------------------------------------
# Exception Raise
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : chg_account_to_comma
# - Desc : 잔고 종목 리스트를 콤마리스트로 변경
# - Input
# 1) account_data : 잔고 데이터
# - Output
# 1) 종목 리스트(콤마 구분자)
# -----------------------------------------------------------------------------
def chg_account_to_comma(account_data):
try:
rtn_data = ""
for account_data_for in account_data:
if rtn_data == '':
rtn_data = rtn_data + account_data_for['market']
else:
rtn_data = rtn_data + ',' + accountDataFor['market']
return rtn_data
# ----------------------------------------
# Exception Raise
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : get_ticker
# - Desc : 현재가 조회
# - Input
# 1) target_itemlist : 대상 종목(콤마 구분자)
# - Output
# 2) 현재가 데이터
# -----------------------------------------------------------------------------
def get_ticker(target_itemlist):
try:
url = "https://api.upbit.com/v1/ticker"
querystring = {"markets": target_itemlist}
response = send_request("GET", url, querystring, "")
rtn_data = response.json()
return rtn_data
# ----------------------------------------
# 모든 함수의 공통 부분(Exception 처리)
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : cancel_order
# - Desc : 미체결 주문 취소
# - Input
# 1) target_item : 대상종목
# 2) side : 매수/매도 구분(BUY/bid:매수, SELL/ask:매도, ALL:전체)
# - Output
# -----------------------------------------------------------------------------
def cancel_order(target_item, side):
try:
# 미체결 주문 조회
order_data = get_order(target_item)
# 매수/매도 구분
for order_data_for in order_data:
if side == "BUY" or side == "buy":
if order_data_for['side'] == "ask":
order_data.remove(order_data_for)
elif side == "SELL" or side == "sell":
if order_data_for['side'] == "bid":
order_data.remove(order_data_for)
# 미체결 주문이 있으면
if len(order_data) > 0:
# 미체결 주문내역 전체 취소
for order_data_for in order_data:
cancel_order_uuid(order_data_for['uuid'])
# ----------------------------------------
# 모든 함수의 공통 부분(Exception 처리)
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : cancel_order_uuid
# - Desc : 미체결 주문 취소 by UUID
# - Input
# 1) order_uuid : 주문 키
# - Output
# 1) 주문 내역 취소
# -----------------------------------------------------------------------------
def cancel_order_uuid(order_uuid):
try:
query = {
'uuid': order_uuid,
}
query_string = urlencode(query).encode()
m = hashlib.sha512()
m.update(query_string)
query_hash = m.hexdigest()
payload = {
'access_key': access_key,
'nonce': str(uuid.uuid4()),
'query_hash': query_hash,
'query_hash_alg': 'SHA512',
}
jwt_token = jwt.encode(payload, secret_key)
authorize_token = 'Bearer {}'.format(jwt_token)
headers = {"Authorization": authorize_token}
res = send_request("DELETE", server_url + "/v1/order", query, headers)
rtn_data = res.json()
return rtn_data
# ----------------------------------------
# 모든 함수의 공통 부분(Exception 처리)
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : get_order
# - Desc : 미체결 주문 조회
# - Input
# 1) target_item : 대상종목
# - Output
# 1) 미체결 주문 내역
# -----------------------------------------------------------------------------
def get_order(target_item):
try:
query = {
'market': target_item,
'state': 'wait',
}
query_string = urlencode(query).encode()
m = hashlib.sha512()
m.update(query_string)
query_hash = m.hexdigest()
payload = {
'access_key': access_key,
'nonce': str(uuid.uuid4()),
'query_hash': query_hash,
'query_hash_alg': 'SHA512',
}
jwt_token = jwt.encode(payload, secret_key)
authorize_token = 'Bearer {}'.format(jwt_token)
headers = {"Authorization": authorize_token}
res = send_request("GET", server_url + "/v1/orders", query, headers)
rtn_data = res.json()
return rtn_data
# ----------------------------------------
# 모든 함수의 공통 부분(Exception 처리)
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : get_rsi
# - Desc : RSI 조회
# - Input
# 1) target_item : 대상 종목
# 2) tick_kind : 캔들 종류 (1, 3, 5, 10, 15, 30, 60, 240 - 분, D-일, W-주, M-월)
# 3) inq_range : 조회 범위
# - Output
# 1) RSI 값
# -----------------------------------------------------------------------------
def get_rsi(target_item, tick_kind, inq_range):
try:
# 캔들 추출
candle_data = get_candle(target_item, tick_kind, inq_range)
df = pd.DataFrame(candle_data)
df = df.reindex(index=df.index[::-1]).reset_index()
df['close'] = df["trade_price"]
# RSI 계산
def rsi(ohlc: pd.DataFrame, period: int = 14):
ohlc["close"] = ohlc["close"]
delta = ohlc["close"].diff()
up, down = delta.copy(), delta.copy()
up[up < 0] = 0
down[down > 0] = 0
_gain = up.ewm(com=(period - 1), min_periods=period).mean()
_loss = down.abs().ewm(com=(period - 1), min_periods=period).mean()
RS = _gain / _loss
return pd.Series(100 - (100 / (1 + RS)), name="RSI")
rsi = round(rsi(df, 14).iloc[-1], 4)
return rsi
# ----------------------------------------
# 모든 함수의 공통 부분(Exception 처리)
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : get_mfi
# - Desc : MFI 조회
# - Input
# 1) target_item : 대상 종목
# 2) tick_kind : 캔들 종류 (1, 3, 5, 10, 15, 30, 60, 240 - 분, D-일, W-주, M-월)
# 3) inq_range : 캔들 조회 범위
# 4) loop_cnt : 지표 반복계산 횟수
# - Output
# 1) MFI 값
# -----------------------------------------------------------------------------
def get_mfi(target_item, tick_kind, inq_range, loop_cnt):
try:
# 캔들 데이터 조회용
candle_datas = []
# MFI 데이터 리턴용
mfi_list = []
# 캔들 추출
candle_data = get_candle(target_item, tick_kind, inq_range)
# 조회 횟수별 candle 데이터 조합
for i in range(0, int(loop_cnt)):
candle_datas.append(candle_data[i:int(len(candle_data))])
# 캔들 데이터만큼 수행
for candle_data_for in candle_datas:
df = pd.DataFrame(candle_data_for)
dfDt = df['candle_date_time_kst'].iloc[::-1]
df['typical_price'] = (df['trade_price'] + df['high_price'] + df['low_price']) / 3
df['money_flow'] = df['typical_price'] * df['candle_acc_trade_volume']
positive_mf = 0
negative_mf = 0
for i in range(0, 14):
if df["typical_price"][i] > df["typical_price"][i + 1]:
positive_mf = positive_mf + df["money_flow"][i]
elif df["typical_price"][i] < df["typical_price"][i + 1]:
negative_mf = negative_mf + df["money_flow"][i]
if negative_mf > 0:
mfi = 100 - (100 / (1 + (positive_mf / negative_mf)))
else:
mfi = 100 - (100 / (1 + (positive_mf)))
mfi_list.append({"type": "MFI", "DT": dfDt[0], "MFI": round(mfi, 4)})
return mfi_list
# ----------------------------------------
# 모든 함수의 공통 부분(Exception 처리)
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : get_macd
# - Desc : MACD 조회
# - Input
# 1) target_item : 대상 종목
# 2) tick_kind : 캔들 종류 (1, 3, 5, 10, 15, 30, 60, 240 - 분, D-일, W-주, M-월)
# 3) inq_range : 캔들 조회 범위
# 4) loop_cnt : 지표 반복계산 횟수
# - Output
# 1) MACD 값
# -----------------------------------------------------------------------------
def get_macd(target_item, tick_kind, inq_range, loop_cnt):
try:
# 캔들 데이터 조회용
candle_datas = []
# MACD 데이터 리턴용
macd_list = []
# 캔들 추출
candle_data = get_candle(target_item, tick_kind, inq_range)
# 조회 횟수별 candle 데이터 조합
for i in range(0, int(loop_cnt)):
candle_datas.append(candle_data[i:int(len(candle_data))])
df = pd.DataFrame(candle_datas[0])
df = df.iloc[::-1]
df = df['trade_price']
# MACD 계산
exp1 = df.ewm(span=12, adjust=False).mean()
exp2 = df.ewm(span=26, adjust=False).mean()
macd = exp1 - exp2
exp3 = macd.ewm(span=9, adjust=False).mean()
for i in range(0, int(loop_cnt)):
macd_list.append(
{"type": "MACD", "DT": candle_datas[0][i]['candle_date_time_kst'], "MACD": round(macd[i], 4),
"SIGNAL": round(exp3[i], 4),
"OCL": round(macd[i] - exp3[i], 4)})
return macd_list
# ----------------------------------------
# 모든 함수의 공통 부분(Exception 처리)
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : get_bb
# - Desc : 볼린저밴드 조회
# - Input
# 1) target_item : 대상 종목
# 2) tick_kind : 캔들 종류 (1, 3, 5, 10, 15, 30, 60, 240 - 분, D-일, W-주, M-월)
# 3) inq_range : 캔들 조회 범위
# 4) loop_cnt : 지표 반복계산 횟수
# - Output
# 1) 볼린저 밴드 값
# -----------------------------------------------------------------------------
def get_bb(target_item, tick_kind, inq_range, loop_cnt):
try:
# 캔들 데이터 조회용
candle_datas = []
# 볼린저밴드 데이터 리턴용
bb_list = []
# 캔들 추출
candle_data = get_candle(target_item, tick_kind, inq_range)
# 조회 횟수별 candle 데이터 조합
for i in range(0, int(loop_cnt)):
candle_datas.append(candle_data[i:int(len(candle_data))])
# 캔들 데이터만큼 수행
for candle_data_for in candle_datas:
df = pd.DataFrame(candle_data_for)
dfDt = df['candle_date_time_kst'].iloc[::-1]
df = df['trade_price'].iloc[::-1]
# 표준편차(곱)
unit = 2
band1 = unit * numpy.std(df[len(df) - 20:len(df)])
bb_center = numpy.mean(df[len(df) - 20:len(df)])
band_high = bb_center + band1
band_low = bb_center - band1
bb_list.append({"type": "BB", "DT": dfDt[0], "BBH": round(band_high, 4), "BBM": round(bb_center, 4),
"BBL": round(band_low, 4)})
return bb_list
# ----------------------------------------
# 모든 함수의 공통 부분(Exception 처리)
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : get_williams
# - Desc : 윌리암스 %R 조회
# - Input
# 1) target_item : 대상 종목
# 2) tick_kind : 캔들 종류 (1, 3, 5, 10, 15, 30, 60, 240 - 분, D-일, W-주, M-월)
# 3) inq_range : 캔들 조회 범위
# 4) loop_cnt : 지표 반복계산 횟수
# - Output
# 1) 윌리암스 %R 값
# -----------------------------------------------------------------------------
def get_williamsR(target_item, tick_kind, inq_range, loop_cnt):
try:
# 캔들 데이터 조회용
candle_datas = []
# 윌리암스R 데이터 리턴용
williams_list = []
# 캔들 추출
candle_data = get_candle(target_item, tick_kind, inq_range)
# 조회 횟수별 candle 데이터 조합
for i in range(0, int(loop_cnt)):
candle_datas.append(candle_data[i:int(len(candle_data))])
# 캔들 데이터만큼 수행
for candle_data_for in candle_datas:
df = pd.DataFrame(candle_data_for)
dfDt = df['candle_date_time_kst'].iloc[::-1]
df = df.iloc[:14]
# 계산식
# %R = (Highest High - Close)/(Highest High - Lowest Low) * -100
hh = numpy.max(df['high_price'])
ll = numpy.min(df['low_price'])
cp = df['trade_price'][0]
w = (hh - cp) / (hh - ll) * -100
williams_list.append(
{"type": "WILLIAMS", "DT": dfDt[0], "HH": round(hh, 4), "LL": round(ll, 4), "CP": round(cp, 4),
"W": round(w, 4)})
return williams_list
# ----------------------------------------
# 모든 함수의 공통 부분(Exception 처리)
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : get_rsi
# - Desc : RSI 조회
# - Input
# 1) candle_data : 캔들 정보
# - Output
# 1) RSI 값
# -----------------------------------------------------------------------------
def get_rsi(candle_datas):
try:
# RSI 데이터 리턴용
rsi_data = []
# 캔들 데이터만큼 수행
for candle_data_for in candle_datas:
df = pd.DataFrame(candle_data_for)
dfDt = df['candle_date_time_kst'].iloc[::-1]
df = df.reindex(index=df.index[::-1]).reset_index()
df['close'] = df["trade_price"]
# RSI 계산
def rsi(ohlc: pd.DataFrame, period: int = 14):
ohlc["close"] = ohlc["close"]
delta = ohlc["close"].diff()
up, down = delta.copy(), delta.copy()
up[up < 0] = 0
down[down > 0] = 0
_gain = up.ewm(com=(period - 1), min_periods=period).mean()
_loss = down.abs().ewm(com=(period - 1), min_periods=period).mean()
RS = _gain / _loss
return pd.Series(100 - (100 / (1 + RS)), name="RSI")
rsi = round(rsi(df, 14).iloc[-1], 4)
rsi_data.append({"type": "RSI", "DT": dfDt[0], "RSI": rsi})
return rsi_data
# ----------------------------------------
# 모든 함수의 공통 부분(Exception 처리)
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : get_mfi
# - Desc : MFI 조회
# - Input
# 1) candle_datas : 캔들 정보
# - Output
# 1) MFI 값
# -----------------------------------------------------------------------------
def get_mfi(candle_datas):
try:
# MFI 데이터 리턴용
mfi_list = []
# 캔들 데이터만큼 수행
for candle_data_for in candle_datas:
df = pd.DataFrame(candle_data_for)
dfDt = df['candle_date_time_kst'].iloc[::-1]
df['typical_price'] = (df['trade_price'] + df['high_price'] + df['low_price']) / 3
df['money_flow'] = df['typical_price'] * df['candle_acc_trade_volume']
positive_mf = 0
negative_mf = 0
for i in range(0, 14):
if df["typical_price"][i] > df["typical_price"][i + 1]:
positive_mf = positive_mf + df["money_flow"][i]
elif df["typical_price"][i] < df["typical_price"][i + 1]:
negative_mf = negative_mf + df["money_flow"][i]
if negative_mf > 0:
mfi = 100 - (100 / (1 + (positive_mf / negative_mf)))
else:
mfi = 100 - (100 / (1 + (positive_mf)))
mfi_list.append({"type": "MFI", "DT": dfDt[0], "MFI": round(mfi, 4)})
return mfi_list
# ----------------------------------------
# 모든 함수의 공통 부분(Exception 처리)
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : get_macd
# - Desc : MACD 조회
# - Input
# 1) candle_datas : 캔들 정보
# 2) loop_cnt : 반복 횟수
# - Output
# 1) MACD 값
# -----------------------------------------------------------------------------
def get_macd(candle_datas, loop_cnt):
try:
# MACD 데이터 리턴용
macd_list = []
df = pd.DataFrame(candle_datas[0])
df = df.iloc[::-1]
df = df['trade_price']
# MACD 계산
exp1 = df.ewm(span=12, adjust=False).mean()
exp2 = df.ewm(span=26, adjust=False).mean()
macd = exp1 - exp2
exp3 = macd.ewm(span=9, adjust=False).mean()
for i in range(0, int(loop_cnt)):
macd_list.append(
{"type": "MACD", "DT": candle_datas[0][i]['candle_date_time_kst'], "MACD": round(macd[i], 4),
"SIGNAL": round(exp3[i], 4),
"OCL": round(macd[i] - exp3[i], 4)})
return macd_list
# ----------------------------------------
# 모든 함수의 공통 부분(Exception 처리)
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : get_bb
# - Desc : 볼린저밴드 조회
# - Input
# 1) candle_datas : 캔들 정보
# - Output
# 1) 볼린저 밴드 값
# -----------------------------------------------------------------------------
def get_bb(candle_datas):
try:
# 볼린저밴드 데이터 리턴용
bb_list = []
# 캔들 데이터만큼 수행
for candle_data_for in candle_datas:
df = pd.DataFrame(candle_data_for)
dfDt = df['candle_date_time_kst'].iloc[::-1]
df = df['trade_price'].iloc[::-1]
# 표준편차(곱)
unit = 2
band1 = unit * numpy.std(df[len(df) - 20:len(df)])
bb_center = numpy.mean(df[len(df) - 20:len(df)])
band_high = bb_center + band1
band_low = bb_center - band1
bb_list.append({"type": "BB", "DT": dfDt[0], "BBH": round(band_high, 4), "BBM": round(bb_center, 4),
"BBL": round(band_low, 4)})
return bb_list
# ----------------------------------------
# 모든 함수의 공통 부분(Exception 처리)
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : get_williams
# - Desc : 윌리암스 %R 조회
# - Input
# 1) candle_datas : 캔들 정보
# - Output
# 1) 윌리암스 %R 값
# -----------------------------------------------------------------------------
def get_williams(candle_datas):
try:
# 윌리암스R 데이터 리턴용
williams_list = []
# 캔들 데이터만큼 수행
for candle_data_for in candle_datas:
df = pd.DataFrame(candle_data_for)
dfDt = df['candle_date_time_kst'].iloc[::-1]
df = df.iloc[:14]
# 계산식
# %R = (Highest High - Close)/(Highest High - Lowest Low) * -100
hh = numpy.max(df['high_price'])
ll = numpy.min(df['low_price'])
cp = df['trade_price'][0]
w = (hh - cp) / (hh - ll) * -100
williams_list.append(
{"type": "WILLIAMS", "DT": dfDt[0], "HH": round(hh, 4), "LL": round(ll, 4), "CP": round(cp, 4),
"W": round(w, 4)})
return williams_list
# ----------------------------------------
# 모든 함수의 공통 부분(Exception 처리)
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : get_indicators
# - Desc : 보조지표 조회
# - Input
# 1) target_item : 대상 종목
# 2) tick_kind : 캔들 종류 (1, 3, 5, 10, 15, 30, 60, 240 - 분, D-일, W-주, M-월)
# 3) inq_range : 캔들 조회 범위
# 4) loop_cnt : 지표 반복계산 횟수
# - Output
# 1) RSI
# 2) MFI
# 3) MACD
# 4) BB
# 5) WILLIAMS %R
# -----------------------------------------------------------------------------
def get_indicators(target_item, tick_kind, inq_range, loop_cnt):
try:
# 보조지표 리턴용
indicator_data = []
# 캔들 데이터 조회용
candle_datas = []
# 캔들 추출
candle_data = get_candle(target_item, tick_kind, inq_range)
if len(candle_data) >= 30:
# 조회 횟수별 candle 데이터 조합
for i in range(0, int(loop_cnt)):
candle_datas.append(candle_data[i:int(len(candle_data))])
# RSI 정보 조회
rsi_data = get_rsi(candle_datas)
# MFI 정보 조회
mfi_data = get_mfi(candle_datas)
# MACD 정보 조회
macd_data = get_macd(candle_datas, loop_cnt)
# BB 정보 조회
bb_data = get_bb(candle_datas)
# WILLIAMS %R 조회
williams_data = get_williams(candle_datas)
if len(rsi_data) > 0:
indicator_data.append(rsi_data)
if len(mfi_data) > 0:
indicator_data.append(mfi_data)
if len(macd_data) > 0:
indicator_data.append(macd_data)
if len(bb_data) > 0:
indicator_data.append(bb_data)
if len(williams_data) > 0:
indicator_data.append(williams_data)
return indicator_data
# ----------------------------------------
# 모든 함수의 공통 부분(Exception 처리)
# ----------------------------------------
except Exception:
raise반응형
공통 모듈에서 아래 Key 부분은 본인 업비트 계정에서 발급받은 Key를 넣어야 동작합니다.
# Keys
access_key = '업비트에서 발급받은 Access Key'
secret_key = '업비트에서 발급받은 Secret Key'
server_url = 'https://api.upbit.com'
자동 매수 로직 buy_bot.py 작성하기
import time
import os
import sys
import logging
import traceback
from decimal import Decimal
# 공통 모듈 Import
sys.path.append(os.path.dirname(os.path.dirname(__file__)))
from module import upbit
# -----------------------------------------------------------------------------
# - Name : start_buytrade
# - Desc : 매수 로직
# - Input
# 1) buy_amt : 매수금액
# 2) except_items : 매수 제외종목
# -----------------------------------------------------------------------------
def start_buytrade(buy_amt, except_items):
try:
#----------------------------------------------------------------------
# 반복 수행
#----------------------------------------------------------------------
while True:
logging.info("*********************************************************")
logging.info("1. 로그레벨 : " + str(log_level))
logging.info("2. 매수금액 : " + str(buy_amt))
logging.info("3. 매수 제외종목 : " + str(except_items))
logging.info("*********************************************************")
#------------------------------------------------------------------
# 매수 제외종목을 제외한 종목 리스트 추출
#------------------------------------------------------------------
target_items = upbit.get_items('KRW', except_items)
for target_item in target_items:
rsi_val = False
mfi_val = False
ocl_val = False
logging.info('체크중....[' + str(target_item['market']) + ']')
#--------------------------------------------------------------
# 종목별 보조지표를 조회
# 1. 조회 기준 : 일캔들, 최근 5개 지표 조회
#--------------------------------------------------------------
indicators_data = upbit.get_indicators(target_item['market'], 'D', 200, 5)
#--------------------------------------------------------------
# 최근 30일 이내에 신규 상장하여 보조 지표를 구하기 어려운 건은 제외
#--------------------------------------------------------------
if len(indicators_data) < 5:
logging.info('캔들 데이터 부족으로 매수 대상에서 제외....[' + str(target_item['market']) + ']')
continue
#--------------------------------------------------------------
# 매수 로직
# 1. RSI : 2일전 < 30미만, 3일전 > 2일전, 1일전 > 2일전, 현재 > 1일전
# 2. MFI : 2일전 < 20미만, 3일전 > 2일전, 1일전 > 2일전, 현재 > 1일전
# 3. MACD(OCL) : 3일전 < 0, 2일전 < 0, 1일전 < 0, 3일전 > 2일전, 1일전 > 2일전, 현재 > 1일전
#--------------------------------------------------------------
#--------------------------------------------------------------
# RSI : 2일전 < 30미만, 3일전 > 2일전, 1일전 > 2일전, 현재 > 1일전
# indicators_data[0][0]['RSI'] : 현재
# indicators_data[0][1]['RSI'] : 1일전
# indicators_data[0][2]['RSI'] : 2일전
# indicators_data[0][3]['RSI'] : 3일전
#--------------------------------------------------------------
if (Decimal(str(indicators_data[0][0]['RSI'])) > Decimal(str(indicators_data[0][1]['RSI']))
and Decimal(str(indicators_data[0][1]['RSI'])) > Decimal(str(indicators_data[0][2]['RSI']))
and Decimal(str(indicators_data[0][3]['RSI'])) > Decimal(str(indicators_data[0][2]['RSI']))
and Decimal(str(indicators_data[0][2]['RSI'])) < Decimal(str(30))):
rsi_val = True
#--------------------------------------------------------------
# MFI : 2일전 < 20미만, 3일전 > 2일전, 1일전 > 2일전, 현재 > 1일전
# indicators_data[1][0]['MFI'] : 현재
# indicators_data[1][1]['MFI'] : 1일전
# indicators_data[1][2]['MFI'] : 2일전
# indicators_data[1][3]['MFI'] : 3일전
#--------------------------------------------------------------
if (Decimal(str(indicators_data[1][0]['MFI'])) > Decimal(str(indicators_data[1][1]['MFI']))
and Decimal(str(indicators_data[1][1]['MFI'])) > Decimal(str(indicators_data[1][2]['MFI']))
and Decimal(str(indicators_data[1][3]['MFI'])) > Decimal(str(indicators_data[1][2]['MFI']))
and Decimal(str(indicators_data[1][2]['MFI'])) < Decimal(str(20))):
mfi_val = True
#--------------------------------------------------------------
# MACD(OCL) : 3일전 < 0, 2일전 < 0, 1일전 < 0, 3일전 > 2일전, 1일전 > 2일전, 현재 > 1일전
# indicators_data[2][0]['OCL'] : 현재
# indicators_data[2][1]['OCL'] : 1일전
# indicators_data[2][2]['OCL'] : 2일전
# indicators_data[2][3]['OCL'] : 3일전
#--------------------------------------------------------------
if (Decimal(str(indicators_data[2][0]['OCL'])) > Decimal(str(indicators_data[2][1]['OCL']))
and Decimal(str(indicators_data[2][1]['OCL'])) > Decimal(str(indicators_data[2][2]['OCL']))
and Decimal(str(indicators_data[2][3]['OCL'])) > Decimal(str(indicators_data[2][2]['OCL']))
and Decimal(str(indicators_data[2][1]['OCL'])) < Decimal(str(0))
and Decimal(str(indicators_data[2][2]['OCL'])) < Decimal(str(0))
and Decimal(str(indicators_data[2][3]['OCL'])) < Decimal(str(0))):
ocl_val = True
#--------------------------------------------------------------
# 매수대상 발견
#--------------------------------------------------------------
if rsi_val and mfi_val and ocl_val:
logging.info('매수대상 발견....[' + str(target_item['market']) + ']')
logging.info(indicators_data[0])
logging.info(indicators_data[1])
logging.info(indicators_data[2])
# ------------------------------------------------------------------
# 매수금액 설정
# 1. M : 수수료를 제외한 최대 가능 KRW 금액만큼 매수
# 2. 금액 : 입력한 금액만큼 매수
# ------------------------------------------------------------------
available_amt = upbit.get_krwbal()['available_krw']
if buy_amt == 'M':
buy_amt = available_amt
# ------------------------------------------------------------------
# 입력 금액이 주문 가능금액보다 작으면 종료
# ------------------------------------------------------------------
if Decimal(str(available_amt)) < Decimal(str(buy_amt)):
logging.info('주문 가능금액[' + str(available_amt) + ']이 입력한 주문금액[' + str(buy_amt) + '] 보다 작습니다.')
continue
# ------------------------------------------------------------------
# 최소 주문 금액(업비트 기준 5000원) 이상일 때만 매수로직 수행
# ------------------------------------------------------------------
if Decimal(str(buy_amt)) < Decimal(str(upbit.min_order_amt)):
logging.info('주문금액[' + str(buy_amt) + ']이 최소 주문금액[' + str(upbit.min_order_amt) + '] 보다 작습니다.')
continue
# ------------------------------------------------------------------
# 시장가 매수
# 실제 매수 로직은 안전을 위해 주석처리 하였습니다.
# 실제 매매를 원하시면 테스트를 충분히 거친 후 주석을 해제하시면 됩니다.
# ------------------------------------------------------------------
logging.info('시장가 매수 시작! [' + str(target_item['market']) + ']')
#rtn_buycoin_mp = upbit.buycoin_mp(target_item['market'], buy_amt)
logging.info('시장가 매수 종료! [' + str(target_item['market']) + ']')
#logging.info(rtn_buycoin_mp)
# ------------------------------------------------------------------
# 매수 완료 종목은 매수 대상에서 제외
# ------------------------------------------------------------------
if except_items.strip():
except_items = except_items + ',' + target_item['market'].split('-')[1]
else:
except_items = target_item['market'].split('-')[1]
# ---------------------------------------
# 모든 함수의 공통 부분(Exception 처리)
# ----------------------------------------
except Exception:
raise
# -----------------------------------------------------------------------------
# - Name : main
# - Desc : 메인
# -----------------------------------------------------------------------------
if __name__ == '__main__':
# noinspection PyBroadException
try:
# ---------------------------------------------------------------------
# 입력 받을 변수
#
# 1. 로그레벨
# 1) 레벨 값 : D:DEBUG, E:ERROR, 그 외:INFO
#
# 2. 매수금액
# 1) M : 수수료를 제외한 최대 가능 금액으로 매수
# 2) 금액 : 입력한 금액만 매수(수수료 포함)
#
# 3. 매수 제외종목
# 1) 종목코드(콤마구분자) : BTC,ETH
# ---------------------------------------------------------------------
# 1. 로그레벨
log_level = input("로그레벨(D:DEBUG, E:ERROR, 그 외:INFO) : ").upper()
buy_amt = input("매수금액(M:최대, 10000:1만원) : ").upper()
except_items = input("매수 제외종목(종목코드, ex:BTC,ETH) : ").upper()
upbit.set_loglevel(log_level)
logging.info("*********************************************************")
logging.info("1. 로그레벨 : " + str(log_level))
logging.info("2. 매수금액 : " + str(buy_amt))
logging.info("3. 매수 제외종목 : " + str(except_items))
logging.info("*********************************************************")
# 매수 로직 시작
start_buytrade(buy_amt, except_items)
except KeyboardInterrupt:
logging.error("KeyboardInterrupt Exception 발생!")
logging.error(traceback.format_exc())
sys.exit(-100)
except Exception:
logging.error("Exception 발생!")
logging.error(traceback.format_exc())
sys.exit(-200)프로그램 실행 시 로그레벨 및 원하는 매수금액 또는 최대 가능 금액(M)을 입력 받도록 했으며 매수를 원하지 않는 종목 역시 입력 받아 제외 할 수 있도록 구현하였습니다.
기본적으로 무한 반복하며 RSI, MFI, MACD 오실레이터의 지표를 추출하여 원하는 조건에 맞는 경우 원하는 금액만큼 시장가 매수하는 로직 입니다.
업비트 기준 최소 주문 금액이 종목별로 1,000원에서 5,000원까지 다양하기 때문에 최소 주문 금액이 5,000원 미만인 경우 수행하지 않도록 로직을 구현하였습니다.
무작정 실행하는 분들이 계실까봐 안전을 위해 실제 시장가 매수 로직은 주석처리 하였으니 실제 적용시에는 주석을 풀고 사용하여야 합니다.
마치며
매수 타이밍을 결정하는 로직이 상당히 보수적인 편이라 프로그램을 돌려도 잦은 매수가 발생하지는 않을 수 있습니다. 전반적으로 급락하였다가 반등하는 경우에 매수가 될 확률이 높습니다. 더 잦은 매매를 원하는 경우에는 로직을 더 완화하여야 하지만 잘 판단하여 로직을 수정해야 할 것입니다.
오늘 만든 프로그램이라 실제로 테스트를 오랜기간 많이 진행해 보지 못했습니다. 당장은 로직을 이해하는 정도로 참고하시는게 좋을 것 같으며 앞으로 발견될 수 있는 오류들을 수정하면서 천천히 완성도를 높여 보도록 하겠습니다. 다음에는 매도 프로그램을 작성해 보도록 하겠습니다.
블로그를 구독하시면 소식을 조금 더 빨리 받아 보실 수 있습니다. 궁금한 점은 댓글로 달아 주시면 최대한 반영해 볼 수 있도록 하겠습니다.
반응형
'프로젝트 > 비트코인 자동매매' 카테고리의 다른 글
| 잔고정보 조회하기(오류 수정) - 업비트 파이썬 비트코인 자동매매 (7) | 2021.10.26 |
|---|---|
| 보조지표를 활용한 코인 자동매도 프로그램 파이썬 업비트 비트코인 자동매매 (5) | 2021.10.26 |
| 트레일링 스탑을 활용한 코인 자동매도 프로그램 - 파이썬 업비트 비트코인 자동매매 (18) | 2021.10.23 |
| 윌리암스 %R 지표 구하기 - 파이썬 업비트 비트코인 자동매매 (5) | 2021.10.07 |
| 코인 자동매매 프로그램 샘플 예제 - 파이썬 업비트 비트코인 자동매매 (17) | 2021.09.07 |
| 보조 지표 한번에 구하기 - 파이썬 업비트 비트코인 자동매매 (11) | 2021.08.10 |
