Первичная версия: стилометрия реплик, LSA и протокол экспериментов

This commit is contained in:
2026-10-24 01:17:36 +03:00
commit c390224500
52 changed files with 10099 additions and 0 deletions

View File

@@ -0,0 +1,13 @@
"""Стилометрия реплик Stardew Valley.
Основные модули:
- ``io`` — чтение и валидация корпуса;
- ``cleaning`` — нормализация текста;
- ``tokenization`` — NLTK-токенизация, Universal POS, резервный режим;
- ``features`` — ручные стилометрические признаки;
- ``lsa`` — TF-IDF и латентно-семантический анализ;
- ``pipelines`` — сборка признаковых пространств и классификаторов;
- ``cv``, ``evaluation``, ``stats`` — кросс-валидация, метрики и статистика.
"""
__version__ = "1.0.0"

123
lab1/stylometry/cleaning.py Normal file
View File

@@ -0,0 +1,123 @@
"""Нормализация текста реплик.
Модуль намеренно не выполняет агрессивную очистку: корпус уже отфильтрован
вручную. Здесь только приводятся к единому виду Unicode, плейсхолдеры и
пробельные символы. Пунктуация, регистр и разговорные элементы сохраняются,
поскольку они являются стилометрическими признаками.
"""
from __future__ import annotations
import re
import unicodedata
from typing import Iterable
# Плейсхолдеры игрового движка. Они не являются авторским текстом, поэтому
# заменяются на нейтральные слова, а их количество учитывается отдельным
# признаком.
PLACEHOLDER_PATTERN = re.compile(r"<([A-Z][A-Z0-9_]*)>")
PLACEHOLDER_MAP: dict[str, str] = {
"PLAYER": "PLAYER",
"FARM": "FARM",
"PET": "PET",
"SPOUSE": "SPOUSE",
"TIME": "TIME",
"NOUN": "NOUN",
"FIRST_LETTER": "FIRST_LETTER",
"NAME": "NAME",
}
# Символы, которые в корпусе могут встречаться в разных типографских
# вариантах. Приводим их к ASCII-эквивалентам, чтобы признаки были
# сопоставимы между репликами.
_UNICODE_REPLACEMENTS: dict[str, str] = {
"\u2026": "...", # многоточие
"\u2018": "'",
"\u2019": "'",
"\u201a": "'",
"\u201b": "'",
"\u201c": '"',
"\u201d": '"',
"\u201e": '"',
"\u2013": "-", # короткое тире
"\u2014": "-", # длинное тире
"\u2212": "-", # математический минус
"\u00a0": " ", # неразрывный пробел
"\u2009": " ", # тонкий пробел
"\u200a": " ",
"\u200b": "", # нулевой пробел
"\u200c": "",
"\u200d": "",
"\ufeff": "",
}
def normalize_unicode(text: str) -> str:
"""Приводит Unicode к предсказуемому виду, не меняя смысл текста."""
text = unicodedata.normalize("NFKC", str(text))
for source, target in _UNICODE_REPLACEMENTS.items():
text = text.replace(source, target)
return text
def normalize_placeholders(text: str) -> str:
"""Заменяет ``<PLAYER>`` и аналогичные метки на слова без угловых скобок."""
def _replace(match: re.Match[str]) -> str:
name = match.group(1)
return PLACEHOLDER_MAP.get(name, "ENTITY")
return PLACEHOLDER_PATTERN.sub(_replace, text)
def normalize_whitespace(text: str) -> str:
"""Схлопывает повторные пробелы и переводы строк."""
text = re.sub(r"[ \t\f\v]+", " ", text)
text = re.sub(r"\s*\n\s*", " ", text)
return text.strip()
def clean_text(text: str) -> str:
"""Основная функция нормализации одной реплики."""
text = normalize_unicode(text)
text = normalize_placeholders(text)
text = normalize_whitespace(text)
return text
def normalized_key(text: str) -> str:
"""Ключ для поиска точных дублей.
Регистр, пунктуация и плейсхолдеры игнорируются. Такой ключ использовался
при аудите корпуса; для обучения он не применяется.
"""
text = text.lower()
text = PLACEHOLDER_PATTERN.sub("", text)
text = re.sub(r"[^a-z0-9' ]+", " ", text)
return normalize_whitespace(text)
def count_placeholders(text: str) -> int:
"""Возвращает число плейсхолдеров в исходной реплике."""
return len(PLACEHOLDER_PATTERN.findall(text))
def placeholder_types(text: str) -> list[str]:
"""Возвращает типы плейсхолдеров в порядке появления."""
return [match.group(1) for match in PLACEHOLDER_PATTERN.finditer(text)]
def is_valid_line(text: str) -> bool:
"""Отбрасывает пустые строки и строки без латинских букв."""
return bool(re.search(r"[A-Za-z]", text))
def tokenize_for_audit(text: str) -> list[str]:
"""Простая токенизация для аудита; не заменяет NLTK-токенизатор."""
return re.findall(r"[A-Za-z]+(?:'[A-Za-z]+)?|\d+|[^\w\s]", text)
def join_tokens(tokens: Iterable[str]) -> str:
"""Склеивает токены через пробел, сохраняя их как отдельные единицы."""
return " ".join(str(token) for token in tokens)

866
lab1/stylometry/cli.py Normal file
View File

@@ -0,0 +1,866 @@
"""Командный интерфейс для аудита, признаков, LSA и экспериментов."""
from __future__ import annotations
import argparse
import json
import sys
from datetime import datetime
from pathlib import Path
from typing import Any, Sequence
import numpy as np
import pandas as pd
from .clustering import run_clustering
from .cv import CVResult, run_nested_cv, run_repeated_cv
from .datasets import DEFAULT_CACHE_PATH, load_feature_frame
from .delta import DeltaClassifier
from .evaluation import (
bootstrap_ci,
classification_metrics,
confusion_matrix_frame,
per_class_report,
top_confusions,
)
from .features import StylometricVectorizer, feature_dictionary
from .io import (
DEFAULT_DATA_PATH,
compare_with_stats,
corpus_fingerprint,
read_corpus,
read_stats,
validate_corpus,
write_csv,
)
from .lsa import LSA_VARIANTS, make_lsa_vectorizer
from .metadata import group_confusions, load_character_metadata, subgroup_report
from .pipelines import (
MODEL_NAMES,
FeatureConfig,
build_pipeline,
preset_config,
)
from .plotting import (
plot_class_distribution,
plot_confusion_matrix,
plot_pca,
plot_scree,
)
from .stats import bootstrap_difference, mcnemar_exact
from .verification import run_verification
PROJECT_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_RESULTS_DIR = PROJECT_ROOT / "results"
DEFAULT_CONFIG_PATH = PROJECT_ROOT / "configs" / "default.toml"
def _package_versions() -> dict[str, str]:
versions = {}
for name in ("numpy", "pandas", "scipy", "sklearn", "nltk", "matplotlib"):
try:
module = __import__(name)
versions[name] = getattr(module, "__version__", "unknown")
except Exception:
versions[name] = "missing"
return versions
def _json_default(value: Any) -> Any:
if isinstance(value, (np.integer,)):
return int(value)
if isinstance(value, (np.floating,)):
return float(value)
if isinstance(value, (np.ndarray,)):
return value.tolist()
if isinstance(value, Path):
return str(value)
return str(value)
def save_json(data: Any, path: str | Path) -> Path:
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", encoding="utf-8") as stream:
json.dump(data, stream, ensure_ascii=False, indent=2, default=_json_default)
return path
def _run_id(prefix: str) -> str:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
return f"{prefix}_{timestamp}"
def _safe_name(value: str) -> str:
return "".join(char if char.isalnum() or char in "-_." else "_" for char in value)
def _markdown_table(frame: pd.DataFrame) -> str:
try:
return frame.to_markdown(index=False)
except Exception:
return "```\n" + frame.to_string(index=False) + "\n```"
def _load_config(path: str | Path | None) -> dict[str, Any]:
if path is None:
path = DEFAULT_CONFIG_PATH
path = Path(path)
if not path.exists():
return {}
try:
import tomllib
with path.open("rb") as stream:
return tomllib.load(stream)
except Exception as error:
print(f"[config] не удалось прочитать {path}: {error}")
return {}
def _feature_config(preset: str, overrides: dict[str, Any] | None = None) -> FeatureConfig:
config = preset_config(preset)
if overrides:
for key, value in overrides.items():
if hasattr(config, key):
setattr(config, key, value)
return config
def _frame(args: argparse.Namespace) -> pd.DataFrame:
data_path = Path(args.data) if getattr(args, "data", None) else DEFAULT_DATA_PATH
cache_path = Path(args.cache) if getattr(args, "cache", None) else DEFAULT_CACHE_PATH
return load_feature_frame(
data_path=data_path,
cache_path=cache_path,
force=getattr(args, "no_cache", False),
verbose=True,
)
def _first_repeat(oof: pd.DataFrame) -> pd.DataFrame:
if oof.empty:
return oof
repeat = oof["repeat"].min()
return oof[oof["repeat"] == repeat].copy()
def _save_cv_result(
result: CVResult,
output_dir: Path,
config: dict[str, Any],
title: str,
) -> pd.DataFrame:
output_dir.mkdir(parents=True, exist_ok=True)
config.setdefault("package_versions", _package_versions())
result.fold_metrics.to_csv(output_dir / "fold_metrics.csv", index=False, encoding="utf-8-sig")
result.oof.to_csv(output_dir / "oof_predictions.csv", index=False, encoding="utf-8-sig")
fold_summary = result.summary()
fold_summary.to_csv(output_dir / "fold_summary.csv", index=False, encoding="utf-8-sig")
pooled, bootstrap = result.pooled_summary()
pooled.to_csv(output_dir / "pooled_metrics.csv", index=False, encoding="utf-8-sig")
save_json(bootstrap, output_dir / "bootstrap_ci.json")
first = _first_repeat(result.oof)
per_class = per_class_report(first["y_true"], first["y_pred"], result.labels)
per_class.to_csv(output_dir / "per_class_report.csv", index=False, encoding="utf-8-sig")
matrix = confusion_matrix_frame(first["y_true"], first["y_pred"], result.labels)
matrix.to_csv(output_dir / "confusion_matrix.csv", encoding="utf-8-sig")
normalized = confusion_matrix_frame(
first["y_true"], first["y_pred"], result.labels, normalize="true"
)
normalized.to_csv(output_dir / "confusion_matrix_normalized.csv", encoding="utf-8-sig")
top_confusions(first["y_true"], first["y_pred"], result.labels).to_csv(
output_dir / "top_confusions.csv", index=False, encoding="utf-8-sig"
)
plot_confusion_matrix(
matrix, output_dir / "figures" / "confusion_matrix.png", title="Матрица ошибок"
)
test_distribution = pd.DataFrame({"character": first["y_true"].values})
plot_class_distribution(
test_distribution,
character_column="character",
path=output_dir / "figures" / "test_class_distribution.png",
)
summary_row = pooled.iloc[0].to_dict() if not pooled.empty else {}
summary_row.update(
{
"model": config.get("model", ""),
"preset": config.get("preset", ""),
"folds": result.n_splits,
"repeats": result.n_repeats,
"n_samples": int(len(result.oof) // max(result.n_repeats, 1)),
"n_classes": int(len(result.labels)),
}
)
save_json({"title": title, "config": config, "metrics": summary_row, "bootstrap": bootstrap}, output_dir / "run_metadata.json")
lines = [
f"# {title}",
"",
f"- Модель: `{config.get('model', '')}`",
f"- Набор признаков: `{config.get('preset', '')}`",
f"- Кросс-валидация: {result.n_splits}-fold, повторов: {result.n_repeats}",
f"- Объектов: {summary_row.get('n_samples', '')}",
f"- Классов: {summary_row.get('n_classes', '')}",
"",
"## Основные метрики",
"",
_markdown_table(fold_summary),
"",
"## Pooled out-of-fold",
"",
_markdown_table(pooled),
"",
"## Доверительные интервалы (bootstrap)",
"",
"```json",
json.dumps(bootstrap, ensure_ascii=False, indent=2, default=_json_default),
"```",
"",
]
(output_dir / "summary.md").write_text("\n".join(lines), encoding="utf-8")
return fold_summary
def cmd_audit(args: argparse.Namespace) -> int:
data_path = Path(args.data) if args.data else DEFAULT_DATA_PATH
frame = read_corpus(data_path)
stats = validate_corpus(frame)
counts = pd.DataFrame(
sorted(stats["class_counts"].items(), key=lambda item: item[1], reverse=True),
columns=["Персонаж", "Реплик"],
)
word_lengths = frame["text"].str.findall(r"[A-Za-z0-9']+").map(len)
bins = pd.cut(
word_lengths,
bins=[0, 5, 10, 15, 20, 30, 1000],
labels=["0–5", "6–10", "11–15", "16–20", "21–30", "31+"],
include_lowest=True,
).value_counts().sort_index()
bin_frame = pd.DataFrame({"Длина реплики (слов)": bins.index.astype(str), "Реплик": bins.values})
comparison = compare_with_stats(frame, read_stats())
output_dir = Path(args.out) if args.out else PROJECT_ROOT / "data" / "processed"
output_dir.mkdir(parents=True, exist_ok=True)
write_csv(frame, output_dir / "lines.csv")
lines = [
"# Аудит корпуса",
"",
f"- Файл: `{data_path}`",
f"- SHA-256: `{corpus_fingerprint(data_path)}`",
f"- Реплик: {stats['n_lines']}",
f"- Персонажей: {stats['n_characters']}",
f"- Словоупотреблений: {stats['n_words']}",
f"- Средняя длина: {stats['mean_words_per_line']:.2f} слова",
f"- Медиана: {stats['median_words_per_line']:.1f} слова",
f"- Минимум/максимум: {stats['min_words_per_line']} / {stats['max_words_per_line']} слов",
f"- Точных дублей текста: {stats['n_duplicate_texts']}",
f"- Реплик с плейсхолдерами: {stats['n_placeholder_lines']}",
f"- Всего плейсхолдеров: {stats['n_placeholders']}",
"",
"## Распределение по персонажам",
"",
_markdown_table(counts),
"",
"## Распределение длин реплик",
"",
_markdown_table(bin_frame),
"",
]
if not comparison.empty:
lines.extend(
[
"## Сверка с stats_per_character.xlsx",
"",
_markdown_table(comparison.reset_index()),
"",
]
)
(output_dir / "audit.md").write_text("\n".join(lines), encoding="utf-8")
plot_class_distribution(frame, path=output_dir / "figures" / "class_distribution.png")
print(f"[audit] реплик: {stats['n_lines']}, персонажей: {stats['n_characters']}")
print(f"[audit] отчёт: {output_dir / 'audit.md'}")
return 0
def cmd_features(args: argparse.Namespace) -> int:
frame = _frame(args)
output_dir = Path(args.out) if args.out else PROJECT_ROOT / "features"
output_dir.mkdir(parents=True, exist_ok=True)
vectorizer = StylometricVectorizer()
matrix = vectorizer.fit_transform(frame)
names = vectorizer.get_feature_names_out()
result = pd.DataFrame(matrix, columns=names)
result.insert(0, "character", frame["character"].values)
result.insert(0, "line_id", frame["line_id"].values)
write_csv(result, output_dir / "stylometric.csv")
dictionary = feature_dictionary(names)
write_csv(dictionary, output_dir / "feature_dictionary.csv")
print(f"[features] матрица: {matrix.shape}; словарь: {output_dir / 'feature_dictionary.csv'}")
return 0
def cmd_lsa(args: argparse.Namespace) -> int:
frame = _frame(args)
output_dir = Path(args.out) if args.out else PROJECT_ROOT / "features"
output_dir.mkdir(parents=True, exist_ok=True)
variant = args.variant
components = args.components
vectorizer = make_lsa_vectorizer(variant, n_components=components)
matrix = vectorizer.fit_transform(frame)
result = pd.DataFrame(matrix, columns=vectorizer.get_feature_names_out())
result.insert(0, "character", frame["character"].values)
result.insert(0, "line_id", frame["line_id"].values)
path = output_dir / f"lsa_{variant}_k{vectorizer.n_components_}.csv"
write_csv(result, path)
top_terms = []
for component_index, terms in enumerate(vectorizer.top_terms(15), start=1):
for rank, (term, weight) in enumerate(terms, start=1):
top_terms.append(
{
"component": component_index,
"rank": rank,
"term": term,
"weight": weight,
}
)
write_csv(pd.DataFrame(top_terms), output_dir / f"lsa_{variant}_top_terms.csv")
variance = pd.DataFrame(
{
"component": np.arange(1, vectorizer.n_components_ + 1),
"explained_variance_ratio": vectorizer.explained_variance_ratio_,
"cumulative_explained_variance": vectorizer.cumulative_explained_variance_,
}
)
write_csv(variance, output_dir / f"lsa_{variant}_variance.csv")
plot_scree(
vectorizer.explained_variance_ratio_,
output_dir / f"lsa_{variant}_scree.png",
title=f"LSA ({variant}): объяснённая дисперсия",
)
print(
f"[lsa] вариант={variant}, компонент={vectorizer.n_components_}, "
f"накопленная дисперсия={vectorizer.cumulative_explained_variance_[-1]:.3f}"
)
print(f"[lsa] матрица: {path}")
return 0
def cmd_run(args: argparse.Namespace) -> int:
frame = _frame(args)
y = frame["character"].values
config_data = _load_config(args.config)
preset = args.preset or config_data.get("experiment", {}).get("preset", "stylometry_pos_lsa")
model = args.model or config_data.get("experiment", {}).get("model", "logreg")
folds = (
args.folds if args.folds is not None
else config_data.get("cv", {}).get("folds", 5)
)
repeats = (
args.repeats if args.repeats is not None
else config_data.get("cv", {}).get("repeats", 5)
)
seed = (
args.seed if args.seed is not None
else config_data.get("cv", {}).get("seed", 42)
)
overrides = dict(config_data.get("features", {}))
for key in ("char_ngram_range", "pos_ngram_range"):
if key in overrides and isinstance(overrides[key], list):
overrides[key] = tuple(overrides[key])
if "lsa_variants" in overrides and isinstance(overrides["lsa_variants"], list):
overrides["lsa_variants"] = tuple(overrides["lsa_variants"])
feature_config = _feature_config(preset, overrides)
pipeline = build_pipeline(feature_config, model)
run_config = {
"preset": preset,
"model": model,
"folds": folds,
"repeats": repeats,
"seed": seed,
"n_samples": int(len(frame)),
"n_classes": int(len(np.unique(y))),
"corpus_sha256": frame.attrs.get("corpus_sha256", ""),
"feature_config": dict(feature_config.__dict__),
"package_versions": _package_versions(),
}
if args.nested:
result = run_nested_cv(
pipeline, frame, y, model_name=model,
outer_splits=folds, inner_splits=3, seed=seed, verbose=True,
)
output_dir = DEFAULT_RESULTS_DIR / _run_id(f"nested_{preset}_{model}")
_save_cv_result(result, output_dir, run_config, f"Вложенная CV: {preset} / {model}")
save_json(result.best_params, output_dir / "best_params.json")
else:
result = run_repeated_cv(
pipeline, frame, y, n_splits=folds, n_repeats=repeats,
seed=seed, model_name=model, verbose=True,
)
output_dir = DEFAULT_RESULTS_DIR / _run_id(f"{preset}_{model}")
_save_cv_result(result, output_dir, run_config, f"CV: {preset} / {model}")
print(f"[run] результаты: {output_dir}")
print(result.summary().to_string(index=False))
return 0
def cmd_benchmark(args: argparse.Namespace) -> int:
frame = _frame(args)
y = frame["character"].values
preset = args.preset
models = args.models.split(",") if args.models else [
"logreg", "linear_svc", "random_forest", "multinomial_nb", "knn"
]
output_dir = DEFAULT_RESULTS_DIR / _run_id(f"benchmark_{preset}")
output_dir.mkdir(parents=True, exist_ok=True)
rows = []
for model in models:
model = model.strip()
if model not in MODEL_NAMES:
print(f"[benchmark] пропускаю неизвестную модель {model}")
continue
print(f"\n[benchmark] модель: {model}")
pipeline = build_pipeline(_feature_config(preset), model)
result = run_repeated_cv(
pipeline, frame, y, n_splits=args.folds, n_repeats=args.repeats,
seed=args.seed, model_name=model, verbose=False,
)
pooled, bootstrap = result.pooled_summary()
first = pooled.iloc[0].to_dict() if not pooled.empty else {}
rows.append(
{
"model": model,
"macro_f1": first.get("macro_f1"),
"balanced_accuracy": first.get("balanced_accuracy"),
"accuracy": first.get("accuracy"),
"mcc": first.get("mcc"),
"cohen_kappa": first.get("cohen_kappa"),
"top3_accuracy": first.get("top3_accuracy"),
"top5_accuracy": first.get("top5_accuracy"),
"macro_f1_ci_lower": bootstrap["macro_f1"]["lower"],
"macro_f1_ci_upper": bootstrap["macro_f1"]["upper"],
}
)
model_dir = output_dir / _safe_name(model)
_save_cv_result(
result,
model_dir,
{
"preset": preset,
"model": model,
"corpus_sha256": frame.attrs.get("corpus_sha256", ""),
},
f"Benchmark: {preset} / {model}",
)
summary = pd.DataFrame(rows).sort_values("macro_f1", ascending=False)
write_csv(summary, output_dir / "summary.csv")
print("\n" + summary.to_string(index=False))
return 0
def cmd_ablate(args: argparse.Namespace) -> int:
frame = _frame(args)
y = frame["character"].values
base = _feature_config(args.preset)
groups = list(base.groups)
if not groups:
print("[ablate] в выбранном наборе нет ручных групп признаков.")
return 1
output_dir = DEFAULT_RESULTS_DIR / _run_id(f"ablation_{args.preset}")
output_dir.mkdir(parents=True, exist_ok=True)
rows = []
variants = [("full", groups)] + [
(f"without_{group}", [value for value in groups if value != group])
for group in groups
]
for name, selected in variants:
config = FeatureConfig(
groups=tuple(selected),
use_char_ngrams=base.use_char_ngrams,
use_pos_ngrams=base.use_pos_ngrams,
lsa_variants=base.lsa_variants,
lsa_components=base.lsa_components,
stylo_scaler=base.stylo_scaler,
stylo_normalize=base.stylo_normalize,
include_stylo=True,
include_lsa=base.include_lsa,
char_max_features=base.char_max_features,
pos_max_features=base.pos_max_features,
char_ngram_range=base.char_ngram_range,
pos_ngram_range=base.pos_ngram_range,
min_df=base.min_df,
random_state=base.random_state,
)
pipeline = build_pipeline(config, args.model)
result = run_repeated_cv(
pipeline, frame, y, n_splits=args.folds, n_repeats=args.repeats,
seed=args.seed, model_name=args.model, verbose=False,
)
pooled, _ = result.pooled_summary()
first = pooled.iloc[0].to_dict() if not pooled.empty else {}
rows.append({"variant": name, "macro_f1": first.get("macro_f1"), "balanced_accuracy": first.get("balanced_accuracy"), "accuracy": first.get("accuracy")})
print(f"[ablate] {name}: macro-F1={first.get('macro_f1', float('nan')):.4f}")
table = pd.DataFrame(rows)
full_value = table.loc[table["variant"] == "full", "macro_f1"].iloc[0]
table["delta_macro_f1"] = table["macro_f1"] - full_value
write_csv(table, output_dir / "ablation.csv")
print(f"[ablate] результаты: {output_dir / 'ablation.csv'}")
return 0
def cmd_lsa_sweep(args: argparse.Namespace) -> int:
frame = _frame(args)
y = frame["character"].values
variants = args.variants.split(",")
components = [int(value) for value in args.components.split(",")]
output_dir = DEFAULT_RESULTS_DIR / _run_id("lsa_sweep")
output_dir.mkdir(parents=True, exist_ok=True)
rows = []
for variant in variants:
for k in components:
print(f"[lsa-sweep] {variant}, k={k}")
config = FeatureConfig(
groups=(),
use_char_ngrams=False,
use_pos_ngrams=False,
lsa_variants=(variant,),
lsa_components=k,
include_stylo=False,
include_lsa=True,
)
pipeline = build_pipeline(config, args.model)
result = run_repeated_cv(
pipeline, frame, y, n_splits=args.folds, n_repeats=args.repeats,
seed=args.seed, model_name=args.model, verbose=False,
)
pooled, _ = result.pooled_summary()
first = pooled.iloc[0].to_dict() if not pooled.empty else {}
rows.append({"variant": variant, "n_components": k, "macro_f1": first.get("macro_f1"), "accuracy": first.get("accuracy"), "top3_accuracy": first.get("top3_accuracy")})
table = pd.DataFrame(rows).sort_values(["macro_f1"], ascending=False)
write_csv(table, output_dir / "lsa_sweep.csv")
print(table.to_string(index=False))
return 0
def cmd_delta(args: argparse.Namespace) -> int:
frame = _frame(args)
y = frame["character"].values
output_dir = DEFAULT_RESULTS_DIR / _run_id("delta")
output_dir.mkdir(parents=True, exist_ok=True)
rows = []
for n_features in [int(value) for value in args.features.split(",")]:
for metric in args.metrics.split(","):
print(f"[delta] MFW={n_features}, metric={metric}")
classifier = DeltaClassifier(n_features=n_features, metric=metric)
result = run_repeated_cv(
classifier, frame, y, n_splits=args.folds, n_repeats=args.repeats,
seed=args.seed, model_name=f"delta_{metric}", verbose=False,
)
pooled, _ = result.pooled_summary()
first = pooled.iloc[0].to_dict() if not pooled.empty else {}
rows.append({"n_features": n_features, "metric": metric, "macro_f1": first.get("macro_f1"), "accuracy": first.get("accuracy"), "balanced_accuracy": first.get("balanced_accuracy")})
table = pd.DataFrame(rows).sort_values("macro_f1", ascending=False)
write_csv(table, output_dir / "delta_sweep.csv")
print(table.to_string(index=False))
return 0
def cmd_verify(args: argparse.Namespace) -> int:
frame = _frame(args)
y = frame["character"].values
feature_config = _feature_config(args.preset)
result = run_verification(frame, y, feature_config, n_splits=args.folds, seed=args.seed)
output_dir = DEFAULT_RESULTS_DIR / _run_id(f"verification_{args.preset}")
output_dir.mkdir(parents=True, exist_ok=True)
write_csv(result.fold_metrics, output_dir / "fold_metrics.csv")
result.scores.to_csv(output_dir / "pair_scores.csv", index=False, encoding="utf-8-sig")
save_json(result.summary, output_dir / "summary.json")
print(json.dumps(result.summary, ensure_ascii=False, indent=2))
return 0
def cmd_cluster(args: argparse.Namespace) -> int:
frame = _frame(args)
y = frame["character"].values
feature_config = _feature_config(args.preset)
result = run_clustering(frame, y, feature_config, n_clusters=args.clusters, seed=args.seed)
output_dir = DEFAULT_RESULTS_DIR / _run_id(f"clustering_{args.preset}")
output_dir.mkdir(parents=True, exist_ok=True)
write_csv(result.metrics, output_dir / "clustering_metrics.csv")
result.centroids.to_csv(output_dir / "character_centroids.csv", encoding="utf-8-sig")
plot_pca(
result.centroids.to_numpy(),
result.centroids.index.tolist(),
output_dir / "figures" / "character_centroids_pca.png",
title="PCA центроидов персонажей",
)
print(result.metrics.to_string(index=False))
return 0
def cmd_length_report(args: argparse.Namespace) -> int:
import pandas as pd
run_dir = Path(args.run_dir)
oof_path = run_dir / "oof_predictions.csv"
if not oof_path.exists():
raise FileNotFoundError(f"Нет файла предсказаний: {oof_path}")
oof = pd.read_csv(oof_path)
lines_path = Path(args.lines) if args.lines else PROJECT_ROOT / "data" / "processed" / "lines.csv"
if not lines_path.exists():
raise FileNotFoundError(
f"Нет файла {lines_path}. Сначала выполните python -m stylometry.cli audit."
)
lines = pd.read_csv(lines_path)
if "n_words" not in lines.columns:
lines["n_words"] = lines["text"].astype(str).str.findall(r"[A-Za-z0-9']+").map(len)
first = oof[oof["repeat"] == oof["repeat"].min()].copy()
first = first.merge(lines[["line_id", "n_words"]], on="line_id", how="left")
first["length_bin"] = pd.cut(
first["n_words"],
bins=[0, 5, 10, 15, 20, 30, 1000],
labels=["0–5", "6–10", "11–15", "16–20", "21–30", "31+"],
include_lowest=True,
)
labels = sorted(first["y_true"].unique())
rows = []
for bin_name, group in first.groupby("length_bin", observed=True):
# В узком диапазоне длины могут отсутствовать некоторые персонажи.
# Macro-F1 считаем только по классам, реально присутствующим в группе.
local_labels = sorted(group["y_true"].unique())
metrics = classification_metrics(
group["y_true"].values, group["y_pred"].values, labels=local_labels
)
interval = bootstrap_ci(
group["y_true"].values, group["y_pred"].values, local_labels,
metric="macro_f1", n_boot=500, seed=args.seed,
)
rows.append(
{
"length_bin": str(bin_name),
"n_samples": int(len(group)),
"macro_f1": metrics["macro_f1"],
"macro_f1_lower": interval["lower"],
"macro_f1_upper": interval["upper"],
"balanced_accuracy": metrics["balanced_accuracy"],
"accuracy": metrics["accuracy"],
"top3_accuracy": metrics.get("top3_accuracy"),
}
)
table = pd.DataFrame(rows)
output_path = run_dir / "length_bins.csv"
write_csv(table, output_path)
print(table.to_string(index=False))
print(f"[length-report] результат: {output_path}")
return 0
def cmd_compare(args: argparse.Namespace) -> int:
import pandas as pd
def _read_first(path: str | Path) -> pd.DataFrame:
frame = pd.read_csv(Path(path) / "oof_predictions.csv")
return frame[frame["repeat"] == frame["repeat"].min()].copy()
first = _read_first(args.run_a).rename(columns={"y_pred": "pred_a"})
second = _read_first(args.run_b).rename(columns={"y_pred": "pred_b"})
merged = first.merge(
second[["line_id", "y_true", "pred_b"]],
on=["line_id", "y_true"],
how="inner",
)
if merged.empty:
raise ValueError("Не удалось сопоставить предсказания по line_id и y_true.")
labels = sorted(merged["y_true"].unique())
metrics_a = classification_metrics(merged["y_true"], merged["pred_a"], labels=labels)
metrics_b = classification_metrics(merged["y_true"], merged["pred_b"], labels=labels)
comparison = {
"n_matched": int(len(merged)),
"model_a": {
"macro_f1": metrics_a["macro_f1"],
"accuracy": metrics_a["accuracy"],
"balanced_accuracy": metrics_a["balanced_accuracy"],
},
"model_b": {
"macro_f1": metrics_b["macro_f1"],
"accuracy": metrics_b["accuracy"],
"balanced_accuracy": metrics_b["balanced_accuracy"],
},
"mcnemar": mcnemar_exact(merged["y_true"], merged["pred_a"], merged["pred_b"]),
"bootstrap_difference_macro_f1": bootstrap_difference(
merged["y_true"], merged["pred_a"], merged["pred_b"],
metric="macro_f1", labels=labels, n_boot=1000, seed=args.seed,
),
}
output = {
"run_a": str(args.run_a),
"run_b": str(args.run_b),
"comparison": comparison,
}
path = Path(args.out) if args.out else Path(args.run_a) / "comparison_with_other.json"
save_json(output, path)
print(json.dumps(comparison, ensure_ascii=False, indent=2, default=_json_default))
print(f"[compare] результат: {path}")
return 0
def cmd_subgroup_report(args: argparse.Namespace) -> int:
import pandas as pd
run_dir = Path(args.run_dir)
oof_path = run_dir / "oof_predictions.csv"
if not oof_path.exists():
raise FileNotFoundError(f"Нет файла предсказаний: {oof_path}")
oof = pd.read_csv(oof_path)
first = oof[oof["repeat"] == oof["repeat"].min()].copy()
labels = sorted(first["y_true"].unique())
metadata = load_character_metadata(args.metadata)
metrics = subgroup_report(
first["y_true"], first["y_pred"], labels, metadata, args.group
)
confusions = group_confusions(
first["y_true"], first["y_pred"], metadata, args.group
)
write_csv(metrics, run_dir / f"subgroup_{args.group}_metrics.csv")
write_csv(confusions, run_dir / f"subgroup_{args.group}_confusions.csv")
print(metrics.to_string(index=False))
print(f"[subgroup-report] результаты: {run_dir}")
return 0
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Стилометрия реплик Stardew Valley: аудит, признаки, LSA и эксперименты."
)
subparsers = parser.add_subparsers(dest="command", required=True)
def add_common(sub: argparse.ArgumentParser) -> None:
sub.add_argument("--data", default=None, help="Путь к dialogues.csv или .xlsx")
sub.add_argument("--cache", default=None, help="Путь к кэшу токенизации")
sub.add_argument("--no-cache", action="store_true", help="Пересобрать кэш")
sub.add_argument("--out", default=None, help="Каталог для результатов")
audit = subparsers.add_parser("audit", help="Аудит корпуса")
add_common(audit)
audit.set_defaults(func=cmd_audit)
features = subparsers.add_parser("features", help="Ручные стилометрические признаки")
add_common(features)
features.set_defaults(func=cmd_features)
lsa = subparsers.add_parser("lsa", help="LSA одного варианта")
add_common(lsa)
lsa.add_argument("--variant", default="word", choices=LSA_VARIANTS)
lsa.add_argument("--components", type=int, default=100)
lsa.set_defaults(func=cmd_lsa)
run = subparsers.add_parser("run", help="Кросс-валидация одной модели")
add_common(run)
run.add_argument("--config", default=None)
run.add_argument("--preset", default=None)
run.add_argument("--model", default=None, choices=MODEL_NAMES)
run.add_argument("--folds", type=int, default=None)
run.add_argument("--repeats", type=int, default=None)
run.add_argument("--seed", type=int, default=None)
run.add_argument("--nested", action="store_true", help="Вложенная кросс-валидация")
run.set_defaults(func=cmd_run)
benchmark = subparsers.add_parser("benchmark", help="Сравнение классификаторов")
add_common(benchmark)
benchmark.add_argument("--preset", default="stylometry_pos_lsa")
benchmark.add_argument("--models", default=None, help="Список через запятую")
benchmark.add_argument("--folds", type=int, default=5)
benchmark.add_argument("--repeats", type=int, default=3)
benchmark.add_argument("--seed", type=int, default=42)
benchmark.set_defaults(func=cmd_benchmark)
ablate = subparsers.add_parser("ablate", help="Поочерёдное исключение групп признаков")
add_common(ablate)
ablate.add_argument("--preset", default="stylometry_pos_lsa")
ablate.add_argument("--model", default="logreg", choices=MODEL_NAMES)
ablate.add_argument("--folds", type=int, default=5)
ablate.add_argument("--repeats", type=int, default=2)
ablate.add_argument("--seed", type=int, default=42)
ablate.set_defaults(func=cmd_ablate)
sweep = subparsers.add_parser("lsa-sweep", help="Перебор вариантов и числа компонент LSA")
add_common(sweep)
sweep.add_argument("--variants", default="word,pos,char")
sweep.add_argument("--components", default="50,100,200")
sweep.add_argument("--model", default="logreg", choices=MODEL_NAMES)
sweep.add_argument("--folds", type=int, default=5)
sweep.add_argument("--repeats", type=int, default=1)
sweep.add_argument("--seed", type=int, default=42)
sweep.set_defaults(func=cmd_lsa_sweep)
delta = subparsers.add_parser("delta", help="Дельта Барроуза")
add_common(delta)
delta.add_argument("--features", default="100,200,500")
delta.add_argument("--metrics", default="manhattan,cosine")
delta.add_argument("--folds", type=int, default=5)
delta.add_argument("--repeats", type=int, default=2)
delta.add_argument("--seed", type=int, default=42)
delta.set_defaults(func=cmd_delta)
verify = subparsers.add_parser("verify", help="Верификация «свой/чужой»")
add_common(verify)
verify.add_argument("--preset", default="stylometry_pos_lsa")
verify.add_argument("--folds", type=int, default=5)
verify.add_argument("--seed", type=int, default=42)
verify.set_defaults(func=cmd_verify)
cluster = subparsers.add_parser("cluster", help="Кластеризация персонажей")
add_common(cluster)
cluster.add_argument("--preset", default="stylometry_pos_lsa")
cluster.add_argument("--clusters", type=int, default=None)
cluster.add_argument("--seed", type=int, default=42)
cluster.set_defaults(func=cmd_cluster)
length_report = subparsers.add_parser(
"length-report", help="Метрики по группам длины реплик"
)
length_report.add_argument("--run-dir", required=True)
length_report.add_argument("--lines", default=None)
length_report.add_argument("--seed", type=int, default=42)
length_report.set_defaults(func=cmd_length_report)
compare = subparsers.add_parser(
"compare", help="Парное сравнение двух запусков"
)
compare.add_argument("--run-a", required=True)
compare.add_argument("--run-b", required=True)
compare.add_argument("--out", default=None)
compare.add_argument("--seed", type=int, default=42)
compare.set_defaults(func=cmd_compare)
subgroup = subparsers.add_parser(
"subgroup-report", help="Метрики по полу, возрасту и виду персонажей"
)
subgroup.add_argument("--run-dir", required=True)
subgroup.add_argument("--metadata", default=None)
subgroup.add_argument(
"--group", default="gender", choices=["gender", "age_group", "species"]
)
subgroup.set_defaults(func=cmd_subgroup_report)
return parser
def main(argv: Sequence[str] | None = None) -> int:
parser = build_parser()
args = parser.parse_args(argv)
return int(args.func(args))
if __name__ == "__main__":
sys.exit(main())

View File

@@ -0,0 +1,90 @@
"""Кластеризация персонажей в стилометрическом пространстве."""
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.cluster import AgglomerativeClustering, KMeans
from sklearn.metrics import (
adjusted_rand_score,
calinski_harabasz_score,
davies_bouldin_score,
normalized_mutual_info_score,
silhouette_score,
)
from sklearn.preprocessing import normalize
from .pipelines import FeatureConfig, build_feature_union
@dataclass
class ClusteringResult:
metrics: pd.DataFrame
centroids: pd.DataFrame
def _centroid_matrix(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_clustering(
X: pd.DataFrame,
y: Sequence,
feature_config: FeatureConfig,
n_clusters: int | None = None,
seed: int = 42,
) -> ClusteringResult:
"""Кластеризует центроиды персонажей и считает внешние метрики."""
y = np.asarray(y)
features = build_feature_union(feature_config)
matrix = features.fit_transform(X)
matrix = matrix.toarray() if issparse(matrix) else np.asarray(matrix)
matrix = normalize(matrix, norm="l2", copy=False)
classes, centroids = _centroid_matrix(matrix, y)
# Если число кластеров равно числу персонажей, кластеризация тривиальна,
# а силуэт не определён. По умолчанию берём примерно корень из числа
# классов: для 34 персонажей это 6 групп.
n_clusters = int(n_clusters or max(2, round(len(classes) ** 0.5)))
n_clusters = min(n_clusters, len(classes) - 1)
rows: list[dict[str, Any]] = []
for name, model in (
(
"agglomerative_cosine",
AgglomerativeClustering(
n_clusters=n_clusters, metric="cosine", linkage="average"
),
),
(
"kmeans",
KMeans(n_clusters=n_clusters, random_state=seed, n_init=20),
),
):
predicted = model.fit_predict(centroids)
n_predicted = len(set(predicted))
valid_partition = 2 <= n_predicted < len(centroids)
rows.append(
{
"method": name,
"n_clusters": n_clusters,
"adjusted_rand_index": float(adjusted_rand_score(classes, predicted)),
"normalized_mutual_info": float(normalized_mutual_info_score(classes, predicted)),
"silhouette_cosine": float(
silhouette_score(centroids, predicted, metric="cosine")
) if valid_partition else 0.0,
"calinski_harabasz": float(calinski_harabasz_score(centroids, predicted))
if valid_partition else 0.0,
"davies_bouldin": float(davies_bouldin_score(centroids, predicted))
if valid_partition else 0.0,
}
)
centroid_frame = pd.DataFrame(centroids, index=classes)
centroid_frame.index.name = "character"
return ClusteringResult(metrics=pd.DataFrame(rows), centroids=centroid_frame)

293
lab1/stylometry/cv.py Normal file
View File

@@ -0,0 +1,293 @@
"""Схемы кросс-валидации и сбор out-of-fold предсказаний."""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Iterable, Sequence
import numpy as np
import pandas as pd
from sklearn.base import clone
from sklearn.model_selection import StratifiedKFold
from .evaluation import (
align_scores,
classification_metrics,
metrics_summary,
summarize_pooled_predictions,
)
from .pipelines import make_grid_search
@dataclass
class CVResult:
"""Результат повторной стратифицированной кросс-валидации."""
fold_metrics: pd.DataFrame
oof: pd.DataFrame
labels: np.ndarray
scores_by_repeat: dict[int, np.ndarray] = field(default_factory=dict)
n_splits: int = 5
n_repeats: int = 1
model_name: str = ""
def summary(self) -> pd.DataFrame:
return metrics_summary(self.fold_metrics)
def pooled_summary(self) -> tuple[pd.DataFrame, dict[str, float]]:
return summarize_pooled_predictions(
self.oof, self.labels, self.scores_by_repeat
)
@dataclass
class NestedCVResult(CVResult):
"""Результат вложенной кросс-валидации с настройкой гиперпараметров."""
best_params: list[dict[str, Any]] = field(default_factory=list)
def _predict_scores(pipeline: Any, X: pd.DataFrame) -> np.ndarray | None:
if hasattr(pipeline, "predict_proba"):
try:
return np.asarray(pipeline.predict_proba(X), dtype=float)
except Exception:
pass
if hasattr(pipeline, "decision_function"):
try:
scores = np.asarray(pipeline.decision_function(X), dtype=float)
if scores.ndim == 1:
scores = np.column_stack([-scores, scores])
return scores
except Exception:
pass
return None
def _model_classes(pipeline: Any) -> np.ndarray:
if hasattr(pipeline, "named_steps") and "model" in pipeline.named_steps:
return np.asarray(pipeline.named_steps["model"].classes_)
return np.asarray(pipeline.classes_)
def validate_folds(
y: Sequence,
splits: Iterable[tuple[np.ndarray, np.ndarray]],
labels: Sequence,
) -> None:
"""Проверяет, что каждый класс есть и в обучении, и в тесте."""
y = np.asarray(y)
labels_set = set(labels)
for fold_index, (train_idx, test_idx) in enumerate(splits):
train_labels = set(y[train_idx])
test_labels = set(y[test_idx])
missing_train = labels_set - train_labels
missing_test = labels_set - test_labels
if missing_train or missing_test:
raise ValueError(
f"Фолд {fold_index}: в обучении нет классов {sorted(missing_train)}, "
f"в тесте нет классов {sorted(missing_test)}. "
"Увеличьте размер выборки или уменьшите число фолдов."
)
def _line_ids(X: pd.DataFrame, indices: np.ndarray) -> np.ndarray:
if "line_id" in X.columns:
return X.iloc[indices]["line_id"].to_numpy()
return np.asarray(indices)
def _order_scores_like_oof(X: pd.DataFrame, rows: list[dict[str, Any]]) -> np.ndarray:
"""Возвращает порядок строк матрицы score-ов, соответствующий OOF-таблице."""
if "line_id" in X.columns:
positions = {
int(value): index for index, value in enumerate(X["line_id"].to_numpy())
}
return np.asarray([positions[int(row["line_id"])] for row in rows], dtype=int)
return np.asarray([int(row["line_id"]) for row in rows], dtype=int)
def run_repeated_cv(
pipeline: Any,
X: pd.DataFrame,
y: Sequence,
n_splits: int = 5,
n_repeats: int = 5,
seed: int = 42,
model_name: str = "",
verbose: bool = True,
) -> CVResult:
"""Повторная стратифицированная кросс-валидация.
Все преобразования внутри ``pipeline`` обучаются только на обучающем
блоке. Out-of-fold предсказания собираются для каждого повтора.
"""
y = np.asarray(y)
labels = np.unique(y)
n_samples = len(y)
fold_rows: list[dict[str, Any]] = []
oof_rows: list[dict[str, Any]] = []
scores_by_repeat: dict[int, np.ndarray] = {}
for repeat in range(n_repeats):
splitter = StratifiedKFold(
n_splits=n_splits, shuffle=True, random_state=seed + repeat
)
splits = list(splitter.split(X, y))
validate_folds(y, splits, labels)
scores_oof = np.full((n_samples, len(labels)), np.nan, dtype=float)
oof_start = len(oof_rows)
for fold, (train_idx, test_idx) in enumerate(splits, start=1):
model = clone(pipeline)
model.fit(X.iloc[train_idx], y[train_idx])
y_pred = model.predict(X.iloc[test_idx])
raw_scores = _predict_scores(model, X.iloc[test_idx])
if raw_scores is not None:
scores = align_scores(
raw_scores, _model_classes(model), labels, fill_value=0.0
)
scores_oof[test_idx] = scores
else:
scores = None
metrics = classification_metrics(
y[test_idx], y_pred, y_score=scores, labels=labels
)
metrics.update(
{
"repeat": repeat,
"fold": fold,
"n_train": int(len(train_idx)),
"n_test": int(len(test_idx)),
}
)
fold_rows.append(metrics)
for row_index, sample_index in enumerate(test_idx):
oof_rows.append(
{
"repeat": repeat,
"fold": fold,
"line_id": int(_line_ids(X, np.asarray([sample_index]))[0]),
"character": str(X.iloc[sample_index].get("character", "")),
"y_true": y[sample_index],
"y_pred": y_pred[row_index],
}
)
if verbose:
print(
f"[cv] повтор {repeat + 1}/{n_repeats}, "
f"фолд {fold}/{n_splits}: macro-F1={metrics['macro_f1']:.4f}"
)
if not np.isnan(scores_oof).any():
repeat_rows = sorted(oof_rows[oof_start:], key=lambda row: int(row["line_id"]))
order = _order_scores_like_oof(X, repeat_rows)
scores_by_repeat[repeat] = scores_oof[order]
oof = pd.DataFrame(oof_rows)
if not oof.empty:
oof = oof.sort_values(["repeat", "line_id"]).reset_index(drop=True)
return CVResult(
fold_metrics=pd.DataFrame(fold_rows),
oof=oof,
labels=labels,
scores_by_repeat=scores_by_repeat,
n_splits=n_splits,
n_repeats=n_repeats,
model_name=model_name,
)
def run_nested_cv(
pipeline: Any,
X: pd.DataFrame,
y: Sequence,
model_name: str,
outer_splits: int = 5,
inner_splits: int = 3,
seed: int = 42,
verbose: bool = True,
) -> NestedCVResult:
"""Вложенная кросс-валидация: настройка гиперпараметров внутри outer-фолдов."""
y = np.asarray(y)
labels = np.unique(y)
n_samples = len(y)
outer = StratifiedKFold(
n_splits=outer_splits, shuffle=True, random_state=seed
)
splits = list(outer.split(X, y))
validate_folds(y, splits, labels)
fold_rows: list[dict[str, Any]] = []
oof_rows: list[dict[str, Any]] = []
best_params: list[dict[str, Any]] = []
scores_oof = np.full((n_samples, len(labels)), np.nan, dtype=float)
for fold, (train_idx, test_idx) in enumerate(splits, start=1):
search = make_grid_search(
pipeline, model_name, cv=inner_splits, scoring="f1_macro", n_jobs=-1
)
search.fit(X.iloc[train_idx], y[train_idx])
model = search.best_estimator_
y_pred = model.predict(X.iloc[test_idx])
raw_scores = _predict_scores(model, X.iloc[test_idx])
if raw_scores is not None:
scores = align_scores(
raw_scores, _model_classes(model), labels, fill_value=0.0
)
scores_oof[test_idx] = scores
else:
scores = None
metrics = classification_metrics(
y[test_idx], y_pred, y_score=scores, labels=labels
)
metrics.update(
{
"repeat": 0,
"fold": fold,
"n_train": int(len(train_idx)),
"n_test": int(len(test_idx)),
}
)
fold_rows.append(metrics)
best_params.append(dict(search.best_params_))
for row_index, sample_index in enumerate(test_idx):
oof_rows.append(
{
"repeat": 0,
"fold": fold,
"line_id": int(_line_ids(X, np.asarray([sample_index]))[0]),
"character": str(X.iloc[sample_index].get("character", "")),
"y_true": y[sample_index],
"y_pred": y_pred[row_index],
}
)
if verbose:
print(
f"[nested-cv] фолд {fold}/{outer_splits}: "
f"macro-F1={metrics['macro_f1']:.4f}, "
f"best={search.best_params_}"
)
scores_by_repeat = {}
if not np.isnan(scores_oof).any():
ordered_rows = sorted(oof_rows, key=lambda row: int(row["line_id"]))
order = _order_scores_like_oof(X, ordered_rows)
scores_by_repeat[0] = scores_oof[order]
oof = pd.DataFrame(oof_rows)
if not oof.empty:
oof = oof.sort_values(["repeat", "line_id"]).reset_index(drop=True)
return NestedCVResult(
fold_metrics=pd.DataFrame(fold_rows),
oof=oof,
labels=labels,
scores_by_repeat=scores_by_repeat,
n_splits=outer_splits,
n_repeats=1,
model_name=model_name,
best_params=best_params,
)

View File

@@ -0,0 +1,86 @@
"""Подготовка таблицы реплик с токенами, тегами и текстами для LSA."""
from __future__ import annotations
from pathlib import Path
import pandas as pd
from .io import DEFAULT_DATA_PATH, corpus_fingerprint, read_corpus
from .tokenization import Tokenizer
DEFAULT_CACHE_PATH = Path(__file__).resolve().parents[1] / "data" / "processed" / "tokenized.pkl"
REQUIRED_COLUMNS = {
"line_id",
"character",
"text",
"tokens",
"ptb",
"upos",
"sentences",
"pos_text",
"fw_text",
"content_text",
}
def build_feature_frame(
data_path: str | Path | None = None,
tokenizer: Tokenizer | None = None,
verbose: bool = True,
) -> pd.DataFrame:
"""Читает корпус и добавляет результаты токенизации и разметки."""
data_path = Path(data_path) if data_path is not None else DEFAULT_DATA_PATH
frame = read_corpus(data_path)
tokenizer = tokenizer or Tokenizer(use_nltk=True)
features = tokenizer.analyze_frame(frame, verbose=verbose)
features.attrs["corpus_path"] = str(data_path)
features.attrs["corpus_sha256"] = corpus_fingerprint(data_path)
features.attrs["nltk_used"] = tokenizer.use_nltk
features.attrs["pos_available"] = tokenizer.pos_available
return features
def save_feature_frame(frame: pd.DataFrame, path: str | Path) -> Path:
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
frame.to_pickle(path)
return path
def load_feature_frame(
data_path: str | Path | None = None,
cache_path: str | Path | None = None,
force: bool = False,
verbose: bool = True,
) -> pd.DataFrame:
"""Загружает кэш токенизации или строит его заново.
Кэш проверяется по SHA-256 исходного файла. Если файл корпуса изменился,
токенизация выполняется повторно.
"""
data_path = Path(data_path) if data_path is not None else DEFAULT_DATA_PATH
cache_path = Path(cache_path) if cache_path is not None else DEFAULT_CACHE_PATH
if cache_path.exists() and not force:
try:
frame = pd.read_pickle(cache_path)
if REQUIRED_COLUMNS <= set(frame.columns):
cached_hash = frame.attrs.get("corpus_sha256")
current_hash = corpus_fingerprint(data_path)
if cached_hash == current_hash:
if verbose:
print(f"[datasets] загружен кэш: {cache_path}")
return frame
if verbose:
print("[datasets] корпус изменился, пересобираю признаки.")
except Exception as error:
if verbose:
print(f"[datasets] не удалось прочитать кэш ({error}), пересобираю.")
frame = build_feature_frame(data_path, verbose=verbose)
save_feature_frame(frame, cache_path)
if verbose:
print(f"[datasets] кэш сохранён: {cache_path}")
return frame

148
lab1/stylometry/delta.py Normal file
View File

@@ -0,0 +1,148 @@
"""Классическая дельта Барроуза и её косинусный вариант.
Дельта Барроуза — стандартный метод стилометрии. Он использует частоты
наиболее частотных слов, z-нормирование по обучающей выборке и расстояние до
центроида класса. Мы реализуем два варианта: манхэттенское расстояние
(классическая дельта) и косинусное расстояние (Cosine Delta).
"""
from __future__ import annotations
import re
from collections import Counter
from typing import Any, Sequence
import numpy as np
import pandas as pd
from scipy.special import softmax
from scipy.spatial.distance import cdist
from sklearn.base import BaseEstimator, ClassifierMixin, TransformerMixin
_TOKEN_RE = re.compile(r"[A-Za-z]+(?:'[A-Za-z]+)?")
def _texts(X: Any) -> list[str]:
if isinstance(X, pd.DataFrame):
if "text" not in X.columns:
raise KeyError("Для Delta нужна колонка 'text'.")
return X["text"].astype(str).tolist()
return [str(value) for value in X]
def _tokens(text: str, lowercase: bool = True) -> list[str]:
tokens = _TOKEN_RE.findall(text)
return [token.lower() for token in tokens] if lowercase else tokens
class MFWVectorizer(BaseEstimator, TransformerMixin):
"""Векторизатор по наиболее частотным словам (Most Frequent Words)."""
def __init__(
self,
n_features: int = 100,
lowercase: bool = True,
zscore: bool = True,
) -> None:
self.n_features = n_features
self.lowercase = lowercase
self.zscore = zscore
def fit(self, X: Any, y=None) -> "MFWVectorizer":
texts = _texts(X)
counter: Counter[str] = Counter()
for text in texts:
counter.update(_tokens(text, self.lowercase))
self.vocabulary_ = [word for word, _ in counter.most_common(self.n_features)]
if not self.vocabulary_:
raise ValueError("Не удалось выделить слова для Delta.")
index = {word: position for position, word in enumerate(self.vocabulary_)}
raw = np.zeros((len(texts), len(self.vocabulary_)), dtype=float)
for row, text in enumerate(texts):
tokens = _tokens(text, self.lowercase)
counts = Counter(tokens)
total = max(len(tokens), 1)
for word, count in counts.items():
position = index.get(word)
if position is not None:
raw[row, position] = count / total
self.mean_ = raw.mean(axis=0)
std = raw.std(axis=0, ddof=0)
self.std_ = np.where(std == 0, 1.0, std)
return self
def transform(self, X: Any) -> np.ndarray:
texts = _texts(X)
index = {word: position for position, word in enumerate(self.vocabulary_)}
raw = np.zeros((len(texts), len(self.vocabulary_)), dtype=float)
for row, text in enumerate(texts):
tokens = _tokens(text, self.lowercase)
counts = Counter(tokens)
total = max(len(tokens), 1)
for word, count in counts.items():
position = index.get(word)
if position is not None:
raw[row, position] = count / total
if self.zscore:
return (raw - self.mean_) / self.std_
return raw
def get_feature_names_out(self, input_features=None) -> np.ndarray:
return np.asarray(self.vocabulary_, dtype=object)
class DeltaClassifier(BaseEstimator, ClassifierMixin):
"""Классификатор на основе расстояния до центроида класса."""
def __init__(
self,
n_features: int = 100,
metric: str = "manhattan",
zscore: bool = True,
lowercase: bool = True,
) -> None:
self.n_features = n_features
self.metric = metric
self.zscore = zscore
self.lowercase = lowercase
def _make_vectorizer(self) -> MFWVectorizer:
return MFWVectorizer(
n_features=self.n_features,
lowercase=self.lowercase,
zscore=self.zscore,
)
def fit(self, X: Any, y: Sequence) -> "DeltaClassifier":
y = np.asarray(y)
self.vectorizer_ = self._make_vectorizer()
matrix = self.vectorizer_.fit_transform(X)
self.classes_ = np.unique(y)
self.centroids_ = np.vstack(
[matrix[y == label].mean(axis=0) for label in self.classes_]
)
return self
def _distances(self, X: Any) -> np.ndarray:
matrix = self.vectorizer_.transform(X)
if self.metric == "manhattan":
return cdist(matrix, self.centroids_, metric="cityblock") / max(matrix.shape[1], 1)
if self.metric == "euclidean":
return cdist(matrix, self.centroids_, metric="euclidean")
if self.metric == "cosine":
return cdist(matrix, self.centroids_, metric="cosine")
raise ValueError(f"Неизвестная метрика Delta: {self.metric}")
def decision_function(self, X: Any) -> np.ndarray:
return -self._distances(X)
def predict(self, X: Any) -> np.ndarray:
distances = self._distances(X)
return self.classes_[np.argmin(distances, axis=1)]
def predict_proba(self, X: Any) -> np.ndarray:
distances = self._distances(X)
scale = np.maximum(distances.std(axis=1, keepdims=True), 1e-6)
return softmax(-distances / scale, axis=1)
def get_feature_names_out(self) -> np.ndarray:
return self.vectorizer_.get_feature_names_out()

View File

@@ -0,0 +1,273 @@
"""Метрики качества классификации и их агрегация."""
from __future__ import annotations
from typing import Iterable, Sequence
import numpy as np
import pandas as pd
from sklearn.metrics import (
accuracy_score,
balanced_accuracy_score,
cohen_kappa_score,
confusion_matrix,
f1_score,
log_loss,
matthews_corrcoef,
precision_recall_fscore_support,
top_k_accuracy_score,
)
def _labels_or_default(y_true: Sequence, y_pred: Sequence, labels: Sequence | None) -> np.ndarray:
if labels is not None:
return np.asarray(labels)
return np.unique(np.concatenate([np.asarray(y_true), np.asarray(y_pred)]))
def brier_score_multiclass(y_true: Sequence, y_score: np.ndarray, labels: Sequence) -> float:
"""Многоклассовая оценка Брайера: средний квадрат отклонения вероятностей."""
label_to_index = {label: index for index, label in enumerate(labels)}
y_true = np.asarray(y_true)
one_hot = np.zeros_like(y_score, dtype=float)
for row, label in enumerate(y_true):
one_hot[row, label_to_index[label]] = 1.0
return float(np.mean(np.sum((y_score - one_hot) ** 2, axis=1)))
def align_scores(
y_score: np.ndarray,
model_classes: Sequence,
labels: Sequence,
fill_value: float = 0.0,
) -> np.ndarray:
"""Приводит столбцы вероятностей/решений к единому порядку классов."""
labels = list(labels)
model_classes = list(model_classes)
result = np.full((y_score.shape[0], len(labels)), fill_value, dtype=float)
for source_index, label in enumerate(model_classes):
if label in labels:
result[:, labels.index(label)] = y_score[:, source_index]
return result
def classification_metrics(
y_true: Sequence,
y_pred: Sequence,
y_score: np.ndarray | None = None,
labels: Sequence | None = None,
top_k: Iterable[int] = (3, 5),
) -> dict[str, float]:
"""Считает набор метрик для одного разбиения."""
y_true = np.asarray(y_true)
y_pred = np.asarray(y_pred)
labels_array = _labels_or_default(y_true, y_pred, labels)
n_classes = len(labels_array)
result: dict[str, float] = {
"n_samples": int(len(y_true)),
"n_classes": int(n_classes),
"accuracy": float(accuracy_score(y_true, y_pred)),
"balanced_accuracy": float(
balanced_accuracy_score(y_true, y_pred)
),
"macro_f1": float(
f1_score(y_true, y_pred, labels=labels_array, average="macro", zero_division=0)
),
"weighted_f1": float(
f1_score(y_true, y_pred, labels=labels_array, average="weighted", zero_division=0)
),
"micro_f1": float(
f1_score(y_true, y_pred, labels=labels_array, average="micro", zero_division=0)
),
"mcc": float(matthews_corrcoef(y_true, y_pred)),
"cohen_kappa": float(cohen_kappa_score(y_true, y_pred, labels=labels_array)),
}
if y_score is not None:
y_score = np.asarray(y_score, dtype=float)
if y_score.ndim == 2 and y_score.shape[1] == n_classes:
for k in top_k:
if 1 < k < n_classes:
result[f"top{k}_accuracy"] = float(
top_k_accuracy_score(y_true, y_score, k=k, labels=labels_array)
)
row_sums = y_score.sum(axis=1)
if np.all(y_score >= 0) and np.allclose(row_sums, 1.0, atol=1e-5):
result["log_loss"] = float(log_loss(y_true, y_score, labels=labels_array))
result["brier_score"] = brier_score_multiclass(y_true, y_score, labels_array)
return result
def per_class_report(
y_true: Sequence,
y_pred: Sequence,
labels: Sequence,
) -> pd.DataFrame:
"""Точность, полнота и F-мера по каждому классу."""
precision, recall, f1, support = precision_recall_fscore_support(
y_true, y_pred, labels=labels, zero_division=0
)
return pd.DataFrame(
{
"class": list(labels),
"precision": precision,
"recall": recall,
"f1": f1,
"support": support,
}
)
def confusion_matrix_frame(
y_true: Sequence,
y_pred: Sequence,
labels: Sequence,
normalize: str | None = None,
) -> pd.DataFrame:
"""Матрица ошибок в виде таблицы с подписями классов."""
matrix = confusion_matrix(y_true, y_pred, labels=labels, normalize=normalize)
return pd.DataFrame(matrix, index=list(labels), columns=list(labels))
def top_confusions(
y_true: Sequence,
y_pred: Sequence,
labels: Sequence,
top_n: int = 30,
) -> pd.DataFrame:
"""Самые частые пары «истинный класс — предсказанный класс»."""
frame = confusion_matrix_frame(y_true, y_pred, labels)
rows = []
for true_label in frame.index:
for predicted_label in frame.columns:
if true_label == predicted_label:
continue
count = int(frame.loc[true_label, predicted_label])
if count > 0:
rows.append(
{
"true_class": true_label,
"predicted_class": predicted_label,
"count": count,
"share_of_true_class": count / max(int(frame.loc[true_label].sum()), 1),
}
)
result = pd.DataFrame(rows)
if result.empty:
return result
return result.sort_values(["count", "share_of_true_class"], ascending=False).head(top_n)
def bootstrap_ci(
y_true: Sequence,
y_pred: Sequence,
labels: Sequence,
metric: str = "macro_f1",
n_boot: int = 1000,
seed: int = 42,
) -> dict[str, float]:
"""Перцентильный доверительный интервал метрики по репликам."""
y_true = np.asarray(y_true)
y_pred = np.asarray(y_pred)
labels_array = np.asarray(labels)
rng = np.random.default_rng(seed)
def _metric(true_values: np.ndarray, predicted_values: np.ndarray) -> float:
if metric == "macro_f1":
return float(
f1_score(
true_values,
predicted_values,
labels=labels_array,
average="macro",
zero_division=0,
)
)
if metric == "accuracy":
return float(accuracy_score(true_values, predicted_values))
if metric == "balanced_accuracy":
return float(balanced_accuracy_score(true_values, predicted_values))
if metric == "mcc":
return float(matthews_corrcoef(true_values, predicted_values))
raise ValueError(f"Неизвестная метрика для bootstrap: {metric}")
values = []
n_samples = len(y_true)
for _ in range(n_boot):
indices = rng.integers(0, n_samples, n_samples)
values.append(_metric(y_true[indices], y_pred[indices]))
values_array = np.asarray(values, dtype=float)
return {
"mean": float(np.mean(values_array)),
"lower": float(np.percentile(values_array, 2.5)),
"upper": float(np.percentile(values_array, 97.5)),
"n_boot": int(n_boot),
}
def metrics_summary(
metrics: pd.DataFrame,
columns: Iterable[str] | None = None,
) -> pd.DataFrame:
"""Среднее и стандартное отклонение метрик по фолдам/повторам."""
if columns is None:
columns = [
column
for column in metrics.columns
if column not in {"repeat", "fold", "n_train", "n_test", "n_samples", "n_classes"}
]
rows = []
for column in columns:
if column not in metrics.columns:
continue
values = pd.to_numeric(metrics[column], errors="coerce").dropna()
if values.empty:
continue
rows.append(
{
"metric": column,
"mean": float(values.mean()),
"std": float(values.std(ddof=1)) if len(values) > 1 else 0.0,
"min": float(values.min()),
"max": float(values.max()),
"n": int(len(values)),
}
)
return pd.DataFrame(rows)
def summarize_pooled_predictions(
oof: pd.DataFrame,
labels: Sequence,
scores_by_repeat: dict[int, np.ndarray] | None = None,
top_k: Iterable[int] = (3, 5),
) -> tuple[pd.DataFrame, dict[str, float]]:
"""Считает метрики по pooled out-of-fold предсказаниям.
Возвращает таблицу метрик по повторам и итоговую bootstrap-оценку по
первому повтору (для доверительного интервала).
"""
rows = []
for repeat, group in oof.groupby("repeat"):
scores = None
if scores_by_repeat is not None and repeat in scores_by_repeat:
scores = scores_by_repeat[repeat]
metrics = classification_metrics(
group["y_true"].values,
group["y_pred"].values,
y_score=scores,
labels=labels,
top_k=top_k,
)
metrics["repeat"] = int(repeat)
rows.append(metrics)
per_repeat = pd.DataFrame(rows)
first_repeat = oof[oof["repeat"] == oof["repeat"].min()]
bootstrap = {
metric: bootstrap_ci(first_repeat["y_true"].values, first_repeat["y_pred"].values, labels, metric)
for metric in ("macro_f1", "accuracy", "balanced_accuracy", "mcc")
}
return per_repeat, bootstrap

183
lab1/stylometry/io.py Normal file
View File

@@ -0,0 +1,183 @@
"""Чтение и валидация корпуса реплик Stardew Valley."""
from __future__ import annotations
import hashlib
from pathlib import Path
from typing import Any
import pandas as pd
from .cleaning import clean_text, count_placeholders, is_valid_line
# Русские названия колонок в исходном CSV/XLSX.
COLUMN_RENAMES = {
"Реплика": "text",
"Персонаж": "character",
"Text": "text",
"Character": "character",
}
DEFAULT_DATA_PATH = Path(__file__).resolve().parents[1] / "data" / "dialogues.csv"
DEFAULT_STATS_PATH = Path(__file__).resolve().parents[1] / "data" / "stats_per_character.xlsx"
class CorpusError(ValueError):
"""Ошибка структуры или кодировки корпуса."""
def _read_csv_robust(path: Path) -> pd.DataFrame:
"""Пробует несколько кодировок и проверяет наличие нужных колонок.
Основная кодировка корпуса — cp1251. UTF-8 добавлен на случай, если файл
будет пересохранён; koi8-r — как редкий запасной вариант.
"""
last_error: Exception | None = None
for encoding in ("cp1251", "utf-8-sig", "utf-8", "koi8-r"):
try:
frame = pd.read_csv(path, sep=";", encoding=encoding, dtype=str)
except (UnicodeDecodeError, UnicodeError) as error:
last_error = error
continue
except pd.errors.ParserError as error:
last_error = error
continue
columns = set(frame.columns)
if {"Реплика", "Персонаж"} <= columns or {"text", "character"} <= columns:
return frame
# Файл прочитался, но это не тот формат. Пробуем следующую кодировку.
last_error = CorpusError(
f"В файле {path} не найдены колонки 'Реплика'/'Персонаж' "
f"или 'text'/'character'. Найдены: {list(frame.columns)}"
)
raise CorpusError(f"Не удалось прочитать CSV {path}: {last_error}")
def _read_excel(path: Path) -> pd.DataFrame:
"""Читает первый лист Excel. Второй лист в исходном файле пустой."""
try:
return pd.read_excel(path, sheet_name=0, dtype=str)
except Exception as error: # openpyxl поднимает разные типы исключений
raise CorpusError(f"Не удалось прочитать Excel {path}: {error}") from error
def read_corpus(path: str | Path | None = None) -> pd.DataFrame:
"""Читает корпус и возвращает таблицу с колонками ``text`` и ``character``.
Дополнительно сохраняются:
- ``text_raw`` — исходная реплика после приведения к строке;
- ``line_id`` — сквозной номер реплики;
- ``n_chars`` — длина нормализованного текста;
- ``placeholder_count`` — число плейсхолдеров.
"""
path = Path(path) if path is not None else DEFAULT_DATA_PATH
if not path.exists():
raise FileNotFoundError(f"Файл корпуса не найден: {path}")
suffix = path.suffix.lower()
if suffix == ".csv":
frame = _read_csv_robust(path)
elif suffix in {".xlsx", ".xls"}:
frame = _read_excel(path)
else:
raise CorpusError(f"Неподдерживаемый формат файла: {suffix}")
frame = frame.rename(columns=COLUMN_RENAMES)
required = {"text", "character"}
if not required <= set(frame.columns):
raise CorpusError(
f"Ожидались колонки {sorted(required)}, найдены {list(frame.columns)}"
)
frame = frame[["text", "character"]].copy()
frame = frame.dropna(subset=["text", "character"])
frame["text"] = frame["text"].astype(str).str.strip()
frame["character"] = frame["character"].astype(str).str.strip()
frame = frame[(frame["text"] != "") & (frame["character"] != "")]
frame["text_raw"] = frame["text"]
frame["text"] = frame["text_raw"].map(clean_text)
frame = frame[frame["text"].map(is_valid_line)].copy()
frame = frame.reset_index(drop=True)
frame.insert(0, "line_id", range(len(frame)))
frame["n_chars"] = frame["text"].str.len().astype(int)
frame["n_words"] = frame["text"].str.findall(r"[A-Za-z0-9']+").map(len).astype(int)
frame["placeholder_count"] = frame["text_raw"].map(count_placeholders).astype(int)
if frame.empty:
raise CorpusError("После нормализации корпус оказался пустым.")
return frame
def read_stats(path: str | Path | None = None) -> pd.DataFrame:
"""Читает таблицу статистики по персонажам, если она есть."""
path = Path(path) if path is not None else DEFAULT_STATS_PATH
if not path.exists():
return pd.DataFrame()
return pd.read_excel(path, sheet_name=0)
def validate_corpus(frame: pd.DataFrame) -> dict[str, Any]:
"""Считает основные характеристики корпуса для отчёта и проверок."""
counts = frame["character"].value_counts().sort_values(ascending=False)
word_counts = frame["text"].str.findall(r"[A-Za-z0-9']+").map(len)
char_counts = frame["text"].str.len()
return {
"n_lines": int(len(frame)),
"n_characters": int(frame["character"].nunique()),
"n_words": int(word_counts.sum()),
"mean_words_per_line": float(word_counts.mean()),
"median_words_per_line": float(word_counts.median()),
"min_words_per_line": int(word_counts.min()),
"max_words_per_line": int(word_counts.max()),
"mean_chars_per_line": float(char_counts.mean()),
"median_chars_per_line": float(char_counts.median()),
"class_counts": counts.to_dict(),
"n_duplicate_texts": int(frame["text"].duplicated().sum()),
"n_placeholder_lines": int((frame["placeholder_count"] > 0).sum()),
"n_placeholders": int(frame["placeholder_count"].sum()),
}
def compare_with_stats(frame: pd.DataFrame, stats: pd.DataFrame) -> pd.DataFrame:
"""Сравнивает фактические counts с таблицей stats_per_character.xlsx.
Возвращает таблицу с колонками ``character``, ``actual``, ``reported``,
``difference``. Если таблица статистики пуста, возвращается пустая таблица.
"""
if stats.empty or "Персонаж" not in stats.columns:
return pd.DataFrame()
reported_column = "реплик" if "реплик" in stats.columns else stats.columns[1]
reported = (
stats[["Персонаж", reported_column]]
.rename(columns={"Персонаж": "character", reported_column: "reported"})
.set_index("character")["reported"]
)
actual = frame["character"].value_counts()
comparison = pd.DataFrame({"actual": actual, "reported": reported})
comparison["difference"] = comparison["actual"] - comparison["reported"]
return comparison.fillna(0).astype({"actual": int, "reported": int, "difference": int})
def corpus_fingerprint(path: str | Path | None = None) -> str:
"""SHA-256 файла корпуса для журнала воспроизводимости."""
path = Path(path) if path is not None else DEFAULT_DATA_PATH
digest = hashlib.sha256()
with path.open("rb") as stream:
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def write_csv(frame: pd.DataFrame, path: str | Path) -> Path:
"""Сохраняет таблицу в UTF-8 с BOM, чтобы её открывал Excel."""
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
frame.to_csv(path, index=False, encoding="utf-8-sig")
return path

175
lab1/stylometry/lexicons.py Normal file
View File

@@ -0,0 +1,175 @@
"""Закрытые классы слов и небольшие лексиконы для стилометрии.
Списки фиксированы заранее и не зависят от корпуса. Это важно для
воспроизводимости: признаки не подбираются по тестовой выборке.
"""
from __future__ import annotations
ARTICLES = {"a", "an", "the"}
DETERMINERS = {
"this", "that", "these", "those",
"my", "your", "his", "her", "its", "our", "their",
"some", "any", "no", "every", "each", "either", "neither",
"both", "all", "few", "many", "much", "more", "most",
"other", "another", "such", "what", "which", "whose",
}
PREPOSITIONS = {
"about", "above", "across", "after", "against", "along", "among",
"around", "at", "before", "behind", "below", "beneath", "beside",
"between", "beyond", "by", "despite", "down", "during", "except",
"for", "from", "in", "inside", "into", "like", "near", "of", "off",
"on", "onto", "out", "outside", "over", "past", "since", "through",
"throughout", "to", "toward", "towards", "under", "until", "up",
"upon", "with", "within", "without",
}
COORDINATORS = {"and", "but", "or", "nor", "for", "so", "yet"}
SUBORDINATORS = {
"because", "although", "though", "while", "whereas", "if", "unless",
"since", "as", "than", "whether", "after", "before", "until", "when",
"where", "once", "whenever", "wherever",
}
WH_WORDS = {
"who", "whom", "whose", "what", "which", "when", "where", "why", "how",
}
PERSONAL_PRONOUNS = {
"i", "me", "my", "mine", "myself",
"you", "your", "yours", "yourself", "yourselves",
"he", "him", "his", "himself",
"she", "her", "hers", "herself",
"it", "its", "itself",
"we", "us", "our", "ours", "ourselves",
"they", "them", "their", "theirs", "themselves",
}
PRONOUN_CATEGORIES: dict[str, set[str]] = {
"first_singular": {"i", "me", "my", "mine", "myself"},
"first_plural": {"we", "us", "our", "ours", "ourselves"},
"second": {"you", "your", "yours", "yourself", "yourselves"},
"third_singular": {"he", "him", "his", "himself", "she", "her", "hers", "herself", "it", "its", "itself"},
"third_plural": {"they", "them", "their", "theirs", "themselves"},
"possessive": {"my", "mine", "your", "yours", "his", "her", "hers", "its", "our", "ours", "their", "theirs"},
"reflexive": {"myself", "yourself", "yourselves", "himself", "herself", "itself", "ourselves", "themselves"},
}
AUXILIARIES = {
"am", "is", "are", "was", "were", "be", "been", "being",
"have", "has", "had", "having",
"do", "does", "did", "doing", "done",
"'m", "'re", "'s", "'ve", "'ll", "'d",
}
MODALS = {
"can", "could", "may", "might", "must", "shall", "should", "will",
"would", "ought", "need", "dare",
}
NEGATIONS = {
"not", "no", "never", "none", "nothing", "nobody", "nowhere",
"neither", "nor", "n't", "cannot",
"can't", "won't", "don't", "doesn't", "didn't", "isn't", "aren't",
"wasn't", "weren't", "haven't", "hasn't", "hadn't", "couldn't",
"wouldn't", "shouldn't", "mustn't", "ain't",
}
CONTRACTIONS = {
"i'm", "i've", "i'll", "i'd",
"you're", "you've", "you'll", "you'd",
"he's", "he'll", "he'd",
"she's", "she'll", "she'd",
"it's", "it'll",
"we're", "we've", "we'll", "we'd",
"they're", "they've", "they'll", "they'd",
"that's", "there's", "what's", "who's", "where's", "how's",
"can't", "won't", "don't", "doesn't", "didn't", "isn't", "aren't",
"wasn't", "weren't", "haven't", "hasn't", "hadn't", "couldn't",
"wouldn't", "shouldn't", "mustn't", "ain't",
}
DISCOURSE_MARKERS = {
"oh", "ah", "eh", "hey", "hi", "hello", "huh", "hmm", "um", "uh",
"well", "so", "like", "just", "really", "actually", "anyway",
"maybe", "perhaps", "please", "thanks", "thank", "sorry", "wow",
"yeah", "yes", "okay", "ok", "right", "still", "even", "only",
"quite", "rather", "pretty", "too", "very",
}
INTERJECTIONS = {
"oh", "ah", "eh", "hey", "hi", "hello", "huh", "hmm", "um", "uh",
"wow", "yeah", "yes", "okay", "ok", "sigh", "giggle", "sniff",
"heh", "haha", "mmm", "phew", "ugh", "ouch", "oops", "yay",
"hooray", "bye", "goodbye",
}
HEDGES = {
"maybe", "perhaps", "probably", "possibly", "guess", "suppose",
"think", "kind", "sort", "like", "somewhat", "rather", "fairly",
"quite",
}
INTENSIFIERS = {
"very", "really", "so", "too", "pretty", "quite", "absolutely",
"totally", "completely", "extremely", "incredibly", "terribly",
"awfully", "highly",
}
POLITENESS = {
"please", "thanks", "thank", "sorry", "excuse", "pardon", "welcome",
"kindly",
}
# Объединённый список функциональных слов. Он используется и для
# относительных частот, и для отбора функциональных слов в LSA.
FUNCTION_WORDS: set[str] = set().union(
ARTICLES,
DETERMINERS,
PREPOSITIONS,
COORDINATORS,
SUBORDINATORS,
WH_WORDS,
PERSONAL_PRONOUNS,
AUXILIARIES,
MODALS,
NEGATIONS,
CONTRACTIONS,
)
FUNCTION_WORD_CATEGORIES: dict[str, set[str]] = {
"articles": ARTICLES,
"determiners": DETERMINERS,
"prepositions": PREPOSITIONS,
"coordinators": COORDINATORS,
"subordinators": SUBORDINATORS,
"wh_words": WH_WORDS,
"pronouns": PERSONAL_PRONOUNS,
"auxiliaries": AUXILIARIES,
"modals": MODALS,
"negations": NEGATIONS,
"contractions": CONTRACTIONS,
"discourse_markers": DISCOURSE_MARKERS,
}
# Запасной список стоп-слов, если NLTK-ресурс stopwords недоступен.
STOPWORD_FALLBACK = set().union(
ARTICLES,
DETERMINERS,
PREPOSITIONS,
COORDINATORS,
SUBORDINATORS,
WH_WORDS,
PERSONAL_PRONOUNS,
AUXILIARIES,
MODALS,
NEGATIONS,
{"also", "just", "then", "there", "here", "now", "very", "too", "only"},
)
# Слова, которые в NLTK помечаются как IN, но синтаксически являются
# подчинительными союзами. Используются при переходе к Universal POS.
SUBORDINATING_IN = SUBORDINATORS

317
lab1/stylometry/lsa.py Normal file
View File

@@ -0,0 +1,317 @@
"""Векторизация текста и латентно-семантический анализ (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)

View File

@@ -0,0 +1,99 @@
"""Анализ ошибок по метаданным персонажей."""
from __future__ import annotations
from pathlib import Path
from typing import Sequence
import pandas as pd
from .evaluation import per_class_report
DEFAULT_METADATA_PATH = Path(__file__).resolve().parents[1] / "data" / "metadata" / "characters.csv"
METADATA_COLUMNS = ("gender", "age_group", "species", "role")
def load_character_metadata(path: str | Path | None = None) -> pd.DataFrame:
"""Читает таблицу пола, возраста, вида и роли персонажей."""
path = Path(path) if path is not None else DEFAULT_METADATA_PATH
if not path.exists():
raise FileNotFoundError(f"Нет файла метаданных: {path}")
frame = pd.read_csv(path)
required = {"character", *METADATA_COLUMNS}
missing = required - set(frame.columns)
if missing:
raise ValueError(f"В метаданных нет колонок: {sorted(missing)}")
frame = frame.copy()
frame["character"] = frame["character"].astype(str).str.strip()
return frame
def subgroup_report(
y_true: Sequence,
y_pred: Sequence,
labels: Sequence,
metadata: pd.DataFrame,
group_column: str = "gender",
) -> pd.DataFrame:
"""Сводка F-меры по подгруппам персонажей."""
if group_column not in metadata.columns:
raise ValueError(f"Нет колонки {group_column!r} в метаданных.")
per_class = per_class_report(y_true, y_pred, labels)
per_class = per_class.merge(
metadata[["character", group_column]],
left_on="class",
right_on="character",
how="left",
)
rows = []
for value, group in per_class.groupby(group_column, dropna=False):
support = int(group["support"].sum())
weighted = float((group["f1"] * group["support"]).sum() / max(support, 1))
rows.append(
{
"group_column": group_column,
"group": str(value),
"n_characters": int(len(group)),
"n_samples": support,
"weighted_f1": weighted,
"mean_class_f1": float(group["f1"].mean()),
"min_class_f1": float(group["f1"].min()),
"max_class_f1": float(group["f1"].max()),
}
)
return pd.DataFrame(rows).sort_values(["weighted_f1"], ascending=False)
def group_confusions(
y_true: Sequence,
y_pred: Sequence,
metadata: pd.DataFrame,
group_column: str = "gender",
) -> pd.DataFrame:
"""Считает, как часто ошибки происходят внутри и между группами."""
frame = pd.DataFrame({"true_character": list(y_true), "pred_character": list(y_pred)})
frame = frame.merge(
metadata[["character", group_column]].rename(
columns={"character": "true_character", group_column: "true_group"}
),
on="true_character",
how="left",
).merge(
metadata[["character", group_column]].rename(
columns={"character": "pred_character", group_column: "pred_group"}
),
on="pred_character",
how="left",
)
errors = frame[frame["true_character"] != frame["pred_character"]]
table = (
errors.groupby(["true_group", "pred_group"], dropna=False)
.size()
.reset_index(name="count")
.sort_values("count", ascending=False)
)
total = max(int(table["count"].sum()), 1)
table["share_of_errors"] = table["count"] / total
table["same_group"] = table["true_group"] == table["pred_group"]
return table

View File

@@ -0,0 +1,365 @@
"""Сборка признаковых пространств и классификаторов."""
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
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]
return FeatureUnion(branches)
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))
if name == "logreg":
return LogisticRegression(
C=params.pop("C", 1.0),
solver="lbfgs",
max_iter=5000,
class_weight="balanced",
)
if name == "linear_svc":
return LinearSVC(
C=params.pop("C", 1.0),
class_weight="balanced",
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),
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),
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,
)

167
lab1/stylometry/plotting.py Normal file
View File

@@ -0,0 +1,167 @@
"""Построение графиков для отчёта и презентации."""
from __future__ import annotations
from pathlib import Path
from typing import Sequence
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt # noqa: E402
import numpy as np # noqa: E402
import pandas as pd # noqa: E402
from sklearn.decomposition import PCA # noqa: E402
from sklearn.manifold import TSNE # noqa: E402
def _save(fig: plt.Figure, path: str | Path) -> Path:
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
fig.tight_layout()
fig.savefig(path, dpi=200, bbox_inches="tight")
plt.close(fig)
return path
def plot_class_distribution(
frame: pd.DataFrame,
character_column: str = "character",
path: str | Path = "results/figures/class_distribution.png",
) -> Path:
counts = frame[character_column].value_counts().sort_values()
fig, ax = plt.subplots(figsize=(8, max(6, len(counts) * 0.25)))
ax.barh(counts.index, counts.values)
ax.set_xlabel("Число реплик")
ax.set_ylabel("Персонаж")
ax.set_title("Распределение реплик по персонажам")
return _save(fig, path)
def plot_scree(
explained_variance_ratio: Sequence[float],
path: str | Path = "results/figures/lsa_scree.png",
title: str = "Доля объяснённой дисперсии компонент LSA",
) -> Path:
values = np.asarray(explained_variance_ratio, dtype=float)
cumulative = np.cumsum(values)
fig, ax = plt.subplots(figsize=(7, 4.5))
ax.plot(np.arange(1, len(values) + 1), cumulative, marker="o", markersize=3)
ax.set_xlabel("Число компонент")
ax.set_ylabel("Накопленная доля дисперсии")
ax.set_title(title)
ax.grid(alpha=0.3)
return _save(fig, path)
def plot_confusion_matrix(
matrix: pd.DataFrame,
path: str | Path = "results/figures/confusion_matrix.png",
title: str = "Матрица ошибок",
annotate: bool | None = None,
) -> Path:
values = matrix.to_numpy(dtype=float)
n_classes = values.shape[0]
if annotate is None:
annotate = n_classes <= 20
fig, ax = plt.subplots(figsize=(max(8, n_classes * 0.35), max(7, n_classes * 0.32)))
image = ax.imshow(values, cmap="Blues", aspect="auto")
ax.set_xticks(np.arange(n_classes))
ax.set_yticks(np.arange(n_classes))
ax.set_xticklabels(matrix.columns, rotation=90, fontsize=7)
ax.set_yticklabels(matrix.index, fontsize=7)
ax.set_xlabel("Предсказанный класс")
ax.set_ylabel("Истинный класс")
ax.set_title(title)
if annotate:
for row in range(n_classes):
for column in range(n_classes):
value = values[row, column]
if value:
ax.text(
column, row, f"{value:.2f}" if value < 1 else f"{int(value)}",
ha="center", va="center", fontsize=6,
color="white" if value > values.max() * 0.5 else "black",
)
fig.colorbar(image, ax=ax, fraction=0.046, pad=0.04)
return _save(fig, path)
def plot_pca(
matrix: np.ndarray,
labels: Sequence[str],
path: str | Path = "results/figures/pca.png",
title: str = "PCA признакового пространства",
) -> Path:
values = np.asarray(matrix, dtype=float)
if values.shape[0] < 2:
raise ValueError("Для PCA нужно хотя бы два объекта.")
coordinates = PCA(n_components=2, random_state=42).fit_transform(values)
fig, ax = plt.subplots(figsize=(9, 7))
unique_labels = sorted(set(labels))
cmap = plt.get_cmap("tab20")
color_map = {label: cmap(index % 20) for index, label in enumerate(unique_labels)}
for label in unique_labels:
mask = np.asarray(labels) == label
ax.scatter(
coordinates[mask, 0], coordinates[mask, 1],
s=12, alpha=0.7, label=label, color=color_map[label],
)
ax.set_xlabel("Первая главная компонента")
ax.set_ylabel("Вторая главная компонента")
ax.set_title(title)
ax.legend(fontsize=6, ncol=2, loc="best", frameon=False)
return _save(fig, path)
def plot_tsne(
matrix: np.ndarray,
labels: Sequence[str],
path: str | Path = "results/figures/tsne.png",
title: str = "t-SNE признакового пространства",
perplexity: float = 30.0,
) -> Path:
values = np.asarray(matrix, dtype=float)
if values.shape[0] < 10:
raise ValueError("Для t-SNE нужно хотя бы десять объектов.")
perplexity = min(perplexity, max(5.0, values.shape[0] / 4.0))
coordinates = TSNE(
n_components=2,
perplexity=perplexity,
init="pca",
learning_rate="auto",
random_state=42,
).fit_transform(values)
fig, ax = plt.subplots(figsize=(9, 7))
unique_labels = sorted(set(labels))
cmap = plt.get_cmap("tab20")
color_map = {label: cmap(index % 20) for index, label in enumerate(unique_labels)}
for label in unique_labels:
mask = np.asarray(labels) == label
ax.scatter(
coordinates[mask, 0], coordinates[mask, 1],
s=10, alpha=0.7, label=label, color=color_map[label],
)
ax.set_xlabel("t-SNE 1")
ax.set_ylabel("t-SNE 2")
ax.set_title(title)
ax.legend(fontsize=6, ncol=2, loc="best", frameon=False)
return _save(fig, path)
def plot_feature_importance(
names: Sequence[str],
values: Sequence[float],
path: str | Path = "results/figures/feature_importance.png",
title: str = "Важность признаков",
top_n: int = 25,
) -> Path:
frame = pd.DataFrame({"feature": list(names), "value": np.asarray(values, dtype=float)})
frame["abs_value"] = frame["value"].abs()
frame = frame.sort_values("abs_value", ascending=False).head(top_n).iloc[::-1]
fig, ax = plt.subplots(figsize=(9, max(5, len(frame) * 0.28)))
ax.barh(frame["feature"], frame["value"])
ax.set_xlabel("Вклад признака")
ax.set_ylabel("Признак")
ax.set_title(title)
return _save(fig, path)

143
lab1/stylometry/stats.py Normal file
View File

@@ -0,0 +1,143 @@
"""Статистические критерии для сравнения моделей."""
from __future__ import annotations
from typing import Sequence
import numpy as np
from scipy.stats import binomtest, friedmanchisquare, wilcoxon
from sklearn.metrics import accuracy_score, f1_score
def mcnemar_exact(
y_true: Sequence,
predictions_a: Sequence,
predictions_b: Sequence,
) -> dict[str, float]:
"""Точный критерий Мак-Немара для двух парных предсказаний."""
y_true = np.asarray(y_true)
correct_a = np.asarray(predictions_a) == y_true
correct_b = np.asarray(predictions_b) == y_true
b = int(np.sum(correct_a & ~correct_b))
c = int(np.sum(~correct_a & correct_b))
n = b + c
p_value = 1.0 if n == 0 else float(binomtest(min(b, c), n, 0.5).pvalue)
return {
"b": b,
"c": c,
"n_discordant": n,
"statistic": float(min(b, c)),
"p_value": p_value,
}
def wilcoxon_signed_rank(
scores_a: Sequence[float],
scores_b: Sequence[float],
) -> dict[str, float]:
"""Парный критерий Уилкоксона для результатов по фолдам."""
scores_a = np.asarray(scores_a, dtype=float)
scores_b = np.asarray(scores_b, dtype=float)
differences = scores_a - scores_b
if np.allclose(differences, 0):
return {"statistic": 0.0, "p_value": 1.0, "n": int(len(scores_a))}
statistic, p_value = wilcoxon(scores_a, scores_b, alternative="two-sided")
return {
"statistic": float(statistic),
"p_value": float(p_value),
"n": int(len(scores_a)),
}
def friedman_test(*score_arrays: Sequence[float]) -> dict[str, float]:
"""Критерий Фридмана для сравнения нескольких моделей."""
arrays = [np.asarray(values, dtype=float) for values in score_arrays]
if len(arrays) < 3:
raise ValueError("Для критерия Фридмана нужно хотя бы три модели.")
if len({len(values) for values in arrays}) != 1:
raise ValueError("Все массивы должны быть одинаковой длины.")
statistic, p_value = friedmanchisquare(*arrays)
return {"statistic": float(statistic), "p_value": float(p_value)}
def holm_bonferroni(p_values: Sequence[float]) -> list[float]:
"""Поправка Холма—Бонферрони для множественных сравнений."""
values = np.asarray(p_values, dtype=float)
n = len(values)
order = np.argsort(values)
adjusted = np.empty(n, dtype=float)
previous = 0.0
for rank, index in enumerate(order):
value = min(1.0, values[index] * (n - rank))
previous = max(previous, value)
adjusted[index] = previous
return adjusted.tolist()
def bootstrap_difference(
y_true: Sequence,
predictions_a: Sequence,
predictions_b: Sequence,
metric: str = "macro_f1",
labels: Sequence | None = None,
n_boot: int = 1000,
seed: int = 42,
) -> dict[str, float]:
"""Доверительный интервал разницы метрик двух моделей."""
y_true = np.asarray(y_true)
predictions_a = np.asarray(predictions_a)
predictions_b = np.asarray(predictions_b)
labels_array = np.asarray(labels) if labels is not None else np.unique(y_true)
def _metric(true_values: np.ndarray, predicted_values: np.ndarray) -> float:
if metric == "macro_f1":
return float(
f1_score(
true_values,
predicted_values,
labels=labels_array,
average="macro",
zero_division=0,
)
)
if metric == "accuracy":
return float(accuracy_score(true_values, predicted_values))
raise ValueError(f"Неизвестная метрика: {metric}")
rng = np.random.default_rng(seed)
differences = []
n_samples = len(y_true)
for _ in range(n_boot):
indices = rng.integers(0, n_samples, n_samples)
differences.append(
_metric(y_true[indices], predictions_a[indices])
- _metric(y_true[indices], predictions_b[indices])
)
values = np.asarray(differences)
return {
"mean_difference": float(values.mean()),
"lower": float(np.percentile(values, 2.5)),
"upper": float(np.percentile(values, 97.5)),
"n_boot": int(n_boot),
}
def cochran_q(correctness_matrix: np.ndarray) -> dict[str, float]:
"""Критерий Кокрана для сравнения нескольких бинарных классификаторов.
``correctness_matrix`` имеет форму (объекты, модели), значения 0/1.
"""
matrix = np.asarray(correctness_matrix, dtype=float)
if matrix.ndim != 2:
raise ValueError("Ожидается матрица (объекты, модели).")
k = matrix.shape[1]
column_sums = matrix.sum(axis=0)
row_sums = matrix.sum(axis=1)
total = matrix.sum()
numerator = (k - 1) * (k * np.sum(column_sums**2) - total**2)
denominator = k * total - np.sum(row_sums**2)
statistic = 0.0 if denominator == 0 else float(numerator / denominator)
from scipy.stats import chi2
p_value = float(chi2.sf(statistic, df=k - 1))
return {"statistic": statistic, "p_value": p_value, "df": k - 1}

View File

@@ -0,0 +1,363 @@
"""Токенизация, разбиение на предложения и частеречная разметка.
По умолчанию используется NLTK: ``word_tokenize``, ``sent_tokenize`` и
``pos_tag``. Если ресурсы NLTK недоступны, включается упрощённый резервный
режим на регулярных выражениях, а POS-признаки отключаются. Это позволяет
запускать аудит и часть признаков даже без скачанных моделей.
"""
from __future__ import annotations
import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Sequence
import pandas as pd
from .lexicons import (
AUXILIARIES,
FUNCTION_WORDS,
MODALS,
NEGATIONS,
PERSONAL_PRONOUNS,
SUBORDINATORS,
STOPWORD_FALLBACK,
)
DEFAULT_NLTK_DIR = Path(__file__).resolve().parents[1] / "nltk_data"
_WORD_RE = re.compile(r"[A-Za-z]+(?:['’][A-Za-z]+)?|\d+|[^\w\s]")
_SENTENCE_RE = re.compile(r"(?<=[.!?])\s+")
_PUNCT_TAGS = {".", ",", ":", "``", "''", "-LRB-", "-RRB-", "HYPH", "NFP"}
_PTB_TAGS = [
"CC", "CD", "DT", "EX", "FW", "IN", "JJ", "JJR", "JJS", "LS", "MD",
"NN", "NNS", "NNP", "NNPS", "PDT", "POS", "PRP", "PRP$", "RB",
"RBR", "RBS", "RP", "SYM", "TO", "UH", "VB", "VBD", "VBG", "VBN",
"VBP", "VBZ", "WDT", "WP", "WP$", "WRB",
]
UPOS_TAGS = [
"ADJ", "ADP", "ADV", "AUX", "CONJ", "DET", "INTJ", "NOUN", "NUM",
"PRON", "PRT", "PUNCT", "SCONJ", "SYM", "VERB", "X",
]
_AUX_WORDS = AUXILIARIES | MODALS | {"be", "been", "being", "am", "is", "are", "was", "were"}
_NEGATION_WORDS = NEGATIONS | {"not", "n't"}
def _check_nltk_resource(resource: str) -> bool:
try:
import nltk
nltk.data.find(resource)
return True
except Exception:
return False
def ensure_nltk_data(
nltk_data_dir: str | Path | None = None,
download: bool = False,
) -> bool:
"""Добавляет локальный каталог NLTK в путь поиска и проверяет ресурсы.
Возвращает ``True``, если доступны и токенизатор, и POS-теггер.
"""
import nltk
directory = Path(nltk_data_dir) if nltk_data_dir is not None else DEFAULT_NLTK_DIR
if directory.exists():
path = str(directory.resolve())
if path not in nltk.data.path:
nltk.data.path.insert(0, path)
required = {
"tokenizer": "tokenizers/punkt_tab",
"tagger": "taggers/averaged_perceptron_tagger_eng",
}
if download:
for package in (
"punkt",
"punkt_tab",
"averaged_perceptron_tagger",
"averaged_perceptron_tagger_eng",
"stopwords",
):
nltk.download(package, download_dir=str(directory), quiet=True)
return all(_check_nltk_resource(resource) for resource in required.values())
def load_stopwords(nltk_data_dir: str | Path | None = None) -> set[str]:
"""Загружает английские стоп-слова NLTK или использует запасной список."""
try:
import nltk
directory = Path(nltk_data_dir) if nltk_data_dir is not None else DEFAULT_NLTK_DIR
if directory.exists():
path = str(directory.resolve())
if path not in nltk.data.path:
nltk.data.path.insert(0, path)
return set(nltk.corpus.stopwords.words("english"))
except Exception:
return set(STOPWORD_FALLBACK)
def penn_to_upos(token: str, tag: str) -> str:
"""Переводит тег Penn Treebank в Universal POS.
Часть решений зависит не только от тега, но и от самого слова: например,
``do`` может быть смысловым глаголом или вспомогательным, а ``as`` —
предлогом или подчинительным союзом.
"""
lower = token.lower()
if tag in _PUNCT_TAGS or token in {".", ",", "!", "?", ";", ":", "...", "…"}:
return "PUNCT"
if tag == "SYM":
return "SYM"
if tag == "CC":
return "CONJ"
if tag == "IN":
return "SCONJ" if lower in SUBORDINATORS else "ADP"
if tag == "TO":
return "PRT"
if tag == "RP" or tag == "POS":
return "PRT"
if tag == "MD":
return "AUX"
if tag == "EX":
return "PRON"
if tag in {"WDT", "WP", "WP$"}:
return "PRON"
if tag == "WRB":
return "ADV"
if tag == "UH":
return "INTJ"
if tag == "CD":
return "NUM"
if tag in {"DT", "PDT"}:
return "DET"
if tag in {"PRP", "PRP$"}:
return "PRON"
if tag.startswith("JJ"):
return "ADJ"
if tag.startswith("RB"):
return "ADV"
if tag.startswith("NN"):
return "NOUN"
if tag.startswith("VB"):
if lower in _AUX_WORDS:
return "AUX"
return "VERB"
if tag == "FW":
return "X"
if tag == "LS":
return "X"
# Уточнения по слову для распространённых разговорных форм.
if lower in _NEGATION_WORDS:
return "PRT"
if lower in _AUX_WORDS:
return "AUX"
if lower in PERSONAL_PRONOUNS:
return "PRON"
return "X"
def collapse_punctuation(tags: Sequence[str]) -> list[str]:
"""Схлопывает подряд идущие знаки препинания в один тег PUNCT."""
result: list[str] = []
for tag in tags:
if tag == "PUNCT" and result and result[-1] == "PUNCT":
continue
result.append(tag)
return result
def is_function_word(token: str, upos: str | None = None) -> bool:
"""Определяет, относится ли слово к закрытому классу."""
lower = token.lower()
if lower in FUNCTION_WORDS:
return True
if upos in {"ADP", "AUX", "CONJ", "DET", "PRON", "PRT", "SCONJ"}:
return True
return False
def is_content_word(token: str, upos: str | None = None, stopwords: set[str] | None = None) -> bool:
"""Определяет, является ли слово содержательным для content-LSA."""
lower = token.lower()
if not re.search(r"[A-Za-z]", token):
return False
if is_function_word(token, upos):
return False
if stopwords and lower in stopwords:
return False
return True
@dataclass
class AnalyzedLine:
"""Результат токенизации и разметки одной реплики."""
tokens: list[str] = field(default_factory=list)
ptb: list[str] = field(default_factory=list)
upos: list[str] = field(default_factory=list)
sentences: list[list[str]] = field(default_factory=list)
pos_text: str = ""
fw_text: str = ""
content_text: str = ""
class Tokenizer:
"""Токенизатор с NLTK и резервным режимом."""
def __init__(
self,
use_nltk: bool = True,
nltk_data_dir: str | Path | None = None,
strict: bool = False,
) -> None:
self.use_nltk = bool(use_nltk)
self.nltk_data_dir = Path(nltk_data_dir) if nltk_data_dir is not None else DEFAULT_NLTK_DIR
self.strict = strict
self.stopwords = load_stopwords(self.nltk_data_dir)
if self.use_nltk:
available = ensure_nltk_data(self.nltk_data_dir)
if not available:
message = (
"Ресурсы NLTK не найдены. Скачайте punkt_tab и "
"averaged_perceptron_tagger_eng или запустите с --no-nltk."
)
if self.strict:
raise RuntimeError(message)
print(f"[tokenization] {message} Включаю резервный режим.")
self.use_nltk = False
self._pos_available = self.use_nltk and _check_nltk_resource(
"taggers/averaged_perceptron_tagger_eng"
)
self._word_tokenize = None
self._sent_tokenize = None
self._pos_tag = None
if self.use_nltk:
from nltk import pos_tag
from nltk.tokenize import sent_tokenize, word_tokenize
self._word_tokenize = word_tokenize
self._sent_tokenize = sent_tokenize
self._pos_tag = pos_tag
@property
def pos_available(self) -> bool:
return self._pos_available
def tokenize(self, text: str) -> list[str]:
if self._word_tokenize is not None:
try:
return [token for token in self._word_tokenize(text) if token.strip()]
except Exception:
pass
return _WORD_RE.findall(text)
def sentence_tokenize(self, text: str) -> list[str]:
if self._sent_tokenize is not None:
try:
sentences = [sentence.strip() for sentence in self._sent_tokenize(text)]
return [sentence for sentence in sentences if sentence]
except Exception:
pass
return [part.strip() for part in _SENTENCE_RE.split(text) if part.strip()]
def pos_tag(self, tokens: Sequence[str]) -> list[tuple[str, str]]:
if self._pos_tag is not None:
try:
return list(self._pos_tag(list(tokens)))
except Exception:
pass
return [(token, "X") for token in tokens]
def analyze(self, text: str) -> AnalyzedLine:
tokens = self.tokenize(text)
tagged = self.pos_tag(tokens)
ptb = [tag for _, tag in tagged]
upos = [penn_to_upos(token, tag) for token, tag in tagged]
sentences: list[list[str]] = []
for sentence in self.sentence_tokenize(text):
sentence_tokens = self.tokenize(sentence)
if sentence_tokens:
sentences.append(sentence_tokens)
if not sentences and tokens:
sentences = [tokens]
pos_sequence = collapse_punctuation(upos)
pos_text = " ".join(pos_sequence) if pos_sequence else "__EMPTY__"
fw_tokens = [
token
for token, tag in zip(tokens, upos)
if is_function_word(token, tag)
]
fw_text = " ".join(fw_tokens) if fw_tokens else "__EMPTY__"
content_tokens = [
token
for token, tag in zip(tokens, upos)
if is_content_word(token, tag, self.stopwords)
]
content_text = " ".join(content_tokens) if content_tokens else "__EMPTY__"
return AnalyzedLine(
tokens=tokens,
ptb=ptb,
upos=upos,
sentences=sentences,
pos_text=pos_text,
fw_text=fw_text,
content_text=content_text,
)
def analyze_frame(
self,
frame: pd.DataFrame,
text_column: str = "text",
verbose: bool = False,
) -> pd.DataFrame:
"""Добавляет к таблице колонки с токенами, тегами и текстами для LSA."""
if text_column not in frame.columns:
raise KeyError(f"Нет колонки {text_column!r}")
records: list[dict[str, object]] = []
total = len(frame)
for position, (_, row) in enumerate(frame.iterrows(), start=1):
analyzed = self.analyze(str(row[text_column]))
records.append(
{
"tokens": analyzed.tokens,
"ptb": analyzed.ptb,
"upos": analyzed.upos,
"sentences": analyzed.sentences,
"pos_text": analyzed.pos_text,
"fw_text": analyzed.fw_text,
"content_text": analyzed.content_text,
}
)
if verbose and position % 500 == 0:
print(f"[tokenization] обработано {position} из {total}")
additions = pd.DataFrame(records, index=frame.index)
result = pd.concat([frame.copy(), additions], axis=1)
return result
def upos_tag_list() -> list[str]:
"""Список универсальных частеречных тегов, используемых в признаках."""
return list(UPOS_TAGS)
def ptb_tag_list() -> list[str]:
"""Список тегов Penn Treebank, используемых в признаках."""
return list(_PTB_TAGS)

View File

@@ -0,0 +1,129 @@
"""Верификация «тот же персонаж / другой персонаж» по центроидам классов."""
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)