130 lines
5.2 KiB
Python
130 lines
5.2 KiB
Python
"""Верификация «тот же персонаж / другой персонаж» по центроидам классов."""
|
||
|
||
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)
|