Como Criar um Bot de Apostas com IA (Tutorial

# modulo_coleta.py
import sqlite3
import requests
import json
from datetime import datetime
from loguru import logger

DATABASE = 'apostas.db'
API_KEY = 'sua_chave_the_odds_api'

def inicializar_banco():
    """Cria tabelas necessárias se não existirem."""
    conn = sqlite3.connect(DATABASE)
    cursor = conn.cursor()
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS jogos (
            id TEXT PRIMARY KEY,
            esporte TEXT,
            liga TEXT,
            time_casa TEXT,
            time_fora TEXT,
            data_jogo TEXT,
            resultado TEXT DEFAULT NULL
        )
    ''')
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS odds_historico (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            jogo_id TEXT,
            casa_aposta TEXT,
            mercado TEXT,
            selecao TEXT,
            odd REAL,
            coletado_em TEXT,
            FOREIGN KEY (jogo_id) REFERENCES jogos(id)
        )
    ''')
    
    conn.commit()
    conn.close()
    logger.info('Banco inicializado com sucesso')

def coletar_e_salvar_odds(esporte: str = 'soccer_brazil_campeonato'):
    """Coleta odds da API e salva no banco local."""
    url = f'https://api.the-odds-api.com/v4/sports/{esporte}/odds'
    params = {
        'apiKey': API_KEY,
        'regions': 'eu',
        'markets': 'h2h',
        'oddsFormat': 'decimal'
    }
    
    try:
        response = requests.get(url, params=params, timeout=10)
        response.raise_for_status()
        jogos = response.json()
        
        conn = sqlite3.connect(DATABASE)
        cursor = conn.cursor()
        agora = datetime.utcnow().isoformat()
        
        for jogo in jogos:
            # Insere ou ignora o jogo (já pode existir de coletas anteriores)
            cursor.execute('''
                INSERT OR IGNORE INTO jogos (id, esporte, liga, time_casa, time_fora, data_jogo)
                VALUES (?, ?, ?, ?, ?, ?)
            ''', (jogo['id'], esporte, jogo.get('sport_title', ''),
                  jogo['home_team'], jogo['away_team'], jogo['commence_time']))
            
            # Salva snapshot das odds atuais
            for bookmaker in jogo.get('bookmakers', []):
                for market in bookmaker.get('markets', []):
                    for outcome in market.get('outcomes', []):
                        cursor.execute('''
                            INSERT INTO odds_historico 
                            (jogo_id, casa_aposta, mercado, selecao, odd, coletado_em)
                            VALUES (?, ?, ?, ?, ?, ?)
                        ''', (jogo['id'], bookmaker['key'], market['key'],
                              outcome['name'], outcome['price'], agora))
        
        conn.commit()
        conn.close()
        logger.info(f'Coletado: {len(jogos)} jogos de {esporte}')
        
    except requests.RequestException as e:
        logger.error(f'Erro na coleta: {e}')

if __name__ == '__main__':
    inicializar_banco()
    coletar_e_salvar_odds()

# modulo_modelo.py
import pandas as pd
import numpy as np
from scipy.stats import poisson
from scipy.optimize import minimize
from typing import Tuple

class ModeloPoisson:
    """
    Modelo de Dixon-Coles simplificado para previsão de futebol.
    Baseado no paper: 'Modelling Association Football Scores' (1997).
    """
    
    def __init__(self):
        self.params = None
        self.times = None
    
    def _log_likelihood(self, params: np.ndarray, dados: pd.DataFrame,
                        times_casa: list, times_fora: list) -> float:
        """Função de log-verossimilhança do modelo de Poisson."""
        n_times = len(self.times)
        ataque = dict(zip(self.times, params[:n_times]))
        defesa = dict(zip(self.times, params[n_times:2*n_times]))
        home_advantage = params[-1]
        
        log_lik = 0
        for _, row in dados.iterrows():
            lambda_casa = np.exp(ataque[row['time_casa']] - 
                                  defesa[row['time_fora']] + home_advantage)
            lambda_fora = np.exp(ataque[row['time_fora']] - 
                                  defesa[row['time_casa']])
            
            log_lik += (poisson.logpmf(row['gols_casa'], lambda_casa) + 
                        poisson.logpmf(row['gols_fora'], lambda_fora))
        
        return -log_lik  # Minimizar negativo = maximizar
    
    def treinar(self, dados: pd.DataFrame):
        """Treina modelo com histórico de jogos."""
        self.times = sorted(list(
            set(dados['time_casa'].unique()) | set(dados['time_fora'].unique())
        ))
        n_times = len(self.times)
        
        # Parâmetros iniciais
        params_iniciais = np.concatenate([
            np.zeros(n_times),   # ataque
            np.zeros(n_times),   # defesa  
            [0.1]                 # home advantage
        ])
        
        resultado = minimize(
            self._log_likelihood,
            params_iniciais,
            args=(dados, self.times, self.times),
            method='L-BFGS-B'
        )
        
        self.params = resultado.x
        print(f'Modelo treinado. Convergiu: {resultado.success}')
    
    def prever(self, time_casa: str, time_fora: str,
               max_gols: int = 10) -> dict:
        """Retorna probabilidades de vitória, empate e derrota."""
        if self.params is None:
            raise ValueError('Treine o modelo antes de prever')
        
        n_times = len(self.times)
        ataque = dict(zip(self.times, self.params[:n_times]))
        defesa = dict(zip(self.times, self.params[n_times:2*n_times]))
        home_adv = self.params[-1]
        
        lambda_casa = np.exp(ataque[time_casa] - defesa[time_fora] + home_adv)
        lambda_fora = np.exp(ataque[time_fora] - defesa[time_casa])
        
        # Matriz de probabilidades de placar
        prob_matrix = np.outer(
            [poisson.pmf(i, lambda_casa) for i in range(max_gols)],
            [poisson.pmf(i, lambda_fora) for i in range(max_gols)]
        )
        
        p_casa = np.sum(np.tril(prob_matrix, -1))
        p_empate = np.sum(np.diag(prob_matrix))
        p_fora = np.sum(np.triu(prob_matrix, 1))
        
        return {
            'p_casa': round(p_casa, 4),
            'p_empate': round(p_empate, 4),
            'p_fora': round(p_fora, 4)
        }

# modulo_banca.py
from typing import Optional

def kelly_criterion(probabilidade_estimada: float,
                    odd_decimal: float,
                    fracao_kelly: float = 0.25) -> float:
    """
    Calcula o percentual da banca a apostar usando Kelly Criterion.
    
    Args:
        probabilidade_estimada: Sua estimativa de probabilidade do evento (0-1)
        odd_decimal: Odd oferecida pela casa no formato decimal
        fracao_kelly: Fração do Kelly pra usar (0.25 = Kelly fracionado, mais conservador)
    
    Returns:
        Percentual da banca a apostar (0 se sem valor)
    """
    # b = lucro líquido por unidade apostada
    b = odd_decimal - 1
    p = probabilidade_estimada
    q = 1 - p  # probabilidade de perder
    
    # Formula de Kelly: f* = (bp - q) / b
    kelly = (b * p - q) / b
    
    if kelly <= 0:
        return 0.0  # Sem valor, não apostar
    
    # Aplicar fração para reduzir variância
    return min(kelly * fracao_kelly, 0.05)  # Máximo de 5% da banca

class GestorBanca:
    """Gerencia a banca e registra apostas."""
    
    def __init__(self, banca_inicial: float):
        self.banca_atual = banca_inicial
        self.banca_inicial = banca_inicial
        self.apostas = []
    
    def calcular_aposta(self, prob_estimada: float,
                         odd: float) -> Optional[float]:
        """Retorna valor a apostar ou None se sem valor."""
        pct_kelly = kelly_criterion(prob_estimada, odd)
        if pct_kelly == 0:
            return None
        return round(self.banca_atual * pct_kelly, 2)
    
    def registrar_resultado(self, valor_apostado: float,
                             odd: float, ganhou: bool):
        """Atualiza banca após resultado."""
        if ganhou:
            lucro = valor_apostado * (odd - 1)
            self.banca_atual += lucro
        else:
            self.banca_atual -= valor_apostado
        
        self.apostas.append({
            'valor': valor_apostado,
            'odd': odd,
            'resultado': 'W' if ganhou else 'L',
            'banca_apos': self.banca_atual
        })
    
    def relatorio(self) -> dict:
        """Resumo da performance."""
        total_apostas = len(self.apostas)
        vitorias = sum(1 for a in self.apostas if a['resultado'] == 'W')
        retorno = (self.banca_atual - self.banca_inicial) / self.banca_inicial
        return {
            'banca_inicial': self.banca_inicial,
            'banca_atual': self.banca_atual,
            'total_apostas': total_apostas,
            'win_rate': f'{vitorias/total_apostas:.1%}' if total_apostas else 'N/A',
            'retorno_total': f'{retorno:.1%}'
        }

# Exemplo
banca = GestorBanca(1000.0)
valor = banca.calcular_aposta(prob_estimada=0.52, odd=2.10)
print(f'Valor sugerido pra apostar: R${valor}')
print(f'Representa {valor/1000:.1%} da banca')

# main.py — Integração de todos os módulos
from modulo_coleta import coletar_e_salvar_odds
from modulo_modelo import ModeloPoisson
from modulo_banca import GestorBanca
from apscheduler.schedulers.blocking import BlockingScheduler
from loguru import logger
import sqlite3
import pandas as pd

MIN_VALUE_EDGE = 0.05   # Só aposta se probabilidade estimada > odd em 5%
PAPER_TRADING = True    # Mude pra False apenas com testes extensivos

def carregar_dados_historicos() -> pd.DataFrame:
    """Carrega jogos com resultado do banco pra treinar o modelo."""
    conn = sqlite3.connect('apostas.db')
    df = pd.read_sql_query('''
        SELECT time_casa, time_fora, gols_casa, gols_fora
        FROM jogos 
        WHERE resultado IS NOT NULL
        ORDER BY data_jogo DESC
        LIMIT 500
    ''', conn)
    conn.close()
    return df

def ciclo_principal():
    """Roda um ciclo completo: coleta → análise → decisão."""
    logger.info('Iniciando ciclo de análise')
    
    # 1. Coleta odds atuais
    coletar_e_salvar_odds()
    
    # 2. Carrega e treina modelo
    historico = carregar_dados_historicos()
    if len(historico) < 100:
        logger.warning('Dados insuficientes pra treinar modelo confiável')
        return
    
    modelo = ModeloPoisson()
    modelo.treinar(historico)
    
    # 3. Analisa jogos disponíveis
    gestor = GestorBanca(banca_inicial=1000.0)
    conn = sqlite3.connect('apostas.db')
    
    # Pega jogos com odds disponíveis que ainda não aconteceram
    jogos_pendentes = pd.read_sql_query('''
        SELECT DISTINCT j.id, j.time_casa, j.time_fora, j.data_jogo,
               o.casa_aposta, o.selecao, o.odd
        FROM jogos j
        JOIN odds_historico o ON j.id = o.jogo_id
        WHERE j.resultado IS NULL
        AND o.mercado = 'h2h'
        AND o.coletado_em = (SELECT MAX(coletado_em) FROM odds_historico)
        AND o.casa_aposta = 'pinnacle'  -- Casa com melhores odds como referência
    ''', conn)
    conn.close()
    
    for _, jogo in jogos_pendentes.iterrows():
        try:
            prob = modelo.prever(jogo['time_casa'], jogo['time_fora'])
        except KeyError:
            continue  # Time sem histórico suficiente
        
        # Checa se há value na seleção disponível
        if jogo['selecao'] == jogo['time_casa']:
            p_estimada = prob['p_casa']
        elif jogo['selecao'] == jogo['time_fora']:
            p_estimada = prob['p_fora']
        else:
            p_estimada = prob['p_empate']
        
        p_implicita = 1 / jogo['odd']
        edge = p_estimada - p_implicita
        
        if edge >= MIN_VALUE_EDGE:
            valor = gestor.calcular_aposta(p_estimada, jogo['odd'])
            if valor and valor > 0:
                modo = 'SIMULADO' if PAPER_TRADING else 'REAL'
                logger.info(
                    f'[{modo}] Value bet encontrado: '
                    f'{jogo["time_casa"]} vs {jogo["time_fora"]} | '
                    f'Seleção: {jogo["selecao"]} | '
                    f'Odd: {jogo["odd"]} | '
                    f'Prob estimada: {p_estimada:.1%} | '
                    f'Edge: {edge:.1%} | '
                    f'Apostar: R${valor:.2f}'
                )

if __name__ == '__main__':
    scheduler = BlockingScheduler()
    scheduler.add_job(ciclo_principal, 'interval', hours=1)
    logger.info('Bot iniciado. Rodando a cada hora.')
    scheduler.start()