"""Верификация «тот же персонаж / другой персонаж» по центроидам классов.""" from __future__ import annotations from dataclasses import dataclass from typing import Any, Sequence import numpy as np import pandas as pd from scipy.sparse import issparse from sklearn.metrics import roc_auc_score, roc_curve from sklearn.metrics.pairwise import cosine_similarity from sklearn.model_selection import StratifiedKFold from sklearn.base import clone from .pipelines import FeatureConfig, build_feature_union def _to_dense(matrix: Any) -> np.ndarray: return matrix.toarray() if issparse(matrix) else np.asarray(matrix) @dataclass class VerificationResult: fold_metrics: pd.DataFrame scores: pd.DataFrame summary: dict[str, float] def equal_error_rate(y_true: Sequence[int], scores: Sequence[float]) -> float: """Равная ошибка: точка, где FPR и FNR совпадают.""" fpr, tpr, _ = roc_curve(y_true, scores) fnr = 1.0 - tpr index = int(np.argmin(np.abs(fpr - fnr))) return float((fpr[index] + fnr[index]) / 2.0) def tar_at_far( y_true: Sequence[int], scores: Sequence[float], far: float = 0.01 ) -> float: """Доля принятых «своих» при заданной доле ложных принятий.""" fpr, tpr, _ = roc_curve(y_true, scores) return float(np.interp(far, fpr, tpr)) def _centroids(matrix: np.ndarray, labels: np.ndarray) -> tuple[np.ndarray, np.ndarray]: classes = np.unique(labels) centroids = np.vstack([matrix[labels == label].mean(axis=0) for label in classes]) return classes, centroids def run_verification( X: pd.DataFrame, y: Sequence, feature_config: FeatureConfig, n_splits: int = 5, seed: int = 42, verbose: bool = True, ) -> VerificationResult: """Строит оценки «свой/чужой» для каждой реплики и каждого персонажа. На каждом фолде центроиды персонажей считаются только по обучающим репликам. Для тестовой реплики косинусная близость к центроиду своего персонажа образует положительный класс, к остальным — отрицательный. """ y = np.asarray(y) splitter = StratifiedKFold(n_splits=n_splits, shuffle=True, random_state=seed) fold_rows: list[dict[str, Any]] = [] score_rows: list[dict[str, Any]] = [] for fold, (train_idx, test_idx) in enumerate(splitter.split(X, y), start=1): features = build_feature_union(feature_config) features = clone(features) train_matrix = _to_dense(features.fit_transform(X.iloc[train_idx])) test_matrix = _to_dense(features.transform(X.iloc[test_idx])) train_labels = y[train_idx] classes, centroids = _centroids(train_matrix, train_labels) similarities = cosine_similarity(test_matrix, centroids) binary_true: list[int] = [] binary_scores: list[float] = [] top1_correct = 0 for row, sample_index in enumerate(test_idx): true_label = y[sample_index] for class_index, class_label in enumerate(classes): score = float(similarities[row, class_index]) is_same = int(class_label == true_label) binary_true.append(is_same) binary_scores.append(score) score_rows.append( { "fold": fold, "line_id": int(X.iloc[sample_index]["line_id"]) if "line_id" in X.columns else int(sample_index), "true_character": true_label, "candidate_character": class_label, "score": score, "is_same": is_same, } ) predicted_label = classes[int(np.argmax(similarities[row]))] top1_correct += int(predicted_label == true_label) binary_true_array = np.asarray(binary_true) binary_scores_array = np.asarray(binary_scores) auc = float(roc_auc_score(binary_true_array, binary_scores_array)) eer = equal_error_rate(binary_true_array, binary_scores_array) fold_rows.append( { "fold": fold, "roc_auc": auc, "eer": eer, "tar_at_far_1pct": tar_at_far(binary_true_array, binary_scores_array, 0.01), "top1_accuracy": top1_correct / max(len(test_idx), 1), "n_test": int(len(test_idx)), } ) if verbose: print( f"[verification] фолд {fold}/{n_splits}: " f"ROC-AUC={auc:.4f}, EER={eer:.4f}, top1={top1_correct / max(len(test_idx), 1):.4f}" ) fold_metrics = pd.DataFrame(fold_rows) scores = pd.DataFrame(score_rows) summary = { column: float(fold_metrics[column].mean()) for column in ("roc_auc", "eer", "tar_at_far_1pct", "top1_accuracy") } return VerificationResult(fold_metrics=fold_metrics, scores=scores, summary=summary)