A/B Testing di Modelli ML: Metodologia, Metriche e Implementazione
Hai addestrato due versioni del tuo modello di raccomandazione. Il nuovo modello, basato su transformer, mostra un AUC del 3% superiore sull'holdout set. Sembra un miglioramento netto, ma questa differenza si traduce davvero in un impatto positivo per gli utenti reali? Il modello potrebbe performare meglio su certe coorti demografiche e peggio su altre. Potrebbe ridurre il click-through rate pur aumentando la soddisfazione a lungo termine. Potrebbe avere una latenza superiore che annulla i benefici di accuratezza. Le metriche offline non mentono, ma raccontano solo una parte della storia.
L'A/B testing per modelli ML e la metodologia che permette di rispondere a queste domande in modo rigoroso, confrontando le versioni di un modello su traffico reale con utenti reali, misurando le metriche di business che contano davvero. Secondo una ricerca di 2025 di Aimpoint Digital Labs, le organizzazioni che adottano strategie strutturate di A/B testing per i modelli ML riducono del 40% il rischio di regressione in produzione rispetto a deployment diretti basati solo su metriche offline. Il mercato MLOps, che varra $4.38 miliardi nel 2026, e destinato a raggiungere $89.18 miliardi entro il 2035, e l'A/B testing e uno dei mattoni fondamentali di questa crescita.
In questa guida costruiremo un sistema completo di A/B testing per modelli ML: dalla teoria statistica al router FastAPI, dal canary deployment al shadow mode, dai test frequentisti all'approccio bayesiano con Thompson Sampling, fino al monitoring delle metriche durante i test con Prometheus e Grafana.
Cosa Imparerai
- Differenza tra A/B testing ML e A/B testing web classico
- Design di esperimenti: sample size, potenza statistica, metriche di successo
- Traffic splitting con router FastAPI e canary deployment progressivo
- Shadow mode: testare senza impatto sugli utenti
- Multi-Armed Bandits e Thompson Sampling come alternativa all'A/B testing classico
- Analisi statistica: p-value, confidence intervals, effect size
- A/B testing bayesiano per decisioni più rapide
- Monitoring durante i test con Prometheus e Grafana
- Best practices e anti-pattern da evitare
A/B Testing ML vs A/B Testing Web: Differenze Critiche
Il test A/B e nato nel web analytics per confrontare varianti di landing page, bottoni e copy. Il framework statistico di base e lo stesso, ma l'A/B testing per modelli ML ha complessità aggiuntive che lo rendono sostanzialmente diverso nella pratica.
Nel web testing si confrontano esperienze visive discrete: la variante A e la variante B sono chiaramente separate. Nei modelli ML, le predizioni sono continue, distribuite e spesso correlate nel tempo. Un modello di raccomandazione che serve lo stesso utente in sessioni diverse non produce predizioni indipendenti: c'è correlazione temporale che viola le assunzioni di indipendenza dei test statistici classici.
Differenze Chiave ML vs Web A/B Testing
- Metriche: nel web si ottimizza CTR o conversion rate; in ML si ottimizzano simultaneamente metriche offline (AUC, RMSE) e metriche di business (revenue, churn rate, NPS), spesso in conflitto.
- Latenza del feedback: nel web il risultato e immediato (click); in ML può richiedere giorni o settimane (churn dopo 30 giorni, revenue dopo un trimestre).
- Distribuzione dell'effetto: un modello può performare meglio in media ma peggio su coorti specifiche (ageism, geographic bias), richiedendo analisi segmentata.
- Effetti di sistema: in sistemi a feedback loop (raccomandazioni, pricing dinamico), il modello B influenza i dati che poi alleneranno il modello C.
- Rischi operativi: un bug in una variante web causa una brutta UX; un bug in un modello ML di fraud detection può causare perdite finanziarie significative.
Design dell'Esperimento: Prima del Codice
Un A/B test mal progettato e peggio di nessun A/B test: fornisce una falsa sensazione di rigore scientifico mentre produce conclusioni sbagliate. Il design dell'esperimento deve precedere qualsiasi implementazione tecnica.
Definire le Metriche di Successo
Ogni esperimento deve avere una Primary Metric unica che determina il vincitore, più zero a due Guardrail Metrics che il modello B non deve peggiorare rispetto ad A. La primary metric deve essere direttamente causalmente collegata all'obiettivo di business.
Esempi di metriche per diversi scenari:
- Modello churn: primary = retention rate a 30 giorni; guardrail = latenza P95, costo campagna
- Modello raccomandazione: primary = revenue per sessione; guardrail = CTR, diversità raccomandazioni
- Modello fraud: primary = tasso di frodi non rilevate; guardrail = false positive rate, latenza
- Modello pricing: primary = margine lordo; guardrail = tasso di conversione, NPS
Calcolo del Sample Size
Il sample size necessario dipende da tre fattori: l'effect size minimo che si vuole rilevare (minimum detectable effect, MDE), il livello di significativita alpha (solitamente 0.05) e la potenza statistica 1-beta (solitamente 0.80).
# sample_size_calculator.py
# Calcolo del sample size per A/B test ML
import numpy as np
from scipy import stats
from scipy.stats import norm
import math
def calculate_sample_size(
baseline_rate: float,
minimum_detectable_effect: float,
alpha: float = 0.05,
power: float = 0.80,
two_tailed: bool = True
) -> int:
"""
Calcola il sample size per un A/B test su proporzioni (es. conversion rate).
Args:
baseline_rate: Tasso attuale del modello A (es. 0.15 per 15% churn)
minimum_detectable_effect: Variazione minima relativa da rilevare (es. 0.05 per +5%)
alpha: Livello di significativita (type I error rate)
power: Potenza statistica (1 - type II error rate)
two_tailed: True per test bidirezionale (default raccomandato)
Returns:
Sample size per ciascuna delle due varianti
"""
p1 = baseline_rate
p2 = baseline_rate * (1 + minimum_detectable_effect)
# Calcolo basato su formula di Cohen
z_alpha = norm.ppf(1 - alpha / (2 if two_tailed else 1))
z_beta = norm.ppf(power)
p_avg = (p1 + p2) / 2
q_avg = 1 - p_avg
numerator = (z_alpha * math.sqrt(2 * p_avg * q_avg) + z_beta * math.sqrt(p1 * (1-p1) + p2 * (1-p2))) ** 2
denominator = (p2 - p1) ** 2
n = math.ceil(numerator / denominator)
return n
def calculate_duration_days(
sample_size_per_variant: int,
daily_requests: int,
traffic_split: float = 0.5
) -> float:
"""Stima la durata del test in giorni."""
requests_per_variant_per_day = daily_requests * traffic_split
return sample_size_per_variant / requests_per_variant_per_day
# --- Esempio pratico: modello churn ---
baseline_churn_rate = 0.18 # 18% churn attuale (modello A)
mde = 0.10 # vogliamo rilevare un miglioramento del 10% relativo
# (da 18% a 16.2%)
n_per_variant = calculate_sample_size(
baseline_rate=baseline_churn_rate,
minimum_detectable_effect=-mde, # negativo = riduzione del churn
alpha=0.05,
power=0.80
)
daily_traffic = 5000 # richieste al giorno
test_duration = calculate_duration_days(n_per_variant, daily_traffic, 0.5)
print(f"Sample size per variante: {n_per_variant:,} campioni")
print(f"Durata stimata del test: {test_duration:.1f} giorni")
print(f"Traffico totale necessario: {n_per_variant * 2:,} richieste")
# Output tipico:
# Sample size per variante: 8,744 campioni
# Durata stimata del test: 3.5 giorni
# Traffico totale necessario: 17,488 richiesteTrappola del Peeking: Non Guardare i Risultati Troppo Presto
Il problema del "peeking" (o optional stopping) e uno degli errori più comuni nell'A/B testing: si guarda ai risultati intermedi e si ferma il test non appena si raggiunge la significativita statistica. Questo aumenta drasticamente il tasso di falsi positivi: se si guarda ai dati ogni giorno, la probabilità di trovare un risultato significativo per caso sale fino al 30% anche se le due varianti sono identiche. Usa sempre un sample size prefissato e controlla i risultati solo alla fine del test, oppure adotta metodi di testing sequenziale come i Sequential Probability Ratio Tests (SPRT).
Traffic Splitting con FastAPI
Il router di A/B testing e il componente centrale dell'infrastruttura. Deve distribuire il traffico in modo deterministico (lo stesso utente deve sempre andare sulla stessa variante per tutta la durata del test), registrare a quale variante e stato assegnato ogni utente e ogni predizione, e essere estremamente veloce per non aggiungere latenza al percorso critico.
# ab_router.py
# Router A/B testing per modelli ML con FastAPI
from fastapi import FastAPI, Request, Header
from pydantic import BaseModel
import hashlib
import json
import time
import logging
from typing import Optional, Literal
from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST
from starlette.responses import Response
logger = logging.getLogger(__name__)
app = FastAPI(title="ML A/B Testing Router")
# --- Prometheus Metrics ---
AB_REQUESTS = Counter(
"ab_test_requests_total",
"Numero totale di richieste per variante",
labelnames=["experiment_id", "variant", "model_version"]
)
AB_LATENCY = Histogram(
"ab_test_latency_seconds",
"Latenza inference per variante",
labelnames=["experiment_id", "variant"],
buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0]
)
AB_PREDICTIONS = Counter(
"ab_test_predictions_total",
"Distribuzione delle predizioni per variante",
labelnames=["experiment_id", "variant", "prediction_bucket"]
)
# --- Configurazione esperimento ---
ACTIVE_EXPERIMENT = {
"experiment_id": "churn_model_v2_vs_v3",
"model_a": {
"name": "churn-model-v2",
"endpoint": "http://model-a-service:8080/predict",
"traffic_weight": 0.5
},
"model_b": {
"name": "churn-model-v3",
"endpoint": "http://model-b-service:8080/predict",
"traffic_weight": 0.5
},
"start_time": "2025-03-01T00:00:00Z",
"end_time": "2025-03-15T00:00:00Z"
}
class PredictionRequest(BaseModel):
user_id: str
features: dict
def assign_variant(user_id: str, experiment_id: str, traffic_split: float = 0.5) -> str:
"""
Assegna deterministicamente un utente a una variante.
Lo stesso user_id + experiment_id producono sempre lo stesso risultato.
"""
hash_input = f"{user_id}:{experiment_id}"
hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)
normalized = (hash_value % 10000) / 10000.0
if normalized < traffic_split:
return "A"
else:
return "B"
async def call_model(endpoint: str, features: dict) -> dict:
"""Chiama il servizio del modello."""
import httpx
async with httpx.AsyncClient(timeout=2.0) as client:
response = await client.post(endpoint, json=features)
response.raise_for_status()
return response.json()
@app.post("/predict")
async def predict(request: PredictionRequest):
"""
Endpoint principale: smista le richieste alle varianti A/B.
"""
exp = ACTIVE_EXPERIMENT
exp_id = exp["experiment_id"]
# Assegna variante in modo deterministico
variant = assign_variant(
user_id=request.user_id,
experiment_id=exp_id,
traffic_split=exp["model_a"]["traffic_weight"]
)
# Seleziona il modello corretto
model_config = exp["model_a"] if variant == "A" else exp["model_b"]
# Registra richiesta
AB_REQUESTS.labels(
experiment_id=exp_id,
variant=variant,
model_version=model_config["name"]
).inc()
# Chiama il modello con misura della latenza
start_time = time.time()
try:
result = await call_model(model_config["endpoint"], request.features)
except Exception as e:
logger.error(f"Errore chiamata modello {variant}: {e}")
raise
latency = time.time() - start_time
AB_LATENCY.labels(experiment_id=exp_id, variant=variant).observe(latency)
# Bucketing della predizione per distribuzione
score = result.get("churn_probability", 0)
bucket = "high" if score > 0.7 else ("medium" if score > 0.3 else "low")
AB_PREDICTIONS.labels(
experiment_id=exp_id, variant=variant, prediction_bucket=bucket
).inc()
return {
"prediction": result,
"variant": variant,
"model_version": model_config["name"],
"experiment_id": exp_id,
"latency_ms": round(latency * 1000, 2)
}
@app.get("/metrics")
async def metrics():
return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)
@app.get("/experiment/status")
async def experiment_status():
"""Ritorna lo stato corrente dell'esperimento."""
return {
"experiment": ACTIVE_EXPERIMENT["experiment_id"],
"active": True
}Canary Deployment: Rilascio Progressivo
Il canary deployment e una strategia di rilascio progressivo in cui il nuovo modello (il "canarino") riceve inizialmente solo una piccola percentuale del traffico produzione, tipicamente 1-5%. Se le metriche rimangono stabili, la percentuale viene aumentata gradualmente: 5% → 10% → 25% → 50% → 100%. In caso di anomalie, il rollback e immediato riportando tutto il traffico al modello stabile.
A differenza del classico A/B test 50/50, il canary e orientato alla riduzione del rischio più che alla rilevazione statistica di differenze. L'obiettivo non e dimostrare che il nuovo modello e migliore con significativita statistica, ma verificare che non causi problemi tecnici o regressioni evidenti prima di scalare il deployment.
# canary_deployment.py
# Implementazione canary deployment con rollback automatico
import asyncio
import time
import logging
from dataclasses import dataclass, field
from typing import Optional
from prometheus_client import Gauge
logger = logging.getLogger(__name__)
# Gauge per monitorare la percentuale di traffico canary
CANARY_TRAFFIC_WEIGHT = Gauge(
"canary_traffic_weight_percent",
"Percentuale di traffico al modello canary",
labelnames=["experiment_id"]
)
ERROR_RATE_GAUGE = Gauge(
"canary_error_rate",
"Tasso di errore del modello canary",
labelnames=["experiment_id"]
)
@dataclass
class CanaryConfig:
experiment_id: str
stable_model_endpoint: str
canary_model_endpoint: str
initial_canary_weight: float = 0.05 # Inizia al 5%
max_canary_weight: float = 1.0 # Target finale: 100%
step_size: float = 0.10 # Incremento per ogni step
step_interval_minutes: int = 30 # Ogni 30 minuti aumenta
max_error_rate: float = 0.02 # Rollback se errori > 2%
max_latency_p99_ms: float = 500.0 # Rollback se P99 > 500ms
current_weight: float = field(init=False)
def __post_init__(self):
self.current_weight = self.initial_canary_weight
class CanaryController:
"""
Controlla progressivamente il traffico al modello canary.
Esegue rollback automatico se le metriche superano le soglie.
"""
def __init__(self, config: CanaryConfig):
self.config = config
self.error_count = 0
self.total_count = 0
self.latencies = []
self.is_rolled_back = False
self.is_promoted = False
def should_route_to_canary(self, user_id: str) -> bool:
"""Determina se questa richiesta va al canary."""
if self.is_rolled_back:
return False
hash_val = int(hashlib.md5(
f"{user_id}:{self.config.experiment_id}".encode()
).hexdigest(), 16)
normalized = (hash_val % 10000) / 10000.0
return normalized < self.config.current_weight
def record_outcome(self, is_canary: bool, success: bool, latency_ms: float):
"""Registra l'esito di una chiamata al canary."""
if not is_canary:
return
self.total_count += 1
if not success:
self.error_count += 1
self.latencies.append(latency_ms)
# Aggiorna metriche Prometheus
error_rate = self.error_count / max(self.total_count, 1)
ERROR_RATE_GAUGE.labels(
experiment_id=self.config.experiment_id
).set(error_rate)
# Controlla soglie per rollback automatico
if error_rate > self.config.max_error_rate and self.total_count > 100:
logger.critical(
f"Error rate {error_rate:.2%} exceeded threshold "
f"{self.config.max_error_rate:.2%}. Initiating rollback."
)
self.rollback()
if len(self.latencies) >= 100:
p99 = sorted(self.latencies)[-1] # semplificato
if p99 > self.config.max_latency_p99_ms:
logger.critical(f"P99 latency {p99:.0f}ms exceeded threshold. Rollback.")
self.rollback()
def advance_canary(self):
"""Incrementa il peso del canary se le metriche sono OK."""
if self.is_rolled_back or self.is_promoted:
return
new_weight = min(
self.config.current_weight + self.config.step_size,
self.config.max_canary_weight
)
self.config.current_weight = new_weight
CANARY_TRAFFIC_WEIGHT.labels(
experiment_id=self.config.experiment_id
).set(new_weight * 100)
logger.info(
f"Canary weight increased to {new_weight:.0%} "
f"for experiment {self.config.experiment_id}"
)
if new_weight >= self.config.max_canary_weight:
self.is_promoted = True
logger.info("Canary fully promoted to production!")
def rollback(self):
"""Esegue rollback immediato al modello stabile."""
self.config.current_weight = 0.0
self.is_rolled_back = True
CANARY_TRAFFIC_WEIGHT.labels(
experiment_id=self.config.experiment_id
).set(0)
logger.warning(f"ROLLBACK executed for {self.config.experiment_id}")Shadow Mode: Testing Senza Impatto sugli Utenti
Lo shadow mode (o shadow deployment) e la tecnica più conservativa e al tempo stesso la più potente per validare un nuovo modello prima di esporlo agli utenti. Il traffico di produzione viene duplicato: il modello A serve le richieste reali e le sue predizioni vengono restituite agli utenti, mentre il modello B riceve le stesse richieste in parallelo ma le sue predizioni vengono scartate o solo loggate.
Questo approccio permette di confrontare i due modelli su traffico reale senza alcun rischio per gli utenti o per il business. E ideale per validare che il nuovo modello: non abbia bug critici, rispetti i requisiti di latenza sotto carico reale, non produca predizioni anomale o out-of-distribution, e si comporti come atteso su tutti i segmenti di utenti.
# shadow_mode.py
# Implementazione shadow deployment con logging asincrono
import asyncio
import httpx
import logging
import json
from datetime import datetime
from typing import Any
logger = logging.getLogger(__name__)
class ShadowModeRouter:
"""
Router che invia le richieste sia al modello produzione che al modello shadow.
Il modello produzione risponde agli utenti; il shadow solo logga.
"""
def __init__(
self,
production_endpoint: str,
shadow_endpoint: str,
shadow_log_file: str = "shadow_predictions.jsonl"
):
self.production_endpoint = production_endpoint
self.shadow_endpoint = shadow_endpoint
self.shadow_log_file = shadow_log_file
async def predict(self, request_data: dict, request_id: str) -> dict:
"""
Invia la richiesta al modello produzione e in parallelo al shadow.
Restituisce solo la risposta del modello produzione.
"""
# Esegui produzione e shadow in parallelo
prod_task = asyncio.create_task(
self._call_model(self.production_endpoint, request_data, "production")
)
shadow_task = asyncio.create_task(
self._call_model(self.shadow_endpoint, request_data, "shadow")
)
# Aspetta la risposta produzione (non bloccante per shadow)
prod_result = await prod_task
# Logga la risposta shadow in background senza bloccare
asyncio.create_task(
self._log_shadow_result(shadow_task, request_id, request_data, prod_result)
)
return prod_result
async def _call_model(
self, endpoint: str, data: dict, label: str
) -> dict:
"""Chiama un endpoint modello con gestione degli errori."""
start = asyncio.get_event_loop().time()
try:
async with httpx.AsyncClient(timeout=2.0) as client:
response = await client.post(endpoint, json=data)
response.raise_for_status()
result = response.json()
result["_latency_ms"] = (asyncio.get_event_loop().time() - start) * 1000
result["_model"] = label
return result
except Exception as e:
logger.error(f"Error calling {label} model: {e}")
return {"error": str(e), "_model": label, "_latency_ms": -1}
async def _log_shadow_result(
self,
shadow_task: asyncio.Task,
request_id: str,
input_data: dict,
production_result: dict
):
"""Logga la risposta shadow per analisi offline."""
try:
shadow_result = await shadow_task
except Exception as e:
shadow_result = {"error": str(e)}
log_entry = {
"timestamp": datetime.utcnow().isoformat(),
"request_id": request_id,
"input_features": input_data,
"production_prediction": production_result.get("prediction"),
"production_latency_ms": production_result.get("_latency_ms"),
"shadow_prediction": shadow_result.get("prediction"),
"shadow_latency_ms": shadow_result.get("_latency_ms"),
"shadow_error": shadow_result.get("error"),
"predictions_agree": (
production_result.get("prediction") == shadow_result.get("prediction")
)
}
# Scrivi su file JSONL per analisi batch
with open(self.shadow_log_file, "a") as f:
f.write(json.dumps(log_entry) + "\n")
# --- Analisi dei risultati shadow ---
def analyze_shadow_results(log_file: str):
"""Analizza i log shadow per validare il nuovo modello."""
import pandas as pd
records = []
with open(log_file) as f:
for line in f:
records.append(json.loads(line))
df = pd.DataFrame(records)
total = len(df)
agreement_rate = df["predictions_agree"].mean()
shadow_errors = df["shadow_error"].notna().sum()
print(f"Totale richieste analizzate: {total:,}")
print(f"Tasso di accordo produzione/shadow: {agreement_rate:.1%}")
print(f"Errori modello shadow: {shadow_errors} ({shadow_errors/total:.1%})")
print(f"Latenza media produzione: {df['production_latency_ms'].mean():.1f}ms")
print(f"Latenza media shadow: {df['shadow_latency_ms'].mean():.1f}ms")
# Identifica i casi di disaccordo per analisi manuale
disagreements = df[~df["predictions_agree"]]
print(f"\nCasi di disaccordo: {len(disagreements)}")
return dfMulti-Armed Bandits: Oltre l'A/B Testing Classico
Il limite principale dell'A/B testing classico e il costo dell'esplorazione: per tutta la durata del test, una frazione degli utenti riceve il modello potenzialmente peggiore. Se il modello B e chiaramente superiore, stiamo "sprecando" le conversioni degli utenti assegnati ad A nelle settimane di test.
I Multi-Armed Bandits (MAB) risolvono il problema exploration-exploitation: invece di mantenere uno split fisso per tutta la durata del test, l'algoritmo adatta dinamicamente il traffico verso il modello che sta performando meglio, massimizzando le conversioni totali durante il test stesso. Una ricerca del 2025 di Aimpoint Digital Labs dimostra che i bandit approach come Thompson Sampling possono ridurre il regret cumulativo del 20-35% rispetto all'A/B testing classico in scenari con effetti forti.
# thompson_sampling_bandit.py
# Multi-Armed Bandit con Thompson Sampling per selezione modello ML
import numpy as np
from dataclasses import dataclass, field
from typing import List, Tuple
import json
import logging
logger = logging.getLogger(__name__)
@dataclass
class ModelArm:
"""Rappresenta un modello come braccio del bandit."""
name: str
endpoint: str
alpha: float = 1.0 # Successi (Beta distribution prior)
beta: float = 1.0 # Fallimenti (Beta distribution prior)
@property
def estimated_success_rate(self) -> float:
"""Stima puntuale del tasso di successo (media della distribuzione Beta)."""
return self.alpha / (self.alpha + self.beta)
@property
def total_observations(self) -> int:
return int(self.alpha + self.beta - 2) # sottrai i prior
def sample(self) -> float:
"""Campiona dalla distribuzione Beta posteriore (Thompson Sampling)."""
return np.random.beta(self.alpha, self.beta)
def update(self, reward: float):
"""
Aggiorna la distribuzione con il nuovo outcome.
reward = 1.0 per successo (churn evitato, conversione, etc.)
reward = 0.0 per fallimento
"""
if reward >= 0.5: # successo
self.alpha += 1
else: # fallimento
self.beta += 1
class ThompsonSamplingBandit:
"""
Multi-Armed Bandit con Thompson Sampling.
Ottimale per selezione adattiva di modelli ML.
"""
def __init__(self, models: List[ModelArm]):
self.models = models
self.selection_history = []
def select_model(self) -> Tuple[int, ModelArm]:
"""
Seleziona il modello campionando dalle distribuzioni Beta.
Il modello con il sample più alto viene selezionato.
"""
samples = [arm.sample() for arm in self.models]
best_idx = int(np.argmax(samples))
self.selection_history.append(best_idx)
return best_idx, self.models[best_idx]
def update(self, arm_idx: int, reward: float):
"""Aggiorna la distribuzione del braccio selezionato."""
self.models[arm_idx].update(reward)
def get_traffic_allocation(self) -> dict:
"""
Stima la distribuzione del traffico corrente
basata sulla storia delle selezioni recenti.
"""
if not self.selection_history:
return {arm.name: 1/len(self.models) for arm in self.models}
recent = self.selection_history[-1000:] # ultime 1000 selezioni
total = len(recent)
allocation = {}
for i, arm in enumerate(self.models):
allocation[arm.name] = recent.count(i) / total
return allocation
def get_status(self) -> dict:
"""Ritorna lo stato corrente del bandit."""
return {
"models": [
{
"name": arm.name,
"estimated_rate": round(arm.estimated_success_rate, 4),
"alpha": arm.alpha,
"beta": arm.beta,
"observations": arm.total_observations
}
for arm in self.models
],
"traffic_allocation": self.get_traffic_allocation(),
"total_selections": len(self.selection_history)
}
def check_convergence(self, min_observations: int = 500) -> Optional[str]:
"""
Verifica se il bandit e convergito verso un vincitore chiaro.
Restituisce il nome del modello vincitore o None se ancora incerto.
"""
for arm in self.models:
if arm.total_observations < min_observations:
return None # Non abbastanza dati
# Controlla se un modello domina chiaramente
rates = [(arm.name, arm.estimated_success_rate) for arm in self.models]
rates.sort(key=lambda x: x[1], reverse=True)
best_name, best_rate = rates[0]
second_name, second_rate = rates[1]
# Margine di 3% di distanza per dichiarare un vincitore
if best_rate - second_rate > 0.03:
logger.info(f"Bandit converged: {best_name} wins ({best_rate:.2%} vs {second_rate:.2%})")
return best_name
return None
# --- Esempio di utilizzo ---
models = [
ModelArm(name="churn-model-v2", endpoint="http://model-v2:8080/predict"),
ModelArm(name="churn-model-v3", endpoint="http://model-v3:8080/predict"),
]
bandit = ThompsonSamplingBandit(models)
# Simulazione di 1000 interazioni
np.random.seed(42)
true_rates = {"churn-model-v2": 0.72, "churn-model-v3": 0.78} # v3 e migliore
for i in range(1000):
arm_idx, selected_model = bandit.select_model()
# Simula outcome (in produzione viene dal feedback reale)
reward = float(np.random.random() < true_rates[selected_model.name])
bandit.update(arm_idx, reward)
if (i + 1) % 200 == 0:
status = bandit.get_status()
print(f"\nStep {i+1}:")
for m in status["models"]:
print(f" {m['name']}: rate={m['estimated_rate']:.3f}, obs={m['observations']}")
winner = bandit.check_convergence(min_observations=100)
if winner:
print(f" => WINNER: {winner}")Analisi Statistica: p-value, Confidence Intervals ed Effect Size
Al termine del periodo di test, l'analisi statistica deve rispondere a tre domande distinte: la differenza osservata e statisticamente significativa? Quanto grande e l'effetto? L'effetto e praticamente rilevante per il business?
# statistical_analysis.py
# Analisi statistica dei risultati di un A/B test ML
import numpy as np
import pandas as pd
from scipy import stats
from scipy.stats import norm, t
import math
from typing import Tuple, Optional
def analyze_ab_test_results(
conversions_a: int,
total_a: int,
conversions_b: int,
total_b: int,
alpha: float = 0.05
) -> dict:
"""
Analisi statistica completa di un A/B test su proporzioni.
Returns:
Dizionario con tutti i risultati statistici
"""
p_a = conversions_a / total_a
p_b = conversions_b / total_b
# --- Test z per differenza di proporzioni ---
# Pooled proportion sotto H0 (le due proporzioni sono uguali)
p_pool = (conversions_a + conversions_b) / (total_a + total_b)
se_pool = math.sqrt(p_pool * (1 - p_pool) * (1/total_a + 1/total_b))
z_statistic = (p_b - p_a) / se_pool
p_value = 2 * (1 - norm.cdf(abs(z_statistic))) # two-tailed
# --- Confidence Interval per la differenza ---
se_diff = math.sqrt(p_a * (1-p_a)/total_a + p_b * (1-p_b)/total_b)
z_critical = norm.ppf(1 - alpha/2)
diff = p_b - p_a
ci_lower = diff - z_critical * se_diff
ci_upper = diff + z_critical * se_diff
# --- Effect size (Cohen's h per proporzioni) ---
phi_a = 2 * math.asin(math.sqrt(p_a))
phi_b = 2 * math.asin(math.sqrt(p_b))
cohens_h = phi_b - phi_a
effect_magnitude = (
"negligible" if abs(cohens_h) < 0.2
else "small" if abs(cohens_h) < 0.5
else "medium" if abs(cohens_h) < 0.8
else "large"
)
# --- Relative lift ---
relative_lift = (p_b - p_a) / p_a if p_a > 0 else 0
# --- Potenza statistica osservata ---
z_beta = (abs(z_statistic) - z_critical)
observed_power = norm.cdf(z_beta)
is_significant = p_value < alpha
return {
"variant_a": {
"conversions": conversions_a,
"total": total_a,
"rate": round(p_a, 4),
"rate_pct": f"{p_a:.2%}"
},
"variant_b": {
"conversions": conversions_b,
"total": total_b,
"rate": round(p_b, 4),
"rate_pct": f"{p_b:.2%}"
},
"difference": {
"absolute": round(diff, 4),
"relative_lift": round(relative_lift, 4),
"relative_lift_pct": f"{relative_lift:.2%}",
"confidence_interval_95": (round(ci_lower, 4), round(ci_upper, 4))
},
"statistics": {
"z_statistic": round(z_statistic, 4),
"p_value": round(p_value, 6),
"is_significant": is_significant,
"alpha": alpha,
"cohens_h": round(cohens_h, 4),
"effect_magnitude": effect_magnitude,
"observed_power": round(observed_power, 4)
},
"conclusion": (
f"Modello B e statisticamente migliore (p={p_value:.4f}, lift={relative_lift:.2%})"
if is_significant and diff > 0
else f"Nessuna differenza significativa rilevata (p={p_value:.4f})"
)
}
# --- Esempio pratico ---
results = analyze_ab_test_results(
conversions_a=1380, # Modello A: 1380 churn evitati
total_a=8500, # su 8500 utenti a rischio
conversions_b=1545, # Modello B: 1545 churn evitati
total_b=8200 # su 8200 utenti
)
print("=== RISULTATI A/B TEST ===")
print(f"Modello A: {results['variant_a']['rate_pct']} retention rate")
print(f"Modello B: {results['variant_b']['rate_pct']} retention rate")
print(f"Lift relativo: {results['difference']['relative_lift_pct']}")
print(f"CI 95%: {results['difference']['confidence_interval_95']}")
print(f"p-value: {results['statistics']['p_value']}")
print(f"Significativo: {results['statistics']['is_significant']}")
print(f"Effect size: {results['statistics']['effect_magnitude']} (h={results['statistics']['cohens_h']})")
print(f"\nConclusione: {results['conclusion']}")A/B Testing Bayesiano
L'approccio frequentista con p-value ha limiti noti: il p-value non e la probabilità che il modello B sia migliore (e la probabilità di osservare dati estremi come questi se H0 fosse vera). L'approccio bayesiano risponde direttamente alla domanda che ci interessa: qual è la probabilità che il modello B sia migliore di A, e di quanto?
L'approccio bayesiano permette anche di fermare il test quando si raggiunge una probabilità sufficientemente alta (es. 95%) che un modello sia il migliore, senza il problema del peeking tipico del frequentismo.
# bayesian_ab_test.py
# A/B Testing Bayesiano per modelli ML
import numpy as np
from scipy import stats
from scipy.stats import beta as beta_dist
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
def bayesian_ab_test(
successes_a: int, trials_a: int,
successes_b: int, trials_b: int,
prior_alpha: float = 1.0,
prior_beta: float = 1.0,
n_samples: int = 100_000,
credible_interval: float = 0.95
) -> dict:
"""
A/B test bayesiano usando distribuzione Beta come prior/posterior.
Modella il tasso di successo come Beta(alpha, beta).
Returns:
Risultati con probabilità che B > A e credible intervals
"""
# Aggiorna i prior con i dati osservati (prior Beta + dati binomiali = posterior Beta)
alpha_a = prior_alpha + successes_a
beta_a = prior_beta + (trials_a - successes_a)
alpha_b = prior_alpha + successes_b
beta_b = prior_beta + (trials_b - successes_b)
# Campiona dalle distribuzioni posterior
samples_a = np.random.beta(alpha_a, beta_a, n_samples)
samples_b = np.random.beta(alpha_b, beta_b, n_samples)
# Probabilità che B sia migliore di A
prob_b_better = np.mean(samples_b > samples_a)
# Distribuzione del lift relativo
lift_samples = (samples_b - samples_a) / samples_a
lift_mean = np.mean(lift_samples)
lift_std = np.std(lift_samples)
# Credible interval per il lift
ci_lower = float(np.percentile(lift_samples, (1 - credible_interval) / 2 * 100))
ci_upper = float(np.percentile(lift_samples, (1 - (1 - credible_interval) / 2) * 100))
# Probabilità di un lift minimo (es. almeno +2%)
prob_lift_2pct = np.mean(lift_samples > 0.02)
# Expected loss: quanto perdiamo se scegliamo il modello sbagliato
expected_loss_a = np.mean(np.maximum(samples_b - samples_a, 0)) # perdita se scegliamo A
expected_loss_b = np.mean(np.maximum(samples_a - samples_b, 0)) # perdita se scegliamo B
return {
"posterior_a": {"alpha": alpha_a, "beta": beta_a, "mean": alpha_a/(alpha_a+beta_a)},
"posterior_b": {"alpha": alpha_b, "beta": beta_b, "mean": alpha_b/(alpha_b+beta_b)},
"prob_b_better_than_a": round(float(prob_b_better), 4),
"lift": {
"mean": round(float(lift_mean), 4),
"std": round(float(lift_std), 4),
f"credible_interval_{int(credible_interval*100)}pct": (
round(ci_lower, 4), round(ci_upper, 4)
),
"prob_lift_above_2pct": round(float(prob_lift_2pct), 4)
},
"expected_loss": {
"choose_a": round(float(expected_loss_a), 6),
"choose_b": round(float(expected_loss_b), 6),
"recommended_choice": "B" if expected_loss_b < expected_loss_a else "A"
},
"decision": (
"Scegli B" if prob_b_better > 0.95
else "Scegli A" if prob_b_better < 0.05
else f"Incerto (P(B>A) = {prob_b_better:.1%}) - continua a raccogliere dati"
)
}
# --- Esempio ---
result = bayesian_ab_test(
successes_a=1380, trials_a=8500,
successes_b=1545, trials_b=8200,
credible_interval=0.95
)
print("=== A/B TEST BAYESIANO ===")
print(f"P(B > A) = {result['prob_b_better_than_a']:.1%}")
print(f"Lift medio: {result['lift']['mean']:.2%}")
print(f"Credible interval 95%: {result['lift']['credible_interval_95pct']}")
print(f"P(lift > 2%): {result['lift']['prob_lift_above_2pct']:.1%}")
print(f"Expected loss se scegli A: {result['expected_loss']['choose_a']:.6f}")
print(f"Expected loss se scegli B: {result['expected_loss']['choose_b']:.6f}")
print(f"Decisione: {result['decision']}")Monitoring Durante i Test con Prometheus e Grafana
Un A/B test attivo in produzione deve essere monitorato continuamente. Non basta aspettare la fine del test per analizzare i risultati: bisogna garantire che entrambe le varianti funzionino correttamente sul piano tecnico (latenza, error rate, disponibilità) e che le metriche di business siano in linea con le aspettative iniziali.
# ab_test_monitoring.yml
# Dashboard Grafana per A/B test ML - configurazione panel
# Esempio di PromQL queries per i panel Grafana:
# 1. Distribuzione del traffico tra varianti (dovrebbe essere ~50/50)
# sum by (variant) (rate(ab_test_requests_total[5m]))
# 2. Latenza P95 per variante
# histogram_quantile(0.95, sum by (variant, le) (rate(ab_test_latency_seconds_bucket[5m])))
# 3. Error rate per variante
# sum by (variant) (rate(ab_test_errors_total[5m])) /
# sum by (variant) (rate(ab_test_requests_total[5m]))
# 4. Distribuzione delle predizioni per variante (prediction drift indicator)
# sum by (variant, prediction_bucket) (rate(ab_test_predictions_total[1h]))
---
# prometheus_ab_alerts.yml
groups:
- name: ab_test_alerts
rules:
# Alert se il traffico non e bilanciato (sbilanciamento > 10%)
- alert: ABTestTrafficImbalance
expr: |
abs(
sum(rate(ab_test_requests_total{variant="A"}[10m]))
/
sum(rate(ab_test_requests_total[10m]))
- 0.5
) > 0.10
for: 5m
labels:
severity: warning
annotations:
summary: "A/B test traffic imbalance detected"
description: "Traffic split deviates more than 10% from 50/50"
# Alert se error rate variante B supera il doppio di A
- alert: ABTestVariantBHighErrors
expr: |
(
sum(rate(ab_test_errors_total{variant="B"}[5m]))
/
sum(rate(ab_test_requests_total{variant="B"}[5m]))
) > 2 * (
sum(rate(ab_test_errors_total{variant="A"}[5m]))
/
sum(rate(ab_test_requests_total{variant="A"}[5m]))
)
for: 10m
labels:
severity: critical
annotations:
summary: "Variant B has significantly higher error rate than A"
description: "Consider rolling back variant B"
# Alert se latenza P95 di B supera 200ms più di A
- alert: ABTestVariantBHighLatency
expr: |
(
histogram_quantile(0.95, sum by (le) (
rate(ab_test_latency_seconds_bucket{variant="B"}[5m])
))
-
histogram_quantile(0.95, sum by (le) (
rate(ab_test_latency_seconds_bucket{variant="A"}[5m])
))
) > 0.2
for: 5m
labels:
severity: warning
annotations:
summary: "Variant B P95 latency is 200ms+ higher than A"Budget <5K EUR/Anno per PMI: Stack A/B Testing Completo
Un sistema completo di A/B testing per modelli ML non richiede budget enterprise. Con lo stack open-source e una piccola VPS, si può avere tutto il necessario:
- FastAPI router + Python stats: Open-source, gratuito
- Prometheus + Grafana: Open-source, gratuito
- VPS per hosting (Hetzner/OVH): 20-40 EUR/mese (240-480 EUR/anno)
- Feature flag service (Unleash self-hosted): Open-source, gratuito
- MLflow per model registry: Open-source, gratuito
- Totale stimato infrastruttura: 300-600 EUR/anno
Best Practices e Anti-Pattern
Checklist Pre-Esperimento
- Calcola il sample size prima di partire: non fare mai test "finchè non vedo qualcosa di interessante". Il sample size prefissato e non negoziabile.
- Definisci UNA primary metric: ottimizzare per due metriche simultaneamente rende la decisione ambigua. Le guardrail metrics esistono per prevenire regressioni, non per eleggere vincitori.
- Testa la validita dell'assignment: prima di lanciare il test reale, fai un A/A test (stesso modello su entrambe le varianti) per verificare che non ci siano bug nell'assegnazione che causino differenze artificiali.
- Documenta le ipotesi: annota perchè ci si aspetta che il modello B sia migliore e di quanto. Questo evita il bias del "ho visto la risposta e ora invento una spiegazione".
- Controlla i nuovi utenti separatamente: i nuovi utenti non hanno storia pregressa con nessuno dei modelli e hanno comportamenti diversi. Analizza i loro risultati separatamente.
Anti-Pattern da Evitare Assolutamente
- Peeking continuo: controllare i risultati ogni giorno e fermare il test alla prima significativita statistica aumenta i falsi positivi fino al 30%. Usa test sequenziali (SPRT) se hai bisogno di stopping anticipato.
- HARKing (Hypothesizing After Results are Known): analizzare i dati per trovare qualsiasi differenza significativa, poi raccontare la storia come se fosse stata ipotizzata a priori. In un test su 20 segmenti, uno risultera significativo solo per caso con alpha = 0.05.
- Ignorare la varianza delle metriche: alcune metriche come il revenue per utente hanno code molto pesanti. Un singolo whale user può far sembrare significativo un effetto che non esiste. Usa bootstrap o test non parametrici per metriche con distribuzione non gaussiana.
- Test troppo brevi: effetti settimanali (utenti che usano il servizio solo di lunedi) e novelty effect (gli utenti reagiscono positivamente alla novità per 1-2 giorni poi tornano alla baseline) richiedono test di almeno 2 settimane per essere compensati.
- Feedback loops nei sistemi ML: in sistemi con feedback loops (raccomandazioni che cambiano il comportamento degli utenti), le predizioni delle due varianti non sono indipendenti. Modella esplicitamente questa correlazione.
Quando Usare Quale Approccio
Guida alla Scelta della Strategia
- Shadow Mode: usa quando il modello e completamente nuovo, non ancora validato, o quando il rischio di un bug e troppo alto. E sempre il primo step prima di qualsiasi test con utenti reali.
- Canary Deployment: usa quando vuoi ridurre il rischio operativo di un nuovo deployment. Ottimo per modelli critici (fraud, pricing) dove una regressione avrebbe impatto finanziario immediato.
- A/B Test Classico (50/50): usa quando vuoi misurare l'effetto di business con massima potenza statistica e il rischio operativo e basso. Richiede sample size sufficiente e feedback loop rapido.
- Multi-Armed Bandit: usa quando il feedback e rapido (entro ore/giorni), il costo dell'esplorazione e alto e preferisci massimizzare le conversioni durante il test. Non ideale per effetti piccoli con feedback lento.
- A/B Bayesiano: usa quando vuoi stopping rules flessibili, interpretare le probabilità direttamente o hai prior information da esperimenti precedenti. Ideale per team che trovano i p-value confusionari.
Conclusioni e Prossimi Passi
L'A/B testing per modelli ML e molto più di un semplice split del traffico. Richiede un design statistico rigoroso prima di ogni implementazione, la scelta della strategia giusta in base al contesto (shadow, canary, 50/50, bandit), un monitoring continuo durante il test e un'analisi statistica corretta alla fine.
La differenza tra un team che fa A/B testing correttamente e uno che lo fa male non e nella complessità del codice, ma nella disciplina del processo: definire le ipotesi prima, non guardare i dati durante, analizzare tutto correttamente dopo. Con lo stack open-source descritto in questa guida (FastAPI, Prometheus, Grafana, scipy, numpy), puoi implementare un sistema production-grade con budget minimo.
Il passo successivo naturale e integrare l'A/B testing con la governance ML: ogni decisione di promuovere un modello in produzione deve essere documentata, auditabile e conforme agli standard etici e normativi. Lo vedremo nell'articolo successivo sulla Governance ML.
Continua la Serie MLOps
- Articolo precedente: Scaling ML su Kubernetes - orchestrare deployment con KubeFlow e Seldon Core
- Articolo successivo: Governance ML: Compliance, Audit, Ethics - AI Act EU, explainability e fairness
- Correlato: Model Drift Detection e Retraining Automatico - rilevare e reagire al degrado del modello
- Correlato: Serving Modelli: FastAPI + Uvicorn in Produzione - costruire API inference scalabili
- Serie correlata: Deep Learning Avanzato - A/B testing per modelli neurali complessi