"""Telegram 매매 알림.""" from __future__ import annotations import logging import os import requests logger = logging.getLogger(__name__) _API = "https://api.telegram.org/bot{token}/sendMessage" def _send(text: str) -> None: token = os.getenv("TELEGRAM_TRADE_TOKEN", "") chat_id = os.getenv("TELEGRAM_CHAT_ID", "") if not token or not chat_id: logger.warning("Telegram 설정 없음, 알림 스킵") return try: requests.post( _API.format(token=token), json={"chat_id": chat_id, "text": text, "parse_mode": "HTML"}, timeout=5, ) except Exception as e: logger.error(f"Telegram 알림 실패: {e}") def notify_buy(ticker: str, price: float, amount: float, invested_krw: int) -> None: _send( f"📈 [매수] {ticker}\n" f"가격: {price:,.0f}원\n" f"수량: {amount}\n" f"투자금: {invested_krw:,}원" ) def notify_sell(ticker: str, price: float, pnl_pct: float, reason: str) -> None: emoji = "✅" if pnl_pct >= 0 else "🔴" _send( f"{emoji} [매도] {ticker}\n" f"가격: {price:,.0f}원\n" f"수익률: {pnl_pct:+.1f}%\n" f"사유: {reason}" ) def notify_error(message: str) -> None: _send(f"⚠️ [오류]\n{message}") def notify_status(positions: dict) -> None: """1시간마다 포지션 현황 요약 전송.""" from datetime import datetime import pyupbit now = datetime.now().strftime("%H:%M") if not positions: _send(f"📊 [{now} 현황]\n보유 포지션 없음 — 매수 신호 대기 중") return lines = [f"📊 [{now} 현황]"] for ticker, pos in positions.items(): current = pyupbit.get_current_price(ticker) if not current: continue pnl = (current - pos["buy_price"]) / pos["buy_price"] * 100 peak = pos["peak_price"] drop = (peak - current) / peak * 100 elapsed = (datetime.now() - pos["entry_time"]).total_seconds() / 3600 emoji = "📈" if pnl >= 0 else "📉" lines.append( f"\n{emoji} {ticker}\n" f" 현재가: {current:,.0f}원\n" f" 수익률: {pnl:+.1f}%\n" f" 최고가 대비: -{drop:.1f}%\n" f" 보유: {elapsed:.1f}h" ) _send("\n".join(lines))