비트코인 자동매매 - 프로젝트 구조 만들기

이번 시간에는 본격적으로 로직을 만들고 살을 붙이기 전에 프로젝트 구조를 만들어 보도록 하겠습니다.

 

비트코인 자동매매 프로그램을 만드는 것이 얼마나 간단한지는 아래 포스트를 참고하시면 확인하실 수 있습니다.

 

2021.06.06 - [프로젝트/비트코인 자동매매] - 비트코인 자동매매 - 시세 종목 조회 하기

비트코인 자동매매 - 시세 종목 조회 하기

 

 

목차 - 클릭하면 이동합니다.

 

프로젝트 구조

앞으로 진행할 프로젝트에서는 아래 프로그램들을 만들어 볼 예정입니다.

 

① 자동 매수 프로그램
② 자동 매도 프로그램
③ 모니터링 프로그램

 

그런데 세 가지 프로그램을 만들면서 공통적으로 사용해야 하는 로직들이 있을텐데 그런 로직을 각 프로그램마다 작성하게 되면 같은 코드를 세 가지 프로그램에 모두 작성해야 하고 수정이 필요한 경우 전부 수정해야 하는 등 여러가지 불편함이 따라오게 됩니다.

 

그래서 공통으로 사용할 수 있는 로직은 한 곳에 모아두고 세 가지 프로그램에서는 참고해서 사용하도록 만들 예정인데요. 이것이 바로 모듈이라는 개념 입니다.

 

복잡하게 생각하실 필요는 없고 실제로도 복잡하지 않게 아래와 같이 아주 간단한 구조로 만들 예정입니다.

 

메인 프로젝트 아래 3가지 프로그램을 만들고 module 폴더 아래에 upbit.py라는 공통 모듈을 만들어서 3가지 프로그램에서 import하여 사용하는 간단한 구조 입니다.

 

프로젝트 구조 만들기

프로젝트 만들기

① trade_bot 으로 메인 프로젝트 이름을 정합니다.
② welcome 스크립트는 만들지 않겠습니다.
③ Create 버튼을 클릭하여 프로젝트를 생성합니다.

 

모듈 폴더 만들기

① 프로젝트에서 마우스를 우클릭한 후 ② New > ③ Directory를 차례로 클릭 합니다.

 

module라는 폴더를 만듭니다.

 

모듈 파일 만들기

① 생성한 module 폴더에서 마우스를 우클릭한 후 ② New > ③ Python File을 클릭합니다.

 

① 모듈 이름 : upbit 라고 만들도록 하겠습니다.
② Python File을 클릭하여 생성합니다.

 

프로그램 파일 만들기

이번에는 ① trade_bot 메인 프로젝트에서 마우스를 우클릭한 후 ② New > ③ Python File을 클릭합니다.

 

① 프로그램 이름 : buy_bot이라는 매수 프로그램을 만들도록 하겠습니다.
② Python File을 클릭하여 생성합니다.

 

공통 모듈 작성하기

로그 모듈

# -----------------------------------------------------------------------------
# - 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

로그는 프로그램이 원하는 대로 작동하는지 확인하거나 에러 발생시 내용을 확인할 수 있는 아주 중요한 부분이며 logging 패키지를 import하면 간단하게 사용하실 수 있습니다.

 

로그는 일반적으로 레벨을 설정하여 사용하는데 우리는 총 세가지 레벨을 이용하려고 합니다.

 

① DEBUG : 상세한 로그를 보고 싶은 경우
② INFO : 정보성 로그를 보고 싶은 경우
③ ERROR : 에러 로그를 보고 싶은 경우

 

앞으로 프로그램을 작성하면서 로그를 출력해야 하는 부분에 레벨을 적절하게 이용하면 편리합니다.

 

개발 또는 에러를 해결하기 위해서 상세한 로그가 필요한 경우에는 debug를 이용해 로그를 출력하면 되고 정보성의 경우 info 에러의 경우 error로 출력하도록 프로그래밍 하면 됩니다.

 

요청 모듈

업비트를 비롯해 대부분의 API를 제공하는 업체에서는 과도한 요청으로 서버에 부하가 발생하는 것을 방지하기 위해 초당 요청 제한 회수를 적용하고 있습니다.

 

요청 수를 초과하게 되면 에러가 발생하지만 패널티가 부과되지는 않습니다. 패널티가 부과되지 않더라도 요청수를 초과하면 에러가 발생하기 때문에 가능한 최대한의 요청 횟수를 활용하기 위해 공통 모듈에 만들고 관리하는 것이 좋습니다.

 

# -----------------------------------------------------------------------------
# - 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 = 1

        # 요청에 대한 응답을 받을 때까지 반복 수행
        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

            # 요청 가능회수가 4개 미만이면 요청 가능회수 확보를 위해 일정시간 대기
            if int(remain_sec) < 4:
                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

잔여 요청 횟수를 매번 체크하여 3회 이하(4회 미만)로 남은 경우에는 잠시 대기하는 로직을 적용하여 요청 회수 초과를 방지하고 있습니다.

 

그래도 초과하는 경우가 발생할 수 있는데 그런 경우라도 에러로 프로그램이 중단되지 않고 다시 시도하도록 되어 있습니다.

 

종목 조회 모듈

업비트에서 거래되는 종목 코드를 가져오는 부분은 매수 로직을 비롯하여 여러가지 다른 로직에서 많이 사용되기 때문에 공통 모듈에 적용하는 것이 좋습니다.

 

업비트에서 제공하는 샘플 코드를 응용해서 조금 더 유연하게 움직이는 로직을 만들어 보겠습니다.

 

# -----------------------------------------------------------------------------
# - 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

① 입력 변수로 마켓 코드를 받아 원하는 마켓의 종목만 조회할 수 있습니다.
② 제외하고 싶은 종목을 입력하면 해당 종목은 제외 후 결과를 출력합니다.

 

전체 코드

import time
import logging
import requests


# -----------------------------------------------------------------------------
# - 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 = 1

        # 요청에 대한 응답을 받을 때까지 반복 수행
        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

            # 요청 가능회수가 4개 미만이면 요청 가능회수 확보를 위해 일정시간 대기
            if int(remain_sec) < 4:
                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

 

프로그램 작성하기

공통 모듈 참조하기

위에서 만든 공통 모듈을 참조하여 매수 프로그램에 종목 조회 기능을 추가해 보겠습니다.

 

from module import upbit

위와 같이 직접 만든 모듈을 import 하여 사용할 수 있습니다.

 

로그 레벨 설정

upbit.set_loglevel('D')

프로그램을 시작하기 전에 로그레벨을 설정합니다. D = DEBUG로 설정하면 DEBUG, INFO, ERROR로 출력되는 모든 로그를 출력합니다.

 

종목 조회

item_list = upbit.get_items("KRW", "BTC")

공통 모듈에 만든 로직을 upbit.로직이름 을 이용해서 간편하게 언제 어디서든지 사용할 수 있습니다.

 

위의 로직은 KRW 마켓에서 거래되는 모든 종목 중 BTC(비트코인)은 제외하고 결과를 리턴하게 됩니다. 

 

결과 출력

logging.debug(item_list)

조회한 결과를 debug로 출력해 보겠습니다. 실제 운영할 때는 로그레벨을 INFO로 설정하여 해당 로그는 출력되지 않도록 하는 것이 좋습니다.

 

전체 코드

import sys
import logging
import traceback
from module import upbit


# -----------------------------------------------------------------------------
# - Name : main
# - Desc : 메인
# -----------------------------------------------------------------------------
if __name__ == '__main__':

    # noinspection PyBroadException
    try:
        # 로그레벨 설정(DEBUG)
        upbit.set_loglevel('D')

        # 종목 조회
        item_list = upbit.get_items("KRW", "BTC")

        # 결과 출력
        logging.info(item_list)

    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)

 

프로그램 실행 및 결과 확인

buy_bot 프로그램에서 ① 실행 버튼을 클릭하면 ② 결과가 정상적으로 출력됨을 확인할 수 있습니다.

 

출력된 결과를 보면 BTC(비트코인)는 제외된 후 KRW 마켓에서 거래되는 코인만 조회 된 것을 확인할 수 있습니다.

 

다음 시간부터는 조금씩 로직을 붙여가며 자동매매 프로그램을 만들어 가는 과정을 살펴 보도록 하겠습니다.