minor fixes, additional tests

SamoraHunter · SamoraHunter · commit f87218baabfd · 2025-09-16T18:09:33.000+01:00
diff --git a/ml_grid/pipeline/data_constant_columns.py b/ml_grid/pipeline/data_constant_columns.py
@@ -99,15 +99,10 @@ def remove_constant_columns_with_debug(
         if verbosity > 0:
             print(f"Constant columns in X_train: {list(constant_columns_train)}")
 
-        test_variances = X_test.var(axis=0)
-        if verbosity > 1:
-            print(f"Variance of X_test columns:\n{test_variances}")
-
-        constant_columns_test = test_variances[test_variances == 0].index
-        if verbosity > 0:
-            print(f"Constant columns in X_test: {list(constant_columns_test)}")
-
-        constant_columns = constant_columns_train.union(constant_columns_test)
+        # A column is constant if it has no variance in the training set.
+        # We should not consider the test set variance, as a small test set
+        # might misleadingly have constant columns.
+        constant_columns = constant_columns_train
 
         X_train = X_train.loc[:, ~X_train.columns.isin(constant_columns)]
         X_test = X_test.loc[:, ~X_test.columns.isin(constant_columns)]
@@ -127,14 +122,10 @@ def remove_constant_columns_with_debug(
         if verbosity > 0:
             print(f"Constant feature indices in X_train: {list(constant_indices_train)}")
 
-        test_variances = X_test.var(axis=var_axis)
-        constant_indices_test = np.where(test_variances == 0)[0]
-        if verbosity > 0:
-            print(f"Constant feature indices in X_test: {list(constant_indices_test)}")
-
-        # Combine indices of constant features from both train and test sets
-        constant_indices = np.union1d(constant_indices_train, constant_indices_test)
-
+        # A feature is constant if it has no variance in the training set.
+        # We should not consider the test set variance, as a small test set
+        # might misleadingly have constant features.
+        constant_indices = constant_indices_train
         # Create a boolean mask for features to keep
         num_features = X_train.shape[1]
         keep_mask = np.ones(num_features, dtype=bool)
diff --git a/ml_grid/pipeline/data_feature_importance_methods.py b/ml_grid/pipeline/data_feature_importance_methods.py
@@ -48,20 +48,17 @@ def handle_feature_importance_methods(
         feature_method = ml_grid_object.local_param_dict.get("feature_selection_method")
 
         if feature_method == "anova" or feature_method is None:
-            print("feature_method ANOVA")
-
-            features = feature_methods.getNfeaturesANOVAF(
-                self, n=target_n_features, X_train=X_train, y_train=y_train
-            )
+            print("feature_method ANOVA") 
+            fm = feature_methods() 
+            features = fm.getNfeaturesANOVAF(n=target_n_features, X_train=X_train, y_train=y_train)
 
         elif feature_method == "markov_blanket":
-            print("feature method Markov")
-
-            features = feature_methods.getNFeaturesMarkovBlanket(
-                self, n=target_n_features, X_train=X_train, y_train=y_train
-            )
+            print("feature method Markov") 
+            fm = feature_methods() 
+            features = fm.getNFeaturesMarkovBlanket(n=target_n_features, X_train=X_train, y_train=y_train)
 
         print(f"target_n_features: {target_n_features}")
+        print(f"Selected features: {features}")
 
         X_train = X_train[features]
 
diff --git a/ml_grid/pipeline/data_feature_methods.py b/ml_grid/pipeline/data_feature_methods.py
@@ -114,6 +114,10 @@ def getNFeaturesMarkovBlanket(
                 "X_train must be a pandas DataFrame for getNFeaturesMarkovBlanket."
             )
         original_columns = X_train.columns
+
+        # Ensure y_train is a pandas Series, as expected by PyImpetus internally
+        if not isinstance(y_train, pd.Series):
+            y_train = pd.Series(y_train)
         
         # Initialize the PyImpetus object with desired parameters
         model = PPIMBC(model=SVC(random_state=27, class_weight="balanced", kernel=svc_kernel),
@@ -132,15 +136,23 @@ def getNFeaturesMarkovBlanket(
         model.fit(X_train.values, y_train)
         
         # Get the feature indices from the Markov blanket (MB)
-        feature_indices = model.MB
+        selected_features = model.MB
 
-        # Map indices back to original column names and truncate by n
-        feature_names = [original_columns[i] for i in feature_indices][:n]
+        # PyImpetus can return column names (str) or indices (int).
+        # We need to handle both cases to get the final list of feature names.
+        if all(isinstance(f, int) for f in selected_features):
+            # It returned indices, so map them to names
+            feature_names = [original_columns[i] for i in selected_features][:n]
+        else:
+            # It returned names directly
+            feature_names = list(selected_features)[:n]
 
         # Fallback: If feature selection returns an empty list, but the model found features,
         # return the single most important one. This prevents pipeline failure.
-        if not feature_names and feature_indices:
-            feature_names = [original_columns[feature_indices[0]]]
+        if not feature_names and selected_features:
+            # Re-evaluate the first selected feature to ensure it's a valid name
+            first_feature = selected_features[0]
+            feature_names = [original_columns[first_feature] if isinstance(first_feature, int) else first_feature]
         
         return feature_names
 
diff --git a/tests/test_column_names.py b/tests/test_column_names.py
@@ -0,0 +1,82 @@
+import unittest
+import pandas as pd
+from ml_grid.pipeline.column_names import get_pertubation_columns, filter_substring_list
+from ml_grid.util.global_params import global_parameters
+
+class TestColumnNames(unittest.TestCase):
+
+    def setUp(self):
+        """Set up common variables for tests."""
+        self.all_df_columns = [
+            'age', 'male', 'bmi_val', 'census_A', 'blood_test_mean', # Corrected to bmi_ and census_
+            'diag_order_num_diagnostic-order', 'drug_order_num_drug-order', 'annotation_1_count',
+            'meta_sp_annotation_1_count_subject_present', 'annotation_mrc_1_count_mrc_cs',
+            'meta_sp_annotation_mrc_1_count_subject_present_mrc_cs', 'core_02_feature',
+            'bed_feature', 'vte_status_feature', 'hosp_site_A', # Corrected to hosp_site_
+            'core_resus_feature', 'news_resus_feature', # Corrected to news_
+            'date_time_stamp_2022', 'ConsultantCode_X', 'outcome_var_1',
+            'some_col__index_level_0', 'Unnamed: 0'
+        ]
+        self.drop_term_list = ['bad_term']
+        # Mute verbose output for tests
+        global_parameters.verbose = 0
+
+    def test_filter_substring_list(self):
+        """Test the filter_substring_list utility function."""
+        string_list = ['test_mean', 'test_median', 'other_val', 'bmi_mean']
+        substr_list = ['_mean', '_median']
+        # 'bmi_mean' should be excluded by the function's logic
+        expected = ['test_mean', 'test_median']
+        result = filter_substring_list(string_list, substr_list)
+        self.assertCountEqual(result, expected)
+
+    def test_get_pertubation_columns_selects_all(self):
+        """Test that all categories are selected when flags are True."""
+        local_param_dict = {
+            "outcome_var_n": 1,
+            "data": {
+                'age': True, 'sex': True, 'bmi': True, 'ethnicity': True,
+                'bloods': True, 'diagnostic_order': True, 'drug_order': True,
+                'annotation_n': True, 'meta_sp_annotation_n': True,
+                'annotation_mrc_n': True, 'meta_sp_annotation_mrc_n': True,
+                'core_02': True, 'bed': True, 'vte_status': True,
+                'hosp_site': True, 'core_resus': True, 'news': True,
+                'date_time_stamp': True, 'appointments': True
+            }
+        }
+        pert_cols, _ = get_pertubation_columns(
+            self.all_df_columns, local_param_dict, self.drop_term_list
+        )
+        # Expect all columns except outcome and special drop columns
+        self.assertEqual(len(pert_cols), 19)
+
+    def test_get_pertubation_columns_selects_none(self):
+        """Test that no categories are selected when flags are False."""
+        local_param_dict = {
+            "outcome_var_n": 1,
+            "data": {key: False for key in [
+                'age', 'sex', 'bmi', 'ethnicity', 'bloods', 'diagnostic_order',
+                'drug_order', 'annotation_n', 'meta_sp_annotation_n',
+                'annotation_mrc_n', 'meta_sp_annotation_mrc_n', 'core_02',
+                'bed', 'vte_status', 'hosp_site', 'core_resus', 'news',
+                'date_time_stamp', 'appointments'
+            ]}
+        }
+        pert_cols, _ = get_pertubation_columns(
+            self.all_df_columns, local_param_dict, self.drop_term_list
+        )
+        self.assertEqual(len(pert_cols), 0)
+
+    def test_drop_list_population(self):
+        """Test that the initial drop_list is populated correctly."""
+        local_param_dict = {"outcome_var_n": 1, "data": {}}
+        _, drop_list = get_pertubation_columns(
+            self.all_df_columns, local_param_dict, self.drop_term_list
+        )
+        # Should contain '__index_level' and 'Unnamed:' columns
+        self.assertIn('some_col__index_level_0', drop_list)
+        self.assertIn('Unnamed: 0', drop_list)
+        self.assertEqual(len(drop_list), 2)
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/test_data_constant_columns.py b/tests/test_data_constant_columns.py
@@ -0,0 +1,105 @@
+import unittest
+import pandas as pd
+import numpy as np
+from ml_grid.pipeline.data_constant_columns import remove_constant_columns, remove_constant_columns_with_debug
+
+class TestRemoveConstantColumns(unittest.TestCase):
+
+    def test_remove_constant_columns_with_constants(self):
+        """Test that constant columns are identified and added to the drop list."""
+        df = pd.DataFrame({
+            'a': [1, 2, 3],
+            'b': [5, 5, 5],
+            'c': ['x', 'y', 'z'],
+            'd': [0, 0, 0]
+        })
+        initial_drop_list = ['e']
+        updated_drop_list = remove_constant_columns(df, initial_drop_list.copy(), verbose=0)
+        self.assertCountEqual(updated_drop_list, ['e', 'b', 'd'])
+
+    def test_remove_constant_columns_no_constants(self):
+        """Test that no columns are added when there are no constants."""
+        df = pd.DataFrame({
+            'a': [1, 2, 3],
+            'b': [4, 5, 6]
+        })
+        updated_drop_list = remove_constant_columns(df, [], verbose=0)
+        self.assertEqual(updated_drop_list, [])
+
+    def test_remove_constant_columns_empty_df(self):
+        """Test with an empty DataFrame."""
+        df = pd.DataFrame()
+        updated_drop_list = remove_constant_columns(df, [], verbose=0)
+        self.assertEqual(updated_drop_list, [])
+
+class TestRemoveConstantColumnsWithDebug(unittest.TestCase):
+
+    def test_pandas_2d_constant_in_train(self):
+        """Test with a constant column in the training DataFrame."""
+        X_train = pd.DataFrame({'a': [1, 2, 3], 'b': [5, 5, 5]})
+        X_test = pd.DataFrame({'a': [4, 5, 6], 'b': [7, 8, 9]})
+        X_test_orig = X_test.copy()
+        
+        train_out, test_out, orig_out = remove_constant_columns_with_debug(
+            X_train, X_test, X_test_orig, verbosity=0
+        )
+        
+        self.assertNotIn('b', train_out.columns)
+        self.assertNotIn('b', test_out.columns)
+        self.assertNotIn('b', orig_out.columns)
+        self.assertIn('a', train_out.columns)
+
+    def test_pandas_2d_constant_in_test(self):
+        """Test that a column constant only in the test set is NOT removed."""
+        X_train = pd.DataFrame({'a': [1, 2, 3], 'b': [7, 8, 9]})
+        X_test = pd.DataFrame({'a': [4, 5, 6], 'b': [5, 5, 5]})
+        X_test_orig = X_test.copy()
+
+        train_out, test_out, orig_out = remove_constant_columns_with_debug(
+            X_train, X_test, X_test_orig, verbosity=0
+        )
+        
+        # 'b' should NOT be removed as it has variance in the training set.
+        self.assertIn('b', train_out.columns)
+        self.assertIn('b', test_out.columns)
+        self.assertIn('b', orig_out.columns)
+        self.assertIn('a', train_out.columns)
+
+    def test_numpy_2d(self):
+        """Test with 2D numpy arrays."""
+        X_train = np.array([[1, 5], [2, 5], [3, 5]])
+        X_test = np.array([[4, 7], [5, 8], [6, 9]])
+        X_test_orig = X_test.copy()
+
+        train_out, test_out, orig_out = remove_constant_columns_with_debug(
+            X_train, X_test, X_test_orig, verbosity=0
+        )
+
+        self.assertEqual(train_out.shape[1], 1)
+        self.assertEqual(test_out.shape[1], 1)
+        self.assertEqual(orig_out.shape[1], 1)
+        self.assertTrue(np.array_equal(train_out, np.array([[1], [2], [3]])))
+
+    def test_numpy_3d_time_series(self):
+        """Test with 3D numpy arrays for time series data."""
+        # Shape: (samples, features, timesteps)
+        X_train = np.array([
+            [[1, 1], [5, 5], [1, 1]],  # Sample 1: Feature 1 varies, Feature 2 is constant
+            [[2, 2], [5, 5], [2, 2]],  # Sample 2
+        ])
+        X_test = np.array([
+            [[3, 3], [9, 9], [3, 3]],
+        ])
+        X_test_orig = X_test.copy()
+
+        train_out, test_out, orig_out = remove_constant_columns_with_debug(
+            X_train, X_test, X_test_orig, verbosity=0
+        )
+
+        # Expecting feature 1 (index 0) and 2 (index 2) to be kept, feature 2 (index 1) to be dropped
+        self.assertEqual(train_out.shape[1], 2)
+        self.assertEqual(test_out.shape[1], 2)
+        self.assertEqual(orig_out.shape[1], 2)
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/test_markov_blanket_feature_reduction.py b/tests/test_markov_blanket_feature_reduction.py
@@ -1,7 +1,8 @@
 import unittest
-from ml_grid.pipeline.data_feature_importance_methods import (
-    feature_importance_methods,
-)  # Replace 'your_module' with the actual module name where MyClass is defined
+import pandas as pd
+from ml_grid.pipeline.data_feature_methods import (
+    feature_methods,
+)
 from sklearn.datasets import make_classification
 from sklearn.model_selection import train_test_split
 
@@ -15,14 +16,18 @@ def test_number_of_features(self):
 
         # Split the data into training and testing sets
         X_train, _, y_train, _ = train_test_split(X, y, test_size=0.33, random_state=42)
+        
+        # Convert numpy array to pandas DataFrame as the method expects it
+        X_train = pd.DataFrame(X_train, columns=[f'feature_{i}' for i in range(X_train.shape[1])])
 
-        # Create an instance of MyClass
-        my_instance = feature_importance_methods()
+        # Create an instance of the correct class
+        my_instance = feature_methods()
 
-        # Call the function to get the top 3 features
+        # Call the function to get the top 5 features
         top_features = my_instance.getNFeaturesMarkovBlanket(5, X_train, y_train)
 
-        # Assert that the number of features returned is approximately 5
+        # Assert that the number of features returned is reasonable.
+        # The method may return fewer than n features.
         self.assertAlmostEqual(len(top_features), 3, delta=1)
 
 
diff --git a/tests/test_synthetic_data_generator.py b/tests/test_synthetic_data_generator.py
@@ -0,0 +1,48 @@
+import unittest
+import pandas as pd
+from ml_grid.util.synthetic_data_generator import generate_time_series, columns
+
+
+class TestGenerateTimeSeries(unittest.TestCase):
+
+    def setUp(self):
+        """Set up a test DataFrame for all test methods."""
+        self.num_clients = 5
+        self.num_rows_per_client = 10
+        self.df = generate_time_series(self.num_clients, self.num_rows_per_client)
+
+    def test_output_is_dataframe(self):
+        """Test that the output is a pandas DataFrame."""
+        self.assertIsInstance(self.df, pd.DataFrame)
+
+    def test_dataframe_shape(self):
+        """Test the shape of the generated DataFrame."""
+        expected_rows = self.num_clients * self.num_rows_per_client
+        expected_cols = len(columns)
+        self.assertEqual(self.df.shape, (expected_rows, expected_cols))
+
+    def test_number_of_unique_clients(self):
+        """Test that the number of unique clients is correct."""
+        self.assertEqual(self.df['client_idcode'].nunique(), self.num_clients)
+
+    def test_timestamp_column_type(self):
+        """Test that the timestamp column has the correct data type."""
+        self.assertTrue(pd.api.types.is_datetime64_any_dtype(self.df['timestamp']))
+
+    def test_sorting(self):
+        """Test that the DataFrame is sorted by client_idcode and timestamp."""
+        # Check if each client's timestamps are sorted
+        for client_id in self.df['client_idcode'].unique():
+            client_df = self.df[self.df['client_idcode'] == client_id]
+            self.assertTrue(client_df['timestamp'].is_monotonic_increasing)
+
+    def test_outcome_variable_is_binary(self):
+        """Test that the outcome variable is binary (0 or 1)."""
+        outcome_col = 'outcome_var_1'
+        self.assertIn(outcome_col, self.df.columns)
+        unique_outcomes = self.df[outcome_col].unique()
+        self.assertTrue(all(item in [0, 1] for item in unique_outcomes))
+
+
+if __name__ == '__main__':
+    unittest.main()
