/
maratgaliulin
/
landcode_classifier
Обзор
Документация
Войти
/
maratgaliulin
/
landcode_classifier
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
methods/classes/ABTestRunner.py
454 строки
21 KB
maratgaliulin
.
17 апр 2026, 07:44
17 апр 2026, 07:44
1e9a134
Код
Авторство
О чём код?
# methods/classes/ABTestRunner.py import torch import pandas as pd import numpy as np from tqdm.auto import tqdm from typing import Dict, List, Tuple, Optional, Callable from dataclasses import dataclass, field from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score, classification_report @dataclass class ABTestResult: """Результаты A/B-тестирования""" # Основные метрики accuracy_model_a: float = 0.0 accuracy_model_b: float = 0.0 accuracy_hybrid: float = 0.0 # Precision/Recall/F1 для разных подходов precision_model_a: float = 0.0 recall_model_a: float = 0.0 f1_model_a: float = 0.0 precision_model_b: float = 0.0 recall_model_b: float = 0.0 f1_model_b: float = 0.0 precision_hybrid: float = 0.0 recall_hybrid: float = 0.0 f1_hybrid: float = 0.0 # Статистика по заменам total_samples: int = 0 samples_handled_by_model_a: int = 0 samples_handled_by_model_b: int = 0 samples_switched_to_b: int = 0 samples_switched_correctly: int = 0 samples_switched_incorrectly: int = 0 # Детальные предсказания predictions_df: Optional[pd.DataFrame] = None # Метрики по классам (редкие vs частые) rare_classes_accuracy_a: float = 0.0 rare_classes_accuracy_b: float = 0.0 rare_classes_accuracy_hybrid: float = 0.0 frequent_classes_accuracy_a: float = 0.0 frequent_classes_accuracy_b: float = 0.0 frequent_classes_accuracy_hybrid: float = 0.0 def to_dict(self) -> Dict: """Преобразование в словарь""" return { 'accuracy_model_a': self.accuracy_model_a, 'accuracy_model_b': self.accuracy_model_b, 'accuracy_hybrid': self.accuracy_hybrid, 'improvement': self.accuracy_hybrid - self.accuracy_model_a, 'precision_model_a': self.precision_model_a, 'recall_model_a': self.recall_model_a, 'f1_model_a': self.f1_model_a, 'precision_model_b': self.precision_model_b, 'recall_model_b': self.recall_model_b, 'f1_model_b': self.f1_model_b, 'precision_hybrid': self.precision_hybrid, 'recall_hybrid': self.recall_hybrid, 'f1_hybrid': self.f1_hybrid, 'total_samples': self.total_samples, 'samples_handled_by_model_a': self.samples_handled_by_model_a, 'samples_handled_by_model_b': self.samples_handled_by_model_b, 'samples_switched_to_b': self.samples_switched_to_b, 'samples_switched_correctly': self.samples_switched_correctly, 'samples_switched_incorrectly': self.samples_switched_incorrectly, 'rare_classes_accuracy_a': self.rare_classes_accuracy_a, 'rare_classes_accuracy_b': self.rare_classes_accuracy_b, 'rare_classes_accuracy_hybrid': self.rare_classes_accuracy_hybrid, 'frequent_classes_accuracy_a': self.frequent_classes_accuracy_a, 'frequent_classes_accuracy_b': self.frequent_classes_accuracy_b, 'frequent_classes_accuracy_hybrid': self.frequent_classes_accuracy_hybrid, } def summary(self) -> str: """Форматированное текстовое резюме""" lines = [] lines.append("="*60) lines.append("РЕЗУЛЬТАТЫ A/B-ТЕСТИРОВАНИЯ") lines.append("="*60) lines.append(f"\n📊 ОБЩАЯ ТОЧНОСТЬ:") lines.append(f" • Модель A: {self.accuracy_model_a:.4f} ({self.accuracy_model_a*100:.2f}%)") lines.append(f" • Модель B: {self.accuracy_model_b:.4f} ({self.accuracy_model_b*100:.2f}%)") lines.append(f" • Гибридный подход: {self.accuracy_hybrid:.4f} ({self.accuracy_hybrid*100:.2f}%)") lines.append(f" • Улучшение: +{(self.accuracy_hybrid - self.accuracy_model_a)*100:.2f} п.п.") lines.append(f"\n🎯 PRECISION/RECALL/F1:") lines.append(f" • Модель A: P={self.precision_model_a:.4f}, R={self.recall_model_a:.4f}, F1={self.f1_model_a:.4f}") lines.append(f" • Модель B: P={self.precision_model_b:.4f}, R={self.recall_model_b:.4f}, F1={self.f1_model_b:.4f}") lines.append(f" • Гибрид: P={self.precision_hybrid:.4f}, R={self.recall_hybrid:.4f}, F1={self.f1_hybrid:.4f}") lines.append(f"\n🔄 СТАТИСТИКА ЗАМЕН:") lines.append(f" • Всего заменено на модель B: {self.samples_switched_to_b} ({self.samples_switched_to_b/self.total_samples*100:.1f}%)") lines.append(f" • Из них правильных: {self.samples_switched_correctly} ({self.samples_switched_correctly/self.samples_switched_to_b*100:.1f}% если были замены)") lines.append(f" • Из них неправильных: {self.samples_switched_incorrectly}") lines.append(f"\n📈 ТОЧНОСТЬ НА РАЗНЫХ ТИПАХ КЛАССОВ:") lines.append(f" • Редкие классы (не 13:021, 02:010):") lines.append(f" - Модель A: {self.rare_classes_accuracy_a:.4f} ({self.rare_classes_accuracy_a*100:.2f}%)") lines.append(f" - Модель B: {self.rare_classes_accuracy_b:.4f} ({self.rare_classes_accuracy_b*100:.2f}%)") lines.append(f" - Гибрид: {self.rare_classes_accuracy_hybrid:.4f} ({self.rare_classes_accuracy_hybrid*100:.2f}%)") lines.append(f" • Частые классы (13:021, 02:010):") lines.append(f" - Модель A: {self.frequent_classes_accuracy_a:.4f} ({self.frequent_classes_accuracy_a*100:.2f}%)") lines.append(f" - Модель B: {self.frequent_classes_accuracy_b:.4f} ({self.frequent_classes_accuracy_b*100:.2f}%)") lines.append(f" - Гибрид: {self.frequent_classes_accuracy_hybrid:.4f} ({self.frequent_classes_accuracy_hybrid*100:.2f}%)") return "\n".join(lines) class ABTestRunner: """ A/B тестирование для сравнения двух моделей классификации """ def __init__(self, model_a, model_b, tokenizer_a, tokenizer_b, device=None): """ Инициализация A/B тестера Parameters: ----------- model_a : torch.nn.Module Первая модель (исходная) model_b : torch.nn.Module Вторая модель (на сбалансированном датасете) tokenizer_a : transformers.PreTrainedTokenizer Токенизатор для модели A tokenizer_b : transformers.PreTrainedTokenizer Токенизатор для модели B device : torch.device Устройство для инференса """ self.model_a = model_a self.model_b = model_b self.tokenizer_a = tokenizer_a self.tokenizer_b = tokenizer_b if device is None: self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') else: self.device = device self.model_a.to(self.device) self.model_b.to(self.device) self.model_a.eval() self.model_b.eval() def _predict_single(self, text: str, model, tokenizer) -> Tuple[int, float]: """ Предсказание для одного текста Returns: -------- Tuple[int, float]: (predicted_label, confidence) """ encoding = tokenizer( text, truncation=True, padding='max_length', max_length=256, return_tensors='pt' ) input_ids = encoding['input_ids'].to(self.device) attention_mask = encoding['attention_mask'].to(self.device) with torch.no_grad(): outputs = model(input_ids, attention_mask) probabilities = torch.softmax(outputs, dim=1) confidence = probabilities[0].max().item() prediction = torch.argmax(outputs, dim=1).item() return prediction, confidence def run_ab_test(self, test_df: pd.DataFrame, text_column: str = 'combined_text', label_column: str = 'label', code_column: str = 'Код расчёта вида использования (ГБУ)', threshold: float = 0.4, frequent_codes: List[str] = None, show_progress: bool = True) -> ABTestResult: """ Запуск A/B тестирования Parameters: ----------- test_df : pd.DataFrame Тестовый датасет text_column : str Название колонки с текстом label_column : str Название колонки с метками code_column : str Название колонки с исходными кодами (для анализа редких/частых классов) threshold : float Порог уверенности для переключения на модель B frequent_codes : List[str] Список частых кодов для отдельного анализа show_progress : bool Показывать прогресс-бар Returns: -------- ABTestResult """ if frequent_codes is None: frequent_codes = ['13:021', '02:010'] results = [] # Прогресс-бар iterator = tqdm(test_df.iterrows(), total=len(test_df), desc="A/B Testing") if show_progress else test_df.iterrows() for idx, row in iterator: text = row[text_column] true_label = row[label_column] code = row[code_column] # Предсказание модели A pred_a, conf_a = self._predict_single(text, self.model_a, self.tokenizer_a) # Гибридное предсказание if conf_a >= threshold: final_pred = pred_a final_conf = conf_a source = 'model_a' pred_b = None conf_b = None else: # Предсказание модели B pred_b, conf_b = self._predict_single(text, self.model_b, self.tokenizer_b) final_pred = pred_b final_conf = conf_b source = 'model_b' # Определяем, была ли замена правильной is_correct_a = (pred_a == true_label) is_correct_final = (final_pred == true_label) results.append({ 'index': idx, 'true_label': true_label, 'code': code, 'pred_a': pred_a, 'conf_a': conf_a, 'pred_b': pred_b, 'conf_b': conf_b, 'final_pred': final_pred, 'final_conf': final_conf, 'source': source, 'is_correct_a': is_correct_a, 'is_correct_final': is_correct_final, }) # Создаём DataFrame с результатами results_df = pd.DataFrame(results) # Вычисляем метрики для модели A accuracy_a = (results_df['pred_a'] == results_df['true_label']).mean() precision_a = precision_score(results_df['true_label'], results_df['pred_a'], average='weighted', zero_division=0) recall_a = recall_score(results_df['true_label'], results_df['pred_a'], average='weighted', zero_division=0) f1_a = f1_score(results_df['true_label'], results_df['pred_a'], average='weighted', zero_division=0) # Метрики для модели B (только там, где она использовалась) mask_b = results_df['pred_b'].notna() if mask_b.any(): accuracy_b = (results_df.loc[mask_b, 'pred_b'] == results_df.loc[mask_b, 'true_label']).mean() precision_b = precision_score( results_df.loc[mask_b, 'true_label'], results_df.loc[mask_b, 'pred_b'], average='weighted', zero_division=0 ) recall_b = recall_score( results_df.loc[mask_b, 'true_label'], results_df.loc[mask_b, 'pred_b'], average='weighted', zero_division=0 ) f1_b = f1_score( results_df.loc[mask_b, 'true_label'], results_df.loc[mask_b, 'pred_b'], average='weighted', zero_division=0 ) else: accuracy_b = 0.0 precision_b = 0.0 recall_b = 0.0 f1_b = 0.0 # Метрики для гибридного подхода accuracy_hybrid = results_df['is_correct_final'].mean() precision_hybrid = precision_score(results_df['true_label'], results_df['final_pred'], average='weighted', zero_division=0) recall_hybrid = recall_score(results_df['true_label'], results_df['final_pred'], average='weighted', zero_division=0) f1_hybrid = f1_score(results_df['true_label'], results_df['final_pred'], average='weighted', zero_division=0) # Статистика по заменам total_samples = len(results_df) samples_handled_by_model_a = (results_df['source'] == 'model_a').sum() samples_handled_by_model_b = (results_df['source'] == 'model_b').sum() samples_switched_to_b = samples_handled_by_model_b switched_correct = results_df[(results_df['source'] == 'model_b') & (results_df['is_correct_final'])].shape[0] switched_incorrect = results_df[(results_df['source'] == 'model_b') & (~results_df['is_correct_final'])].shape[0] # Анализ по редким и частым классам rare_mask = ~test_df[code_column].isin(frequent_codes) frequent_mask = test_df[code_column].isin(frequent_codes) # Преобразуем маски в numpy массивы для безопасной индексации rare_mask_array = rare_mask.values if hasattr(rare_mask, 'values') else rare_mask frequent_mask_array = frequent_mask.values if hasattr(frequent_mask, 'values') else frequent_mask # Для редких классов if rare_mask_array.sum() > 0: rare_indices = results_df.index[rare_mask_array] rare_results = results_df.loc[rare_indices] rare_classes_accuracy_a = (rare_results['pred_a'] == rare_results['true_label']).mean() rare_classes_accuracy_hybrid = rare_results['is_correct_final'].mean() # Для модели B (только где есть предсказания) rare_mask_b = rare_mask_array & mask_b.values if hasattr(mask_b, 'values') else rare_mask_array & mask_b if rare_mask_b.sum() > 0: rare_b_indices = results_df.index[rare_mask_b] rare_b_results = results_df.loc[rare_b_indices] rare_classes_accuracy_b = (rare_b_results['pred_b'] == rare_b_results['true_label']).mean() else: rare_classes_accuracy_b = 0.0 else: rare_classes_accuracy_a = 0.0 rare_classes_accuracy_b = 0.0 rare_classes_accuracy_hybrid = 0.0 # Для частых классов if frequent_mask_array.sum() > 0: frequent_indices = results_df.index[frequent_mask_array] frequent_results = results_df.loc[frequent_indices] frequent_classes_accuracy_a = (frequent_results['pred_a'] == frequent_results['true_label']).mean() frequent_classes_accuracy_hybrid = frequent_results['is_correct_final'].mean() # Для модели B frequent_mask_b = frequent_mask_array & mask_b.values if hasattr(mask_b, 'values') else frequent_mask_array & mask_b if frequent_mask_b.sum() > 0: frequent_b_indices = results_df.index[frequent_mask_b] frequent_b_results = results_df.loc[frequent_b_indices] frequent_classes_accuracy_b = (frequent_b_results['pred_b'] == frequent_b_results['true_label']).mean() else: frequent_classes_accuracy_b = 0.0 else: frequent_classes_accuracy_a = 0.0 frequent_classes_accuracy_b = 0.0 frequent_classes_accuracy_hybrid = 0.0 # Формируем результат result = ABTestResult( accuracy_model_a=accuracy_a, accuracy_model_b=accuracy_b, accuracy_hybrid=accuracy_hybrid, precision_model_a=precision_a, recall_model_a=recall_a, f1_model_a=f1_a, precision_model_b=precision_b, recall_model_b=recall_b, f1_model_b=f1_b, precision_hybrid=precision_hybrid, recall_hybrid=recall_hybrid, f1_hybrid=f1_hybrid, total_samples=total_samples, samples_handled_by_model_a=samples_handled_by_model_a, samples_handled_by_model_b=samples_handled_by_model_b, samples_switched_to_b=samples_switched_to_b, samples_switched_correctly=switched_correct, samples_switched_incorrectly=switched_incorrect, predictions_df=results_df, rare_classes_accuracy_a=rare_classes_accuracy_a, rare_classes_accuracy_b=rare_classes_accuracy_b, rare_classes_accuracy_hybrid=rare_classes_accuracy_hybrid, frequent_classes_accuracy_a=frequent_classes_accuracy_a, frequent_classes_accuracy_b=frequent_classes_accuracy_b, frequent_classes_accuracy_hybrid=frequent_classes_accuracy_hybrid, ) return result def find_optimal_threshold(self, val_df: pd.DataFrame, text_column: str = 'combined_text', label_column: str = 'label', thresholds: List[float] = None, show_progress: bool = True) -> Dict: """ Поиск оптимального порога уверенности Parameters: ----------- val_df : pd.DataFrame Валидационный датасет text_column : str Название колонки с текстом label_column : str Название колонки с метками thresholds : List[float] Список порогов для тестирования show_progress : bool Показывать прогресс-бар Returns: -------- Dict: Результаты по каждому порогу """ if thresholds is None: thresholds = [0.3, 0.35, 0.4, 0.45, 0.5, 0.55, 0.6] results = {} for threshold in thresholds: print(f"\n📊 Тестирование порога: {threshold}") result = self.run_ab_test( test_df=val_df, text_column=text_column, label_column=label_column, threshold=threshold, show_progress=show_progress ) results[threshold] = result # Вывод сводной таблицы print("\n" + "="*80) print("СВОДНАЯ ТАБЛИЦА ПО ПОРОГАМ") print("="*80) print(f"{'Порог':<8} {'Acc A':<10} {'Acc B':<10} {'Acc Hybrid':<12} {'Замены':<10} {'Правильные замены':<18}") print("-"*80) for threshold, res in results.items(): switch_rate = res.samples_switched_to_b / res.total_samples * 100 correct_switch_rate = res.samples_switched_correctly / res.samples_switched_to_b * 100 if res.samples_switched_to_b > 0 else 0 print(f"{threshold:<8.2f} {res.accuracy_model_a*100:<10.2f}% {res.accuracy_model_b*100:<10.2f}% " f"{res.accuracy_hybrid*100:<12.2f}% {switch_rate:<10.1f}% {correct_switch_rate:<18.1f}%") return results