""" Trade Signal Notifier — отправка сигналов через Telegram/Webhook. Режимы: - Telegram: бот отправляет сообщение при BUY/SELL сигналах - Webhook: POST запрос на указанный URL Использование: from utils.notifier import Notifier n = Notifier() n.send_signal(ticker='SBER', signal='BUY', confidence=0.75, price=294.13) n.send_summary(summary={'total_buy': 3, 'total_sell': 1, 'total_neutral': 10}) """ import os import json import logging from datetime import datetime from typing import Optional logger = logging.getLogger(__name__) class Notifier: """Уведомитель о торговых сигналах. Поддерживает Telegram и Webhook. Настройки читает из окружения (.env) либо переданных параметров. """ def __init__( self, telegram_token: Optional[str] = None, telegram_chat_id: Optional[str] = None, webhook_url: Optional[str] = None, min_confidence: float = 0.55, ): self.telegram_token = telegram_token or os.environ.get('TELEGRAM_BOT_TOKEN') self.telegram_chat_id = telegram_chat_id or os.environ.get('TELEGRAM_CHAT_ID') self.webhook_url = webhook_url or os.environ.get('NOTIFIER_WEBHOOK_URL') self.min_confidence = min_confidence self._telegram_enabled = bool(self.telegram_token and self.telegram_chat_id) self._webhook_enabled = bool(self.webhook_url) if not (self._telegram_enabled or self._webhook_enabled): logger.info( "Notifier: no channels configured. " "Set TELEGRAM_BOT_TOKEN+TELEGRAM_CHAT_ID or NOTIFIER_WEBHOOK_URL" ) # ------------------------------------------------------------------ # Публичные методы # ------------------------------------------------------------------ def send_signal( self, ticker: str, signal: str, confidence: float, price: float, p_long: float = 0.0, p_short: float = 0.0, ) -> None: """Отправляет уведомление о сигнале (только BUY/SELL, если confidence >= порога).""" if signal == 'NEUTRAL': return if confidence < self.min_confidence: return message = self._format_signal(ticker, signal, confidence, price, p_long, p_short) self._notify(message) def send_summary(self, summary: dict) -> None: """Отправляет сводку после мониторинга всех тикеров.""" message = self._format_summary(summary) self._notify(message) def send_alert(self, text: str) -> None: """Отправляет произвольное предупреждение (ошибки, аномалии).""" self._notify(f"⚠️ *AI Strategy Alert*\n{text}") def send_ticker_disabled(self, ticker: str, reason: str, pnl: float = 0.0, deposit: float = 0.0) -> None: """Уведомление о блокировке тикера через auto-disable (audit 2026-08-03). Args: ticker: заблокированный тикер. reason: причина блокировки (например 'auto_disable (PnL -4367/92035...)'). pnl: суммарный PnL за окно. deposit: текущий депозит. """ pnl_str = f"{pnl:+.0f}" if pnl else "N/A" dep_str = f"{deposit:.0f}" if deposit else "N/A" pnl_pct = f"{pnl/deposit*100:+.1f}%" if deposit > 0 else "N/A" message = ( f"🚫 *AI Strategy: Ticker Auto-Disabled*\n" f"Ticker: `{ticker}`\n" f"Reason: {reason}\n" f"PnL за окно: {pnl_str} ₽ ({pnl_pct} от депозита {dep_str} ₽)\n" f"Cooldown: 7 дней" ) self._notify(message) # ------------------------------------------------------------------ # Форматирование # ------------------------------------------------------------------ @staticmethod def _format_signal( ticker: str, signal: str, confidence: float, price: float, p_long: float, p_short: float, ) -> str: emoji = '🟢' if signal == 'BUY' else '🔴' return ( f"{emoji} *{ticker}*: {signal}\n" f"├ Price: {price:.2f}\n" f"├ Confidence: {confidence:.1%}\n" f"├ P(Long): {p_long:.1%}\n" f"└ P(Short): {p_short:.1%}\n" f"🕐 {datetime.now().strftime('%H:%M %d.%m.%Y')}" ) @staticmethod def _format_summary(summary: dict) -> str: total = summary.get('total_buy', 0) + summary.get('total_sell', 0) + summary.get('total_neutral', 0) lines = [ f"📊 *Monitor Summary*", f"├ Total: {total} tickers", ] if summary.get('total_buy'): lines.append(f"├ 🟢 BUY: {summary['total_buy']}") if summary.get('total_sell'): lines.append(f"├ 🔴 SELL: {summary['total_sell']}") if summary.get('total_neutral'): lines.append(f"├ ⚪ NEUTRAL: {summary['total_neutral']}") if summary.get('signals'): lines.append(f"└ Signals:") for s in summary['signals']: lines.append(f" {s['emoji']} {s['ticker']}: {s['signal']} ({s['confidence']:.1%})") lines.append(f"🕐 {datetime.now().strftime('%H:%M %d.%m.%Y')}") return '\n'.join(lines) # ------------------------------------------------------------------ # Отправка # ------------------------------------------------------------------ def _notify(self, message: str) -> None: """Отправляет сообщение через все настроенные каналы.""" errors = [] if self._telegram_enabled: try: self._send_telegram(message) except Exception as e: errors.append(f"Telegram: {e}") logger.error("Telegram notification failed: %s", e) if self._webhook_enabled: try: self._send_webhook(message) except Exception as e: errors.append(f"Webhook: {e}") logger.error("Webhook notification failed: %s", e) if errors: logger.warning("Notifier errors: %s", '; '.join(errors)) def _send_telegram(self, text: str) -> None: """Отправляет сообщение через Telegram Bot API.""" import requests url = f"https://api.telegram.org/bot{self.telegram_token}/sendMessage" payload = { 'chat_id': self.telegram_chat_id, 'text': text, 'parse_mode': 'Markdown', 'disable_web_page_preview': True, } resp = requests.post(url, json=payload, timeout=10) resp.raise_for_status() data = resp.json() if not data.get('ok'): raise RuntimeError(f"Telegram API error: {data.get('description', 'unknown')}") logger.debug("Telegram notification sent") def _send_webhook(self, message: str) -> None: """Отправляет POST запрос на webhook URL.""" import requests payload = { 'text': message, 'source': 'ai_strategy', 'timestamp': datetime.utcnow().isoformat(), } resp = requests.post( self.webhook_url, json=payload, timeout=10, headers={'Content-Type': 'application/json'}, ) resp.raise_for_status() logger.debug("Webhook notification sent")