그리드 트레이딩 봇 구현

그리드 트레이딩이란?

그리드 트레이딩(Grid Trading)은 일정 가격 간격으로 매수·매도 주문을 격자(grid)처럼 배치해 횡보장에서 꾸준히 수익을 내는 자동매매 전략입니다. 가격이 오르면 미리 걸어둔 매도 주문이 체결되고, 내리면 매수 주문이 체결되면서 반복적으로 차익을 실현합니다.

방향 예측이 필요 없어 초보자도 접근하기 쉽고, 24시간 운영되는 암호화폐 시장에서 특히 인기 있는 전략입니다. 바이낸스, 업비트 등 주요 거래소에서 그리드 봇 기능을 기본 제공할 정도로 대중화되었습니다.

그리드 전략의 핵심 파라미터

파라미터	설명	권장 범위
상한가(Upper)	그리드 최상단 가격	저항선 기준
하한가(Lower)	그리드 최하단 가격	지지선 기준
그리드 수(N)	격자 개수	10~50개
그리드 간격	등차(산술) 또는 등비(기하)	변동성에 따라 선택
주문당 금액	각 그리드 레벨의 투자금	총 자본 / 그리드 수

등차 vs 등비 그리드

그리드 간격을 어떻게 설정하느냐에 따라 전략 특성이 달라집니다.

import numpy as np

def arithmetic_grid(lower: float, upper: float,
                    num_grids: int) -> list:
    """등차 그리드: 동일한 가격 간격"""
    return np.linspace(lower, upper, num_grids + 1).tolist()

def geometric_grid(lower: float, upper: float,
                   num_grids: int) -> list:
    """등비 그리드: 동일한 % 간격 — 넓은 범위에 적합"""
    ratio = (upper / lower) ** (1 / num_grids)
    return [lower * (ratio ** i) for i in range(num_grids + 1)]

# 예시: BTC 80,000~100,000 구간, 10개 그리드
arith = arithmetic_grid(80000, 100000, 10)
geo = geometric_grid(80000, 100000, 10)

# 등차: [80000, 82000, 84000, ..., 100000] 간격 2000
# 등비: [80000, 81823, 83690, ..., 100000] 간격 점점 증가

등비 그리드는 낮은 가격대에서 더 촘촘하게, 높은 가격대에서 더 넓게 배치되어 퍼센트 기준 수익률이 균일합니다. 변동성이 큰 암호화폐에서는 등비 그리드가 더 효율적입니다.

파이썬 그리드 봇 핵심 엔진

파이썬 자동매매 시스템에 통합할 수 있는 그리드 봇 엔진입니다.

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
import time

class Side(Enum):
    BUY = "buy"
    SELL = "sell"

@dataclass
class GridOrder:
    level: int
    price: float
    side: Side
    qty: float
    order_id: Optional[str] = None
    filled: bool = False

class GridBot:
    def __init__(self, exchange, symbol: str,
                 lower: float, upper: float,
                 num_grids: int, total_capital: float,
                 grid_type: str = "geometric"):
        self.exchange = exchange
        self.symbol = symbol
        self.lower = lower
        self.upper = upper
        self.num_grids = num_grids
        self.total_capital = total_capital
        self.grid_type = grid_type

        self.grid_levels = self._build_grid()
        self.qty_per_grid = total_capital / num_grids
        self.orders: list[GridOrder] = []
        self.realized_pnl = 0.0
        self.trade_count = 0

    def _build_grid(self) -> list:
        if self.grid_type == "arithmetic":
            return arithmetic_grid(
                self.lower, self.upper, self.num_grids)
        return geometric_grid(
            self.lower, self.upper, self.num_grids)

    def initialize(self):
        """현재가 기준 초기 주문 배치"""
        current_price = self.exchange.get_price(self.symbol)

        for i, level_price in enumerate(self.grid_levels):
            qty = self.qty_per_grid / level_price

            if level_price < current_price:
                # 현재가 아래 → 매수 주문
                order = GridOrder(
                    level=i, price=level_price,
                    side=Side.BUY, qty=qty)
            elif level_price > current_price:
                # 현재가 위 → 매도 주문
                order = GridOrder(
                    level=i, price=level_price,
                    side=Side.SELL, qty=qty)
            else:
                continue

            order.order_id = self.exchange.create_limit_order(
                self.symbol, order.side.value,
                order.qty, order.price)
            self.orders.append(order)

        print(f"그리드 초기화 완료: {len(self.orders)}개 주문")

    def check_and_reorder(self):
        """체결된 주문 확인 후 반대 주문 생성"""
        for order in self.orders:
            if order.filled:
                continue

            status = self.exchange.get_order_status(
                order.order_id)
            if status != 'filled':
                continue

            order.filled = True
            self.trade_count += 1

            # 반대 방향 주문 생성
            if order.side == Side.BUY:
                # 매수 체결 → 한 단계 위에서 매도
                next_level = order.level + 1
                if next_level < len(self.grid_levels):
                    sell_price = self.grid_levels[next_level]
                    new_order = GridOrder(
                        level=next_level,
                        price=sell_price,
                        side=Side.SELL,
                        qty=order.qty)
                    profit = (sell_price - order.price) * order.qty
                    self.realized_pnl += profit
                    print(f"매수 체결 @{order.price:.2f} → "
                          f"매도 주문 @{sell_price:.2f} "
                          f"(예상 수익: ${profit:.2f})")
            else:
                # 매도 체결 → 한 단계 아래에서 매수
                next_level = order.level - 1
                if next_level >= 0:
                    buy_price = self.grid_levels[next_level]
                    new_order = GridOrder(
                        level=next_level,
                        price=buy_price,
                        side=Side.BUY,
                        qty=order.qty)
                    print(f"매도 체결 @{order.price:.2f} → "
                          f"매수 주문 @{buy_price:.2f}")

            if 'new_order' in locals() and new_order:
                new_order.order_id = 
                    self.exchange.create_limit_order(
                        self.symbol, new_order.side.value,
                        new_order.qty, new_order.price)
                self.orders.append(new_order)

    def status(self) -> dict:
        """봇 상태 조회"""
        return {
            'total_trades': self.trade_count,
            'realized_pnl': f"${self.realized_pnl:.2f}",
            'active_orders': sum(
                1 for o in self.orders if not o.filled),
            'grid_range': f"{self.lower}~{self.upper}",
            'grid_count': self.num_grids
        }

동적 그리드: 시장에 맞춰 자동 조절

고정 그리드의 한계는 가격이 범위를 벗어나면 작동을 멈춘다는 점입니다. 동적 그리드는 이를 해결합니다.

class DynamicGridBot(GridBot):
    def __init__(self, *args, atr_period=14,
                 atr_multiplier=3.0, **kwargs):
        super().__init__(*args, **kwargs)
        self.atr_period = atr_period
        self.atr_multiplier = atr_multiplier

    def calculate_range(self, prices) -> tuple:
        """ATR 기반 동적 그리드 범위 계산"""
        high = prices['high'].tail(self.atr_period)
        low = prices['low'].tail(self.atr_period)
        close = prices['close'].tail(self.atr_period)

        tr = np.maximum(
            high - low,
            np.maximum(
                abs(high - close.shift(1)),
                abs(low - close.shift(1))
            )
        )
        atr = tr.mean()
        current = close.iloc[-1]

        new_lower = current - atr * self.atr_multiplier
        new_upper = current + atr * self.atr_multiplier
        return new_lower, new_upper

    def rebalance_grid(self, prices):
        """가격 범위 이탈 시 그리드 재구성"""
        current = prices['close'].iloc[-1]

        if current <= self.lower or current >= self.upper:
            # 기존 주문 전부 취소
            for order in self.orders:
                if not order.filled:
                    self.exchange.cancel_order(order.order_id)

            # ATR 기반 새 범위 설정
            self.lower, self.upper = 
                self.calculate_range(prices)
            self.grid_levels = self._build_grid()
            self.orders = []
            self.initialize()
            print(f"그리드 재구성: {self.lower:.0f}~"
                  f"{self.upper:.0f}")

수익성 분석: 그리드 전략은 언제 유리한가?

시장 상황	그리드 성과	이유
횡보장 (레인지)	⭐ 최적	매수·매도 반복 체결로 꾸준한 수익
완만한 상승장	✅ 양호	매수→매도 체결 + 보유 자산 평가익
급등장	⚠️ 보통	일찍 매도해 기회비용 발생
완만한 하락장	⚠️ 주의	매수만 반복, 평가손 누적
급락장	❌ 위험	하한 돌파 시 대량 미실현 손실

리스크 관리 필수 설정

서킷브레이커와 결합해 급락장 방어선을 구축하는 것이 중요합니다.

• 손절 라인: 하한가 아래 5~10%에 전체 포지션 손절 설정. 그리드 하단 돌파 시 추가 하락 리스크가 큽니다.
• 자본 분배: 전체 자본의 50~70%만 그리드에 투입. 나머지는 그리드 재구성이나 기회 포착용 예비금으로 보유합니다.
• 그리드 수 조절: 그리드가 너무 촘촘하면 수수료 대비 수익이 작고, 너무 넓으면 체결 빈도가 낮습니다. 수수료의 3~5배 이상 간격을 권장합니다.
• 변동성 모니터링: 변동성이 급등하면 그리드 범위를 넓히고, 급감하면 좁혀서 효율을 유지합니다.
• 최대 보유량 제한: 한 방향 매수가 과도하게 누적되지 않도록 총 포지션 상한을 설정하세요.

거래소 API 연동 예시

import ccxt

def run_grid_bot():
    """ccxt 기반 그리드 봇 실행"""
    exchange = ccxt.binance({
        'apiKey': 'YOUR_API_KEY',
        'secret': 'YOUR_SECRET',
        'options': {'defaultType': 'spot'}
    })

    bot = DynamicGridBot(
        exchange=exchange,
        symbol='BTC/USDT',
        lower=78000,
        upper=95000,
        num_grids=20,
        total_capital=10000,
        grid_type='geometric'
    )

    bot.initialize()

    while True:
        try:
            bot.check_and_reorder()
            status = bot.status()
            print(f"거래: {status['total_trades']}건 | "
                  f"수익: {status['realized_pnl']}")
            time.sleep(10)  # 10초 간격 체크
        except Exception as e:
            print(f"에러: {e}")
            time.sleep(60)

마무리

그리드 트레이딩은 방향 예측 없이 가격 변동 자체에서 수익을 추출하는 체계적인 자동매매 전략입니다. 등비 그리드로 효율을 높이고, ATR 기반 동적 범위 조절로 시장 변화에 적응하며, 엄격한 리스크 관리로 급락 리스크를 통제하면 안정적인 자동매매 시스템을 구축할 수 있습니다. 파이썬과 ccxt 라이브러리를 활용하면 주요 거래소 대부분에 즉시 배포할 수 있습니다.