스캘핑 봇 개발 일지 #5 빗썸 자동거래 봇 완성: 파이썬 핵심 구성 요소와 공격적 스캘핑 전략 가이드

드디어 빗썸 자동거래 프로그램이 완성되었습니다! 많은 분들이 꿈꾸는 자신만의 트레이딩 시스템을 파이썬으로 구축하는 여정은 쉽지 않지만, 그만큼 보람 있는 일입니다. 이 가이드는 제가 직접 경험하며 완성한 자동화 시스템의 핵심 구성 요소들을 단계별로 소개합니다. 안정적이고 효율적인 봇 개발을 위한 필수적인 지식들을 함께 살펴보시죠.

1단계: 데이터 수집 (Data Collection)

자동화 시스템의 첫걸음은 정확하고 신뢰할 수 있는 데이터입니다. 실시간 시장 데이터는 물론, 때로는 웹사이트 정보나 기타 필요한 데이터를 꾸준히 수집해야 합니다. 이 과정에서 ccxt와 같이 다양한 암호화폐 거래소와의 연동을 최소화하여 거래 및 데이터 수집을 가능하게 하는 유용한 라이브러리가 있습니다. 하지만 제 봇의 경우, 빗썸과의 교신 안정성 및 최적화를 위해 빗썸 공식 API를 직접 활용하여 데이터를 수집했습니다. 데이터를 어떻게 효율적으로 가져올 것인지가 성공적인 전략 구현의 바탕이 됩니다.

2단계: 전략 구현 (Strategy Implementation)

수집된 데이터는 단순한 숫자에 불과합니다. 이 데이터를 바탕으로 언제, 어떻게 거래할지 결정하는 의사결정 로직, 즉 '전략'을 구현하는 것이 핵심입니다.

제가 구현한 트레이딩 전략은 공격적인 스캘핑(Aggressive Scalping) 방식으로, 두 가지 인기 있는 기술적 지표인 지수 이동평균선(EMA)과 상대 강도 지수(RSI)를 결합합니다. 핵심 아이디어는 암호화폐가 새로운 상승 추세에 진입하면서 동시에 '과매도' 상태일 때 단기적인 거래 기회를 찾는 것입니다. 이러한 두 가지 요소를 활용하는 접근 방식은 빠른 가격 반등을 포착하기 위해 설계되었습니다.

전략 구현에는 pandas가 데이터를 효율적으로 분석하고 처리하는 데 필수적이며, pandas-ta (또는 ta 라이브러리)는 EMA나 RSI와 같은 기술적 지표들을 손쉽게 계산할 수 있도록 도와줍니다.

• 예시:
  • EMA 크로스: 단기 이동평균선(EMA)이 장기 이동평균선을 상향 돌파하면 매수 신호로 판단할 수 있습니다.
  • RSI: 상대 강도 지수(RSI)가 30 이하로 떨어지면 과매도 상태로 판단하여 매수 신호를 포착할 수 있습니다.

3단계: 작업 실행 (Execution)

전략이 매수 또는 매도 신호를 보내면, 실제 행동으로 옮겨야 합니다. 트레이딩 봇의 경우, 거래소의 API를 호출하여 주문을 실행하는 단계입니다. requests 라이브러리를 직접 사용하여 API에 요청을 보내거나, ccxt와 같은 거래소 통합 라이브러리를 통해 주문을 실행할 수 있습니다.

 

저는 최초 ccxt를 사용하여 구축하였으나 빗썸과 연결 시 지속적인 오류로 빗썸 API와 직접 requests를 통해 통신하는 방식으로 구현하였습니다. 또, IS_PAPER와 같은 플래그를 사용하여 실제 거래와 가상 거래(페이퍼 트레이딩)를 전환할 수 있도록 했습니다.  참고로, ccxt는 페이퍼 트레이딩을 지원하지 않습니다.

4단계: 로깅 및 알림 (Logging & Notifications: Telegram)

자동화 시스템은 백그라운드에서 끊임없이 실행됩니다. 따라서 시스템의 진행 상황을 투명하게 모니터링하는 것이 매우 중요합니다. 파이썬의 표준 logging 모듈을 사용하여 시스템의 상태, 이벤트, 오류 등을 체계적으로 기록하고 거래 및 운영상태를 모니터링하는 한편 telegram과 연동하여 주요 거래 신호나 예상치 못한 오류 발생 시 텔레그램으로 즉시 알림을 받도록 설정하여 시스템의 안정적인 운영을 도모하였습니다.

5단계: 수익추적 (Firebase)

마지막으로 실현될 이익 (또는 손실)을 데이터화하여 활용가능하도록 파이어베이스 DB와 연동하여 거래가 일어날 때마다 저장하도록 설정하였습니다.

프로젝트 과정 및 참고 링크

코딩, 삶에 스며들다 - 스캘핑 봇 개발 일지 #1 나만의 빗썸 자동매매 스캘핑 봇: 개발부터 Oracle Cloud 배포까지 (Python, Docker, Telegram)

나만의 빗썸 자동매매 스캘핑 봇: 개발부터 Oracle Cloud 배포까지 (Python, Docker, Telegram)

코딩, 삶에 스며들다 - 스캘핑 봇 개발 일지 #2. 속도 향상! Docker 컨테이너에서 Hot Reloading 환경 구축하기 (feat. Nodemon)

스캘핑 봇 개발 일지 #2. 속도 향상! Docker 컨테이너에서 Hot Reloading 환경 구축하기 (feat. Nodemon)

코딩, 삶에 스며들다 - 스캘핑 봇 개발 일지 #3. 속도 향상! Docker 컨테이너에서 Hot Reloading 환경 구축하기 (feat. Nodemon)

스캘핑 봇 개발 일지 #3. 속도 향상! Docker 컨테이너에서 Hot Reloading 환경 구축하기 (feat. Nodemon)

코딩, 삶에 스며들다 - 스캘핑 봇 개발 일지 #4. 파이썬 트레이딩 봇: 파이어스토어로 수익 추적 및 데이터 관리

스캘핑 봇 개발 일지 #4. 파이썬 트레이딩 봇: 파이어스토어로 수익 추적 및 데이터 관리

https://apidocs.bithumb.com/reference

마켓 코드 조회

프로젝트 간소버전

import os
import time
import requests
import hashlib
import hmac
import base64
import urllib.parse
from collections import OrderedDict
import pandas as pd
import pandas_ta as ta
import logging
from dotenv import load_dotenv

# 환경 변수 불러오기
load_dotenv()

# 로깅 설정
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# --- 설정 및 전역 변수 ---
IS_PAPER = True  # 테스트를 위해 True로 설정
EMA_SHORT = 12
EMA_LONG = 26
RSI_PERIOD = 14
RSI_OVERBOUGHT = 70
RSI_OVERSOLD = 30
POLL_INTERVAL_SEC = 60 # 60초 간격으로 반복

# --- Bithumb API 호출 함수 (requests 사용) ---
def fetch_krw_balance():
    """Bithumb API를 사용해 계좌 KRW 잔고를 가져옵니다."""
    api_key = os.getenv("API_KEY")
    api_secret = os.getenv("API_SECRET")

    if not api_key or not api_secret:
        logger.error("API_KEY 또는 API_SECRET 환경 변수가 설정되지 않았습니다.")
        return None

    endpoint = "/info/balance"
    url = f"https://api.bithumb.com{endpoint}"
    nonce = str(int(time.time() * 1000))
    
    payload = OrderedDict([
        ('endpoint', endpoint),
        ('currency', 'ALL'),
    ])
    
    query_string = urllib.parse.urlencode(payload)
    message = (endpoint + '\x00' + query_string + '\x00' + nonce).encode('utf-8')
    secret_bytes = base64.b64decode(api_secret)
    h = hmac.new(secret_bytes, message, hashlib.sha512)
    signature = base64.b64encode(h.digest())
    
    headers = {
        'Api-Key': api_key,
        'Api-Sign': signature.decode('utf-8'),
        'Api-Nonce': nonce,
        'Content-Type': 'application/x-www-form-urlencoded',
    }
    
    try:
        response = requests.post(url, data=query_string, headers=headers)
        response.raise_for_status()
        data = response.json()
        
        if data.get('status') == '0000':
            balance = data.get('data', {}).get('total_krw')
            return float(balance)
        else:
            logger.error(f"API 오류: {data.get('message')}")
            return None
    except Exception as e:
        logger.error(f"잔고 확인 중 오류 발생: {e}")
        return None

def execute_trade(signal, price, trade_volume):
    """
    매수(bid) 또는 매도(ask) 주문을 실행합니다.
    실제 API 호출을 위한 기본 구조를 포함합니다.
    """
    if IS_PAPER:
        logger.info(f"[가상] {signal.upper()} 신호! {trade_volume:.8f} BTC 주문을 시뮬레이션합니다.")
        return True # 가상 거래 성공으로 간주
    
    api_key = os.getenv("API_KEY")
    api_secret = os.getenv("API_SECRET")

    endpoint = "/trade/market_buy" if signal == "BUY" else "/trade/market_sell"
    url = f"https://api.bithumb.com{endpoint}"
    nonce = str(int(time.time() * 1000))

    payload = OrderedDict([
        ('endpoint', endpoint),
        ('order_currency', 'BTC'),
        ('payment_currency', 'KRW'),
        ('units', trade_volume)
    ])

    query_string = urllib.parse.urlencode(payload)
    message = (endpoint + '\x00' + query_string + '\x00' + nonce).encode('utf-8')
    secret_bytes = base64.b64decode(api_secret)
    h = hmac.new(secret_bytes, message, hashlib.sha512)
    signature = base64.b64encode(h.digest())

    headers = {
        'Api-Key': api_key,
        'Api-Sign': signature.decode('utf-8'),
        'Api-Nonce': nonce,
        'Content-Type': 'application/x-www-form-urlencoded',
    }
    
    try:
        response = requests.post(url, data=query_string, headers=headers)
        response.raise_for_status()
        data = response.json()
        
        if data.get('status') == '0000':
            logger.info(f"{signal.upper()} 주문 성공: {data.get('order_id')}")
            return True
        else:
            logger.error(f"주문 실패: {data.get('message')}")
            return False
    except Exception as e:
        logger.error(f"주문 실행 중 오류 발생: {e}")
        return False

# --- 데이터베이스 및 메시징 함수 (placeholder) ---
def record_trade_to_db(trade_record):
    """(placeholder) Firestore와 같은 DB에 거래 내역을 저장합니다."""
    logger.info("DB에 거래 내역을 저장합니다.")
    # 여기에 데이터베이스 저장 로직을 추가하세요.
    pass

def send_telegram_message(message):
    """(placeholder) 텔레그램으로 메시지를 전송합니다."""
    logger.info(f"텔레그램 메시지 전송: {message}")
    # 여기에 텔레그램 API 호출 로직을 추가하세요.
    pass

# --- 전략 및 메인 로직 ---
def fetch_simulated_ohlcv():
    """전략 테스트를 위한 가상 OHLCV 데이터를 생성합니다."""
    logger.info("가상 데이터 수집 중...")
    return pd.DataFrame({
        'close': [100, 101, 102, 103, 104, 105, 106, 107, 108, 109, 110]
    })

def apply_strategy(df):
    """EMA 지표를 사용해 매수/매도 신호를 생성합니다."""
    if len(df) < max(EMA_SHORT, EMA_LONG):
        return None
    df['ema12'] = ta.ema(df['close'], length=EMA_SHORT)
    df['ema26'] = ta.ema(df['close'], length=EMA_LONG)
    
    last_row = df.iloc[-1]
    
    if last_row['ema12'] > last_row['ema26']:
        return "BUY"
    return "SELL"

def main_loop():
    """메인 루프: 데이터 수집 -> 전략 적용 -> 주문 실행."""
    # 초기 잔고 확인
    krw_balance = fetch_krw_balance()
    if krw_balance:
        logger.info(f"시작 잔고: {krw_balance:,.2f} KRW")
    
    while True:
        try:
            # 1. 가상 데이터 수집
            ohlcv_df = fetch_simulated_ohlcv()
            
            # 2. 전략 적용 및 신호 생성
            signal = apply_strategy(ohlcv_df)
            current_price = ohlcv_df['close'].iloc[-1]
            
            # 3. 신호에 따라 주문 실행
            if signal:
                trade_volume = 5000 / current_price # 최소 주문 금액 기준
                if execute_trade(signal, current_price, trade_volume):
                    # 가상 거래 기록
                    trade_record = {
                        'signal': signal,
                        'price': current_price,
                        'volume': trade_volume,
                        'timestamp': time.time(),
                    }
                    record_trade_to_db(trade_record)
                    send_telegram_message(f"주문 실행: {signal.upper()} at {current_price}")
            
        except Exception as e:
            logger.error(f"메인 루프 오류: {e}")
            
        time.sleep(POLL_INTERVAL_SEC)

if __name__ == '__main__':
    logger.info("자동화된 봇 테스트 시작!")
    # 아래 주석을 해제하여 테스트를 시작하세요.
    # main_loop()
    print("--- 테스트 완료 ---")

프로젝트를 마치며

이렇게 파이썬 자동거래 시스템을 구축할 수 있습니다. 앞으로는 해당 프로그램의 성과에 대해서 공유 해 볼까 합니다.