Web Scraping de Odds com Python: Tutorial

# Criar ambiente virtual
python -m venv odds-env
source odds-env/bin/activate  # Linux/Mac
# odds-env\Scripts\activate  # Windows

# Instalar dependências
pip install requests beautifulsoup4 selenium pandas sqlalchemy
pip install webdriver-manager  # gerencia ChromeDriver automaticamente
pip install python-dotenv      # pra variáveis de ambiente

odds-scraper/
├── .env                 # chaves de API
├── config.py            # configurações gerais
├── collectors/
│   ├── odds_api.py      # coleta via API
│   └── scraper.py       # coleta via scraping
├── storage/
│   └── database.py      # SQLite / PostgreSQL
├── analysis/
│   └── compare.py       # comparação de odds
└── main.py              # orquestrador

import requests
import os
from dotenv import load_dotenv
from datetime import datetime

load_dotenv()

API_KEY = os.getenv("ODDS_API_KEY")
BASE_URL = "https://api.the-odds-api.com/v4/sports"

def get_odds(sport: str = "soccer_brazil_serie_a", regions: str = "us,eu", markets: str = "h2h,spreads,totals") -> list[dict]:
    """Coleta odds de todas as casas disponíveis pra um esporte."""
    url = f"{BASE_URL}/{sport}/odds"
    params = {
        "apiKey": API_KEY,
        "regions": regions,
        "markets": markets,
        "oddsFormat": "decimal",
        "dateFormat": "iso",
    }
    response = requests.get(url, params=params, timeout=30)
    response.raise_for_status()
    
    remaining = response.headers.get("x-requests-remaining", "?")
    print(f"[{datetime.now():%H:%M}] Requests restantes: {remaining}")
    
    return response.json()

def extract_best_odds(events: list[dict]) -> list[dict]:
    """Pra cada evento, encontra a melhor odd em cada outcome."""
    results = []
    for event in events:
        game = {
            "id": event["id"],
            "home": event["home_team"],
            "away": event["away_team"],
            "start": event["commence_time"],
            "best_odds": {},
        }
        for bookmaker in event.get("bookmakers", []):
            for market in bookmaker.get("markets", []):
                for outcome in market.get("outcomes", []):
                    key = f"{market['key']}_{outcome['name']}"
                    current_best = game["best_odds"].get(key, {"price": 0})
                    if outcome["price"] > current_best.get("price", 0):
                        game["best_odds"][key] = {
                            "price": outcome["price"],
                            "bookmaker": bookmaker["title"],
                        }
        results.append(game)
    return results

# Uso
events = get_odds("soccer_brazil_serie_a")
best = extract_best_odds(events)
for game in best:
    print(f"\n{game['home']} vs {game['away']}")
    for market, info in game["best_odds"].items():
        print(f"  {market}: {info['price']:.2f} ({info['bookmaker']})")

from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from webdriver_manager.chrome import ChromeDriverManager
from bs4 import BeautifulSoup
import time
import json

def create_driver() -> webdriver.Chrome:
    """Cria driver Chrome em modo headless."""
    options = Options()
    options.add_argument("--headless=new")
    options.add_argument("--no-sandbox")
    options.add_argument("--disable-dev-shm-usage")
    options.add_argument("user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36")
    
    service = Service(ChromeDriverManager().install())
    return webdriver.Chrome(service=service, options=options)

def scrape_odds_page(url: str) -> list[dict]:
    """Faz scraping de uma pagina de odds."""
    driver = create_driver()
    results = []
    
    try:
        driver.get(url)
        # Esperar conteúdo dinâmico carregar
        WebDriverWait(driver, 15).until(
            EC.presence_of_element_located((By.CSS_SELECTOR, ".event-row, .match-row, [data-testid='event']"))
        )
        time.sleep(2)  # buffer pra JS terminar
        
        soup = BeautifulSoup(driver.page_source, "html.parser")
        
        # Seletor genérico - adapte pro site alvo
        events = soup.select(".event-row, .match-row")
        for event in events:
            teams = event.select(".team-name, .participant")
            odds_els = event.select(".odd-value, .price, [data-odd]")
            
            if len(teams) >= 2 and len(odds_els) >= 2:
                results.append({
                    "home": teams[0].get_text(strip=True),
                    "away": teams[1].get_text(strip=True),
                    "odds": [el.get_text(strip=True) for el in odds_els],
                })
    finally:
        driver.quit()
    
    return results

# Exemplo de uso (adapte a URL pro site que voce quer)
odds = scrape_odds_page("https://example-odds-site.com/football/brazil")
for match in odds:
    print(f"{match['home']} vs {match['away']}: {match['odds']}")

from sqlalchemy import create_engine, Column, Integer, Float, String, DateTime
from sqlalchemy.orm import declarative_base, sessionmaker
from datetime import datetime

Base = declarative_base()

class OddsRecord(Base):
    __tablename__ = "odds"
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    event_id = Column(String, index=True)
    sport = Column(String)
    home_team = Column(String)
    away_team = Column(String)
    bookmaker = Column(String)
    market = Column(String)
    outcome = Column(String)
    price = Column(Float)
    collected_at = Column(DateTime, default=datetime.utcnow)
    commence_time = Column(DateTime)

# Setup banco
engine = create_engine("sqlite:///odds_data.db")
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

def save_odds(events: list[dict], sport: str):
    """Salva odds no banco."""
    session = Session()
    count = 0
    for event in events:
        for bookmaker in event.get("bookmakers", []):
            for market in bookmaker.get("markets", []):
                for outcome in market.get("outcomes", []):
                    record = OddsRecord(
                        event_id=event["id"],
                        sport=sport,
                        home_team=event["home_team"],
                        away_team=event["away_team"],
                        bookmaker=bookmaker["title"],
                        market=market["key"],
                        outcome=outcome["name"],
                        price=outcome["price"],
                        commence_time=datetime.fromisoformat(event["commence_time"].replace("Z", "+00:00")),
                    )
                    session.add(record)
                    count += 1
    session.commit()
    session.close()
    print(f"Salvos {count} registros de odds")

# Pipeline completo
events = get_odds("soccer_brazil_serie_a")
save_odds(events, "soccer_brazil_serie_a")

import pandas as pd

def find_best_odds_per_event(db_path: str = "sqlite:///odds_data.db") -> pd.DataFrame:
    """Encontra a melhor odd pra cada outcome de cada evento."""
    engine = create_engine(db_path)
    query = """
        SELECT event_id, home_team, away_team, market, outcome,
               MAX(price) as best_price,
               bookmaker
        FROM odds
        WHERE collected_at > datetime('now', '-2 hours')
        GROUP BY event_id, market, outcome
        ORDER BY event_id, market
    """
    df = pd.read_sql(query, engine)
    return df

# Ver resultados
df = find_best_odds_per_event()
print(df.to_string(index=False))

import schedule
import time
from collectors.odds_api import get_odds, extract_best_odds
from storage.database import save_odds

SPORTS = [
    "soccer_brazil_serie_a",
    "soccer_brazil_serie_b",
    "soccer_epl",
]

def collect_and_store():
    """Pipeline principal: coleta e armazena."""
    for sport in SPORTS:
        try:
            events = get_odds(sport)
            save_odds(events, sport)
            best = extract_best_odds(events)
            
            # Checa oportunidades de arbitragem
            for game in best:
                check_arbitrage(game)
                
            print(f"OK: {sport} - {len(events)} eventos")
        except Exception as e:
            print(f"ERRO: {sport} - {e}")

def check_arbitrage(game: dict):
    """Verifica se existe oportunidade de arbitragem."""
    h2h_odds = {}
    for key, info in game["best_odds"].items():
        if key.startswith("h2h_"):
            h2h_odds[key] = info["price"]
    
    if len(h2h_odds) >= 2:
        inv_sum = sum(1 / odd for odd in h2h_odds.values())
        if inv_sum < 1.0:
            margin = (1 - inv_sum) * 100
            print(f"  ARBITRAGEM! {game['home']} vs {game['away']} - margem: {margin:.2f}%")

# Rodar a cada 30 minutos
schedule.every(30).minutes.do(collect_and_store)

print("Pipeline iniciado. Ctrl+C pra parar.")
collect_and_store()  # roda uma vez imediatamente
while True:
    schedule.run_pending()
    time.sleep(60)