389 lines
14 KiB
Python
389 lines
14 KiB
Python
"""Сборка признаковых пространств и классификаторов."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass, field
|
||
from typing import Any
|
||
|
||
from sklearn.dummy import DummyClassifier
|
||
from sklearn.ensemble import ExtraTreesClassifier, RandomForestClassifier
|
||
from sklearn.linear_model import LogisticRegression
|
||
from sklearn.model_selection import GridSearchCV
|
||
from sklearn.naive_bayes import MultinomialNB
|
||
from sklearn.neighbors import KNeighborsClassifier
|
||
from sklearn.pipeline import FeatureUnion, Pipeline
|
||
from sklearn.preprocessing import MinMaxScaler, Normalizer, StandardScaler
|
||
from sklearn.svm import LinearSVC
|
||
|
||
from .features import FEATURE_GROUPS, StylometricVectorizer
|
||
from .lsa import (
|
||
LSA_VARIANTS,
|
||
make_char_ngram_vectorizer,
|
||
make_lsa_vectorizer,
|
||
make_pos_ngram_vectorizer,
|
||
)
|
||
|
||
MODEL_NAMES = (
|
||
"dummy",
|
||
"multinomial_nb",
|
||
"logreg",
|
||
"linear_svc",
|
||
"random_forest",
|
||
"extra_trees",
|
||
"knn",
|
||
)
|
||
|
||
|
||
@dataclass
|
||
class FeatureConfig:
|
||
"""Описание признакового пространства."""
|
||
|
||
groups: tuple[str, ...] = field(default_factory=lambda: tuple(FEATURE_GROUPS))
|
||
use_char_ngrams: bool = True
|
||
use_pos_ngrams: bool = True
|
||
lsa_variants: tuple[str, ...] = ("word", "pos")
|
||
lsa_components: int = 100
|
||
stylo_scaler: str = "standard" # standard | minmax | none
|
||
stylo_normalize: bool = True
|
||
# Веса блоков в FeatureUnion: stylo, char, pos, lsa_word, lsa_pos и т.д.
|
||
# Пустой словарь означает равный вес блоков.
|
||
block_weights: dict[str, float] = field(default_factory=dict)
|
||
include_stylo: bool = True
|
||
include_lsa: bool = True
|
||
char_max_features: int = 5000
|
||
pos_max_features: int = 500
|
||
char_ngram_range: tuple[int, int] = (2, 5)
|
||
pos_ngram_range: tuple[int, int] = (1, 3)
|
||
min_df: int = 3
|
||
random_state: int = 42
|
||
|
||
|
||
def _stylo_branch(config: FeatureConfig) -> Pipeline:
|
||
if config.stylo_scaler == "standard":
|
||
scaler: Any = StandardScaler()
|
||
elif config.stylo_scaler == "minmax":
|
||
scaler = MinMaxScaler()
|
||
elif config.stylo_scaler == "none":
|
||
scaler = "passthrough"
|
||
else:
|
||
raise ValueError(f"Неизвестный масштабировщик: {config.stylo_scaler}")
|
||
steps: list[tuple[str, Any]] = [
|
||
("extract", StylometricVectorizer(groups=config.groups)),
|
||
("scale", scaler),
|
||
]
|
||
if config.stylo_normalize:
|
||
# L2-нормировка уравнивает вклад блоков разной размерности.
|
||
steps.append(("norm", Normalizer()))
|
||
return Pipeline(steps)
|
||
|
||
|
||
def build_feature_union(config: FeatureConfig) -> Any:
|
||
"""Собирает блоки признаков в единое пространство."""
|
||
branches: list[tuple[str, Any]] = []
|
||
|
||
if config.include_stylo:
|
||
branches.append(("stylo", _stylo_branch(config)))
|
||
|
||
if config.use_char_ngrams:
|
||
branches.append(
|
||
(
|
||
"char",
|
||
Pipeline(
|
||
[
|
||
(
|
||
"tfidf",
|
||
make_char_ngram_vectorizer(
|
||
ngram_range=config.char_ngram_range,
|
||
max_features=config.char_max_features,
|
||
min_df=config.min_df,
|
||
),
|
||
),
|
||
("norm", Normalizer()),
|
||
]
|
||
),
|
||
)
|
||
)
|
||
|
||
if config.use_pos_ngrams:
|
||
branches.append(
|
||
(
|
||
"pos",
|
||
Pipeline(
|
||
[
|
||
(
|
||
"tfidf",
|
||
make_pos_ngram_vectorizer(
|
||
ngram_range=config.pos_ngram_range,
|
||
max_features=config.pos_max_features,
|
||
min_df=config.min_df,
|
||
),
|
||
),
|
||
("norm", Normalizer()),
|
||
]
|
||
),
|
||
)
|
||
)
|
||
|
||
if config.include_lsa:
|
||
for variant in config.lsa_variants:
|
||
if variant not in LSA_VARIANTS:
|
||
raise ValueError(f"Неизвестный вариант LSA: {variant}")
|
||
branches.append(
|
||
(
|
||
f"lsa_{variant}",
|
||
Pipeline(
|
||
[
|
||
(
|
||
"lsa",
|
||
make_lsa_vectorizer(
|
||
variant=variant,
|
||
n_components=config.lsa_components,
|
||
min_df=config.min_df,
|
||
random_state=config.random_state,
|
||
),
|
||
),
|
||
("norm", Normalizer()),
|
||
]
|
||
),
|
||
)
|
||
)
|
||
|
||
if not branches:
|
||
raise ValueError("Не выбран ни один блок признаков.")
|
||
|
||
if len(branches) == 1:
|
||
return branches[0][1]
|
||
|
||
union_kwargs: dict[str, Any] = {}
|
||
if config.block_weights:
|
||
branch_names = {name for name, _ in branches}
|
||
unknown = set(config.block_weights) - branch_names
|
||
if unknown:
|
||
raise ValueError(
|
||
f"Неизвестные блоки в block_weights: {sorted(unknown)}. "
|
||
f"Доступны: {sorted(branch_names)}"
|
||
)
|
||
union_kwargs["transformer_weights"] = dict(config.block_weights)
|
||
return FeatureUnion(branches, **union_kwargs)
|
||
|
||
|
||
def make_classifier(name: str, **params: Any) -> Any:
|
||
"""Создаёт классификатор по короткому имени."""
|
||
if name == "dummy":
|
||
strategy = params.pop("strategy", "most_frequent")
|
||
return DummyClassifier(strategy=strategy)
|
||
if name == "multinomial_nb":
|
||
return MultinomialNB(
|
||
alpha=params.pop("alpha", 1.0),
|
||
fit_prior=params.pop("fit_prior", True),
|
||
)
|
||
if name == "logreg":
|
||
return LogisticRegression(
|
||
C=params.pop("C", 1.0),
|
||
solver=params.pop("solver", "lbfgs"),
|
||
max_iter=params.pop("max_iter", 5000),
|
||
class_weight=params.pop("class_weight", "balanced"),
|
||
)
|
||
if name == "linear_svc":
|
||
return LinearSVC(
|
||
C=params.pop("C", 1.0),
|
||
loss=params.pop("loss", "squared_hinge"),
|
||
tol=params.pop("tol", 1e-4),
|
||
class_weight=params.pop("class_weight", "balanced"),
|
||
max_iter=params.pop("max_iter", 20_000),
|
||
dual="auto",
|
||
)
|
||
if name == "random_forest":
|
||
return RandomForestClassifier(
|
||
n_estimators=params.pop("n_estimators", 400),
|
||
max_depth=params.pop("max_depth", None),
|
||
min_samples_leaf=params.pop("min_samples_leaf", 1),
|
||
min_samples_split=params.pop("min_samples_split", 2),
|
||
max_features=params.pop("max_features", "sqrt"),
|
||
class_weight=params.pop("class_weight", "balanced_subsample"),
|
||
n_jobs=-1,
|
||
random_state=params.pop("random_state", 42),
|
||
)
|
||
if name == "extra_trees":
|
||
return ExtraTreesClassifier(
|
||
n_estimators=params.pop("n_estimators", 400),
|
||
max_depth=params.pop("max_depth", None),
|
||
min_samples_leaf=params.pop("min_samples_leaf", 1),
|
||
min_samples_split=params.pop("min_samples_split", 2),
|
||
max_features=params.pop("max_features", "sqrt"),
|
||
class_weight=params.pop("class_weight", "balanced"),
|
||
n_jobs=-1,
|
||
random_state=params.pop("random_state", 42),
|
||
)
|
||
if name == "knn":
|
||
return KNeighborsClassifier(
|
||
n_neighbors=params.pop("n_neighbors", 5),
|
||
metric=params.pop("metric", "cosine"),
|
||
weights=params.pop("weights", "distance"),
|
||
n_jobs=-1,
|
||
)
|
||
raise ValueError(f"Неизвестный классификатор: {name}. Доступны: {MODEL_NAMES}")
|
||
|
||
|
||
def build_pipeline(
|
||
feature_config: FeatureConfig,
|
||
model_name: str,
|
||
**model_params: Any,
|
||
) -> Pipeline:
|
||
"""Собирает полный конвейер: признаки -> классификатор.
|
||
|
||
Для MultinomialNB исключаются LSA-компоненты (они могут быть
|
||
отрицательными) и используется MinMax-масштабирование, чтобы матрица
|
||
оставалась неотрицательной.
|
||
"""
|
||
if model_name == "dummy":
|
||
return Pipeline([("model", make_classifier(model_name, **model_params))])
|
||
|
||
if model_name == "multinomial_nb":
|
||
config = FeatureConfig(
|
||
groups=feature_config.groups,
|
||
use_char_ngrams=feature_config.use_char_ngrams,
|
||
use_pos_ngrams=feature_config.use_pos_ngrams,
|
||
lsa_variants=(),
|
||
lsa_components=feature_config.lsa_components,
|
||
stylo_scaler="minmax",
|
||
stylo_normalize=feature_config.stylo_normalize,
|
||
include_stylo=feature_config.include_stylo,
|
||
include_lsa=False,
|
||
char_max_features=feature_config.char_max_features,
|
||
pos_max_features=feature_config.pos_max_features,
|
||
char_ngram_range=feature_config.char_ngram_range,
|
||
pos_ngram_range=feature_config.pos_ngram_range,
|
||
min_df=feature_config.min_df,
|
||
random_state=feature_config.random_state,
|
||
)
|
||
features = build_feature_union(config)
|
||
return Pipeline([("features", features), ("model", make_classifier(model_name, **model_params))])
|
||
|
||
features = build_feature_union(feature_config)
|
||
return Pipeline([("features", features), ("model", make_classifier(model_name, **model_params))])
|
||
|
||
|
||
def preset_config(name: str = "stylometry_pos_lsa") -> FeatureConfig:
|
||
"""Готовые наборы признаков для типовых экспериментов."""
|
||
all_groups = tuple(FEATURE_GROUPS)
|
||
if name == "stylometry":
|
||
return FeatureConfig(
|
||
groups=all_groups,
|
||
use_char_ngrams=False,
|
||
use_pos_ngrams=False,
|
||
lsa_variants=(),
|
||
stylo_normalize=False,
|
||
include_lsa=False,
|
||
)
|
||
if name == "stylometry_pos":
|
||
return FeatureConfig(
|
||
groups=all_groups,
|
||
use_char_ngrams=True,
|
||
use_pos_ngrams=True,
|
||
lsa_variants=(),
|
||
include_lsa=False,
|
||
)
|
||
if name in {"stylometry_lsa", "stylometry_pos_lsa"}:
|
||
return FeatureConfig(
|
||
groups=all_groups,
|
||
use_char_ngrams=True,
|
||
use_pos_ngrams=True,
|
||
lsa_variants=("word", "pos"),
|
||
include_lsa=True,
|
||
)
|
||
if name == "all":
|
||
return FeatureConfig(
|
||
groups=all_groups,
|
||
use_char_ngrams=True,
|
||
use_pos_ngrams=True,
|
||
lsa_variants=("word", "word2", "content", "fw", "char", "pos"),
|
||
include_lsa=True,
|
||
)
|
||
if name == "char":
|
||
return FeatureConfig(
|
||
groups=(),
|
||
use_char_ngrams=True,
|
||
use_pos_ngrams=False,
|
||
lsa_variants=(),
|
||
include_stylo=False,
|
||
include_lsa=False,
|
||
)
|
||
if name == "pos":
|
||
return FeatureConfig(
|
||
groups=(),
|
||
use_char_ngrams=False,
|
||
use_pos_ngrams=True,
|
||
lsa_variants=(),
|
||
include_stylo=False,
|
||
include_lsa=False,
|
||
)
|
||
if name == "fw":
|
||
return FeatureConfig(
|
||
groups=("function_words", "pronouns"),
|
||
use_char_ngrams=False,
|
||
use_pos_ngrams=False,
|
||
lsa_variants=(),
|
||
stylo_normalize=False,
|
||
include_lsa=False,
|
||
)
|
||
if name in {
|
||
"lsa_word", "lsa_word2", "lsa_content", "lsa_fw", "lsa_char", "lsa_pos"
|
||
}:
|
||
variant = name.removeprefix("lsa_")
|
||
return FeatureConfig(
|
||
groups=(),
|
||
use_char_ngrams=False,
|
||
use_pos_ngrams=False,
|
||
lsa_variants=(variant,),
|
||
include_stylo=False,
|
||
include_lsa=True,
|
||
)
|
||
if name == "length":
|
||
return FeatureConfig(
|
||
groups=("length",),
|
||
use_char_ngrams=False,
|
||
use_pos_ngrams=False,
|
||
lsa_variants=(),
|
||
stylo_normalize=False,
|
||
include_lsa=False,
|
||
)
|
||
raise ValueError(f"Неизвестный набор признаков: {name}")
|
||
|
||
|
||
PARAM_GRIDS: dict[str, dict[str, list[Any]]] = {
|
||
"logreg": {"model__C": [0.1, 1.0, 10.0]},
|
||
"linear_svc": {"model__C": [0.1, 1.0, 10.0]},
|
||
"random_forest": {
|
||
"model__max_depth": [None, 30],
|
||
"model__min_samples_leaf": [1, 2],
|
||
},
|
||
"extra_trees": {
|
||
"model__max_depth": [None, 30],
|
||
"model__min_samples_leaf": [1, 2],
|
||
},
|
||
"multinomial_nb": {"model__alpha": [0.1, 1.0, 3.0]},
|
||
"knn": {"model__n_neighbors": [3, 5, 9]},
|
||
}
|
||
|
||
|
||
def make_grid_search(
|
||
pipeline: Pipeline,
|
||
model_name: str,
|
||
cv: int = 3,
|
||
scoring: str = "f1_macro",
|
||
n_jobs: int = -1,
|
||
) -> GridSearchCV:
|
||
"""GridSearchCV с сеткой, специфичной для модели."""
|
||
if model_name not in PARAM_GRIDS:
|
||
raise ValueError(f"Для модели {model_name} нет сетки гиперпараметров.")
|
||
return GridSearchCV(
|
||
pipeline,
|
||
PARAM_GRIDS[model_name],
|
||
cv=cv,
|
||
scoring=scoring,
|
||
n_jobs=n_jobs,
|
||
refit=True,
|
||
return_train_score=False,
|
||
)
|