지금까지 몇 차례에 걸쳐 업비트 API와 파이썬을 이용한 비트코인 자동매매 프로그램을 만드는데 사용하게 될 보조지표를 구하는 방법들을 개별적으로 살펴 보았습니다.
하지만 개별적으로 보조지표들을 조회하는 방법은 불필요한 조회를 여러번 하게 되어 속도에 영향을 미칠 수 있기 때문에 실제 자동매매 프로그램을 작성할 때에는 보조지표를 개별적으로 조회하기 보다는 한 번에 모두 조회하는 것이 좋습니다.
지금까지 알아본 보조지표들과 대부분의 보조지표들은 모두 캔들 데이터를 기반으로 계산을 하게 되는데 4개의 보조지표를 각각 계산하게 된다면 동일한 캔들 데이터를 업비트 API를 4회 호출하는 불필요한 과정을 거치게 되고 업비트 API 호출은 초당 조회 횟수제한이 있기 때문에 속도에 큰 영향을 미칠 수 있습니다.
그래서 보조지표 계산에 공통적으로 사용되는 캔들 정보는 한 번만 조회하고 보조지표들을 구하는 방법을 사용하면 보조지표 계산 시간을 줄일 수 있습니다.
보조지표를 한번에 구하는 방법은 개별적으로 보조지표를 구하는 방법에서 캔들 데이터만 공유하여 사용하도록 되어 있기 때문에 로직상 큰 변화는 없습니다. 각종 지표에 대한 계산 로직은 아래 링크를 통해 확인하실 수 있습니다.
2021.07.27 - [프로젝트/비트코인 자동매매] - RSI 상대강도지수 구하기 - 파이썬 업비트 비트코인 자동매매
RSI 상대강도지수 구하기 - 파이썬 업비트 비트코인 자동매매
주식과는 마찬가지로 코인 시장에서도 차트를 이용한 매매 기법을 사용하실 수 있습니다. 그 중에서도 RSI(상대강도지수)는 상당히 많이 사용되는 보조지표중에 하나 입니다. 앞으로 업비트 API
technfin.tistory.com
2021.07.29 - [프로젝트/비트코인 자동매매] - MFI 자금흐름지수 구하기 - 파이썬 업비트 비트코인 자동매매
MFI 자금흐름지수 구하기 - 파이썬 업비트 비트코인 자동매매
지난 시간에 업비트API와 파이썬을 사용하여 과매수/과매도 구간을 확인하는데 유용한 지표인 RSI 보조지표를 구하는 방법에 대해서 살펴 보았습니다. 오늘은 가격 상승과 하락 변화량에 대한 상
technfin.tistory.com
2021.08.03 - [프로젝트/비트코인 자동매매] - MACD 지표 구하기 - 파이썬 업비트 비트코인 자동매매
MACD 지표 구하기 - 파이썬 업비트 비트코인 자동매매
MACD는 보조 지표중에서 장기 및 단기 이동평균선의 차이를 이용하여 매매 시점을 포착하는데 유용하게 사용되는 지표중에 하나 입니다. 차트를 보면 거의 대부분 단기 이동평균선과 장기 이동
technfin.tistory.com
2021.08.04 - [프로젝트/비트코인 자동매매] - 볼린저밴드 지표 구하기 - 파이썬 업비트 비트코인 자동매매
볼린저밴드 지표 구하기 - 파이썬 업비트 비트코인 자동매매
이번 시간에는 요청이 있었던 업비트API와 파이썬을 이용하여 볼린저밴드 지표값을 구하는 방법에 대해서 살펴 보려고 합니다. 볼린저밴드 외에도 다른 지표들을 계산하는 방법은 아래 링크를
technfin.tistory.com
목차 - 클릭하면 이동합니다.
업비트 API를 이용한 파이썬 코드
import logging
import requests
import time
import smtplib
import jwt
import sys
import uuid
import hashlib
import math
import numpy
import pandas as pd
from datetime import datetime, timedelta
from decimal import Decimal
from urllib.parse import urlencode
# Keys
access_key = '업비트에서 발급받은 Access Key'
secret_key = '업비트에서 발급받은 Secret Key'
server_url = 'https://api.upbit.com'
# -----------------------------------------------------------------------------
# - 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_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_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_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
# -----------------------------------------------------------------------------
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)
# 조회 횟수별 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)
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)
return indicator_data
# ----------------------------------------
# 모든 함수의 공통 부분(Exception 처리)
# ----------------------------------------
except Exception:
raise보조지표를 구하는 로직은 필요한 곳에서 매번 작성하는 것 보다 공통 모듈에 만들고 필요할 때마다 호출하여 사용하는 것이 편리합니다.
위의 코드에서는 upbit.py라는 이름으로 공통 모듈을 만들어서 여러 곳에서 사용되는 반복 로직을 모두 공통 모듈에 작성하고 필요한 곳에서는 단순한 호출만 하고 있는데요.
해당 구조를 만드는 방법은 아래 포스트를 참고하시면 도움이 될 것 같습니다.
2021.06.06 - [프로젝트/비트코인 자동매매] - 비트코인 자동매매 - 프로젝트 구조 만들기
비트코인 자동매매 - 프로젝트 구조 만들기
이번 시간에는 본격적으로 로직을 만들고 살을 붙이기 전에 프로젝트 구조를 만들어 보도록 하겠습니다. 비트코인 자동매매 프로그램을 만드는 것이 얼마나 간단한지는 아래 포
technfin.tistory.com
로직 호출
import os
import sys
import logging
import math
import traceback
# 공통 모듈 Import
sys.path.append(os.path.dirname(os.path.dirname(__file__)))
from lib import upbit as upbit # noqa
# -----------------------------------------------------------------------------
# - Name : main
# - Desc : 메인
# -----------------------------------------------------------------------------
if __name__ == '__main__':
# noinspection PyBroadException
try:
print("***** USAGE ******")
print("[1] 로그레벨(D:DEBUG, E:ERROR, 그외:INFO)")
# 로그레벨(D:DEBUG, E:ERROR, 그외:INFO)
upbit.set_loglevel('I')
# ---------------------------------------------------------------------
# Logic Start!
# ---------------------------------------------------------------------
# 보지지표 조회
indicator_data = upbit.get_indicators('KRW-BTC', '30', '200', 2)
for indicator_data_for in indicator_data:
logging.info(indicator_data_for)
except KeyboardInterrupt:
logging.error("KeyboardInterrupt Exception 발생!")
logging.error(traceback.format_exc())
sys.exit(1)
except Exception:
logging.error("Exception 발생!")
logging.error(traceback.format_exc())
sys.exit(1)indicator_data = upbit.get_indicators('KRW-BTC', '30', '200', 2)
위와 같이 원하는 곳에서 공통모듈을 호출하면 4가지 보지지표(RSI, MFI, MACD, 볼린저밴드) 값을 한 번에 조회할 수 있습니다. 위의 예시는 비트코인(KRW-BTC)의 30분봉을 기준으로 총 2개의 보조지표 데이터들을 구하는 방법 입니다.
호출 결과
모듈을 호출하면 총 4개의 보조지표가 원하는 갯수만큼 리턴되고 type 을 통해 구분하여 추출한 후 원하는 로직에 사용하면 됩니다.
① RSI 보조지표 조회 결과
[{'type': 'RSI', 'DT': '2021-08-10T02:00:00', 'RSI': 61.9977}, {'type': 'RSI', 'DT': '2021-08-10T01:30:00', 'RSI': 62.1116}]
② MFI 보조지표 조회 결과
[{'type': 'MFI', 'DT': '2021-08-10T02:00:00', 'MFI': 67.7791}, {'type': 'MFI', 'DT': '2021-08-10T01:30:00', 'MFI': 70.9995}]
③ MACD 보조지표 조회 결과
[{'type': 'MACD', 'DT': '2021-08-10T02:00:00', 'MACD': 421805.1652, 'SIGNAL': 440410.9886, 'OCL': -18605.8234}, {'type': 'MACD', 'DT': '2021-08-10T01:30:00', 'MACD': 439121.6133, 'SIGNAL': 445062.4445, 'OCL': -5940.8311}]
④ 볼린저밴드 보조지표 조회 결과
[{'type': 'BB', 'DT': '2021-08-10T02:00:00', 'BBH': 53442704.0403, 'BBM': 52095350.0, 'BBL': 50747995.9597}, {'type': 'BB', 'DT': '2021-08-10T01:30:00', 'BBH': 53499999.9834, 'BBM': 51993900.0, 'BBL': 50487800.0166}]
다음 시간에는 보조지표를 이용하여 자동매매를 하는 간단한 예제 프로그램을 작성해 보도록 하겠습니다. 우측 상단 버튼을 눌러 블로그를 구독해 주시면 조금 더 빨리 소식을 받아보실 수 있습니다.
'프로젝트 > 비트코인 자동매매' 카테고리의 다른 글
| 보조지표를 활용한 코인 자동매수 프로그램 - 파이썬 업비트 비트코인 자동매매 (19) | 2021.10.21 |
|---|---|
| 윌리암스 %R 지표 구하기 - 파이썬 업비트 비트코인 자동매매 (5) | 2021.10.07 |
| 코인 자동매매 프로그램 샘플 예제 - 파이썬 업비트 비트코인 자동매매 (17) | 2021.09.07 |
| 볼린저밴드 지표 구하기 - 파이썬 업비트 비트코인 자동매매 (9) | 2021.08.04 |
| MACD 지표 구하기 - 파이썬 업비트 비트코인 자동매매 (4) | 2021.08.03 |
| MFI 자금흐름지수 구하기 - 파이썬 업비트 비트코인 자동매매 (7) | 2021.07.29 |
