"""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))