P2 and P3 Initial

Author: raptor419

streamline/p2_impute_scale/interface.py

@@ -0,0 +1,80 @@
+# Phase 2: Imputation & Scaling — base interfaces (no typing, no shared base)
+
+class Imputer(object):
+    """
+    Base interface for imputers.
+    Contract:
+      - fit(X, y, feature_meta) learns imputation statistics.
+      - transform(X) applies them without reordering columns.
+    """
+
+    def __init__(self, component_id="imputer", random_state=None, **kwargs):
+        self.id = component_id
+        self.random_state = random_state
+        self.params = dict(kwargs)
+
+        # capability flags (override in subclasses or set via set_params)
+        self.supports_nan_in_fit = True     # can learn with NaNs present
+        self.preserves_dtype = False        # try to keep dtype if possible
+
+    def get_params(self):
+        return dict(self.params)
+
+    def set_params(self, **params):
+        self.params.update(params)
+        return self
+
+    def state_dict(self):
+        return dict(self.params)
+
+    def load_state_dict(self, state):
+        if state is None:
+            state = {}
+        self.params.update(state)
+
+    # --- to implement ---
+    def fit(self, X, y=None, feature_meta=None):
+        raise NotImplementedError("Imputer.fit must be implemented")
+
+    def transform(self, X):
+        raise NotImplementedError("Imputer.transform must be implemented")
+
+
+class Scaler(object):
+    """
+    Base interface for scalers.
+    Contract:
+      - fit(X, y, feature_meta) learns scaling params.
+      - transform(X) applies them without reordering columns.
+    """
+
+    def __init__(self, component_id="scaler", random_state=None, **kwargs):
+        self.id = component_id
+        self.random_state = random_state
+        self.params = dict(kwargs)
+
+        # capability flags
+        self.requires_dense = True               # most scalers need dense inputs
+        self.scale_only_quantitative = True      # ignore categoricals by default
+
+    def get_params(self):
+        return dict(self.params)
+
+    def set_params(self, **params):
+        self.params.update(params)
+        return self
+
+    def state_dict(self):
+        return dict(self.params)
+
+    def load_state_dict(self, state):
+        if state is None:
+            state = {}
+        self.params.update(state)
+
+    # --- to implement ---
+    def fit(self, X, y=None, feature_meta=None):
+        raise NotImplementedError("Scaler.fit must be implemented")
+
+    def transform(self, X):
+        raise NotImplementedError("Scaler.transform must be implemented")

streamline/p3_feature_learning/interface.py

@@ -0,0 +1,66 @@
+# Phase 3: Feature Learning — base interface (no typing, flags on self)
+
+class FeatureLearner(object):
+    """
+    Base interface for optional feature learning / transformation steps.
+    Examples: PCA/ICA/NMF, polynomial features, random features, FIBERS, etc.
+
+    Contract:
+      - fit(X, y, feature_meta) learns transform parameters.
+      - transform(X) applies them; must preserve row order.
+      - get_feature_names_out(input_features) returns names for output columns.
+      - get_parent_map(output_features) maps produced features -> source columns.
+    """
+
+    def __init__(self, component_id="feature_learner", random_state=None, **kwargs):
+        # identifiers & params
+        self.id = component_id
+        self.random_state = random_state
+        self.params = dict(kwargs)
+
+        # capability flags (override in subclasses or via set_params)
+        self.needs_quantitative = False    # True if input must be numeric-only
+        self.is_supervised = False         # True if y is required during fit
+        self.produces_sparse = False       # True if transform returns sparse
+
+    # ---------- lifecycle ----------
+    def get_params(self):
+        return dict(self.params)
+
+    def set_params(self, **params):
+        self.params.update(params)
+        return self
+
+    def state_dict(self):
+        return dict(self.params)
+
+    def load_state_dict(self, state):
+        if state is None:
+            state = {}
+        self.params.update(state)
+
+    # ---------- fit/transform ----------
+    def fit(self, X, y=None, feature_meta=None):
+        raise NotImplementedError("FeatureLearner.fit must be implemented")
+
+    def transform(self, X):
+        raise NotImplementedError("FeatureLearner.transform must be implemented")
+
+    def fit_transform(self, X, y=None, feature_meta=None):
+        self.fit(X, y, feature_meta)
+        return self.transform(X)
+
+    # ---------- names & lineage ----------
+    def get_feature_names_out(self, input_features):
+        """
+        Return names for columns produced by transform().
+        Default: identity (no change).
+        """
+        return list(input_features)
+
+    def get_parent_map(self, output_features):
+        """
+        Map each produced feature -> list of parent input feature names.
+        Default: identity mapping.
+        """
+        return dict((name, [name]) for name in output_features)

streamline/p3_feature_learning/learner.py

@@ -0,0 +1,148 @@
+import os
+import glob
+import time
+import pickle
+import dask
+import pandas as pd
+from pathlib import Path
+from joblib import Parallel, delayed
+
+from streamline.utils.runners import runner_fn, num_cores
+from streamline.utils.cluster import get_cluster
+
+# concrete learners that implement the interface
+from .registry.pca import PCAFeatureLearner
+
+class FeatureLearnJob(object):
+    def __init__(
+        self,
+        cv_train_path,
+        cv_test_path,
+        experiment_path,
+        overwrite_cv=True,
+        outcome_label="Class",
+        instance_label=None,
+        random_state=None,
+        feature_learner="pca",
+        # PCA params
+        n_components=None,
+        whiten=False,
+        svd_solver="auto",
+        passthrough=False,
+        prefix="pca",
+    ):
+        self.cv_train_path = cv_train_path
+        self.cv_test_path = cv_test_path
+        self.experiment_path = experiment_path
+
+        self.overwrite_cv = overwrite_cv
+        self.outcome_label = outcome_label
+        self.instance_label = instance_label
+        self.random_state = random_state
+
+        self.feature_learner = feature_learner
+
+        # PCA params
+        self.n_components = n_components
+        self.whiten = whiten
+        self.svd_solver = svd_solver
+        self.passthrough = passthrough
+        self.prefix = prefix
+
+        self.dataset_name = None
+        self.cv_count = None
+        self.job_start_time = time.time()
+
+    def _make_learner(self):
+        if (self.feature_learner or "").lower() == "pca":
+            return PCAFeatureLearner(
+                component_id="pca",
+                random_state=self.random_state,
+                n_components=self.n_components,
+                whiten=self.whiten,
+                svd_solver=self.svd_solver,
+                passthrough=self.passthrough,
+                prefix=self.prefix,
+            )
+        raise ValueError("Unknown feature_learner: %s" % self.feature_learner)
+
+    def run(self):
+        data_train = pd.read_csv(self.cv_train_path, na_values="NA", sep=",")
+        data_test = pd.read_csv(self.cv_test_path, na_values="NA", sep=",")
+
+        self.dataset_name = self.cv_train_path.split("/")[-3]
+        self.cv_count = self.cv_train_path.split("/")[-1].split("_")[-2]
+
+        y_train = data_train[self.outcome_label]
+        y_test = data_test[self.outcome_label]
+
+        i_train = None; i_test = None
+        if self.instance_label is not None and self.instance_label in data_train.columns:
+            i_train = data_train[self.instance_label]
+            i_test = data_test[self.instance_label]
+
+        drop_cols = [self.outcome_label] + ([self.instance_label] if i_train is not None else [])
+        X_train = data_train.drop(drop_cols, axis=1)
+        X_test = data_test.drop(drop_cols, axis=1)
+
+        learner = self._make_learner()
+        learner.fit(X_train, y=None, feature_meta=None)
+        Z_train = learner.transform(X_train)
+        Z_test = learner.transform(X_test)
+
+        if i_train is None:
+            new_train = pd.concat([pd.DataFrame(y_train, columns=[self.outcome_label]), Z_train], axis=1)
+            new_test = pd.concat([pd.DataFrame(y_test, columns=[self.outcome_label]), Z_test], axis=1)
+        else:
+            new_train = pd.concat(
+                [pd.DataFrame(y_train, columns=[self.outcome_label]),
+                 pd.DataFrame(i_train, columns=[self.instance_label]),
+                 Z_train],
+                axis=1
+            )
+            new_test = pd.concat(
+                [pd.DataFrame(y_test, columns=[self.outcome_label]),
+                 pd.DataFrame(i_test, columns=[self.instance_label]),
+                 Z_test],
+                axis=1
+            )
+
+        fl_dir = os.path.join(self.experiment_path, self.dataset_name, "feature_learning")
+        if not os.path.exists(fl_dir):
+            os.makedirs(fl_dir)
+
+        with open(os.path.join(fl_dir, "feature_learner_%s_cv%s.pickle" % (self.feature_learner, str(self.cv_count))), "wb") as f:
+            pickle.dump(learner, f)
+
+        if hasattr(learner, "explained_variance_ratio_") and learner.explained_variance_ratio_ is not None:
+            evr_path = os.path.join(fl_dir, "pca_evr_cv" + str(self.cv_count) + ".csv")
+            pd.Series(learner.explained_variance_ratio_).to_csv(evr_path, index=False, header=["explained_variance_ratio"])
+
+        self.write_cv_files(new_train, new_test)
+
+        jobs_dir = os.path.join(self.experiment_path, "jobsCompleted")
+        if not os.path.exists(jobs_dir): os.makedirs(jobs_dir)
+        with open(os.path.join(jobs_dir, "job_feature_learning_" + self.dataset_name + "_" + str(self.cv_count) + ".txt"), "w") as f:
+            f.write("complete")
+
+        runtime_dir = os.path.join(self.experiment_path, self.dataset_name, "runtime")
+        if not os.path.exists(runtime_dir): os.makedirs(runtime_dir)
+        with open(os.path.join(runtime_dir, "runtime_feature_learning" + str(self.cv_count) + ".txt"), "w") as f:
+            f.write(str(time.time() - self.job_start_time))
+
+    def write_cv_files(self, data_train, data_test):
+        if self.overwrite_cv:
+            os.remove(self.cv_train_path)
+            os.remove(self.cv_test_path)
+        else:
+            dataset_dir = os.path.join(self.experiment_path, self.dataset_name)
+            os.rename(
+                self.cv_train_path,
+                os.path.join(dataset_dir, "CVDatasets", self.dataset_name + "_PreFL_" + str(self.cv_count) + "_Train.csv")
+            )
+            os.rename(
+                self.cv_test_path,
+                os.path.join(dataset_dir, "CVDatasets", self.dataset_name + "_PreFL_" + str(self.cv_count) + "_Test.csv")
+            )
+        data_train.to_csv(self.cv_train_path, index=False)
+        data_test.to_csv(self.cv_test_path, index=False)