318 lines
11 KiB
Python
318 lines
11 KiB
Python
"""Векторизация текста и латентно-семантический анализ (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)
|