2585d47140
Run status reporter thread every 60 minutes, sends current price, PnL, drop from peak, and holding time for each position. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
83 lines
2.4 KiB
Python
83 lines
2.4 KiB
Python
"""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"📈 <b>[매수]</b> {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} <b>[매도]</b> {ticker}\n"
|
|
f"가격: {price:,.0f}원\n"
|
|
f"수익률: {pnl_pct:+.1f}%\n"
|
|
f"사유: {reason}"
|
|
)
|
|
|
|
|
|
def notify_error(message: str) -> None:
|
|
_send(f"⚠️ <b>[오류]</b>\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"📊 <b>[{now} 현황]</b>\n보유 포지션 없음 — 매수 신호 대기 중")
|
|
return
|
|
|
|
lines = [f"📊 <b>[{now} 현황]</b>"]
|
|
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} <b>{ticker}</b>\n"
|
|
f" 현재가: {current:,.0f}원\n"
|
|
f" 수익률: {pnl:+.1f}%\n"
|
|
f" 최고가 대비: -{drop:.1f}%\n"
|
|
f" 보유: {elapsed:.1f}h"
|
|
)
|
|
_send("\n".join(lines))
|