Bithumb/src/bithumb/notifications/telegram.py

"""텔레그램 Bot API 알림."""

from __future__ import annotations

import logging
from datetime import datetime
from typing import Any

import requests

logger = logging.getLogger(__name__)

_TELEGRAM_API = "https://api.telegram.org/bot{token}/sendMessage"


class TelegramNotifier:
    """매매 체결 등 운영 알림을 텔레그램으로 전송한다."""

    def __init__(
        self,
        bot_token: str,
        chat_id: str,
        *,
        enabled: bool = True,
        timeout_sec: float = 10.0,
    ) -> None:
        """알림 클라이언트를 초기화한다.

        Args:
            bot_token: Bot API 토큰.
            chat_id: 대상 채팅 ID.
            enabled: False면 전송하지 않음.
            timeout_sec: HTTP 타임아웃(초).
        """
        self.bot_token = (bot_token or "").strip()
        self.chat_id = (chat_id or "").strip()
        self.enabled = enabled
        self.timeout_sec = timeout_sec
        self._session = requests.Session()

    @property
    def is_active(self) -> bool:
        """토큰·채팅 ID가 있고 enabled일 때 True."""
        return bool(self.enabled and self.bot_token and self.chat_id)

    def send_message(self, text: str) -> bool:
        """텍스트 메시지를 전송한다.

        Args:
            text: 본문 (HTML 미사용, plain text).

        Returns:
            성공 시 True.
        """
        if not self.is_active:
            return False
        url = _TELEGRAM_API.format(token=self.bot_token)
        try:
            resp = self._session.post(
                url,
                json={
                    "chat_id": self.chat_id,
                    "text": text,
                    "disable_web_page_preview": True,
                },
                timeout=self.timeout_sec,
            )
            if resp.status_code != 200:
                logger.warning(
                    "텔레그램 전송 실패 status=%s body=%s",
                    resp.status_code,
                    resp.text[:200],
                )
                return False
            data = resp.json()
            if not data.get("ok"):
                logger.warning("텔레그램 API 오류: %s", data)
                return False
            return True
        except requests.RequestException as exc:
            logger.warning("텔레그램 전송 예외: %s", exc)
            return False

    def notify_trade_execution(
        self,
        *,
        mode: str,
        symbol: str,
        coin_name: str,
        technique_id: str,
        side: str,
        signal_type: str,
        datetime_str: str,
        signal_price: float,
        trade: dict[str, Any],
        portfolio: dict[str, Any],
        trades_today_count: int,
        daily_max_trades: int,
        cluster_size: int,
    ) -> bool:
        """체결 1건 알림을 포맷하여 전송한다."""
        if not trade.get("executed"):
            return False

        side_label = "매수" if side == "buy" else "매도"
        mode_label = "LIVE" if mode == "live" else "PAPER"
        price = float(trade.get("price", signal_price))
        order_krw = float(trade.get("order_krw", 0))
        order_coin = float(trade.get("order_coin", 0))
        fee_krw = float(trade.get("fee_krw", 0))
        cash = float(portfolio.get("cash_krw", 0))
        coin_qty = float(portfolio.get("coin_qty", 0))
        equity = cash + coin_qty * price

        lines = [
            f"[Bithumb] {side_label} 체결 ({mode_label})",
            f"{coin_name} ({symbol}) | {technique_id}",
            f"시각: {datetime_str}",
            f"신호: {signal_type or side}",
            f"체결가: {_fmt_krw(price)}",
        ]
        if side == "buy":
            lines.append(f"주문: {_fmt_krw(order_krw)} ({order_coin:.8f} {symbol})")
        else:
            lines.append(f"주문: {order_coin:.8f} {symbol} ({_fmt_krw(order_krw)})")
        lines.append(f"수수료: {_fmt_krw(fee_krw)}")
        if cluster_size > 1:
            lines.append(f"클러스터 분할: {cluster_size}건")
        lines.append(f"잔고: 현금 {_fmt_krw(cash)} | {symbol} {coin_qty:.8f}")
        lines.append(f"평가(추정): {_fmt_krw(equity)}")
        lines.append(f"오늘 체결: {trades_today_count}/{daily_max_trades}")

        return self.send_message("\n".join(lines))

    def notify_vol_breakout_trade(
        self,
        *,
        mode: str,
        symbol: str,
        side: str,
        price: float,
        order_krw: float,
        order_coin: float,
        equity_krw: float,
        reason: str = "signal_vol_breakout",
        ts: str | None = None,
        pnl_krw: float | None = None,
        pnl_pct: float | None = None,
    ) -> bool:
        """vol_breakout 현물 롱 체결 알림 (Binance live_engine 형식)."""
        mode_txt = "실거래" if mode == "live" else "페이퍼"
        market = f"{symbol.upper()}KRW"
        time_txt = (ts or datetime.now().strftime("%Y-%m-%d %H:%M:%S"))[:19]

        if side == "buy":
            text = (
                f"[{mode_txt}] 롱 진입(매수)\n"
                f"{market} @ {price:,.2f}\n"
                f"수량 {order_coin:.6f} (≈{order_krw:,.0f}원)\n"
                f"사유 {reason}\n"
                f"시각 {time_txt}"
            )
        else:
            pnl_line = ""
            if pnl_krw is not None and pnl_pct is not None:
                pnl_line = (
                    f"손익 {pnl_krw:+,.0f}원 ({pnl_pct:+.3f}%)\n"
                )
            text = (
                f"[{mode_txt}] 롱 청산(매도)\n"
                f"{market} @ {price:,.2f}\n"
                f"{pnl_line}"
                f"자본 {equity_krw:,.0f}원\n"
                f"사유 {reason}\n"
                f"시각 {time_txt}"
            )
        return self.send_message(text)

    def notify_trade_failure(
        self,
        *,
        mode: str,
        symbol: str,
        technique_id: str,
        side: str,
        datetime_str: str,
        reason: str,
    ) -> bool:
        """live 체결 실패 알림."""
        side_label = "매수" if side == "buy" else "매도"
        mode_label = "LIVE" if mode == "live" else "PAPER"
        text = (
            f"[Bithumb] {side_label} 실패 ({mode_label})\n"
            f"{symbol} | {technique_id}\n"
            f"시각: {datetime_str}\n"
            f"사유: {reason}"
        )
        return self.send_message(text)

    def notify_ops_error(
        self,
        *,
        mode: str,
        symbol: str,
        technique_id: str,
        stage: str,
        error: str,
        detail: str = "",
    ) -> bool:
        """운영 tick·체결 등 예외 발생 알림 (프로세스는 계속 실행)."""
        mode_label = "LIVE" if mode == "live" else "PAPER"
        lines = [
            f"[Bithumb] 운영 오류 ({mode_label})",
            f"{symbol} | {technique_id}",
            f"단계: {stage}",
            f"원인: {error}",
        ]
        if detail:
            lines.append(detail)
        lines.append("프로세스는 계속 실행됩니다.")
        return self.send_message("\n".join(lines))


def _fmt_krw(value: float) -> str:
    """원화 금액 포맷."""
    return f"{round(value):,}원"


def create_telegram_notifier(
    bot_token: str,
    chat_id: str,
    *,
    enabled: bool = True,
) -> TelegramNotifier:
    """설정값으로 TelegramNotifier를 생성한다."""
    return TelegramNotifier(bot_token, chat_id, enabled=enabled)