(Python) 업비트 자동 매매 프로그램 만들기(5) - 전략 구현

Summary

이번 게시물에서는 인기 있는 기술 분석 도구인 볼린져 밴드 전략을 이용해서 매수/매도 시점을 확인하는 방법을 알아보겠습니다.

 

그런데 유의해야 할 사항은 아래 내용이 간단하게 구현한 것이지, 절대 전문적이지 않습니다.

볼린져 밴드 자체도 훌륭한 보조 지표로써 활용할 수 있지만 단순히 해당 전략만으로 매매를 진행하는 것을 아주 위험하오니

"이렇게 흘러가는 구나" 하는 코드의 흐름을 익히고 자신만의 전략을 구현하는 것은 조금 더 노력이 필요합니다.

 

 

 

1) 볼린져 밴드(Bollinger band) 전략이란?

• 볼린져 밴드는 세 개의 선을 그리는 전략
  1) 중심 밴드 (Middle Band) : 주가의 단순 이동평균선이며, 20일 이동 평균을 사용
  2) 상단 밴드 (Upper Band) : 일반적으로 중심 밴드에서 2배의 표준 편차를 더한 값
  3) 하단 밴드 (Lower Band) : 중심 밴드에서 2배의 표준 편차를 뺀 값

 

• 볼린져 밴드의 해석 및 사용
  1) 변동성 측정 : 밴드가 확장될 때는 가격의 변동성이 높고, 밴드가 수축될 때는 변동성이 낮다는 것을 의미
  2) 과매수 / 과매도 판단 : 주가가 상단 밴드에 있는 경우 과매수 상태, 하단 밴드에 위치할 경우 과매도 상태로 판단
  3) 밴드 터치와 반등 : 주가가 밴드에 닿았다가 반대 방향으로 움직이기 시작하면 종종 가격의 반전 신호로 해석 가능

 

위의 그림을 보면 업비트 사이트에서 제공하는 볼린져 밴드 지표를 이용해서 리플의 15분봉 차트를 그린 결과입니다.

상단선과 하단선을 뚫고 나오는 차트가 있는 것을 확인할 수 있네요. 

 

좀 더 자세한 개념을 이해하고 사용해야 하겠지만 저희는 전문 트레이더는 아니기에 간단히 전략만 구현해 보겠습니다.

상단선을 뚫으면 매도하고, 하단선을 뚫으면 매수하는 방식을 알아봅시다.

 

 

 

2) 볼린져 밴드 계산

• 상단, 하단 밴드 계산

def get_bollinger_bands(prices, window=20, multiplier=2):

    ### 20일 기간 동안 이동 평균선 계산 (Middle Band)
    sma = prices.rolling(window).mean()

    ### 20일 동안의 표준 편차 계산
    rolling_std = prices.rolling(window).std()

    ### 중간 밴드 + (표준편차 * 2)
    upper_band = sma + (rolling_std * multiplier)
    
    ### 중간 밴드 - (표준편차 * 2)
    lower_band = sma - (rolling_std * multiplier)

    return upper_band, lower_band

 

• 함수의 인자는 3개로 구성
  prices : 가격 데이터
  window : 이동 평균을 계산하기 위한 기간. (기본값은 20일)
  multiplier : 상단 밴드와 하단 밴드를 계산할 때 사용되는 표준 편차의 배수. (기본값은 2배)

 

• 코드 설명
  1) sma : 20일 기간 동안 이동 평균선 계산 (Middle Band)
  2) rolling_std : 20일 동안의 표준 편차 계산
  3) upper_band : 중간 밴드 + (표준편차 * 2)
  4) lower_band : 중간 밴드 - (표준편차 * 2)
  5) return 값 : 상단 밴드, 하단 밴드

 

 

 

3) 매수/매도 타이밍 판단

• 볼린져 밴드의 상단/하단 밴드를 확인 후 현재 가격과 비교해서 매수/매도 판단

def trading_signal(prices):
    ### get_bollinger_bands() 함수를 통해 상단/하단 밴드 요청
    upper_band, lower_band = get_bollinger_bands(prices)

    band_high = upper_band.iloc[-1]
    band_low = lower_band.iloc[-1]
    cur_price = get_cur_price(coin)

    ### 상단/하단/현재가 출력
    print(f"HIGH : {band_high} / LOW : {band_low} / PRICE : {cur_price}")

    ### 현재 가격이 상단 밴드보다 큰 경우는 'BUY' 신호를, 
    ### 하단 밴드보다 낮은 경우는 'SELL' 신호를,
    ### 밴드안에 있는 경우는 'HOLD' 신호를 return
    if cur_price > band_high:
        print("Sell signal")
        return 'SELL'
    elif cur_price < band_low:
        print("Buy signal")
        return 'BUY'
    else:
        return 'HOLD'

 

• 코드 설명
  1) get_bollinger_bands() 함수를 통해 상단/하단 밴드 요청
  2) print() 문으로 상단/하단/현재가 출력
  3) 현재 가격이 상단밴드보다 큰 경우 '매도', 하단 밴드보다 낮은 경우 '매수' 신호 return
  

 

 

 

4) 거래 시작

• trading_signal() 함수를 통해 받은 signal 정보를 토대로 매매 진행

def trading(coin):
    ### pyupbit.get_ohlcv() 함수를 이용해서 업비트 사이트의 20일 데이터 받아오기
    prices_data = pyupbit.get_ohlcv(coin, interval="day", count=20)
    prices = prices_data['close']
    
    ### 받아온 가격 데이터의 종가만을 추출하여 trading_signal() 함수에 전달
    signal = trading_signal(prices)
    balance_cash = get_balance_cash()
    balance_coin = get_balance_coin(coin)

    ### signal 이 "BUY" 이고 현재 보유 현금이 10000원 보다 많은 경우 매수 주문
    ### signal 이 "SELL" 이고 현재 보유 코인이 1개 보다 많은 경우 매도 주문
    if signal == 'BUY' and balance_cash > 10000:
        print("BUY Order")
        return order_buy_market(coin, 10000)
    elif signal == 'SELL' and balance_coin > 0:
        print("SELL Order")
        return order_sell_market(coin, balance_coin)
    else:
        print("Hold position.")
    return None

 

• 코드 설명
  1) pyupbit.get_ohlcv() 함수를 이용해서 업비트 사이트의 20일 데이터 받아오기
  2) 받아온 가격 데이터의 종가만을 추출하여 trading_signal() 함수에 전달
  3) signal 이 "BUY" 이고 현재 보유 현금이 10000원 보다 많은 경우 매수 주문
  4) signal 이 "SELL" 이고 현재 보유 코인이 1개 보다 많은 경우 매도 주문

 

 

 

5) 전체 코드

import pyupbit
import time

# Configuration
coin = "KRW-XRP"
with open("key.txt") as f:
    access_key, secret_key = [line.strip() for line in f.readlines()]

# Upbit Initialization
upbit = pyupbit.Upbit(access_key, secret_key)


def fetch_data(fetch_func, max_retries=20, delay=0.5):
    for i in range(max_retries):
        res = fetch_func() # fetch_func() 함수를 호출하여 데이터
        if res is not None: # 가져온 데이터가 None이 아닌 경우 루프를 종료
            return res
        time.sleep(delay) # 데이터를 가져오지 못한 경우 0.5초 동안 대기
    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

# 볼린져 밴드와 시그널 생성 함수
def get_bollinger_bands(prices, window=20, multiplier=2):

    ### 20일 기간 동안 이동 평균선 계산 (Middle Band)
    sma = prices.rolling(window).mean()

    ### 20일 동안의 표준 편차 계산
    rolling_std = prices.rolling(window).std()

    ### 중간 밴드 + (표준편차 * 2)
    upper_band = sma + (rolling_std * multiplier)

    ### 중간 밴드 - (표준편차 * 2)
    lower_band = sma - (rolling_std * multiplier)

    return upper_band, lower_band

def trading_signal(prices):

    ### get_bollinger_bands() 함수를 통해 상단/하단 밴드 요청
    upper_band, lower_band = get_bollinger_bands(prices)

    band_high = upper_band.iloc[-1]
    band_low = lower_band.iloc[-1]
    cur_price = get_cur_price(coin)

    ### 상단/하단/현재가 출력
    print(f"HIGH : {band_high} / LOW : {band_low} / {coin} PRICE : {cur_price}")

    ### 현재 가격이 상단 밴드보다 큰 경우는 'BUY' 신호를, 
    ### 하단 밴드보다 낮은 경우는 'SELL' 신호를,
    ### 밴드안에 있는 경우는 'HOLD' 신호를 return
    if cur_price > band_high:
        print("Sell signal")
        return 'SELL'
    elif cur_price < band_low:
        print("Buy signal")
        return 'BUY'
    else:
        return 'HOLD'

def trading(ticker):

    ### pyupbit.get_ohlcv() 함수를 이용해서 업비트 사이트의 20일 데이터 받아오기
    prices_data = pyupbit.get_ohlcv(ticker, interval="day", count=20)
    prices = prices_data['close']
    
    ### 받아온 가격 데이터의 종가만을 추출하여 trading_signal() 함수에 전달
    signal = trading_signal(prices)
    balance_cash = get_balance_cash()
    balance_coin = get_balance_coin(ticker)

    ### signal 이 "BUY" 이고 현재 보유 현금이 10000원 보다 많은 경우 매수 주문
    ### signal 이 "SELL" 이고 현재 보유 코인이 1개 보다 많은 경우 매도 주문
    if signal == 'BUY' and balance_cash > 10000:
        print("BUY Order")
        return order_buy_market(ticker, 10000)
    elif signal == 'SELL' and balance_coin > 0:
        print("SELL Order")
        return order_sell_market(ticker, balance_coin)
    else:
        print("Hold position.")
    return None


def run():
    ret = trading(coin)
    if ret is not None:
        print(ret)

while True:
    run()
    time.sleep(10)

trading() 함수는 현재 가격을 판단하여 계산한 결과를 토대로 매수/매도를 진행하던지 아니면 HOLD 되고 종료합니다.
이를 계속 켜놔야 현재 가격이 밴드를 뚫고 나가는 순간 매매를 진행할 수 있습니다.

 

run() 함수를 만들어서 while 문 안에 넣어주고 10초의 딜레이를 주었습니다.

 

코드를 실행하면 현재 가격을 조회하며 밴드를 돌파하는지 확인하기 시작합니다.

---

오늘의 결론 : 

업비트 사이트의 각종 API와 파이썬 코드를 사용해서 볼린져 밴드를 활용한 매매를 자동으로 진행해주는 방법을 살펴보았습니다.

 

다시 한번 강조드리지만 자신만의 전략을 잘 구현한 뒤 많은 재산을 투자해서 사용하기 보단 소액으로 돌려보거나
과거 데이터를 통해 백테스팅과 같은 기법으로 충분한 검증이 필요합니다!!

 

다음 게시물에서는 다양한 로그를 날짜별로 출력하는 방법을 알아보도록 하겠습니다.