fix: issue #60 — W-Dist センチネル廃止し Gudhi の実 Wasserstein に統一

tda_ml/metrics.py

@@ -32,24 +32,22 @@ def compute_recall_specificity_gmean_mcc_wdist(
     *,
     points: np.ndarray | None = None,
     gt_inliers: np.ndarray | None = None,
-    empty_pred_wdist: float = 9.99,
 ) -> tuple[float, float, float, float, float]:
     """
     Return the four classification metrics plus W-Dist.
 
-    W-Dist is computed only when ``points`` and ``gt_inliers`` are provided.
-    If every predicted label is outlier (no predicted inliers), returns
-    ``empty_pred_wdist`` (default ``9.99``), distinct from
-    :func:`tda_ml.persistence.compute_w_distance` returning ``999.0`` for an empty point cloud.
+    W-Dist is the H1 1-Wasserstein distance between persistence diagrams of
+    predicted inliers and ``gt_inliers``.  It is computed only when ``points``
+    and ``gt_inliers`` are provided; otherwise ``w_dist`` is ``0.0``.
+    An empty predicted-inlier set uses the standard OT distance
+    ``W(empty_diagram, PD(gt_inliers))`` — see
+    :func:`tda_ml.persistence.compute_w_distance`.
     """
     recall, specificity, gmean, mcc = compute_recall_specificity_gmean_mcc(
         labels_gt, labels_pred
     )
     w_dist = 0.0
     if points is not None and gt_inliers is not None:
         pred_inliers = points[np.asarray(labels_pred) == 0]
-        if len(pred_inliers) > 0:
-            w_dist = float(compute_w_distance(pred_inliers, gt_inliers))
-        else:
-            w_dist = float(empty_pred_wdist)
+        w_dist = float(compute_w_distance(pred_inliers, gt_inliers))
     return recall, specificity, gmean, mcc, w_dist

tda_ml/persistence.py

@@ -27,26 +27,23 @@ def wasserstein_h1(
 
 def compute_w_distance(points_pred, points_gt):
     """
-    Computes Wasserstein distance between persistence diagrams (dim 1) of two point clouds.
+    Wasserstein distance between H1 persistence diagrams of two point clouds.
 
     Args:
         points_pred: (N, 2) array or tensor
         points_gt: (M, 2) array or tensor
 
     Returns:
-        distance: float. Returns ``999.0`` when ``points_pred`` is empty (no diagram to match),
-        and ``0.0`` when ``points_gt`` is empty.
+        1-Wasserstein distance (``order=1``, ``internal_p=2``) via Gudhi.
+        Empty point clouds yield empty H1 diagrams; the distance is the
+        standard optimal-transport cost (including matching to the diagonal),
+        not a sentinel penalty.
     """
     if isinstance(points_pred, torch.Tensor):
         points_pred = points_pred.detach().cpu().numpy()
     if isinstance(points_gt, torch.Tensor):
         points_gt = points_gt.detach().cpu().numpy()
 
-    if len(points_pred) == 0:
-        return 999.0
-    if len(points_gt) == 0:
-        return 0.0
-
     def get_pd(pts):
         if len(pts) < 3:
             return []
@@ -59,33 +56,24 @@ def get_pd(pts):
     pd_pred = get_pd(points_pred)
     pd_gt = get_pd(points_gt)
 
-    if len(pd_pred) == 0 and len(pd_gt) == 0:
-        return 0.0
-    if len(pd_pred) == 0:
-        pd_pred = np.empty((0, 2))
-    if len(pd_gt) == 0:
-        pd_gt = np.empty((0, 2))
-
     return wasserstein_h1(pd_pred, pd_gt, order=1, internal_p=2)
 
 
 def compute_bottleneck_distance(points_pred, points_gt):
     """
-    Computes Bottleneck distance between persistence diagrams (dim 1).
+    Bottleneck distance between H1 persistence diagrams of two point clouds.
 
     Hold note (B-3 / GitHub #24): currently unused in active tda_ml/scripts/tests,
     but referenced in legacy code under `trash/`; keep until removal is fully validated.
+
+    Empty point clouds yield empty H1 diagrams; distance is computed via Gudhi
+    without sentinel penalties (same policy as :func:`compute_w_distance`).
     """
     if isinstance(points_pred, torch.Tensor):
         points_pred = points_pred.detach().cpu().numpy()
     if isinstance(points_gt, torch.Tensor):
         points_gt = points_gt.detach().cpu().numpy()
 
-    if len(points_pred) == 0:
-        return 99.0
-    if len(points_gt) == 0:
-        return 0.0
-
     def get_pd(pts):
         if len(pts) < 3:
             return []
@@ -98,15 +86,12 @@ def get_pd(pts):
     pd_pred = get_pd(points_pred)
     pd_gt = get_pd(points_gt)
 
-    if len(pd_pred) == 0 and len(pd_gt) == 0:
-        return 0.0
     if len(pd_pred) == 0:
         pd_pred = np.empty((0, 2))
     if len(pd_gt) == 0:
         pd_gt = np.empty((0, 2))
 
-    dist = gudhi.bottleneck_distance(pd_pred, pd_gt)
-    return dist
+    return float(gudhi.bottleneck_distance(pd_pred, pd_gt))
 
 
 def compute_betti_error(points_pred, points_gt):

tests/test_metrics.py

@@ -6,6 +6,7 @@
     compute_recall_specificity_gmean_mcc,
     compute_recall_specificity_gmean_mcc_wdist,
 )
+from tda_ml.persistence import compute_w_distance
 
 
 class TestMetrics(unittest.TestCase):
@@ -47,12 +48,19 @@ def test_compute_metrics_with_wdist_empty_pred_inliers(self):
         y_true = np.array([0, 0, 1, 1], dtype=int)
         y_pred = np.array([1, 1, 1, 1], dtype=int)
         points = np.array([[0.0, 0.0], [0.5, 0.2], [1.0, 1.0], [-1.0, -1.0]])
-        gt_inliers = np.array([[0.0, 0.0], [0.5, 0.2]])
+        gt_inliers = self._circle_gt_inliers()
 
         _, _, _, _, wdist = compute_recall_specificity_gmean_mcc_wdist(
-            y_true, y_pred, points=points, gt_inliers=gt_inliers, empty_pred_wdist=9.99
+            y_true, y_pred, points=points, gt_inliers=gt_inliers
         )
-        self.assertAlmostEqual(wdist, 9.99, places=6)
+        expected = compute_w_distance(np.empty((0, 2)), gt_inliers)
+        self.assertAlmostEqual(wdist, expected, places=6)
+        self.assertTrue(np.isfinite(wdist))
+
+    @staticmethod
+    def _circle_gt_inliers(n: int = 12) -> np.ndarray:
+        theta = np.linspace(0, 2 * np.pi, n, endpoint=False)
+        return np.stack([np.cos(theta), np.sin(theta)], axis=1)
 
     def test_compute_metrics_wdist_extreme_scale_inputs(self):
         """極小/極大スケールの点群でも指標計算が有限値で返る。"""

tests/test_persistence.py

@@ -0,0 +1,36 @@
+import unittest
+
+import numpy as np
+
+from tda_ml.persistence import compute_bottleneck_distance, compute_w_distance
+
+
+class TestPersistenceWasserstein(unittest.TestCase):
+    def _circle_points(self, n: int = 20) -> np.ndarray:
+        theta = np.linspace(0, 2 * np.pi, n, endpoint=False)
+        return np.stack([np.cos(theta), np.sin(theta)], axis=1)
+
+    def test_empty_pred_returns_finite_wasserstein_not_sentinel(self):
+        circle = self._circle_points()
+        w = compute_w_distance([], circle)
+        self.assertTrue(np.isfinite(w))
+        self.assertLess(w, 10.0)
+
+    def test_empty_gt_symmetric_with_empty_pred(self):
+        circle = self._circle_points()
+        w_pred_empty = compute_w_distance([], circle)
+        w_gt_empty = compute_w_distance(circle, [])
+        self.assertAlmostEqual(w_pred_empty, w_gt_empty, places=6)
+
+    def test_both_empty_point_clouds(self):
+        self.assertEqual(compute_w_distance([], []), 0.0)
+
+    def test_bottleneck_empty_pred_not_sentinel(self):
+        circle = self._circle_points()
+        b = compute_bottleneck_distance([], circle)
+        self.assertTrue(np.isfinite(b))
+        self.assertLess(b, 10.0)
+
+
+if __name__ == "__main__":
+    unittest.main()