Add mutual information metric

sdgx/metrics/column/base.py

@@ -59,10 +59,8 @@ def calculate(
         cls, real_data: pd.Series | pd.DataFrame, synthetic_data: pd.Series | pd.DataFrame
     ):
         """Calculate the metric value between columns between real table and synthetic table.
-
         Args:
             real_data(pd.DataFrame or pd.Series): the real (original) data table / column.
-
             synthetic_data(pd.DataFrame or pd.Series): the synthetic (generated) data table / column.
         """
         # This method should first check the input

sdgx/metrics/multi_table/multitable_mi_sim.py

@@ -0,0 +1,87 @@
+import time
+from datetime import datetime
+
+import numpy as np
+import pandas as pd
+from scipy.stats import entropy
+from sklearn.metrics.cluster import normalized_mutual_info_score
+
+from sdgx.metrics.multi_table.base import MultiTableMetric
+from sdgx.metrics.pair_column.mi_sim import MISim
+
+
+class MISim(MultiTableMetric):
+    """MISim : Mutual Information Similarity
+
+    This class is used to calculate the Mutual Information Similarity between the target columns of real data and synthetic data.
+
+    Currently, we support discrete and continuous(need to be discretized) columns as inputs.
+    """
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.lower_bound = 0
+        self.upper_bound = 1
+        self.metric_name = "mutual_information_similarity"
+        self.numerical_bins = 50
+
+    @classmethod
+    def calculate(
+        real_data: pd.DataFrame, synthetic_data: pd.DataFrame, metadata: dict
+    ) -> pd.DataFrame:
+        """
+        Calculate the JSD value between a real column and a synthetic column.
+        Args:
+            real_data (pd.DataFrame): The real data.
+            synthetic_data (pd.DataFrame): The synthetic data.
+            metadata(dict): The metadata that describes the data type of each column
+
+        Returns:
+            MI_similarity (float): The metric value.
+        """
+
+        # 传入概率分布数组
+
+        columns = synthetic_data.columns
+        n = len(columns)
+        mi_sim_instance = MISim()
+        nMI_sim = np.zeros((n, n))
+
+        for i in range(len(columns)):
+            for j in range(len(columns)):
+                syn_data = pd.concat(
+                    [synthetic_data[columns[i]], synthetic_data[columns[j]]], axis=1
+                )
+                real_data = pd.concat([real_data[columns[i]], real_data[columns[j]]], axis=1)
+
+                nMI_sim[i][j] = mi_sim_instance.calculate(real_data, syn_data, metadata)
+
+        MI_sim = np.sum(nMI_sim) / n / n
+        # test
+        MISim.check_output(MI_sim)
+
+        return MI_sim
+
+    @classmethod
+    def check_output(raw_metric_value: float):
+        """Check the output value.
+
+        Args:
+            raw_metric_value (float):  the calculated raw value of the JSD metric.
+        """
+        # instance = cls()
+        if raw_metric_value < self.lower_bound or raw_metric_value > self.upper_bound:
+            raise ValueError
+
+    # @classmethod
+    # def normailized_mutual_information(cls, p: float, q: float):
+    #     """Calculate the jensen_shannon_divergence of p and q.
+
+    #     Args:
+    #         p (float): the input parameter p.
+
+    #         q (float): the input parameter q.
+    #     """
+    #     n_MI = None
+
+    #     return n_MI

sdgx/metrics/pair_column/base.py

@@ -0,0 +1,79 @@
+import pandas as pd
+
+from sdgx.log import logger
+
+
+class PairMetric(object):
+    """PairMetric
+    Metrics used to evaluate the quality of synthetic data columns.
+    """
+
+    upper_bound = None
+    lower_bound = None
+    metric_name = "Correlation"
+
+    def __init__(self) -> None:
+        pass
+
+    @classmethod
+    def check_input(
+        cls, real_data: pd.DataFrame, synthetic_data: pd.DataFrame, real_metadata, syn_metadata
+    ):
+        """Input check for table input.
+        Args:
+            real_data(pd.DataFrame ): the real (original) data table.
+            synthetic_data(pd.DataFrame): the synthetic (generated) data table .
+        """
+        # Input parameter must not contain None value
+        if real_data is None or synthetic_data is None:
+            raise TypeError("Input contains None.")
+        # check column_names
+        real_cols = real_data.columns
+        syn_cols = synthetic_data.columns
+        if set(real_cols) != set(syn_cols):
+            raise TypeError("Columns of Dataframe are Different.")
+
+        # check column_types
+        for col in real_cols:
+            if real_metadata[col] != syn_metadata[col]:
+                raise TypeError("Columns of Dataframe are Different.")
+
+        # if type is pd.DataFrame, return directly
+        if isinstance(real_data, pd.DataFrame):
+            return real_data, synthetic_data
+
+        # if type is not pd.Series or pd.DataFrame tranfer it to Series
+        try:
+            real_data = pd.DataFrame(real_data)
+            synthetic_data = pd.DataFrame(synthetic_data)
+            return real_data, synthetic_data
+        except Exception as e:
+            logger.error(f"An error occurred while converting to pd.DataFrame: {e}")
+
+        return None, None
+
+    @classmethod
+    def calculate(cls, real_data: pd.DataFrame, synthetic_data: pd.DataFrame):
+        """Calculate the metric value between pair-columns between real table and synthetic table.
+
+        Args:
+            real_data(pd.DataFrame or pd.Series): the real (original) data pair.
+
+            synthetic_data(pd.DataFrame or pd.Series): the synthetic (generated) data pair.
+        """
+        # This method should first check the input
+        # such as:
+        real_data, synthetic_data = PairMetric.check_input(real_data, synthetic_data)
+
+        raise NotImplementedError()
+
+    @classmethod
+    def check_output(cls, raw_metric_value: float):
+        """Check the output value.
+
+        Args:
+            raw_metric_value (float):  the calculated raw value of the JSD metric.
+        """
+        raise NotImplementedError()
+
+    pass

sdgx/metrics/pair_column/mi_sim.py

@@ -0,0 +1,120 @@
+import time
+from datetime import datetime
+
+import numpy as np
+import pandas as pd
+from scipy.stats import entropy
+from sklearn.metrics.cluster import normalized_mutual_info_score
+
+from sdgx.metrics.pair_column.base import PairMetric
+
+
+def Jaccard_index(A, B):
+    return min(A, B) / max(A, B)
+
+
+def time2int(datetime, form):
+    time_array = time.strptime(datetime, form)
+    time_stamp = int(time.mktime(time_array))
+    return time_stamp
+
+
+class MISim(PairMetric):
+    """MISim : Mutual Information Similarity
+
+    This class is used to calculate the Mutual Information Similarity between the target columns of real data and synthetic data.
+
+    Currently, we support discrete and continuous(need to be discretized) columns as inputs.
+    """
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.lower_bound = 0
+        self.upper_bound = 1
+        self.metric_name = "mutual_information_similarity"
+        self.numerical_bins = 50
+
+    @classmethod
+    def calculate(
+        real_data: pd.DataFrame,
+        synthetic_data: pd.DataFrame,
+        metadata: dict,
+    ) -> float:
+        """
+        Calculate the JSD value between a real column and a synthetic column.
+        Args:
+            real_data (pd.DataFrame): The real data.
+            synthetic_data (pd.DataFrame): The synthetic data.
+            metadata(dict): The metadata that describes the data type of each column
+        Returns:
+            MI_similarity (float): The metric value.
+        """
+
+        # 传入概率分布数组
+
+        columns = synthetic_data.columns
+        n = len(columns)
+
+        for col in columns:
+            # data_type = self.metadata[col]
+            if data_type == "numerical":
+                # max_value = real_data[col].max()
+                # min_value = real_data[col].min()
+                real_data[col] = pd.cut(
+                    real_data[col], self.numerical_bins, labels=range(self.numerical_bins)
+                )
+                synthetic_data[col] = pd.cut(
+                    synthetic_data[col], self.numerical_bins, labels=range(self.numerical_bins)
+                )
+
+            elif data_type == "datetime":
+                real_data[col] = real_data[col].apply(time2int)
+                synthetic_data[col] = synthetic_data[col].apply(time2int)
+                real_data[col] = pd.cut(
+                    real_data[col], self.numerical_bins, labels=range(self.numerical_bins)
+                )
+                synthetic_data[col] = pd.cut(
+                    synthetic_data[col], self.numerical_bins, labels=range(self.numerical_bins)
+                )
+
+        # nMI_sim = np.zeros((n,n))
+
+        # for i in range(len(columns)):
+        #     for j in range(len(columns)):
+        syn_MI = normalized_mutual_info_score(
+            synthetic_data[columns[0]], synthetic_data[columns[1]]
+        )
+        real_MI = normalized_mutual_info_score(real_data[columns[0]], real_data[columns[1]])
+
+        MI_sim = Jaccard_index(syn_MI, real_MI)
+        """
+        MI_sim = np.sum(nMI_sim)/n/n
+        # test
+        MISim.check_output(MI_sim)
+        """
+        MISim.check_output(MI_sim)
+        return MI_sim
+
+    @classmethod
+    def check_output(raw_metric_value: float):
+        """Check the output value.
+
+        Args:
+            raw_metric_value (float):  the calculated raw value of the JSD metric.
+        """
+        # instance = cls()
+        if raw_metric_value < self.lower_bound or raw_metric_value > self.upper_bound:
+            raise ValueError
+
+    # @classmethod
+    # def normailized_mutual_information(cls, p: float, q: float):
+    #     """Calculate the jensen_shannon_divergence of p and q.
+
+    #     Args:
+    #         p (float): the input parameter p.
+
+    #         q (float): the input parameter q.
+    #     """
+    #     n_MI = None
+
+    #     return n_MI

sdgx/metrics/single_table/single_mi_sim.py

@@ -0,0 +1,84 @@
+import time
+from datetime import datetime
+
+import numpy as np
+import pandas as pd
+from scipy.stats import entropy
+from sklearn.metrics.cluster import normalized_mutual_info_score
+
+from sdgx.metrics.pair_column.mi_sim import MISim
+from sdgx.metrics.single_table.base import SingleTableMetric
+
+
+class SinTabMISim(SingleTableMetric):
+    """MISim : Mutual Information Similarity
+
+    This class is used to calculate the Mutual Information Similarity between the target columns of real data and synthetic data.
+
+    Currently, we support discrete and continuous(need to be discretized) columns as inputs.
+    """
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.lower_bound = 0
+        self.upper_bound = 1
+        self.metric_name = "mutual_information_similarity"
+        self.numerical_bins = 50
+
+    @classmethod
+    def calculate(real_data: pd.DataFrame, synthetic_data: pd.DataFrame, metadata) -> pd.DataFrame:
+        """
+        Calculate the JSD value between a real column and a synthetic column.
+        Args:
+            real_data (pd.DataFrame): The real data.
+            synthetic_data (pd.DataFrame): The synthetic data.
+            metadata(dict): The metadata that describes the data type of each column
+        Returns:
+            MI_similarity (float): The metric value.
+        """
+
+        # 传入概率分布数组
+
+        columns = synthetic_data.columns
+        n = len(columns)
+        mi_sim_instance = MISim()
+        nMI_sim = np.zeros((n, n))
+
+        for i in range(len(columns)):
+            for j in range(len(columns)):
+                syn_data = pd.concat(
+                    [synthetic_data[columns[i]], synthetic_data[columns[j]]], axis=1
+                )
+                real_data = pd.concat([real_data[columns[i]], real_data[columns[j]]], axis=1)
+
+                nMI_sim[i][j] = mi_sim_instance.calculate(real_data, syn_data, metadata)
+
+        MI_sim = np.sum(nMI_sim) / n / n
+        # test
+        MISim.check_output(MI_sim)
+
+        return MI_sim
+
+    @classmethod
+    def check_output(raw_metric_value: float):
+        """Check the output value.
+
+        Args:
+            raw_metric_value (float):  the calculated raw value of the JSD metric.
+        """
+        # instance = cls()
+        if raw_metric_value < self.lower_bound or raw_metric_value > self.upper_bound:
+            raise ValueError
+
+    # @classmethod
+    # def normailized_mutual_information(cls, p: float, q: float):
+    #     """Calculate the jensen_shannon_divergence of p and q.
+
+    #     Args:
+    #         p (float): the input parameter p.
+
+    #         q (float): the input parameter q.
+    #     """
+    #     n_MI = None
+
+    #     return n_MI