working through last rocauc bug

Author: czaloom

src/valor_lite/cache/compute.py

@@ -107,7 +107,6 @@ def sort(
     table_sort_override : Callable[[pa.Table], pa.Table], optional
         Option to override sort function for singular cache fragments.
     """
-
     if source.count_tables() == 1:
         for tbl in source.iterate_tables(columns=columns):
             if table_sort_override is not None:

src/valor_lite/classification/computation.py

@@ -7,12 +7,15 @@
 
 
 def compute_rocauc(
-    ids: NDArray[np.int64],
-    scores: NDArray[np.float64],
     rocauc: NDArray[np.float64],
+    array: NDArray[np.float64],
     gt_count_per_label: NDArray[np.uint64],
     pd_count_per_label: NDArray[np.uint64],
     n_labels: int,
+    accumulated_tp: NDArray[np.uint64],
+    accumulated_fp: NDArray[np.uint64],
+    prev_fpr: NDArray[np.float64],
+    prev_tpr: NDArray[np.float64],
 ) -> tuple[NDArray[np.float64], NDArray[np.uint64], NDArray[np.uint64]]:
     """
     Compute ROCAUC.
@@ -44,9 +47,9 @@ def compute_rocauc(
     NDArray[np.uint64]
         Final cumulative sum for TP's. Used as intermediate in chunking operations.
     """
-    gt_labels = ids[:, 1]
-    pd_labels = ids[:, 2]
-    mask_matching_labels = np.isclose(gt_labels, pd_labels)
+    pd_labels = array[:, 0].astype(np.int64)
+    scores = array[:, 1]
+    mask_matching_labels = array[:, 2] > 0.5
 
     positive_count = gt_count_per_label
     negative_count = pd_count_per_label - gt_count_per_label
@@ -57,44 +60,92 @@ def compute_rocauc(
         if pd_count_per_label[label_idx] == 0 or n_masked_pds == 0:
             continue
 
-        true_positives = np.zeros(n_masked_pds, dtype=np.uint64)
-        false_positives = np.zeros_like(true_positives)
-        tp_scores = np.zeros_like(true_positives, dtype=np.float64)
-
         true_positives = mask_matching_labels[mask_pds]
         false_positives = ~mask_matching_labels[mask_pds]
         tp_scores = scores[mask_pds]
 
-        cumulative_fp = np.cumsum(false_positives)
-        cumulative_tp = np.cumsum(true_positives)
+        cumulative_fp = np.cumsum(false_positives) + accumulated_fp[label_idx]
+        cumulative_tp = np.cumsum(true_positives) + accumulated_tp[label_idx]
 
-        fpr = np.zeros_like(true_positives, dtype=np.float64)
+        accumulated_fp[label_idx] = cumulative_fp[-1]
+        accumulated_tp[label_idx] = cumulative_tp[-1]
+
+        fpr = np.zeros(n_masked_pds, dtype=np.float64)
         np.divide(
             cumulative_fp,
             negative_count[label_idx],
             where=negative_count[label_idx] > 0,
             out=fpr,
         )
-        tpr = np.zeros_like(true_positives, dtype=np.float64)
+        tpr = np.zeros(n_masked_pds, dtype=np.float64)
         np.divide(
             cumulative_tp,
             positive_count[label_idx],
             where=positive_count[label_idx] > 0,
             out=tpr,
         )
 
+        if prev_fpr[label_idx] > -0.5 and prev_tpr[label_idx] > -0.5:
+            fpr = np.concatenate([prev_fpr[label_idx:label_idx+1], fpr])
+            tpr = np.concatenate([prev_tpr[label_idx:label_idx+1], tpr])
+
         # sort by -tpr, -score
         indices = np.lexsort((-tpr, -tp_scores))
         fpr = fpr[indices]
         tpr = tpr[indices]
 
+        sfpr = fpr.copy()
+        stpr = tpr.copy()
+
         # running max of tpr
         np.maximum.accumulate(tpr, out=tpr)
 
+
+        prev_fpr[label_idx] = fpr[-1]
+        prev_tpr[label_idx] = tpr[-1]
+
         # compute rocauc
-        rocauc[label_idx] = npc.trapezoid(x=fpr, y=tpr, axis=0)
+        rocauc[label_idx] += npc.trapezoid(x=fpr, y=tpr, axis=0)
+
+        print()
+        print(label_idx, rocauc[label_idx])
+        print("====")
+        print(
+            f"{'FP':4}",
+            f"{'TP':4}",
+            f"{'CFP':4}",
+            f"{'CTP':4}",
+            f"{'FPR':4}",
+            f"{'TPR':4}",
+            f"{'SFPR':4}",
+            f"{'STPR':4}",
+            f"{'SCO':4}",
+        )
+        for f, t, af, at, fr, tr, sf, st, s in zip(
+            false_positives,
+            true_positives,
+            cumulative_fp,
+            cumulative_tp,
+            fpr, 
+            tpr, 
+            sfpr,
+            stpr,
+            tp_scores,
+        ):
+            print(
+                f"{f:.2f}", 
+                f"{t:.2f}",
+                f"{af:.2f}", 
+                f"{at:.2f}", 
+                f"{fr:.2f}", 
+                f"{tr:.2f}", 
+                f"{sf:.2f}", 
+                f"{st:.2f}", 
+                f"{s:.2f}",
+            )
+            
 
-    return rocauc, None, None
+    return rocauc, accumulated_fp, accumulated_tp, prev_fpr, prev_tpr
 
 
 def compute_counts(

src/valor_lite/classification/evaluator.py

@@ -37,15 +37,15 @@ class Evaluator(Base):
     def __init__(
         self,
         reader: MemoryCacheReader | FileCacheReader,
-        sorted_reader: MemoryCacheReader | FileCacheReader,
+        rocauc_reader: MemoryCacheReader | FileCacheReader,
         info: EvaluatorInfo,
         label_counts: NDArray[np.uint64],
         index_to_label: dict[int, str],
         path: str | Path | None,
     ):
         self._path = Path(path) if path else None
         self._reader = reader
-        self._sorted_reader = sorted_reader
+        self._rocauc_reader = rocauc_reader
         self._info = info
         self._label_counts = label_counts
         self._index_to_label = index_to_label
@@ -82,8 +82,8 @@ def load(
 
         # load cache
         reader = FileCacheReader.load(cls._generate_cache_path(path))
-        sorted_reader = FileCacheReader.load(
-            cls._generate_sorted_cache_path(path)
+        rocauc_reader = FileCacheReader.load(
+            cls._generate_rocauc_cache_path(path)
         )
 
         # build evaluator meta
@@ -101,7 +101,7 @@ def load(
         return cls(
             path=path,
             reader=reader,
-            sorted_reader=sorted_reader,
+            rocauc_reader=rocauc_reader,
             info=info,
             label_counts=label_counts,
             index_to_label=index_to_label,
@@ -216,8 +216,7 @@ def delete(self):
         if self._path and self._path.exists():
             self.delete_at_path(self._path)
 
-    @staticmethod
-    def iterate_values(reader: MemoryCacheReader | FileCacheReader):
+    def iterate_values(self):
         columns = [
             "datum_id",
             "gt_label_id",
@@ -226,7 +225,7 @@ def iterate_values(reader: MemoryCacheReader | FileCacheReader):
             "winner",
             "match",
         ]
-        for tbl in reader.iterate_tables(columns=columns):
+        for tbl in self._reader.iterate_tables(columns=columns):
             ids = np.column_stack(
                 [
                     tbl[col].to_numpy()
@@ -242,11 +241,8 @@ def iterate_values(reader: MemoryCacheReader | FileCacheReader):
             matches = tbl["match"].to_numpy()
             yield ids, scores, winners, matches
 
-    @staticmethod
-    def iterate_values_with_tables(
-        reader: MemoryCacheReader | FileCacheReader,
-    ):
-        for tbl in reader.iterate_tables():
+    def iterate_values_with_tables(self):
+        for tbl in self._reader.iterate_tables():
             ids = np.column_stack(
                 [
                     tbl[col].to_numpy()
@@ -266,13 +262,6 @@ def compute_rocauc(self) -> dict[MetricType, list[Metric]]:
         """
         Compute ROCAUC.
 
-        Parameters
-        ----------
-        rows_per_chunk : int, default=10_000
-            The number of sorted rows to return in each chunk.
-        read_batch_size : int, default=1_000
-            The maximum number of rows to load in-memory per file.
-
         Returns
         -------
         dict[MetricType, list[Metric]]
@@ -281,42 +270,42 @@ def compute_rocauc(self) -> dict[MetricType, list[Metric]]:
         n_labels = self.info.number_of_labels
 
         rocauc = np.zeros(n_labels, dtype=np.float64)
-        cumulative_fp = np.zeros(n_labels, dtype=np.uint64)
-        cumulative_tp = np.zeros(n_labels, dtype=np.uint64)
-
-        tpr = np.zeros(n_labels, dtype=np.float64)
-        fpr = np.zeros(n_labels, dtype=np.float64)
+        accumulated_fp = np.zeros(n_labels, dtype=np.uint64)
+        accumulated_tp = np.zeros(n_labels, dtype=np.uint64)
 
-        positive_count = self._label_counts[:, 0]
-        negative_count = self._label_counts[:, 1] - self._label_counts[:, 0]
+        prev_tpr = np.ones(n_labels, dtype=np.float64) * -1
+        prev_fpr = np.ones(n_labels, dtype=np.float64) * -1
 
-        for ids, scores, winners, matches in self.iterate_values(
-            self._sorted_reader
-        ):
-            for id_row, score in zip(ids, scores):
-                glabel = id_row[1]
-                plabel = id_row[2]
-                if glabel < 0 or plabel < 0:
-                    continue
-                elif glabel == plabel:
-                    cumulative_tp[plabel] += 1
-                    tpr_new = cumulative_tp[plabel] / positive_count[plabel]
-                    tpr[plabel]
-                else:
-                    cumulative_fp[plabel] += 1
-
-            batch_rocauc, cumulative_fp, cumulative_tp = compute_rocauc(
-                ids=ids,
-                scores=scores,
+        for loopid, array in enumerate(self._rocauc_reader.iterate_arrays(
+            numeric_columns=[
+                "pd_label_id",
+                "score",
+                "match",
+            ]
+        )):
+            # for id_row, score in zip(ids, scores):
+            #     glabel = id_row[1]
+            #     plabel = id_row[2]
+            #     if glabel < 0 or plabel < 0:
+            #         continue
+            #     elif glabel == plabel:
+            #         cumulative_tp[plabel] += 1
+            #         tpr_new = cumulative_tp[plabel] / positive_count[plabel]
+            #         tpr[plabel]
+            #     else:
+            #         cumulative_fp[plabel] += 1
+            print("loop", loopid)
+            rocauc, accumulated_fp, accumulated_tp, prev_fpr, prev_tpr = compute_rocauc(
                 rocauc=rocauc,
-                cumulative_fp=cumulative_fp,
-                cumulative_tp=cumulative_tp,
+                array=array,
                 gt_count_per_label=self._label_counts[:, 0],
                 pd_count_per_label=self._label_counts[:, 1],
-                n_datums=self.info.number_of_datums,
                 n_labels=self.info.number_of_labels,
+                accumulated_fp=accumulated_fp,
+                accumulated_tp=accumulated_tp,
+                prev_fpr=prev_fpr,
+                prev_tpr=prev_tpr,
             )
-            rocauc += batch_rocauc
 
         mean_rocauc = rocauc.mean()
 

src/valor_lite/classification/loader.py

@@ -24,7 +24,7 @@ def __init__(
     ):
         self._path = Path(path) if path else None
         self._writer = writer
-        self._sorted_writer = sorted_writer
+        self._rocauc_writer = sorted_writer
         self._datum_metadata_fields = datum_metadata_fields
 
         # internal state
@@ -53,7 +53,7 @@ def in_memory(
             batch_size=batch_size,
         )
         sorted_writer = MemoryCacheWriter.create(
-            schema=cls._generate_sorted_schema(),
+            schema=cls._generate_rocauc_schema(),
             batch_size=batch_size,
         )
         return cls(
@@ -104,8 +104,8 @@ def persistent(
             compression=compression,
         )
         sorted_writer = FileCacheWriter.create(
-            path=cls._generate_sorted_cache_path(path),
-            schema=cls._generate_sorted_schema(),
+            path=cls._generate_rocauc_cache_path(path),
+            schema=cls._generate_rocauc_schema(),
             batch_size=batch_size,
             rows_per_file=rows_per_file,
             compression=compression,
@@ -214,6 +214,8 @@ def finalize(
         self._writer.flush()
         if self._writer.count_rows() == 0:
             raise EmptyCacheError()
+        elif self._rocauc_writer.count_rows() > 0:
+            raise RuntimeError("data already finalized")
 
         # sort in-place and locally
         self._writer.sort_by(
@@ -229,23 +231,20 @@ def finalize(
         reader = self._writer.to_reader()
         sort(
             source=reader,
-            sink=self._sorted_writer,
+            sink=self._rocauc_writer,
             batch_size=batch_size,
             sorting=[
                 ("score", "descending"),
+                ("match", "descending"),
                 ("pd_label_id", "ascending"),
-                ("gt_label_id", "ascending"),
             ],
             columns=[
-                "datum_id",
-                "gt_label_id",
                 "pd_label_id",
                 "score",
-                "winner",
                 "match",
             ],
         )
-        sorted_reader = self._sorted_writer.to_reader()
+        rocauc_reader = self._rocauc_writer.to_reader()
 
         # generate evaluator meta
         (index_to_label, label_counts, info,) = self.generate_meta(
@@ -254,7 +253,7 @@ def finalize(
 
         return Evaluator(
             reader=reader,
-            sorted_reader=sorted_reader,
+            rocauc_reader=rocauc_reader,
             info=info,
             label_counts=label_counts,
             index_to_label=index_to_label,