Bot de Telegram para Cotações: Como Criar com Python e brapi API

🤖 brapi Bot - Demonstração
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

Você: /cotacao PETR4

Bot: 📊 PETR4 - PETROBRAS PN
     ━━━━━━━━━━━━━━━━━━━━━━━
     💰 Preço: R$ 37.85
     📈 Variação: +1.20% (+R$ 0.45)
     📉 Mín/Máx: R$ 37.20 - R$ 38.10
     📊 Volume: 45.678.900
     ⏰ Atualizado: 14:30:00

Você: /alerta PETR4 > 40.00

Bot: ✅ Alerta configurado!
     Você será notificado quando
     PETR4 ultrapassar R$ 40.00

python --version

pip install python-telegram-bot requests python-dotenv apscheduler

telegram-bot/
├── bot.py              # Código principal
├── handlers.py         # Comandos do bot
├── brapi_client.py     # Cliente da API brapi
├── alerts.py           # Sistema de alertas
├── config.py           # Configurações
├── .env                # Variáveis de ambiente
└── requirements.txt    # Dependências

# Arquivo .env
TELEGRAM_TOKEN=123456789:ABCdefGHIjklMNOpqrsTUVwxyz
BRAPI_TOKEN=seu_token_brapi_aqui

python-telegram-bot==20.7
requests==2.31.0
python-dotenv==1.0.0
APScheduler==3.10.4

# brapi_client.py

import requests
from typing import Optional, List, Dict
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
    day_high: float
    day_low: float
    previous_close: float
    market_cap: Optional[float] = None
    pe_ratio: Optional[float] = None
    dividend_yield: Optional[float] = None


class BrapiClient:
    """Cliente para a API brapi.dev"""
    
    def __init__(self, token: str):
        self.token = token
        self.base_url = "https://brapi.dev/api"
    
    def get_quote(self, symbol: str) -> Optional[StockQuote]:
        """Obtém cotação de uma ação"""
        try:
            url = f"{self.base_url}/quote/{symbol}"
            params = {"token": self.token}
            
            response = requests.get(url, params=params, timeout=10)
            data = response.json()
            
            if not data.get('results'):
                return None
            
            result = data['results'][0]
            
            return StockQuote(
                symbol=result.get('symbol', symbol),
                name=result.get('shortName', ''),
                price=result.get('regularMarketPrice', 0),
                change=result.get('regularMarketChange', 0),
                change_percent=result.get('regularMarketChangePercent', 0),
                volume=result.get('regularMarketVolume', 0),
                day_high=result.get('regularMarketDayHigh', 0),
                day_low=result.get('regularMarketDayLow', 0),
                previous_close=result.get('regularMarketPreviousClose', 0),
                market_cap=result.get('marketCap'),
                pe_ratio=result.get('trailingPE'),
                dividend_yield=result.get('dividendYield'),
            )
            
        except Exception as e:
            print(f"Erro ao buscar {symbol}: {e}")
            return None
    
    def get_multiple_quotes(self, symbols: List[str]) -> List[StockQuote]:
        """Obtém cotações de múltiplas ações"""
        try:
            symbols_str = ','.join(symbols)
            url = f"{self.base_url}/quote/{symbols_str}"
            params = {"token": self.token}
            
            response = requests.get(url, params=params, timeout=30)
            data = response.json()
            
            quotes = []
            for result in data.get('results', []):
                quotes.append(StockQuote(
                    symbol=result.get('symbol', ''),
                    name=result.get('shortName', ''),
                    price=result.get('regularMarketPrice', 0),
                    change=result.get('regularMarketChange', 0),
                    change_percent=result.get('regularMarketChangePercent', 0),
                    volume=result.get('regularMarketVolume', 0),
                    day_high=result.get('regularMarketDayHigh', 0),
                    day_low=result.get('regularMarketDayLow', 0),
                    previous_close=result.get('regularMarketPreviousClose', 0),
                ))
            
            return quotes
            
        except Exception as e:
            print(f"Erro ao buscar múltiplas ações: {e}")
            return []
    
    def search_stocks(self, query: str) -> List[Dict]:
        """Busca ações por nome ou ticker"""
        try:
            url = f"{self.base_url}/quote/list"
            params = {
                "token": self.token,
                "search": query,
                "limit": 5
            }
            
            response = requests.get(url, params=params, timeout=10)
            data = response.json()
            
            return data.get('stocks', [])
            
        except Exception as e:
            print(f"Erro na busca: {e}")
            return []

# alerts.py

import json
from pathlib import Path
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
from enum import Enum


class AlertCondition(Enum):
    ABOVE = ">"
    BELOW = "<"
    EQUALS = "="


@dataclass
class PriceAlert:
    """Representa um alerta de preço"""
    user_id: int
    symbol: str
    condition: str  # ">", "<", "="
    target_price: float
    active: bool = True
    
    def check(self, current_price: float) -> bool:
        """Verifica se o alerta foi acionado"""
        if not self.active:
            return False
        
        if self.condition == ">":
            return current_price > self.target_price
        elif self.condition == "<":
            return current_price < self.target_price
        elif self.condition == "=":
            return abs(current_price - self.target_price) < 0.01
        
        return False


class AlertManager:
    """Gerenciador de alertas de preço"""
    
    def __init__(self, storage_file: str = "alerts.json"):
        self.storage_file = Path(storage_file)
        self.alerts: List[PriceAlert] = []
        self._load()
    
    def _load(self):
        """Carrega alertas do arquivo"""
        if self.storage_file.exists():
            try:
                with open(self.storage_file, 'r') as f:
                    data = json.load(f)
                    self.alerts = [PriceAlert(**a) for a in data]
            except Exception as e:
                print(f"Erro ao carregar alertas: {e}")
                self.alerts = []
    
    def _save(self):
        """Salva alertas no arquivo"""
        try:
            with open(self.storage_file, 'w') as f:
                json.dump([asdict(a) for a in self.alerts], f, indent=2)
        except Exception as e:
            print(f"Erro ao salvar alertas: {e}")
    
    def add_alert(self, user_id: int, symbol: str, 
                  condition: str, target_price: float) -> PriceAlert:
        """Adiciona novo alerta"""
        alert = PriceAlert(
            user_id=user_id,
            symbol=symbol.upper(),
            condition=condition,
            target_price=target_price
        )
        self.alerts.append(alert)
        self._save()
        return alert
    
    def get_user_alerts(self, user_id: int) -> List[PriceAlert]:
        """Retorna alertas de um usuário"""
        return [a for a in self.alerts if a.user_id == user_id and a.active]
    
    def remove_alert(self, user_id: int, symbol: str) -> bool:
        """Remove alertas de um símbolo para um usuário"""
        initial_count = len(self.alerts)
        self.alerts = [
            a for a in self.alerts 
            if not (a.user_id == user_id and a.symbol == symbol.upper())
        ]
        self._save()
        return len(self.alerts) < initial_count
    
    def check_alerts(self, symbol: str, current_price: float) -> List[PriceAlert]:
        """Verifica quais alertas foram acionados"""
        triggered = []
        
        for alert in self.alerts:
            if alert.symbol == symbol.upper() and alert.check(current_price):
                triggered.append(alert)
                alert.active = False  # Desativa após acionar
        
        if triggered:
            self._save()
        
        return triggered
    
    def get_all_symbols(self) -> List[str]:
        """Retorna todos os símbolos com alertas ativos"""
        return list(set(a.symbol for a in self.alerts if a.active))

# handlers.py

from telegram import Update
from telegram.ext import ContextTypes
from brapi_client import BrapiClient, StockQuote
from alerts import AlertManager
import re


def format_quote(quote: StockQuote) -> str:
    """Formata cotação para exibição no Telegram"""
    emoji_var = "📈" if quote.change >= 0 else "📉"
    sinal = "+" if quote.change >= 0 else ""
    
    msg = f"""📊 *{quote.symbol}* - {quote.name}
━━━━━━━━━━━━━━━━━━━━━━━

💰 *Preço:* R$ {quote.price:.2f}
{emoji_var} *Variação:* {sinal}{quote.change_percent:.2f}% ({sinal}R$ {quote.change:.2f})
📉 *Mín/Máx:* R$ {quote.day_low:.2f} - R$ {quote.day_high:.2f}
📊 *Volume:* {quote.volume:,.0f}
🔙 *Fechamento ant:* R$ {quote.previous_close:.2f}"""
    
    if quote.pe_ratio:
        msg += f"\n📐 *P/L:* {quote.pe_ratio:.2f}"
    
    if quote.dividend_yield:
        msg += f"\n💵 *Dividend Yield:* {quote.dividend_yield*100:.2f}%"
    
    if quote.market_cap:
        market_cap_bi = quote.market_cap / 1_000_000_000
        msg += f"\n🏢 *Market Cap:* R$ {market_cap_bi:.1f} bi"
    
    return msg


async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Comando /start - Boas-vindas"""
    welcome = """🤖 *Bem-vindo ao Bot de Cotações!*

Comandos disponíveis:

📊 `/cotacao PETR4` - Ver cotação
📋 `/carteira` - Ver sua carteira
🔔 `/alerta PETR4 > 40.00` - Criar alerta
📝 `/alertas` - Ver seus alertas
❌ `/remover PETR4` - Remover alerta
🔍 `/buscar petro` - Buscar ações
📈 `/comparar PETR4 VALE3` - Comparar ações

_Powered by brapi.dev_"""
    
    await update.message.reply_text(welcome, parse_mode='Markdown')


async def quote_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Comando /cotacao - Consulta cotação"""
    if not context.args:
        await update.message.reply_text(
            "❌ Use: `/cotacao PETR4`", 
            parse_mode='Markdown'
        )
        return
    
    symbol = context.args[0].upper()
    brapi: BrapiClient = context.bot_data['brapi']
    
    # Feedback de carregamento
    loading_msg = await update.message.reply_text(f"⏳ Buscando {symbol}...")
    
    quote = brapi.get_quote(symbol)
    
    if quote:
        await loading_msg.edit_text(format_quote(quote), parse_mode='Markdown')
    else:
        await loading_msg.edit_text(f"❌ Ação `{symbol}` não encontrada.", parse_mode='Markdown')


async def portfolio_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Comando /carteira - Mostra carteira do usuário"""
    user_id = update.effective_user.id
    
    # Carteira salva (simplificado - em produção usar banco de dados)
    portfolios = context.bot_data.get('portfolios', {})
    user_portfolio = portfolios.get(user_id, ['PETR4', 'VALE3', 'ITUB4'])
    
    if not user_portfolio:
        await update.message.reply_text(
            "📋 Sua carteira está vazia.\n"
            "Use `/adicionar PETR4` para adicionar ações.",
            parse_mode='Markdown'
        )
        return
    
    loading_msg = await update.message.reply_text("⏳ Carregando carteira...")
    
    brapi: BrapiClient = context.bot_data['brapi']
    quotes = brapi.get_multiple_quotes(user_portfolio)
    
    if not quotes:
        await loading_msg.edit_text("❌ Erro ao carregar carteira.")
        return
    
    total_change = sum(q.change_percent for q in quotes) / len(quotes)
    emoji_total = "📈" if total_change >= 0 else "📉"
    
    msg = f"📋 *SUA CARTEIRA*\n━━━━━━━━━━━━━━━━━━━━━━━\n\n"
    
    for quote in quotes:
        emoji = "🟢" if quote.change_percent >= 0 else "🔴"
        sinal = "+" if quote.change_percent >= 0 else ""
        msg += f"{emoji} *{quote.symbol}*: R$ {quote.price:.2f} ({sinal}{quote.change_percent:.2f}%)\n"
    
    sinal_total = "+" if total_change >= 0 else ""
    msg += f"\n━━━━━━━━━━━━━━━━━━━━━━━\n"
    msg += f"{emoji_total} *Média:* {sinal_total}{total_change:.2f}%"
    
    await loading_msg.edit_text(msg, parse_mode='Markdown')


async def alert_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Comando /alerta - Configura alerta de preço"""
    if len(context.args) < 3:
        await update.message.reply_text(
            "❌ Use: `/alerta PETR4 > 40.00`\n\n"
            "Condições: `>` (acima), `<` (abaixo), `=` (igual)",
            parse_mode='Markdown'
        )
        return
    
    symbol = context.args[0].upper()
    condition = context.args[1]
    
    try:
        target_price = float(context.args[2].replace(',', '.'))
    except ValueError:
        await update.message.reply_text("❌ Preço inválido.")
        return
    
    if condition not in ['>', '<', '=']:
        await update.message.reply_text(
            "❌ Condição inválida. Use: `>`, `<` ou `=`",
            parse_mode='Markdown'
        )
        return
    
    alert_manager: AlertManager = context.bot_data['alerts']
    user_id = update.effective_user.id
    
    alert = alert_manager.add_alert(user_id, symbol, condition, target_price)
    
    condition_text = {
        '>': 'ultrapassar',
        '<': 'cair abaixo de',
        '=': 'atingir'
    }
    
    await update.message.reply_text(
        f"✅ *Alerta configurado!*\n\n"
        f"Você será notificado quando\n"
        f"*{symbol}* {condition_text[condition]} R$ {target_price:.2f}",
        parse_mode='Markdown'
    )


async def list_alerts_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Comando /alertas - Lista alertas do usuário"""
    alert_manager: AlertManager = context.bot_data['alerts']
    user_id = update.effective_user.id
    
    alerts = alert_manager.get_user_alerts(user_id)
    
    if not alerts:
        await update.message.reply_text("📭 Você não tem alertas configurados.")
        return
    
    msg = "🔔 *SEUS ALERTAS*\n━━━━━━━━━━━━━━━━━━━━━━━\n\n"
    
    for alert in alerts:
        condition_text = {
            '>': 'acima de',
            '<': 'abaixo de',
            '=': 'igual a'
        }
        msg += f"• *{alert.symbol}* {condition_text[alert.condition]} R$ {alert.target_price:.2f}\n"
    
    msg += f"\n_Use `/remover TICKER` para cancelar_"
    
    await update.message.reply_text(msg, parse_mode='Markdown')


async def remove_alert_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Comando /remover - Remove alerta"""
    if not context.args:
        await update.message.reply_text(
            "❌ Use: `/remover PETR4`",
            parse_mode='Markdown'
        )
        return
    
    symbol = context.args[0].upper()
    alert_manager: AlertManager = context.bot_data['alerts']
    user_id = update.effective_user.id
    
    if alert_manager.remove_alert(user_id, symbol):
        await update.message.reply_text(f"✅ Alertas de *{symbol}* removidos.", parse_mode='Markdown')
    else:
        await update.message.reply_text(f"❌ Nenhum alerta encontrado para *{symbol}*.", parse_mode='Markdown')


async def search_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Comando /buscar - Busca ações"""
    if not context.args:
        await update.message.reply_text(
            "❌ Use: `/buscar petrobras`",
            parse_mode='Markdown'
        )
        return
    
    query = ' '.join(context.args)
    brapi: BrapiClient = context.bot_data['brapi']
    
    results = brapi.search_stocks(query)
    
    if not results:
        await update.message.reply_text(f"❌ Nenhuma ação encontrada para '{query}'")
        return
    
    msg = f"🔍 *Resultados para '{query}':*\n\n"
    
    for stock in results[:5]:
        msg += f"• *{stock.get('stock', '')}* - {stock.get('name', '')}\n"
    
    msg += "\n_Use `/cotacao TICKER` para ver detalhes_"
    
    await update.message.reply_text(msg, parse_mode='Markdown')


async def compare_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Comando /comparar - Compara duas ações"""
    if len(context.args) < 2:
        await update.message.reply_text(
            "❌ Use: `/comparar PETR4 VALE3`",
            parse_mode='Markdown'
        )
        return
    
    symbols = [arg.upper() for arg in context.args[:2]]
    brapi: BrapiClient = context.bot_data['brapi']
    
    loading_msg = await update.message.reply_text("⏳ Comparando...")
    
    quotes = brapi.get_multiple_quotes(symbols)
    
    if len(quotes) < 2:
        await loading_msg.edit_text("❌ Não foi possível obter dados de uma ou mais ações.")
        return
    
    q1, q2 = quotes[0], quotes[1]
    
    def winner(val1, val2, higher_better=True):
        if higher_better:
            return "🏆" if val1 > val2 else "  "
        else:
            return "🏆" if val1 < val2 else "  "
    
    msg = f"""📊 *COMPARATIVO*
━━━━━━━━━━━━━━━━━━━━━━━

|  | *{q1.symbol}* | *{q2.symbol}* |
|---|---|---|
| Preço | R$ {q1.price:.2f} | R$ {q2.price:.2f} |
| Variação | {q1.change_percent:+.2f}% | {q2.change_percent:+.2f}% |
| Volume | {q1.volume/1e6:.1f}M | {q2.volume/1e6:.1f}M |

{winner(q1.change_percent, q2.change_percent)} Melhor desempenho hoje: *{q1.symbol if q1.change_percent > q2.change_percent else q2.symbol}*
"""
    
    await loading_msg.edit_text(msg, parse_mode='Markdown')

# bot.py

import os
import asyncio
from dotenv import load_dotenv
from telegram import Update
from telegram.ext import (
    Application, 
    CommandHandler, 
    ContextTypes
)
from apscheduler.schedulers.asyncio import AsyncIOScheduler

from brapi_client import BrapiClient
from alerts import AlertManager
from handlers import (
    start_command,
    quote_command,
    portfolio_command,
    alert_command,
    list_alerts_command,
    remove_alert_command,
    search_command,
    compare_command,
)

# Carrega variáveis de ambiente
load_dotenv()

TELEGRAM_TOKEN = os.getenv('TELEGRAM_TOKEN')
BRAPI_TOKEN = os.getenv('BRAPI_TOKEN')


async def check_alerts_job(context: ContextTypes.DEFAULT_TYPE):
    """Job que verifica alertas periodicamente"""
    alert_manager: AlertManager = context.bot_data['alerts']
    brapi: BrapiClient = context.bot_data['brapi']
    
    # Pega todos os símbolos com alertas ativos
    symbols = alert_manager.get_all_symbols()
    
    if not symbols:
        return
    
    # Busca cotações
    quotes = brapi.get_multiple_quotes(symbols)
    
    # Verifica cada cotação
    for quote in quotes:
        triggered_alerts = alert_manager.check_alerts(quote.symbol, quote.price)
        
        for alert in triggered_alerts:
            # Envia notificação
            condition_text = {
                '>': 'ultrapassou',
                '<': 'caiu abaixo de',
                '=': 'atingiu'
            }
            
            msg = f"""🔔 *ALERTA ACIONADO!*

*{quote.symbol}* {condition_text[alert.condition]} R$ {alert.target_price:.2f}

💰 Preço atual: R$ {quote.price:.2f}
📈 Variação: {quote.change_percent:+.2f}%"""
            
            try:
                await context.bot.send_message(
                    chat_id=alert.user_id,
                    text=msg,
                    parse_mode='Markdown'
                )
            except Exception as e:
                print(f"Erro ao enviar alerta: {e}")


async def daily_summary_job(context: ContextTypes.DEFAULT_TYPE):
    """Job que envia resumo diário às 18:00"""
    # Em produção, iterar sobre todos os usuários
    # Aqui simplificado para demonstração
    pass


def main():
    """Função principal"""
    # Cria aplicação
    app = Application.builder().token(TELEGRAM_TOKEN).build()
    
    # Inicializa clientes
    app.bot_data['brapi'] = BrapiClient(BRAPI_TOKEN)
    app.bot_data['alerts'] = AlertManager()
    app.bot_data['portfolios'] = {}
    
    # Registra handlers
    app.add_handler(CommandHandler("start", start_command))
    app.add_handler(CommandHandler("cotacao", quote_command))
    app.add_handler(CommandHandler("carteira", portfolio_command))
    app.add_handler(CommandHandler("alerta", alert_command))
    app.add_handler(CommandHandler("alertas", list_alerts_command))
    app.add_handler(CommandHandler("remover", remove_alert_command))
    app.add_handler(CommandHandler("buscar", search_command))
    app.add_handler(CommandHandler("comparar", compare_command))
    
    # Configura job de verificação de alertas (a cada 5 minutos)
    job_queue = app.job_queue
    job_queue.run_repeating(check_alerts_job, interval=300, first=10)
    
    # Job de resumo diário às 18:00
    # job_queue.run_daily(daily_summary_job, time=datetime.time(18, 0))
    
    print("🤖 Bot iniciado! Pressione Ctrl+C para parar.")
    
    # Inicia o bot
    app.run_polling(allowed_updates=Update.ALL_TYPES)


if __name__ == '__main__':
    main()

# Instala dependências
pip install -r requirements.txt

# Roda o bot
python bot.py

# No servidor
git clone seu-repositorio
cd telegram-bot
pip install -r requirements.txt

# Cria serviço systemd
sudo nano /etc/systemd/system/telegram-bot.service

[Unit]
Description=Telegram Bot Cotações
After=network.target

[Service]
Type=simple
User=ubuntu
WorkingDirectory=/home/ubuntu/telegram-bot
ExecStart=/usr/bin/python3 bot.py
Restart=always

[Install]
WantedBy=multi-user.target

sudo systemctl enable telegram-bot
sudo systemctl start telegram-bot

import matplotlib.pyplot as plt
import io

async def chart_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Gera gráfico da ação"""
    if not context.args:
        return
    
    symbol = context.args[0].upper()
    
    # Busca histórico (necessário endpoint com range)
    # ... código para buscar histórico ...
    
    # Gera gráfico
    plt.figure(figsize=(10, 6))
    plt.plot(dates, prices)
    plt.title(f'{symbol} - Últimos 30 dias')
    plt.xlabel('Data')
    plt.ylabel('Preço (R$)')
    
    # Salva em buffer
    buf = io.BytesIO()
    plt.savefig(buf, format='png')
    buf.seek(0)
    
    await update.message.reply_photo(photo=buf)
    plt.close()

from telegram import InlineQueryResultArticle, InputTextMessageContent

async def inline_query(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Permite consultar em qualquer chat com @seu_bot PETR4"""
    query = update.inline_query.query.upper()
    
    if not query:
        return
    
    brapi: BrapiClient = context.bot_data['brapi']
    quote = brapi.get_quote(query)
    
    if quote:
        results = [
            InlineQueryResultArticle(
                id=quote.symbol,
                title=f"{quote.symbol} - R$ {quote.price:.2f}",
                description=f"{quote.name} | {quote.change_percent:+.2f}%",
                input_message_content=InputTextMessageContent(
                    format_quote(quote),
                    parse_mode='Markdown'
                )
            )
        ]
        await update.inline_query.answer(results)