"""Дополнительные эксперименты для итогового отчёта. Стадии: - nested: вложенный выбор конфигурации признаков и модели; - topic: контроль содержания через маскирование содержательных слов; - calibration: коррекция априорных вероятностей классов; - ensemble: мягкое голосование и stacking; - extra: дополнительные конфигурации и отложенная выборка. Запуск: python scripts/run_final_experiments.py --stages nested topic calibration ensemble extra """ from __future__ import annotations import argparse import json import re import sys from dataclasses import replace from pathlib import Path from typing import Any, Iterable, Sequence import numpy as np import pandas as pd from sklearn.ensemble import VotingClassifier, StackingClassifier from sklearn.linear_model import LogisticRegression from sklearn.metrics import accuracy_score, confusion_matrix, f1_score from sklearn.model_selection import StratifiedKFold, train_test_split PROJECT_ROOT = Path(__file__).resolve().parents[1] if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) from stylometry.cv import run_repeated_cv from stylometry.datasets import load_feature_frame from stylometry.evaluation import ( bootstrap_ci, classification_metrics, confusion_matrix_frame, per_class_report, top_confusions, ) from stylometry.pipelines import ( FeatureConfig, build_pipeline, preset_config, ) DEFAULT_OUT = PROJECT_ROOT / "results" / "final" def save_json(data: Any, path: Path) -> None: 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=str) def save_cv(result, directory: Path, title: str, config: dict[str, Any]) -> None: directory.mkdir(parents=True, exist_ok=True) result.fold_metrics.to_csv(directory / "fold_metrics.csv", index=False, encoding="utf-8-sig") result.oof.to_csv(directory / "oof_predictions.csv", index=False, encoding="utf-8-sig") pooled, bootstrap = result.pooled_summary() pooled.to_csv(directory / "pooled_metrics.csv", index=False, encoding="utf-8-sig") save_json(bootstrap, directory / "bootstrap_ci.json") first = result.oof[result.oof["repeat"] == result.oof["repeat"].min()] per_class_report(first["y_true"], first["y_pred"], result.labels).to_csv( directory / "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(directory / "confusion_matrix.csv", encoding="utf-8-sig") top_confusions(first["y_true"], first["y_pred"], result.labels).to_csv( directory / "top_confusions.csv", index=False, encoding="utf-8-sig" ) summary = { "title": title, "config": config, "fold_mean_macro_f1": float(result.fold_metrics["macro_f1"].mean()), "fold_std_macro_f1": float(result.fold_metrics["macro_f1"].std(ddof=1)), "n_splits": result.n_splits, "n_repeats": result.n_repeats, "bootstrap": bootstrap, } save_json(summary, directory / "summary.json") (directory / "summary.md").write_text( f"# {title}\n\n" f"Macro-F1 по фолдам: {summary['fold_mean_macro_f1']:.4f} " f"± {summary['fold_std_macro_f1']:.4f}\n\n" f"Pooled:\n\n{pooled.to_string(index=False)}\n", encoding="utf-8", ) def lean_config(base: FeatureConfig, lsa_components: int = 300) -> FeatureConfig: groups = tuple( group for group in base.groups if group not in {"function_words", "pos", "vocabulary", "readability"} ) return replace( base, groups=groups, lsa_variants=("content",), lsa_components=lsa_components, use_char_ngrams=True, use_pos_ngrams=True, include_lsa=True, ) def masked_frame(frame: pd.DataFrame) -> pd.DataFrame: """Заменяет содержательные слова на CONTENT, сохраняя служебные слова.""" content_tags = {"NOUN", "VERB", "ADJ", "ADV", "NUM"} def _mask_row(row: pd.Series) -> str: tokens = list(row["tokens"]) tags = list(row["upos"]) result = [] for token, tag in zip(tokens, tags): if tag in content_tags and re.search(r"[A-Za-z]", str(token)): result.append("CONTENT") else: result.append(str(token)) return " ".join(result) if result else "__EMPTY__" result = frame.copy() result["masked_text"] = result.apply(_mask_row, axis=1) result["text"] = result["masked_text"] return result # --------------------------------------------------------------------------- # Nested selection # --------------------------------------------------------------------------- def nested_selection( frame: pd.DataFrame, y: np.ndarray, out_dir: Path, folds: int = 5, inner_folds: int = 3, repeats: int = 1, seed: int = 42, ) -> None: base = preset_config("stylometry_pos_lsa") candidates: list[tuple[str, FeatureConfig, str]] = [ ("et_full", base, "extra_trees"), ("et_no_lsa", preset_config("stylometry_pos"), "extra_trees"), ("et_lean_content300", lean_config(base, 300), "extra_trees"), ("logreg_lean_content300", lean_config(base, 300), "logreg"), ("rf_char10000", replace(base, char_max_features=10000), "random_forest"), ("rf_char15000", replace(base, char_max_features=15000), "random_forest"), ("et_char10000", replace(base, char_max_features=10000), "extra_trees"), ("et_lean_content400", lean_config(base, 400), "extra_trees"), ] labels = np.unique(y) rows: list[dict[str, Any]] = [] oof_rows: list[dict[str, Any]] = [] selection_rows: list[dict[str, Any]] = [] for repeat in range(repeats): outer = StratifiedKFold(n_splits=folds, shuffle=True, random_state=seed + repeat) for fold, (train_idx, test_idx) in enumerate(outer.split(frame, y), start=1): print(f"[nested] repeat={repeat} fold={fold}: inner selection", flush=True) inner = StratifiedKFold( n_splits=inner_folds, shuffle=True, random_state=seed + repeat * 100 + fold, ) inner_scores: dict[str, list[float]] = {name: [] for name, _, _ in candidates} for inner_train, inner_test in inner.split(frame.iloc[train_idx], y[train_idx]): for name, config, model_name in candidates: pipeline = build_pipeline(config, model_name) pipeline.fit( frame.iloc[train_idx].iloc[inner_train], y[train_idx][inner_train], ) prediction = pipeline.predict( frame.iloc[train_idx].iloc[inner_test] ) inner_scores[name].append( f1_score( y[train_idx][inner_test], prediction, labels=labels, average="macro", zero_division=0, ) ) means = { name: float(np.mean(values)) for name, values in inner_scores.items() } best_name = max(means, key=means.get) best_config, best_model = next( (config, model) for name, config, model in candidates if name == best_name ) selection_rows.append( { "repeat": repeat, "fold": fold, "selected": best_name, "inner_mean": means[best_name], **{f"inner_{name}": value for name, value in means.items()}, } ) pipeline = build_pipeline(best_config, best_model) pipeline.fit(frame.iloc[train_idx], y[train_idx]) prediction = pipeline.predict(frame.iloc[test_idx]) metrics = classification_metrics( y[test_idx], prediction, labels=labels ) metrics.update({"repeat": repeat, "fold": fold, "selected": best_name}) rows.append(metrics) for sample_index, predicted in zip(test_idx, prediction): oof_rows.append( { "repeat": repeat, "fold": fold, "line_id": int(frame.iloc[sample_index]["line_id"]), "y_true": y[sample_index], "y_pred": predicted, "selected": best_name, } ) fold_metrics = pd.DataFrame(rows) oof = pd.DataFrame(oof_rows).sort_values(["repeat", "line_id"]).reset_index(drop=True) out_dir.mkdir(parents=True, exist_ok=True) fold_metrics.to_csv(out_dir / "fold_metrics.csv", index=False, encoding="utf-8-sig") oof.to_csv(out_dir / "oof_predictions.csv", index=False, encoding="utf-8-sig") pd.DataFrame(selection_rows).to_csv( out_dir / "selection.csv", index=False, encoding="utf-8-sig" ) first = oof[oof["repeat"] == oof["repeat"].min()] per_class_report(first["y_true"], first["y_pred"], labels).to_csv( out_dir / "per_class_report.csv", index=False, encoding="utf-8-sig" ) confusion_matrix_frame(first["y_true"], first["y_pred"], labels).to_csv( out_dir / "confusion_matrix.csv", encoding="utf-8-sig" ) pooled = [] for repeat, group in oof.groupby("repeat"): metrics = classification_metrics(group["y_true"], group["y_pred"], labels=labels) metrics["repeat"] = repeat pooled.append(metrics) pooled_frame = pd.DataFrame(pooled) pooled_frame.to_csv(out_dir / "pooled_metrics.csv", index=False, encoding="utf-8-sig") save_json( { "fold_mean_macro_f1": float(fold_metrics["macro_f1"].mean()), "fold_std_macro_f1": float(fold_metrics["macro_f1"].std(ddof=1)), "pooled": pooled_frame.to_dict(orient="records"), "selection_counts": pd.DataFrame(selection_rows)["selected"].value_counts().to_dict(), }, out_dir / "summary.json", ) print( f"[nested] macro-F1={fold_metrics['macro_f1'].mean():.4f} " f"± {fold_metrics['macro_f1'].std(ddof=1):.4f}" ) # --------------------------------------------------------------------------- # Topic control # --------------------------------------------------------------------------- def topic_control( frame: pd.DataFrame, y: np.ndarray, out_dir: Path, folds: int = 5, repeats: int = 3, seed: int = 42, ) -> None: style_groups = ( "length", "punctuation", "function_words", "pronouns", "pos", "syntax", "dialogue", ) config = FeatureConfig( groups=style_groups, use_char_ngrams=True, char_ngram_range=(2, 5), char_max_features=5000, use_pos_ngrams=True, pos_ngram_range=(1, 3), pos_max_features=500, lsa_variants=("fw", "pos"), lsa_components=100, stylo_normalize=True, include_stylo=True, include_lsa=True, ) masked = masked_frame(frame) for name, data in (("original", frame), ("masked", masked)): for model in ("extra_trees", "logreg"): print(f"[topic] {name} / {model}", flush=True) pipeline = build_pipeline(config, model) result = run_repeated_cv( pipeline, data, y, n_splits=folds, n_repeats=repeats, seed=seed, model_name=model, verbose=False, ) save_cv( result, out_dir / f"{name}_{model}", f"Topic control: {name} / {model}", {"condition": name, "model": model, "groups": style_groups}, ) # --------------------------------------------------------------------------- # Calibration # --------------------------------------------------------------------------- def _prior_adjust(scores: np.ndarray, priors: np.ndarray, alpha: float) -> np.ndarray: adjusted = scores / np.power(np.maximum(priors, 1e-12), alpha)[None, :] row_sums = adjusted.sum(axis=1, keepdims=True) return adjusted / np.maximum(row_sums, 1e-12) def calibration( frame: pd.DataFrame, y: np.ndarray, out_dir: Path, folds: int = 5, repeats: int = 5, seed: int = 42, ) -> None: config = preset_config("stylometry_pos_lsa") pipeline = build_pipeline(config, "extra_trees") print("[calibration] base CV", flush=True) result = run_repeated_cv( pipeline, frame, y, n_splits=folds, n_repeats=repeats, seed=seed, model_name="extra_trees", verbose=False, ) save_cv( result, out_dir / "base_extra_trees", "Calibration base: extra_trees", {"model": "extra_trees", "preset": "stylometry_pos_lsa"}, ) labels = result.labels label_to_index = {label: index for index, label in enumerate(labels)} priors = np.bincount( np.array([label_to_index[value] for value in y]), minlength=len(labels) ).astype(float) priors /= priors.sum() alphas = [0.0, 0.25, 0.5, 0.75, 1.0] matrix: dict[int, np.ndarray] = {} truths: dict[int, np.ndarray] = {} for repeat, group in result.oof.groupby("repeat"): scores = result.scores_by_repeat.get(int(repeat)) if scores is None or scores.shape[0] != len(group): continue matrix[int(repeat)] = scores truths[int(repeat)] = group["y_true"].to_numpy() rows = [] for repeat in sorted(matrix): other = [value for value in matrix if value != repeat] table: dict[float, float] = {} for alpha in alphas: values = [] for other_repeat in other: prediction_indices = np.argmax( _prior_adjust(matrix[other_repeat], priors, alpha), axis=1 ) values.append( f1_score( truths[other_repeat], labels[prediction_indices], labels=labels, average="macro", zero_division=0, ) ) table[alpha] = float(np.mean(values)) best_alpha = max(table, key=table.get) baseline_pred = labels[np.argmax(matrix[repeat], axis=1)] corrected_pred = labels[ np.argmax(_prior_adjust(matrix[repeat], priors, best_alpha), axis=1) ] baseline = f1_score( truths[repeat], baseline_pred, labels=labels, average="macro", zero_division=0 ) corrected = f1_score( truths[repeat], corrected_pred, labels=labels, average="macro", zero_division=0 ) rows.append( { "repeat": repeat, "selected_alpha": best_alpha, "baseline_macro_f1": baseline, "corrected_macro_f1": corrected, "delta": corrected - baseline, **{f"alpha_{alpha}": table[alpha] for alpha in alphas}, } ) table = pd.DataFrame(rows) out_dir.mkdir(parents=True, exist_ok=True) table.to_csv(out_dir / "calibration.csv", index=False, encoding="utf-8-sig") save_json( { "mean_baseline": float(table["baseline_macro_f1"].mean()), "mean_corrected": float(table["corrected_macro_f1"].mean()), "mean_delta": float(table["delta"].mean()), "selected_alpha": table["selected_alpha"].tolist(), }, out_dir / "summary.json", ) print(table.to_string(index=False)) # --------------------------------------------------------------------------- # Ensemble # --------------------------------------------------------------------------- def _base_estimators() -> list[tuple[str, Any]]: base = preset_config("stylometry_pos_lsa") return [ ("et_full", build_pipeline(base, "extra_trees")), ( "rf_char10000", build_pipeline(replace(base, char_max_features=10000), "random_forest"), ), ( "logreg_lean_content300", build_pipeline(lean_config(base, 300), "logreg"), ), ] def ensemble( frame: pd.DataFrame, y: np.ndarray, out_dir: Path, folds: int = 5, repeats: int = 3, seed: int = 42, with_stacking: bool = True, ) -> None: estimators = _base_estimators() voting = VotingClassifier(estimators=estimators, voting="soft", n_jobs=1) print("[ensemble] soft voting", flush=True) result = run_repeated_cv( voting, frame, y, n_splits=folds, n_repeats=repeats, seed=seed, model_name="soft_voting", verbose=False, ) save_cv( result, out_dir / "soft_voting", "Ensemble: soft voting", {"estimators": [name for name, _ in estimators], "voting": "soft"}, ) if with_stacking: print("[ensemble] stacking", flush=True) stacking = StackingClassifier( estimators=estimators, final_estimator=LogisticRegression( max_iter=3000, class_weight="balanced" ), stack_method="predict_proba", cv=3, n_jobs=1, passthrough=False, ) stacked = run_repeated_cv( stacking, frame, y, n_splits=folds, n_repeats=1, seed=seed, model_name="stacking", verbose=False, ) save_cv( stacked, out_dir / "stacking", "Ensemble: stacking", {"estimators": [name for name, _ in estimators], "meta": "logreg"}, ) # --------------------------------------------------------------------------- # Extra configurations # --------------------------------------------------------------------------- def extra( frame: pd.DataFrame, y: np.ndarray, out_dir: Path, folds: int = 5, repeats: int = 3, seed: int = 42, ) -> None: base = preset_config("stylometry_pos_lsa") configs: list[tuple[str, FeatureConfig, str]] = [ ("rf_char15000", replace(base, char_max_features=15000), "random_forest"), ("logreg_lean_content400", lean_config(base, 400), "logreg"), ("logreg_lean_content500", lean_config(base, 500), "logreg"), ( "et_block_weights", replace( base, block_weights={ "stylo": 1.0, "char": 0.7, "pos": 0.7, "lsa_word": 0.3, "lsa_pos": 0.3, }, ), "extra_trees", ), ] for name, config, model in configs: print(f"[extra] {name}", flush=True) pipeline = build_pipeline(config, model) result = run_repeated_cv( pipeline, frame, y, n_splits=folds, n_repeats=repeats, seed=seed, model_name=model, verbose=False, ) save_cv(result, out_dir / name, f"Extra: {name}", {"model": model}) # Отложенная выборка: одно стратифицированное разбиение 80/20. train_idx, test_idx = train_test_split( np.arange(len(frame)), test_size=0.2, stratify=y, random_state=seed ) holdout_rows = [] for name, config, model in ( ("et_full_holdout", base, "extra_trees"), ( "rf_char10000_holdout", replace(base, char_max_features=10000), "random_forest", ), ("logreg_lean_content300_holdout", lean_config(base, 300), "logreg"), ): print(f"[extra] holdout {name}", flush=True) pipeline = build_pipeline(config, model) pipeline.fit(frame.iloc[train_idx], y[train_idx]) prediction = pipeline.predict(frame.iloc[test_idx]) metrics = classification_metrics(y[test_idx], prediction, labels=np.unique(y)) metrics["name"] = name metrics["train_n"] = len(train_idx) metrics["test_n"] = len(test_idx) holdout_rows.append(metrics) pd.DataFrame(holdout_rows).to_csv( out_dir / "holdout_metrics.csv", index=False, encoding="utf-8-sig" ) # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def main() -> int: parser = argparse.ArgumentParser() parser.add_argument( "--stages", nargs="+", default=["nested", "topic", "calibration", "ensemble", "extra"], choices=["nested", "topic", "calibration", "ensemble", "extra"], ) parser.add_argument("--out", default=str(DEFAULT_OUT)) parser.add_argument("--folds", type=int, default=5) parser.add_argument("--repeats", type=int, default=3) parser.add_argument("--seed", type=int, default=42) parser.add_argument("--no-stacking", action="store_true") args = parser.parse_args() frame = load_feature_frame(verbose=True) y = frame["character"].to_numpy() output = Path(args.out) output.mkdir(parents=True, exist_ok=True) if "nested" in args.stages: nested_selection( frame, y, output / "nested_selection", folds=args.folds, inner_folds=3, repeats=args.repeats, seed=args.seed, ) if "topic" in args.stages: topic_control( frame, y, output / "topic_control", folds=args.folds, repeats=args.repeats, seed=args.seed, ) if "calibration" in args.stages: calibration( frame, y, output / "calibration", folds=args.folds, repeats=5, seed=args.seed, ) if "ensemble" in args.stages: ensemble( frame, y, output / "ensemble", folds=args.folds, repeats=args.repeats, seed=args.seed, with_stacking=not args.no_stacking, ) if "extra" in args.stages: extra( frame, y, output / "extra", folds=args.folds, repeats=args.repeats, seed=args.seed, ) return 0 if __name__ == "__main__": raise SystemExit(main())