자동매매 봇에 실시간 알림이 필요한 이유
자동매매 봇을 24시간 운영하면 체결, 손절, 에러 등 중요한 이벤트를 놓치기 쉽습니다. 텔레그램 알림 시스템을 연동하면 어디서든 스마트폰으로 봇 상태를 실시간 모니터링할 수 있습니다.
이 글에서는 파이썬 자동매매 봇에 텔레그램 알림을 연동하는 방법을 실전 코드와 함께 단계별로 설명합니다. 단순 메시지 전송부터 구조화된 알림 시스템 설계까지 다룹니다.
텔레그램 봇 생성 및 설정
먼저 텔레그램 봇을 생성하고 API 토큰을 발급받아야 합니다.
1단계: BotFather로 봇 생성
텔레그램에서 @BotFather를 검색하고, /newbot 명령어로 봇을 생성합니다. 봇 이름과 사용자명을 설정하면 API 토큰이 발급됩니다.
2단계: Chat ID 확인
알림을 받을 채팅방의 Chat ID가 필요합니다. 봇에 메시지를 보낸 후 아래 URL로 확인할 수 있습니다.
https://api.telegram.org/bot{YOUR_TOKEN}/getUpdates파이썬 텔레그램 알림 클래스 구현
아래는 자동매매 봇에 바로 통합할 수 있는 알림 클래스입니다.
import requests
import time
from datetime import datetime
from enum import Enum
class AlertLevel(Enum):
INFO = "ℹ️"
SUCCESS = "✅"
WARNING = "⚠️"
ERROR = "🚨"
TRADE = "💰"
class TelegramNotifier:
"""자동매매 봇용 텔레그램 알림 시스템"""
BASE_URL = "https://api.telegram.org/bot{token}"
def __init__(self, token, chat_id, bot_name="MyBot"):
self.token = token
self.chat_id = chat_id
self.bot_name = bot_name
self.base_url = self.BASE_URL.format(token=token)
self._rate_limit = {}
self.MIN_INTERVAL = 1 # 최소 전송 간격(초)
def send(self, message, level=AlertLevel.INFO, parse_mode="HTML"):
"""메시지 전송 (레이트 리밋 적용)"""
now = time.time()
if now - self._rate_limit.get('last', 0) < self.MIN_INTERVAL:
time.sleep(self.MIN_INTERVAL)
header = f"{level.value} [{self.bot_name}]"
timestamp = datetime.now().strftime("%H:%M:%S")
full_msg = f"{header}n{message}n{timestamp}"
try:
resp = requests.post(
f"{self.base_url}/sendMessage",
json={
"chat_id": self.chat_id,
"text": full_msg,
"parse_mode": parse_mode
},
timeout=10
)
self._rate_limit['last'] = time.time()
return resp.json().get('ok', False)
except Exception as e:
print(f"텔레그램 전송 실패: {e}")
return False매매 이벤트별 알림 템플릿
자동매매에서 자주 발생하는 이벤트별 알림 메서드를 구현합니다. 각 이벤트에 맞는 구조화된 메시지를 보내면 스마트폰에서 한눈에 상황을 파악할 수 있습니다.
class TradingAlerts(TelegramNotifier):
"""매매 특화 알림 메서드 모음"""
def order_filled(self, symbol, side, qty, price, pnl=None):
"""주문 체결 알림"""
emoji = "🟢" if side == "buy" else "🔴"
msg = (
f"{emoji} 주문 체결n"
f"종목: {symbol}n"
f"방향: {side.upper()} | 수량: {qty}n"
f"체결가: {price:,.2f}"
)
if pnl is not None:
pnl_emoji = "📈" if pnl >= 0 else "📉"
msg += f"n{pnl_emoji} 손익: {pnl:+,.2f}"
self.send(msg, AlertLevel.TRADE)
def stop_loss_triggered(self, symbol, entry_price, exit_price, loss):
"""손절 발동 알림"""
loss_pct = (exit_price - entry_price) / entry_price * 100
msg = (
f"🛑 손절 발동n"
f"종목: {symbol}n"
f"진입가: {entry_price:,.2f} → 손절가: {exit_price:,.2f}n"
f"손실: {loss_pct:+.2f}% ({loss:+,.2f})"
)
self.send(msg, AlertLevel.WARNING)
def daily_summary(self, total_trades, win_rate, daily_pnl, balance):
"""일간 성과 요약"""
msg = (
f"📊 일간 리포트n"
f"총 거래: {total_trades}건n"
f"승률: {win_rate:.1f}%n"
f"일간 손익: {daily_pnl:+,.2f}n"
f"잔고: {balance:,.2f}"
)
self.send(msg, AlertLevel.INFO)
def error_alert(self, error_type, detail):
"""에러 알림"""
msg = (
f"에러 발생n"
f"유형: {error_type}n"
f"상세: {detail[:200]}"
)
self.send(msg, AlertLevel.ERROR)
def bot_status(self, status, uptime_hours, open_positions):
"""봇 상태 알림"""
msg = (
f"🤖 봇 상태: {status}n"
f"가동 시간: {uptime_hours:.1f}시간n"
f"열린 포지션: {open_positions}개"
)
self.send(msg, AlertLevel.INFO)레이트 리밋과 알림 큐 관리
텔레그램 Bot API는 초당 약 30건의 메시지 제한이 있습니다. 자동매매에서 급격한 변동 시 대량 알림이 발생할 수 있으므로, 알림 큐와 중복 필터링을 구현해야 합니다.
from collections import deque
from threading import Thread, Lock
import hashlib
class QueuedNotifier(TradingAlerts):
"""큐 기반 알림: 레이트 리밋 준수 + 중복 제거"""
def __init__(self, *args, max_per_minute=20, **kwargs):
super().__init__(*args, **kwargs)
self.queue = deque(maxlen=100)
self.lock = Lock()
self.max_per_minute = max_per_minute
self.sent_hashes = {} # 중복 방지용
self.DEDUP_WINDOW = 300 # 5분 내 동일 메시지 차단
self._start_worker()
def _msg_hash(self, message):
return hashlib.md5(message.encode()).hexdigest()[:8]
def send(self, message, level=AlertLevel.INFO, **kwargs):
"""큐에 추가 (중복 체크 후)"""
h = self._msg_hash(message)
now = time.time()
# 5분 내 동일 메시지 필터링
if h in self.sent_hashes:
if now - self.sent_hashes[h] < self.DEDUP_WINDOW:
return # 중복 → 무시
with self.lock:
self.queue.append((message, level, kwargs))
def _start_worker(self):
"""백그라운드 워커: 큐에서 메시지를 꺼내 전송"""
def worker():
sent_count = 0
minute_start = time.time()
while True:
# 분당 제한 리셋
if time.time() - minute_start > 60:
sent_count = 0
minute_start = time.time()
if self.queue and sent_count < self.max_per_minute:
with self.lock:
msg, level, kwargs = self.queue.popleft()
super(QueuedNotifier, self).send(msg, level, **kwargs)
self.sent_hashes[self._msg_hash(msg)] = time.time()
sent_count += 1
else:
time.sleep(0.5)
t = Thread(target=worker, daemon=True)
t.start()자동매매 봇과 통합하기
실제 자동매매 봇에 알림을 통합하는 예시입니다.
import ccxt
# 텔레그램 알림 초기화
alerts = QueuedNotifier(
token="YOUR_BOT_TOKEN",
chat_id="YOUR_CHAT_ID",
bot_name="BTC-Scalper"
)
# 거래소 연결
exchange = ccxt.binance({
'apiKey': 'KEY', 'secret': 'SECRET'
})
def execute_trade(symbol, side, amount):
"""매매 실행 + 알림"""
try:
order = exchange.create_market_order(symbol, side, amount)
price = order.get('average', 0)
# 체결 알림
alerts.order_filled(symbol, side, amount, price)
return order
except ccxt.InsufficientFunds as e:
alerts.error_alert("잔고 부족", str(e))
except ccxt.NetworkError as e:
alerts.error_alert("네트워크 오류", str(e))
except Exception as e:
alerts.error_alert("알 수 없는 오류", str(e))
return None
# 봇 시작 알림
alerts.send("봇 가동을 시작합니다.", AlertLevel.SUCCESS)
# 예시: 전략 루프
while True:
signal = get_trading_signal() # 사용자 정의 전략
if signal == "buy":
execute_trade("BTC/USDT", "buy", 0.001)
elif signal == "sell":
execute_trade("BTC/USDT", "sell", 0.001)
time.sleep(60)알림 레벨별 필터링 설정
모든 알림을 다 받으면 피로감이 생깁니다. 알림 레벨에 따라 필터링하면 중요한 알림만 받을 수 있습니다.
| 레벨 | 용도 | 권장 빈도 |
|---|---|---|
| 🚨 ERROR | API 오류, 봇 중단 | 즉시 전송 |
| ⚠️ WARNING | 손절, 큰 낙폭 | 즉시 전송 |
| 💰 TRADE | 주문 체결 | 건별 또는 묶음 |
| ℹ️ INFO | 일간 리포트, 상태 | 1일 1~2회 |
장애 상황 대비: 폴백 알림
텔레그램 API 자체가 다운될 수도 있습니다. 중요한 알림은 폴백 채널(이메일, 디스코드 웹훅 등)을 함께 설정하세요.
class MultiChannelNotifier:
"""다중 채널 폴백 알림"""
def __init__(self, primary, fallbacks=None):
self.primary = primary
self.fallbacks = fallbacks or []
def send(self, message, level=AlertLevel.INFO):
success = self.primary.send(message, level)
if not success:
for fb in self.fallbacks:
if fb.send(message, level):
break # 하나라도 성공하면 중단마치며
텔레그램 알림은 자동매매 봇 운영에서 선택이 아닌 필수입니다. 체결 알림, 손절 알림, 에러 알림을 체계적으로 설정하면 봇을 떠나있어도 안심할 수 있습니다. 자동매매 로그 분석과 함께 운영하면 더욱 견고한 트레이딩 시스템을 구축할 수 있습니다.