Testarea A/B a modelelor ML: Metodologie, metrici și implementare
Ați antrenat două versiuni ale modelului dumneavoastră de recomandare. Noul model, bazat pe transformator, arată o AUC cu 3% mai mare pe setul de rezistență. Pare o îmbunătățire clară, dar această diferență se traduce într-adevăr într-un impact pozitiv pentru utilizatorii reali? Modelul poate performa mai bine pe anumite cohorte demografice și mai rău pe altele. Ar putea reduce rata de clic, crescând în același timp satisfacția pe termen lung. Poate avea latență mai mare, ceea ce anulează beneficiile de precizie. Valorile offline nu mint, spun ei doar o parte din poveste.
L'Testare A/B pentru modele ML și metodologia care vă permite să răspundeți la aceste întrebări riguros, comparând versiuni ale unui model pe trafic real cu utilizatori reali, măsurând valorile de afaceri care contează cu adevărat. Conform cercetărilor efectuate de 2025 de către Aimpoint Digital Labs, organizații care adoptă strategii structurate de testare A/B pentru modelele ML acestea reduc cu 40% risc de regresie în producție față de implementări directe bazate doar pe valori offline. Piața MLOps, care merită 4,38 miliarde USD în 2026 și se așteaptă să ajungă la 89,18 miliarde USD până în 2035, iar A/B testarea este unul dintre elementele fundamentale ale acestei creșteri.
În acest ghid vom construi un sistem complet de testare A/B pentru modelele ML: din teorie statistici către routerul FastAPI, de la implementarea canary la modul umbră, de la testele Frequentist la abordarea bayesiană cu Thompson Sampling, până la monitorizarea metricilor în timpul teste cu Prometeu si Grafana.
Ce vei învăța
- Diferența dintre testarea ML A/B și testarea web A/B clasică
- Proiectarea experimentelor: dimensiunea eșantionului, puterea statistică, metrica succesului
- Divizarea traficului cu routere FastAPI și implementare progresivă Canary
- Modul umbră: testați fără a afecta utilizatorii
- Multi-Armed Bandits și Thompson Sampling ca alternativă la testarea A/B clasică
- Analiză statistică: valoarea p, intervale de încredere, mărimea efectului
- Testare Bayesian A/B pentru decizii mai rapide
- Monitorizare în timpul testării cu Prometheus și Grafana
- Cele mai bune practici și anti-modele de evitat
Testarea ML A/B vs testarea Web A/B: diferențe critice
Testarea A/B a luat naștere în analiza web pentru a compara variațiile paginilor de destinație, butoanelor și copie. Cadrul statistic de bază este același, dar testarea A/B pentru modelele ML a făcut-o complexități suplimentare care o fac substanțial diferită în practică.
În testarea web, sunt comparate experiențe vizuale discrete: varianta A și varianta B sunt clar separate. În modelele ML, predicțiile sunt continue, distribuite și adesea corelate în timp. Un model de recomandare care deservește același utilizator în diferite sesiuni nu produce predicţii independente: există corelaţie temporală ceea ce încalcă ipotezele de independenţă ale testelor statistice clasice.
Diferențele cheie ML vs Web A/B Testing
- Valori: pe web, CTR-ul sau rata de conversie este optimizată; in ML da optimizați simultan valorile offline (AUC, RMSE) și valorile de afaceri (venituri, rata de abandon, NPS), adesea conflictuale.
- Latența feedback-ului: pe web rezultatul este imediat (click); în ML poate dura zile sau săptămâni (renuntare după 30 de zile, venituri după un trimestru).
- Distribuția efectelor: un model poate avea rezultate mai bune în medie dar mai rău pe anumite cohorte (vârstă, părtinire geografică), necesitând analiză segmentată.
- Efecte de sistem: în sisteme de buclă de feedback (recomandări, stabilirea prețurilor dinamice), modelul B influențează datele care vor antrena apoi modelul C.
- Riscuri operaționale: un bug într-o variantă web provoacă un UX prost; o eroare într-un model ML de detectare a fraudei poate cauza pierderi financiare semnificative.
Proiectarea experimentului: înainte de cod
Un test A/B prost proiectat este mai rău decât niciun test A/B: oferă un sentiment fals al rigoare științifică producând în același timp concluzii incorecte. Proiectarea experimentului trebuie preced orice implementare tehnică.
Definiți valorile de succes
Fiecare experiment trebuie să aibă un Valori primare singurul care determină câștigătorul, plus zero la doi Valori de gardă acel model B nu trebuie să fie mai rău decât A. Metrica primară trebuie să fie directă legate cauzal de obiectivul de afaceri.
Exemple de valori pentru diferite scenarii:
- Model de ridicare: primar = rata de retenție de 30 de zile; balustradă = latență P95, costul campaniei
- Model de recomandare: primar = venit pe sesiune; balustradă = CTR, recomandări diversitate
- Model de fraudă: primar = rata de fraudă nedetectată; balustradă = rată fals pozitivă, latență
- Model de preț: primar = marja bruta; balustradă = rata de conversie, NPS
Calculul dimensiunii eșantionului
Dimensiunea necesară a eșantionului depinde de trei factori:dimensiunea efectului minim pe care doriți să le detectați (efect minim detectabil, MDE), nivelul de semnificație alfa (de obicei 0,05) și la putere statistică 1-beta (de obicei 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 richiestePeeking Capcana: Nu te uita la rezultate prea curând
Problema „peeking” (sau oprire opțională) este una dintre cele mai frecvente erori în testarea A/B: ne uităm la rezultatele intermediare și oprim testul de îndată ce se ajunge la semnificație statistici. Acest lucru crește dramatic rata fals pozitive: dacă te uiți la date În fiecare zi, probabilitatea de a găsi un rezultat semnificativ din întâmplare crește la 30% chiar dacă cele două variante sunt identice. Utilizați întotdeauna o dimensiune a eșantionului predeterminat și verifică rezultatele numai la sfârșitul testului, sau adoptă metode de testare secvenţială, cum ar fi testele cu raportul de probabilitate secvenţială (SPRT).
Împărțirea traficului cu FastAPI
Routerul de testare A/B este componenta centrală a infrastructurii. Trebuie distribuit trafic într-un mod determinist (același utilizator trebuie să meargă întotdeauna pe același variantă pe toată durata testului), înregistrați la ce variantă ați fost repartizat fiecare utilizator și fiecare predicție și fiți extrem de rapid pentru a nu adăuga latență spre calea critică.
# 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
}Desfășurare Canary: lansare progresivă
Il desfăşurare canar și o strategie de eliberare progresivă unde noul model („canarul”) primește inițial doar un mic procent din traficul de producție, de obicei 1-5%. Dacă valorile rămân stabile, procentul crește treptat: 5% → 10% → 25% → 50% → 100%. În cazul unor anomalii, rollback-ul este imediat, readucend tot traficul înapoi la modelul stabil.
Spre deosebire de clasicul test 50/50 A/B, canarul este orientat reducerea riscului mai mult decât să detectarea statistică a diferenţelor. Scopul nu este de a demonstra că noul model este mai bun cu semnificație statistică, dar verificați dacă nu este cauza probleme tehnice vizibile sau regresii înainte de a scala implementarea.
# 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}")Modul umbră: testare fără impact asupra utilizatorilor
Lo modul umbră (sau desfășurarea în umbră) este cea mai conservatoare tehnică și, în același timp, cel mai puternic pentru validarea unui nou model înainte de a-l expune utilizatorilor. Traficul de producție este duplicat: Modelul A servește cererile reale și propriile sale predicțiile sunt returnate utilizatorilor, în timp ce modelul B primește aceleași solicitări în paralel dar vin previziunile lui eliminate sau doar autentificate.
Această abordare vă permite să comparați cele două modele pe trafic real fără niciunul risc pentru utilizatori sau pentru afacere. Este ideal pentru a valida că noul model: nu are erori critice, îndeplinește cerințele de latență la încărcare reală, nu produce predicții anormale sau în afara distribuției și se comportă conform așteptărilor pe toate segmentele de utilizatori.
# 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 dfBandiți cu mai multe arme: dincolo de testarea A/B clasică
Principala limitare a testării A/B clasice este costul explorării: pe durata testului, o fracțiune de utilizatori pot primi modelul mai rău. Dacă modelul B este net superior, „irosim” conversii de utilizatori alocați lui A în săptămânile de testare.
I Bandiți cu mai multe armate (MAB) ele rezolvă problema explorare-exploatare: în loc să se mențină o divizare fixă pe toată durata testului, algoritmul se adaptează dinamic trafic către modelul care funcționează mai bine, maximizând conversiile totale în timpul testului în sine. O căutare din 2025 de către Aimpoint Digital Labs demonstrează că abordări bandit, cum ar fi Thompson Sampling poate reduce regretul cumulat al 20-35% comparativ cu testarea A/B clasică în scenarii cu efecte puternice.
# 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}")Analiză statistică: valoarea p, intervale de încredere și dimensiunea efectului
La sfârșitul perioadei de testare, analiza statistică trebuie să răspundă la trei întrebări distincte: Este diferența observată semnificativă statistic? Cât de mare este efectul? Este efectul practic relevant pentru afacere?
# 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']}")Testare Bayesian A/B
Abordarea frecventistă a valorii p are limitări cunoscute: valoarea p nu este probabilitatea acel model B este mai bun (și probabilitatea de a observa date extreme ca aceasta dacă H0 ar fi adevărată). Abordarea Bayesian răspunde direct la întrebarea care ne interesează: care este probabilitatea ca modelul B să fie mai bun decât A, si cu cat?
Abordarea bayesiană vă permite, de asemenea, să opriți testul când ajunge la a probabilitate suficient de mare (de exemplu, 95%) ca un model să fie cel mai bun, fără problema peeking-ului tipic frecventismului.
# 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']}")Monitorizare în timpul testelor cu Prometheus și Grafana
Un test A/B activ în producție trebuie monitorizat continuu. Nu este suficient așteptați până la sfârșitul testului pentru a analiza rezultatele: trebuie garantat că ambele variantele functioneaza corect la nivel tehnic (latenta, rata de eroare, disponibilitate) și că valorile de afaceri sunt în conformitate cu așteptările inițiale.
# 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"Bugetul <5.000 EUR/an pentru IMM-uri: testare completă A/B Stack
Un sistem complet de testare A/B pentru modelele ML nu necesită un buget al întreprinderii. Cu stiva open-source și un VPS mic, puteți avea tot ce aveți nevoie:
- Router FastAPI + statistici Python: Open-source, gratuit
- Prometheus + Grafana: Open-source, gratuit
- VPS pentru hosting (Hetzner/OVH): 20-40 EUR/lună (240-480 EUR/an)
- Serviciu de semnalizare a caracteristicilor (Unleash self-hosted): Open-source, gratuit
- MLflow pentru registrul modelului: Open-source, gratuit
- Total infrastructură estimată: 300-600 EUR/an
Cele mai bune practici și anti-modele
Lista de verificare pre-experiment
- Calculați dimensiunea eșantionului înainte de a începe: nu faci niciodată teste „până când Nu văd nimic interesant.” Mărimea eșantionului este fixă și nenegociabilă.
- Definiți O valoare principală: optimizați pentru două valori simultan face decizia ambiguă. Valorile de gardă există pentru pentru a preveni regresiile, nu pentru a alege câștigătorii.
- Testați validitatea misiunii: înainte de a lansa testul real, faceți un test A/A (același model pe ambele variante) pentru a verifica că nu există erori în sarcină care provoacă diferențe artificiale.
- Documentează-ți ipotezele: rețineți de ce ne așteptăm la modelul B este mai bine si cu cat. Acest lucru evită părtinirea „Am văzut răspunsul și acum”. Voi inventa o explicație.”
- Verificați separat utilizatorii noi: utilizatorii noi nu au istoria anterioară cu niciunul dintre modele și au comportamente diferite. Analizează rezultatele lor separat.
Anti-modele de evitat absolut
- Privire continuă: verificați rezultatele în fiecare zi e oprirea testului la prima semnificație statistică crește fals pozitive până la 30%. Utilizați teste secvențiale (SPRT) dacă aveți nevoie de oprire timpurie.
- HARKing (Ipoteza după ce rezultatele sunt cunoscute): analiza datele pentru a găsi orice diferențe semnificative, apoi spuneți povestea de parcă ar fi fost emisă ipoteza a priori. Într-un test de 20 de segmente, unul va ieși în evidență semnificativ doar întâmplător cu alfa = 0,05.
- Ignorați varianța valorii: unele valori precum venitul per utilizator au cozi foarte grele. Un singur utilizator de balenă poate face acest lucru un efect care nu există pare semnificativ. Utilizați bootstrap sau non-testare parametrice pentru metrici cu distribuție non-Gauss.
- Teste prea scurte: efecte săptămânale (utilizatorii care folosesc serviciu numai luni) și efect de noutate (utilizatorii reacționează pozitiv la noutate timp de 1-2 zile apoi revenirea la valoarea inițială) necesită testarea a cel puțin 2 saptamani de compensat.
- Bucle de feedback în sistemele ML: în sistemele cu bucle de feedback (recomandări care modifică comportamentul utilizatorului), previziunile de două variante nu sunt independente. Modelați această corelație în mod explicit.
Când să folosiți care abordare
Ghid pentru alegerea unei strategii
- Modul umbră: utilizați când modelul este complet nou, nu a fost încă validat sau când riscul unei erori este prea mare. Și întotdeauna primul pas înainte de orice testare cu utilizatori reali.
- Desfășurare Canary: utilizați atunci când doriți să reduceți riscul operațional a unei noi implementări. Excelent pentru modele critice (fraudă, prețuri) unde o regresie ar avea un impact financiar imediat.
- Test A/B clasic (50/50): utilizați atunci când doriți să măsurați efectul afaceri cu putere statistică maximă și risc operațional scăzut. Necesită o dimensiune suficientă a eșantionului și o buclă rapidă de feedback.
- Bandit cu mai multe arme: utilizați atunci când feedback-ul este rapid (în câteva ore/zile), costul explorării este mare și preferați să maximizați conversiile în timpul testului. Nu este ideal pentru efecte mici cu feedback lent.
- Bayesian A/B: utilizați reguli de oprire flexibile oricând doriți, să interpreteze direct probabilitățile sau să aibă informații anterioare din experimente precedente. Ideal pentru echipele care consideră valorile p confuze.
Concluzii și pașii următori
Testarea A/B pentru modelele ML este mult mai mult decât simpla împărțire a traficului. Este nevoie de proiectare statistică riguroasă înainte de fiecare implementare, alegere strategia potrivită în funcție de context (umbră, canar, 50/50, bandit), monitorizare continuă în timpul testului și analiză statistică corectă la final.
Diferența dintre o echipă care face corect testele A/B și una care o face prost nu este în complexitatea codului, ci în disciplina procesului: definirea mai întâi ipotezele, nu te uita la datele în timpul, analizează totul corect după aceea. Cu stiva open-source descrisă în acest ghid (FastAPI, Prometheus, Grafana, scipy, numpy), puteți implementa un sistem de producție cu buget minim.
Următorul pas natural este integrarea testării A/B cu guvernarea ML: fiecare decizie de a promova un model în producție trebuie să fie documentată, auditabile și conforme cu standardele etice și de reglementare. O vom vedea în articol următorul despre ML Governance.
Seria MLOps continuă
- Articolul precedent: Scalare ML pe Kubernetes - orchestrați implementările cu KubeFlow și Seldon Core
- Articolul următor: Guvernare ML: conformitate, audit, etică - AI Act EU, explicabilitate și corectitudine
- Înrudit: Detectarea derivei modelului și reinstruire automată - detectează și reacționează la degradarea modelului
- Înrudit: Modele de servire: FastAPI + Uvicorn în producție - construiți API-uri de inferență scalabile
- Serii înrudite: Învățare profundă avansată - Testare A/B pentru modele neuronale complexe