A/B Testing para Modelos de Machine Learning: Experimentación ML

class ExperimentoML:
    def __init__(self, nombre, hipotesis, criterios_exito):
        self.nombre = nombre
        self.hipotesis = hipotesis
        self.criterios_exito = criterios_exito
        self.variantes = {}

experimento = ExperimentoML(
 (como se discute en [AI-powered Test Generation: The Future Is Already Here](/blog/ai-powered-test-generation))    nombre="modelo_ranking_v2",
    hipotesis="Nuevo modelo transformer aumentará CTR en 5%",
    criterios_exito={
        'aumento_ctr': 0.05,
        'latencia_p95': 200,  # ms
        'significancia_minima': 0.95
    }
)

experimento.agregar_variante('control', modelo_v1, asignacion_trafico=0.5)
experimento.agregar_variante('tratamiento', modelo_v2, asignacion_trafico=0.5)

class DivisorTrafico:
    def asignar_variante(self, user_id):
        """Asignación basada en hash consistente"""
        valor_hash = hashlib.md5(
            f"{self.experimento.nombre}:{user_id}".encode()
        ).hexdigest()

        hash_int = int(valor_hash, 16)
        umbral = hash_int % 100

        acumulativo = 0
        for nombre_variante, variante in self.experimento.variantes.items():
            acumulativo += variante['asignacion_trafico'] * 100
            if umbral < acumulativo:
                return nombre_variante, variante['modelo']

from scipy import stats

class ProbadorSignificancia:
    def __init__(self, alpha=0.05):
        self.alpha = alpha

    def t_test(self, datos_control, datos_tratamiento):
        """Prueba t de dos muestras para métricas continuas"""
        t_stat, p_value = stats.ttest_ind(datos_control, datos_tratamiento)

        return {
            't_statistic': t_stat,
            'p_value': p_value,
            'es_significativo': p_value < self.alpha,
            'media_control': np.mean(datos_control),
            'media_tratamiento': np.mean(datos_tratamiento),
            'lift_relativo': (np.mean(datos_tratamiento) - np.mean(datos_control)) / np.mean(datos_control)
        }

    def prueba_chi_cuadrado(self, conversiones_control, total_control, conversiones_tratamiento, total_tratamiento):
        """Prueba chi-cuadrado para métricas binarias"""
        tabla_contingencia = np.array([
            [conversiones_control, total_control - conversiones_control],
            [conversiones_tratamiento, total_tratamiento - conversiones_tratamiento]
        ])

        chi2, p_value, dof, expected = stats.chi2_contingency(tabla_contingencia)

        tasa_control = conversiones_control / total_control
        tasa_tratamiento = conversiones_tratamiento / total_tratamiento

        return {
            'chi2_statistic': chi2,
            'p_value': p_value,
            'es_significativo': p_value < self.alpha,
            'tasa_control': tasa_control,
            'tasa_tratamiento': tasa_tratamiento,
            'lift_relativo': (tasa_tratamiento - tasa_control) / tasa_control
        }

class EvaluadorOffline:
    def validacion_holdout(self, modelo_viejo, modelo_nuevo):
        """Evaluar en datos retenidos"""
        X_test, y_test = self.datos['X_test'], self.datos['y_test']

        predicciones_viejas = modelo_viejo.predict(X_test)
        predicciones_nuevas = modelo_nuevo.predict(X_test)

        return {
            'auc_modelo_viejo': roc_auc_score(y_test, predicciones_viejas),
            'auc_modelo_nuevo': roc_auc_score(y_test, predicciones_nuevas),
            'mejora_auc': roc_auc_score(y_test, predicciones_nuevas) - roc_auc_score(y_test, predicciones_viejas)
        }

class EvaluadorOnline:
    def monitorear_tiempo_real(self):
        """Monitoreo en tiempo real con alertas"""
        resumen = self.calcular_rendimiento_actual()

        alertas = []

        if resumen['tratamiento']['tasa_error'] > resumen['control']['tasa_error'] * 1.5:
            alertas.append({
                'severidad': 'CRITICA',
                'mensaje': f"Tasa de error de tratamiento 50% mayor"
            })

        return {'resumen': resumen, 'alertas': alertas}

    def verificar_barandillas(self):
        """Asegurar que métricas críticas no se degraden"""
        barandillas = {
            'aumento_tasa_error': 0.10,
            'aumento_latencia_p95': 0.20,
            'disminucion_ingresos': 0.05
        }

        violaciones = []
        return {'aprobado': len(violaciones) == 0, 'violaciones': violaciones}

class ThompsonSampling:
    """Asignación adaptativa de tráfico basada en rendimiento"""
    def __init__(self, variantes):
        self.variantes = {
            nombre: {'alpha': 1, 'beta': 1}
            for nombre in variantes
        }

    def seleccionar_variante(self):
        """Thompson sampling: muestra de distribuciones posteriores"""
        muestras = {
            nombre: np.random.beta(params['alpha'], params['beta'])
            for nombre, params in self.variantes.items()
        }

        return max(muestras, key=muestras.get)

    def actualizar(self, nombre_variante, recompensa):
        """Actualizar posterior basado en recompensa observada"""
        if recompensa > 0:
            self.variantes[nombre_variante]['alpha'] += 1
        else:
            self.variantes[nombre_variante]['beta'] += 1

class DespliegueGradual:
    def __init__(self, experimento):
        self.experimento = experimento
        self.asignacion_actual = 0.05  # Empezar con 5%

    def deberia_aumentar_trafico(self, horas_corriendo, metricas):
        """Decidir si aumentar tráfico a tratamiento"""
        if horas_corriendo < 24:
            return False

        if not metricas['barandillas_aprobadas']:
            return False

        if metricas['significancia_estadistica'] and metricas['lift_positivo']:
            return True

        return False

# Etapas de despliegue
# Etapa 1: 5% durante 24-48 horas
# Etapa 2: 20% durante 48 horas
# Etapa 3: 50% durante 48 horas
# Etapa 4: 100% (despliegue completo)