A/B testování ML modelů: Metodika, metriky a implementace
Trénovali jste dvě verze svého modelu doporučení. Nový model, založený na transformátor, vykazuje o 3 % vyšší AUC na sadě výdrže. Vypadá to na jasné zlepšení, ale skutečně se tento rozdíl promítá do pozitivního dopadu na skutečné uživatele? Modelka v některých demografických kohortách může fungovat lépe a v jiných hůře. Mohlo by to snížit míru prokliku a zároveň zvýšit dlouhodobou spokojenost. Může mít latenci vyšší, což ruší výhody přesnosti. Offline metriky nelžou, říkají jen část příběhu.
L'A/B testování pro ML modely a metodologii, která vám umožní reagovat tyto otázky důsledně srovnávat verze modelu se skutečným provozem skutečných uživatelů, měření obchodních metrik, na kterých opravdu záleží. Podle výzkumu od 2025 od Aimpoint Digital Labs, organizací, které přijímají strategie strukturovaného testování A/B u modelů ML snižují o 40% riziko regrese ve výrobě oproti přímému nasazení pouze na základě offline metrik. Trh MLOps, který stojí za to 4,38 miliardy $ v roce 2026 a očekává se, že do roku 2035 dosáhne 89,18 miliardy $ a A/B testování je jedním ze základních stavebních kamenů tohoto růstu.
V této příručce vytvoříme kompletní A/B testovací systém pro ML modely: z teorie statistiky do routeru FastAPI, od nasazení canary po režim stínu, z testů Frequentist na Bayesovský přístup s Thompson Sampling, až po sledování metrik během testy s Prometheem a Grafanou.
Co se naučíte
- Rozdíl mezi ML A/B testováním a klasickým webovým A/B testováním
- Návrh experimentů: velikost vzorku, statistická síla, metriky úspěšnosti
- Rozdělení provozu pomocí směrovačů FastAPI a progresivní nasazení canary
- Stínový režim: test bez dopadu na uživatele
- Multi-Armed Bandits a Thompson Sampling jako alternativa ke klasickému A/B testování
- Statistická analýza: p-hodnota, intervaly spolehlivosti, velikost účinku
- Bayesovské A/B testování pro rychlejší rozhodování
- Monitorování během testování s Prometheus a Grafana
- Doporučené postupy a anti-vzory, kterým je třeba se vyhnout
ML A/B testování vs Web A/B testování: kritické rozdíly
A/B testování se zrodilo ve webové analýze pro porovnání variant vstupních stránek, tlačítek a kopírovat. Základní statistický rámec je stejný, ale A/B testování pro ML modely ano další složitosti, které jej v praxi podstatně odlišují.
Při testování webu se porovnávají diskrétní vizuální zážitky: varianta A a varianta B jsou jasně odděleny. V modelech ML jsou předpovědi spojité, distribuované a často koreluje v čase. Model doporučení, který slouží stejnému uživateli v různých relacích nevytváří nezávislé předpovědi: existuje časová korelace což porušuje předpoklady nezávislosti klasických statistických testů.
Klíčové rozdíly ML vs Web A/B testování
- Metriky: na webu je optimalizována CTR nebo konverzní poměr; v ML ano současně optimalizovat offline metriky (AUC, RMSE) a obchodní metriky (výnosy, míra odchodu, NPS), často protichůdné.
- Latence zpětné vazby: na webu je výsledek okamžitý (kliknutí); v ML může trvat dny nebo týdny (uvolnění po 30 dnech, tržby po čtvrtletí).
- Distribuce efektu: model může mít v průměru lepší výkon ale horší u konkrétních kohort (ageismus, geografické zkreslení), které vyžadují segmentovanou analýzu.
- Systémové efekty: v systémech zpětné vazby (doporučení, dynamické oceňování), model B ovlivňuje data, která pak budou trénovat model C.
- Provozní rizika: chyba ve webové variantě způsobuje špatné uživatelské prostředí; chyba v modelu ML detekce podvodů může způsobit značné finanční ztráty.
Návrh experimentu: Před kódem
Špatně navržený A/B test je horší než žádný A/B test: poskytuje falešný dojem vědeckou přísnost a zároveň vyvozovat nesprávné závěry. Návrh experimentu musí předcházet jakékoli technické realizaci.
Definujte metriky úspěchu
Každý experiment musí mít a Primární metriky jediný, který určuje vítěz plus nula ku dvěma Metriky zábradlí ten model B nesmí být horší než A. Primární metrika musí být přímo v příčinné souvislosti s obchodním cílem.
Příklady metrik pro různé scénáře:
- Model odvalu: primární = 30denní míra retence; zábradlí = latence P95, náklady na kampaň
- Model doporučení: primární = tržby za relaci; mantinel = CTR, doporučení diverzity
- Model podvodu: primární = míra nezjištěných podvodů; zábradlí = míra falešné pozitivity, latence
- Cenový model: primární = hrubá marže; mantinel = konverzní poměr, NPS
Výpočet velikosti vzorku
Potřebná velikost vzorku závisí na třech faktorech:velikost efektu minimální které chcete detekovat (minimální detekovatelný efekt, MDE), úroveň významnosti alfa (obvykle 0,05) a la statistická síla 1-beta (obvykle 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 richiestePokukující past: Nedívejte se na výsledky příliš brzy
Problém „nakouknutí“ (nebo volitelného zastavení) je jednou z nejčastějších chyb v A/B testování: podíváme se na průběžné výsledky a zastavíme test, jakmile je dosaženo významnosti statistiky. To dramaticky zvyšuje míru falešně pozitivních výsledků: pokud se podíváte na data Každým dnem stoupá pravděpodobnost náhodného nalezení významného výsledku 30 %, i když jsou obě varianty totožné. Vždy používejte velikost vzorku předem stanoveny a zkontrolovat výsledky až na konci testu, případně přijmout metody sekvenčního testování, jako jsou sekvenční testy poměru pravděpodobnosti (SPRT).
Rozdělení provozu pomocí FastAPI
Směrovač pro testování A/B je ústřední součástí infrastruktury. Musí distribuovat provoz deterministickým způsobem (stejný uživatel musí vždy jet na stejném varianta po celou dobu trvání testu), zaznamenejte si, ke které variantě jste byli zařazeni každého uživatele a každou předpověď a buďte extrémně rychlí, abyste nezvýšili latenci na kritickou cestu.
# 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: Progressive Release
Il kanárské nasazení a strategie progresivního vydávání, kde nový model („kanárek“) zpočátku dostává jen malé procento produkčního provozu, obvykle 1–5 %. Pokud metriky zůstanou stabilní, procento se postupně zvyšuje: 5 % → 10 % → 25 % → 50 % → 100 %. V případě anomálií je rollback okamžitý, čímž se veškerý provoz vrátí zpět do stabilního modelu.
Na rozdíl od klasického testu 50/50 A/B je kanárek orientovaný snížení rizika více než do statistické zjišťování rozdílů. Cílem není demonstrovat že nový model je lepší se statistickou významností, ale ověřte, že tomu tak není způsobit znatelné technické problémy nebo regrese před škálováním nasazení.
# 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}")Stínový režim: Testování bez dopadu na uživatele
Lo stínový režim (neboli nasazení stínů) je nejkonzervativnější technika a zároveň nejvýkonnější pro ověřování nového modelu před jeho vystavením uživatelům. Produkční provoz je duplikován: Model A obsluhuje skutečné požadavky i své vlastní předpovědi jsou vráceny uživatelům, zatímco model B přijímá stejné požadavky paralelně, ale přicházejí jeho předpovědi zahozen nebo pouze přihlášen.
Tento přístup vám umožňuje porovnávat dva modely na reálném provozu bez jakéhokoli riziko pro uživatele nebo obchod. Je ideální pro ověření, že nový model: nemá kritické chyby, splňuje požadavky na latenci při reálném zatížení, nevytváří anomální předpovědi nebo předpovědi mimo distribuci a chová se podle očekávání napříč všemi segmenty uživatelů.
# 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 banditi: Beyond Classical A/B Testing
Hlavním omezením klasického A/B testování je náklady na průzkum: po dobu trvání testu model potenciálně obdrží zlomek uživatelů horší. Pokud je model B jednoznačně lepší, „plýtváme“ konverzemi uživatelů přiřazených k A v testovacích týdnech.
I Mnohorucí bandité (MAB) problém vyřeší průzkum-využívání: místo zachování pevného rozdělení po celou dobu trvání testu, algoritmu dynamicky přizpůsobovat návštěvnost modelu, který pracuje lépe, maximalizovat celkové konverze během samotného testu. Hledání z roku 2025 od Aimpoint Digital Labs ukazuje, že přístupy banditů, jako je Thompson Sampling může snížit kumulativní lítost 20-35% oproti klasickému A/B testování ve scénářích se silnými účinky.
# 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}")Statistická analýza: p-hodnota, intervaly spolehlivosti a velikost účinku
Na konci testovacího období musí statistická analýza odpovědět na tři různé otázky: Je pozorovaný rozdíl statisticky významný? Jak velký je efekt? Je efekt prakticky relevantní pro podnikání?
# 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']}")Bayesovské A/B testování
Frekvenční přístup p-hodnoty má známá omezení: p-hodnota není pravděpodobnost že model B je lepší (a pravděpodobnost pozorování extrémních dat, jako je tento kdyby H0 byla pravda). Přístup Bayesovský odpovídá přímo na otázku, která nás zajímá: jaká je pravděpodobnost, že model B je lepší než A, a o kolik?
Bayesovský přístup také umožňuje zastavit test, když dosáhne a dostatečně vysoká pravděpodobnost (např. 95 %), že model je nejlepší, bez problému s pokukováním typickým pro frekventismus.
# 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']}")Monitorování během testů s Prometheus a Grafana
Aktivní A/B test ve výrobě musí být nepřetržitě monitorován. To nestačí počkejte s analýzou výsledků do konce testu: musí být zaručeno, že obojí varianty fungují správně na technické úrovni (latence, chybovost, dostupnost) a že obchodní metriky jsou v souladu s původními očekáváními.
# 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"Rozpočet <5 000 EUR/rok pro malé a střední podniky: Dokončete testování zásobníku A/B
Kompletní A/B testovací systém pro modely ML nevyžaduje podnikový rozpočet. S open-source stackem a malým VPS můžete mít vše, co potřebujete:
- FastAPI router + statistiky Pythonu: Open source, zdarma
- Prometheus + Grafana: Open source, zdarma
- VPS pro hosting (Hetzner/OVH): 20-40 EUR/měsíc (240-480 EUR/rok)
- Služba příznaku funkce (Unleash self-hosted): Open source, zdarma
- MLflow pro registr modelů: Open source, zdarma
- Celková odhadovaná infrastruktura: 300-600 EUR/rok
Osvědčené postupy a anti-vzorce
Kontrolní seznam před experimentem
- Před zahájením vypočítejte velikost vzorku: nikdy nedělej testy, dokud Nevidím nic zajímavého." Velikost vzorku je pevná a nelze s ní obchodovat.
- Definujte JEDNU primární metriku: optimalizovat pro dvě metriky zároveň činí rozhodnutí nejednoznačným. Metriky zábradlí existují pro zabránit regresi, ne volit vítěze.
- Otestujte platnost zadání: před spuštěním skutečného testu, proveďte A/A test (stejný model na obou variantách), abyste ověřili, že ne v zadání jsou chyby, které způsobují umělé rozdíly.
- Zdokumentujte své hypotézy: všimněte si, proč očekáváme model B je lepší a o kolik. Vyhnete se tak zkreslení „Viděl jsem odpověď a teď“. Vymyslím vysvětlení."
- Zkontrolujte nové uživatele samostatně: noví uživatelé nemají předchozí historie bez žádného z modelů a mají odlišné chování. Analyzujte jejich výsledky samostatně.
Anti-vzorce, kterým se absolutně vyhnout
- Průběžné prohlížení: kontrolovat výsledky každý den e zastavení testu při první statistické významnosti zvyšuje falešně pozitivní výsledky až 30 %. Pokud potřebujete včasné ukončení, použijte sekvenční testy (SPRT).
- HARKing (hypotetizace poté, co jsou výsledky známy): analyzovat data najít nějaké významné rozdíly, pak vyprávět příběh jako by to bylo a priori předpokládané. V testu 20 segmentů vynikne jeden významný pouze náhodou s alfa = 0,05.
- Ignorovat metrický rozptyl: některé metriky jako např tržby na uživatele mají velmi těžké fronty. Může to udělat jeden uživatel velryby efekt, který neexistuje, se zdá významný. Použijte bootstrap nebo netestování parametrické pro metriky s negaussovským rozdělením.
- Testy jsou příliš krátké: týdenní efekty (uživatelé, kteří používají služba pouze v pondělí) a efekt novinky (uživatelé reagují pozitivně). na novost na 1–2 dny a poté se vraťte na výchozí úroveň) vyžadují testování minimálně 2 týdny na kompenzaci.
- Smyčky zpětné vazby v systémech ML: v systémech se zpětnovazebními smyčkami (doporučení, která mění chování uživatelů), předpovědi dvě varianty nejsou nezávislé. Modelujte tuto korelaci explicitně.
Kdy použít který přístup
Průvodce výběrem strategie
- Režim stínu: používat, když je model zcela nový, ještě neověřené, nebo když je riziko chyby příliš vysoké. A vždy první krok před jakýmkoli testováním se skutečnými uživateli.
- Nasazení Canary: použijte, když chcete snížit provozní riziko nového nasazení. Skvělé pro kritické modely (podvody, ceny), kde regrese by měla okamžitý finanční dopad.
- Klasický A/B test (50/50): použijte, když chcete změřit účinek podnikání s maximální statistickou silou a nízkým operačním rizikem. Vyžaduje dostatečnou velikost vzorku a rychlou zpětnovazební smyčku.
- Víceruký bandita: používat, když je zpětná vazba rychlá (během hodin/dnů), náklady na průzkum jsou vysoké a dáváte přednost maximalizaci konverzí během testu. Není ideální pro malé efekty s pomalou zpětnou vazbou.
- Bayesian A/B: používejte flexibilní pravidla zastavení, kdykoli budete chtít, interpretovat pravděpodobnosti přímo nebo mít předchozí informace z experimentů precedenty. Ideální pro týmy, které považují p-hodnoty za matoucí.
Závěry a další kroky
A/B testování pro modely ML je mnohem více než pouhé rozdělování provozu. Vyžaduje rigorózní statistický návrh před každou implementací, volbou správná strategie založená na kontextu (stín, kanár, 50/50, bandita), průběžné sledování během testu a správná statistická analýza na konci.
Rozdíl mezi týmem, který dělá A/B testování správně, a týmem, který to dělá špatně není to ve složitosti kódu, ale v disciplíně procesu: definování nejprve hypotézy, na data se nedívejte, vše správně analyzujte až poté. Se zásobníkem open source popsaným v této příručce (FastAPI, Prometheus, Grafana, scipy, numpy), můžete implementovat systém produkční úrovně s minimálním rozpočtem.
Přirozeným dalším krokem je integrace A/B testování s ML governance: každé rozhodnutí prosadit model do výroby musí být zdokumentováno, auditovatelné a v souladu s etickými a regulačními standardy. To uvidíme v článku další o ML Governance.
Série MLOps pokračuje
- Předchozí článek: Škálování ML na Kubernetes - zorganizovat nasazení s KubeFlow a Seldon Core
- Další článek: ML Governance: Compliance, audit, etika - AI Act EU, vysvětlitelnost a spravedlnost
- Související: Detekce driftu modelu a automatické přeškolení - detekovat a reagovat na degradaci modelu
- Související: Servírovací modely: FastAPI + Uvicorn ve výrobě - vytvářet škálovatelná inferenční API
- Související série: Pokročilé hluboké učení - A/B testování pro komplexní neurální modely