feat: add unit tests for pattern_based_weighted_mean_metric and distance_correlation_complement

Author: anhkhoangoho

qolmat/benchmark/metrics.py

@@ -858,7 +858,7 @@ def distance_correlation_complement(
     df1 = df1[df_mask].fillna(0.0)
     df2 = df2[df_mask].fillna(0.0)
 
-    return 1.0 - pd.Series(dcor.distance_correlation(df1.values, df2.values), index=["All"])
+    return 1.0 - pd.Series([dcor.distance_correlation(df1.values, df2.values)], index=["All"])
 
 
 def pattern_based_weighted_mean_metric(
@@ -912,10 +912,16 @@ def pattern_based_weighted_mean_metric(
 
         if len(df1_pattern) >= min_num_row:
             df2_pattern = df2.loc[df1_pattern.index, df1_pattern.columns]
-            weights.append(len(df1_pattern))
+            weights.append(1.0 / len(df1_pattern))
             scores.append(
                 metric(df1_pattern, df2_pattern, ~df1_pattern.isna(), **kwargs).values[0]
             )
 
+    if len(scores) == 0:
+        raise Exception(
+            "Not found enough patterns. "
+            + f"Number of row for each pattern must be larger than min_num_row={min_num_row}."
+        )
+
     weighted_scores = np.array(scores) * np.array(weights)
-    return pd.Series(np.mean(weighted_scores), index=["All"])
+    return pd.Series(np.sum(weighted_scores) / np.sum(weights), index=["All"])

qolmat/benchmark/missing_patterns.py

@@ -2,9 +2,8 @@
 
 import functools
 import logging
-from typing import Callable, List, Optional, Tuple, Union
+from typing import Callable, List, Optional, Tuple
 
-import random
 import numpy as np
 import pandas as pd
 from sklearn.model_selection import GroupShuffleSplit

tests/benchmark/test_metrics.py

@@ -333,3 +333,28 @@ def test_value_error_get_correlation_f_oneway_matrix(
     assert metrics.mean_diff_corr_matrix_categorical_vs_numerical_features(
         df1, df2, df_mask
     ).equals(pd.Series([np.nan], index=["col1"]))
+
+
+@pytest.mark.parametrize("df1", [df_incomplete])
+@pytest.mark.parametrize("df2", [df_imputed])
+@pytest.mark.parametrize("df_mask", [df_mask])
+def test_distance_correlation_complement(
+    df1: pd.DataFrame, df2: pd.DataFrame, df_mask: pd.DataFrame
+) -> None:
+    result = metrics.distance_correlation_complement(df1, df2, df_mask)
+    expected = pd.Series([0.001559], index=["All"])
+    np.testing.assert_allclose(result, expected, atol=1e-3)
+
+
+@pytest.mark.parametrize("df1", [df_incomplete])
+@pytest.mark.parametrize("df2", [df_imputed])
+@pytest.mark.parametrize("df_mask", [df_mask])
+def test_pattern_based_weighted_mean_metric(
+    df1: pd.DataFrame, df2: pd.DataFrame, df_mask: pd.DataFrame
+) -> None:
+
+    result = metrics.pattern_based_weighted_mean_metric(
+        df1, df2, df_mask, metric=metrics.distance_correlation_complement, min_num_row=5
+    )
+    expected = pd.Series([2 / 3], index=["All"])
+    np.testing.assert_allclose(result, expected, atol=1e-3)