Alertas de Preço em Tempo Real com Python 2026: Tutorial Completo

┌─────────────────────────────────────────────────────────────┐
│                 SISTEMA DE ALERTAS                           │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  Monitoramento                                               │
│  ├─ Preço acima de X                                        │
│  ├─ Preço abaixo de Y                                       │
│  ├─ Variação diária > Z%                                    │
│  └─ Cruzamento de médias                                    │
│                                                              │
│  Notificações                                                │
│  ├─ 📧 Email                                                │
│  ├─ 📱 Telegram                                             │
│  ├─ 💻 Desktop (Windows/Mac/Linux)                         │
│  └─ 📝 Log em arquivo                                       │
│                                                              │
│  Execução                                                    │
│  ├─ Script rodando continuamente                            │
│  ├─ Cron job (Linux/Mac)                                    │
│  └─ Task Scheduler (Windows)                                │
│                                                              │
└─────────────────────────────────────────────────────────────┘

pip install requests schedule plyer python-telegram-bot

price-alerts/
├── src/
│   ├── alerts/
│   │   ├── base.py           # Classe base de alerta
│   │   ├── price_alert.py    # Alertas de preço
│   │   └── technical.py      # Alertas técnicos
│   ├── notifiers/
│   │   ├── base.py           # Interface de notificação
│   │   ├── telegram.py       # Notificação Telegram
│   │   ├── email.py          # Notificação Email
│   │   └── desktop.py        # Notificação Desktop
│   └── monitor.py            # Monitor principal
├── config.py                 # Configurações
├── main.py                   # Script principal
└── requirements.txt

import requests
from typing import Optional
from dataclasses import dataclass

@dataclass
class StockQuote:
    """Representa uma cotação de ação"""
    symbol: str
    name: str
    price: float
    change: float
    change_percent: float
    volume: int
    previous_close: float
    market_time: str

class BrapiClient:
    """Cliente para API da brapi"""
    
    BASE_URL = "https://brapi.dev/api"
    
    def __init__(self, token: str):
        self.token = token
    
    def get_quote(self, ticker: str) -> Optional[StockQuote]:
        """Busca cotação de uma ação"""
        
        url = f"{self.BASE_URL}/quote/{ticker}"
        params = {"token": self.token}
        
        try:
            response = requests.get(url, params=params, timeout=10)
            response.raise_for_status()
            data = response.json()
            
            if not data.get("results"):
                return None
            
            result = data["results"][0]
            
            return StockQuote(
                symbol=result.get("symbol", ticker),
                name=result.get("longName", result.get("shortName", "")),
                price=result.get("regularMarketPrice", 0),
                change=result.get("regularMarketChange", 0),
                change_percent=result.get("regularMarketChangePercent", 0),
                volume=result.get("regularMarketVolume", 0),
                previous_close=result.get("regularMarketPreviousClose", 0),
                market_time=result.get("regularMarketTime", "")
            )
        
        except Exception as e:
            print(f"Erro ao buscar {ticker}: {e}")
            return None
    
    def get_multiple_quotes(self, tickers: list[str]) -> dict[str, StockQuote]:
        """Busca cotação de múltiplas ações"""
        
        # Juntar tickers com vírgula
        tickers_str = ",".join(tickers)
        url = f"{self.BASE_URL}/quote/{tickers_str}"
        params = {"token": self.token}
        
        try:
            response = requests.get(url, params=params, timeout=10)
            response.raise_for_status()
            data = response.json()
            
            quotes = {}
            for result in data.get("results", []):
                symbol = result.get("symbol", "")
                quotes[symbol] = StockQuote(
                    symbol=symbol,
                    name=result.get("longName", result.get("shortName", "")),
                    price=result.get("regularMarketPrice", 0),
                    change=result.get("regularMarketChange", 0),
                    change_percent=result.get("regularMarketChangePercent", 0),
                    volume=result.get("regularMarketVolume", 0),
                    previous_close=result.get("regularMarketPreviousClose", 0),
                    market_time=result.get("regularMarketTime", "")
                )
            
            return quotes
        
        except Exception as e:
            print(f"Erro ao buscar cotações: {e}")
            return {}


# Teste
if __name__ == "__main__":
    client = BrapiClient("seu_token")
    quote = client.get_quote("PETR4")
    print(f"{quote.symbol}: R$ {quote.price:.2f} ({quote.change_percent:+.2f}%)")

# src/alerts/base.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional
from enum import Enum

class AlertType(Enum):
    PRICE_ABOVE = "price_above"
    PRICE_BELOW = "price_below"
    CHANGE_PERCENT = "change_percent"
    VOLUME_SPIKE = "volume_spike"
    CROSSING_UP = "crossing_up"
    CROSSING_DOWN = "crossing_down"

@dataclass
class AlertTrigger:
    """Resultado de um alerta disparado"""
    ticker: str
    alert_type: AlertType
    message: str
    current_value: float
    threshold: float
    triggered_at: str

class Alert(ABC):
    """Classe base para alertas"""
    
    def __init__(
        self,
        ticker: str,
        alert_type: AlertType,
        threshold: float,
        enabled: bool = True,
        one_time: bool = False
    ):
        self.ticker = ticker
        self.alert_type = alert_type
        self.threshold = threshold
        self.enabled = enabled
        self.one_time = one_time
        self.triggered = False
    
    @abstractmethod
    def check(self, quote) -> Optional[AlertTrigger]:
        """Verifica se o alerta deve ser disparado"""
        pass
    
    def reset(self):
        """Reseta o estado do alerta"""
        self.triggered = False

# src/alerts/price_alert.py
from datetime import datetime
from typing import Optional
from .base import Alert, AlertType, AlertTrigger

class PriceAboveAlert(Alert):
    """Alerta quando preço ultrapassa valor"""
    
    def __init__(self, ticker: str, threshold: float, **kwargs):
        super().__init__(ticker, AlertType.PRICE_ABOVE, threshold, **kwargs)
    
    def check(self, quote) -> Optional[AlertTrigger]:
        if not self.enabled or (self.one_time and self.triggered):
            return None
        
        if quote.price > self.threshold:
            self.triggered = True
            return AlertTrigger(
                ticker=self.ticker,
                alert_type=self.alert_type,
                message=f"🚀 {self.ticker} ultrapassou R$ {self.threshold:.2f}! Preço atual: R$ {quote.price:.2f}",
                current_value=quote.price,
                threshold=self.threshold,
                triggered_at=datetime.now().isoformat()
            )
        
        return None


class PriceBelowAlert(Alert):
    """Alerta quando preço cai abaixo de valor"""
    
    def __init__(self, ticker: str, threshold: float, **kwargs):
        super().__init__(ticker, AlertType.PRICE_BELOW, threshold, **kwargs)
    
    def check(self, quote) -> Optional[AlertTrigger]:
        if not self.enabled or (self.one_time and self.triggered):
            return None
        
        if quote.price < self.threshold:
            self.triggered = True
            return AlertTrigger(
                ticker=self.ticker,
                alert_type=self.alert_type,
                message=f"📉 {self.ticker} caiu abaixo de R$ {self.threshold:.2f}! Preço atual: R$ {quote.price:.2f}",
                current_value=quote.price,
                threshold=self.threshold,
                triggered_at=datetime.now().isoformat()
            )
        
        return None


class ChangePercentAlert(Alert):
    """Alerta quando variação diária excede percentual"""
    
    def __init__(self, ticker: str, threshold: float, **kwargs):
        super().__init__(ticker, AlertType.CHANGE_PERCENT, threshold, **kwargs)
    
    def check(self, quote) -> Optional[AlertTrigger]:
        if not self.enabled or (self.one_time and self.triggered):
            return None
        
        if abs(quote.change_percent) >= self.threshold:
            self.triggered = True
            direction = "📈 subiu" if quote.change_percent > 0 else "📉 caiu"
            return AlertTrigger(
                ticker=self.ticker,
                alert_type=self.alert_type,
                message=f"{self.ticker} {direction} {abs(quote.change_percent):.2f}% hoje! Preço: R$ {quote.price:.2f}",
                current_value=quote.change_percent,
                threshold=self.threshold,
                triggered_at=datetime.now().isoformat()
            )
        
        return None


class VolumeSpikeAlert(Alert):
    """Alerta quando volume é muito acima do normal"""
    
    def __init__(self, ticker: str, threshold: float, avg_volume: int, **kwargs):
        super().__init__(ticker, AlertType.VOLUME_SPIKE, threshold, **kwargs)
        self.avg_volume = avg_volume
    
    def check(self, quote) -> Optional[AlertTrigger]:
        if not self.enabled or (self.one_time and self.triggered):
            return None
        
        if self.avg_volume > 0:
            volume_ratio = quote.volume / self.avg_volume
            
            if volume_ratio >= self.threshold:
                self.triggered = True
                return AlertTrigger(
                    ticker=self.ticker,
                    alert_type=self.alert_type,
                    message=f"📊 {self.ticker} com volume {volume_ratio:.1f}x acima da média! Volume: {quote.volume:,}",
                    current_value=volume_ratio,
                    threshold=self.threshold,
                    triggered_at=datetime.now().isoformat()
                )
        
        return None

# src/notifiers/base.py
from abc import ABC, abstractmethod
from src.alerts.base import AlertTrigger

class Notifier(ABC):
    """Interface para notificadores"""
    
    @abstractmethod
    def send(self, trigger: AlertTrigger) -> bool:
        """Envia notificação. Retorna True se sucesso."""
        pass

# src/notifiers/desktop.py
from plyer import notification
from .base import Notifier
from src.alerts.base import AlertTrigger

class DesktopNotifier(Notifier):
    """Notificação desktop (Windows/Mac/Linux)"""
    
    def __init__(self, app_name: str = "Alertas de Ações"):
        self.app_name = app_name
    
    def send(self, trigger: AlertTrigger) -> bool:
        try:
            notification.notify(
                title=f"Alerta: {trigger.ticker}",
                message=trigger.message,
                app_name=self.app_name,
                timeout=10
            )
            return True
        except Exception as e:
            print(f"Erro na notificação desktop: {e}")
            return False

# src/notifiers/telegram.py
import requests
from .base import Notifier
from src.alerts.base import AlertTrigger

class TelegramNotifier(Notifier):
    """Notificação via Telegram"""
    
    def __init__(self, bot_token: str, chat_id: str):
        self.bot_token = bot_token
        self.chat_id = chat_id
        self.api_url = f"https://api.telegram.org/bot{bot_token}"
    
    def send(self, trigger: AlertTrigger) -> bool:
        try:
            url = f"{self.api_url}/sendMessage"
            
            # Formatar mensagem
            message = f"""
🔔 *ALERTA DE PREÇO*

*{trigger.ticker}*
{trigger.message}

⏰ {trigger.triggered_at}
            """
            
            payload = {
                "chat_id": self.chat_id,
                "text": message,
                "parse_mode": "Markdown"
            }
            
            response = requests.post(url, json=payload, timeout=10)
            return response.status_code == 200
        
        except Exception as e:
            print(f"Erro na notificação Telegram: {e}")
            return False


# Como obter bot_token e chat_id:
"""
1. Crie um bot com @BotFather no Telegram
   - Envie /newbot
   - Siga as instruções
   - Copie o token fornecido

2. Obtenha seu chat_id:
   - Inicie conversa com seu bot
   - Acesse: https://api.telegram.org/bot<TOKEN>/getUpdates
   - Procure "chat":{"id": 123456789}
"""

# src/notifiers/email.py
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from .base import Notifier
from src.alerts.base import AlertTrigger

class EmailNotifier(Notifier):
    """Notificação via Email"""
    
    def __init__(
        self,
        smtp_server: str,
        smtp_port: int,
        sender_email: str,
        sender_password: str,
        recipient_email: str
    ):
        self.smtp_server = smtp_server
        self.smtp_port = smtp_port
        self.sender_email = sender_email
        self.sender_password = sender_password
        self.recipient_email = recipient_email
    
    def send(self, trigger: AlertTrigger) -> bool:
        try:
            # Criar mensagem
            msg = MIMEMultipart("alternative")
            msg["Subject"] = f"🔔 Alerta: {trigger.ticker}"
            msg["From"] = self.sender_email
            msg["To"] = self.recipient_email
            
            # Corpo do email
            text_content = f"""
ALERTA DE PREÇO

Ticker: {trigger.ticker}
{trigger.message}

Valor atual: R$ {trigger.current_value:.2f}
Limite configurado: R$ {trigger.threshold:.2f}
Hora: {trigger.triggered_at}

---
Sistema de Alertas brapi.dev
            """
            
            html_content = f"""
<html>
<body>
    <h2>🔔 Alerta de Preço</h2>
    <table>
        <tr><td><strong>Ticker:</strong></td><td>{trigger.ticker}</td></tr>
        <tr><td><strong>Mensagem:</strong></td><td>{trigger.message}</td></tr>
        <tr><td><strong>Valor atual:</strong></td><td>R$ {trigger.current_value:.2f}</td></tr>
        <tr><td><strong>Limite:</strong></td><td>R$ {trigger.threshold:.2f}</td></tr>
        <tr><td><strong>Hora:</strong></td><td>{trigger.triggered_at}</td></tr>
    </table>
    <hr>
    <p><small>Sistema de Alertas brapi.dev</small></p>
</body>
</html>
            """
            
            msg.attach(MIMEText(text_content, "plain"))
            msg.attach(MIMEText(html_content, "html"))
            
            # Enviar
            with smtplib.SMTP(self.smtp_server, self.smtp_port) as server:
                server.starttls()
                server.login(self.sender_email, self.sender_password)
                server.sendmail(
                    self.sender_email,
                    self.recipient_email,
                    msg.as_string()
                )
            
            return True
        
        except Exception as e:
            print(f"Erro na notificação Email: {e}")
            return False


# Configuração para Gmail:
"""
smtp_server = "smtp.gmail.com"
smtp_port = 587
sender_email = "[email protected]"
sender_password = "sua_senha_de_app"  # Não é a senha normal!

Para criar senha de app no Gmail:
1. Ative 2FA em sua conta Google
2. Acesse: https://myaccount.google.com/apppasswords
3. Gere uma senha para "Mail"
"""

# src/monitor.py
import time
import logging
from datetime import datetime
from typing import Optional
from src.data_client import BrapiClient
from src.alerts.base import Alert, AlertTrigger
from src.notifiers.base import Notifier

# Configurar logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[
        logging.FileHandler("alerts.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

class PriceMonitor:
    """Monitor de preços com alertas"""
    
    def __init__(
        self,
        brapi_token: str,
        notifiers: list[Notifier] = None,
        check_interval: int = 60  # segundos
    ):
        self.client = BrapiClient(brapi_token)
        self.notifiers = notifiers or []
        self.check_interval = check_interval
        self.alerts: list[Alert] = []
        self.running = False
    
    def add_alert(self, alert: Alert):
        """Adiciona um alerta"""
        self.alerts.append(alert)
        logger.info(f"Alerta adicionado: {alert.ticker} - {alert.alert_type.value}")
    
    def remove_alert(self, ticker: str, alert_type = None):
        """Remove alertas por ticker"""
        self.alerts = [
            a for a in self.alerts
            if not (a.ticker == ticker and (alert_type is None or a.alert_type == alert_type))
        ]
    
    def add_notifier(self, notifier: Notifier):
        """Adiciona um notificador"""
        self.notifiers.append(notifier)
    
    def _notify(self, trigger: AlertTrigger):
        """Envia notificação por todos os canais"""
        for notifier in self.notifiers:
            try:
                success = notifier.send(trigger)
                if success:
                    logger.info(f"Notificação enviada: {type(notifier).__name__}")
                else:
                    logger.warning(f"Falha na notificação: {type(notifier).__name__}")
            except Exception as e:
                logger.error(f"Erro ao notificar: {e}")
    
    def check_alerts(self):
        """Verifica todos os alertas"""
        
        # Agrupar tickers únicos
        tickers = list(set(a.ticker for a in self.alerts if a.enabled))
        
        if not tickers:
            return
        
        # Buscar cotações
        quotes = self.client.get_multiple_quotes(tickers)
        
        # Verificar cada alerta
        for alert in self.alerts:
            if not alert.enabled:
                continue
            
            quote = quotes.get(alert.ticker)
            if not quote:
                continue
            
            trigger = alert.check(quote)
            
            if trigger:
                logger.info(f"ALERTA DISPARADO: {trigger.message}")
                self._notify(trigger)
                
                # Desativar se for one-time
                if alert.one_time:
                    alert.enabled = False
    
    def run(self):
        """Inicia o monitoramento"""
        self.running = True
        logger.info(f"Monitor iniciado. Verificando a cada {self.check_interval}s")
        logger.info(f"Alertas ativos: {len([a for a in self.alerts if a.enabled])}")
        
        while self.running:
            try:
                self.check_alerts()
            except Exception as e:
                logger.error(f"Erro no ciclo de verificação: {e}")
            
            time.sleep(self.check_interval)
    
    def stop(self):
        """Para o monitoramento"""
        self.running = False
        logger.info("Monitor parado")
    
    def status(self) -> dict:
        """Retorna status do monitor"""
        return {
            "running": self.running,
            "total_alerts": len(self.alerts),
            "active_alerts": len([a for a in self.alerts if a.enabled]),
            "notifiers": len(self.notifiers),
            "check_interval": self.check_interval
        }

# config.py
import os

# Token da brapi (obtenha em https://brapi.dev)
BRAPI_TOKEN = os.getenv("BRAPI_TOKEN", "seu_token_aqui")

# Telegram (opcional)
TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN", "")
TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID", "")

# Email (opcional)
EMAIL_CONFIG = {
    "smtp_server": "smtp.gmail.com",
    "smtp_port": 587,
    "sender_email": os.getenv("EMAIL_SENDER", ""),
    "sender_password": os.getenv("EMAIL_PASSWORD", ""),
    "recipient_email": os.getenv("EMAIL_RECIPIENT", "")
}

# Intervalo de verificação em segundos
CHECK_INTERVAL = 60

# Alertas configurados
ALERTS_CONFIG = [
    # Preço acima
    {"type": "price_above", "ticker": "PETR4", "threshold": 40.00},
    {"type": "price_above", "ticker": "VALE3", "threshold": 70.00},
    
    # Preço abaixo
    {"type": "price_below", "ticker": "PETR4", "threshold": 30.00},
    {"type": "price_below", "ticker": "ITUB4", "threshold": 28.00},
    
    # Variação diária
    {"type": "change_percent", "ticker": "MGLU3", "threshold": 5.0},
    {"type": "change_percent", "ticker": "PETR4", "threshold": 3.0},
]

# main.py
from config import *
from src.monitor import PriceMonitor
from src.alerts.price_alert import (
    PriceAboveAlert,
    PriceBelowAlert,
    ChangePercentAlert
)
from src.notifiers.desktop import DesktopNotifier
from src.notifiers.telegram import TelegramNotifier
from src.notifiers.email import EmailNotifier

def create_alert(config: dict):
    """Cria alerta a partir de configuração"""
    alert_type = config["type"]
    ticker = config["ticker"]
    threshold = config["threshold"]
    
    if alert_type == "price_above":
        return PriceAboveAlert(ticker, threshold)
    elif alert_type == "price_below":
        return PriceBelowAlert(ticker, threshold)
    elif alert_type == "change_percent":
        return ChangePercentAlert(ticker, threshold)
    else:
        raise ValueError(f"Tipo de alerta desconhecido: {alert_type}")

def main():
    print("""
    ╔═══════════════════════════════════════════╗
    ║     SISTEMA DE ALERTAS DE PREÇO           ║
    ║              brapi.dev                    ║
    ╚═══════════════════════════════════════════╝
    """)
    
    # Criar monitor
    monitor = PriceMonitor(
        brapi_token=BRAPI_TOKEN,
        check_interval=CHECK_INTERVAL
    )
    
    # Adicionar notificadores
    monitor.add_notifier(DesktopNotifier())
    
    if TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID:
        monitor.add_notifier(TelegramNotifier(
            bot_token=TELEGRAM_BOT_TOKEN,
            chat_id=TELEGRAM_CHAT_ID
        ))
        print("✓ Notificações Telegram ativadas")
    
    if EMAIL_CONFIG["sender_email"]:
        monitor.add_notifier(EmailNotifier(**EMAIL_CONFIG))
        print("✓ Notificações Email ativadas")
    
    # Adicionar alertas
    for config in ALERTS_CONFIG:
        alert = create_alert(config)
        monitor.add_alert(alert)
    
    print(f"\n📊 Alertas configurados: {len(ALERTS_CONFIG)}")
    print(f"⏱️  Intervalo de verificação: {CHECK_INTERVAL}s")
    print("\nPressione Ctrl+C para parar\n")
    
    # Iniciar monitoramento
    try:
        monitor.run()
    except KeyboardInterrupt:
        print("\n\nEncerrando...")
        monitor.stop()

if __name__ == "__main__":
    main()

# cli.py
import argparse
import json
from src.monitor import PriceMonitor
from src.data_client import BrapiClient
from src.alerts.price_alert import PriceAboveAlert, PriceBelowAlert, ChangePercentAlert
from src.notifiers.desktop import DesktopNotifier
from config import BRAPI_TOKEN

def cmd_quote(args):
    """Mostra cotação atual"""
    client = BrapiClient(BRAPI_TOKEN)
    
    for ticker in args.tickers:
        quote = client.get_quote(ticker)
        if quote:
            print(f"{quote.symbol}: R$ {quote.price:.2f} ({quote.change_percent:+.2f}%)")
        else:
            print(f"{ticker}: Não encontrado")

def cmd_watch(args):
    """Monitora preço com alerta"""
    monitor = PriceMonitor(BRAPI_TOKEN, check_interval=args.interval)
    monitor.add_notifier(DesktopNotifier())
    
    if args.above:
        monitor.add_alert(PriceAboveAlert(args.ticker, args.above))
        print(f"Alerta: {args.ticker} > R$ {args.above:.2f}")
    
    if args.below:
        monitor.add_alert(PriceBelowAlert(args.ticker, args.below))
        print(f"Alerta: {args.ticker} < R$ {args.below:.2f}")
    
    if args.change:
        monitor.add_alert(ChangePercentAlert(args.ticker, args.change))
        print(f"Alerta: {args.ticker} variação > {args.change}%")
    
    print(f"\nMonitorando... (Ctrl+C para parar)")
    
    try:
        monitor.run()
    except KeyboardInterrupt:
        monitor.stop()

def main():
    parser = argparse.ArgumentParser(description="Sistema de Alertas de Preço")
    subparsers = parser.add_subparsers(dest="command")
    
    # Comando: quote
    quote_parser = subparsers.add_parser("quote", help="Mostra cotação")
    quote_parser.add_argument("tickers", nargs="+", help="Códigos das ações")
    
    # Comando: watch
    watch_parser = subparsers.add_parser("watch", help="Monitora com alertas")
    watch_parser.add_argument("ticker", help="Código da ação")
    watch_parser.add_argument("--above", type=float, help="Alerta se preço acima de X")
    watch_parser.add_argument("--below", type=float, help="Alerta se preço abaixo de X")
    watch_parser.add_argument("--change", type=float, help="Alerta se variação > X%")
    watch_parser.add_argument("--interval", type=int, default=60, help="Intervalo em segundos")
    
    args = parser.parse_args()
    
    if args.command == "quote":
        cmd_quote(args)
    elif args.command == "watch":
        cmd_watch(args)
    else:
        parser.print_help()

if __name__ == "__main__":
    main()

# Ver cotação
python cli.py quote PETR4 VALE3 ITUB4

# Monitorar com alerta
python cli.py watch PETR4 --above 40 --below 30

# Monitorar variação
python cli.py watch MGLU3 --change 5 --interval 30

# Editar crontab
crontab -e

# Executar a cada 5 minutos durante horário de pregão (10h-18h)
*/5 10-17 * * 1-5 /usr/bin/python3 /caminho/para/main.py >> /var/log/alerts.log 2>&1

# /etc/systemd/system/price-alerts.service
[Unit]
Description=Price Alerts Service
After=network.target

[Service]
Type=simple
User=seu_usuario
WorkingDirectory=/caminho/para/price-alerts
ExecStart=/usr/bin/python3 main.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

# Ativar serviço
sudo systemctl enable price-alerts
sudo systemctl start price-alerts
sudo systemctl status price-alerts

# src/alerts/technical.py
from typing import Optional
from collections import deque
from .base import Alert, AlertType, AlertTrigger

class MovingAverageCrossAlert(Alert):
    """Alerta de cruzamento de médias móveis"""
    
    def __init__(
        self,
        ticker: str,
        short_period: int = 20,
        long_period: int = 50,
        **kwargs
    ):
        super().__init__(ticker, AlertType.CROSSING_UP, 0, **kwargs)
        self.short_period = short_period
        self.long_period = long_period
        self.prices = deque(maxlen=long_period + 1)
        self.last_short_above = None
    
    def check(self, quote) -> Optional[AlertTrigger]:
        if not self.enabled:
            return None
        
        # Adicionar preço
        self.prices.append(quote.price)
        
        # Precisa de dados suficientes
        if len(self.prices) < self.long_period:
            return None
        
        # Calcular médias
        prices_list = list(self.prices)
        short_ma = sum(prices_list[-self.short_period:]) / self.short_period
        long_ma = sum(prices_list) / self.long_period
        
        short_above = short_ma > long_ma
        
        # Detectar cruzamento
        if self.last_short_above is not None:
            if short_above and not self.last_short_above:
                # Cruzamento para cima (golden cross)
                self.last_short_above = short_above
                return AlertTrigger(
                    ticker=self.ticker,
                    alert_type=AlertType.CROSSING_UP,
                    message=f"📈 {self.ticker}: Média de {self.short_period} cruzou ACIMA da média de {self.long_period}!",
                    current_value=quote.price,
                    threshold=0,
                    triggered_at=quote.market_time
                )
            
            elif not short_above and self.last_short_above:
                # Cruzamento para baixo (death cross)
                self.last_short_above = short_above
                return AlertTrigger(
                    ticker=self.ticker,
                    alert_type=AlertType.CROSSING_DOWN,
                    message=f"📉 {self.ticker}: Média de {self.short_period} cruzou ABAIXO da média de {self.long_period}!",
                    current_value=quote.price,
                    threshold=0,
                    triggered_at=quote.market_time
                )
        
        self.last_short_above = short_above
        return None

class AlertWithCooldown(Alert):
    def __init__(self, cooldown_minutes: int = 30, **kwargs):
        super().__init__(**kwargs)
        self.cooldown = cooldown_minutes * 60
        self.last_trigger = 0
    
    def check(self, quote):
        import time
        if time.time() - self.last_trigger < self.cooldown:
            return None
        
        trigger = super().check(quote)
        if trigger:
            self.last_trigger = time.time()
        return trigger