Skip to content

evaluation_field

evaluation_field

Classes:

Name Description
EvaluationField

Per-column evaluation metadata and distribution scores.

EvaluationField pydantic-model

Bases: BaseModel

Per-column evaluation metadata and distribution scores.

Fields:

name pydantic-field

Column name from the original dataframe.

reference_field_features pydantic-field

Field type and descriptive statistics for the reference column.

output_field_features pydantic-field

Field type and descriptive statistics for the output column.

reference_distribution pydantic-field

Binned distribution dict for the reference column.

output_distribution pydantic-field

Binned distribution dict for the output column.

distribution_distance pydantic-field

Jensen-Shannon distance between the two distributions.

distribution_stability pydantic-field

Graded score derived from the distribution distance.

column_statistics pydantic-field

PII entity counts and transform metadata, if available.

from_series(name, reference, output, column_statistics=None) staticmethod

Build an EvaluationField from paired reference/output column.

Normally called internally by EvaluationDataset; direct use is rarely needed.

Parameters:

Name Type Description Default
name str

Column name.

 required
reference Series

Reference column data.

 required
output Series

Output (synthetic) column data.

 required
column_statistics ColumnStatistics | None

PII entity metadata to attach, if available.

 None

Returns:

Type Description
EvaluationField

A fully populated EvaluationField with computed distributions
EvaluationField

and stability score.
Source code in src/nemo_safe_synthesizer/evaluation/data_model/evaluation_field.py 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
@staticmethod
def from_series(
    name: str,
    reference: pd.Series,
    output: pd.Series,
    column_statistics: ColumnStatistics | None = None,
) -> EvaluationField:
    """Build an ``EvaluationField`` from paired reference/output column.

    Normally called internally by ``EvaluationDataset``; direct use is
    rarely needed.

    Args:
        name: Column name.
        reference: Reference column data.
        output: Output (synthetic) column data.
        column_statistics: PII entity metadata to attach, if available.

    Returns:
        A fully populated ``EvaluationField`` with computed distributions
        and stability score.
    """
    reference_field_features = describe_field(name, reference)
    output_field_features = describe_field(name, output)
    # TODO This was a config setting to explicitly force fields to be categorical.
    # if is_categorical:
    #     reference_field_features.type = FieldType.CATEGORICAL
    #     output_field_features.type = FieldType.CATEGORICAL

    # TODO Synthesizer only, but not making conditional until more new config/control is baked up.
    if reference_field_features.type == FieldType.NUMERIC and output_field_features.type == FieldType.NUMERIC:
        bins = stats.get_numeric_distribution_bins(reference, output)
        reference_distribution = stats.get_numeric_field_distribution(reference, bins)
        output_distribution = stats.get_numeric_field_distribution(output, bins)
        distribution_distance = stats.compute_distribution_distance(reference_distribution, output_distribution)
        distribution_stability = EvaluationField.get_field_distribution_stability(distribution_distance)
    else:
        if is_integer_dtype(reference) or is_integer_dtype(output):
            try:
                # If the other column contains float values or has None values with object dtype,
                # first cast it to float, round the values, and then convert to pd.Int64Dtype.
                # This allows missing values to be properly handled and enables meaningful comparisons.
                reference = reference.astype(float).round().astype(pd.Int64Dtype())
                output = output.astype(float).round().astype(pd.Int64Dtype())
            except ValueError:
                # The other column has something weird that is not a float, just keep going.
                pass
        if (
            reference_field_features.count == 0
            or output_field_features.count == 0
            or reference_field_features.type in HIGHLY_UNIQUE_TYPES
            or output_field_features.type in HIGHLY_UNIQUE_TYPES
        ):
            reference_distribution = None
            output_distribution = None
            distribution_distance = None
            distribution_stability = None
        else:
            reference_distribution = stats.get_categorical_field_distribution(reference)
            output_distribution = stats.get_categorical_field_distribution(output)
            distribution_distance = stats.compute_distribution_distance(reference_distribution, output_distribution)
            distribution_stability = EvaluationField.get_field_distribution_stability(distribution_distance)

    return EvaluationField(
        name=name,
        reference_field_features=reference_field_features,
        output_field_features=output_field_features,
        reference_distribution=reference_distribution,
        output_distribution=output_distribution,
        distribution_distance=distribution_distance,
        distribution_stability=distribution_stability,
        column_statistics=column_statistics,
    )

get_average_divergence(fields) staticmethod

Compute the mean Jensen-Shannon divergence across a list of fields.

Source code in src/nemo_safe_synthesizer/evaluation/data_model/evaluation_field.py 
123
124
125
126
127
128
129
130
131
132
133
@staticmethod
def get_average_divergence(fields: list[EvaluationField]) -> float:
    """Compute the mean Jensen-Shannon divergence across a list of fields."""
    if len(fields) > 0:
        average_divergence = reduce(
            lambda x, y: x + y,
            [f.distribution_distance for f in fields if f.distribution_distance is not None],
            0.0,  # ENGPROD-6, default accumulator value may be required
        ) / len(fields)
        return average_divergence
    return 0.0

text_js_scaling_func(average_divergence) staticmethod

Scale average JS divergence for text data using a linear equation.

Parameters:

Name Type Description Default
average_divergence float

Mean JS divergence across text fields.

 required

Returns:

Type Description
float

A score in the range [1.5, 10].
Source code in src/nemo_safe_synthesizer/evaluation/data_model/evaluation_field.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@staticmethod
def text_js_scaling_func(average_divergence: float) -> float:
    """Scale average JS divergence for text data using a linear equation.

    Args:
        average_divergence: Mean JS divergence across text fields.

    Returns:
        A score in the range ``[1.5, 10]``.
    """
    # Scaling with linear equation penalizes the lower range scores drastically, setting the lower values to 15 instead of 0.
    # More explained in this doc.
    if average_divergence > 0.8:
        score = 1.5
    elif average_divergence < 0.12:
        score = 10.0
    else:
        score = -12.5 * average_divergence + 11.5
    return score

tabular_js_scaling_func(average_divergence) staticmethod

Scale average JS divergence for tabular data using a quadratic equation.

Parameters:

Name Type Description Default
average_divergence float

Mean JS divergence across tabular fields.

 required

Returns:

Type Description
float

A score in the range [0, 10].
Source code in src/nemo_safe_synthesizer/evaluation/data_model/evaluation_field.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
@staticmethod
def tabular_js_scaling_func(average_divergence: float) -> float:
    """Scale average JS divergence for tabular data using a quadratic equation.

    Args:
        average_divergence: Mean JS divergence across tabular fields.

    Returns:
        A score in the range ``[0, 10]``.
    """
    if average_divergence > 0.99:
        score = 0.0
    elif average_divergence < 0.02:
        score = 10.0
    else:
        score = 7.44 * average_divergence**2 - 16.646 * average_divergence + 10.305
    return score

get_field_distribution_stability(average_divergence, js_scaling_func=None) staticmethod

Convert an average JS divergence into a graded EvaluationScore.

Parameters:

Name Type Description Default
average_divergence float

Mean JS divergence across fields.

 required
js_scaling_func Callable[[float], float] | None

Scaling function mapping divergence to a 0--10 score. Defaults to tabular_js_scaling_func.

 None

Returns:

Type Description
EvaluationScore

A finalized EvaluationScore with grade and scaled score.
Source code in src/nemo_safe_synthesizer/evaluation/data_model/evaluation_field.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
@staticmethod
def get_field_distribution_stability(
    average_divergence: float,
    js_scaling_func: Callable[[float], float] | None = None,
) -> EvaluationScore:
    """Convert an average JS divergence into a graded ``EvaluationScore``.

    Args:
        average_divergence: Mean JS divergence across fields.
        js_scaling_func: Scaling function mapping divergence to a 0--10
            score. Defaults to ``tabular_js_scaling_func``.

    Returns:
        A finalized ``EvaluationScore`` with grade and scaled score.
    """
    js_scaling_func = js_scaling_func or EvaluationField.tabular_js_scaling_func
    try:
        if np.isnan(average_divergence):
            return EvaluationScore()
        score = js_scaling_func(average_divergence)
        return EvaluationScore.finalize_grade(average_divergence, score)
    except Exception as e:
        logger.exception("Failed to calculate Field Distribution Stability SQS")
        return EvaluationScore(notes=str(e))