""" BET.CUTTALO.COM - DATA COLLECTOR Raccoglie dati sportivi da API gratuite per training rete neurale """ import requests import pandas as pd import json import os from datetime import datetime, timedelta from typing import Dict, List, Optional import time DATA_DIR = os.path.join(os.path.dirname(__file__), '..', 'data') os.makedirs(DATA_DIR, exist_ok=True) class FootballDataCollector: """ Raccoglie dati da football-data.org (API gratuita) - 10 richieste/minuto free tier - Storico partite, classifiche, statistiche """ BASE_URL = "https://api.football-data.org/v4" # Competizioni disponibili gratis COMPETITIONS = { 'PL': 'Premier League', 'PD': 'La Liga', 'SA': 'Serie A', 'BL1': 'Bundesliga', 'FL1': 'Ligue 1', 'CL': 'Champions League', 'EC': 'European Championship', 'WC': 'World Cup' } def __init__(self, api_key: Optional[str] = None): self.api_key = api_key or os.environ.get('FOOTBALL_DATA_API_KEY', '') self.headers = {'X-Auth-Token': self.api_key} if self.api_key else {} self.request_count = 0 self.last_request = datetime.now() def _rate_limit(self): """Rispetta rate limit: 10 req/min""" self.request_count += 1 if self.request_count >= 10: elapsed = (datetime.now() - self.last_request).seconds if elapsed < 60: time.sleep(60 - elapsed) self.request_count = 0 self.last_request = datetime.now() def _get(self, endpoint: str) -> Dict: """GET request con rate limiting""" self._rate_limit() url = f"{self.BASE_URL}/{endpoint}" try: response = requests.get(url, headers=self.headers, timeout=30) response.raise_for_status() return response.json() except Exception as e: print(f"Errore API: {e}") return {} def get_matches(self, competition: str, season: int = 2024) -> List[Dict]: """Ottieni tutte le partite di una competizione""" data = self._get(f"competitions/{competition}/matches?season={season}") return data.get('matches', []) def get_standings(self, competition: str, season: int = 2024) -> List[Dict]: """Ottieni classifica""" data = self._get(f"competitions/{competition}/standings?season={season}") standings = data.get('standings', []) if standings: return standings[0].get('table', []) return [] def get_team_stats(self, team_id: int) -> Dict: """Ottieni statistiche squadra""" return self._get(f"teams/{team_id}") def collect_historical_data(self, seasons: List[int] = None) -> pd.DataFrame: """ Raccoglie dati storici per training Ritorna DataFrame con feature per ogni partita """ if seasons is None: seasons = [2021, 2022, 2023, 2024] all_matches = [] for comp_code in ['PL', 'SA', 'PD', 'BL1']: # Top 4 campionati print(f"Raccogliendo {self.COMPETITIONS[comp_code]}...") for season in seasons: print(f" Stagione {season}...") matches = self.get_matches(comp_code, season) standings = self.get_standings(comp_code, season) # Crea mappa posizione squadre team_positions = {} for team in standings: team_positions[team['team']['id']] = { 'position': team['position'], 'points': team['points'], 'won': team['won'], 'draw': team['draw'], 'lost': team['lost'], 'goalsFor': team['goalsFor'], 'goalsAgainst': team['goalsAgainst'], 'goalDifference': team['goalDifference'] } for match in matches: if match['status'] != 'FINISHED': continue home_id = match['homeTeam']['id'] away_id = match['awayTeam']['id'] home_stats = team_positions.get(home_id, {}) away_stats = team_positions.get(away_id, {}) # Feature engineering match_data = { 'match_id': match['id'], 'date': match['utcDate'], 'competition': comp_code, 'season': season, 'home_team': match['homeTeam']['name'], 'away_team': match['awayTeam']['name'], 'home_team_id': home_id, 'away_team_id': away_id, # Risultato (target) 'home_goals': match['score']['fullTime']['home'], 'away_goals': match['score']['fullTime']['away'], 'result': self._get_result(match), # Feature casa 'home_position': home_stats.get('position', 10), 'home_points': home_stats.get('points', 0), 'home_won': home_stats.get('won', 0), 'home_draw': home_stats.get('draw', 0), 'home_lost': home_stats.get('lost', 0), 'home_goals_for': home_stats.get('goalsFor', 0), 'home_goals_against': home_stats.get('goalsAgainst', 0), 'home_goal_diff': home_stats.get('goalDifference', 0), # Feature trasferta 'away_position': away_stats.get('position', 10), 'away_points': away_stats.get('points', 0), 'away_won': away_stats.get('won', 0), 'away_draw': away_stats.get('draw', 0), 'away_lost': away_stats.get('lost', 0), 'away_goals_for': away_stats.get('goalsFor', 0), 'away_goals_against': away_stats.get('goalsAgainst', 0), 'away_goal_diff': away_stats.get('goalDifference', 0), # Feature derivate 'position_diff': home_stats.get('position', 10) - away_stats.get('position', 10), 'points_diff': home_stats.get('points', 0) - away_stats.get('points', 0), } all_matches.append(match_data) time.sleep(1) # Pausa tra stagioni df = pd.DataFrame(all_matches) # Salva dati output_path = os.path.join(DATA_DIR, 'historical_matches.csv') df.to_csv(output_path, index=False) print(f"Salvati {len(df)} match in {output_path}") return df def _get_result(self, match: Dict) -> int: """0 = away win, 1 = draw, 2 = home win""" home = match['score']['fullTime']['home'] away = match['score']['fullTime']['away'] if home > away: return 2 elif home < away: return 0 return 1 def get_upcoming_matches(self) -> List[Dict]: """Ottieni partite future per predizioni""" upcoming = [] for comp_code in ['PL', 'SA', 'PD', 'BL1']: data = self._get(f"competitions/{comp_code}/matches?status=SCHEDULED") matches = data.get('matches', [])[:10] # Prossime 10 per campionato standings = self.get_standings(comp_code) team_positions = {t['team']['id']: t for t in standings} for match in matches: home_id = match['homeTeam']['id'] away_id = match['awayTeam']['id'] home_stats = team_positions.get(home_id, {}) away_stats = team_positions.get(away_id, {}) upcoming.append({ 'match_id': match['id'], 'date': match['utcDate'], 'competition': comp_code, 'competition_name': self.COMPETITIONS[comp_code], 'home_team': match['homeTeam']['name'], 'away_team': match['awayTeam']['name'], 'home_team_id': home_id, 'away_team_id': away_id, 'home_position': home_stats.get('position', 10), 'away_position': away_stats.get('position', 10), 'home_points': home_stats.get('points', 0), 'away_points': away_stats.get('points', 0), 'home_goal_diff': home_stats.get('goalDifference', 0), 'away_goal_diff': away_stats.get('goalDifference', 0), }) return upcoming class OddsCollector: """ Raccoglie quote da The Odds API (gratuito 500 req/mese) """ BASE_URL = "https://api.the-odds-api.com/v4" def __init__(self, api_key: Optional[str] = None): self.api_key = api_key or os.environ.get('ODDS_API_KEY', '') def get_odds(self, sport: str = 'soccer_epl') -> List[Dict]: """Ottieni quote per uno sport""" if not self.api_key: return [] url = f"{self.BASE_URL}/sports/{sport}/odds" params = { 'apiKey': self.api_key, 'regions': 'eu', 'markets': 'h2h', 'oddsFormat': 'decimal' } try: response = requests.get(url, params=params, timeout=30) response.raise_for_status() return response.json() except Exception as e: print(f"Errore Odds API: {e}") return [] if __name__ == "__main__": # Test raccolta dati collector = FootballDataCollector() # Raccogli dati storici print("Raccogliendo dati storici...") df = collector.collect_historical_data(seasons=[2023, 2024]) print(f"\nRaccolti {len(df)} match") print(df.head())