자동매매 리스크 관리 시스템

자동매매 리스크 관리가 중요한 이유

아무리 높은 승률의 전략이라도 리스크 관리 없이는 한 번의 큰 손실로 계좌가 파괴될 수 있습니다. 자동매매 봇은 24시간 쉬지 않고 매매하기 때문에, 사람이 개입할 틈 없이 연속 손실이 발생할 수 있습니다. 체계적인 리스크 관리 시스템은 수익을 극대화하는 것이 아니라 생존 확률을 높이는 핵심 장치입니다.

이 글에서는 파이썬으로 자동매매 봇에 적용할 수 있는 다계층 리스크 관리 시스템을 구현합니다.

리스크 관리 4계층 구조

실전 자동매매에서는 단일 손절만으로는 부족합니다. 4계층 방어선을 구축해야 합니다.

계층	역할	예시
1. 주문 단위	개별 포지션 손절	ATR 기반 스톱로스
2. 일간 한도	하루 최대 손실 제한	계좌의 3% 초과 시 매매 중단
3. 포트폴리오	전체 노출도 관리	상관관계 높은 포지션 제한
4. 비상 차단	시스템 레벨 킬스위치	MDD 초과 시 전체 청산

1계층: 주문 단위 리스크 관리

가장 기본적인 방어선은 개별 주문의 손절과 포지션 크기를 제어하는 것입니다.

class OrderRiskManager:
    """주문 단위 리스크 관리"""

    def __init__(self, account_balance, risk_per_trade=0.01):
        self.balance = account_balance
        self.risk_per_trade = risk_per_trade  # 1건당 최대 1% 리스크

    def calculate_position_size(self, entry_price, stop_loss_price):
        """리스크 기반 포지션 사이징"""
        risk_amount = self.balance * self.risk_per_trade
        price_risk = abs(entry_price - stop_loss_price)

        if price_risk == 0:
            return 0

        position_size = risk_amount / price_risk
        return round(position_size, 6)

    def validate_order(self, symbol, side, qty, entry, stop_loss):
        """주문 전 리스크 검증"""
        max_loss = qty * abs(entry - stop_loss)
        max_loss_pct = max_loss / self.balance * 100

        if max_loss_pct > self.risk_per_trade * 100:
            return {
                'approved': False,
                'reason': f'리스크 초과: {max_loss_pct:.2f}% > {self.risk_per_trade*100}%'
            }

        return {
            'approved': True,
            'max_loss': max_loss,
            'max_loss_pct': max_loss_pct
        }

2계층: 일간 손실 한도

하루에 연속으로 손절이 발생하면 감정적으로 대응하기 어렵습니다. 자동매매 봇도 마찬가지로, 일간 최대 손실에 도달하면 자동으로 매매를 중단해야 합니다.

from datetime import datetime, timedelta

class DailyRiskManager:
    """일간 리스크 관리: 최대 손실 및 거래 횟수 제한"""

    def __init__(self, max_daily_loss_pct=3.0, max_daily_trades=20,
                 max_consecutive_losses=5):
        self.max_daily_loss_pct = max_daily_loss_pct
        self.max_daily_trades = max_daily_trades
        self.max_consecutive_losses = max_consecutive_losses
        self.daily_pnl = 0.0
        self.daily_trades = 0
        self.consecutive_losses = 0
        self.last_reset = datetime.now().date()
        self.is_locked = False

    def _check_reset(self):
        """날짜 변경 시 일간 카운터 리셋"""
        today = datetime.now().date()
        if today > self.last_reset:
            self.daily_pnl = 0.0
            self.daily_trades = 0
            self.consecutive_losses = 0
            self.is_locked = False
            self.last_reset = today

    def record_trade(self, pnl, balance):
        """거래 결과 기록 + 한도 체크"""
        self._check_reset()
        self.daily_pnl += pnl
        self.daily_trades += 1

        if pnl < 0:
            self.consecutive_losses += 1
        else:
            self.consecutive_losses = 0

        # 잠금 조건 체크
        daily_loss_pct = abs(self.daily_pnl) / balance * 100
        if self.daily_pnl < 0 and daily_loss_pct >= self.max_daily_loss_pct:
            self.is_locked = True
            return {'locked': True, 'reason': f'일간 손실 한도 도달: -{daily_loss_pct:.1f}%'}

        if self.daily_trades >= self.max_daily_trades:
            self.is_locked = True
            return {'locked': True, 'reason': f'일간 거래 횟수 초과: {self.daily_trades}회'}

        if self.consecutive_losses >= self.max_consecutive_losses:
            self.is_locked = True
            return {'locked': True, 'reason': f'연속 손실 {self.consecutive_losses}회'}

        return {'locked': False}

    def can_trade(self):
        """매매 가능 여부"""
        self._check_reset()
        return not self.is_locked

3계층: 포트폴리오 노출도 관리

여러 종목을 동시에 매매하는 경우, 전체 포트폴리오의 리스크를 관리해야 합니다. 비트코인과 이더리움처럼 상관관계가 높은 자산에 동시 롱 포지션을 잡으면 실질적으로 레버리지를 높이는 것과 같습니다.

class PortfolioRiskManager:
    """포트폴리오 레벨 리스크 관리"""

    def __init__(self, max_total_exposure=0.5, max_per_asset=0.15,
                 max_correlated_exposure=0.25):
        self.max_total_exposure = max_total_exposure  # 총 자산의 50%
        self.max_per_asset = max_per_asset            # 자산당 15%
        self.max_correlated = max_correlated_exposure  # 상관자산 25%
        self.positions = {}

    def add_position(self, symbol, size_usd, group=None):
        """포지션 등록"""
        self.positions[symbol] = {
            'size': size_usd,
            'group': group or symbol  # 상관 그룹
        }

    def check_new_position(self, symbol, size_usd, balance, group=None):
        """신규 포지션 리스크 체크"""
        group = group or symbol
        current_total = sum(p['size'] for p in self.positions.values())
        current_group = sum(
            p['size'] for p in self.positions.values()
            if p['group'] == group
        )

        # 전체 노출도 체크
        new_total = (current_total + size_usd) / balance
        if new_total > self.max_total_exposure:
            return {
                'approved': False,
                'reason': f'전체 노출도 초과: {new_total:.0%} > {self.max_total_exposure:.0%}'
            }

        # 단일 자산 집중도 체크
        asset_exposure = size_usd / balance
        if asset_exposure > self.max_per_asset:
            return {
                'approved': False,
                'reason': f'단일 자산 초과: {asset_exposure:.0%} > {self.max_per_asset:.0%}'
            }

        # 상관 그룹 노출도 체크
        new_group = (current_group + size_usd) / balance
        if new_group > self.max_correlated:
            return {
                'approved': False,
                'reason': f'상관 그룹 초과: {new_group:.0%} > {self.max_correlated:.0%}'
            }

        return {'approved': True}

4계층: 비상 킬스위치

최후의 방어선은 킬스위치입니다. MDD(최대낙폭)가 임계값을 넘으면 모든 포지션을 청산하고 봇을 정지합니다.

class KillSwitch:
    """비상 정지 시스템"""

    def __init__(self, max_mdd_pct=15.0, initial_balance=None):
        self.max_mdd_pct = max_mdd_pct
        self.peak_balance = initial_balance or 0
        self.is_triggered = False

    def update(self, current_balance):
        """잔고 업데이트 + 킬스위치 체크"""
        if current_balance > self.peak_balance:
            self.peak_balance = current_balance

        if self.peak_balance == 0:
            return {'triggered': False}

        drawdown = (self.peak_balance - current_balance) / self.peak_balance * 100

        if drawdown >= self.max_mdd_pct:
            self.is_triggered = True
            return {
                'triggered': True,
                'drawdown': drawdown,
                'peak': self.peak_balance,
                'current': current_balance,
                'action': 'EMERGENCY_CLOSE_ALL'
            }

        return {
            'triggered': False,
            'drawdown': drawdown,
            'remaining': self.max_mdd_pct - drawdown
        }

    def emergency_close(self, exchange, positions):
        """전체 포지션 긴급 청산"""
        results = []
        for pos in positions:
            try:
                side = 'sell' if pos['side'] == 'long' else 'buy'
                order = exchange.create_market_order(
                    pos['symbol'], side, pos['amount']
                )
                results.append({'symbol': pos['symbol'], 'status': 'closed'})
            except Exception as e:
                results.append({'symbol': pos['symbol'], 'error': str(e)})
        return results

통합 리스크 관리 시스템

4개 계층을 하나로 통합한 리스크 게이트웨이입니다. 모든 주문은 이 게이트웨이를 거쳐야 합니다.

class RiskGateway:
    """통합 리스크 관리 게이트웨이"""

    def __init__(self, balance, config=None):
        cfg = config or {}
        self.balance = balance
        self.order_rm = OrderRiskManager(balance, cfg.get('risk_per_trade', 0.01))
        self.daily_rm = DailyRiskManager(
            max_daily_loss_pct=cfg.get('max_daily_loss', 3.0),
            max_consecutive_losses=cfg.get('max_consec_loss', 5)
        )
        self.portfolio_rm = PortfolioRiskManager(
            max_total_exposure=cfg.get('max_exposure', 0.5)
        )
        self.kill_switch = KillSwitch(
            max_mdd_pct=cfg.get('max_mdd', 15.0),
            initial_balance=balance
        )

    def approve_order(self, symbol, side, entry, stop_loss, group=None):
        """주문 승인 프로세스: 4계층 순차 검증"""

        # 4계층: 킬스위치 확인
        ks = self.kill_switch.update(self.balance)
        if ks['triggered']:
            return {'approved': False, 'layer': 4, 'reason': '킬스위치 발동'}

        # 2계층: 일간 한도 확인
        if not self.daily_rm.can_trade():
            return {'approved': False, 'layer': 2, 'reason': '일간 한도 도달'}

        # 1계층: 포지션 사이즈 계산
        qty = self.order_rm.calculate_position_size(entry, stop_loss)
        validation = self.order_rm.validate_order(
            symbol, side, qty, entry, stop_loss
        )
        if not validation['approved']:
            return {'approved': False, 'layer': 1, 'reason': validation['reason']}

        # 3계층: 포트폴리오 노출도
        size_usd = qty * entry
        port_check = self.portfolio_rm.check_new_position(
            symbol, size_usd, self.balance, group
        )
        if not port_check['approved']:
            return {'approved': False, 'layer': 3, 'reason': port_check['reason']}

        return {
            'approved': True,
            'qty': qty,
            'max_loss': validation['max_loss'],
            'max_loss_pct': validation['max_loss_pct']
        }

# 사용 예시
gateway = RiskGateway(balance=10000, config={
    'risk_per_trade': 0.01,
    'max_daily_loss': 3.0,
    'max_mdd': 15.0
})

result = gateway.approve_order(
    symbol='BTC/USDT',
    side='buy',
    entry=65000,
    stop_loss=64000,
    group='crypto_major'
)
print(result)
# {'approved': True, 'qty': 0.1, 'max_loss': 100, 'max_loss_pct': 1.0}

리스크 파라미터 설정 가이드

파라미터	보수적	중립	공격적
건당 리스크	0.5%	1.0%	2.0%
일간 최대 손실	2%	3%	5%
최대 MDD	10%	15%	25%
총 노출도	30%	50%	80%
연속 손실 한도	3회	5회	8회

초보자는 보수적 설정으로 시작하여 충분한 백테스트 후 점진적으로 조정하는 것을 권장합니다.

마치며

자동매매에서 리스크 관리는 선택이 아니라 생존의 문제입니다. 4계층 방어선 — 주문 단위 손절, 일간 한도, 포트폴리오 노출도, 킬스위치 — 를 모두 갖춘 봇만이 장기적으로 살아남습니다. 포지션 사이징 전략과 함께 적용하면 더욱 견고한 트레이딩 시스템을 구축할 수 있습니다.