"""Векторизация текста и латентно-семантический анализ (LSA). LSA реализован как последовательность: TF-IDF -> усечённое сингулярное разложение (TruncatedSVD) -> L2-нормализация. Все параметры, включая число компонент, должны настраиваться внутри кросс-валидации, иначе возникает утечка информации из тестовой выборки. """ from __future__ import annotations from typing import Any import numpy as np import pandas as pd from sklearn.base import BaseEstimator, TransformerMixin from sklearn.decomposition import TruncatedSVD from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.preprocessing import normalize LSA_VARIANTS = ("word", "word2", "content", "fw", "char", "pos") _DEFAULT_CONFIGS: dict[str, dict[str, Any]] = { "word": { "column": "text", "analyzer": "word", "ngram_range": (1, 1), "token_pattern": r"(?u)\b\w[\w']*\b", "lowercase": True, "min_df": 3, "max_df": 0.9, "max_features": 5000, }, "word2": { "column": "text", "analyzer": "word", "ngram_range": (1, 2), "token_pattern": r"(?u)\b\w[\w']*\b", "lowercase": True, "min_df": 3, "max_df": 0.9, "max_features": 8000, }, "content": { "column": "content_text", "analyzer": "word", "ngram_range": (1, 2), "token_pattern": r"\S+", "lowercase": True, "min_df": 3, "max_df": 0.9, "max_features": 5000, }, "fw": { "column": "fw_text", "analyzer": "word", "ngram_range": (1, 2), "token_pattern": r"\S+", "lowercase": True, "min_df": 3, "max_df": 0.95, "max_features": 2000, }, "char": { "column": "text", "analyzer": "char_wb", "ngram_range": (2, 5), "token_pattern": None, "lowercase": True, "min_df": 3, "max_df": 0.95, "max_features": 5000, }, "pos": { "column": "pos_text", "analyzer": "word", "ngram_range": (1, 4), "token_pattern": r"\S+", "lowercase": False, "min_df": 3, "max_df": 0.95, "max_features": 3000, }, } def _as_texts(X: Any, column: str) -> list[str]: if isinstance(X, pd.DataFrame): if column not in X.columns: raise KeyError(f"Нет колонки {column!r} для векторизации.") return X[column].astype(str).tolist() return [str(value) for value in X] class ColumnTfidfVectorizer(BaseEstimator, TransformerMixin): """TF-IDF по одной колонке таблицы признаков.""" def __init__( self, column: str = "text", analyzer: str = "word", ngram_range: tuple[int, int] = (1, 1), token_pattern: str | None = r"(?u)\b\w[\w']*\b", lowercase: bool = True, min_df: int | float = 1, max_df: int | float = 1.0, max_features: int | None = None, sublinear_tf: bool = True, stop_words: str | None = None, ) -> None: self.column = column self.analyzer = analyzer self.ngram_range = ngram_range self.token_pattern = token_pattern self.lowercase = lowercase self.min_df = min_df self.max_df = max_df self.max_features = max_features self.sublinear_tf = sublinear_tf self.stop_words = stop_words def _make_vectorizer(self) -> TfidfVectorizer: kwargs: dict[str, Any] = { "analyzer": self.analyzer, "ngram_range": self.ngram_range, "lowercase": self.lowercase, "min_df": self.min_df, "max_df": self.max_df, "max_features": self.max_features, "sublinear_tf": self.sublinear_tf, "stop_words": self.stop_words, } if self.analyzer == "word" and self.token_pattern is not None: kwargs["token_pattern"] = self.token_pattern return TfidfVectorizer(**kwargs) def fit(self, X: Any, y=None) -> "ColumnTfidfVectorizer": texts = _as_texts(X, self.column) if not texts: raise ValueError("Пустая выборка для TF-IDF.") self.vectorizer_ = self._make_vectorizer() self.vectorizer_.fit(texts) self.n_features_in_ = len(self.vectorizer_.get_feature_names_out()) return self def transform(self, X: Any) -> Any: if not hasattr(self, "vectorizer_"): raise RuntimeError("Сначала вызовите fit.") return self.vectorizer_.transform(_as_texts(X, self.column)) def get_feature_names_out(self, input_features=None) -> np.ndarray: if not hasattr(self, "vectorizer_"): raise RuntimeError("Сначала вызовите fit.") return self.vectorizer_.get_feature_names_out() class LSAVectorizer(BaseEstimator, TransformerMixin): """TF-IDF + TruncatedSVD + L2-нормализация.""" def __init__( self, column: str = "text", analyzer: str = "word", ngram_range: tuple[int, int] = (1, 1), token_pattern: str | None = r"(?u)\b\w[\w']*\b", lowercase: bool = True, min_df: int | float = 3, max_df: int | float = 0.9, max_features: int | None = 5000, n_components: int = 100, sublinear_tf: bool = True, stop_words: str | None = None, normalize_rows: bool = True, random_state: int = 42, ) -> None: self.column = column self.analyzer = analyzer self.ngram_range = ngram_range self.token_pattern = token_pattern self.lowercase = lowercase self.min_df = min_df self.max_df = max_df self.max_features = max_features self.n_components = n_components self.sublinear_tf = sublinear_tf self.stop_words = stop_words self.normalize_rows = normalize_rows self.random_state = random_state def fit(self, X: Any, y=None) -> "LSAVectorizer": self.tfidf_ = ColumnTfidfVectorizer( column=self.column, analyzer=self.analyzer, ngram_range=self.ngram_range, token_pattern=self.token_pattern, lowercase=self.lowercase, min_df=self.min_df, max_df=self.max_df, max_features=self.max_features, sublinear_tf=self.sublinear_tf, stop_words=self.stop_words, ) matrix = self.tfidf_.fit_transform(X) max_components = min(matrix.shape[0] - 1, matrix.shape[1] - 1) if max_components < 1: raise ValueError( "Слишком маленькая матрица для LSA: нужно хотя бы два документа " "и два признака." ) self.n_components_ = int(min(self.n_components, max_components)) self.svd_ = TruncatedSVD( n_components=self.n_components_, algorithm="randomized", n_iter=10, random_state=self.random_state, ) self.svd_.fit(matrix) self.explained_variance_ratio_ = self.svd_.explained_variance_ratio_ self.cumulative_explained_variance_ = np.cumsum(self.explained_variance_ratio_) self.n_features_in_ = matrix.shape[1] return self def transform(self, X: Any) -> np.ndarray: if not hasattr(self, "svd_"): raise RuntimeError("Сначала вызовите fit.") matrix = self.tfidf_.transform(X) result = self.svd_.transform(matrix) if self.normalize_rows: result = normalize(result, norm="l2", copy=False) return np.asarray(result) def fit_transform(self, X: Any, y=None, **fit_params) -> np.ndarray: return self.fit(X, y).transform(X) def get_feature_names_out(self, input_features=None) -> np.ndarray: return np.array([f"lsa_{index}" for index in range(self.n_components_)], dtype=object) def top_terms(self, n: int = 10) -> list[list[tuple[str, float]]]: """Возвращает топ-``n`` терминов для каждой компоненты. Знак компоненты не интерпретируется: он произволен. Для отчёта важнее набор терминов и их относительный вклад. """ if not hasattr(self, "svd_"): raise RuntimeError("Сначала вызовите fit.") terms = self.tfidf_.get_feature_names_out() result: list[list[tuple[str, float]]] = [] for component in self.svd_.components_: order = np.argsort(component)[::-1][:n] result.append([(str(terms[index]), float(component[index])) for index in order]) return result def reconstruction_error(self, X: Any) -> float: """Относительная ошибка восстановления TF-IDF матрицы.""" matrix = self.tfidf_.transform(X) reconstructed = self.svd_.inverse_transform(self.svd_.transform(matrix)) denominator = np.linalg.norm(matrix.toarray()) if denominator == 0: return 0.0 return float(np.linalg.norm(matrix.toarray() - reconstructed) / denominator) def make_lsa_vectorizer( variant: str = "word", n_components: int = 100, **overrides: Any, ) -> LSAVectorizer: """Создаёт LSA-векторизатор одного из заранее определённых вариантов.""" if variant not in _DEFAULT_CONFIGS: raise ValueError( f"Неизвестный вариант LSA: {variant}. Доступны: {LSA_VARIANTS}" ) config = dict(_DEFAULT_CONFIGS[variant]) config.update(overrides) return LSAVectorizer(n_components=n_components, **config) def make_text_vectorizer(kind: str = "word", **overrides: Any) -> ColumnTfidfVectorizer: """Создаёт TF-IDF-векторизатор без понижения размерности.""" if kind not in _DEFAULT_CONFIGS: raise ValueError(f"Неизвестный вариант TF-IDF: {kind}") config = dict(_DEFAULT_CONFIGS[kind]) config.update(overrides) return ColumnTfidfVectorizer(**config) def make_pos_ngram_vectorizer(**overrides: Any) -> ColumnTfidfVectorizer: """TF-IDF по последовательностям Universal POS.""" config = { "column": "pos_text", "analyzer": "word", "ngram_range": (1, 3), "token_pattern": r"\S+", "lowercase": False, "min_df": 3, "max_df": 0.95, "max_features": 500, "sublinear_tf": True, } config.update(overrides) return ColumnTfidfVectorizer(**config) def make_char_ngram_vectorizer(**overrides: Any) -> ColumnTfidfVectorizer: """TF-IDF по символьным n-граммам.""" config = { "column": "text", "analyzer": "char_wb", "ngram_range": (2, 5), "token_pattern": None, "lowercase": True, "min_df": 3, "max_df": 0.95, "max_features": 5000, "sublinear_tf": True, } config.update(overrides) return ColumnTfidfVectorizer(**config)