트레이딩 봇 개발 완벽 가이드: 서버 설치부터 24시간 자동 운영까지

# src/strategy.py
import pandas as pd
import ta

class MovingAverageCrossStrategy:
    """
    단기 이동평균이 장기 이동평균을 상향 돌파 → 매수 신호
    단기 이동평균이 장기 이동평균을 하향 돌파 → 매도 신호
    """
    def __init__(self, short_window: int = 20, long_window: int = 50):
        self.short_window = short_window
        self.long_window = long_window

    def generate_signal(self, df: pd.DataFrame) -> str:
        """
        df: OHLCV 데이터프레임 (open, high, low, close, volume)
        return: 'BUY', 'SELL', 'HOLD'
        """
        if len(df) < self.long_window:
            return 'HOLD'

        # 이동평균 계산
        df['ma_short'] = df['close'].rolling(self.short_window).mean()
        df['ma_long'] = df['close'].rolling(self.long_window).mean()

        # 크로스오버 감지
        prev = df.iloc[-2]
        curr = df.iloc[-1]

        # 골든 크로스: 단기가 장기를 상향 돌파
        if prev['ma_short'] <= prev['ma_long'] and curr['ma_short'] > curr['ma_long']:
            return 'BUY'

        # 데드 크로스: 단기가 장기를 하향 돌파
        if prev['ma_short'] >= prev['ma_long'] and curr['ma_short'] < curr['ma_long']:
            return 'SELL'

        return 'HOLD'

# src/exchange.py
import ccxt
import os
from loguru import logger

class ExchangeConnector:
    def __init__(self):
        # API 키는 반드시 환경변수로 관리
        self.exchange = ccxt.binance({
            'apiKey': os.getenv('EXCHANGE_API_KEY'),
            'secret': os.getenv('EXCHANGE_SECRET_KEY'),
            'enableRateLimit': True,  # API 호출 제한 준수
            'options': {
                'defaultType': 'spot',  # 현물 거래
            }
        })

    def get_ohlcv(self, symbol: str, timeframe: str = '1h', limit: int = 100):
        """OHLCV 캔들 데이터 수집"""
        try:
            ohlcv = self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
            df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
            df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
            return df
        except Exception as e:
            logger.error(f"OHLCV 수집 실패: {e}")
            return None

    def get_balance(self, currency: str = 'USDT') -> float:
        """잔고 조회"""
        balance = self.exchange.fetch_balance()
        return balance['free'].get(currency, 0)

    def create_market_order(self, symbol: str, side: str, amount: float):
        """시장가 주문 실행"""
        try:
            order = self.exchange.create_market_order(symbol, side, amount)
            logger.info(f"주문 완료: {side} {amount} {symbol}")
            return order
        except Exception as e:
            logger.error(f"주문 실패: {e}")
            raise

# src/risk_manager.py

class RiskManager:
    def __init__(self, config):
        self.max_position_pct = config['MAX_POSITION_PCT']   # 전체 자금 대비 최대 포지션 (예: 0.1 = 10%)
        self.stop_loss_pct = config['STOP_LOSS_PCT']         # 손절 기준 (예: 0.02 = 2%)
        self.take_profit_pct = config['TAKE_PROFIT_PCT']     # 익절 기준 (예: 0.05 = 5%)
        self.max_daily_loss_pct = config['MAX_DAILY_LOSS_PCT']  # 일일 최대 손실 한도 (예: 0.05 = 5%)
        self.daily_loss = 0.0
        self.trade_count_today = 0
        self.max_trades_per_day = config.get('MAX_TRADES_PER_DAY', 10)

    def can_trade(self, current_balance: float) -> tuple[bool, str]:
        """거래 가능 여부 판단"""
        # 일일 손실 한도 초과 여부
        if self.daily_loss >= current_balance * self.max_daily_loss_pct:
            return False, f"일일 손실 한도 초과 ({self.daily_loss:.2f})"

        # 일일 거래 횟수 초과 여부
        if self.trade_count_today >= self.max_trades_per_day:
            return False, f"일일 최대 거래 횟수 초과 ({self.trade_count_today}회)"

        return True, "OK"

    def calculate_position_size(self, balance: float, price: float) -> float:
        """
        켈리 공식 기반 포지션 사이징
        여기서는 단순화하여 고정 비율 사용
        """
        max_amount = balance * self.max_position_pct
        quantity = max_amount / price
        return quantity

    def get_stop_loss_price(self, entry_price: float, side: str) -> float:
        if side == 'buy':
            return entry_price * (1 - self.stop_loss_pct)
        else:
            return entry_price * (1 + self.stop_loss_pct)

    def get_take_profit_price(self, entry_price: float, side: str) -> float:
        if side == 'buy':
            return entry_price * (1 + self.take_profit_pct)
        else:
            return entry_price * (1 - self.take_profit_pct)

    def update_daily_pnl(self, pnl: float):
        """손익 기록"""
        if pnl < 0:
            self.daily_loss += abs(pnl)
        self.trade_count_today += 1

# backtest/backtest.py
import pandas as pd
from dataclasses import dataclass

@dataclass
class TradeResult:
    entry_time: str
    exit_time: str
    side: str
    entry_price: float
    exit_price: float
    pnl: float
    pnl_pct: float

class Backtester:
    def __init__(self, strategy, initial_capital: float = 10_000_000):
        self.strategy = strategy
        self.initial_capital = initial_capital
        self.capital = initial_capital
        self.trades = []

    def run(self, df: pd.DataFrame) -> dict:
        position = None
        entry_price = 0

        for i in range(50, len(df)):  # 충분한 데이터 확보 후 시작
            window = df.iloc[:i]
            signal = self.strategy.generate_signal(window)
            current_price = df.iloc[i]['close']
            current_time = df.iloc[i]['timestamp']

            # 포지션 없을 때 매수 신호
            if position is None and signal == 'BUY':
                position = 'long'
                entry_price = current_price
                amount = self.capital * 0.1 / current_price  # 자금의 10%

            # 포지션 있을 때 매도 신호
            elif position == 'long' and signal == 'SELL':
                pnl = (current_price - entry_price) * amount
                pnl_pct = (current_price - entry_price) / entry_price * 100
                self.capital += pnl
                self.trades.append(TradeResult(
                    entry_time=str(entry_price),
                    exit_time=str(current_time),
                    side='long',
                    entry_price=entry_price,
                    exit_price=current_price,
                    pnl=pnl,
                    pnl_pct=pnl_pct
                ))
                position = None

        return self._calculate_metrics()

    def _calculate_metrics(self) -> dict:
        if not self.trades:
            return {}

        pnls = [t.pnl for t in self.trades]
        wins = [p for p in pnls if p > 0]
        losses = [p for p in pnls if p <= 0]

        total_return = (self.capital - self.initial_capital) / self.initial_capital * 100
        win_rate = len(wins) / len(pnls) * 100
        avg_win = sum(wins) / len(wins) if wins else 0
        avg_loss = sum(losses) / len(losses) if losses else 0
        profit_factor = abs(sum(wins) / sum(losses)) if losses else float('inf')

        return {
            '총 거래 수': len(self.trades),
            '승률': f'{win_rate:.1f}%',
            '총 수익률': f'{total_return:.2f}%',
            '최종 자본': f'{self.capital:,.0f}원',
            '평균 이익': f'{avg_win:,.0f}원',
            '평균 손실': f'{avg_loss:,.0f}원',
            '수익 팩터': f'{profit_factor:.2f}',
        }

# main.py
import schedule
import time
from loguru import logger
from dotenv import load_dotenv
import os

load_dotenv()

from src.exchange import ExchangeConnector
from src.strategy import MovingAverageCrossStrategy
from src.risk_manager import RiskManager
from src.order_manager import OrderManager
from src.notifier import TelegramNotifier

def setup():
    exchange = ExchangeConnector()
    strategy = MovingAverageCrossStrategy(short_window=20, long_window=50)
    risk_manager = RiskManager({
        'MAX_POSITION_PCT': float(os.getenv('MAX_POSITION_PCT', 0.1)),
        'STOP_LOSS_PCT': float(os.getenv('STOP_LOSS_PCT', 0.02)),
        'TAKE_PROFIT_PCT': float(os.getenv('TAKE_PROFIT_PCT', 0.05)),
        'MAX_DAILY_LOSS_PCT': 0.05,
    })
    notifier = TelegramNotifier()
    return exchange, strategy, risk_manager, notifier

def run_bot(exchange, strategy, risk_manager, notifier):
    """봇의 핵심 실행 루프 (1시간마다 호출)"""
    symbol = os.getenv('TRADING_SYMBOL', 'BTC/USDT')
    logger.info(f"[{symbol}] 전략 실행 시작")

    try:
        # 1. 데이터 수집
        df = exchange.get_ohlcv(symbol, timeframe='1h', limit=100)
        if df is None:
            return

        # 2. 잔고 확인
        balance = exchange.get_balance('USDT')

        # 3. 리스크 체크
        can_trade, reason = risk_manager.can_trade(balance)
        if not can_trade:
            logger.warning(f"거래 불가: {reason}")
            return

        # 4. 신호 생성
        signal = strategy.generate_signal(df)
        current_price = df.iloc[-1]['close']
        logger.info(f"신호: {signal}, 현재가: {current_price:,.0f}")

        # 5. 주문 실행
        if signal == 'BUY':
            amount = risk_manager.calculate_position_size(balance, current_price)
            order = exchange.create_market_order(symbol, 'buy', amount)
            sl_price = risk_manager.get_stop_loss_price(current_price, 'buy')
            tp_price = risk_manager.get_take_profit_price(current_price, 'buy')
            notifier.send(
                f"🟢 매수 완료\n"
                f"심볼: {symbol}\n"
                f"체결가: {current_price:,.0f}\n"
                f"수량: {amount:.6f}\n"
                f"손절가: {sl_price:,.0f}\n"
                f"익절가: {tp_price:,.0f}"
            )
        elif signal == 'SELL':
            # 포지션 청산 로직 (생략)
            pass

    except Exception as e:
        logger.error(f"봇 실행 오류: {e}")
        notifier.send(f"⚠️ 봇 오류 발생: {e}")

def main():
    logger.add("logs/bot.log", rotation="1 day", retention="7 days")
    logger.info("트레이딩 봇 시작")

    exchange, strategy, risk_manager, notifier = setup()
    notifier.send("🤖 트레이딩 봇이 시작되었습니다.")

    # 1시간마다 실행
    schedule.every().hour.at(":00").do(
        run_bot, exchange, strategy, risk_manager, notifier
    )

    # 즉시 1회 실행
    run_bot(exchange, strategy, risk_manager, notifier)

    while True:
        schedule.run_pending()
        time.sleep(1)

if __name__ == "__main__":
    main()

# src/notifier.py
import requests
import os

class TelegramNotifier:
    def __init__(self):
        self.token = os.getenv('TELEGRAM_BOT_TOKEN')
        self.chat_id = os.getenv('TELEGRAM_CHAT_ID')
        self.base_url = f"https://api.telegram.org/bot{self.token}"

    def send(self, message: str):
        try:
            url = f"{self.base_url}/sendMessage"
            payload = {
                "chat_id": self.chat_id,
                "text": message,
                "parse_mode": "HTML"
            }
            requests.post(url, json=payload, timeout=10)
        except Exception as e:
            print(f"텔레그램 전송 실패: {e}")

def send_daily_report(exchange, notifier):
    """매일 자정 성과 리포트 발송"""
    balance = exchange.get_balance('USDT')
    # DB에서 오늘 거래 내역 조회 (생략)

    report = f"""
📊 <b>일일 거래 리포트</b>

💰 현재 잔고: {balance:,.2f} USDT
📈 오늘 거래: N회
✅ 익절: N회 | ❌ 손절: N회
📉 오늘 손익: +N USDT (+N%)

🤖 봇 상태: 정상 운영 중
    """
    notifier.send(report)

# 매일 자정에 실행
schedule.every().day.at("00:00").do(send_daily_report, exchange, notifier)

# 수수료를 무시하면 수익 전략이 손실 전략이 됨
거래당 수수료 = 0.1% (매수) + 0.1% (매도) = 0.2%
하루 10번 거래 시 = 2% 수수료 지출
월 20일 = 40% 수수료... 원금 갉아먹기