[python] "업비트 API 활용" 비트코인 자동 매매 프로그램 : 초보자 가이드

이번 포스팅에서는 파이썬으로 「업비트 API를 활용한 비트코인 자동 매매 프로그램」을 만드는 방법을 다뤄봤습니다.

 

비트코인 자동 매매의 개념, 업비트 API 소개, 그리고 트레이딩 전략을 반영한 코드를 순서대로 작성하였고

자동 매매 프로그램의 리스크에 대한 유의사항도 고민해 봤습니다.

 

요즘 비트코인 열차가 전고점을 돌파해 안드로메다를 향하고 있는데요.

저희 모두 열심히 공부해서 이번 열차에는 탑승하도록 하시죠😝

---

들어가기에 앞서 - 비트코인 자동 매매 프로그램이 필요한 이유

비트코인은 24시간, 주말에도 계속 거래되기 때문에, 사람이 쉬는 시간에도 가격 변동이 일어나죠. 그런데 암호화폐 시장은 변동성이 크고 가격이 급격히 오르거나 떨어질 때가 많아서 잠시만 자리를 비워도 매수 타이밍을 놓칠 수 있어요. 또, 매매 타이밍을 잡는 일이 쉽지 않아서 심리적으로도 부담이 큽니다.

자동 매수 프로그램은 이런 문제를 해결해 줍니다.

미리 정해둔 조건에 따라 가격이 특정 구간에 도달했을 때 자동으로 매수를 실행해 주기 때문에, 사용자가 일일이 차트를 보거나 신경 쓰지 않아도 됩니다. 예를 들어 "비트코인 가격이 일정 수준까지 떨어지면 매수한다"는 조건을 설정해 두면, 그 가격에 도달했을 때 프로그램이 알아서 매수해 줍니다. 이렇게 하면 바쁜 일상 속에서도 거래 기회를 놓치지 않을 수 있죠.

또한 자동 매수 프로그램 덕분에 실시간으로 가격에 휘둘리지 않고, 미리 정해둔 전략에 따라 차분하게 거래할 수 있습니다. 불필요한 감정 개입을 줄이고, 계획적으로 매매를 진행할 수 있어서 훨씬 안정적인 거래가 가능해집니다.

---

1. 개발 환경 설정하기

• Python 설치 : Python이 설치되지 않았다면 설치 방법을 안내합니다.
• 필요한 라이브러리 설치 : requests, pandas, datetime 등의 라이브러리를 설치합니다.

pip install requests pandas

• 업비트 API 키 발급 : 업비트에 가입 후 API 키(Access Key, Secret Key)를 발급받는 방법을 설명합니다.

---

2. 업비트 API 이해하기

• API 개요 : 업비트 API의 기본적인 작동 원리와 주요 엔드포인트(가격 확인, 잔고 확인, 주문 실행)에 대해 설명합니다.
• API 연결하기

import requests

access_key = 'YOUR_ACCESS_KEY'
secret_key = 'YOUR_SECRET_KEY'
base_url = 'https://api.upbit.com/v1'

# 예시: 잔고 조회
headers = {
    'Authorization': f'Bearer {access_key}:{secret_key}'
}

def get_balance():
    response = requests.get(f'{base_url}/accounts', headers=headers)
    return response.json()

 

---

3. 트레이딩 로직 만들기

• 트레이딩 전략 설정 : 초보자를 위해 간단한 이동평균 교차 전략을 예로 듭니다. 이 전략은 단기 이동평균선이 장기 이동평균선을 상향 돌파할 때 매수하고, 하향 돌파할 때 매도하는 방식입니다.
• 가격 데이터 가져오기

def get_price_data(crypto='KRW-BTC', count=50):
    url = f"{base_url}/candles/days"
    params = {"market": crypto, "count": count}
    response = requests.get(url, params=params, headers=headers)
    return [candle['trade_price'] for candle in response.json()]

• 이동평균 계산하기

def calculate_moving_average(prices, window):
    return sum(prices[-window:]) / window

• 매수 및 매도 조건 설정

def should_buy(prices, short_window=5, long_window=10):
    short_ma = calculate_moving_average(prices, short_window)
    long_ma = calculate_moving_average(prices, long_window)
    return short_ma > long_ma

def should_sell(prices, short_window=5, long_window=10):
    short_ma = calculate_moving_average(prices, short_window)
    long_ma = calculate_moving_average(prices, long_window)
    return short_ma < long_ma

---

4. 업비트 API로 주문 실행하기

• 매수 및 매도 함수

def place_order(crypto, side, volume):
    order_url = f"{base_url}/orders"
    order_data = {
        'market': crypto,
        'side': side,
        'volume': volume,
        'ord_type': 'market'
    }
    response = requests.post(order_url, json=order_data, headers=headers)
    return response.json()

# 예시 사용법
buy_order = place_order('KRW-BTC', 'bid', '0.001')
sell_order = place_order('KRW-BTC', 'ask', '0.001')

---

5. 봇(bot) 자동화하기

• 전략 반복 실행하기 : 루프를 설정해 시장 상황을 지속적으로 체크하고 매매 조건이 충족될 때마다 자동으로 거래를 실행합니다.
• 봇 코드 예시

import time

crypto = 'KRW-BTC'
volume = '0.001'  # 예시 거래량

while True:
    prices = get_price_data(crypto)
    if should_buy(prices):
        print("비트코인 매수 중...")
        place_order(crypto, 'bid', volume)
    elif should_sell(prices):
        print("비트코인 매도 중...")
        place_order(crypto, 'ask', volume)
    time.sleep(60)  # 1분마다 실행

---

6. 오류 처리 및 로그 기록

• 오류 처리 : API 요청에 대한 기본 오류 처리 방식을 설명합니다.
• 거래 로그 기록

import logging

logging.basicConfig(filename='trading_bot.log', level=logging.INFO)

def log_trade(action, crypto, volume):
    logging.info(f"{action} {crypto} - Volume: {volume}")

• 거래 함수에 로그 기능 추가

def place_order(crypto, side, volume):
    # 동일한 함수 내용...
    log_trade(side, crypto, volume)
    return response.json()

---

7. 봇(bot) 테스트하기

백테스팅

백테스팅은 과거 데이터를 이용해 프로그램이 어떻게 작동할지를 테스트하는 방법입니다. 다음은 과거 데이터를 사용해 이동평균 전략을 백테스팅하는 예시입니다.

import pandas as pd

# 가상의 과거 데이터 예시 (날짜와 가격만 사용)
# 실제로는 업비트 API에서 과거 데이터를 가져오거나 csv 파일로 가져옵니다.
data = {
    "date": ["2024-01-01", "2024-01-02", "2024-01-03", "2024-01-04", "2024-01-05"],
    "price": [100, 102, 101, 103, 105]
}
df = pd.DataFrame(data)

# 이동평균 계산 함수
def calculate_moving_average(prices, window):
    return sum(prices[-window:]) / window

# 백테스팅 실행
short_window = 3
long_window = 5
for i in range(long_window, len(df)):
    short_ma = calculate_moving_average(df['price'][:i], short_window)
    long_ma = calculate_moving_average(df['price'][:i], long_window)
    if short_ma > long_ma:
        print(f"{df['date'][i]}: 매수 신호 발생")
    elif short_ma < long_ma:
        print(f"{df['date'][i]}: 매도 신호 발생")

 

이 코드는 가상의 과거 데이터를 사용해 단기 이동평균선이 장기 이동평균선을 돌파할 때 매수, 반대 상황에 매도 신호를 출력합니다. 실제로는 더 많은 데이터가 필요하며, 조건에 따라 매매 시뮬레이션을 추가해 볼 수도 있습니다.

---

테스트 모드 실행

테스트 모드에서는 실제로 주문을 넣지 않고 print문을 사용하여 프로그램이 예상대로 작동하는지 확인해 볼 수 있습니다. 예를 들어 매매 조건이 만족될 때만 메시지를 출력하는 방식으로 테스트할 수 있습니다.

import time

crypto = 'KRW-BTC'
volume = '0.001'

def should_buy(prices):
    return prices[-1] < prices[-2]  # 가격이 떨어질 때 매수 예시 조건

def should_sell(prices):
    return prices[-1] > prices[-2]  # 가격이 오를 때 매도 예시 조건

while True:
    # 가상의 현재가 리스트 (실제 데이터에서는 API 호출로 대체)
    prices = [100, 99, 101, 100, 98]

    if should_buy(prices):
        print("매수 조건 충족 - 매수 테스트 실행")
    elif should_sell(prices):
        print("매도 조건 충족 - 매도 테스트 실행")

    time.sleep(60)  # 1분마다 실행

 

이 코드에서는 가상의 가격 데이터를 사용해 매수와 매도 조건이 만족되었을 때 매수·매도 테스트가 실행되도록 했습니다. 실제로 프로그램을 실전에서 돌리기 전, 조건이 예상대로 작동하는지 검증할 때 유용합니다.

---

8. 리스크 및 고려 사항

1) 네트워크 이슈 대비

네트워크 연결이 불안정할 경우 자동 매매 봇이 오류를 발생시키지 않도록 예외 처리를 추가할 수 있습니다. 아래 코드는 try-except 구문을 사용해 API 호출이 실패할 경우 잠시 대기 후 다시 시도하도록 구성한 예시입니다.

import requests

def get_price(crypto='KRW-BTC'):
    try:
        response = requests.get(f'https://api.upbit.com/v1/ticker?markets={crypto}')
        response.raise_for_status()
        return response.json()[0]['trade_price']
    except requests.exceptions.RequestException as e:
        print(f"네트워크 오류 발생: {e}")
        time.sleep(60)  # 오류 발생 시 1분 대기 후 재시도
        return get_price(crypto)  # 재귀 호출로 재시도

---

2)  API 속도 이슈 대비

API 호출을 너무 자주 하면 거래소에서 일시적으로 차단당할 수 있습니다. 따라서 요청 간격을 두어 호출 수를 제한해야 합니다.

import time

crypto = 'KRW-BTC'

while True:
    # 가격 데이터를 가져오는 API 호출
    price = get_price(crypto)
    print(f"{crypto} 현재가: {price}")

    # 다음 요청을 위한 1분 대기
    time.sleep(60)

 

time.sleep(60)을 사용하여 매 1분마다 가격 데이터를 확인하게 함으로써 API 요청 빈도를 낮추고 속도 제한을 피할 수 있습니다. 이 외에도 매매 조건이 충족될 때만 API를 호출하는 방식으로 요청 수를 줄일 수 있습니다.

---

3) 시장 변동성에 대비한 손실 방지(손절매 기능 추가)

시장 변동성이 클 경우, 큰 손실을 방지하기 위해 손절매 조건을 추가할 수 있습니다. 예를 들어, 일정 수준의 손실이 발생할 경우 자동으로 매도하여 손실을 줄일 수 있습니다.

buy_price = 100  # 매수 가격 (예시)
loss_threshold = 0.05  # 손절매 조건: 5% 손실

def check_loss(price):
    if price <= buy_price * (1 - loss_threshold):
        print("손절매 조건 충족 - 매도 실행")
        # 매도 함수 호출 (place_order 함수 등 사용)
    else:
        print("손절매 조건 미충족 - 매수 유지")

# 현재가를 가져와 손절매 여부 확인
current_price = get_price(crypto)
check_loss(current_price)

 

이 코드는 매수 후 5% 이상의 손실이 발생할 경우 손절매 조건을 만족하게 하여 매도하도록 구성되어 있습니다. 손절매는 큰 손실을 방지하는 중요한 기능 중 하나입니다.

 

9. 전체 소스코드 (참고용)

앞서 설명한 내용을 모두 반영하여 완성한 비트코인 자동 매매 프로그램의 전체 코드입니다.

 

이 코드는 백테스팅, 오류 처리, 네트워크 문제 대응, API 속도 제한, 손절매 조건까지 포함해 실제 거래 시 발생할 수 있는 상황에 대응할 수 있도록 작성했습니다. API 키와 설정을 입력한 후 바로 실행할 수 있도록 구성하였습니다.

• 백테스팅 : backtest() 함수를 통해 과거 데이터를 바탕으로 매수·매도 신호가 언제 발생하는지 확인합니다.
• 네트워크 오류 처리 : get_price() 함수에서 네트워크 오류 발생 시 1분 대기 후 다시 시도하는 구조로 작성되어 있습니다.
• 손절매 조건 : check_loss() 함수로 매수 후 일정 손실이 발생할 경우 손절매를 실행하도록 합니다.
• API 속도 제한 대응 : time.sleep(60)을 통해 매 1분 간격으로 실행되도록 하여 과도한 API 호출을 방지합니다.

import requests
import time
import logging
import pandas as pd

# Upbit API 설정 (본인의 API 키로 입력)
access_key = 'YOUR_ACCESS_KEY'
secret_key = 'YOUR_SECRET_KEY'
base_url = 'https://api.upbit.com/v1'

# 로그 설정
logging.basicConfig(filename='trading_bot.log', level=logging.INFO)

# 과거 데이터로 백테스팅
def backtest(prices, short_window=5, long_window=10):
    for i in range(long_window, len(prices)):
        short_ma = calculate_moving_average(prices[:i], short_window)
        long_ma = calculate_moving_average(prices[:i], long_window)
        if short_ma > long_ma:
            print(f"{i}번째 데이터: 매수 신호 발생")
        elif short_ma < long_ma:
            print(f"{i}번째 데이터: 매도 신호 발생")

# 잔고 조회
def get_balance():
    headers = {
        'Authorization': f'Bearer {access_key}:{secret_key}'
    }
    response = requests.get(f'{base_url}/accounts', headers=headers)
    return response.json()

# 현재가 데이터 조회
def get_price(crypto='KRW-BTC'):
    try:
        response = requests.get(f'{base_url}/ticker?markets={crypto}')
        response.raise_for_status()
        return response.json()[0]['trade_price']
    except requests.exceptions.RequestException as e:
        print(f"네트워크 오류 발생: {e}")
        time.sleep(60)  # 오류 발생 시 1분 대기 후 재시도
        return get_price(crypto)  # 재시도

# 가격 데이터 가져오기 (과거 데이터 사용 시)
def get_price_data(crypto='KRW-BTC', count=50):
    url = f"{base_url}/candles/days"
    params = {"market": crypto, "count": count}
    headers = {
        'Authorization': f'Bearer {access_key}:{secret_key}'
    }
    response = requests.get(url, params=params, headers=headers)
    return [candle['trade_price'] for candle in response.json()]

# 이동평균 계산
def calculate_moving_average(prices, window):
    return sum(prices[-window:]) / window

# 매수 조건
def should_buy(prices, short_window=5, long_window=10):
    short_ma = calculate_moving_average(prices, short_window)
    long_ma = calculate_moving_average(prices, long_window)
    return short_ma > long_ma

# 매도 조건
def should_sell(prices, short_window=5, long_window=10):
    short_ma = calculate_moving_average(prices, short_window)
    long_ma = calculate_moving_average(prices, long_window)
    return short_ma < long_ma

# 손절매 조건 확인
def check_loss(current_price, buy_price, loss_threshold=0.05):
    if current_price <= buy_price * (1 - loss_threshold):
        print("손절매 조건 충족 - 매도 실행")
        place_order('KRW-BTC', 'ask', '0.001')  # 예시 매도 주문
    else:
        print("손절매 조건 미충족 - 매수 유지")

# 매수/매도 주문
def place_order(crypto, side, volume):
    order_url = f"{base_url}/orders"
    headers = {
        'Authorization': f'Bearer {access_key}:{secret_key}'
    }
    order_data = {
        'market': crypto,
        'side': side,
        'volume': volume,
        'ord_type': 'market'
    }
    response = requests.post(order_url, json=order_data, headers=headers)
    log_trade(side, crypto, volume)
    return response.json()

# 거래 로그 기록
def log_trade(action, crypto, volume):
    logging.info(f"{action} {crypto} - Volume: {volume}")

# 자동 매매 봇 실행
def run_bot():
    crypto = 'KRW-BTC'
    volume = '0.001'  # 거래량 설정
    buy_price = None  # 마지막 매수 가격 초기화

    while True:
        try:
            prices = get_price_data(crypto)
            current_price = get_price(crypto)

            if should_buy(prices):
                print("매수 조건 충족 - 매수 실행")
                place_order(crypto, 'bid', volume)
                buy_price = current_price  # 매수 가격 저장
            elif should_sell(prices) and buy_price:
                print("매도 조건 충족 - 매도 실행")
                place_order(crypto, 'ask', volume)
                buy_price = None  # 매도 후 매수 가격 초기화
            elif buy_price:
                # 손절매 조건 체크
                check_loss(current_price, buy_price)

            time.sleep(60)  # 1분마다 실행하여 API 속도 제한 대응
        except Exception as e:
            logging.error(f"Error: {e}")
            time.sleep(60)  # 오류 발생 시 1분 대기 후 재시도

# 백테스팅 실행
print("백테스팅 시작...")
historical_prices = get_price_data()  # 과거 데이터 가져오기
backtest(historical_prices)

# 프로그램 시작
if __name__ == "__main__":
    print("비트코인 자동 매매 봇 시작")
    run_bot()