""" Importa il dataset (data/processed/matches.csv) in un database MySQL "tigertips" con tabelle normalizzate, pronte per essere usate dal futuro sito web. Schema: leagues - le leghe (codice football-data, nome, paese) teams - le squadre (nome unico) matches - le partite: risultato, statistiche, quote, probabilità implicite Lo script è idempotente: si può rilanciare dopo ogni nuovo download e aggiorna le partite esistenti (chiave unica: data + squadra casa + squadra ospite). Uso: python scripts/import_db.py python scripts/import_db.py --host 127.0.0.1 --port 3306 --user root --password "" """ import argparse import sys from pathlib import Path import numpy as np import pandas as pd import pymysql import config PROJECT_ROOT = Path(__file__).resolve().parent.parent CSV_PATH = PROJECT_ROOT / "data" / "processed" / "matches.csv" DB_NAME = "tigertips" LEAGUES = { "I1": ("Serie A", "Italia"), "I2": ("Serie B", "Italia"), "E0": ("Premier League", "Inghilterra"), "SP1": ("La Liga", "Spagna"), "D1": ("Bundesliga", "Germania"), "F1": ("Ligue 1", "Francia"), } SCHEMA = """ CREATE TABLE IF NOT EXISTS leagues ( id TINYINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, code VARCHAR(5) NOT NULL UNIQUE, name VARCHAR(50) NOT NULL, country VARCHAR(30) NOT NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; CREATE TABLE IF NOT EXISTS teams ( id SMALLINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, name VARCHAR(60) NOT NULL UNIQUE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; CREATE TABLE IF NOT EXISTS matches ( id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY, league_id TINYINT UNSIGNED NOT NULL, season VARCHAR(7) NOT NULL, -- es. '2023/24' match_date DATE NOT NULL, match_time TIME NULL, home_team_id SMALLINT UNSIGNED NOT NULL, away_team_id SMALLINT UNSIGNED NOT NULL, -- Risultato (ht = primo tempo, result: H/D/A) home_goals TINYINT UNSIGNED NULL, away_goals TINYINT UNSIGNED NULL, result CHAR(1) NULL, ht_home_goals TINYINT UNSIGNED NULL, ht_away_goals TINYINT UNSIGNED NULL, ht_result CHAR(1) NULL, -- Statistiche di gioco home_shots TINYINT UNSIGNED NULL, away_shots TINYINT UNSIGNED NULL, home_shots_target TINYINT UNSIGNED NULL, away_shots_target TINYINT UNSIGNED NULL, home_corners TINYINT UNSIGNED NULL, away_corners TINYINT UNSIGNED NULL, home_fouls TINYINT UNSIGNED NULL, away_fouls TINYINT UNSIGNED NULL, home_yellow TINYINT UNSIGNED NULL, away_yellow TINYINT UNSIGNED NULL, home_red TINYINT UNSIGNED NULL, away_red TINYINT UNSIGNED NULL, -- Quote 1X2 pre-match b365_home DECIMAL(7,3) NULL, b365_draw DECIMAL(7,3) NULL, b365_away DECIMAL(7,3) NULL, pinnacle_home DECIMAL(7,3) NULL, pinnacle_draw DECIMAL(7,3) NULL, pinnacle_away DECIMAL(7,3) NULL, avg_home DECIMAL(7,3) NULL, avg_draw DECIMAL(7,3) NULL, avg_away DECIMAL(7,3) NULL, max_home DECIMAL(7,3) NULL, max_draw DECIMAL(7,3) NULL, max_away DECIMAL(7,3) NULL, -- Quote 1X2 di chiusura (closing) b365c_home DECIMAL(7,3) NULL, b365c_draw DECIMAL(7,3) NULL, b365c_away DECIMAL(7,3) NULL, pinnaclec_home DECIMAL(7,3) NULL, pinnaclec_draw DECIMAL(7,3) NULL, pinnaclec_away DECIMAL(7,3) NULL, avgc_home DECIMAL(7,3) NULL, avgc_draw DECIMAL(7,3) NULL, avgc_away DECIMAL(7,3) NULL, maxc_home DECIMAL(7,3) NULL, maxc_draw DECIMAL(7,3) NULL, maxc_away DECIMAL(7,3) NULL, -- Over/Under 2.5 b365_over25 DECIMAL(7,3) NULL, b365_under25 DECIMAL(7,3) NULL, avg_over25 DECIMAL(7,3) NULL, avg_under25 DECIMAL(7,3) NULL, -- Probabilità implicite normalizzate (senza margine del bookmaker) imp_prob_home DECIMAL(6,4) NULL, imp_prob_draw DECIMAL(6,4) NULL, imp_prob_away DECIMAL(6,4) NULL, imp_prob_source VARCHAR(20) NULL, UNIQUE KEY uq_match (match_date, home_team_id, away_team_id), KEY idx_league_season (league_id, season), KEY idx_date (match_date), KEY idx_home_team (home_team_id), KEY idx_away_team (away_team_id), CONSTRAINT fk_matches_league FOREIGN KEY (league_id) REFERENCES leagues (id), CONSTRAINT fk_matches_home FOREIGN KEY (home_team_id) REFERENCES teams (id), CONSTRAINT fk_matches_away FOREIGN KEY (away_team_id) REFERENCES teams (id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; """ # Mappa colonna CSV -> colonna DB (nell'ordine dell'INSERT) CSV_TO_DB = { "FTHG": "home_goals", "FTAG": "away_goals", "FTR": "result", "HTHG": "ht_home_goals", "HTAG": "ht_away_goals", "HTR": "ht_result", "HS": "home_shots", "AS": "away_shots", "HST": "home_shots_target", "AST": "away_shots_target", "HC": "home_corners", "AC": "away_corners", "HF": "home_fouls", "AF": "away_fouls", "HY": "home_yellow", "AY": "away_yellow", "HR": "home_red", "AR": "away_red", "B365H": "b365_home", "B365D": "b365_draw", "B365A": "b365_away", "PSH": "pinnacle_home", "PSD": "pinnacle_draw", "PSA": "pinnacle_away", "AvgH": "avg_home", "AvgD": "avg_draw", "AvgA": "avg_away", "MaxH": "max_home", "MaxD": "max_draw", "MaxA": "max_away", "B365CH": "b365c_home", "B365CD": "b365c_draw", "B365CA": "b365c_away", "PSCH": "pinnaclec_home", "PSCD": "pinnaclec_draw", "PSCA": "pinnaclec_away", "AvgCH": "avgc_home", "AvgCD": "avgc_draw", "AvgCA": "avgc_away", "MaxCH": "maxc_home", "MaxCD": "maxc_draw", "MaxCA": "maxc_away", "B365>2.5": "b365_over25", "B365<2.5": "b365_under25", "Avg>2.5": "avg_over25", "Avg<2.5": "avg_under25", "ImpProbH": "imp_prob_home", "ImpProbD": "imp_prob_draw", "ImpProbA": "imp_prob_away", "ImpProbSource": "imp_prob_source", } def connect(args, database: str | None = None) -> pymysql.Connection: return pymysql.connect( host=args.host, port=args.port, user=args.user, password=args.password, database=database, charset="utf8mb4", autocommit=False, ) def to_py(value): """Converte NaN/tipi numpy in tipi Python accettati dal driver MySQL.""" if value is None or (isinstance(value, float) and np.isnan(value)) or pd.isna(value): return None if isinstance(value, (np.integer,)): return int(value) if isinstance(value, (np.floating,)): return float(value) return value def main() -> int: parser = argparse.ArgumentParser(description="Importa matches.csv nel database MySQL tigertips") config.add_db_args(parser) args = parser.parse_args() if not CSV_PATH.exists(): print(f"Dataset non trovato: {CSV_PATH}\nEsegui prima: python scripts/download_data.py") return 1 df = pd.read_csv(CSV_PATH, low_memory=False) df["Date"] = pd.to_datetime(df["Date"]).dt.date print(f"Dataset: {len(df)} partite da importare") # --- Creazione database e tabelle --- conn = connect(args) with conn.cursor() as cur: cur.execute(f"CREATE DATABASE IF NOT EXISTS {DB_NAME} " "CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci") conn.commit() conn.close() conn = connect(args, DB_NAME) cur = conn.cursor() for statement in SCHEMA.split(";"): if statement.strip(): cur.execute(statement) conn.commit() # --- Leghe --- cur.executemany( "INSERT INTO leagues (code, name, country) VALUES (%s, %s, %s) " "ON DUPLICATE KEY UPDATE name = VALUES(name), country = VALUES(country)", [(code, name, country) for code, (name, country) in LEAGUES.items()], ) conn.commit() cur.execute("SELECT code, id FROM leagues") league_ids = dict(cur.fetchall()) # --- Squadre --- team_names = sorted(set(df["HomeTeam"]) | set(df["AwayTeam"])) cur.executemany("INSERT IGNORE INTO teams (name) VALUES (%s)", [(n,) for n in team_names]) conn.commit() cur.execute("SELECT name, id FROM teams") team_ids = dict(cur.fetchall()) print(f"Squadre: {len(team_ids)}") # --- Partite (upsert a blocchi) --- db_cols = ["league_id", "season", "match_date", "match_time", "home_team_id", "away_team_id"] + list(CSV_TO_DB.values()) placeholders = ", ".join(["%s"] * len(db_cols)) updates = ", ".join(f"{c} = VALUES({c})" for c in db_cols if c not in ("match_date", "home_team_id", "away_team_id")) sql = (f"INSERT INTO matches ({', '.join(db_cols)}) VALUES ({placeholders}) " f"ON DUPLICATE KEY UPDATE {updates}") rows = [] for _, r in df.iterrows(): time_val = to_py(r.get("Time")) rows.append(tuple( [league_ids[r["Div"]], r["Season"], r["Date"], time_val, team_ids[r["HomeTeam"]], team_ids[r["AwayTeam"]]] + [to_py(r.get(csv_col)) for csv_col in CSV_TO_DB] )) batch_size = 1000 for start in range(0, len(rows), batch_size): cur.executemany(sql, rows[start:start + batch_size]) conn.commit() print(f" importate {min(start + batch_size, len(rows))}/{len(rows)}", end="\r") print() # --- Riepilogo --- cur.execute("SELECT COUNT(*) FROM matches") total = cur.fetchone()[0] cur.execute(""" SELECT l.name, COUNT(*), MIN(m.match_date), MAX(m.match_date) FROM matches m JOIN leagues l ON l.id = m.league_id GROUP BY l.name ORDER BY COUNT(*) DESC """) print(f"\n=== Database '{DB_NAME}' ===") print(f"Partite in tabella: {total}") for name, count, dmin, dmax in cur.fetchall(): print(f" {name:<16} {count:>6} {dmin} -> {dmax}") cur.close() conn.close() return 0 if __name__ == "__main__": sys.exit(main())