A/B Тестирование Моделей Machine Learning: ML Эксперименты

class МЛЭксперимент:
    def __init__(self, название, гипотеза, критерии_успеха):
        self.название = название
        self.гипотеза = гипотеза
        self.критерии_успеха = критерии_успеха
        self.варианты = {}

эксперимент = МЛЭксперимент(
    название="модель_ранжирования_v2",
    гипотеза="Новая transformer модель увеличит CTR на 5%",
    критерии_успеха={
        'увеличение_ctr': 0.05,
        'латентность_p95': 200,  # мс
        'минимальная_значимость': 0.95
    }
)

эксперимент.добавить_вариант('контроль', модель_v1, выделение_трафика=0.5)
эксперимент.добавить_вариант('лечение', модель_v2, выделение_трафика=0.5)

class РазделительТрафика:
    def назначить_вариант(self, user_id):
        """Назначение на основе консистентного хеша"""
        значение_хеша = hashlib.md5(
            f"{self.эксперимент.название}:{user_id}".encode()
        ).hexdigest()

        хеш_инт = int(значение_хеша, 16)
        порог = хеш_инт % 100

        накопительный = 0
        for название_варианта, вариант in self.эксперимент.варианты.items():
            накопительный += вариант['выделение_трафика'] * 100
            if порог < накопительный:
                return название_варианта, вариант['модель']

from scipy import stats

class ТестерЗначимости:
    def __init__(self, alpha=0.05):
        self.alpha = alpha

    def t_тест(self, данные_контроль, данные_лечение):
        """T-тест двух выборок для непрерывных метрик"""
        t_stat, p_значение = stats.ttest_ind(данные_контроль, данные_лечение)

        return {
            't_статистика': t_stat,
            'p_значение': p_значение,
            'значимо': p_значение < self.alpha,
            'среднее_контроль': np.mean(данные_контроль),
            'среднее_лечение': np.mean(данные_лечение),
            'относительный_прирост': (np.mean(данные_лечение) - np.mean(данные_контроль)) / np.mean(данные_контроль)
        }

    def тест_хи_квадрат(self, конверсии_контроль, всего_контроль, конверсии_лечение, всего_лечение):
        """Тест хи-квадрат для бинарных метрик"""
        таблица_сопряженности = np.array([
            [конверсии_контроль, всего_контроль - конверсии_контроль],
            [конверсии_лечение, всего_лечение - конверсии_лечение]
        ])

        chi2, p_значение, dof, expected = stats.chi2_contingency(таблица_сопряженности)

        ставка_контроль = конверсии_контроль / всего_контроль
        ставка_лечение = конверсии_лечение / всего_лечение

        return {
            'chi2_статистика': chi2,
            'p_значение': p_значение,
            'значимо': p_значение < self.alpha,
            'ставка_контроль': ставка_контроль,
            'ставка_лечение': ставка_лечение,
            'относительный_прирост': (ставка_лечение - ставка_контроль) / ставка_контроль
        }

class ОценщикОффлайн:
    def валидация_holdout(self, старая_модель, новая_модель):
        """Оценить на отложенных данных"""
        X_test, y_test = self.данные['X_test'], self.данные['y_test']

        старые_предсказания = старая_модель.predict(X_test)
        новые_предсказания = новая_модель.predict(X_test)

        return {
            'auc_старой_модели': roc_auc_score(y_test, старые_предсказания),
            'auc_новой_модели': roc_auc_score(y_test, новые_предсказания),
            'улучшение_auc': roc_auc_score(y_test, новые_предсказания) - roc_auc_score(y_test, старые_предсказания)
        }

class ОценщикОнлайн:
    def мониторить_реальное_время(self):
        """Мониторинг в реальном времени с алертами"""
        сводка = self.рассчитать_текущую_производительность()

        алерты = []

        if сводка['лечение']['процент_ошибок'] > сводка['контроль']['процент_ошибок'] * 1.5:
            алерты.append({
                'серьезность': 'КРИТИЧЕСКАЯ',
                'сообщение': f"Процент ошибок лечения на 50% выше"
            })

        return {'сводка': сводка, 'алерты': алерты}

    def проверить_ограждения(self):
        """Убедиться что критические метрики не деградируют"""
        ограждения = {
            'увеличение_процента_ошибок': 0.10,
            'увеличение_латентности_p95': 0.20,
            'уменьшение_дохода': 0.05
        }

        нарушения = []
        return {'пройдено': len(нарушения) == 0, 'нарушения': нарушения}

class ThompsonSampling:
    """Адаптивное выделение трафика на основе производительности"""
    def __init__(self, варианты):
        self.варианты = {
            название: {'alpha': 1, 'beta': 1}
            for название in варианты
        }

    def выбрать_вариант(self):
        """Thompson sampling: выборка из апостериорных распределений"""
        образцы = {
            название: np.random.beta(параметры['alpha'], параметры['beta'])
            for название, параметры in self.варианты.items()
        }

        return max(образцы, key=образцы.get)

    def обновить(self, название_варианта, награда):
        """Обновить апостериорное на основе наблюдаемой награды"""
        if награда > 0:
            self.варианты[название_варианта]['alpha'] += 1
        else:
            self.варианты[название_варианта]['beta'] += 1

class ПостепенноеРазвертывание:
    def __init__(self, эксперимент):
        self.эксперимент = эксперимент
        self.текущее_выделение = 0.05  # Начать с 5%

    def должен_увеличить_трафик(self, часов_работы, метрики):
        """Решить стоит ли увеличивать трафик к лечению"""
        if часов_работы < 24:
            return False

        if not метрики['ограждения_проходят']:
            return False

        if метрики['статистическая_значимость'] and метрики['положительный_прирост']:
            return True

        return False

# Этапы развертывания
# Этап 1: 5% в течение 24-48 часов
# Этап 2: 20% в течение 48 часов
# Этап 3: 50% в течение 48 часов
# Этап 4: 100% (полное развертывание)