Validity and Stability Pipeline Development FS Benchmark

experimental/feature_selection_benchmark/extra_benchmark/feature_selection_benchmark_runner.py

@@ -1,27 +1,39 @@
 """Shared infrastructure and entry point for feature selection benchmark evaluation.
 
 Usage:
-    python fs_benchmark_runner.py --mode validity --seed 42
-    python fs_benchmark_runner.py --mode stability --seed 42
+    python feature_selection_benchmark_runner.py \
+        --mode validity \
+        --method_name FSBench__RandomFeatureSelector__5__0__lgbm__3600 \
+        --data_foundry_task_id "UserTask|1386903908|anneal/019d3f7b-494a-71fa-8eb2-25d01dfb7792|/work/dlclarge1/purucker-fs_benchmark/.openml/tabarena_tasks" \
+        --repeat 0 \
+        --noise 1.0 \
+        --noise_type gaussian
+
+    python feature_selection_benchmark_runner.py \
+        --mode stability \
+        --method_name FSBench__RandomFeatureSelector__5__0__lgbm__3600 \
+        --data_foundry_task_id "UserTask|1386903908|anneal/019d3f7b-494a-71fa-8eb2-25d01dfb7792|/work/dlclarge1/purucker-fs_benchmark/.openml/tabarena_tasks" \
+        --repeat 0 \
+        --noise 1.0 \
+        --noise_type gaussian
 """
 
 from __future__ import annotations
 
 import argparse
 import time
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any
+from pathlib import Path
+from typing import Any
 
 import numpy as np
+import pandas as pd
 from tabarena.benchmark.feature_selection_methods.feature_selection_benchmark_utils import (
     selector_and_config_from_string,
 )
 from tabarena.benchmark.task.openml import OpenMLTaskWrapper
 from tabflow_slurm.run_tabarena_experiment import _parse_task_id
 
-if TYPE_CHECKING:
-    import pandas as pd
-
 
 @dataclass
 class FeatureSelectionResult:
@@ -36,6 +48,9 @@ class FeatureSelectionResult:
         repeat: Repeat number for the FS metric.
 
         selected_features: Names of selected features from the original_features.
+        num_classes: Number of classes in the target variable (for classification tasks).
+        num_samples: Number of samples in the dataset.
+
         elapsed_time_fs: Runtime measurement (seconds).
 
         mode: Evaluation mode ("validity" or "stability").
@@ -50,6 +65,9 @@ class FeatureSelectionResult:
     repeat: int
 
     selected_features: list[int]
+    num_classes: int
+    num_samples: int
+
     elapsed_time_fs: float
 
     mode: str
@@ -62,11 +80,11 @@ def _augment_dataset(
     original_features = list(X.columns)
 
     if mode == "validity":
-        from validity_fs_metric import get_dataset_for_validity
+        from validity_fs_metric import get_dataset_for_validity  # noqa: PLC0415
 
         X = get_dataset_for_validity(X=X, rng=rng, **kwargs)
     elif mode == "stability":
-        from stability_fs_metric import get_dataset_for_stability
+        from stability_fs_metric import get_dataset_for_stability  # noqa: PLC0415
 
         X, y = get_dataset_for_stability(X=X, y=y, rng=rng, **kwargs)
     else:
@@ -79,7 +97,7 @@ def _augment_dataset(
     return X, y, original_features
 
 
-def run_benchmark(
+def run_benchmark(  # noqa: D417
     *,
     data_foundry_task_id: str,
     mode: str,
@@ -109,7 +127,7 @@ def run_benchmark(
     feature_selector, config = selector_and_config_from_string(preprocessing_name=method_name)
 
     # Augment dataset with new feature based on mode.
-    X, y, original_features = _augment_dataset(mode=mode, X=X, rng=rng, **kwargs)
+    X, y, original_features = _augment_dataset(mode=mode, X=X, y=y, rng=rng, **kwargs)
 
     # Run Feature Selection
     start_time = time.monotonic()
@@ -135,6 +153,13 @@ def run_benchmark(
         mode_kwargs=kwargs,
     )
 
+def get_cache_path(args) -> Path:
+    """Generate the cache path based on arguments."""
+    cache_dir = Path(__file__).parent / "results"
+    cache_dir.mkdir(parents=True, exist_ok=True)  # Ensure the directory exists
+    cache_path = cache_dir / f"{args.mode}_{args.method_name}_{args.data_foundry_task_id.split('|')[3].split('/')[0]}_{args.repeat}.csv"
+    return cache_path
+
 
 def parse_args() -> argparse.Namespace:
     """Parse CLI arguments for the FS benchmark runner."""
@@ -149,14 +174,15 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument(
         "--method_name",
         type=str,
-        default="FSBench__AccuracyFeatureSelector__5__0__lgbm__3600",
+        default="FSBench__RandomFeatureSelector__5__0__lgbm__3600",
         help="Feature Selection Method name [default: FSBench__AccuracyFeatureSelector__5__0__lgbm__3600]",
     )
     parser.add_argument(
         "--data_foundry_task_id",
         type=str,
-        default="anneal/019d3f7b-494a-71fa-8eb2-25d01dfb7792",
-        help="TabArena/OpenML task identifier [default: anneal/019d3f7b-494a-71fa-8eb2-25d01dfb7792]",
+        default="UserTask|1386903908|anneal/019d3f7b-494a-71fa-8eb2-25d01dfb7792|/Users/schaefer.bastian/.openml/tabarena_tasks",
+        help="TabArena/OpenML task metadata identifier [default: UserTask|1386903908|anneal/019d3f7b-494a-71fa-8eb2-25d"
+             "01dfb7792|/Users/schaefer.bastian/.openml/tabarena_tasks]",
     )
     parser.add_argument("--repeat", type=int, default=0, help="Repeat [default: 0]")
 
@@ -165,27 +191,41 @@ def parse_args() -> argparse.Namespace:
         "--noise",
         type=float,
         default=1.0,
-        nargs="+",
         help="Noise features relative to original count (validity mode only) [default: 1.0]",
     )
     parser.add_argument(
-        "--nose_type",
+        "--noise_type",
         type=str,
         choices=["gaussian", "uniform"],
         default="gaussian",
         help="Type of noise features to add (validity mode only) [default: random]",
     )
+    parser.add_argument(
+        "--ignore_cache",
+        type=bool,
+        default=False,
+        help="Whether to ignore existing cache and rerun the benchmark (useful for debugging) [default: False]",
+    )
     return parser.parse_args()
 
 
 if __name__ == "__main__":
     args = parse_args()
 
-    run_benchmark(
-        data_foundry_task_id=args.data_foundry_task_id,
-        mode=args.mode,
-        method_name=args.method_name,
-        repeat=args.repeat,
-        noise=args.noise,
-        nose_type=args.nose_type,
-    )
+    cache_path = get_cache_path(args)
+
+    if cache_path.exists() and not args.ignore_cache:
+        print(f"Cache exists at {cache_path}. Skipping operation.")
+    else:
+        result = run_benchmark(
+            data_foundry_task_id=args.data_foundry_task_id,
+            mode=args.mode,
+            method_name=args.method_name,
+            repeat=args.repeat,
+            noise=args.noise,
+            noise_type=args.noise_type,
+        )
+
+        print(result)
+        result = pd.DataFrame([result.__dict__])
+        result.to_csv(cache_path, index=False)

experimental/feature_selection_benchmark/extra_benchmark/run_setup_slurm_jobs_extra_benchmark.py

@@ -0,0 +1,81 @@
+from pathlib import Path
+
+import pandas as pd
+import submitit
+from experimental.feature_selection_benchmark.tabarena_setup.fr_cluster_setup import (
+    ALL_TASK_METADATA,
+    FS_TIME_LIMIT,
+    FSBenchmarkConfig,
+)
+
+
+def run_extra_pipeline(mode, method_name, task_id, noise, noise_type):
+    """Function to run the extra pipeline."""
+    print(
+        f"Running extra pipeline with mode={mode}, method_name={method_name}, task_id={task_id}, noise={noise}, noise_type={noise_type}")
+    # Add your pipeline logic here
+
+
+if __name__ == "__main__":
+    method_names = FSBenchmarkConfig().get_default_preprocessing_configs(
+        fs_methods=[
+            "AccuracyFeatureSelector",
+            "RandomFeatureSelector",
+            "ANOVAFeatureSelector",
+            "CFSFeatureSelector",
+            "Chi2FeatureSelector",
+            "DISRFeatureSelector",
+            "GainRatioFeatureSelector",
+            "GiniFeatureSelector",
+            "ImpurityFeatureSelector",
+            "InformationGainFeatureSelector",
+            "INTERACTFeatureSelector",
+            "MarkovBlanketFeatureSelector",
+            "MIFeatureSelector",
+            "mRMRFeatureSelector",
+            "PearsonCorrelationFeatureSelector",
+            "ReliefFFeatureSelector",
+            "RFImportanceFeatureSelector",
+            "SequentialBackwardEliminationFeatureSelector",
+            "SequentialForwardSelectionFeatureSelector",
+            "SymmetricalUncertaintyFeatureSelector",
+            "LassoFeatureSelector",  # just for regression but with label encoder for classification?
+            "LaplacianScoreFeatureSelector",  # OOM, Segmentation fault issues
+            "ConsistencyFeatureSelector",
+            # selected_indices = np.where(S)[0].tolist(), UnboundLocalError: cannot access local variable 'S' where it is not associated with a value
+            "JMIFeatureSelector",
+            # time limit computed incorrectly, and error at remaining.remove(best_idx), ValueError: list.remove(x): x not in list
+            "OneRFeatureSelector",
+            # major OOM errors (tries to allocate one major array), wrong time limit computation,  max(accuracies, key=accuracies.get) -> max() iterable argument is empty
+            "ElasticNetFeatureSelector",  # Only for classification
+            "CMIMFeatureSelector",  # problems with time limit and fallback of features
+            # "tTestFeatureSelector", # Does not work for regression
+            "CARTFeatureSelector",  # Only implemented for classification, OOM problems as well
+        ]
+    )
+    task_ids = pd.read_csv(ALL_TASK_METADATA)["task_id_str"].tolist()
+    # Define the parameter grid
+    modes = ["validity", "stability"]
+    noises = [0.5, 0.75, 1.0]
+    noise_types = ["gaussian"]
+
+    # Create a SLURM executor
+    executor = submitit.AutoExecutor(folder=Path("slurm_logs"))
+    executor.update_parameters(
+        timeout_min=FS_TIME_LIMIT,  # Job timeout in minutes
+        slurm_partition="default",  # SLURM partition
+        cpus_per_task=8,  # Number of CPUs per task
+        mem_gb=32,  # Memory in GB
+    )
+
+    # Submit jobs
+    with executor.batch():
+        for mode in modes:
+            for method_name in method_names:
+                for task_id in task_ids:
+                    if mode == "validity":
+                        for noise in noises:
+                            for noise_type in noise_types:
+                                executor.submit(run_extra_pipeline, mode, method_name, task_id, noise, noise_type)
+                    else:
+                        executor.submit(run_extra_pipeline, mode, method_name, task_id, noise, noise_type)

tabarena/tabarena/benchmark/experiment/experiment_runner.py

@@ -82,11 +82,19 @@ def __init__(
         self.eval_metric: Scorer = get_metric(metric=self.eval_metric_name, problem_type=self.task.problem_type)
         self.model: AbstractExecModel | None = None
         self.task_split_idx = self.task.get_split_idx(fold=self.fold, repeat=self.repeat, sample=self.sample)
-        self.X, self.y, self.X_test, self.y_test = self.task.get_train_test_split(fold=self.fold, repeat=self.repeat, sample=self.sample)
+
+        if self.task.lazy_load_data:
+            assert input_format == "openml", "Lazy load data only works with input_format='openml'"
+            self.X, self.y, self.X_test, self.y_test = None, None, None, None
+            _, y, _, _ = self.task.get_train_test_split(fold=self.fold, repeat=self.repeat, sample=self.sample)
+        else:
+            self.X, self.y, self.X_test, self.y_test = self.task.get_train_test_split(fold=self.fold, repeat=self.repeat, sample=self.sample)
+            y = self.y
+
         if input_format == "csv":
             self.X = self.task.to_csv_format(X=self.X)
             self.X_test = self.task.to_csv_format(X=self.X_test)
-        self.label_cleaner = LabelCleaner.construct(problem_type=self.task.problem_type, y=self.y)
+        self.label_cleaner = LabelCleaner.construct(problem_type=self.task.problem_type, y=y)
         if cacher is None:
             cacher = CacheFunctionDummy()
         self.cacher = cacher
@@ -105,8 +113,19 @@ def split_seed(self):
         """We use the split index as a source for a seed that creates different randomness per split."""
         return self.task_split_idx
 
+    def _lazy_load_for_run_model_fit(self):
+        X, y, X_test, _ = self.task.get_train_test_split(fold=self.fold, repeat=self.repeat, sample=self.sample)
+        return X, y, X_test
+
+
     def run_model_fit(self) -> dict:
-        return self.model.fit_custom(X=self.X, y=self.y, X_test=self.X_test, split_seed=self.split_seed)
+        if self.task.lazy_load_data:
+            lazy_load_function = self._lazy_load_for_run_model_fit
+            X, y, X_test = None, None, None
+        else:
+            lazy_load_function = None
+            X, y, X_test = self.X, self.y, self.X_test
+        return self.model.fit_custom(X=X, y=y, X_test=X_test, split_seed=self.split_seed, lazy_load_function=lazy_load_function)
 
     def run(self) -> dict:
         out = self._run()
@@ -157,11 +176,19 @@ def _run(self) -> dict:
                 self.handle_failure(exc=exc)
             raise
         out = self.post_fit(out=out)
+
+        if self.task.lazy_load_data:
+            _, _, _, y_test = self.task.get_train_test_split(fold=self.fold, repeat=self.repeat, sample=self.sample)
+        else:
+            y_test = self.y_test
         out["metric_error"] = self.evaluate(
-            y_true=self.y_test,
+            y_true=y_test,
             y_pred=out["predictions"],
             y_pred_proba=out["probabilities"],
         )
+        if self.task.lazy_load_data:
+            del y_test
+
         out = self.post_evaluate(out=out)
         out["experiment_metadata"] = self._experiment_metadata(time_start=time_start, time_start_str=time_start_str)
         out = self.convert_to_output(out=out)
@@ -269,7 +296,12 @@ def post_evaluate(self, out: dict) -> dict:
                 simulation_artifact["pred_proba_dict_test"] = self.label_cleaner.transform_proba(out["probabilities"], as_pandas=True)
                 if self.task.problem_type == "binary":
                     simulation_artifact["pred_proba_dict_test"] = simulation_artifact["pred_proba_dict_test"].iloc[:, 1]
-            simulation_artifact["y_test"] = self.label_cleaner.transform(self.y_test)
+
+            if self.task.lazy_load_data:
+                _, _, _, y_test = self.task.get_train_test_split(fold=self.fold, repeat=self.repeat, sample=self.sample)
+            else:
+                y_test = self.y_test
+            simulation_artifact["y_test"] = self.label_cleaner.transform(y_test)
 
             if self.optimize_simulation_artifacts_memory:
                 # optimize memory
@@ -296,7 +328,12 @@ def post_evaluate(self, out: dict) -> dict:
             simulation_artifact["metric"] = self.eval_metric_name
 
             if self.compute_bag_info and (self.model.can_get_per_child_oof and self.model.can_get_per_child_val_idx):
-                simulation_artifact["bag_info"] = self.model.bag_artifact(X_test=self.X_test)
+                if self.task.lazy_load_data:
+                    _, _, X_test, _ = self.task.get_train_test_split(fold=self.fold, repeat=self.repeat, sample=self.sample)
+                else:
+                    X_test = self.X_test
+
+                simulation_artifact["bag_info"] = self.model.bag_artifact(X_test=X_test)
 
 
             simulation_artifact["pred_proba_dict_val"] = {self.method: simulation_artifact["pred_proba_dict_val"]}