자동매매 서킷브레이커란?
서킷브레이커(Circuit Breaker)는 자동매매 시스템에서 비정상적인 상황을 감지하면 거래를 자동 중단하는 안전장치입니다. 주식시장의 서킷브레이커에서 차용한 개념으로, 급격한 손실, API 오류, 이상 가격 등이 발생했을 때 봇이 무한히 손실을 키우는 것을 방지합니다.
2022년 테라-루나 폭락, 2024년 엔캐리 청산 등 극단적 시장 이벤트에서 서킷브레이커 없이 운영된 자동매매 봇들이 치명적 손실을 입었습니다. 리스크 관리 시스템의 최후 방어선으로서 서킷브레이커는 선택이 아닌 필수입니다.
서킷브레이커 트리거 유형
효과적인 서킷브레이커는 여러 유형의 이상 상황을 감지해야 합니다.
- 일일 손실 한도: 당일 누적 손실이 설정 금액 또는 비율을 초과하면 즉시 중단합니다.
- 연속 손실 횟수: 연속으로 N번 이상 손절이 발생하면 냉각 시간을 부여합니다.
- 급격한 가격 변동: 1분 내 5% 이상 급등락 시 주문을 보류합니다.
- API 에러율: 최근 주문 중 실패율이 임계값을 넘으면 시스템 점검 모드로 전환합니다.
- 포지션 크기 초과: 총 포지션이 자본 대비 설정 비율을 초과하면 신규 진입을 차단합니다.
- 스프레드 이상: 호가 스프레드가 평소 대비 급격히 벌어지면 유동성 부족으로 판단합니다.
파이썬 서킷브레이커 구현
아래는 여러 트리거를 통합한 서킷브레이커 클래스입니다.
from enum import Enum
from datetime import datetime, timedelta
from collections import deque
class BreakerState(Enum):
CLOSED = "정상" # 거래 가능
OPEN = "차단" # 거래 중단
HALF_OPEN = "시험" # 제한적 거래
class CircuitBreaker:
def __init__(self, config=None):
cfg = config or {}
# 일일 손실 한도 (%)
self.max_daily_loss_pct = cfg.get('max_daily_loss_pct', 5.0)
# 연속 손실 한도
self.max_consecutive_losses = cfg.get('max_consecutive_losses', 5)
# 가격 급변 임계값 (%)
self.price_spike_pct = cfg.get('price_spike_pct', 5.0)
# API 에러율 한도 (%)
self.max_error_rate = cfg.get('max_error_rate', 30.0)
# 냉각 시간 (분)
self.cooldown_min = cfg.get('cooldown_min', 30)
# 반 개방 시 최대 주문 수
self.half_open_max_orders = cfg.get('half_open_max_orders', 3)
self.state = BreakerState.CLOSED
self.trip_reason = None
self.trip_time = None
self.daily_pnl = 0.0
self.starting_capital = 0.0
self.consecutive_losses = 0
self.recent_orders = deque(maxlen=50)
self.half_open_orders = 0
self.price_history = deque(maxlen=120)
def initialize(self, capital):
"""일일 시작 시 초기화"""
self.starting_capital = capital
self.daily_pnl = 0.0
self.consecutive_losses = 0
self.state = BreakerState.CLOSED
self.trip_reason = None
def can_trade(self):
"""거래 가능 여부 확인"""
self._check_auto_recovery()
if self.state == BreakerState.CLOSED:
return True
elif self.state == BreakerState.HALF_OPEN:
return self.half_open_orders < self.half_open_max_orders
return False
def record_trade(self, pnl, success=True):
"""거래 결과 기록"""
self.daily_pnl += pnl
self.recent_orders.append({
'time': datetime.now(),
'pnl': pnl,
'success': success
})
if pnl < 0:
self.consecutive_losses += 1
else:
self.consecutive_losses = 0
if self.state == BreakerState.HALF_OPEN:
self.half_open_orders += 1
if pnl < 0:
self._trip("반개방 중 추가 손실 발생")
elif self.half_open_orders >= self.half_open_max_orders:
self._recover()
self._check_triggers()
def record_price(self, price, timestamp=None):
"""가격 기록 및 급변 감지"""
self.price_history.append({
'price': price,
'time': timestamp or datetime.now()
})
self._check_price_spike()
def record_api_call(self, success):
"""API 호출 결과 기록"""
self.recent_orders.append({
'time': datetime.now(),
'pnl': 0,
'success': success
})
self._check_error_rate()
def _check_triggers(self):
"""모든 트리거 확인"""
# 1. 일일 손실 한도
if self.starting_capital > 0:
loss_pct = abs(self.daily_pnl) / self.starting_capital * 100
if self.daily_pnl < 0 and loss_pct >= self.max_daily_loss_pct:
self._trip(f"일일 손실 한도 초과: -{loss_pct:.1f}%")
return
# 2. 연속 손실
if self.consecutive_losses >= self.max_consecutive_losses:
self._trip(f"연속 {self.consecutive_losses}회 손실")
return
def _check_price_spike(self):
"""가격 급변 감지"""
if len(self.price_history) < 2:
return
recent = self.price_history[-1]['price']
one_min_ago = None
cutoff = datetime.now() - timedelta(minutes=1)
for p in self.price_history:
if p['time'] >= cutoff:
one_min_ago = p['price']
break
if one_min_ago:
change = abs(recent - one_min_ago) / one_min_ago * 100
if change >= self.price_spike_pct:
self._trip(f"1분 내 {change:.1f}% 급변 감지")
def _check_error_rate(self):
"""API 에러율 확인"""
recent = [o for o in self.recent_orders
if o['time'] > datetime.now() - timedelta(minutes=5)]
if len(recent) < 5:
return
errors = sum(1 for o in recent if not o['success'])
rate = errors / len(recent) * 100
if rate >= self.max_error_rate:
self._trip(f"API 에러율 {rate:.0f}% (최근 5분)")
def _trip(self, reason):
"""서킷브레이커 발동"""
self.state = BreakerState.OPEN
self.trip_reason = reason
self.trip_time = datetime.now()
print(f"🚨 서킷브레이커 발동: {reason}")
print(f" {self.cooldown_min}분 냉각 후 반개방")
def _check_auto_recovery(self):
"""냉각 시간 경과 시 반개방 전환"""
if (self.state == BreakerState.OPEN and self.trip_time):
elapsed = (datetime.now() - self.trip_time).seconds / 60
if elapsed >= self.cooldown_min:
self.state = BreakerState.HALF_OPEN
self.half_open_orders = 0
print("⚠️ 반개방 모드: 제한적 거래 시작")
def _recover(self):
"""정상 복구"""
self.state = BreakerState.CLOSED
self.trip_reason = None
self.trip_time = None
print("✅ 서킷브레이커 정상 복구")
def status(self):
"""현재 상태 반환"""
return {
'state': self.state.value,
'daily_pnl': self.daily_pnl,
'consecutive_losses': self.consecutive_losses,
'trip_reason': self.trip_reason,
'can_trade': self.can_trade()
}자동매매 봇에 통합하기
서킷브레이커를 기존 자동매매 봇에 통합하는 패턴입니다.
class TradingBot:
def __init__(self, exchange, strategy, capital):
self.exchange = exchange
self.strategy = strategy
self.breaker = CircuitBreaker({
'max_daily_loss_pct': 3.0,
'max_consecutive_losses': 4,
'price_spike_pct': 5.0,
'cooldown_min': 30
})
self.breaker.initialize(capital)
def run_tick(self, market_data):
"""매 틱마다 실행"""
# 1. 가격 기록
self.breaker.record_price(market_data['price'])
# 2. 거래 가능 여부 확인
if not self.breaker.can_trade():
status = self.breaker.status()
print(f"거래 차단: {status['trip_reason']} "
f"(상태: {status['state']})")
return
# 3. 전략 시그널 생성
signal = self.strategy.generate_signal(market_data)
if not signal:
return
# 4. 주문 실행
try:
order = self.exchange.create_order(
signal['symbol'], signal['side'],
signal['qty'], signal['price']
)
self.breaker.record_api_call(success=True)
# 체결 후 PnL 계산 및 기록
pnl = self.calculate_pnl(order)
self.breaker.record_trade(pnl, success=True)
except Exception as e:
self.breaker.record_api_call(success=False)
print(f"주문 실패: {e}")서킷브레이커 설정 가이드
전략 유형별 권장 서킷브레이커 설정값입니다.
- 스캘핑 전략: 일일 손실 한도 2~3%, 연속 손실 3~5회, 냉각 시간 15분. 빈번한 거래 특성상 타이트한 설정이 필요합니다.
- 스윙 트레이딩: 일일 손실 한도 5~7%, 연속 손실 3~4회, 냉각 시간 60분. 포지션 보유 기간이 길어 여유 있는 설정이 적합합니다.
- 차익거래: 일일 손실 한도 1~2%, API 에러율 20%, 냉각 시간 10분. 무위험에 가까운 전략이므로 손실 자체가 이상 신호입니다.
- 모멘텀 전략: 일일 손실 한도 4~5%, 가격 급변 임계값 3%, 냉각 시간 30분. 변동성 장세에서도 안정적으로 운영할 수 있습니다.
알림 시스템 연동
서킷브레이커가 발동되면 즉시 트레이더에게 알림을 보내야 합니다.
import requests
def send_telegram_alert(bot_token, chat_id, breaker):
"""서킷브레이커 발동 시 텔레그램 알림"""
status = breaker.status()
msg = (
f"🚨 서킷브레이커 발동!nn"
f"사유: {status['trip_reason']}n"
f"상태: {status['state']}n"
f"일일 손익: ${status['daily_pnl']:+,.2f}n"
f"연속 손실: {status['consecutive_losses']}회nn"
f"냉각 시간 경과 후 반개방 모드로 전환됩니다."
)
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
requests.post(url, json={
'chat_id': chat_id,
'text': msg,
'parse_mode': 'HTML'
})마무리
서킷브레이커는 자동매매 시스템의 생존을 결정짓는 핵심 모듈입니다. 아무리 뛰어난 전략이라도 블랙 스완 이벤트에서 계좌를 지키지 못하면 의미가 없습니다. 일일 손실 한도, 연속 손실 감지, 가격 급변 탐지, API 에러율 모니터링 등 다중 방어선을 구축하고, 냉각 시간과 반개방 모드로 점진적 복구 메커니즘을 갖추는 것이 실전 운영의 핵심입니다. 오늘 구현한 서킷브레이커를 여러분의 자동매매 봇에 반드시 적용하시기 바랍니다.