""" Corso Machine Learning Engineer - Modulo 03, Lezione 7 "Boosting: AdaBoost, Gradient Boosting e cenni su XGBoost" OBIETTIVO DI QUESTO FILE ------------------------ Tre esercizi su un dominio nuovo rispetto all'articolo: il controllo delle letture dei contatori elettrici di un condominio. Dieci letture, qualcuna anomala, e un paio di letture TRASCRITTE MALE dall'operatore - cioè etichette sbagliate, che sono il veleno specifico del boosting. 1. un giro di AdaBoost fatto a mano, con un risultato notevole da verificare: dopo l'aggiornamento dei pesi, l'errore pesato dello STESSO classificatore debole vale esattamente 0.5. Il boosting rende inutile il modello che ha appena scelto, e così è costretto a cambiare; 2. il Gradient Boosting con perdita quadratica e il modello debole più semplice che esista (una costante): le previsioni seguono la formula chiusa F_m = media * (1 - (1-nu)^m). Da qui si vede perché learning rate e numero di alberi sono due manopole della stessa cosa; 3. perché AdaBoost insegue le etichette sbagliate: il peso di un esempio sempre classificato male viene moltiplicato per 1/(2*err) a ogni giro, e più il modello debole è bravo, più quella crescita è violenta. COME SI USA ----------- python3 esercizi.py Il file gira anche così com'è: stampa quali esercizi mancano ancora. AMBIENTE DICHIARATO ------------------- Python 3.12+ e NumPy 2.3. Nessun altro pacchetto. python3 -m venv .venv && source .venv/bin/activate pip install "numpy>=2.0" Tutti e tre gli esercizi sono DETERMINISTICI e i numeri scelti danno frazioni esatte: 1/6, 1/14, 5/3. Carta e penna bastano. """ import numpy as np # ---------------------------------------------------------------------- # I dati: dieci letture di contatore, pesi iniziali uniformi. # # Il primo classificatore debole (una soglia sul consumo notturno) sbaglia su # tre letture: le posizioni 2, 5 e 9. L'errore pesato è quindi 0.3. # ---------------------------------------------------------------------- N_LETTURE = 10 PESI_INIZIALI = np.full(N_LETTURE, 1.0 / N_LETTURE) SBAGLIATE = np.zeros(N_LETTURE, dtype=bool) SBAGLIATE[[2, 5, 9]] = True # ---------------------------------------------------------------------- # ESERCIZIO 1 - Un giro di AdaBoost, a mano # # Dati i pesi w e la maschera degli esempi sbagliati: # # err = somma dei pesi degli esempi sbagliati (i pesi sommano a 1) # alpha = 0.5 * ln( (1 - err) / err ) # w_i <- w_i * exp(+alpha) se l'esempio i è sbagliato # w_i <- w_i * exp(-alpha) se è corretto # e poi si normalizza, dividendo per la somma Z. # # Scrivi la funzione. Poi guarda l'ultima riga stampata dal runner. # # Domanda: perché l'errore pesato dello stesso classificatore, calcolato con # i pesi NUOVI, vale esattamente 0.5? E che cosa implica per il giro dopo? # ---------------------------------------------------------------------- def passo_adaboost(pesi, sbagliate): """Un singolo giro di AdaBoost: errore, peso del modello, pesi aggiornati. Parametri --------- pesi : np.ndarray (n,) - pesi correnti, sommano a 1. sbagliate : np.ndarray (n,) booleano - True dove il modello debole sbaglia. Ritorna ------- tupla (err, alpha, nuovi_pesi) con nuovi_pesi normalizzati a somma 1, oppure None se l'esercizio non è ancora svolto. Casi limite da conoscere, perché in produzione arrivano: - err = 0: alpha è +infinito. Le librerie si fermano e tengono quel solo modello, che da solo spiega tutto (sospetta data leakage); - err >= 0.5: alpha è <= 0. AdaBoost nella formulazione originale si ferma: un modello debole peggiore del caso non serve. Se capita al primo giro, il problema non è il boosting, sono le feature. """ # TODO (esercizio 1): sostituisci `return None` con il giro completo. return None def errore_pesato(pesi, sbagliate): """Errore pesato di un classificatore. Già pronta.""" return float(pesi[sbagliate].sum() / pesi.sum()) # ---------------------------------------------------------------------- # ESERCIZIO 2 - Gradient Boosting con perdita quadratica # # Prendi il modello debole più povero immaginabile: una COSTANTE. Con perdita # quadratica, la costante migliore per un vettore di residui è la loro media. # # Partendo da F_0 = 0, il giro m fa: # residui r = y - F_m # costante c = media(r) # F_{m+1} = F_m + nu * c # # Scrivi il ciclo e confronta con la formula chiusa # # F_m = media(y) * ( 1 - (1 - nu)^m ) # # Domanda: dimezzando nu, quanti giri servono per arrivare allo stesso punto? # ---------------------------------------------------------------------- LETTURE_KWH = np.array([8.0, 12.0, 9.0, 11.0, 14.0, 6.0, 10.0, 13.0, 7.0, 10.0]) def boosting_costante(y, nu, M): """Gradient boosting con perdita quadratica e modelli deboli costanti. Ritorna ------- np.ndarray (M+1,) con la previsione (unica, perché il modello è una costante) dopo 0, 1, ..., M giri; oppure None se non svolto. Suggerimento: tieni una variabile scalare `F`, parti da 0.0 e a ogni giro aggiungi `nu * (y - F).mean()`. Non serve NumPy per altro che la media. """ # TODO (esercizio 2): sostituisci `return None` con il ciclo. return None def formula_chiusa(y, nu, m): """F_m secondo la formula chiusa. Già pronta: serve come controllo.""" return float(y.mean() * (1.0 - (1.0 - nu) ** m)) # ---------------------------------------------------------------------- # ESERCIZIO 3 - Perché AdaBoost insegue le letture trascritte male # # Nel giro di AdaBoost, dopo la normalizzazione, il peso di un esempio viene # moltiplicato per un fattore che dipende SOLO dall'errore del modello debole: # # esempio sbagliato : exp(+alpha) / Z = 1 / (2 * err) # esempio corretto : exp(-alpha) / Z = 1 / (2 * (1 - err)) # # dove Z = 2 * sqrt( err * (1 - err) ) è il fattore di normalizzazione. # (Verificarlo è una riga di algebra: exp(alpha) = sqrt((1-err)/err).) # # Scrivi i due moltiplicatori e guarda la tabella. # # Domanda: con err = 0.10, di quanto cresce in dieci giri il peso di una # lettura trascritta male, che per definizione nessun modello potrà mai # indovinare? # ---------------------------------------------------------------------- def moltiplicatori_di_peso(err): """Fattori di crescita del peso per esempio sbagliato e corretto. Ritorna ------- tupla (fattore_sbagliato, fattore_corretto), oppure None se l'esercizio non è ancora svolto. Controllo di sanità: con err = 0.5 entrambi i fattori devono valere 1.0, cioè nessun aggiornamento. Ha senso: un modello che indovina a caso non insegna niente, e AdaBoost lo scarta. """ # TODO (esercizio 3): sostituisci `return None` con i due fattori. return None # ---------------------------------------------------------------------- # 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 - un giro di AdaBoost") risultato = passo_adaboost(PESI_INIZIALI, SBAGLIATE) if risultato is None: _da_fare(1) return False err, alpha, nuovi = risultato print(f" err = {err:.6f} alpha = {alpha:.6f}" f" exp(alpha) = {np.exp(alpha):.6f}") print(f" pesi nuovi, esempi sbagliati: {nuovi[SBAGLIATE].round(6)}" f" (frazione esatta 1/6)") print(f" pesi nuovi, esempi corretti : {nuovi[~SBAGLIATE][:3].round(6)} ..." f" (frazione esatta 1/14)") print(f" somma dei pesi nuovi = {nuovi.sum():.6f}") print(f" errore pesato dello STESSO modello con i pesi nuovi =" f" {errore_pesato(nuovi, SBAGLIATE):.6f}") print(" Esattamente 0.5: il modello appena scelto è diventato inutile.") print(" AdaBoost non 'cerca' modelli diversi, li RENDE necessari.") return True def esegui_esercizio_2(): _sezione("Esercizio 2 - Gradient Boosting a costanti") media = float(LETTURE_KWH.mean()) print(f" media delle letture = {media:.4f} kWh" " (il bersaglio, per un modello costante)") for nu in (1.0, 0.5, 0.1): previsioni = boosting_costante(LETTURE_KWH, nu, 50) if previsioni is None: _da_fare(2) return False scelti = [1, 5, 20, 50] celle = " ".join( f"m={m:<3}{previsioni[m]:7.4f}" for m in scelti) print(f" nu = {nu:<4} {celle}") massimo_scarto = max(abs(previsioni[m] - formula_chiusa(LETTURE_KWH, nu, m)) for m in range(51)) print(f" scarto massimo dalla formula chiusa:" f" {massimo_scarto:.2e}") print(" Con nu = 1.0 il modello arriva alla media in un solo giro: zero") print(" regolarizzazione. Con nu = 0.1 ci arriva in circa 50 giri.") print(" Il prodotto nu * M è approssimativamente invariante: dimezzare il") print(" learning rate richiede raddoppiare gli alberi. Lo si vede dalla") print(" formula: (1-nu)^M resta costante se M * nu resta costante.") return True def esegui_esercizio_3(): _sezione("Esercizio 3 - la lettura trascritta male") print(" err | x peso se sbagliato | x peso se corretto |" " dopo 10 giri (sbagliato)") for err in (0.45, 0.30, 0.20, 0.10, 0.05): coppia = moltiplicatori_di_peso(err) if coppia is None: _da_fare(3) return False su, giu = coppia print(f" {err:5.2f} | {su:20.6f} | {giu:18.6f} | {su ** 10:22.1f}") print(" Con err = 0.5 i fattori sono 1.0 e 1.0: nessun aggiornamento.") print(" Più il modello debole è bravo, più brutalmente AdaBoost") print(" riscrive i pesi di cio' che ha sbagliato. Se quel qualcosa è una") print(" lettura trascritta male, nessun modello potrà indovinarla, e il") print(" boosting spendera' i giri successivi a inseguirla.") print(" Difese, in ordine di efficacia: (1) pulire le etichette;") print(" (2) learning rate basso con early stopping in validazione;") print(" (3) una perdita robusta (Huber per la regressione) o") print(" Gradient Boosting invece di AdaBoost, che pesa i residui in") print(" modo meno esplosivo.") return True def main(): print("Corso MLE - Modulo 03, Lezione 7 - esercizi di codice") fatti = sum(bool(f()) for f in (esegui_esercizio_1, esegui_esercizio_2, esegui_esercizio_3)) 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 # ====================================================================== # # --- Esercizio 1 ------------------------------------------------------- # def passo_adaboost(pesi, sbagliate): # err = float(pesi[sbagliate].sum() / pesi.sum()) # alpha = 0.5 * np.log((1.0 - err) / err) # fattori = np.where(sbagliate, np.exp(alpha), np.exp(-alpha)) # nuovi = pesi * fattori # return err, float(alpha), nuovi / nuovi.sum() # # Atteso, tutto in frazioni esatte: # err = 3 * 0.1 = 0.300000 # alpha = 0.5 * ln(7/3) = 0.5 * 0.8472979 = 0.4236489 # exp(alpha) = sqrt(7/3) = 1.5275252 ; exp(-alpha) = sqrt(3/7) = 0.6546537 # pesi non normalizzati: 0.1527525 (x3) e 0.0654654 (x7) # Z = 3*0.1527525 + 7*0.0654654 = 0.4582576 + 0.4582576 = 0.9165151 # (e Z = 2*sqrt(err*(1-err)) = 2*sqrt(0.21) = 0.9165151: coincide) # pesi normalizzati: 0.1666667 = 1/6 per i tre sbagliati # 0.0714286 = 1/14 per i sette corretti # verifica: 3 * 1/6 = 0.5 e 7 * 1/14 = 0.5 # errore pesato del modello con i pesi NUOVI = 0.500000 # # Perché esattamente 0.5, sempre, qualunque fosse err? Perché Z si spezza in # due metà uguali per costruzione: la massa degli sbagliati, err*sqrt((1-e)/e) # = sqrt(e(1-e)), e la massa dei corretti, (1-err)*sqrt(e/(1-e)) = sqrt(e(1-e)). # Sono la stessa quantità, quindi ciascuna vale Z/2. Non è un caso dei # numeri scelti: è un'identità. # Implicazione: il modello debole appena aggiunto, sui pesi del giro # successivo, è equivalente al lancio di una monetina. Il giro dopo è # costretto a guardare altrove. È tutto il meccanismo di AdaBoost, in una riga. # # --- Esercizio 2 ------------------------------------------------------- # def boosting_costante(y, nu, M): # F = 0.0 # storia = [F] # for _ in range(M): # F = F + nu * float((y - F).mean()) # storia.append(F) # return np.array(storia) # # Atteso: media delle letture = (8+12+9+11+14+6+10+13+7+10)/10 = 100/10 = 10.0 # nu = 1.0 : m=1 -> 10.0000 ; m=5, 20, 50 -> 10.0000 (ci arriva subito) # nu = 0.5 : m=1 -> 5.0000 ; m=5 -> 10*(1-0.5^5) = 9.6875 # m=20 -> 10.0000 (a meno di 1e-5) ; m=50 -> 10.0000 # nu = 0.1 : m=1 -> 1.0000 ; m=5 -> 10*(1-0.9^5) = 4.0951 # m=20 -> 10*(1-0.9^20) = 8.7842 ; m=50 -> 10*(1-0.9^50) = 9.9485 # scarto massimo dalla formula chiusa: 0 oppure dell'ordine di 1e-15 (solo # arrotondamento in virgola mobile). # # Dimostrazione della formula chiusa, per induzione. F_0 = 0 e # media(y - F_m) = media(y) - F_m, perché F_m è una costante. Quindi # F_{m+1} = F_m + nu*(media(y) - F_m) = (1-nu)*F_m + nu*media(y). # Posto g_m = media(y) - F_m si ottiene g_{m+1} = (1-nu) g_m, cioè # g_m = (1-nu)^m * media(y), da cui F_m = media(y)*(1 - (1-nu)^m). # # Due letture del risultato, entrambe utili in produzione: # (a) il learning rate è una frenata geometrica. Con nu = 0.1 dopo m giri # resta il 0.9^m dell'errore iniziale: serve m ~ 22 per abbatterlo al # 10%, m ~ 44 per l'1%. Da qui la regola "nu piccolo, molti alberi"; # (b) nu e M non sono indipendenti: cercarli entrambi in una griglia # quadrata è spreco. Si fissa nu basso (0.05-0.1) e si cerca M con # l'early stopping sulla validazione. # Attenzione a non generalizzare troppo: qui il modello debole è una costante, # quindi il boosting converge alla media e si ferma. Con alberi veri il # bersaglio non è una costante e il boosting, oltre un certo M, inizia a # memorizzare il rumore: per quello serve l'early stopping, non la formula. # # --- Esercizio 3 ------------------------------------------------------- # def moltiplicatori_di_peso(err): # return 1.0 / (2.0 * err), 1.0 / (2.0 * (1.0 - err)) # # Atteso: # err | sbagliato | corretto | sbagliato^10 # 0.45 | 1.111111 | 0.909091 | 2.9 # 0.30 | 1.666667 | 0.714286 | 165.4 # 0.20 | 2.500000 | 0.625000 | 9536.7 # 0.10 | 5.000000 | 0.555556 | 9765625.0 # 0.05 | 10.000000 | 0.526316 | 10000000000.0 # # I numeri delle ultime righe sono idealizzati: nella realtà i pesi sommano a # 1, quindi appena una lettura trascritta male assorbe massa l'errore del # modello debole SALE, i fattori si avvicinano a 1 e la crescita si smorza da # sola. Ma la direzione resta, e la lezione è quella: AdaBoost con modelli # deboli molto bravi (err piccolo) è estremamente aggressivo verso cio' che # non riesce a classificare, e un'etichetta sbagliata non si può classificare # per definizione. Il risultato è un ensemble che spende la sua capacità sui # dati sporchi invece che sul segnale. # La derivazione dei fattori, in una riga: exp(alpha) = sqrt((1-e)/e), e # Z = 2*sqrt(e(1-e)), quindi exp(alpha)/Z = sqrt((1-e)/e)/(2 sqrt(e(1-e))) # = 1/(2e). Analogamente exp(-alpha)/Z = 1/(2(1-e)). # Controllo: con e = 0.5 entrambi valgono 1, e infatti alpha = 0.