""" Corso Machine Learning Engineer - Modulo 03, Lezione 1 "Il processo di un progetto di ML: split, data leakage e metriche che mentono" OBIETTIVO DI QUESTO FILE ------------------------ Tre esercizi sul metodo, non sugli algoritmi. Il dominio è nuovo rispetto all'articolo: il controllo qualità delle saldature di un pannello metallico. Ogni pannello viene misurato e il laboratorio registra TRE righe identiche per lo stesso pannello (tre letture dello stesso strumento). Quelle righe duplicate sono la sorgente di data leakage più comune nei dati industriali reali. 1. le metriche si ricavano dalle quattro caselle della matrice di confusione: scriverle una volta a mano serve a non confondere mai più precision e recall; 2. la soglia giusta non è 0.5 ne' la formula teorica: è quella che minimizza il costo, e va cercata; 3. se le righe dello stesso pannello finiscono da entrambe le parti dello split, il modello viene valutato su dati che ha già visto. Qui l'effetto è estremo e DIMOSTRABILE: accuratezza esattamente 1.000. COME SI USA ----------- python3 esercizi.py Il file gira anche così com'è: stampa quali esercizi mancano ancora. Completa i TODO uno alla volta e rilancia. Le soluzioni di riferimento sono commentate in fondo al file. AMBIENTE DICHIARATO ------------------- Python 3.12+ e NumPy 2.3 (solo NumPy, nessun scikit-learn: il punto della lezione è capire il metodo, non chiamare una libreria). python3 -m venv .venv && source .venv/bin/activate pip install "numpy>=2.0" Gli esercizi 1 e 2 sono DETERMINISTICI: i numeri attesi scritti in fondo si ricavano a mano con carta e penna. L'esercizio 3 usa un generatore con seed fissato: il primo numero (1.000) è garantito dalla costruzione dei dati, il secondo dipende dal seed ed è il valore da leggere a schermo. """ import numpy as np SEED = 7 # Costi illustrativi, in unità arbitrarie, dichiarati come inventati: # lasciare passare una saldatura difettosa costa 40, fermare per niente un # pannello buono costa 2. COSTO_FN = 40.0 COSTO_FP = 2.0 # ---------------------------------------------------------------------- # ESERCIZIO 1 - Le quattro caselle # # Un collaudo automatico su 1200 saldature produce questa matrice: # # previsto difettoso previsto buono # reale difettoso 18 42 # reale buono 6 1134 # # Scrivi la funzione che ne ricava accuratezza, precision, recall e F1. # Nessuna libreria: solo le quattro formule. # # Domanda a cui rispondere a parole: l'accuratezza di questo modello batte # quella del collaudatore che dichiara "buono" ogni pezzo? # ---------------------------------------------------------------------- def metriche_da_matrice(vp, fp, fn, vn): """Ricava le quattro metriche dalle quattro caselle della matrice. Parametri --------- vp, fp, fn, vn : int veri positivi, falsi positivi, falsi negativi, veri negativi. "Positivo" = "saldatura difettosa", la classe rara che vogliamo trovare. Ritorna ------- dict con chiavi "accuratezza", "precision", "recall", "f1", oppure None se l'esercizio non è ancora svolto. Attenzione alle divisioni per zero: se il modello non dichiara mai "difettoso" allora vp + fp = 0 e la precision non è definita. In quel caso restituisci 0.0, dichiarandolo (è una convenzione, non un teorema). """ # TODO (esercizio 1): sostituisci `return None` con il calcolo. return None # ---------------------------------------------------------------------- # ESERCIZIO 2 - La soglia che costa meno # # Dieci saldature, con il punteggio del modello e il verdetto vero del # laboratorio. Cerca la soglia che minimizza il costo totale. # # Il costo di una soglia t è: # COSTO_FN * (quanti difettosi hanno punteggio < t) # + COSTO_FP * (quanti buoni hanno punteggio >= t) # # Le soglie da provare sono i punteggi stessi: tra due punteggi consecutivi la # previsione non cambia, quindi il costo non cambia. Provarne altre è sprecato. # # Domanda: la soglia trovata coincide con la soglia teorica # COSTO_FP / (COSTO_FP + COSTO_FN)? Se no, che cosa ti dice del modello? # ---------------------------------------------------------------------- PUNTEGGI = np.array([0.95, 0.80, 0.62, 0.55, 0.40, 0.33, 0.21, 0.12, 0.07, 0.02]) DIFETTOSE = np.array([1, 1, 0, 1, 0, 1, 0, 0, 0, 0]) def costo_totale(y, punteggi, soglia): """Costo di una soglia. Già scritta: serve all'esercizio 2 e al runner.""" previsto_difettoso = punteggi >= soglia falsi_negativi = int((~previsto_difettoso & (y == 1)).sum()) falsi_positivi = int((previsto_difettoso & (y == 0)).sum()) return COSTO_FN * falsi_negativi + COSTO_FP * falsi_positivi def soglia_di_costo_minimo(y, punteggi): """Restituisce la soglia, tra i punteggi osservati, a costo minimo. Ritorna ------- tupla (soglia, costo) oppure None se l'esercizio non è ancora svolto. Suggerimento: cicla su `np.unique(punteggi)` e tieni il minimo. Se due soglie pareggiano, scegli la più alta: a parità di costo, meno allarmi. """ # TODO (esercizio 2): sostituisci `return None` con la ricerca. return None # ---------------------------------------------------------------------- # ESERCIZIO 3 - Lo split che mente # # 90 pannelli, ciascuno misurato tre volte: 270 righe, di cui 180 sono copie # esatte di altre. Due modi di dividere i dati: # # (a) SBAGLIATO - si prende una riga per pannello e la si mette nel test. # È esattamente cio' che fa uno split casuale per riga quando i # duplicati non sono dichiarati. Ogni riga di test ha due gemelle # identiche nel training: il vicino più prossimo sta a distanza ZERO e # ha la stessa etichetta. L'accuratezza è 1.000 per costruzione, non # per bravura. # # (b) GIUSTO - tutte e tre le righe di un pannello stanno dalla stessa parte. # Il modello viene valutato su pannelli che non ha mai visto. # # Scrivi tu lo split (b). Lo split (a) e il classificatore sono già pronti. # ---------------------------------------------------------------------- def genera_pannelli(n_pannelli, rng): """Dati simulati, e dichiarati tali. Per ogni pannello: due misure (penetrazione in mm, indice di porosità) e un verdetto. Ogni pannello compare in TRE righe identiche. Ritorna (X, y, pannello_di_riga) con 3 * n_pannelli righe. """ penetrazione = rng.normal(4.0, 0.8, n_pannelli) porosita = rng.normal(2.0, 0.9, n_pannelli) # regola vera + rumore sull'etichetta (annotazione imperfetta del laboratorio) rischio = 1.4 * porosita - 1.0 * penetrazione + rng.normal(0.0, 0.6, n_pannelli) verdetto = (rischio > -1.0).astype(int) X_pannelli = np.column_stack([penetrazione, porosita]) X = np.repeat(X_pannelli, 3, axis=0) y = np.repeat(verdetto, 3) pannello_di_riga = np.repeat(np.arange(n_pannelli), 3) return X, y, pannello_di_riga def split_per_riga(pannello_di_riga, rng): """Split SBAGLIATO: una riga per pannello finisce nel test. Già scritto. Restituisce una maschera booleana `in_test`. """ in_test = np.zeros(len(pannello_di_riga), dtype=bool) for p in np.unique(pannello_di_riga): righe = np.where(pannello_di_riga == p)[0] in_test[rng.choice(righe)] = True return in_test def split_per_pannello(pannello_di_riga, frazione_test, rng): """Split GIUSTO: i pannelli, non le righe, vanno da una parte o dall'altra. Parametri --------- pannello_di_riga : np.ndarray (n_righe,) - id del pannello di ogni riga. frazione_test : float tra 0 e 1 - quota di PANNELLI nel test. rng : np.random.Generator Ritorna ------- np.ndarray booleano (n_righe,): True dove la riga va nel test, oppure None se l'esercizio non è ancora svolto. Suggerimento in tre passi: 1. `pannelli = np.unique(pannello_di_riga)`; 2. mescolali con `rng.permutation` e prendi i primi `round(frazione_test * len(pannelli))`; 3. `np.isin(pannello_di_riga, pannelli_di_test)`. """ # TODO (esercizio 3): sostituisci `return None` con lo split per gruppo. return None def knn1_predici(X_tr, y_tr, X_te): """Classificatore a un solo vicino, scritto a mano. Già pronto. Per ogni riga di test cerca la riga di training più vicina in distanza euclidea e ne copia l'etichetta. È il modello più esposto al leakage da duplicati: se il gemello identico è nel training, non sbaglia mai. """ differenze = X_te[:, None, :] - X_tr[None, :, :] distanze = np.sqrt((differenze ** 2).sum(axis=2)) return y_tr[distanze.argmin(axis=1)] # ---------------------------------------------------------------------- # Runner # ---------------------------------------------------------------------- def _sezione(titolo): print() print(titolo) print("-" * len(titolo)) def _da_fare(numero): print(f" [ ] Esercizio {numero} non ancora svolto: completa il TODO qui sopra.") def esegui_esercizio_1(): _sezione("Esercizio 1 - le quattro caselle") vp, fp, fn, vn = 18, 6, 42, 1134 m = metriche_da_matrice(vp, fp, fn, vn) if m is None: _da_fare(1) return False print(f" accuratezza {m['accuratezza']:.4f} precision {m['precision']:.4f}" f" recall {m['recall']:.4f} F1 {m['f1']:.4f}") pigro = metriche_da_matrice(0, 0, vp + fn, fp + vn) print(f" collaudatore pigro ('tutto buono'): accuratezza {pigro['accuratezza']:.4f}") print(" Il modello guadagna un punto percentuale di accuratezza e trova") print(" meno di un difetto su tre: l'accuratezza non è la metrica giusta.") return True def esegui_esercizio_2(): _sezione("Esercizio 2 - la soglia che costa meno") risultato = soglia_di_costo_minimo(DIFETTOSE, PUNTEGGI) if risultato is None: _da_fare(2) return False soglia, costo = risultato teorica = COSTO_FP / (COSTO_FP + COSTO_FN) print(f" soglia a costo minimo: {soglia:.2f} -> costo {costo:.1f}") print(f" soglia 0.50 (default): costo {costo_totale(DIFETTOSE, PUNTEGGI, 0.50):.1f}") print(f" soglia teorica {teorica:.4f}: costo " f"{costo_totale(DIFETTOSE, PUNTEGGI, teorica):.1f}") print(f" 'ferma tutto' (soglia 0.00): costo " f"{costo_totale(DIFETTOSE, PUNTEGGI, 0.0):.1f}") print(f" 'non fermare nulla' (soglia 1.01): costo " f"{costo_totale(DIFETTOSE, PUNTEGGI, 1.01):.1f}") return True def esegui_esercizio_3(rng): _sezione("Esercizio 3 - lo split che mente") X, y, pannello = genera_pannelli(90, rng) print(f" righe: {len(y)} pannelli: {len(np.unique(pannello))}" f" difettosi: {y.mean():.3f}") in_test = split_per_riga(pannello, rng) pred = knn1_predici(X[~in_test], y[~in_test], X[in_test]) acc_sbagliato = float((pred == y[in_test]).mean()) print(f" (a) split per riga -> accuratezza {acc_sbagliato:.3f}" f" [{int(in_test.sum())} righe di test]") in_test_g = split_per_pannello(pannello, 1 / 3, rng) if in_test_g is None: _da_fare(3) return False pred_g = knn1_predici(X[~in_test_g], y[~in_test_g], X[in_test_g]) acc_giusto = float((pred_g == y[in_test_g]).mean()) pannelli_condivisi = len( set(pannello[in_test_g]).intersection(set(pannello[~in_test_g]))) print(f" (b) split per pannello -> accuratezza {acc_giusto:.3f}" f" [{int(in_test_g.sum())} righe di test]") print(f" pannelli presenti da entrambe le parti: {pannelli_condivisi}" " (deve essere 0)") print(f" ottimismo del leakage: {acc_sbagliato - acc_giusto:+.3f} di accuratezza") return True def main(): print("Corso MLE - Modulo 03, Lezione 1 - esercizi di codice") fatti = 0 if esegui_esercizio_1(): fatti += 1 if esegui_esercizio_2(): fatti += 1 if esegui_esercizio_3(np.random.default_rng(SEED)): fatti += 1 print() print(f"Esercizi completati: {fatti} su 3.") if fatti < 3: print("Completa i TODO e rilancia: python3 esercizi.py") if __name__ == "__main__": main() # ====================================================================== # SOLUZIONI DI RIFERIMENTO # Decommenta una funzione alla volta, incollala al posto del TODO # corrispondente e rilancia. # ====================================================================== # # --- Esercizio 1 ------------------------------------------------------- # def metriche_da_matrice(vp, fp, fn, vn): # totale = vp + fp + fn + vn # accuratezza = (vp + vn) / totale # precision = vp / (vp + fp) if (vp + fp) else 0.0 # recall = vp / (vp + fn) if (vp + fn) else 0.0 # f1 = (2 * precision * recall / (precision + recall) # if (precision + recall) else 0.0) # return {"accuratezza": accuratezza, "precision": precision, # "recall": recall, "f1": f1} # # Atteso, verificabile a mano: # accuratezza = (18 + 1134) / 1200 = 1152 / 1200 = 0.9600 # precision = 18 / (18 + 6) = 18 / 24 = 0.7500 # recall = 18 / (18 + 42) = 18 / 60 = 0.3000 # F1 = 2 * 0.75 * 0.30 / 1.05 = 0.45 / 1.05 = 0.4286 # collaudatore pigro: (0 + 1140) / 1200 = 0.9500 # Il modello vale un punto percentuale di accuratezza in più del pigro e # lascia passare 42 difetti su 60. Con il 5% di positivi l'accuratezza è # quasi tutta merito dei negativi: non misura cio' che ci interessa. # # --- Esercizio 2 ------------------------------------------------------- # def soglia_di_costo_minimo(y, punteggi): # migliore_soglia, migliore_costo = None, float("inf") # for soglia in np.sort(np.unique(punteggi))[::-1]: # costo = costo_totale(y, punteggi, soglia) # if costo < migliore_costo: # migliore_costo, migliore_soglia = costo, float(soglia) # return migliore_soglia, migliore_costo # # Atteso, verificabile a mano (il costo per ciascuna delle dieci soglie): # 0.95 -> 120 0.80 -> 80 0.62 -> 82 0.55 -> 42 0.40 -> 44 # 0.33 -> 4 0.21 -> 6 0.12 -> 8 0.07 -> 10 0.02 -> 12 # Minimo: soglia 0.33, costo 4.0 (due falsi allarmi, nessun difetto sfuggito). # Confronti: soglia 0.50 costa 42.0; 'ferma tutto' costa 12.0; 'non fermare # nulla' costa 160.0. La soglia teorica per probabilità CALIBRATE è # COSTO_FP / (COSTO_FP + COSTO_FN) = 2 / 42 = 0.0476, e qui costa 10.0: # peggio del minimo empirico. Non è un errore della formula, è la prova che # i punteggi del modello non sono probabilità calibrate. La formula dice # dove sarebbe la soglia se lo fossero; la ricerca dice dov'è davvero. # # --- Esercizio 3 ------------------------------------------------------- # def split_per_pannello(pannello_di_riga, frazione_test, rng): # pannelli = np.unique(pannello_di_riga) # mescolati = rng.permutation(pannelli) # quanti = int(round(frazione_test * len(pannelli))) # return np.isin(pannello_di_riga, mescolati[:quanti]) # # Output misurato con SEED = 7 (Python 3.14.4, NumPy 2.3.5): # # righe: 270 pannelli: 90 difettosi: 0.389 # (a) split per riga -> accuratezza 1.000 [90 righe di test] # (b) split per pannello -> accuratezza 0.700 [90 righe di test] # pannelli presenti da entrambe le parti: 0 (deve essere 0) # ottimismo del leakage: +0.300 di accuratezza # # Il primo numero è 1.000 ESATTAMENTE, e non dipende dal seed: lo split (a) # mette nel test una sola riga per pannello, quindi le altre due copie # identiche restano nel training. Il vicino più prossimo di ogni riga di test # sta a distanza 0 e porta la stessa etichetta, percio' il 1-NN non può # sbagliare. Un'accuratezza perfetta che non dice nulla sul modello: dice # soltanto che i duplicati sono stati divisi male. # # Il secondo numero, 0.700, è la prestazione vera: il 1-NN su pannelli mai # visti indovina sette volte su dieci. Dipende dal seed, perché dipende da # quali 30 pannelli finiscono nel test; con SEED = 7 esce tondo. # L'ottimismo del leakage è quindi +0.300 di accuratezza: trenta punti # percentuali di bugia, su un dataset in cui nessuno ha copiato nulla e # nessuna riga è sbagliata. L'unico errore è stato dividere per riga invece # che per pannello. # # Il controllo "pannelli presenti da entrambe le parti" deve stampare 0: se # stampa un numero diverso lo split è ancora sbagliato, e il +0.300 che # misuri non è l'ottimismo del leakage - è il tuo bug.