업비트 비트코인 GPT 자동매매 6편 : 자동 매매 메인루프 구성

이번 편에서는 5편까지 만든 “GPT 판단 → 주문 실행” 코드를 스케줄러 없이도 1시간마다 자동 반복하도록 메인 루프를 구성합니다.
안전장치, 예외 처리, 중복 체결 방지까지 함께 다룹니다.

사전 준비물

• 1편의 프로젝트 폴더, venv, requirements 설치 완료
• 2편의 .env 설정 및 API 연동 확인 완료
• 3편의 pyupbit 시세 수집 가능
• 4편의 GPT 판단 JSON 응답 구성
• 5편의 매수/매도 함수 구현

루프 설계 포인트

• 매 시각에 맞춰 1시간 간격으로 실행
• 동일 신호가 반복될 때는 중복 체결 방지
• API 오류나 일시적 네트워크 문제에 대한 재시도와 대기
• 비용 제어를 위한 호출 주기 유지 + 토큰 길이 관리
• 종료 신호(CTRL+C) 시 안전 종료

메인 루프 예제 코드

아래 코드는 3~5편에서 만든 로직을 감싸 안정적인 운영 루프를 구현한 예시입니다.
원하는 파일명으로 저장하세요. 예: bot_loop.py

import os
import time
import json
import random
import logging
from datetime import datetime, timedelta

import pyupbit
from openai import OpenAI
from dotenv import load_dotenv

# ─────────────────────────────────────────────────────────
# 기본 설정
# ─────────────────────────────────────────────────────────
load_dotenv()

UPBIT_ACCESS = os.getenv("UPBIT_ACCESS_KEY")
UPBIT_SECRET = os.getenv("UPBIT_SECRET_KEY")
OPENAI_KEY = os.getenv("OPENAI_API_KEY")

SYMBOL = "KRW-BTC"
INTERVAL = "day"          # 30일 일봉
COUNT = 30                # 최근 30개
LOOP_SECONDS = 3600       # 1시간 사이클
JITTER_RANGE = (5, 20)    # 호출 시각 분산(초) → 레이트리밋 완화
COOLDOWN_MIN = 30         # 동일 신호 재체결 최소 대기(분)
MIN_ORDER_KRW = 5000      # 업비트 최소 체결 금액
FEE_FACTOR = 0.9995       # 수수료 여유
MAX_RETRY = 3             # 단발 오류 재시도 횟수

# ─────────────────────────────────────────────────────────
# 로그 설정
# ─────────────────────────────────────────────────────────
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    handlers=[
        logging.StreamHandler(),
        logging.FileHandler("bot.log", encoding="utf-8")
    ],
)

# ─────────────────────────────────────────────────────────
# 클라이언트 준비
# ─────────────────────────────────────────────────────────
upbit = pyupbit.Upbit(UPBIT_ACCESS, UPBIT_SECRET)
client = OpenAI(api_key=OPENAI_KEY)

SYSTEM_PROMPT = """
You are a Bitcoin trading analyst.
Analyze the last 30 days of OHLC data and decide whether to BUY, SELL, or HOLD.
Respond only in JSON format:
{"decision":"buy"} or {"decision":"sell"} or {"decision":"hold"}
"""

# ─────────────────────────────────────────────────────────
# 유틸
# ─────────────────────────────────────────────────────────
def sleep_with_jitter(seconds: int):
    jitter = random.randint(*JITTER_RANGE)
    time.sleep(seconds + jitter)

def safe_get_ohlcv(retry=MAX_RETRY):
    for i in range(retry):
        try:
            return pyupbit.get_ohlcv(SYMBOL, count=COUNT, interval=INTERVAL)
        except Exception as e:
            logging.warning(f"시세 수집 실패({i+1}/{retry}): {e}")
            time.sleep(3 * (i + 1))
    raise RuntimeError("시세 수집 반복 실패")

def ask_gpt(chart_json: str, retry=MAX_RETRY) -> str:
    for i in range(retry):
        try:
            resp = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[
                    {"role": "system", "content": SYSTEM_PROMPT},
                    {"role": "user", "content": chart_json}
                ],
                response_format={"type": "json_object"}
            )
            content = resp.choices[0].message.content
            data = json.loads(content)
            decision = data.get("decision", "hold").lower()
            if decision not in ("buy", "sell", "hold"):
                decision = "hold"
            return decision
        except Exception as e:
            logging.warning(f"GPT 응답 실패({i+1}/{retry}): {e}")
            time.sleep(3 * (i + 1))
    logging.error("GPT 응답 반복 실패 → HOLD 처리")
    return "hold"

def buy_all_krw():
    krw = upbit.get_balance("KRW")
    if krw * FEE_FACTOR > MIN_ORDER_KRW:
        logging.info(f"매수 실행: KRW={krw:.0f}")
        upbit.buy_market_order(SYMBOL, krw * FEE_FACTOR)
        return True
    logging.info("매수 스킵: 잔고 부족")
    return False

def sell_all_btc():
    btc = upbit.get_balance(SYMBOL)
    price = pyupbit.get_orderbook(SYMBOL)["orderbook_units"][0]["ask_price"]
    if btc * price > MIN_ORDER_KRW:
        logging.info(f"매도 실행: BTC={btc:.6f}, price≈{price}")
        upbit.sell_market_order(SYMBOL, btc)
        return True
    logging.info("매도 스킵: 보유량 부족")
    return False

# ─────────────────────────────────────────────────────────
# 메인 루프
# ─────────────────────────────────────────────────────────
def run():
    last_decision = None
    last_trade_at = None

    logging.info("GPT Auto Trading Loop 시작")

    try:
        while True:
            start_ts = datetime.now()
            logging.info("사이클 시작")

            # 1) 시세 데이터 수집
            try:
                df = safe_get_ohlcv()
            except Exception as e:
                logging.error(f"시세 수집 치명 오류: {e}")
                sleep_with_jitter(LOOP_SECONDS)
                continue

            # 2) GPT 판단
            chart_json = df.to_json()
            decision = ask_gpt(chart_json)
            logging.info(f"GPT 판단: {decision.upper()}")

            # 3) 중복 체결 방지 (쿨다운)
            can_trade = True
            if last_decision == decision and decision in ("buy", "sell"):
                if last_trade_at and datetime.now() < last_trade_at + timedelta(minutes=COOLDOWN_MIN):
                    can_trade = False
                    left = (last_trade_at + timedelta(minutes=COOLDOWN_MIN) - datetime.now()).seconds
                    logging.info(f"중복 체결 방지로 스킵 ({left}초 후 재허용)")

            # 4) 주문 실행
            if can_trade:
                if decision == "buy":
                    if buy_all_krw():
                        last_decision = decision
                        last_trade_at = datetime.now()
                elif decision == "sell":
                    if sell_all_btc():
                        last_decision = decision
                        last_trade_at = datetime.now()
                else:
                    logging.info("HOLD 상태 → 대기")

            # 5) 사이클 종료 및 대기
            elapsed = (datetime.now() - start_ts).seconds
            sleep_for = max(0, LOOP_SECONDS - elapsed)
            logging.info(f"사이클 종료, 다음 실행까지 대기 {sleep_for}초")
            sleep_with_jitter(sleep_for)

    except KeyboardInterrupt:
        logging.info("수동 종료 신호 감지 → 안전 종료")
    except Exception as e:
        logging.exception(f"예상치 못한 예외로 종료: {e}")

if __name__ == "__main__":
    run()

코드 설명 핵심 포인트

• sleep_with_jitter
  • 호출 시점을 5~20초 무작위로 분산해 레이트리밋과 동시호출 리스크를 줄입니다.
• safe_get_ohlcv, ask_gpt
  • 일시적인 오류는 점증적 대기로 재시도합니다.
• COOLDOWN_MIN
  • 직전과 같은 신호가 반복되어도 30분 이내에는 재체결 금지합니다.
• FEE_FACTOR
  • 매수 시 수수료를 감안해 잔액*0.9995만 사용합니다.
• logging
  • 콘솔과 bot.log에 동시에 기록해 장애 원인을 추적하기 쉽습니다.
• KeyboardInterrupt
  • 로컬에서 CTRL+C로 안전하게 종료할 수 있습니다.

운영 팁

• 비용 관리
  • 1시간 간격은 토큰 비용과 신호 과민반응을 줄이는 무난한 선택입니다.
  • 30일 일봉 JSON은 비교적 짧아 비용이 안정적입니다.
• 예외 대비
  • 일시 오류는 재시도, 반복 실패는 HOLD로 안전 귀결.
  • 갑작스런 가격 급변에는 쿨다운이 과체결을 막아줍니다.
• 점진 도입
  • 반드시 초소액으로 충분히 검증한 후 규모를 늘리세요.

다음 편 예고

• 7편 — 리스크 관리 및 중복 매매 방지 전략 심화
  • 1회 주문 상한, 일일 손실 한도, 일일 체결 횟수 제한, 슬리피지 점검, 알림(디스코드/슬랙) 추가

관련 포스팅

업비트 비트코인 GPT 자동매매 4편 : GPT로 의사결정하기

 

업비트 비트코인 GPT 자동매매 5편 : 자동매매 실행하기