import gc import numpy as np import pandas as pd from tqdm import tqdm import os, warnings, traceback, argparse, multiprocessing as mp from utils import ( list_local_datasets, load_dataset, run_ad, run_cl, set_analysis_algorithms ) import tracemalloc, time from multiprocessing import shared_memory from sklearn.preprocessing import RobustScaler warnings.filterwarnings("ignore", category=FutureWarning) # Force single-threaded BLAS/MKL so subprocesses don't over-subscribe CPU cores os.environ["OMP_NUM_THREADS"] = "1" os.environ["OPENBLAS_NUM_THREADS"] = "1" os.environ["MKL_NUM_THREADS"] = "1" os.environ["NUMEXPR_NUM_THREADS"] = "1" N_JOBS = 1 TIMEOUT = 5 * 60 # 5 minutes def run_with_timeout(func, args=(), kwargs={}, timeout=TIMEOUT, seed=None): # Move large arrays into shared memory so the subprocess can access them # without pickling (avoids expensive serialization for big datasets) shms = [] new_args = [] for arg in args: if isinstance(arg, np.ndarray) and arg.nbytes > 50 * 1024**2: # >50 MB shm = shared_memory.SharedMemory(create=True, size=arg.nbytes) shared_arr = np.ndarray(arg.shape, dtype=arg.dtype, buffer=shm.buf) np.copyto(shared_arr, arg) # Pass metadata; the child process reconstructs the array from shared memory. new_args.append((shm.name, arg.shape, arg.dtype)) shms.append(shm) else: new_args.append(arg) q = mp.Queue() p = mp.Process(target=_wrapper_shm, args=(q, func, new_args, kwargs, seed)) p.start(); p.join(timeout) # Always release shared memory blocks, even on timeout or error for shm in shms: shm.close(); shm.unlink() if p.is_alive(): p.terminate(); p.join() raise TimeoutError("Algorithm exceeded time limit.") if q.empty(): raise RuntimeError("No result returned from subprocess.") status, payload = q.get() if status == 'success': return (*payload[0], payload[1], payload[2]) # (*scores, duration, peak_memory) if status == 'memory': raise MemoryError(payload) raise Exception(payload) def _wrapper_shm(queue, func, args, kwargs, seed=None): """Subprocess entry point: reconstruct shared-memory arrays, run func, report back.""" try: if seed is not None: np.random.seed(seed) real_args = [] shms = [] for arg in args: # Detect descriptor tuples injected by run_with_timeout if isinstance(arg, tuple) and len(arg) == 3 and isinstance(arg[0], str): name, shape, dtype = arg shm = shared_memory.SharedMemory(name=name) arr = np.ndarray(shape, dtype=dtype, buffer=shm.buf) real_args.append(arr) shms.append(shm) else: real_args.append(arg) # Measure Python-level peak memory during the algorithm run. tracemalloc.start() t0 = time.perf_counter() result = func(*real_args, **kwargs) t1 = time.perf_counter() current, peak = tracemalloc.get_traced_memory() tracemalloc.stop() for shm in shms: shm.close() queue.put(('success', (result, t1 - t0, peak))) except MemoryError as e: queue.put(('memory', str(e))) except Exception: queue.put(('error', traceback.format_exc())) def _set_status(row, name, status, cols, scores, duration): """Write metric scores, elapsed time, and status into a result row dict.""" for col, score in zip(cols, scores): row[col] = score row[f"{name}_time"] = duration row[f"{name}_status"] = status return row def _upsert_row(df, row, dataset, rep, output_file): """Replace the matching row in df (by dataset + rep) and persist to CSV.""" mask = (df['dataset'] == dataset) & (df['rep'] == rep) df = pd.concat([df[~mask], pd.DataFrame([row])], ignore_index=True) df.to_csv(output_file, index=False) return df def _run_loop(df, row, dataset, rep, algorithms, task, metric_suffixes, run_fn, run_args_fn, output_file, skip_algos, seed): for name, ctor in tqdm(algorithms.items(), desc=f"{task}", leave=False): skip_key = (dataset, name) # Skip algorithms that previously timed out or ran out of memory on this dataset if skip_key in skip_algos: status = skip_algos[skip_key] print(f" Skipping {task}: {name} - {dataset} - (prev {skip_algos[skip_key]})") cols = [f"{name}_{s}" for s in metric_suffixes] row = _set_status(row, name, status, cols, [np.nan] * len(cols), np.nan) df = _upsert_row(df, row, dataset, rep, output_file) continue # Skip if primary metric column already has a value (resume-friendly) first_col = f"{name}_{metric_suffixes[0]}" if first_col in row and not pd.isna(row[first_col]): print(f" Skipping {task}: {name} - {dataset} - (already computed)") continue cols = [f"{name}_{s}" for s in metric_suffixes] try: *scores, duration, memory = run_with_timeout(run_fn, args=run_args_fn(name, ctor), seed=seed) row = _set_status(row, name, "success", cols, scores, duration) row[f"{name}_memory_mb"] = memory / (1024**2) except TimeoutError: print(f" Failed {task}: {name} - {dataset} - timeout") row = _set_status(row, name, "timeout", cols, [np.nan]*len(cols), np.nan) # Skip this algorithm in later repetitions of the same dataset skip_algos[skip_key] = "timeout" except MemoryError: print(f" Failed {task}: {name} - {dataset} - memory") row = _set_status(row, name, "memory", cols, [np.nan]*len(cols), np.nan) # Skip this algorithm in later repetitions of the same dataset skip_algos[skip_key] = "memory" except Exception as e: print(f" Failed {task}: {name} - {dataset} - {e}") row = _set_status(row, name, "error", cols, [np.nan]*len(cols), np.nan) df = _upsert_row(df, row, dataset, rep, output_file) return df, row def run_analysis(mode="ad", n_reps=5, max_samples=10000, datasets_dir="datasets/adbench", dataset_index=None, rep_index=None, output_file=None, random_seed=0): """ Run anomaly detection / clustering experiments on REAL datasets read from a local folder. Args: mode: Tasks to run — 'ad' | 'cl' n_reps: Number of independent repetitions max_samples: Stratified subsampling cap for large datasets (None = no cap) datasets_dir: Local folder containing the .npz files (X, y) dataset_index: If set, run only this dataset (0-based, into the sorted folder listing) rep_index: If set, run only this repetition (0-based); requires dataset_index output_file: CSV path override """ assert mode in ("ad", "cl"), "mode must be 'ad' or 'cl'" source_name = os.path.basename(os.path.normpath(datasets_dir)) results_dir = os.path.join("results", f"results_real_{source_name}") os.makedirs(results_dir, exist_ok=True) output_file = output_file or os.path.join(results_dir, f"results_real_{source_name}.csv") datasets = list_local_datasets(datasets_dir) if dataset_index is not None: datasets = [datasets[dataset_index]] columns = ["dataset", "n_samples", "n_features", "outlier_fraction", "rep"] # Load existing results so interrupted runs can be resumed if os.path.exists(output_file) and os.path.getsize(output_file) > 0: try: df = pd.read_csv(output_file, na_values=["", "nan", "NaN"]).replace("", np.nan) except pd.errors.EmptyDataError: df = pd.DataFrame(columns=columns) else: df = pd.DataFrame(columns=columns) reps = [rep_index] if rep_index is not None else list(range(n_reps)) skip_algos = {} # key: (dataset, algo_name) → failure reason; persists across reps total = len(datasets) * len(reps) with tqdm(total=total, desc="Overall progress") as pbar: for i, dataset in enumerate(datasets): for rep in reps: seed = random_seed + rep np.random.seed(seed) Xraw, yo, contamination = load_dataset( dataset, datasets_dir=datasets_dir, max_samples=max_samples, seed=seed ) Xo = RobustScaler().fit_transform(Xraw) n_samples, n_features = Xo.shape if mode == "ad": outlier_fraction = float(yo.mean()) else: outlier_fraction = sum(yo==-1)/n_samples n_classes = len(np.unique(yo)) print(f"\n[real|{source_name}] Dataset {i+1}/{len(datasets)}: {dataset} - " f"{n_samples} samples, {n_features} dims, {n_classes} classes - " f"{outlier_fraction*100:.1f}% outliers | rep {rep+1}/{n_reps} (seed={seed})") mask = (df['dataset'] == dataset) & (df['rep'] == rep) if not df.empty and mask.any(): row = df.loc[mask].iloc[0].to_dict() else: row = {"dataset": dataset, "n_samples": n_samples, "n_features": n_features, "n_classes": n_classes, "outlier_fraction": outlier_fraction, "rep": rep, "seed": seed} ad_algorithms, clu_algorithms = set_analysis_algorithms(n_clusters=n_classes, size=n_samples, dims=n_features, n_jobs=N_JOBS, seed=seed) if mode in ("ad"): df, row = _run_loop( df, row, dataset, rep, algorithms=ad_algorithms, task="AD", metric_suffixes=["roc", "ap"], run_fn=run_ad, run_args_fn=lambda n, ctor: (ctor, Xo, yo), output_file=output_file, skip_algos=skip_algos, seed=seed) if mode in ("cl"): df, row = _run_loop( df, row, dataset, rep, algorithms=clu_algorithms, task="CL", metric_suffixes=["adjRand", "ami"], run_fn=run_cl, run_args_fn=lambda n, cfg: (n, cfg, Xo, yo), output_file=output_file, skip_algos=skip_algos, seed=seed) gc.collect() pbar.update(1) print("\nDone!") if __name__ == "__main__": parser = argparse.ArgumentParser(description="Run AD and CL experiments on real datasets.") parser.add_argument("--mode", default="ad", choices=["ad", "cl"]) parser.add_argument("--n_reps", type=int, default=5) parser.add_argument("--max_samples", type=int, default=10000, help="Stratified subsample cap for large datasets; 0 disables capping") parser.add_argument("--datasets_dir", type=str, default="datasets/Classical", help="Folder with .npz (X, y) files") parser.add_argument("--dataset_index", type=int, default=None) parser.add_argument("--rep_index", type=int, default=None) parser.add_argument("--output_file", type=str, default=None) parser.add_argument("--random_seed", type=int, default=0) args = parser.parse_args() max_samples = None if args.max_samples in (0, None) else args.max_samples run_analysis(args.mode, args.n_reps, max_samples, args.datasets_dir, args.dataset_index, args.rep_index, args.output_file, args.random_seed)