"""Командный интерфейс для аудита, признаков, 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", "groups"): 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) model_params = dict(config_data.get("model_params", {})) if args.model_params: try: extra_params = json.loads(args.model_params) except json.JSONDecodeError as error: raise ValueError(f"Некорректный JSON в --model-params: {error}") from error if not isinstance(extra_params, dict): raise ValueError("--model-params должен быть JSON-объектом.") model_params.update(extra_params) pipeline = build_pipeline(feature_config, model, **model_params) 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__), "model_params": model_params, "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, block_weights=dict(base.block_weights), 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( "--model-params", default=None, help='JSON с параметрами модели, например \'{"C": 0.3, "class_weight": "balanced"}\'', ) 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())