ENH: exceptions to warnings in RPCA

Author: Hông-Lan Botterman

examples/benchmark.md

@@ -234,11 +234,13 @@ results
 df_plot = results.loc["KL_columnwise",'TEMP']
 plt.barh(df_plot.index, df_plot, color=tab10(0))
 plt.title('TEMP')
+plt.xlabel("KL")
 plt.show()
 
 df_plot = results.loc["KL_columnwise",'PRES']
 plt.barh(df_plot.index, df_plot, color=tab10(0))
 plt.title('PRES')
+plt.xlabel("KL")
 plt.show()
 ```
 

qolmat/imputations/rpca/rpca_noisy.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import warnings
 from typing import Dict, List, Optional, Tuple, Union
 
 import numpy as np
@@ -9,7 +10,6 @@
 
 from qolmat.imputations.rpca import rpca_utils
 from qolmat.imputations.rpca.rpca import RPCA
-from qolmat.utils.exceptions import CostFunctionRPCANotMinimized
 
 
 class RPCANoisy(RPCA):
@@ -457,4 +457,7 @@ def _check_cost_function_minimized(
                 function_str += f"{eta} ||XH||_{self.norm}"
 
         if (round(cost_start, 4) - round(cost_end, 4)) <= -1e-2:
-            raise CostFunctionRPCANotMinimized(function_str, float(cost_start), float(cost_end))
+            warnings.warn(
+                f"RPCA algorithm may provide bad results. Function {function_str} increased from"
+                f" {cost_start} to {cost_end} instead of decreasing!"
+            )

qolmat/imputations/rpca/rpca_pcp.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import warnings
 from typing import Optional, Tuple, Union
 
 import numpy as np
@@ -8,7 +9,6 @@
 
 from qolmat.imputations.rpca import rpca_utils
 from qolmat.imputations.rpca.rpca import RPCA
-from qolmat.utils.exceptions import CostFunctionRPCANotMinimized
 
 
 class RPCAPCP(RPCA):
@@ -114,4 +114,7 @@ def _check_cost_function_minimized(
         cost_end = np.linalg.norm(low_rank, "nuc") + lam * np.sum(np.abs(anomalies))
         if round(cost_start, 4) - round(cost_end, 4) <= -1e-2:
             function_str = "||D||_* + lam ||A||_1"
-            raise CostFunctionRPCANotMinimized(function_str, float(cost_start), float(cost_end))
+            warnings.warn(
+                f"RPCA algorithm may provide bad results. Function {function_str} increased from"
+                f" {cost_start} to {cost_end} instead of decreasing!"
+            )

qolmat/utils/exceptions.py

@@ -1,5 +1,4 @@
 from typing import Any, List, Tuple, Type
-from numpy.typing import NDArray
 
 
 class PyTorchExtraNotInstalled(Exception):
@@ -31,14 +30,6 @@ def __init__(self, subset: Any):
         super().__init__(f"Provided subset `{subset}` should be None or a list!")
 
 
-class CostFunctionRPCANotMinimized(Exception):
-    def __init__(self, name_fct: str, value_start: float, value_end: float):
-        super().__init__(
-            f"RPCA algorithm may provide bad results. Function {name_fct} increased from"
-            f" {value_start} to {value_end} instead of decreasing!"
-        )
-
-
 class NotDimension2(Exception):
     def __init__(self, shape: Tuple[int, ...]):
         super().__init__(f"Provided matrix is of shape {shape}, which is not of dimension 2!")

tests/imputations/rpca/test_rpca_noisy.py

@@ -1,3 +1,5 @@
+import warnings
+
 import numpy as np
 import pytest
 from numpy.typing import NDArray
@@ -6,7 +8,6 @@
 from qolmat.imputations.rpca.rpca_noisy import RPCANoisy
 from qolmat.utils import utils
 from qolmat.utils.data import generate_artificial_ts
-from qolmat.utils.exceptions import CostFunctionRPCANotMinimized
 
 X_complete = np.array([[1, 2], [3, 1]], dtype=float)
 X_incomplete = np.array([[1, 2], [3, np.nan]], dtype=float)
@@ -45,13 +46,35 @@ def synthetic_temporal_data():
         )
     ],
 )
-def test_check_cost_function_minimized_raise_expection(
+def test_check_cost_function_minimized_warning(
+    obs: NDArray, lr: NDArray, ano: NDArray, omega: NDArray, lam: float, tau: float, norm: str
+):
+    """Test warning when the cost function is not minimized."""
+    with pytest.warns(UserWarning):
+        RPCANoisy()._check_cost_function_minimized(obs, lr, ano, omega, lam, tau)
+
+
+@pytest.mark.parametrize(
+    "obs, lr, ano, omega, lam, tau, norm",
+    [
+        (
+            np.array([[1, 1], [1, 1]], dtype=float),
+            np.array([[0, 0], [0, 0]], dtype=float),
+            np.array([[0, 0], [0, 0]], dtype=float),
+            True * np.ones((2, 2)),
+            5,
+            0,
+            "L2",
+        )
+    ],
+)
+def test_check_cost_function_minimized_no_warning(
     obs: NDArray, lr: NDArray, ano: NDArray, omega: NDArray, lam: float, tau: float, norm: str
 ):
-    """Test if exception is raised when the cost function is not minimized."""
-    rpca = RPCANoisy()
-    with pytest.raises(CostFunctionRPCANotMinimized):
-        rpca._check_cost_function_minimized(obs, lr, ano, omega, lam, tau)
+    """Test no warning when the cost function is minimized."""
+    with warnings.catch_warnings(record=True) as record:
+        RPCANoisy()._check_cost_function_minimized(obs, lr, ano, omega, lam, tau)
+    assert len(record) == 0
 
 
 @pytest.mark.parametrize("X", [X_complete])

tests/imputations/rpca/test_rpca_pcp.py

@@ -1,11 +1,12 @@
+import warnings
+
 import numpy as np
 import pytest
 from numpy.typing import NDArray
 
 from qolmat.imputations.rpca.rpca_pcp import RPCAPCP
 from qolmat.utils import utils
 from qolmat.utils.data import generate_artificial_ts
-from qolmat.utils.exceptions import CostFunctionRPCANotMinimized
 
 X_complete = np.array([[1, 2], [3, 1]], dtype=float)
 X_incomplete = np.array([[1, 2], [3, np.nan], [np.nan, 4]], dtype=float)
@@ -41,41 +42,62 @@ def synthetic_temporal_data():
         )
     ],
 )
-def test_check_cost_function_minimized_raise_expection(
+def test_check_cost_function_minimized_warning(
+    obs: NDArray, lr: NDArray, ano: NDArray, lam: float
+):
+    """Test warning when the cost function is minimized."""
+    with pytest.warns(UserWarning):
+        RPCAPCP()._check_cost_function_minimized(obs, lr, ano, lam)
+
+
+@pytest.mark.parametrize(
+    "obs, lr, ano, lam",
+    [
+        (
+            np.array([[1, 1], [1, 1]], dtype=float),
+            np.array([[0, 0], [0, 0]], dtype=float),
+            np.array([[2, 2], [2, 2]], dtype=float),
+            0,
+        )
+    ],
+)
+def test_check_cost_function_minimized_no_warning(
     obs: NDArray, lr: NDArray, ano: NDArray, lam: float
 ):
-    function_str = "||D||_* + lam ||A||_1"
-    rpca = RPCAPCP()
-    with pytest.raises(
-        CostFunctionRPCANotMinimized,
-        match="PCA algorithm may provide bad results. "
-        f"{function_str} is larger at the end "
-        "of the algorithm than at the start.",
-    ):
-        rpca._check_cost_function_minimized(obs, lr, ano, lam)
+    """Test no warning when the cost function is minimized."""
+    with warnings.catch_warnings(record=True) as record:
+        RPCAPCP()._check_cost_function_minimized(obs, lr, ano, lam)
+    assert len(record) == 0
 
 
 @pytest.mark.parametrize("X", [X_complete])
 def test_rpca_rpca_pcp_get_params_scale(X: NDArray):
+    """Test the parameters are well scaled."""
     rpca_pcp = RPCAPCP(max_iterations=max_iterations, mu=0.5, lam=0.1)
     result_dict = rpca_pcp.get_params_scale(X)
     result = list(result_dict.values())
     params_expected = [1 / 7, np.sqrt(2) / 2]
     np.testing.assert_allclose(result, params_expected, atol=1e-4)
 
 
-# The problem is ill-conditioned and the result depends on the parameter mu
 @pytest.mark.parametrize("X, mu", [(X_complete, small_mu)])
 def test_rpca_rpca_pcp_zero_lambda_small_mu(X: NDArray, mu: float):
+    """Test RPCA PCP results if lambda equals zero.
+    The problem is ill-conditioned and the result depends
+    on the parameter mu; case when mu is small.
+    """
     rpca_pcp = RPCAPCP(lam=0, mu=mu)
     X_result, A_result = rpca_pcp.decompose_rpca_signal(X)
     np.testing.assert_allclose(X_result, np.full_like(X, 0), atol=1e-4)
     np.testing.assert_allclose(A_result, X, atol=1e-4)
 
 
-# The problem is ill-conditioned and the result depends on the parameter mu
 @pytest.mark.parametrize("X, mu", [(X_complete, large_mu)])
 def test_rpca_rpca_pcp_zero_lambda_large_mu(X: NDArray, mu: float):
+    """Test RPCA PCP results if lambda equals zero.
+    The problem is ill-conditioned and the result depends
+    on the parameter mu; case when mu is large.
+    """
     rpca_pcp = RPCAPCP(lam=0, mu=mu)
     X_result, A_result = rpca_pcp.decompose_rpca_signal(X)
     np.testing.assert_allclose(X_result, X, atol=1e-4)
@@ -84,13 +106,16 @@ def test_rpca_rpca_pcp_zero_lambda_large_mu(X: NDArray, mu: float):
 
 @pytest.mark.parametrize("X, mu", [(X_complete, large_mu)])
 def test_rpca_rpca_pcp_large_lambda_small_mu(X: NDArray, mu: float):
+    """Test RPCA PCP results with large lambda and small mu."""
     rpca_pcp = RPCAPCP(lam=1e3, mu=mu)
     X_result, A_result = rpca_pcp.decompose_rpca_signal(X)
     np.testing.assert_allclose(X_result, X, atol=1e-4)
     np.testing.assert_allclose(A_result, np.full_like(X, 0), atol=1e-4)
 
 
 def test_rpca_temporal_signal(synthetic_temporal_data):
+    """Test RPCA PCP results for time series data.
+    Check if the cost function is smaller at the end than at the start."""
     signal = synthetic_temporal_data
     period = 100
     lam = 0.1