"""Legacy robustness experiment retained for reusable masking utilities. EXPERIMENT STATUS: SUPERSEDED. The model evaluated by this script is an older baseline and its severity metrics do not describe ``final_pipeline.py``. Use ``ml_polish_benchmark.py`` for final clean and missing-value results. Legacy architecture from ``benchmark_grouped.py``: * label: combined raw + engine-relative features with Logistic Regression, * severity: raw features with Extra Trees. For every cross-validation seed, both clean and artificially masked validation engines are evaluated with exactly the same fitted fold models. Training rows remain clean. This isolates the inference-time effect of missing measurements, which are present in test.csv but absent from val.csv. """ from __future__ import annotations import argparse import json from pathlib import Path import numpy as np import pandas as pd from sklearn.metrics import accuracy_score from benchmark_grouped import ( FAULT_LABELS, FREQ_COLS, NOT_APPLICABLE, make_pipeline, make_splits, macro_f1, ml_points, raw_score, validate_data, ) DEFAULT_SEEDS = [7, 21, 42, 77, 123] SCENARIO_CLEAN = "clean" def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--data", type=Path, default=Path("val.csv")) parser.add_argument( "--output-dir", type=Path, default=Path("robustness_outputs") ) parser.add_argument("--seeds", nargs="+", type=int, default=DEFAULT_SEEDS) parser.add_argument("--n-splits", type=int, default=5) parser.add_argument("--missing-rate", type=float, default=0.05) parser.add_argument("--mask-random-state", type=int, default=2026) parser.add_argument("--model-random-state", type=int, default=42) parser.add_argument( "--n-jobs", type=int, default=-1, help="Parallel workers for Extra Trees; -1 uses all CPU cores.", ) return parser.parse_args() def missing_scenario_name(missing_rate: float) -> str: percent = 100.0 * missing_rate formatted = f"{percent:.2f}".rstrip("0").rstrip(".").replace(".", "p") return f"masked_{formatted}pct" def mask_spectral_cells( df: pd.DataFrame, missing_rate: float, random_state: int, ) -> tuple[pd.DataFrame, pd.DataFrame]: """Mask an exact fraction of currently observed spectral cells. Returns a masked copy and a manifest containing every newly masked cell. Metadata and labels are never modified. """ if not 0.0 <= missing_rate < 1.0: raise ValueError("missing_rate must satisfy 0 <= missing_rate < 1") missing_columns = sorted(set(FREQ_COLS).difference(df.columns)) if missing_columns: raise ValueError(f"Missing spectral columns: {missing_columns}") masked = df.copy(deep=True) spectra = masked[FREQ_COLS].to_numpy(dtype=float, copy=True) observed_positions = np.flatnonzero(~np.isnan(spectra.ravel())) n_to_mask = int(round(missing_rate * len(observed_positions))) rng = np.random.default_rng(random_state) selected = np.sort( rng.choice(observed_positions, size=n_to_mask, replace=False) ) # np.put mutates the original array even if its memory layout is not # contiguous; chained ``ravel()[selected]`` may silently modify a copy. np.put(spectra, selected, np.nan) masked.loc[:, FREQ_COLS] = spectra row_positions, col_positions = np.unravel_index(selected, spectra.shape) manifest = pd.DataFrame( { "row_index": row_positions, "engine_id": df.iloc[row_positions]["engine_id"].to_numpy(), "cylinder": df.iloc[row_positions]["cylinder"].to_numpy(), "frequency_column": np.asarray(FREQ_COLS, dtype=object)[col_positions], } ) return masked, manifest def evaluate_seed( df_clean: pd.DataFrame, df_masked: pd.DataFrame, cv_seed: int, n_splits: int, model_random_state: int, n_jobs: int, output_dir: Path, masked_scenario: str, ) -> tuple[list[dict[str, object]], list[dict[str, object]]]: splits = make_splits( df_clean, splitter_name="stratified-group", n_splits=n_splits, random_state=cv_seed, ) y_label = df_clean["label"].reset_index(drop=True) y_severity = df_clean["severity"].reset_index(drop=True) true_fault = y_label.isin(FAULT_LABELS).to_numpy() scenarios = { SCENARIO_CLEAN: df_clean, masked_scenario: df_masked, } oof = { scenario: { "label": np.full(len(df_clean), "", dtype=object), "severity": np.full(len(df_clean), NOT_APPLICABLE, dtype=object), "fold": np.full(len(df_clean), -1, dtype=int), } for scenario in scenarios } fold_rows: list[dict[str, object]] = [] for fold, (train_idx, valid_idx) in enumerate(splits, start=1): X_train = df_clean.iloc[train_idx] y_train_label = y_label.iloc[train_idx] label_pipeline = make_pipeline( model_name="logistic", feature_set="combined", random_state=model_random_state + fold, n_jobs=n_jobs, ) label_pipeline.fit(X_train, y_train_label) severity_train_mask = y_train_label.isin(FAULT_LABELS).to_numpy() severity_pipeline = make_pipeline( model_name="extra_trees", feature_set="raw", random_state=model_random_state + 100 + fold, n_jobs=n_jobs, ) severity_pipeline.fit( X_train.iloc[np.flatnonzero(severity_train_mask)], y_severity.iloc[train_idx].iloc[np.flatnonzero(severity_train_mask)], ) fold_true_label = y_label.iloc[valid_idx] fold_true_fault = fold_true_label.isin(FAULT_LABELS).to_numpy() for scenario, scenario_df in scenarios.items(): X_valid = scenario_df.iloc[valid_idx] label_pred = label_pipeline.predict(X_valid) severity_pred = severity_pipeline.predict(X_valid) emitted = np.full(len(valid_idx), NOT_APPLICABLE, dtype=object) predicted_fault = np.isin(label_pred, FAULT_LABELS) emitted[predicted_fault] = severity_pred[predicted_fault] oof[scenario]["label"][valid_idx] = label_pred oof[scenario]["severity"][valid_idx] = emitted oof[scenario]["fold"][valid_idx] = fold fold_f1 = macro_f1(fold_true_label, label_pred) fold_severity = float( accuracy_score( y_severity.iloc[valid_idx].to_numpy()[fold_true_fault], emitted[fold_true_fault], ) ) fold_score = raw_score(fold_f1, fold_severity) added_missing = int( ( df_clean.iloc[valid_idx][FREQ_COLS].notna() & scenario_df.iloc[valid_idx][FREQ_COLS].isna() ) .to_numpy() .sum() ) fold_rows.append( { "cv_seed": cv_seed, "fold": fold, "scenario": scenario, "valid_engines": len( set(df_clean.iloc[valid_idx]["engine_id"]) ), "valid_rows": len(valid_idx), "valid_fault_rows": int(fold_true_fault.sum()), "added_missing_cells": added_missing, "macro_f1": fold_f1, "severity_accuracy_submission": fold_severity, "raw_score": fold_score, "ml_points": ml_points(fold_score), } ) run_rows: list[dict[str, object]] = [] for scenario in scenarios: label_pred = oof[scenario]["label"] severity_pred = oof[scenario]["severity"] if (label_pred == "").any() or (oof[scenario]["fold"] < 1).any(): raise AssertionError("Incomplete OOF predictions.") label_f1 = macro_f1(y_label, label_pred) severity_accuracy = float( accuracy_score( y_severity.to_numpy()[true_fault], severity_pred[true_fault] ) ) score = raw_score(label_f1, severity_accuracy) scenario_fold_rows = [ row for row in fold_rows if row["cv_seed"] == cv_seed and row["scenario"] == scenario ] run_rows.append( { "cv_seed": cv_seed, "scenario": scenario, "macro_f1_oof": label_f1, "severity_accuracy_submission_oof": severity_accuracy, "raw_score_oof": score, "ml_points_oof": ml_points(score), "macro_f1_fold_std": float( np.std([row["macro_f1"] for row in scenario_fold_rows], ddof=1) ), "raw_score_fold_std": float( np.std([row["raw_score"] for row in scenario_fold_rows], ddof=1) ), } ) oof_frame = df_clean[["engine_id", "cylinder", "label", "severity"]].copy() oof_frame["fold"] = oof[scenario]["fold"] oof_frame["predicted_label"] = label_pred oof_frame["predicted_severity_submission"] = severity_pred oof_frame.to_csv( output_dir / f"oof__seed_{cv_seed}__{scenario}.csv", index=False ) return run_rows, fold_rows def aggregate_runs(run_frame: pd.DataFrame) -> pd.DataFrame: return ( run_frame.groupby("scenario", sort=False) .agg( seeds=("cv_seed", "nunique"), macro_f1_mean=("macro_f1_oof", "mean"), macro_f1_std=("macro_f1_oof", "std"), macro_f1_min=("macro_f1_oof", "min"), macro_f1_max=("macro_f1_oof", "max"), severity_mean=("severity_accuracy_submission_oof", "mean"), severity_std=("severity_accuracy_submission_oof", "std"), severity_min=("severity_accuracy_submission_oof", "min"), severity_max=("severity_accuracy_submission_oof", "max"), raw_score_mean=("raw_score_oof", "mean"), raw_score_std=("raw_score_oof", "std"), raw_score_min=("raw_score_oof", "min"), raw_score_max=("raw_score_oof", "max"), ml_points_mean=("ml_points_oof", "mean"), ml_points_min=("ml_points_oof", "min"), ) .reset_index() ) def calculate_deltas( run_frame: pd.DataFrame, masked_scenario: str ) -> pd.DataFrame: metrics = [ "macro_f1_oof", "severity_accuracy_submission_oof", "raw_score_oof", "ml_points_oof", ] clean = run_frame[run_frame["scenario"] == SCENARIO_CLEAN].set_index("cv_seed") masked = run_frame[run_frame["scenario"] == masked_scenario].set_index("cv_seed") rows = [] for cv_seed in clean.index: row: dict[str, object] = {"cv_seed": int(cv_seed)} for metric in metrics: row[f"delta_{metric}"] = float( masked.loc[cv_seed, metric] - clean.loc[cv_seed, metric] ) rows.append(row) return pd.DataFrame(rows) def main() -> None: args = parse_args() df_clean = pd.read_csv(args.data).reset_index(drop=True) validate_data(df_clean, args.n_splits) args.output_dir.mkdir(parents=True, exist_ok=True) df_masked, mask_manifest = mask_spectral_cells( df_clean, missing_rate=args.missing_rate, random_state=args.mask_random_state, ) masked_scenario = missing_scenario_name(args.missing_rate) mask_manifest.to_csv(args.output_dir / "mask_manifest.csv", index=False) all_run_rows: list[dict[str, object]] = [] all_fold_rows: list[dict[str, object]] = [] for cv_seed in args.seeds: print(f"Running CV seed {cv_seed} (clean + {masked_scenario}) ...", flush=True) run_rows, fold_rows = evaluate_seed( df_clean=df_clean, df_masked=df_masked, cv_seed=cv_seed, n_splits=args.n_splits, model_random_state=args.model_random_state, n_jobs=args.n_jobs, output_dir=args.output_dir, masked_scenario=masked_scenario, ) all_run_rows.extend(run_rows) all_fold_rows.extend(fold_rows) run_frame = pd.DataFrame(all_run_rows).sort_values(["scenario", "cv_seed"]) fold_frame = pd.DataFrame(all_fold_rows).sort_values( ["scenario", "cv_seed", "fold"] ) summary_frame = aggregate_runs(run_frame) delta_frame = calculate_deltas(run_frame, masked_scenario) run_frame.to_csv(args.output_dir / "robustness_runs.csv", index=False) fold_frame.to_csv(args.output_dir / "robustness_folds.csv", index=False) summary_frame.to_csv(args.output_dir / "robustness_summary.csv", index=False) delta_frame.to_csv(args.output_dir / "robustness_deltas.csv", index=False) metadata = { "data": str(args.data), "rows": len(df_clean), "engines": int(df_clean["engine_id"].nunique()), "cv_seeds": args.seeds, "n_splits": args.n_splits, "splitter": "StratifiedGroupKFold", "missing_rate": args.missing_rate, "mask_random_state": args.mask_random_state, "masked_cells": len(mask_manifest), "label_pipeline": "combined features + Logistic Regression", "severity_pipeline": "raw features + Extra Trees", "training_data_masked": False, "validation_data_masked": True, } (args.output_dir / "metadata.json").write_text( json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8" ) run_columns = [ "cv_seed", "scenario", "macro_f1_oof", "severity_accuracy_submission_oof", "raw_score_oof", "ml_points_oof", ] print("\nResults per CV seed:") print(run_frame[run_columns].to_string(index=False, float_format="{:.4f}".format)) print("\nAggregate robustness:") print(summary_frame.to_string(index=False, float_format="{:.4f}".format)) print("\nMasked minus clean deltas:") print(delta_frame.to_string(index=False, float_format="{:.4f}".format)) print(f"\nDetailed outputs saved to: {args.output_dir.resolve()}") if __name__ == "__main__": main()