First attempt at emplemnting OTOS. Still needs work

Author: zoj613

pyloras/_otos.py

@@ -1,10 +1,10 @@
-import ot
 from imblearn.over_sampling.base import BaseOverSampler
 from imblearn.utils import Substitution
 from imblearn.utils._docstring import (
     _random_state_docstring,
     _n_jobs_docstring,
 )
+from sklearn.svm import LinearSVC
 import numpy as np
 
 from ._common import check_random_state, safe_random_state
@@ -20,16 +20,76 @@ def __init__(
         self,
         *,
         sampling_strategy="auto",
-        svm_regularization=1.0,
-        ot_regularization=1.0,
+        svc_reg=1.0,
+        ot_reg=1.0,
         tradeoff=1.0,
         random_state=None,
+        max_iter=100,
     ):
         super().__init__(sampling_strategy=sampling_strategy)
-        self.svm_regularization = svm_regularization
-        self.ot_regularization = ot_regularization
+        self.svc_reg = svc_reg
+        self.ot_reg = ot_reg
         self.tradeoff = tradeoff
         self.random_state = random_state
+        self.max_iter = max_iter
 
-    def fit_resample(self, X, y):
-        return X, y
+    def _fit_resample(self, X, y):
+        import ot
+        random_state = check_random_state(self.random_state)
+        X_res = [X.copy()]
+        y_res = [y.copy()]
+        svc = LinearSVC(
+            loss="hinge", C=self.svc_reg, random_state=safe_random_state(random_state)
+        )
+        for minority_class, samples_to_make in self.sampling_strategy_.items():
+            if samples_to_make == 0:
+                continue
+            X_p = X[y == minority_class]
+            X_n = X[y != minority_class]
+            n_p = X_p.shape[0]
+            n_n = X_n.shape[0]
+            n_r = samples_to_make
+            one_r = np.ones((n_r, 1))
+            one_n = np.ones((n_n, 1))
+            # set initial distribution for mu_r and mu_p
+            mu_r = np.asarray([1.0 / n_r] * n_r)
+            mu_p = np.asarray([1.0 / n_p] * n_p)
+            T = mu_r[:, None] @ mu_p[:, None].T
+            # manufactor a binary classification problem
+            _y = np.empty_like(y)
+            _y[y == minority_class] = 0
+            _y[y != minority_class] = 1
+            svc.fit(X, _y)
+            w = svc.coef_.T
+
+            hingelosses = np.concatenate(
+                [
+                    np.atleast_1d(max(1 - y_i * svc.coef_ @ x_row, 0.0))
+                    for y_i, x_row in zip(y[y == minority_class], X_p)
+                ]
+            )
+            mu_p = np.exp(hingelosses)
+            mu_p /= mu_p.sum()
+
+            D_r = np.diag(1 / mu_r)
+            X_r = D_r @ T @ X_p
+            # C_p = np.apply_along_axis(c_row, axis=-1, arr=X_r)
+            C_p = np.asarray(
+                [
+                    [np.linalg.norm(x_row - row) for row in X_p]
+                    for x_row in X_r
+                ]
+            )
+            wwT = w @ w.T
+            Theta = self.tradeoff * C_p.T - X_p @ np.kron(one_r.T, wwT @ X_n.T @ one_n + n_n * w) @ D_r
+            Phi = X_p @ wwT @ X_p.T
+            Psi = D_r.T @ D_r
+
+            print(mu_r.sum(), mu_p.sum())
+            for _ in range(self.max_iter):
+                transport_cost = Theta.T + n_n * Psi @ T @ Phi
+                T = ot.sinkhorn(mu_r, mu_p, transport_cost, self.ot_reg)
+            X_res.append(D_r @ T @ X_p)
+            y_res.append([minority_class] * n_r)
+        print(z)
+        return np.concatenate(X_res), np.concatenate(y_res)

pyproject.toml

@@ -27,6 +27,9 @@ classifiers = [
     "Operating System :: MacOS :: MacOS X",
 ]
 
+[project.optional-dependencies]
+otos = ["POT"]
+
 [project.urls]
 source = "https://github.com/zoj613/pyloras"
 tracker = "https://github.com/zoj613/pyloras/issues"

requirements-dev.txt

@@ -2,5 +2,6 @@
 build==0.10.0
 imbalanced-learn==0.10.1
 numpy==1.23.2
+POT==0.9.0
 pytest==7.3.1
 pytest-cov==4.0.0