upbit-trader/core/notify.py

"""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,
    max_budget: int = 0, per_position: int = 0,
) -> None:
    budget_line = (
        f"운용예산: {max_budget:,}원 (포지션당 {per_position:,}원)\n"
        if max_budget else ""
    )
    _send(
        f"📈 <b>[매수]</b> {ticker}\n"
        f"가격: {price:,.2f}원\n"
        f"수량: {amount:.8f}\n"
        f"투자금: {invested_krw:,.2f}원\n"
        f"{budget_line}"
    )


def notify_sell(
    ticker: str, price: float, pnl_pct: float, reason: str,
    krw_profit: float = 0.0, fee_krw: float = 0.0,
    cum_profit: float = 0.0,
) -> None:
    trade_emoji = "✅" if pnl_pct >= 0 else "❌"
    cum_emoji   = "💚" if cum_profit >= 0 else "🔴"
    _send(
        f"{trade_emoji} <b>[매도]</b> {ticker}\n"
        f"가격: {price:,.2f}원\n"
        f"수익률: {pnl_pct:+.2f}%\n"
        f"실손익: {krw_profit:+,.2f}원 (수수료 {fee_krw:,.2f}원)\n"
        f"{cum_emoji} 누적손익: {cum_profit:+,.2f}원\n"
        f"사유: {reason}"
    )


def notify_signal(ticker: str, signal_price: float, vol_mult: float) -> None:
    """거래량 축적 신호 감지 알림."""
    _send(
        f"🔍 <b>[축적감지]</b> {ticker}\n"
        f"신호가: {signal_price:,.2f}원\n"
        f"거래량: {vol_mult:.1f}x 급증 + 2h 횡보\n"
        f"진입 목표: {signal_price * 1.048:,.2f}원 (+4.8%)"
    )


def notify_error(message: str) -> None:
    _send(f"⚠️ <b>[오류]</b>\n{message}")


def notify_status(
    positions: dict,
    max_budget: int = 0,
    per_position: int = 0,
    cum_profit: float = 0.0,
) -> None:
    """정각마다 시장 레짐 + 1시간 이상 보유 포지션 현황 전송."""
    from datetime import datetime
    import pyupbit
    from .market_regime import get_regime

    now = datetime.now().strftime("%H:%M")
    cum_sign = "+" if cum_profit >= 0 else ""

    # 시장 레짐
    regime = get_regime()
    regime_line = (
        f"{regime['emoji']} 시장: {regime['name'].upper()} "
        f"(score {regime['score']:+.2f}%) "
        f"| 조건 TREND≥{regime['trend_pct']}% / VOL≥{regime['vol_mult']}x\n"
    )

    # 1시간 이상 보유 포지션만 필터
    long_positions = {
        ticker: pos for ticker, pos in positions.items()
        if (datetime.now() - pos["entry_time"]).total_seconds() >= 3600
    }

    cum_emoji = "💚" if cum_profit >= 0 else "🔴"
    budget_info = (
        f"💰 운용예산: {max_budget:,}원 | 포지션당: {per_position:,}원\n"
        f"{cum_emoji} 누적손익: {cum_sign}{cum_profit:,.0f}원\n"
        if max_budget else ""
    )

    # 포지션 없어도 레짐 정보는 전송
    header = f"📊 <b>[{now} 현황]</b>\n{regime_line}{budget_info}"

    if not long_positions:
        _send(header + "1h+ 보유 포지션 없음")
        return

    lines = [header]
    for ticker, pos in long_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"{emoji} <b>{ticker}</b>\n"
            f"  현재가: {current:,.2f}원\n"
            f"  수익률: {pnl:+.2f}%\n"
            f"  최고가 대비: -{drop:.2f}%\n"
            f"  보유: {elapsed:.2f}h"
        )
    _send("\n".join(lines))