업비트 자동매매 시스템 코드 설명 (Python + PyUpbit)

1. 시스템 개요

이 코드는 PyUpbit API와 스케줄링 모듈을 이용하여 전략 기반 자동매매를 수행하는 Python 스크립트입니다. 특정 전략에 따라 매수/매도 시점을 판단하고, 조건이 충족되면 시장가 주문을 자동 실행합니다.

2. 주요 구성 요소

• dotenv: 환경 변수(API 키 등) 로드
• pyupbit: 실시간 시세, 캔들 데이터, 주문 기능 제공
• schedule: 전략 실행 주기 설정
• SQLite: 거래 기록 저장 (log_trade)
• Telegram: 실시간 알림 수신

3. 핵심 함수 설명

📌 get_dynamic_budget()

def get_dynamic_budget(strategy, base_budget):
    if "btc" in strategy.lower() or "eth" in strategy.lower():
        return 15000
    return 5000

→ 전략별로 매수 금액을 고정함 (고가 코인은 1.5만원, 그 외는 5천원)

📌 run_strategy()

전략 실행 로직의 핵심이며, 다음 과정을 수행합니다:

1. OHLCV 데이터 로드 (interval, count=200)
2. 전략에 따라 시그널 판단 (get_signal(df, amount))
3. 매수: 시장가 주문 + 로그 + 알림
4. 매도: 수익률 계산 후 매도 + 로그 + 알림

📌 schedule_strategies()

등록된 모든 전략을 종목별로 순차 실행합니다.

• magicsplit_conservative, magicsplit_aggressive 등 variant 지원
• TRX는 dca_5profit 및 magicsplit 전략만 허용

4. 매수 조건 보호 로직

if signal == "buy" and amount < 5000:
    print("⛔ 최소 주문 금액 미만")
    return

→ 업비트의 최소 주문 단가 기준을 만족하지 못하면 매수하지 않음

5. 체결 실패 감지

if volume == 0:
    send_message("체결 실패 경고")

→ 시장가 주문이 체결되지 않은 경우 텔레그램으로 경고 전송

6. 개선 포인트 및 확장

• 전략별 누적 수익률 시각화
• 백테스트 연동 (interactive_backtest.py)
• 손절/익절 기준 커스터마이징
• 하루 단위 요약 리포트 자동 전송

7. 마무리

이 시스템은 단일 파일로 작동하면서도, 다양한 전략을 동시에 운영하고 수익률 분석까지 가능한 구조로 되어 있습니다. 백테스트, 실전매매, 실시간 모니터링까지 모두 Python으로 통합 가능합니다.

 

import time import schedule import pyupbit import os import pandas as pd from dotenv import load_dotenv from datetime import datetime, timedelta import sqlite3 from collections import defaultdict # 전략 불러오기 from strategies.dca_5profit import get_signal as get_dca_5profit_signal from strategies.momentum import get_signal as get_momentum_signal from strategies.moving_average import get_signal as get_moving_average_signal from strategies.rsi import get_signal as get_rsi_signal from strategies.trend_following import get_signal as get_trend_following_signal from strategies.magicsplit import get_signal as get_magicsplit_signal, VARIANT_CONFIGS from telegram_bot import send_message, listen_for_commands from log_signal import log_signal from order_executor import market_buy, market_sell from db_logger import log_trade, init_db, get_last_trade_time, log_trade_reason # 환경 변수 로딩 load_dotenv() ACCESS_KEY = os.getenv("UPBIT_ACCESS_KEY") SECRET_KEY = os.getenv("UPBIT_SECRET_KEY") upbit = pyupbit.Upbit(ACCESS_KEY, SECRET_KEY) init_db() # 전략별 기본 예산 STRATEGY_BUDGETS = {     "dca_5profit": 10000,     "momentum": 10000,     "moving_average": 10000,     "rsi": 8000,     "trend_following": 15000 } STRATEGY_INTERVALS = {     "dca_5profit": "minute10",     "momentum": "minute15",     "moving_average": "minute30",     "rsi": "minute1",     "trend_following": "minute60" } DUPLICATE_BUY_COOLDOWN = {     "dca_5profit": 10,     "momentum": 30,     "moving_average": 30,     "rsi": 10,     "trend_following": 60 } # magicsplit variants 추가 for variant in VARIANT_CONFIGS:     name = f"magicsplit_{variant}"     STRATEGY_BUDGETS[name] = 10000     STRATEGY_INTERVALS[name] = "minute5"     DUPLICATE_BUY_COOLDOWN[name] = 5 # 이익 실현/손절 조건 TAKE_PROFIT = 0.05 STOP_LOSS = -0.03 POSITION_HISTORY = {} TRADE_REASON_LOG = "trade_reason_log.csv" def get_dynamic_budget(strategy, base_budget):     try:         total_base = sum(STRATEGY_BUDGETS.values())         krw_balance = upbit.get_balance("KRW")         if krw_balance is None or total_base == 0:             return base_budget         scale = min(1.0, krw_balance / total_base)         return int(base_budget * scale)     except Exception as e:         print(f"[BUDGET] 예산 계산 오류: {e}")         return base_budget def run_strategy(name, func, ticker):     print(f"⏱️ [{name}] 전략 실행 중... ({ticker})")     try:         interval = STRATEGY_INTERVALS.get(name, "day")         df = pyupbit.get_ohlcv(ticker, interval=interval, count=50)         if df is None or len(df) < 2:             print("❌ OHLCV 데이터 부족")             return         budget = get_dynamic_budget(name, STRATEGY_BUDGETS.get(name, 10000))         result = func(df, amount=budget)         if result is None or "signal" not in result:             print(f"💤 [{name}] 시그널 없음")             return         signal = result["signal"]         reason = result.get("reason", "N/A")         amount = result.get("amount", budget)         price = pyupbit.get_current_price(ticker)         if price is None:             print("❌ 가격 조회 실패")             return         last_time, last_side = get_last_trade_time(ticker, name)         cooldown = DUPLICATE_BUY_COOLDOWN.get(name, 30)         time_diff = (datetime.now() - last_time).total_seconds() / 60         if signal == "buy" and last_side == "buy" and time_diff < cooldown:             print(f"⛔ {name} 최근 매수({time_diff:.1f}분 전), 재매수 차단")             return         if signal == "buy":             if amount < 5000:                 print("⛔ 최소 주문 금액 미만")                 return             result_order = market_buy(ticker, amount)             if result_order:                 volume = float(result_order.get('executed_volume', 0))                 # ✅ 수량 0.0일 경우 경고 출력                 if volume == 0:                     print(f"⚠️ [WARNING] 체결 수량이 0입니다 — 주문 실패 가능 (amount={amount}, price={price})")                     send_message(f"⚠️ <b>[{name}] {ticker} 체결 실패</b>\n금액: {amount:,}원\n사유: 수량 0 (체결 실패 가능성)")                     return                 # ✅ 체결 성공 시 기존처럼 처리                 log_trade(ticker, "buy", volume, price, name)                 log_trade_reason(ticker, "buy", name, reason)                 log_signal(name, ticker, signal, price)                 POSITION_HISTORY[name + ticker] = (price, volume)                 send_message(f"📈 <b>[{name}] {ticker} 매수 완료</b>\n수량: {volume} ({amount:,}원)\n이유: {reason}")                 elif signal == "sell":             balance = upbit.get_balance(ticker)             if balance is None or balance < 0.0001:                 print("⛔ 매도할 잔고 부족")                 return             key = name + ticker             buy_price, volume = POSITION_HISTORY.get(key, (None, None))             if buy_price:                 pnl = (price - buy_price) / buy_price                 if pnl < STOP_LOSS:                     print(f"📉 [{name}] 손절 조건 실행: {pnl*100:.2f}%")                 elif pnl > TAKE_PROFIT:                     print(f"💰 [{name}] 익절 조건 실행: {pnl*100:.2f}%")             result_order = market_sell(ticker, balance)             if result_order:                 log_trade(ticker, "sell", balance, price, name)                 log_trade_reason(ticker, "sell", name, reason)                 log_signal(name, ticker, signal, price)                 send_message(f"📉 <b>[{name}] {ticker} 매도 완료</b>\n수량: {balance}\n이유: {reason}")                 POSITION_HISTORY.pop(key, None)     except Exception as e:         send_message(f"⚠️ <b>[{name}] 전략 실행 오류 ({ticker})</b>: {e}") def schedule_strategies():     strategies = {         "dca_5profit": get_dca_5profit_signal,         "momentum": get_momentum_signal,         "moving_average": get_moving_average_signal,         "rsi": get_rsi_signal,         "trend_following": get_trend_following_signal,     }     # magicsplit 전략 variant 별로 추가     for variant in VARIANT_CONFIGS:         name = f"magicsplit_{variant}"         strategies[name] = lambda df, amount, v=variant: get_magicsplit_signal(df, variant=v, amount=amount)     tickers = ["KRW-BTC", "KRW-ETH", "KRW-TRX"]     for name, func in strategies.items():         for ticker in tickers:             if ticker == "KRW-TRX" and "dca_5profit" not in name and "magicsplit" not in name:                 continue             run_strategy(name, func, ticker) def main():     print("🚀 전략 다중 자동매매 루프 시작")     schedule_strategies()     listen_for_commands(lambda cmd: None)  # 명령어 핸들 생략     while True:         schedule.run_pending()         time.sleep(1) if __name__ == "__main__":     main()