Leakage Module

BasePredictabilityMetric

class bias_amplification.metrics.PredMetrics.BasePredictabilityMetric(model_params: Dict[str, Module], train_params: Dict[str, Any], eval_metric: Callable | str = 'mse', threshold: bool = True, normalized: bool = False, test_size=0.2)[source]

Bases: ABC

Base class for predictability metrics. It provides base functionality for computing leakage with fairness evaluation.

Methods

`calcLambda`(model, x, y, **kwargs)	Calculate the lambda value for a given attacker model, input data and target data.
`calcLeak`(feat_train, data_train, pred_train, ...)	Calculates the leakage of the attacker model for a given mode and given data splits.
`computeBiasAmp`(feat_train, data_train, ...)	This function computes the average amortized leakage for a given protected attribute, ground truth data and predicted values.
`compute_leakage`(lambda_d, lambda_m[, normalized])	Compute the leakage value for a given predictability metric.
`defineModel`()	Define the attacker_model_D and attacker_model_M models by the subclasses.
`initialize_eval_metrics`(metric)	Initializes the evaluation metric for the predictor model.
`permuteData`(data, model_acc[, mode])	This function permutes data for quality equalization to maintain the accuracy of the model.
`shuffle_data`(x, y)	Randomly shuffles x and y data.
`split`(feat, data, pred[, test_size])	Splits the data into training and testing sets.
`train`(x, y, attacker_mode)	Trains the attacker model for a given mode (eg.
`train_batch_with_loss`(attacker_model, ...)	Trains a batch of data with a given loss function.
`train_epochs`(attacker_model, optimizer, ...)	Trains the attacker model for an epoch and returns the average loss.
`train_setup`(attacker_mode)	Setup model, criterion and optimizer for training.

__init__(model_params: Dict[str, Module], train_params: Dict[str, Any], eval_metric: Callable | str = 'mse', threshold: bool = True, normalized: bool = False, test_size=0.2)[source]

calcLambda(model: Module, x: tensor, y: tensor, **kwargs) → tensor[source]: Calculate the lambda value for a given attacker model, input data and target data.

calcLeak(feat_train: tensor, data_train: tensor, pred_train: tensor, feat_test: tensor, data_test: tensor, pred_test: tensor, mode: str | None = None) → tensor[source]

Calculates the leakage of the attacker model for a given mode and given data splits.

Parameters:

feattorch.tensor: Protected Attribute.
datatorch.tensor: Ground truth data.
predtorch.tensor: Predicted Values.
modeLiteral[“AtoT”,”TtoA”]: Sets Direction of calculation.

Returns:

leakagetorch.tensor: Evaluated as if normalized, returns (λ_M - λ_D) / (λ_M + λ_D), otherwise returns λ_M - λ_D

computeBiasAmp(feat_train: tensor, data_train: tensor, pred_train: tensor, mode: str | None = None, num_trials: int = 10, method: str = 'mean', feat_test: tensor | None = None, data_test: tensor | None = None, pred_test: tensor | None = None) → Tuple[tensor, tensor][source]

This function computes the average amortized leakage for a given protected attribute, ground truth data and predicted values. If the given dataset is not split into training and testing sets, the data is split into training and testing sets. The leakage is calculated for each trial. The leakage is then averaged based on provided method and the standard deviation is calculated.

Parameters:

feat_train: torch.tensor: The protected attribute training data.
data_train: torch.tensor: The ground truth training data.
pred_train: torch.tensor: The predicted training data.
mode: Optional[str]: The mode to be used.
num_trials: int: The number of trials to be used.
method: str: The method to be used.
feat_test: torch.tensor: The protected attribute testing data.
data_test: torch.tensor: The ground truth testing data.
pred_test: torch.tensor: The predicted testing data.

Returns:

Tuple[torch.tensor, torch.tensor]: The formatted amortized leakage and the standard deviation of the form “leakage ± standard deviation”.

compute_leakage(lambda_d: tensor, lambda_m: tensor, normalized: bool = False) → tensor[source]: Compute the leakage value for a given predictability metric.

abstract defineModel()[source]: Define the attacker_model_D and attacker_model_M models by the subclasses.

initialize_eval_metrics(metric: Callable | str) → None[source]: Initializes the evaluation metric for the predictor model.

permuteData(data: tensor, model_acc: float, mode: str = 'AtoT') → tensor[source]

This function permutes data for quality equalization to maintain the accuracy of the model. Ground truth data assumed to be binary values in a pytorch tensor but intended to work for any NxM type array.

Parameters:

datatorch.tensor: Original ground truth data.
model_accfloat: Accuracy of the model (A’ or T’) w.r.t (A or T) - used for quality equalization.
modestr: Mode identifier for the attacker model

Returns:

new_datatorch.tensor: Randomly pertubed data matching the specified model accuracy.

shuffle_data(x: tensor, y: tensor) → Tuple[tensor, tensor][source]: Randomly shuffles x and y data.

split(feat: tensor, data: tensor, pred: tensor, test_size: float | None = None) → Tuple[tensor, tensor, tensor, tensor, tensor, tensor][source]: Splits the data into training and testing sets.

train(x: tensor, y: tensor, attacker_mode: str) → tensor[source]

Trains the attacker model for a given mode (eg. AtoT, TtoA).

Parameters:

xtorch.tensor: Input data.
ytorch.tensor: Target data.
attacker_modestr: Mode of the attacker model.

Returns:

None: Trains the attacker model for a given mode.

train_batch_with_loss(attacker_model: Module, optimizer: Optimizer, criterion: Module, x_batch: tensor, y_batch: tensor) → float[source]

Trains a batch of data with a given loss function.

Parameters:

attacker_modeltorch.nn.Module: The attacker model to be trained.
optimizertorch.optim.Optimizer: The optimizer to be used.
criteriontorch.nn.Module: The loss function to be used.
x_batchtorch.tensor: The input data batch.
y_batchtorch.tensor: The target data batch.

Returns:

lossfloat: The loss value.

train_epochs(attacker_model: Module, optimizer: Optimizer, criterion: Module, x: tensor, y: tensor) → float[source]

Trains the attacker model for an epoch and returns the average loss.

Parameters:

attacker_modeltorch.nn.Module: The attacker model to be trained.
optimizertorch.optim.Optimizer: The optimizer to be used.
criteriontorch.nn.Module: The loss function to be used.
xtorch.tensor: The input data.
ytorch.tensor: The target data.

Returns:

avg_lossfloat: The average loss.

train_setup(attacker_mode: str) → Tuple[Module, Module, Optimizer][source]

Setup model, criterion and optimizer for training.

Parameters:

attacker_modestr: Mode identifier for the attacker model

Returns:

Tuple[Attacker Model, loss criterion, and optimizer]

Leakage

class bias_amplification.metrics.PredMetrics.Leakage(attacker_model: Module, train_params: Dict[str, Any], eval_metric: Callable | str = 'mse', threshold: bool = True, normalized: bool = False)[source]

Bases: BasePredictabilityMetric

This class inherits from the BasePredictabilityMetric class and implements the Leakage metric.

Methods

`computeBiasAmp`(feat_train, data_train, ...)	This function calls the base class method to compute the amortized leakage for the Leakage metric.
`defineModel`()	This function defines the attacker models for the Leakage metric.

Examples

from bias_amplification.metrics.PredMetrics import Leakage
from bias_amplification.attacker_models import simpleDenseModel
import torch

# Create attacker model
attacker_model = simpleDenseModel(
    1, 1, 1, numFirst=1,
    activations=["sigmoid", "sigmoid", "sigmoid"]
)

# Initialize Leakage metric
leakage = Leakage(
    attacker_model=attacker_model,
    train_params={
        "learning_rate": 0.01,
        "loss_function": "bce",
        "epochs": 100,
        "batch_size": 64
    },
    model_acc=0.8,
    eval_metric="accuracy"
)

# Calculate amortized leakage
result = leakage.getAmortizedLeakage(
    features, ground_truth, predictions,
    num_trials=10
)

__init__(attacker_model: Module, train_params: Dict[str, Any], eval_metric: Callable | str = 'mse', threshold: bool = True, normalized: bool = False) → None[source]

Parameters:

model_paramsdict

{“attacker” : model}

train_paramsdict

{

“learning_rate”: The learning rate hyperparameter, “loss_function”: The loss function to be used.

Existing options: [“mse”, “cross-entropy”],

“epochs”: Number of training epochs to be set, “batch_size: Number of batches per epoch

}

model_accfloat

The accuracy of the model being used for quality equalization.

eval_metricUnion[Callable,str]

This metric is used to evaluate the attacker model accuracy. Options include [“accuracy”, “mse”, “bce”].

thresholdbool

Whether to use a threshold on the predictions.

normalizedbool

Default is False to use the raw leakage value.

Returns:

None: Initializes the class.

computeBiasAmp(feat_train: tensor, data_train: tensor, pred_train: tensor, num_trials: int = 10, method: str = 'mean', feat_test: tensor | None = None, data_test: tensor | None = None, pred_test: tensor | None = None) → Tuple[tensor, tensor][source]

This function calls the base class method to compute the amortized leakage for the Leakage metric.

Parameters:

feat_train: torch.tensor: The protected attribute training data.
data_train: torch.tensor: The ground truth training data.
pred_train: torch.tensor: The predicted training data.
num_trials: int: The number of trials to be used.
method: str: The method to be used.
feat_test: torch.tensor: The protected attribute testing data.
data_test: torch.tensor: The ground truth testing data.
pred_test: torch.tensor: The predicted testing data.

Returns:

Tuple[torch.tensor, torch.tensor]: The formatted amortized leakage and the standard deviation of the form “leakage ± standard deviation” for the Leakage metric.

defineModel() → None[source]: This function defines the attacker models for the Leakage metric. The attacker model for ground truth data is defined as the attacker_D model. The attacker model for predicted values is defined as the attacker_M model.

DPA (Directional Predictability Amplification)

class bias_amplification.metrics.PredMetrics.DPA(attacker_AtoT: Module, attacker_TtoA: Module, train_params: Dict[str, Any], eval_metric: Callable | str = 'mse', threshold: bool = True, normalized: bool = True)[source]

Bases: BasePredictabilityMetric

Methods

`computeBiasAmp`(feat_train, data_train, ...)	This function calls the base class method to compute the amortized leakage for the DPA metric.
`computeBiasAmpBidirectional`(A, T, A_pred, T_pred)	This function computes the bidirectional leakage for a given protected attribute, ground truth data and predicted values.
`defineModel`()	This function defines the attacker models for the DPA metric.

Examples

from bias_amplification.metrics.PredMetrics import DPA
from bias_amplification.attacker_models import simpleDenseModel

# Create attacker models
attacker_AtoT = simpleDenseModel(1, 1, 1, numFirst=1)
attacker_TtoA = simpleDenseModel(1, 1, 1, numFirst=1)

# Initialize DPA metric
dpa = DPA(
    attacker_AtoT=attacker_AtoT,
    attacker_TtoA=attacker_TtoA,
    train_params={...},
    model_acc={"AtoT": 0.8, "TtoA": 0.7},
    eval_metric="accuracy"
)

# Calculate bidirectional leakage
atot_result, ttoa_result = dpa.calcBidirectional(
    protected_attr, target,
    pred_protected, pred_target
)

__init__(attacker_AtoT: Module, attacker_TtoA: Module, train_params: Dict[str, Any], eval_metric: Callable | str = 'mse', threshold: bool = True, normalized: bool = True) → None[source]

Parameters:

model_paramsdict

Dictionary of the following forms- {“attacker_AtoT” : attacker_AtoT, “attacker_TtoA” : attacker_TtoA}

train_paramsdict

{ “learning_rate”: The learning rate hyperparameter, “loss_function”: The loss function to be used.

Existing options: [“mse”, “cross-entropy”],

“epochs”: Number of training epochs to be set, “batch_size: Number of batches per epoch }

model_accUnion[float, dict]

The accuracy of the model being used for quality equalization. For bidirectional case, send dict of the form {‘AtoT’: acc_AtoT, ‘TtoA’: acc_TtoA}

eval_metricUnion[Callable,str]

This metric is used to evaluate the attacker model accuracy. Options include [“accuracy”, “mse”, “bce”].

thresholdbool

Whether to use a threshold on the predictions.

normalizedbool

Default is True in order to use normalization for the leakage calculation.

Returns:

None: Initializes the class.

computeBiasAmp(feat_train: tensor, data_train: tensor, pred_train: tensor, mode: Literal['AtoT', 'TtoA'], num_trials: int = 10, method: str = 'mean', feat_test: tensor | None = None, data_test: tensor | None = None, pred_test: tensor | None = None) → Tuple[tensor, tensor][source]

This function calls the base class method to compute the amortized leakage for the DPA metric. The mode is used to determine the direction of the leakage calculation. If mode is “AtoT”, the leakage is calculated from the protected attribute to the ground truth data. If mode is “TtoA”, the leakage is calculated from the ground truth data to the protected attribute.

Parameters:

feat_train: torch.tensor: The protected attribute training data.
data_train: torch.tensor: The ground truth training data.
pred_train: torch.tensor: The predicted training data.
mode: Literal[“AtoT”, “TtoA”]: The mode to be used.
num_trials: int: The number of trials to be used.
method: str: The method to be used.
feat_test: torch.tensor: The protected attribute testing data.
data_test: torch.tensor: The ground truth testing data.
pred_test: torch.tensor: The predicted testing data.

Returns:

Tuple[torch.tensor, torch.tensor]: The formatted amortized leakage and the standard deviation of the form “leakage ± standard deviation” for the DPA metric.

computeBiasAmpBidirectional(A: tensor, T: tensor, A_pred: tensor, T_pred: tensor, num_trials: int = 10, method: str = 'mean') → Tuple[Tuple[tensor, tensor], Tuple[tensor, tensor]][source]

This function computes the bidirectional leakage for a given protected attribute, ground truth data and predicted values.

Parameters:

A: torch.tensor: The protected attribute.
T: torch.tensor: The ground truth data.
A_pred: torch.tensor: The predicted protected attribute.
T_pred: torch.tensor: The predicted ground truth data.
num_trials: int: The number of trials to be used.
method: str: The method to be used.

Returns:

Tuple[Tuple[torch.tensor, torch.tensor], Tuple[torch.tensor, torch.tensor]]: The tuple contains (AtoT_leakage, TtoA_leakage). AtoT_leakage gives the average amortized leakage for the AtoT direction. TtoA_leakage gives the average amortized leakage for the TtoA direction.

defineModel() → None[source]: This function defines the attacker models for the DPA metric. The attacker models for ground truth data are defined as the attacker_D_AtoT and attacker_D_TtoA models. The attacker models for predicted values are defined as the attacker_M_AtoT and attacker_M_TtoA models.