(Python) 업비트 자동 매매 프로그램 만들기(7) - 모듈 작업

Summary

이번 게시물은 파이썬으로 업비트의 API 를 사용해 자동 매매 프로그램 만들기 최종편입니다.

전체 코드를 모듈화 하여 유지 관리가 용이하도록 해보겠습니다.

 

그러고보니 지난 게시물을 작성한 뒤 거의 4개월 정도가 지났네요.. 제 실수로 저장된 대부분의 스크립트, 파이썬 코드 등이 날아가 버렸습니다. 하하하... 제 멘탈도 같이 말이죠. 그래서 다른 스크립트들 복구하는게 급했기 때문에 조금 오래 걸렸던 거 같습니다.

 

 

모듈화 작업이란?

지금까지의 기억을 되짚어 보시면 업비트 API를 사용하기 위한 환경설정, 정보를 조회하는 방법, 주문을 시도하는 방법, 거래 타이밍을 잡는 방법, 로그를 출력하는 방법까지 크게 5단계에 걸친 포스팅이었습니다.

 

이러한 내용을 토대로 전체 코드를 모듈화하도록 하겠습니다.

파이썬의 모듈화는 코드를 기능별로 분할해 재사용성을 높이고 관리를 용이하게 합니다. 파일 단위로 구성된 모듈을 import문을 사용하여 다른 코드에서 불러와 사용할 수 있습니다. 이를 통해 프로젝트의 구조를 체계적으로 관리할 수 있습니다.

 

모듈화

 

각 모듈명과 내용은 아래와 같습니다.

• config.py : 설정 관련 변수를 저장합니다.
• upbit_api.py : Upbit API 호출을 처리합니다.
• trading_strategy.py : 거래 전략, 시그널 계산, 거래 로직을 포함합니다.
• mlog.py : my log 라는 뜻으로 로그를 출력하기 위한 모듈입니다.
• main.py : 프로그램의 시작점을 정의하고 무한 루프로 실행됩니다.

 

 

Module 1) config - 설정 변수

config 모듈은 말 그대로 사용자가 필요한 설정 값들을 관리하는 모듈으로 특별한 내용은 없습니다.

거래할 코인 심볼, API 연동을 위한 키값, 보유 현금에 따른 매수금 지정 정도만 추가했습니다.

동작에 필요한 설정값이 있다면 추가해주세요.

# 거래할 Symbol
coin = "KRW-DOGE"

# 개인 Access, Secret Key
with open("key.txt") as f:
    access_key, secret_key = [line.strip() for line in f.readlines()]

# 보유 현금에 따른 매수금액 지정
def calculate_trade_unit(cash):
    if cash <= 300000:
        return 6000
    elif 300000 < cash <= 1000000:
        return 10000
    elif 1000000 < cash <= 1500000:
        return 15000
    elif 1500000 < cash <= 2000000:
        return 20000
    elif 2000000 < cash <= 3000000:
        return 25000
    elif 3000000 < cash <= 4000000:
        return 35000
    elif 4000000 < cash <= 8000000:
        return 50000
    elif 8000000 < cash:
        return 100000
    else:
        return 0

 

 

Module 2) mlog - 로그 출력

이 부분은 개인 로그 출력 방법이 있으시면 패스하셔도 됩니다.

해당 모듈을 사용하기 위해서는 'LOG_PATH'를 지정해야 합니다.

리눅스를 사용하는 경우 '/' 기호로 path 구분을 해주시면 되고,

윈도우를 사용하는 경우 '\\' 기호로 path 구분을 해주시면 됩니다.

import os
import logging
from logging import handlers
from datetime import datetime

# Logging Configuration
LOG_PATH = "D:\\Path\\to\\Logging"  # WINDOWS
# LOG_PATH = "/Path/to/Logging"  # LINUX
LOG_FILE = "upbit_bot"  # FILE NAME

# 로그 출력 포멧 설정
log_formatter = logging.Formatter('%(message)s')
log_handler = handlers.TimedRotatingFileHandler(
    filename=os.path.join(LOG_PATH, LOG_FILE),
    when='midnight',
    interval=1,
    encoding='utf-8'
)

log_handler.setFormatter(log_formatter)
log_handler.suffix = "%Y%m%d"

logger = logging.getLogger()
logger.setLevel(logging.INFO)
logger.addHandler(log_handler)

""" 로그 사용방법
    : 1) log함수를 사용할 파일에서 mlog.py 모듈을 import
    : 2) 로그를 출력하고 싶은 곳에서 'log(loglevel, msg1, msg2 ...)' 입력
    : 3) 일반적인 로그 레벨 종류 
        DG = 전체적인 코드의 흐름, 함수 및 APi 입출력, 에러 등 대부분의 값을 출력
        TR = 코인의 현재가격, 보유 코인수, 수익률 등 사용자가 거래함에 있어 확인할 정보들 출력
        WA = 경고성 메세지 출력. (에러 및 장애가 아니라 사용자에게 주의를 주는 로그)
    : ex) log('TR', "Logging Start")
    : TR|2024-03-13 22:37:44|Logging Start
"""
def log(level, *args):
    now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    messages = "|".join(str(arg) for arg in args)

    levels = ('TR', 'DG', 'WA')
    if level not in levels:
        logger.info(f"TR|{now}|Log Level Error")
    else:
        logger.info(f"{level}|{now}|{messages}")

""" 데코레이터 사용방법
    : 1) 데코레이터를 사용할 파일에서 mlog.py 모듈을 import
    : 2) 로그를 출력하고 싶은 함수 Define 위에 '@log_function_call' 입력
    : ex) 
    @log_function_call
    def get_cur_price(ticker):
        res = pyupbit.get_current_price(ticker)
        log('DG', 'response', f'price : '+ res)
        return res
    : DG|2024-03-13 22:54:21|request|get_cur_price(KRW-XRP)
    : DG|2024-03-13 22:54:21|response|price : 963.6
    
"""
def log_function_call(func):
    """Decorator to log function calls."""
    def wrapper(*args, **kwargs):
        params = ", ".join([str(arg) for arg in args])
        log("DG", "request", f"{func.__name__}({params})")
        return func(*args, **kwargs)
    return wrapper

 

 

Module 3) upbit_api - API 사용

해당 모듈은 Upbit API 사용을 위한 모듈입니다. 

가격, 개인 지갑, 주문 등 다양한 함수의 집합입니다.

import pyupbit
import time
import config
from mlog import *

# API 호출을 위한 Upbit 객체 초기화
upbit = pyupbit.Upbit(config.access_key, config.secret_key)

# API 호출 실패시 0.5초 주기로 최대 20번 재시도
def fetch_data(fetch_func, max_retries=20, delay=0.5):
    for i in range(max_retries):
        res = fetch_func() 
        if res is not None:
            return res
        time.sleep(delay)
    return None

# 현재 가격 조회
def get_cur_price(ticker):
    return fetch_data(lambda: pyupbit.get_current_price(ticker))

# 현재 보유 현금 조회
def get_balance_cash():
    return fetch_data(lambda: upbit.get_balance("KRW"))

# 현재 보유 코인 조회
def get_balance_coin(ticker):
    return fetch_data(lambda: upbit.get_balance(ticker))

# 코인의 평균 매수가 조회
def get_buy_avg(ticker):
    return fetch_data(lambda: upbit.get_avg_buy_price(ticker))

# 코인의 주문 정보 조회
def get_order_info(ticker):
    try:
        orders = fetch_data(lambda: upbit.get_order(ticker))
        if orders and "error" not in orders[0]:
            return orders[-1]
    except Exception as e:
        print(e)
    return None


# 시장가 매수
def order_buy_market(ticker, buy_amount):  
    """
    : ticker : 코인 이름
    : buy_amount : 매수할 금액
    """
    if buy_amount < 5000:  
        print("amount is better than 5000")
        return 0
    
    res = fetch_data(lambda: upbit.buy_market_order(ticker,buy_amount))

    if 'error' in res:
        print(res)
        res = 0

    return res

# 시장가 매도
def order_sell_market(ticker, volume):
    """
    : ticker : 코인 이름
    : volume : 매도할 수량
    """
    res = fetch_data(lambda: upbit.sell_market_order(ticker, volume))

    if 'error' in res:
        print(res)
        res = 0

    return res

# 지정가 매수
def order_buy_limit(ticker, price, volume):
    """
    : ticker : 코인 이름
    : price : 매수할 가격
    : volume : 매수할 수량
    """
    res = fetch_data(lambda: upbit.buy_limit_order(ticker, price, volume))

    if 'error' in res:
        print(res)
        res = 0

    return res

# 지정가 매도
def order_sell_limit(ticker, price, volume):
    """
    : ticker : 코인 이름
    : price : 매도할 가격
    : volume : 매도할 수량
    """
    res = fetch_data(lambda: upbit.sell_limit_order(ticker, price, volume))
    if 'error' in res:
        print(res)
        res = 0
        return res
    return res

# 주문 취소
def order_cancel(ticker):
    order_info = get_order_info(ticker)
    try:
        order_uuid = order_info['uuid']
        res = upbit.cancel_order(order_uuid)
        if 'error' in res:
            print(res)
            res = 0
            return res
        print(res)
    except Exception as e:
        res = 0 
        print(e)
    return res

 

 

Module 4) trading_strategy - 매수/매도 시점 계산

사실상 프로그램의 두뇌와 같은 모듈입니다. 

언제 매수/매도 할지를 결정하는 모듈로 개인에 맞게 전략을 수정하기 위해서는 해당 파일을 수정하시면 됩니다.

import pyupbit
from mlog import *
import upbit_api
import config

### 볼린져 밴드 20일 평균선, 2배수 상단/하단 밴드 형성
def get_bollinger_bands(prices, window=20, multiplier=2):
    # 기간(window) 동안의 단순 이동 평균 계산
    sma = prices.rolling(window).mean()
    # 지정된 기간의 표준 편차 계산
    rolling_std = prices.rolling(window).std()

    # 상단/하단 밴드 계산
    upper_band = sma + (rolling_std * multiplier)
    lower_band = sma - (rolling_std * multiplier)

    return upper_band, lower_band

### 현재가격이 볼린져 밴드 하단/상단을 돌파하는지 감시
def trading_signal(prices, ticker):
    upper_band, lower_band = get_bollinger_bands(prices)
    band_high = upper_band.iloc[-1]
    band_low = lower_band.iloc[-1]

    cur_price = upbit_api.get_cur_price(ticker)
    if cur_price > band_high:
        ret = 'SELL'
    elif cur_price < band_low:
        ret = 'BUY'
    else:
        ret = 'HOLD'

    log('DG', 'response', 'Signal : ' + ret)
    return ret

### 20일치 종가를 얻어와서 볼린져 밴드 분석해서 BUY/SELL/HOLD 판단 요청
@log_function_call
def trading(ticker):
    prices_data = pyupbit.get_ohlcv(ticker, interval="day", count=20)
    prices = prices_data['close']
    signal = trading_signal(prices, ticker)

    # 보유 현금
    balance_cash = upbit_api.get_balance_cash()

    # 보유 코인
    balance_coin = upbit_api.get_balance_coin(ticker)
    
    # 매수금액
    buy_unit = config.calculate_trade_unit(balance_cash)

    if signal == 'BUY' and balance_cash > buy_unit:
        ret = upbit_api.order_buy_market(ticker, buy_unit)
        log('TR', 'response', f'Buy Order : {ret}')
    elif signal == 'SELL' and balance_coin > 0:
        ret = upbit_api.order_sell_market(ticker, balance_coin)
        log('TR', 'response', f'Sell Order : {ret}')
    else:
        log('DG', 'response', 'Hold Position')
        ret = None
    return ret

 

 

Module 5) main - 프로그램 시작

프로그램을 실행하면 다양한 모듈들을 호출하여 거래를 시작하고 결과를 로그로 출력하도록 하는 메인 모듈입니다.

개인에 맞춰 로그 및 코드의 방향성을 수정해주시면 되겠습니다.

import time
import config
import trading_strategy
from mlog import log
from upbit_api import *

coin = config.coin

def get_margin(cur_price, buy_avg_price, cur_balance):
    margin_rate = float((cur_price - buy_avg_price) / buy_avg_price) * 100.0
    margin = cur_balance * buy_avg_price * (margin_rate / 100.0)
    return margin, margin_rate

def run():
    try:
        # BUY/SELL 거래 or HOLD
        ret = trading_strategy.trading(coin)

        # 현재 보유 현금 (소수 1번째 반올림)
        cur_cash = int(round(get_balance_cash(),0))

        # 현재 코인 가격
        cur_price = get_cur_price(coin)

        # 현재 코인 보유 수량(소수 3번째 반올림)
        cur_balance = round(get_balance_coin(coin),2)

        # 평균 매수가격
        buy_avg_price = get_buy_avg(coin)

        # 평가손익, 수익률
        margin, margins = get_margin(cur_price, buy_avg_price, cur_balance)
        
        log('TR', f'  잔액 : {cur_cash:,d}원') # ',d'는 천 단위 구분자 추가
        log('TR', f'보유량 : {cur_balance:,.2f} {coin}')
        log('TR', f'평단가 : {buy_avg_price:,.2f}원') # '.2f'는 소수점 아래 두 자리까지 표시
        log('TR', f'현재가 : {cur_price:,.2f}원')
        log('TR', f'수익률 : {margins:,.2f}%') # 소수점 아래 두 자리 수익률
        log('TR', f'수익금 : {int(round(margin,0)):,d}원')

        if ret is not None:
            print(ret)
    except ZeroDivisionError as Div_zero:
        log('WA', '보유 코인 수 확인', Div_zero)
    except Exception as e:
        log('DG', 'Running Error', e)


if __name__ == "__main__":
    while True:
        run()
        time.sleep(10)

 

 

실행결과

코드를 작성 후 main.py 를 실행합니다.

로그 파일을 모니터링하며 거래가 되길 기다려봅니다.

 

※해당 프로그램으로 수익을 거둔 것은 아닙니다※

10초마다 로그 출력

 

 

이렇게 해서 간단한 자동매매 봇을 만들어 보았습니다.

하지만 아쉽게도 해당 프로그램으로 부자가 될 수는 없습니다.

 

API 호출시 에러에 대한 처리를 하지 않았습니다. 이는 실제 본인 지갑과 연결해서 어떠한 에러들이 발생하는지 확인하면서 try ~ except 구문을 활용하여 예외 처리 및 로깅을 추가해 주셔야합니다.

 

또한 거래 전략도 좀 더 치밀하게 구현하여야 합니다.

볼린져 밴드는 단순한 보조지표로 활용해야하며, 실제 거래 전략에는 복잡한 전략이 포함되어야 합니다.

 

이 코드를 기반으로 본인만의 프로그램을 만드셔서 모두 성투하시기 바라겠습니다.