feat: sanitize wide/sparse datasets before clustering

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),

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
 

config.yaml

@@ -9,6 +9,18 @@
 # Can also be overridden at runtime with: python run_pipeline.py --data <path>
 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.

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.

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`

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",