Issue 2921 and 2920 overlaping check

sdv/evaluation/utils.py

@@ -4,10 +4,13 @@
 import warnings
 
 import pandas as pd
+from pandas.api.types import is_datetime64_any_dtype, is_numeric_dtype
 
 from sdv._utils import _cast_to_iterable, _check_is_dict_of_dataframes
 from sdv.metadata import Metadata
 
+MISSING_VALUE_PLACEHOLDER = '__sdv_missing_value__'
+
 
 def _validate_referential_integrity_inputs(
     metadata, synthetic_data, table_name, foreign_key_name, num_rows
@@ -152,3 +155,206 @@ def print_referential_integrity(
             result = f'❌ Unable to find the linked {parent_table_name} row'
 
         sys.stdout.write(f'{heading}\n{result}\n\n')
+
+
+def _validate_data(real_data, synthetic_data, table_name, column_names):
+    """Validate that both datasets contain the table and columns to check."""
+    if not isinstance(table_name, str):
+        raise TypeError(f"'table_name' must be a string, got {type(table_name).__name__}.")
+
+    if not isinstance(column_names, list) or not all(
+        isinstance(column_name, str) for column_name in column_names
+    ):
+        raise TypeError("'column_names' must be a list of strings.")
+
+    if not column_names:
+        raise ValueError("'column_names' must contain at least one column name.")
+
+    for argument_name, data in [('real_data', real_data), ('synthetic_data', synthetic_data)]:
+        if table_name not in data:
+            raise ValueError(f"Table '{table_name}' is not present in '{argument_name}'.")
+
+        missing = [column for column in column_names if column not in data[table_name].columns]
+        if missing:
+            missing_columns = "', '".join(missing)
+            raise ValueError(
+                f"The columns '{missing_columns}' are not present in table '{table_name}' "
+                f"of '{argument_name}'."
+            )
+
+
+def _align_dtypes(real_column, synthetic_column):
+    """Make sure data types of columns being evaluated match.
+
+    Args:
+        real_column (pd.Series):
+            The column of real data.
+        synthetic_column (pd.Series):
+            The column of synthetic data.
+
+    Returns:
+        tuple[pd.Series, pd.Series]:
+            The real and synthetic column, cast to a comparable dtype.
+    """
+    if real_column.dtype == synthetic_column.dtype:
+        return real_column, synthetic_column
+
+    if is_numeric_dtype(real_column) and is_numeric_dtype(synthetic_column):
+        return real_column.astype('float64'), synthetic_column.astype('float64')
+
+    if is_datetime64_any_dtype(real_column) or is_datetime64_any_dtype(synthetic_column):
+        return (
+            pd.to_datetime(real_column, errors='coerce'),
+            pd.to_datetime(synthetic_column, errors='coerce'),
+        )
+
+    return real_column.astype(str), synthetic_column.astype(str)
+
+
+def _get_combinations(data):
+    """Get the set of unique combinations of values in the data."""
+    combinations = data.astype('object')
+    combinations = combinations.where(combinations.notna(), MISSING_VALUE_PLACEHOLDER)
+
+    return set(combinations.itertuples(index=False, name=None))
+
+
+def _compute_overlap(real_data, synthetic_data, table_name, column_names):
+    """Get the number of combinations shared by both datasets and the percentage they represent.
+
+    Args:
+        real_data (dict):
+            A dictionary mapping a table name to a pandas DataFrame containing real data.
+        synthetic_data (dict):
+            A dictionary mapping a table name to a pandas DataFrame containing synthetic data.
+        table_name (str):
+            The name of the table that contains the columns to check.
+        column_names (list[str]):
+            The column names to combine.
+
+    Returns:
+        tuple[int, float]:
+            The number of shared combinations and their percentage of all combinations.
+    """
+    real_values = real_data[table_name][column_names].copy()
+    synthetic_values = synthetic_data[table_name][column_names].copy()
+    for column_name in column_names:
+        real_values[column_name], synthetic_values[column_name] = _align_dtypes(
+            real_values[column_name], synthetic_values[column_name]
+        )
+
+    real_combinations = _get_combinations(real_values)
+    synthetic_combinations = _get_combinations(synthetic_values)
+
+    num_common = len(real_combinations & synthetic_combinations)
+    num_total = len(real_combinations | synthetic_combinations)
+    percent = round(num_common / num_total * 100, 2) if num_total else 0.0
+
+    return num_common, percent
+
+
+def get_combination_overlap(real_data, synthetic_data, table_name, column_names, verbose=True):
+    """Calculate the overlap of combinations of column values between real and synthetic data.
+
+    Args:
+        real_data (dict):
+            A dictionary mapping a table name to a pandas DataFrame containing real data.
+        synthetic_data (dict):
+            A dictionary mapping a table name to a pandas DataFrame containing synthetic data.
+        table_name (str):
+            The name of the table that contains the columns to check.
+        column_names (list[str]):
+            A list of strings representing the column names to check. Combinations of these
+            columns will be checked.
+        verbose (bool):
+            Whether to print out the interpretation of the results. Defaults to ``True``.
+
+    Returns:
+        int:
+            The number of unique combinations that appear in both the real and synthetic data.
+
+    Raises:
+        TypeError:
+            If ``table_name`` is not a string or ``column_names`` is not a list of strings.
+        ValueError:
+            If the table or any of the columns is missing from the data.
+    """
+    _validate_data(real_data, synthetic_data, table_name, column_names)
+    num_common, percent = _compute_overlap(real_data, synthetic_data, table_name, column_names)
+
+    if verbose:
+        sys.stdout.write(f'Number of common combinations: {num_common} ({percent}%)\n')
+        if num_common == 0:
+            sys.stdout.write(
+                '✅ The synthetic data does not contain any of the same combinations from the '
+                'real data\n'
+            )
+        elif percent <= 2:
+            sys.stdout.write(
+                '⚠️ The synthetic data contains a few of the same combinations as the real '
+                'data. This might be due to random chance.\n'
+            )
+        else:
+            sys.stdout.write(
+                '❌ The synthetic data contains a significant number of the same combinations '
+                'as the real data. This might be due to a small number of possible '
+                'combinations, a large sample of synthetic data, or a misconfiguration in your '
+                'synthesizer.\n'
+            )
+
+    return num_common
+
+
+def get_pii_overlap(real_data, synthetic_data, table_name, pii_column_name, verbose=True):
+    """Calculate the overlap of PII values between the real and synthetic data.
+
+    Args:
+        real_data (dict):
+            A dictionary mapping a table name to a pandas DataFrame containing real data.
+        synthetic_data (dict):
+            A dictionary mapping a table name to a pandas DataFrame containing synthetic data.
+        table_name (str):
+            The name of the table that contains the PII column to check.
+        pii_column_name (str):
+            The name of the column that contains PII values to check.
+        verbose (bool):
+            Whether to print out the interpretation of the results. Defaults to ``True``.
+
+    Returns:
+        int:
+            The number of unique PII values that appear in both the real and synthetic data.
+
+    Raises:
+        TypeError:
+            If ``table_name`` or ``pii_column_name`` is not a string.
+        ValueError:
+            If the table or the column is missing from the data.
+    """
+    if not isinstance(pii_column_name, str):
+        raise TypeError(
+            f"'pii_column_name' must be a string, got {type(pii_column_name).__name__}."
+        )
+
+    column_names = [pii_column_name]
+    _validate_data(real_data, synthetic_data, table_name, column_names)
+    num_common, percent = _compute_overlap(real_data, synthetic_data, table_name, column_names)
+
+    if verbose:
+        sys.stdout.write(f'Number of common data points: {num_common} ({percent}%)\n')
+        if num_common == 0:
+            sys.stdout.write(
+                '✅ The synthetic data does not contain any PII values from the real data\n'
+            )
+        elif percent <= 2:
+            sys.stdout.write(
+                '⚠️ The synthetic data contains a few PII values from the real data. '
+                'This might be due to random chance.\n'
+            )
+        else:
+            sys.stdout.write(
+                '❌ The synthetic data contains a significant number of the same PII values of '
+                'as the real data. This might be due to a small number of possible PII values, '
+                'a large sample of synthetic data, or a misconfiguration in your synthesizer.\n'
+            )
+
+    return num_common

tests/integration/evaluation/test_utils.py

@@ -1,8 +1,15 @@
 import pandas as pd
 import pytest
 
-from sdv.evaluation.utils import print_referential_integrity
+from sdv.evaluation.utils import (
+    get_combination_overlap,
+    get_pii_overlap,
+    print_referential_integrity,
+)
 from sdv.metadata import Metadata
+from sdv.single_table.copulas import GaussianCopulaSynthesizer
+
+NO_PII_OVERLAP_MESSAGE = '✅ The synthetic data does not contain any PII values from the real data'
 
 
 @pytest.fixture()
@@ -137,3 +144,120 @@ def test_print_referential_integrity_with_null_foreign_key(data_metadata_parent_
     # Assert
     captured = capsys.readouterr().out
     assert captured.count('✅ Foreign key is null; no linked parent row expected') == 1
+
+
+def _get_demographic_data():
+    """Return real data and metadata for a table of quasi-identifiers."""
+    real_data = pd.DataFrame({
+        'date_of_birth': [
+            '1990-01-01',
+            '1985-06-15',
+            '1972-11-30',
+            '2001-03-22',
+            '1968-09-08',
+        ],
+        'zipcode': [94301, 10001, 60614, 78701, 2139],
+        'gender': ['M', 'F', 'F', 'M', 'F'],
+    })
+
+    metadata = Metadata()
+    metadata.add_table('customer')
+    metadata.add_column('date_of_birth', 'customer', sdtype='datetime', datetime_format='%Y-%m-%d')
+    metadata.add_column('zipcode', 'customer', sdtype='numerical')
+    metadata.add_column('gender', 'customer', sdtype='categorical')
+
+    return real_data, metadata
+
+
+def test_get_combination_overlap_end_to_end():
+    """Test the overlap of a synthesizer's output against the real data."""
+    # Setup
+    real_data, metadata = _get_demographic_data()
+    synthesizer = GaussianCopulaSynthesizer(metadata)
+    synthesizer.fit(real_data)
+    synthetic_data = synthesizer.sample(10)
+    column_names = ['date_of_birth', 'zipcode', 'gender']
+
+    # Run
+    result = get_combination_overlap(
+        real_data={'customer': real_data},
+        synthetic_data={'customer': synthetic_data},
+        table_name='customer',
+        column_names=column_names,
+        verbose=False,
+    )
+
+    # Assert
+    real_combinations = set(real_data[column_names].itertuples(index=False, name=None))
+    synthetic_combinations = set(synthetic_data[column_names].itertuples(index=False, name=None))
+    assert isinstance(result, int)
+    assert result == len(real_combinations & synthetic_combinations)
+
+
+def test_get_combination_overlap_with_identical_data(capsys):
+    """Test that data copied from the real data overlaps completely."""
+    # Setup
+    real_data, _ = _get_demographic_data()
+    column_names = ['date_of_birth', 'zipcode', 'gender']
+
+    # Run
+    result = get_combination_overlap(
+        real_data={'customer': real_data},
+        synthetic_data={'customer': real_data.copy()},
+        table_name='customer',
+        column_names=column_names,
+    )
+
+    # Assert
+    captured = capsys.readouterr()
+    assert result == 5
+    assert 'Number of common combinations: 5 (100.0%)' in captured.out
+
+
+def test_get_combination_overlap_detects_a_single_shared_row():
+    """Test that one row copied from the real data is detected."""
+    # Setup
+    real_data, _ = _get_demographic_data()
+    synthetic_data = pd.concat(
+        [
+            real_data.iloc[[2]],
+            pd.DataFrame({
+                'date_of_birth': ['1993-04-04'],
+                'zipcode': [30301],
+                'gender': ['F'],
+            }),
+        ],
+        ignore_index=True,
+    )
+
+    # Run
+    result = get_combination_overlap(
+        real_data={'customer': real_data},
+        synthetic_data={'customer': synthetic_data},
+        table_name='customer',
+        column_names=['date_of_birth', 'zipcode', 'gender'],
+        verbose=False,
+    )
+
+    # Assert
+    assert result == 1
+
+
+def test_get_pii_overlap_with_anonymized_column(capsys):
+    """Test that a fully anonymized PII column reports no overlap."""
+    # Setup
+    real_data = pd.DataFrame({'ssn': ['111-11-1111', '222-22-2222', '333-33-3333']})
+    synthetic_data = pd.DataFrame({'ssn': ['999-99-9999', '888-88-8888']})
+
+    # Run
+    result = get_pii_overlap(
+        real_data={'customer': real_data},
+        synthetic_data={'customer': synthetic_data},
+        table_name='customer',
+        pii_column_name='ssn',
+    )
+
+    # Assert
+    captured = capsys.readouterr()
+    assert result == 0
+    assert captured.out == f'Number of common data points: 0 (0.0%)\n{NO_PII_OVERLAP_MESSAGE}\n'