import pandas as pd
from fairlearn.metrics import demographic_parity_difference, equalized_odds_difference

# Load historical distribution data
df = pd.read_csv('food_bank_distribution.csv')

# Step 1: Check demographic balance
demographic_balance = df['neighborhood_income_bracket'].value_counts(normalize=True)
print("Demographic Representation:\n", demographic_balance)

# Step 2: Assess a preliminary model's fairness
# Assume `y_true` (actual need) and `y_pred` (model allocation) exist
dpd = demographic_parity_difference(y_true, y_pred, 
                                     sensitive_features=df['neighborhood_income_bracket'])
eod = equalized_odds_difference(y_true, y_pred,
                                  sensitive_features=df['neighborhood_income_bracket'])

print(f"Demographic Parity Difference: {dpd:.4f}")
print(f"Equalized Odds Difference: {eod:.4f}")

# A high disparity score (>0.1) signals unfair skew, requiring mitigation.

This audit provides a quantitative baseline, a crucial first step offered by specialized **data science analytics services**.

import shap
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split

# Split data
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# Train model
model = RandomForestClassifier(n_estimators=100, max_depth=5, random_state=42)
model.fit(X_train, y_train)

# Create SHAP explainer
explainer = shap.TreeExplainer(model)
shap_values = explainer.shap_values(X_test)

# Visualize global feature importance
shap.summary_plot(shap_values[1], X_test, plot_type="bar")

# Visualize individual prediction for the first test instance
shap.force_plot(explainer.expected_value[1], shap_values[1][0], X_test.iloc[0], matplotlib=True)

This visualization explicitly shows how features like "distance to clinic" or "previous hospitalizations" influence a "high-risk" prediction, enabling domain experts to validate the model's logic.

import shap
import pandas as pd
from sklearn.ensemble import RandomForestClassifier

# Assume model is already trained
model = RandomForestClassifier().fit(X_train, y_train)

# Create an explainer
explainer = shap.TreeExplainer(model)

# Select a specific case to explain
single_instance = X_test.iloc[[0]] 

# Calculate SHAP values
shap_values_single = explainer.shap_values(single_instance)

# Generate a visual force plot
shap.initjs() # For Jupyter notebooks
shap.force_plot(explainer.expected_value[1], 
                shap_values_single[1][0], 
                single_instance.iloc[0])

**Measurable Benefit:** This increases stakeholder trust and accountability, potentially reducing challenge rates to automated decisions by providing a clear, auditable rationale.

from fairlearn.metrics import demographic_parity_difference, equalized_odds_difference

y_pred = model.predict(X_test)

dpd = demographic_parity_difference(y_test, y_pred, 
                                     sensitive_features=demographic_data['race'])
eod = equalized_odds_difference(y_test, y_pred,
                                  sensitive_features=demographic_data['race'])

print(f"Demographic Parity Difference: {dpd:.4f}")
print(f"Equalized Odds Difference: {eod:.4f}")

from fairlearn.reductions import ExponentiatedGradient, DemographicParity

mitigator = ExponentiatedGradient(
    estimator=RandomForestClassifier(),
    constraints=DemographicParity()
)
mitigator.fit(X_train, y_train, sensitive_features=demographic_train['race'])
fair_predictions = mitigator.predict(X_test)

**Measurable Benefit:** Directly quantifies and reduces unfair disparity, leading to more equitable outcomes in critical applications like healthcare screening or loan approvals.

from alibi_detect.cd import TabularDrift
from alibi_detect.utils.saving import save_detector, load_detector
import numpy as np

# Initialize detector on reference (training) data
cd = TabularDrift(X_train_ref, p_val=0.05)

# Save the detector for reuse
save_detector(cd, './drift_detector')

# Later, load and check a new production batch
cd_loaded = load_detector('./drift_detector')
preds = cd_loaded.predict(X_prod_batch, return_p_val=True)

if preds['data']['is_drift']:
    print(f"Drift detected! p-value: {preds['data']['p_val']}")
    trigger_retraining_pipeline()

**Measurable Benefit:** Proactively maintains model integrity and fairness in production, preventing degradation that could harm vulnerable populations dependent on the service.

import pandas as pd
import numpy as np

# Load applicant data
df = pd.read_csv('applicant_data.csv')

# Analyze representation
total_applicants = df.shape[0]
gender_representation = df['gender'].value_counts(normalize=True)
print("Gender Representation:\n", gender_representation)

# Analyze selection rates (if historical decisions exist)
selection_by_gender = df.groupby('gender')['hired'].mean()
print("\nHistorical Selection Rate by Gender:\n", selection_by_gender)

# Calculate Disparate Impact Ratio (80% Rule)
min_selection_rate = selection_by_gender.min()
max_selection_rate = selection_by_gender.max()
disparate_impact_ratio = min_selection_rate / max_selection_rate
print(f"\nDisparate Impact Ratio: {disparate_impact_ratio:.3f}")
# A ratio < 0.8 indicates potential adverse impact.

from sklearn.utils.class_weight import compute_sample_weight

# 'sensitive_attribute' could be 'gender' or 'race'
# 'balanced' weights are inversely proportional to class frequencies in the sensitive attribute
sample_weights = compute_sample_weight('balanced', df['sensitive_attribute'])

# Use in model training (example with a subset)
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.model_selection import train_test_split

X = df.drop(columns=['hired', 'sensitive_attribute'])
y = df['hired']
X_train, X_val, y_train, y_val, weights_train, _ = train_test_split(
    X, y, sample_weights, test_size=0.2, random_state=42
)

model = GradientBoostingClassifier(random_state=42)
model.fit(X_train, y_train, sample_weight=weights_train)

**Benefit:** The model learns from a more balanced perspective of the data, reducing its tendency to perpetuate historical biases.

from sklearn.metrics import roc_curve
# Example for one group 'A'
fpr_a, tpr_a, thresholds_a = roc_curve(y_val[group_a], y_proba[group_a])
# Find threshold for TPR = 0.8
target_tpr = 0.8
idx_a = np.argmin(np.abs(tpr_a - target_tpr))
threshold_a = thresholds_a[idx_a]
# Apply threshold_a for group A in production

import pandas as pd
import geopandas as gpd
from sklearn.preprocessing import StandardScaler

# Load and merge datasets (conceptual)
census_df = pd.read_csv('census_tract_data.csv')
health_df = gpd.read_file('health_clinic_locations.geojson')
environment_df = pd.read_parquet('air_quality.parquet')

# Spatial join: average distance to clinics per tract
# ... (spatial operations) ...
merged_df['avg_clinic_distance_km'] = calculated_distances

# Create composite features
merged_df['economic_stress_score'] = (
    merged_df['unemployment_rate'] * 0.3 +
    merged_df['median_income_pct_rank'] * 0.7  # pct_rank reverses scale
)
merged_df['environmental_burden'] = (
    merged_df['pm2_5_concentration'] * 0.5 +
    merged_df['tree_canopy_pct'].rank(pct=True) * 0.5
)

# Standardize features for modeling
scaler = StandardScaler()
features_to_scale = ['economic_stress_score', 'environmental_burden', 'avg_clinic_distance_km']
merged_df[features_to_scale] = scaler.fit_transform(merged_df[features_to_scale])

import pandas as pd
import great_expectations as ge
from datetime import datetime

# Load raw service data
df = pd.read_json('social_service_records.json', lines=True)

# Data Validation Suite with Great Expectations
context = ge.data_context.DataContext()
suite = context.create_expectation_suite("service_data_suite", overwrite=True)

# Define critical data quality expectations
expectation_configuration = [
    {
        "expectation_type": "expect_column_values_to_not_be_null",
        "kwargs": {"column": "client_id"}
    },
    {
        "expectation_type": "expect_column_values_to_be_between",
        "kwargs": {"column": "household_size", "min_value": 1, "max_value": 20}
    },
    {
        "expectation_type": "expect_column_values_to_be_in_set",
        "kwargs": {"column": "service_type", "value_set": ["Housing", "Food", "Employment", "Healthcare"]}
    }
]

for config in expectation_configuration:
    suite.add_expectation(ge.core.ExpectationConfiguration(**config))

# Run validation
validation_result = context.run_validation_operator(
    "action_list_operator", assets_to_validate=[(df, "service_data_suite")]
)
if not validation_result["success"]:
    log_data_quality_issues(validation_result)
    trigger_data_steward_alert()

from sklearn.metrics import f1_score
from fairlearn.metrics import demographic_parity_difference

# ... after training a model ...
y_pred_val = model.predict(X_val)

# Calculate performance metric
performance = f1_score(y_val, y_pred_val, average='weighted')

# Calculate fairness metric
fairness_gap = demographic_parity_difference(y_val, y_pred_val, 
                                             sensitive_features=demographic_val['zipcode'])

# Composite score (example, weights can be adjusted)
composite_score = 0.7 * performance - 0.3 * abs(fairness_gap)
print(f"Model Performance (F1): {performance:.3f}")
print(f"Fairness Gap (DPD): {fairness_gap:.3f}")
print(f"Composite Ethical Score: {composite_score:.3f}")

import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from fairlearn.metrics import demographic_parity_difference, equalized_odds_difference

# Load and prepare data
df = pd.read_csv('health_data.csv')
X = df.drop(columns=['preventable_admission', 'zipcode_group'])
y = df['preventable_admission']
sensitive = df['zipcode_group']  # Groups based on median income

X_train, X_test, y_train, y_test, sens_train, sens_test = train_test_split(
    X, y, sensitive, test_size=0.3, random_state=42, stratify=y
)

# Train baseline model
baseline_model = RandomForestClassifier(n_estimators=200, random_state=42)
baseline_model.fit(X_train, y_train)
y_pred_base = baseline_model.predict(X_test)

# Assess fairness
dpd_base = demographic_parity_difference(y_test, y_pred_base, sensitive_features=sens_test)
eod_base = equalized_odds_difference(y_test, y_pred_base, sensitive_features=sens_test)

print(f"Baseline Model")
print(f"  Demographic Parity Difference: {dpd_base:.3f}")  # e.g., 0.18
print(f"  Equalized Odds Difference: {eod_base:.3f}")      # e.g., 0.15
# Values >> 0 indicate significant unfairness.

from fairlearn.reductions import GridSearch, DemographicParity
from sklearn.linear_model import LogisticRegression
import numpy as np

# Define the mitigator
mitigator = GridSearch(
    estimator=LogisticRegression(solver='liblinear', max_iter=1000),
    constraints=DemographicParity(), # Constrain demographic parity difference
    grid_size=30 # Number of constraint weightings to try
)

# Fit the mitigator. This learns multiple models with different trade-offs.
mitigator.fit(X_train, y_train, sensitive_features=sens_train)

# Predict with the mitigated model
y_pred_mitigated = mitigator.predict(X_test)

# Evaluate the mitigated model
dpd_mit = demographic_parity_difference(y_test, y_pred_mitigated, sensitive_features=sens_test)
eod_mit = equalized_odds_difference(y_test, y_pred_mitigated, sensitive_features=sens_test)

print(f"\nMitigated Model")
print(f"  Demographic Parity Difference: {dpd_mit:.3f}")  # e.g., 0.04
print(f"  Equalized Odds Difference: {eod_mit:.3f}")      # e.g., 0.07

# Conceptual: The mitigator object contains models across the trade-off curve.
# The final model was selected based on a policy decision to prioritize
# fairness (DPD < 0.05) while maintaining acceptable accuracy (F1 > 0.75).

import great_expectations as ge
import pandas as pd

context = ge.data_context.DataContext()

# Create a validator for incoming shelter data
batch_kwargs = {'path': 'new_shelter_intake.csv', 'datasource': 'filesystem'}
batch = context.get_batch(batch_kwargs, 'shelter_expectation_suite')

# Run predefined validation rules (e.g., required columns, value ranges)
results = context.run_validation_operator("action_list_operator", [batch])

if not results["success"]:
    # Automated alert to data steward; halt pipeline
    send_alert_to_slack(results)
    raise DataQualityException("Validation failed for shelter intake data.")

from aif360.algorithms.preprocessing import Reweighing
from aif360.datasets import StandardDataset

# Convert pandas DataFrame to AIF360 StandardDataset
aif_dataset = StandardDataset(df=training_df, label_name='target',
                              favorable_classes=[1],
                              protected_attribute_names=['race'],
                              privileged_classes=[[1]]) # Assuming 1 is privileged

# Apply reweighing
RW = Reweighing(unprivileged_groups=[{'race': 0}], # Unprivileged group
                 privileged_groups=[{'race': 1}])
dataset_transformed = RW.fit_transform(aif_dataset)

# Extract the transformed dataframe with instance weights
fair_training_df = dataset_transformed.convert_to_dataframe()[0]
sample_weights = fair_training_df['instance_weights'] # Use these in model.fit()

This engineering rigor, a core offering of specialized **data science consulting firms**, creates a reliable foundation for ethical modeling.

# Example: Deploying a model as a FastAPI service for real-time predictions
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import joblib
import pandas as pd
import numpy as np

app = FastAPI(title="Social Impact Predictor API")

# Load the trained model and feature processor
model = joblib.load('optimized_fairness_model.pkl')
feature_processor = joblib.load('feature_processor.pkl')

class PredictionRequest(BaseModel):
    features: dict  # JSON dictionary of input features

@app.post("/predict", summary="Get a prediction for a case")
async def predict(request: PredictionRequest):
    try:
        # Convert request to DataFrame and process features
        input_df = pd.DataFrame([request.features])
        processed_features = feature_processor.transform(input_df)

        # Get prediction and probability
        prediction = model.predict(processed_features)[0]
        probability = model.predict_proba(processed_features)[0].max()

        # Log prediction for audit (non-PII features only)
        log_prediction_audit(request.features, prediction, probability)

        return {
            "prediction": int(prediction),
            "probability": float(probability),
            "recommended_action": "High Priority Outreach" if prediction == 1 else "Routine Follow-up"
        }
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

from sdv.tabular import CTGAN
from sdv.evaluation import evaluate

# Load the sensitive, original dataset (e.g., patient health records)
real_data = pd.read_csv('sensitive_patient_data.csv')

# Step 1: Train a generative model (CTGAN is good for complex, heterogeneous data)
model = CTGAN(epochs=300, verbose=True)
model.fit(real_data)

# Step 2: Generate a synthetic dataset of the desired size
synthetic_data = model.sample(num_rows=20000)

# Step 3: Evaluate the fidelity of the synthetic data
quality_report = evaluate(synthetic_data, real_data)
print(f"Overall Quality Score: {quality_report.score:.3f}")
# A score close to 1.0 indicates high fidelity.

# The synthetic dataset can now be used freely for model development and testing.

**Measurable Benefit:** This approach directly **addresses data scarcity** by enabling the creation of large, high-quality training sets and facilitates safe data sharing between collaborating organizations, all while **mitigating privacy risks** to near zero.

# Simplified client training loop (using a framework like Flower - flwr)
import flwr as fl
import torch

class ShelterClient(fl.client.NumPyClient):
    def fit(self, parameters, config):
        # 1. Set local model parameters from server
        set_model_params(model, parameters)
        # 2. Train locally for 'config["local_epochs"]' epochs
        train_loss = train_local_model(model, local_data, epochs=config["local_epochs"])
        # 3. Return updated parameters and metadata
        updated_params = get_model_params(model)
        num_samples = len(local_data)
        return updated_params, num_samples, {"loss": train_loss}

**Benefit:** Enables collaborative model development on **data that cannot be centralized** due to legal or ethical constraints, effectively solving scarcity through distributed intelligence.

from pipeline.dp import BoundedMean
import numpy as np

# Assume 'ages' is a sensitive array
ages = np.array([...]) 

# Define a differentially private mean calculator
# epsilon (ε) is the privacy budget. Lower = more privacy, less accuracy.
dp_mean_calculator = BoundedMean(epsilon=0.5, lower_bound=0, upper_bound=120)

# Add data
for age in ages:
    dp_mean_calculator.add_entry(age)

# Get the private mean result
private_mean_age = dp_mean_calculator.result()
print(f"Differentially Private Mean Age: {private_mean_age:.1f}")

**Measurable Benefit:** Allows organizations to **quantify and control privacy loss (ε)**, making transparent trade-offs between utility and privacy protection, and enabling safe data collaboration.

import flwr as fl
import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset
import warnings
warnings.filterwarnings("ignore")

# Define a simple neural network model
class ReadmissionRiskModel(nn.Module):
    def __init__(self, input_size):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(input_size, 32),
            nn.ReLU(),
            nn.Linear(32, 16),
            nn.ReLU(),
            nn.Linear(16, 1),
            nn.Sigmoid()
        )
    def forward(self, x):
        return self.net(x)

def train_local_model(model, trainloader, epochs, device):
    """Train the model on local clinic data."""
    criterion = nn.BCELoss()
    optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
    model.train()
    for epoch in range(epochs):
        for features, labels in trainloader:
            features, labels = features.to(device), labels.to(device)
            optimizer.zero_grad()
            outputs = model(features).squeeze()
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()
    return loss.item()

class ClinicClient(fl.client.NumPyClient):
    def __init__(self, model, trainloader, device):
        self.model = model
        self.trainloader = trainloader
        self.device = device

    def get_parameters(self):
        """Return current model parameters as a list of NumPy ndarrays."""
        return [val.cpu().numpy() for _, val in self.model.state_dict().items()]

    def set_parameters(self, parameters):
        """Set model parameters from a list of NumPy ndarrays."""
        params_dict = zip(self.model.state_dict().keys(), parameters)
        state_dict = {k: torch.tensor(v) for k, v in params_dict}
        self.model.load_state_dict(state_dict, strict=True)

    def fit(self, parameters, config):
        """Train the model locally for one federation round."""
        # 1. Set parameters received from the global server
        self.set_parameters(parameters)
        # 2. Train the model
        epochs = config.get("local_epochs", 1)
        device = torch.device("cuda" if torch.cuda.is_available() and config.get("use_cuda", False) else "cpu")
        self.model.to(device)
        loss = train_local_model(self.model, self.trainloader, epochs, device)
        # 3. Return updated parameters and results
        updated_params = self.get_parameters()
        num_examples = len(self.trainloader.dataset)
        return updated_params, num_examples, {"train_loss": loss}

# Client startup script (runs at each clinic)
if __name__ == "__main__":
    # Load this clinic's private data
    clinic_X, clinic_y = load_clinic_specific_data() # Returns torch Tensors
    train_dataset = TensorDataset(clinic_X, clinic_y)
    trainloader = DataLoader(train_dataset, batch_size=32, shuffle=True)

    # Initialize local model
    input_dim = clinic_X.shape[1]
    local_model = ReadmissionRiskModel(input_dim)
    device = torch.device("cpu")

    # Start Flower client
    client = ClinicClient(local_model, trainloader, device)
    fl.client.start_numpy_client(server_address="[SERVER_IP]:8080", client=client)

# Pseudo-code for a data validation and profiling step
def profile_ingested_data(raw_df, sensitive_column='race'):
    profile = {}
    # Check for missing values in key columns
    profile['missing_rates'] = raw_df.isnull().mean().to_dict()
    # Check representation in sensitive column
    profile['representation'] = raw_df[sensitive_column].value_counts(normalize=True).to_dict()
    # Calculate potential label imbalance if target exists
    if 'target' in raw_df.columns:
        profile['target_balance'] = raw_df['target'].value_counts(normalize=True).to_dict()
    # Use a library like pandas-profiling for automated report
    # profile_report = pandas_profiling.ProfileReport(raw_df)
    return profile

# If representation for any group is < 5%, trigger a data sourcing review.

# Extend the prediction API to include explanations
@app.post("/predict_with_explanation")
async def predict_with_explanation(request: PredictionRequest):
    # ... get prediction as before ...
    # Generate SHAP explanation
    import shap
    background = shap.sample(X_train_reference, 100) # Representative background
    explainer = shap.KernelExplainer(model.predict_proba, background)
    shap_values = explainer.shap_values(processed_features, nsamples=100)

    # Format top contributing features
    feature_names = processed_features.columns.tolist()
    exp_dict = {feature_names[i]: float(shap_values[0][i]) 
                for i in np.argsort(np.abs(shap_values[0]))[-5:]} # Top 5

    return {
        "prediction": prediction,
        "explanation": exp_dict,
        "base_value": float(explainer.expected_value[0])
    }

This creates an immutable, queryable audit trail for every decision impacting individuals.

# Scheduled job to evaluate production model performance
def monitor_production_model(current_production_batch, current_predictions):
    from evidently.report import Report
    from evidently.metric_preset import DataDriftPreset, ClassificationPreset

    # Create a report for data drift and performance
    report = Report(metrics=[DataDriftPreset(), ClassificationPreset()])
    # Reference dataset is the training/validation data used for model creation
    report.run(reference_data=reference_dataset, 
               current_data=current_production_batch.assign(prediction=current_predictions))

    report.save_html('./monitoring_reports/latest_report.html')

    # Check for alerts
    if report['metrics'][0].result['dataset_drift']: # DataDriftPreset result
        send_alert("Data Drift Detected", severity="HIGH")
    # Check fairness metrics (extended example)
    fairness_dpd = demographic_parity_difference(...)
    if fairness_dpd > 0.1:
        send_alert(f"Fairness Degradation: DPD = {fairness_dpd:.3f}", severity="MEDIUM")

from fairlearn.datasets import fetch_adult
from sklearn.model_selection import train_test_split
from fairlearn.metrics import MetricFrame, selection_rate

# Load sample data
data = fetch_adult()
X = data.data
y = (data.target == '>50K').astype(int) # Binary target
sensitive = data.data['sex']

# Split data
X_train, X_test, y_train, y_test, sens_train, sens_test = train_test_split(
    X, y, sensitive, test_size=0.2, random_state=42
)

# Analyze disparity in the training labels (outcome) by sensitive attribute
metric_frame_train = MetricFrame(metrics={'selection_rate': selection_rate},
                                 y_true=y_train,
                                 sensitive_features=sens_train)

print("Selection Rate (Positive Outcome) by Group in Training Data:")
print(metric_frame_train.by_group)
print(f"\nOverall Disparity: {metric_frame_train.difference():.3f}")

**Measurable Benefit:** This provides a quantitative, documented baseline of existing bias in the data, informing necessary mitigation strategies and building a case for stakeholder trust.

import shap
import json

# After model training, create an explainer
explainer = shap.Explainer(model, X_train_sample)

# For a batch of production predictions, calculate and store explanations
shap_values_batch = explainer(X_production_batch)

# Log summary data (e.g., mean absolute SHAP values per feature)
mean_abs_shap = pd.DataFrame({
    'feature': X_production_batch.columns,
    'mean_abs_shap': np.abs(shap_values_batch.values).mean(axis=0)
}).sort_values('mean_abs_shap', ascending=False)

log_to_audit_trail({
    'batch_id': batch_uuid,
    'timestamp': datetime.now().isoformat(),
    'global_explanation': mean_abs_shap.to_dict('records'),
    'model_version': 'v2.1'
})

import pandas as pd
import numpy as np

def ethical_data_audit(df, target_col, sensitive_cols, privileged_groups):
    """
    Perform an initial ethical audit on a dataset.

    Args:
        df: Pandas DataFrame.
        target_col: Name of the target/label column.
        sensitive_cols: List of sensitive attribute column names.
        privileged_groups: Dict mapping sensitive_col to privileged value(s).

    Returns:
        Audit report dictionary.
    """
    audit_report = {}

    for sens_col in sensitive_cols:
        # 1. Check representation
        audit_report[f'{sens_col}_representation'] = df[sens_col].value_counts(normalize=True).to_dict()

        # 2. Check outcome disparity (if target exists)
        if target_col in df.columns:
            privileged_vals = privileged_groups.get(sens_col, [])
            if privileged_vals:
                # Simple disparity ratio: min(selection_rate) / max(selection_rate)
                selection_rates = df.groupby(sens_col)[target_col].mean()
                disparity_ratio = selection_rates.min() / selection_rates.max()
                audit_report[f'{sens_col}_disparity_ratio'] = disparity_ratio

                # Flag if ratio violates 80% rule (a common heuristic)
                if disparity_ratio < 0.8:
                    audit_report[f'{sens_col}_adverse_impact_flag'] = True
                else:
                    audit_report[f'{sens_col}_adverse_impact_flag'] = False

    # 3. Check for missing data patterns by sensitive group
    for sens_col in sensitive_cols:
        missing_by_group = df.groupby(sens_col).apply(lambda x: x.isnull().mean().mean())
        audit_report[f'{sens_col}_missing_data_pattern'] = missing_by_group.to_dict()

    return audit_report

# Example usage:
# audit = ethical_data_audit(loan_data, target_col='approved', 
#                            sensitive_cols=['race', 'gender'],
#                            privileged_groups={'race': ['white'], 'gender': ['male']})
# print(audit)

def validate_model_fairness(model, X_val, y_val, sensitive_val, threshold=0.05):
    """
    Validate a model against a fairness threshold.
    Returns True if passes, False and details if fails.
    """
    from fairlearn.metrics import demographic_parity_difference

    y_pred = model.predict(X_val)
    dpd = demographic_parity_difference(y_val, y_pred, sensitive_features=sensitive_val)

    if abs(dpd) <= threshold:
        return True, {"dpd": dpd, "message": f"Passes fairness check (DPD <= {threshold})."}
    else:
        return False, {"dpd": dpd, 
                       "message": f"Fails fairness check. DPD = {dpd:.3f} > {threshold}."}

# Use in model selection or CI/CD pipeline
is_fair, details = validate_model_fairness(candidate_model, X_val, y_val, sens_val, threshold=0.05)
if not is_fair:
    raise ModelValidationError(f"Model rejected on fairness grounds: {details['message']}")