Allow conditional independence tests to handle zero variance inputs

hyppo/conditional/_utils.py

@@ -6,13 +6,14 @@
 class _CheckInputs:
     """Checks inputs for all independence tests"""
 
-    def __init__(self, x, y, z, reps=None, max_dims=None, ignore_z_var=False):
+    def __init__(self, x, y, z=None, reps=None, max_dims=None, ignore_z_var=False):
         self.x = x
         self.y = y
         self.z = z
         self.reps = reps
         self.max_dims = max_dims
         self.ignore_z_var = ignore_z_var # to allow for constant z input
+        self.is_zero_variance = False
 
     def __call__(self):
         check_ndarray_xyz(self.x, self.y, self.z)
@@ -59,8 +60,8 @@ def check_dim_xyz(self, max_dims):
                 raise ValueError(
                     f"x, y, z must have be univariate and have shape [n,{max_dims}]"
                 )
-
         self._check_nd_indeptest()
+        self._check_variance()
 
         return self.x, self.y, self.z
 
@@ -86,10 +87,9 @@ def _check_min_samples(self):
 
     def _check_variance(self):
         if np.var(self.x) == 0:
-        # or np.var(self.y) == 0 or np.var(self.z) == 0:
-            raise ValueError("Test cannot be run. Input array x has 0 variance.")
+            self.is_zero_variance = True
         if np.var(self.y) == 0:
-            raise ValueError("Test cannot be run. Input array y has 0 variance")
-        if not self.ignore_z_var:
+            self.is_zero_variance = True
+        if not self.ignore_z_var and self.z is not None:
             if np.var(self.z) == 0:
-                raise ValueError("Test cannot be run. Input array z has 0 variance")
+                self.is_zero_variance = True

hyppo/conditional/cdcorr.py

@@ -172,6 +172,12 @@ def test(
         check_input = _CheckInputs(x, y, z, reps=reps, ignore_z_var=True)
         x, y, z = check_input()
 
+        if check_input.is_zero_variance:
+            self.stat = 0.0
+            self.pvalue = 1.0
+            self.null_dist = None
+            return ConditionalIndependenceTestOutput(0.0, 1.0)
+
         if not self.is_distance:
             x, y = compute_dist(x, y, metric=self.compute_distance, **self.kwargs)
             z = self._compute_kde(z)

hyppo/conditional/kci.py

@@ -112,6 +112,10 @@ def test(self, x, y):
 
         T = len(y)
 
+        if np.var(x) == 0 or np.var(y) == 0:
+            self.stat = 0.0
+            return ConditionalIndependenceTestOutput(0.0, 1.0)
+
         Kx, Ky = self.compute_kern(x, y)
         stat = self.statistic(x, y)
 

hyppo/conditional/pcorr.py

@@ -148,6 +148,12 @@ def test(
         check_input = _CheckInputs(x, y, z, reps=reps, max_dims=1)
         x, y, z = check_input()
 
+        if check_input.is_zero_variance:
+            self.stat = 0.0
+            self.pvalue = 1.0
+            self.null_dist = None
+            return ConditionalIndependenceTestOutput(0.0, 1.0)
+
         if auto:  # run t-stat
             stat = self.statistic(x, y, z)
 

hyppo/conditional/pdcorr.py

@@ -181,6 +181,12 @@ def test(
         check_input = _CheckInputs(x, y, z, reps=reps)
         x, y, z = check_input()
 
+        if check_input.is_zero_variance:
+            self.stat = 0.0
+            self.pvalue = 1.0
+            self.null_dist = None
+            return ConditionalIndependenceTestOutput(0.0, 1.0)
+
         if not self.is_distance:
             x, y, z = compute_dist(x, y, z, metric=self.compute_distance, **self.kwargs)
             self.is_distance = True

hyppo/conditional/tests/test_utils.py

@@ -64,7 +64,9 @@ def test_constant_input(self):
         y = np.arange(20).reshape(-1, 1)
         z = np.ones(20).reshape(-1, 1)
 
-        assert_raises(ValueError, _CheckInputs(x, y, z, ignore_z_var=False))
+        check_input = _CheckInputs(x, y, z, ignore_z_var=False)
+        check_input()
+        assert check_input.is_zero_variance == True
 
         try:
             _CheckInputs(x, y, z, ignore_z_var=True)