diff --git a/agents/feature_selector.py b/agents/feature_selector.py index 330952d..095e263 100644 --- a/agents/feature_selector.py +++ b/agents/feature_selector.py @@ -69,6 +69,7 @@ def __init__( corr_threshold: float = 0.85, ae_bottleneck_cap: int = 32, ae_max_iter: int = 200, + max_features_for_vif: int = 150, ): # FeatureSelectionAgent owns its ML skills (PCA, AE, VIF). # LLM reasoning is requested through the bus → Orchestrator. @@ -77,6 +78,13 @@ def __init__( self.corr_threshold = corr_threshold self.ae_bottleneck_cap = ae_bottleneck_cap self.ae_max_iter = ae_max_iter + # Generic high-dimensionality guard: the VIF gate is O(rounds × cols × + # OLS), so when a DENSE matrix has more than this many columns we first + # keep only the top-N features by combined PCA+AE importance, then run + # VIF on that bounded set. Feature identity is preserved (unlike a PCA + # projection) so the LLM still selects real, interpretable columns. + # Set to 0/None to disable the cap. + self.max_features_for_vif = max_features_for_vif def run( self, @@ -149,6 +157,25 @@ def run( X[col] = np.log1p(X[col]) X = X.select_dtypes(include=[np.number]) + + # ── Defensive NaN handling ───────────────────────────────────────────── + # StandardScaler / PCA / MLPRegressor / the VIF OLS gate all reject NaN. + # The orchestrator already prunes all-null and mostly-null columns at + # load, but features can still arrive with sparse gaps (e.g. a parquet + # loaded directly, or engineered ratios). Drop any fully-empty column + # that slipped through, then median-impute the rest so the math below + # never crashes. + from skills.data_cleaner import impute_missing + # ±inf (divide-by-zero ratios) is as fatal as NaN — treat it as missing. + X = X.replace([np.inf, -np.inf], np.nan) + all_nan_cols = [c for c in X.columns if X[c].isna().all()] + if all_nan_cols: + print(f' [FeatureSelector] Dropping {len(all_nan_cols)} all-NaN/all-inf column(s) before scaling.') + X = X.drop(columns=all_nan_cols) + if X.isna().any().any(): + X, _imp = impute_missing(X, strategy='median', verbose=False) + print(f' [FeatureSelector] Median-imputed NaN/inf in {len(_imp["imputed"])} column(s).') + feature_names = list(X.columns) n_features = len(feature_names) @@ -190,10 +217,27 @@ def run( combined = 0.5 * _normalise(pca_score) + 0.5 * _normalise(recon_error) ranked_idx = np.argsort(combined)[::-1] # highest first - # ── Step 5: VIF gate ─────────────────────────────────────────────────── + # ── Step 4.5: High-dimensionality prefilter (generic) ────────────────── + # The VIF gate is O(rounds × cols × OLS); on a wide DENSE matrix that + # dominates the runtime. When there are more than max_features_for_vif + # columns, keep only the top-N by combined PCA+AE importance BEFORE the + # gate. This bounds VIF cost for ANY dataset while preserving real + # feature names so the LLM still selects interpretable columns. import pandas as pd - X_for_vif = pd.DataFrame(X_scaled, columns=feature_names) + cap = self.max_features_for_vif + removed_by_prefilter: list[str] = [] + if cap and n_features > cap: + keep_idx = ranked_idx[:cap] + keep_names = [feature_names[i] for i in keep_idx] + keep_set = set(keep_names) + removed_by_prefilter = [f for f in feature_names if f not in keep_set] + print(f' High-dim prefilter: {n_features} → {cap} features ' + f'(top by PCA+AE importance) before VIF gate.') + X_for_vif = pd.DataFrame(X_scaled[:, keep_idx], columns=keep_names) + else: + X_for_vif = pd.DataFrame(X_scaled, columns=feature_names) + # ── Step 5: VIF gate ─────────────────────────────────────────────────── print(f' Running VIF gate (threshold={effective_vif})...') X_clean, removed_by_vif = remove_high_vif( X_for_vif, @@ -203,7 +247,9 @@ def run( ) clean_feature_names = list(X_clean.columns) n_after_vif = len(clean_feature_names) - print(f' After VIF gate: {n_after_vif} features (removed {len(removed_by_vif)})') + print(f' After VIF gate: {n_after_vif} features (removed {len(removed_by_vif)}' + + (f', prefiltered {len(removed_by_prefilter)}' if removed_by_prefilter else '') + + ')') # Failure mode per docs/agents/feature_selector.md: < 10 features survive VIF → blocked if n_after_vif < 10: @@ -425,6 +471,7 @@ def run( issues=issues, metrics={ "n_input_features": n_features, + "n_prefiltered_high_dim": len(removed_by_prefilter), "n_after_vif": n_after_vif, "n_removed_by_vif": len(removed_by_vif), "n_selected": len(selected), diff --git a/agents/orchestrator.py b/agents/orchestrator.py index 730f08a..bf9f2c0 100644 --- a/agents/orchestrator.py +++ b/agents/orchestrator.py @@ -505,6 +505,7 @@ def _llm_handler(agent: str, purpose: str, prompt: str, max_tokens: int, self.bus, ae_bottleneck_cap=config.get('ae_bottleneck_cap', 32), ae_max_iter=config.get('ae_max_iter', 200), + max_features_for_vif=config.get('max_features_for_vif', 150), ) self.cluster_agent = ClusteringAgent(config, self.bus) self.naming_agent = PersonaNamingAgent(self.bus) @@ -514,6 +515,59 @@ def _llm_handler(agent: str, purpose: str, prompt: str, max_tokens: int, self._timings: dict[str, list[float]] = defaultdict(list) self._pipeline_start: float = 0.0 + def _sanitize_loaded_df(self, df, *, source_label: str, user_intent=None): + """Drop low-value columns from a freshly-loaded dataset before any agent + touches it. + + Wide, sparse exports (e.g. a 150-column feature CSV where 60+ columns are + ~99% null and 50+ are constant) otherwise stall the pipeline: NaNs crash + StandardScaler/PCA in FeatureSelector, and constant/rank-deficient columns + slow the VIF gate. Pruning them here is the single, visible place that + guards every downstream stage. + + Controlled by config['data_cleaning']: + enabled (bool, default True) + max_null_frac (float, default 0.5) — drop columns more than this empty + drop_constant (bool, default True) — drop zero-variance columns + + Imputation is intentionally NOT done here: the raw-CSV path feeds + FeatureEngineer (which aggregates raw events, so a median-filled raw + measurement would distort sums/means). FeatureSelector imputes its own + NaNs just before the math that needs it. + """ + from skills.data_cleaner import drop_low_value_columns + + dc_cfg = dict(self.config.get('data_cleaning') or {}) + if dc_cfg.get('enabled', True) is False: + return df + + protect = ['_row_id'] + text_col = getattr(user_intent, 'text_column', None) if user_intent else None + if text_col: + protect.append(text_col) + + cleaned, report = drop_low_value_columns( + df, + max_null_frac=float(dc_cfg.get('max_null_frac', 0.5)), + drop_constant=bool(dc_cfg.get('drop_constant', True)), + protect_cols=protect, + verbose=False, + ) + if report['n_dropped'] > 0: + print( + f" [Orchestrator] Data cleaning ({source_label}): " + f"{report['n_cols_before']} → {report['n_cols_after']} columns " + f"(dropped {len(report['dropped_all_null'])} all-null, " + f"{len(report['dropped_high_null'])} >{report['max_null_frac']:.0%}-null, " + f"{len(report['dropped_constant'])} constant, " + f"{len(report['dropped_duplicate'])} duplicate)." + ) + try: + self.bus.emit('data_cleaning', source=source_label, **report) + except Exception: # noqa: BLE001 + pass + return cleaned + def run( self, features_path: str = 'data/processed/customer_features.parquet', @@ -686,6 +740,9 @@ def run( print(f'\nLoading raw transaction data: {raw_data_path}') full_raw_df = _load_df(raw_data_path) print(f' {len(full_raw_df):,} transactions × {len(full_raw_df.columns)} columns') + full_raw_df = self._sanitize_loaded_df( + full_raw_df, source_label='raw_csv', user_intent=state.user_intent, + ) # Subsample for DatasetExaminer only (it only needs schema + stats) if len(full_raw_df) > 50_000: raw_df = full_raw_df.sample(50_000, random_state=42) @@ -701,6 +758,9 @@ def run( if 'cluster' in features_df.columns: features_df = features_df.drop(columns=['cluster']) print(f' {len(features_df)} customers × {len(features_df.columns)} features') + features_df = self._sanitize_loaded_df( + features_df, source_label='pre_engineered', user_intent=state.user_intent, + ) raw_df = features_df full_raw_df = None diff --git a/config.yaml b/config.yaml index 0aca4d7..221e55e 100644 --- a/config.yaml +++ b/config.yaml @@ -9,6 +9,18 @@ # Can also be overridden at runtime with: python run_pipeline.py --data dataset_path: ~ +# ── Data cleaning (column sanitisation at load) ────────────────────────────── +# Wide, sparse exports (e.g. a 150-column "features" CSV where 60+ columns are +# ~99% null and 50+ are constant) otherwise stall the pipeline: NaNs crash the +# scaler/PCA in feature selection, and constant/rank-deficient columns slow the +# VIF gate. The Orchestrator prunes these once, at load, for BOTH the raw-CSV +# and pre-engineered-parquet paths. What was dropped is printed and emitted on +# the bus (Evidence tab) so the decision is fully visible. +data_cleaning: + enabled: true # set false to feed the raw table through unchanged + max_null_frac: 0.5 # drop any column more than this fraction empty (0.5 = >50% null) + drop_constant: true # drop zero-variance columns (a single unique non-null value) + # ── Data modality ──────────────────────────────────────────────────────────── # What kind of data is being clustered. # auto — detect automatically (text if a long free-text column is found) @@ -121,6 +133,16 @@ ae_bottleneck_cap: 32 # raise it if the AE convergence warning appears and you want more accuracy. ae_max_iter: 200 +# High-dimensionality guard for the VIF gate. +# The VIF gate is O(rounds × cols × OLS), so a wide DENSE feature matrix (many +# real, varying columns) can make it slow. When the matrix has more columns +# than this, FeatureSelector first keeps only the top-N features by combined +# PCA + autoencoder importance, THEN runs VIF on that bounded set. Real feature +# names are preserved (unlike a PCA projection), so the LLM still selects +# interpretable columns. Most engineered matrices (~60–150 cols) never hit this. +# Set to 0 (or null) to disable the cap. +max_features_for_vif: 150 + # ── Classifier ──────────────────────────────────────────────────────────────── # Classifier model for validation. diff --git a/docs/skills/data_cleaner.md b/docs/skills/data_cleaner.md new file mode 100644 index 0000000..29d26aa --- /dev/null +++ b/docs/skills/data_cleaner.md @@ -0,0 +1,65 @@ +# data_cleaner — Column Sanitisation for Wide / Sparse Tables + +**File:** `skills/data_cleaner.py` +**Used by:** [Orchestrator](../agents/orchestrator.md) (at load), [FeatureSelectionAgent](../agents/feature_selector.md) (imputation) + +## Purpose + +Prunes columns that carry no usable clustering signal *before* any heavy math +runs, and (optionally) imputes the remaining gaps. A raw "features" export often +arrives wide and sparse — e.g. 150+ columns where 60+ are ~99% null and 50+ are +constant. Feeding that straight into the pipeline either: + +- crashes `StandardScaler` / PCA / the autoencoder on NaN, or +- stalls the VIF/OLS gate on rank-deficient, all-zero columns. + +Dropping these once, at load, is faster, avoids the crash/stall, and yields +cleaner clusters because the surviving columns actually vary across entities. + +## What gets dropped + +| Category | Rule | +|----------|------| +| Duplicate columns | identical column **name** — keep the first occurrence | +| All-null columns | missing fraction `>= 1.0` | +| Mostly-null columns | missing fraction `> max_null_frac` | +| Constant columns | a single unique non-null value (zero variance) | + +`protect_cols` are never dropped (the Orchestrator protects `_row_id` and the +text column for text modality). + +## API + +```python +from skills.data_cleaner import drop_low_value_columns, impute_missing, sanitize + +cleaned, report = drop_low_value_columns( + df, + max_null_frac=0.5, + drop_constant=True, + drop_duplicate=True, + protect_cols=['_row_id'], +) +# report keys: n_cols_before, n_cols_after, n_dropped, n_rows, max_null_frac, +# dropped_duplicate [name...], dropped_all_null [name...], +# dropped_high_null [[name, frac]...], dropped_constant [name...] + +filled, imp = impute_missing(df, strategy='median') # numeric NaNs only → imp['imputed'] +cleaned, rep = sanitize(df, max_null_frac=0.5, impute='median') # prune + impute +``` + +## Configuration (`config.yaml: data_cleaning`) + +| Knob | Default | Effect | +|------|---------|--------| +| `enabled` | `true` | Master switch for the load-time prune | +| `max_null_frac` | `0.5` | Drop columns more than this fraction empty | +| `drop_constant` | `true` | Drop single-unique-value columns | + +## Imputation policy + +The Orchestrator does **not** impute at load: the raw-CSV path feeds +`FeatureEngineerAgent`, which aggregates raw events, so median-filling a raw +measurement would distort sums/means. Instead, `FeatureSelectionAgent` imputes +its own numeric NaNs (median) immediately before scaling/PCA/AE/VIF — the only +place the value actually has to be finite. diff --git a/skill.md b/skill.md index 72c3739..87fb113 100644 --- a/skill.md +++ b/skill.md @@ -113,6 +113,51 @@ pairs = flag_high_correlation(df, threshold=0.85) --- +## `data_cleaner` — Column Sanitisation for Wide / Sparse Tables + +**File**: `skills/data_cleaner.py` +**Used by**: `Orchestrator` (at load, both paths), `FeatureSelectionAgent` (imputation) + +### Purpose +Prunes columns that carry no usable clustering signal BEFORE any heavy math +runs, then optionally imputes the remaining gaps. A raw "features" export is +often wide and sparse (e.g. 150+ columns, 60+ ~99% null, 50+ constant); feeding +that straight in either crashes `StandardScaler`/PCA on NaN or stalls the VIF +gate on rank-deficient columns. + +### API + +```python +from skills.data_cleaner import drop_low_value_columns, impute_missing, sanitize + +cleaned, report = drop_low_value_columns( + df, + max_null_frac=0.5, # drop columns >50% null + drop_constant=True, # drop zero-variance columns + drop_duplicate=True, # drop duplicate column names (keep first) + protect_cols=['_row_id'], +) +# report: dropped_all_null, dropped_high_null [[name, frac]], dropped_constant, +# dropped_duplicate, n_cols_before, n_cols_after, n_dropped + +filled, imp = impute_missing(df, strategy='median') # numeric NaNs only +cleaned, rep = sanitize(df, max_null_frac=0.5, impute='median') # both steps +``` + +### Defaults (config.yaml: `data_cleaning`) + +| Knob | Default | Effect | +|------|---------|--------| +| `enabled` | `true` | Master switch for the load-time prune | +| `max_null_frac` | `0.5` | Drop columns more than this fraction empty | +| `drop_constant` | `true` | Drop single-unique-value columns | + +Imputation is NOT applied to the raw-CSV path (median-filling a raw measurement +would distort FeatureEngineer's sums/means); FeatureSelector imputes its own +NaNs just before scaling/PCA/VIF. + +--- + ## `silhouette_optimizer` — Data-Driven Cluster Count Selection **File**: `skills/silhouette_optimizer.py` diff --git a/skills/__init__.py b/skills/__init__.py index aff03ef..724bccd 100644 --- a/skills/__init__.py +++ b/skills/__init__.py @@ -8,6 +8,7 @@ from skills.vif_checker import compute_vif, remove_high_vif, flag_high_correlation from skills.silhouette_optimizer import optimize_k, SilhouetteResult from skills.algo_recommender import recommend_algorithm, AlgoRecommendation +from skills.data_cleaner import drop_low_value_columns, impute_missing, sanitize __all__ = [ # Bus @@ -17,6 +18,10 @@ "compute_vif", "remove_high_vif", "flag_high_correlation", + # Data cleaning + "drop_low_value_columns", + "impute_missing", + "sanitize", # Silhouette "optimize_k", "SilhouetteResult", diff --git a/skills/data_cleaner.py b/skills/data_cleaner.py new file mode 100644 index 0000000..e682ca5 --- /dev/null +++ b/skills/data_cleaner.py @@ -0,0 +1,222 @@ +""" +Data Cleaner — Column Sanitisation for Wide / Sparse Feature Matrices + +Contract: docs/skills/data_cleaner.md. + +Prunes columns that carry no usable signal for clustering BEFORE the heavy +math (scaling, PCA, autoencoder, VIF/OLS) ever sees them: + + - duplicate columns — identical column name kept once + - all-null columns — 100% missing + - mostly-null columns — missing fraction > max_null_frac + - constant / zero-variance — a single unique non-null value + +and (optionally) imputes the remaining gaps so downstream estimators that +reject NaN (StandardScaler, PCA, MLPRegressor, the VIF/OLS gate) don't crash. + +Why this exists +─────────────── +A raw "feature" table exported from an upstream system often arrives wide and +sparse — e.g. 150+ columns where 60+ are ~99% null and 50+ are constant. Feeding +that straight into the pipeline either: + * crashes StandardScaler/PCA on NaN, or + * stalls the VIF gate on rank-deficient, all-zero columns. +Dropping these columns first is faster, avoids the crash/stall, and produces +cleaner clusters because the surviving columns actually vary across entities. + +The skill is pure (no I/O, no LLM). It returns the cleaned frame plus a +structured report the Orchestrator surfaces on the bus / Evidence tab so the +user can see exactly which columns were removed and why. +""" +from __future__ import annotations + +import numpy as np +import pandas as pd + + +def drop_low_value_columns( + df: pd.DataFrame, + *, + max_null_frac: float = 0.5, + drop_constant: bool = True, + drop_duplicate: bool = True, + protect_cols: tuple[str, ...] | list[str] = (), + verbose: bool = True, +) -> tuple[pd.DataFrame, dict]: + """ + Drop columns that carry no usable clustering signal. + + Parameters + ---------- + df : pd.DataFrame + The raw table (entity-level feature matrix OR raw event table). + max_null_frac : float + Drop any column whose missing fraction is strictly greater than this. + 0.5 → drop columns that are more than half empty. Set to 1.0 to drop + only fully-empty (100% null) columns. + drop_constant : bool + Drop columns with a single unique non-null value (zero variance). + drop_duplicate : bool + Drop duplicate column *names* (keep the first occurrence). Always run + before the constant check so `df[col]` is unambiguous. + protect_cols : list[str] + Column names that must never be dropped (e.g. the entity id `_row_id`, + or the text column for text modality). + verbose : bool + Print a one-line summary of what was removed. + + Returns + ------- + (cleaned_df, report) + cleaned_df : DataFrame with the low-value columns removed. + report : dict describing exactly what was dropped (see below). + """ + protect = {str(c) for c in protect_cols} + report: dict = { + "n_cols_before": int(df.shape[1]), + "n_rows": int(df.shape[0]), + "max_null_frac": float(max_null_frac), + "dropped_duplicate": [], + "dropped_all_null": [], + "dropped_high_null": [], # list of [name, null_fraction] + "dropped_constant": [], + } + + work = df + + # ── 1. Duplicate column names (keep first) ──────────────────────────────── + if drop_duplicate: + dup_mask = work.columns.duplicated() + if dup_mask.any(): + report["dropped_duplicate"] = [str(c) for c in work.columns[dup_mask]] + work = work.loc[:, ~dup_mask] + + null_frac = work.isna().mean() + drop_set: set = set() + + # ── 2. All-null and mostly-null columns ─────────────────────────────────── + for col in work.columns: + if str(col) in protect: + continue + frac = float(null_frac[col]) + if frac >= 1.0: + report["dropped_all_null"].append(str(col)) + drop_set.add(col) + elif frac > max_null_frac: + report["dropped_high_null"].append([str(col), round(frac, 4)]) + drop_set.add(col) + + # ── 3. Constant / zero-variance columns (among survivors) ───────────────── + if drop_constant: + for col in work.columns: + if col in drop_set or str(col) in protect: + continue + # nunique(dropna=True) <= 1 → all non-null values are identical + # (also catches columns that are entirely null, already caught above). + if work[col].nunique(dropna=True) <= 1: + report["dropped_constant"].append(str(col)) + drop_set.add(col) + + if drop_set: + work = work.drop(columns=[c for c in work.columns if c in drop_set]) + + report["n_cols_after"] = int(work.shape[1]) + report["n_dropped"] = report["n_cols_before"] - report["n_cols_after"] + + if verbose and report["n_dropped"] > 0: + print( + f" [DataCleaner] {report['n_cols_before']} → {report['n_cols_after']} columns " + f"(dropped {len(report['dropped_all_null'])} all-null, " + f"{len(report['dropped_high_null'])} >{max_null_frac:.0%}-null, " + f"{len(report['dropped_constant'])} constant, " + f"{len(report['dropped_duplicate'])} duplicate)." + ) + + return work, report + + +def impute_missing( + df: pd.DataFrame, + *, + strategy: str = "median", + verbose: bool = True, +) -> tuple[pd.DataFrame, dict]: + """ + Fill remaining NaNs in numeric columns so NaN-intolerant estimators + (StandardScaler, PCA, MLPRegressor, the VIF/OLS gate) don't crash. + + Non-numeric columns are left untouched — categorical handling belongs to + the agents that consume them (FeatureEngineer / TextPreparer). + + Parameters + ---------- + strategy : {"median", "mean", "zero"} + How to fill numeric gaps. "median" is robust to the skew that is common + in these feature tables. If the chosen statistic is itself NaN (a column + that is entirely null), 0.0 is used. + + Returns + ------- + (filled_df, report) where report['imputed'] maps column → fill value. + """ + report: dict = {"strategy": strategy, "imputed": {}, "n_inf_replaced": 0} + work = df.copy() + num_cols = work.select_dtypes(include=[np.number]).columns + + # ±inf (e.g. divide-by-zero ratios) is just as fatal to the scaler/PCA as + # NaN — treat it as missing so it is imputed alongside the genuine gaps. + if num_cols.size: + inf_mask = np.isinf(work[num_cols].to_numpy(dtype=float, na_value=np.nan)) + n_inf = int(inf_mask.sum()) + if n_inf: + report["n_inf_replaced"] = n_inf + work[num_cols] = work[num_cols].replace([np.inf, -np.inf], np.nan) + + for col in num_cols: + if not work[col].isna().any(): + continue + if strategy == "mean": + fill = work[col].mean() + elif strategy == "zero": + fill = 0.0 + else: # median (default) + fill = work[col].median() + if pd.isna(fill): + fill = 0.0 + work[col] = work[col].fillna(fill) + report["imputed"][str(col)] = float(fill) + + if verbose and report["imputed"]: + print(f" [DataCleaner] Median-imputed NaNs in {len(report['imputed'])} numeric column(s).") + + return work, report + + +def sanitize( + df: pd.DataFrame, + *, + max_null_frac: float = 0.5, + drop_constant: bool = True, + impute: str | None = None, + protect_cols: tuple[str, ...] | list[str] = (), + verbose: bool = True, +) -> tuple[pd.DataFrame, dict]: + """ + Convenience wrapper: drop low-value columns, then (optionally) impute. + + Set ``impute`` to one of {"median", "mean", "zero"} to fill the remaining + numeric gaps; leave it None to only prune columns (the right choice for raw + event tables, where imputing a raw measurement would distort aggregates). + """ + cleaned, drop_report = drop_low_value_columns( + df, + max_null_frac=max_null_frac, + drop_constant=drop_constant, + protect_cols=protect_cols, + verbose=verbose, + ) + report = {"drop": drop_report} + if impute: + cleaned, imp_report = impute_missing(cleaned, strategy=impute, verbose=verbose) + report["impute"] = imp_report + return cleaned, report