left-icon

Ethical AI for Developers Succinctly®
by Ed Freitas

Previous
Chapter

of
A
A
A

CHAPTER 3

The AI Development Lifecycle and Ethics


Overview

Ethical guidelines are best integrated into every stage of the AI development lifecycle, not merely treated as a final checklist item. A proactive, ethics-by-design approach ensures that potential harm is identified and mitigated early, where it is easiest and cheapest to fix.

Stage 1: Data collection and ethical preparation

This is arguably the most important stage, as the data is the foundation upon which all subsequent ethical properties are built.

Garbage in, garbage out is the technical maxim. Bias in, discrimination out is the ethical one.

Let’s have a look at some of the ethical challenges involved with data collection:

·     Data provenance: Where did the data come from? Was it collected with informed consent?

·     Representativeness: Does the data accurately reflect the population the model will serve? Imbalances can lead to poor performance and unfair outcomes for underrepresented groups.

·     Sensitive attributes: Are sensitive features (such as race, gender) necessary? If so, how will they be protected or used to check for bias?

Ethical coding practice: Data auditing

Let's put this into perspective with an example. A responsible developer should audit the data for imbalances and potential biases.

The following Python script demonstrates a basic (simulated) check for data imbalance in a protected attribute, a crucial first step in ethical data preparation.

Code Listing 3-a: Ethical coding practice—data auditing (data-auditing.py)

import pandas as pd

from sklearn.model_selection import train_test_split

# 1. Simulate a dataset with a protected attribute (e.g., 'Gender').

data = {

    'Feature_A': [10, 20, 30, 40, 50, 60, 70, 80, 90, 100],

    'Feature_B': [5, 15, 25, 35, 45, 55, 65, 75, 85, 95],

    'Gender': ['Male'] * 8 + ['Female'] * 2,

    # Intentional imbalance: 80% Male, 20% Female

    'Target': [0, 1, 0, 1, 0, 1, 0, 1, 0, 1]

}

df = pd.DataFrame(data)

# 2. Check for an imbalance in the protected attribute.

protected_attribute = 'Gender'

imbalance = df[protected_attribute].value_counts(normalize=True) * 100

print(f"--- Data Imbalance Check for '{protected_attribute}' ---")

print(imbalance)

# Ethical comment:

if imbalance.min() < 30:

    print(

    "\n[ETHICAL WARNING]: The protected attribute is highly imbalanced. "

    "Training a model on this data may lead to biased outcomes against

     the minority group.")

else:

    print(

    "\n[ETHICAL NOTE]: Imbalance is within acceptable limits for this  

    attribute.")

# 3. Split data (standard practice).

X = df.drop('Target', axis=1)

y = df['Target']

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)

To run this code, first ensure you have a recent version of Python 3 installed (in my case, I already have version 3.12.4 installed, which I'll be using throughout this ebook). Then, download the project repository.

Python installation

On Windows, you can install Python using the Python Manager or download the standalone installer. The following is a quick overview of the process using a standalone installer (version 3.14.2).

Python 3.14.2 installer (main screen)

Figure 3-a: Python 3.14.2 installer (main screen)

Typically, the easiest option is to click Install Now.

An alternative to installing a standalone Python instance is to install the popular Anaconda distribution, which includes Python and over 300 compatible packages, such as NumPy, scikit-learn, and Pandas.

Once Python is installed, and you've placed the downloaded code repository in a specific folder, open Visual Studio Code in that folder. Follow these steps to create a virtual environment, activate it, and install the following dependencies.

Code Listing 3-b: Setup steps

# Run this from the local folder where you downloaded the code repository.

# Execute this from the terminal within Visual Studio Code.

# This creates the Python virtual environment.

python -m venv venv

venv\Scripts\activate.bat # This activates the Python virtual environment.

# On a Unix/Linux system, it would be: venv/Scripts/activate

pip install pandas scikit-learn # This installs the required dependencies.

Once everything has been installed as described, you can execute the data-auditing.py script as follows.

Code Listing 3-c: Data auditing execution

# Execute this from the terminal within Visual Studio Code.

py data-auditing.py

Running the script produces the following results.

Code Listing 3-d: Data auditing execution results (data-auditing.py)

--- Data Imbalance Check for 'Gender' ---

Gender

Male      80.0

Female    20.0

Name: proportion, dtype: float64

[ETHICAL WARNING]: The protected attribute is highly imbalanced. Training a model on this data may lead to biased outcomes against the minority group.

As you can see, given the imbalance in the male-to-female ratio, a warning was issued.

Stage 2: Model design and training

Once the data is ready, let’s move to the model itself. Ethical choices here revolve around complexity, performance, and the selection of appropriate metrics.

Relying solely on overall accuracy can hide poor performance for minority groups. Ethical metrics such as the equal opportunity difference should be considered.

Interpretability versus performance represents the trade-off between a highly accurate "black box" model and a less precise but fully transparent model.

Robustness ensures the model is not easily fooled by small, malicious changes to the input data.

Stage 3: Deployment and integration

The moment the model is put into production, ethical risks become real-world issues. Planning for monitoring and human intervention can empower you to manage these risks effectively, fostering confidence in your ethical oversight:

·     Human oversight: Defining the human-in-the-loop process and at what stage the AI should defer the task to a human.

·     Rollback strategy: Having a clear and tested plan to revert to a previous, stable version immediately if the model shows harmful behavior.

·     User communication: Informing the end user that they are interacting with an AI system and outlining the system's limitations.

Stage 4: Monitoring and maintenance

Most AI systems are not static; they degrade over time due to changes in the distribution of real-world data (data drift) or in the relationship between features and the target (concept drift). Ethical monitoring is therefore continuous. The two main challenges in this stage are:

·     Drift detection: Monitoring not just for performance drift, but also for fairness drift—where the model's bias increases over time.

·     Feedback loops: Preventing the model's decisions from creating new, biased data that further entrenches the original bias (a biased policing model leading to more arrests in one area, which then generates more data to justify the original bias).

Single codebase with four stages combined

Now, using the original idea behind the data auditing example, let's refactor the code to include the four stages mentioned, combined into a single codebase. Please have a good look at the following code; don't worry, we'll go over each part in detail after the listing.

Code Listing 3-e: Codebase with four stages (data-auditing-4stages.py)

"""

Stage 1: Data Collection & Preparation

Stage 2: Model Development & Training

Stage 3: Model Evaluation & Validation

Stage 4: Deployment & Monitoring

"""

from __future__ import annotations

import hashlib

import json

from dataclasses import dataclass, asdict

from typing import Dict, List, Tuple

import numpy as np

import pandas as pd

from sklearn.compose import ColumnTransformer

from sklearn.linear_model import LogisticRegression

from sklearn.metrics import (

    accuracy_score,

    confusion_matrix,

    f1_score,

    precision_score,

    recall_score,

)

from sklearn.model_selection import train_test_split

from sklearn.pipeline import Pipeline

from sklearn.preprocessing import OneHotEncoder, StandardScaler

# STAGE 1 — DATA COLLECTION & PREPARATION (ETHICAL)

@dataclass(frozen=True)

class DataProvenance:

    dataset_name: str

    source: str

    collection_method: str

    collection_date_range: str

    purpose: str

    consent_obtained: bool

    consent_scope: str

    pii_present: bool

    retention_policy: str

    notes: str = ""

def fingerprint_dataframe(df: pd.DataFrame) -> str:

    payload = df.to_csv(index=False).encode("utf-8")

    return hashlib.sha256(payload).hexdigest()

def audit_protected_attribute_distribution(

    df: pd.DataFrame,

    col: str,

    min_pct: float = 30.0,

    min_count: int = 5,

) -> Dict[str, object]:

    counts = df[col].value_counts(dropna=False)

    pct = df[col].value_counts(normalize=True, dropna=False) * 100

    report = {

        "attribute": col,

        "counts": counts.to_dict(),

        "percentages": pct.round(2).to_dict(),

        "warnings": [],

    }

    if float(pct.min()) < min_pct:

        report["warnings"].append(

            f"Imbalance: smallest group is

              {float(pct.min()):.2f}% (< {min_pct}%). "

            "Bias risk for underrepresented groups."

        )

    if int(counts.min()) < min_count:

        report["warnings"].append(

            f"Small-n risk: smallest group has

             {int(counts.min())} samples (< {min_count}). "

            "Metrics will be noisy; consider collecting more data."

        )

    return report

def audit_missingness(df: pd.DataFrame) -> pd.Series:

    return (df.isna().mean() * 100).round(2)

def audit_target_rate_by_group(df: pd.DataFrame, group_col: str, target_col: str) -> pd.DataFrame:

    out = (

        df.groupby(group_col)[target_col]

        .agg(["count", "mean"])

        .rename(columns={"mean": "target_rate"})

        .reset_index()

    )

    out["target_rate"] = (out["target_rate"] * 100).round(2)

    return out

# --- Simulated dataset (original example) ---

data = {

    "Feature_A": [10, 20, 30, 40, 50, 60, 70, 80, 90, 100],

    "Feature_B": [5, 15, 25, 35, 45, 55, 65, 75, 85, 95],

    "Gender": ["Male"] * 8 + ["Female"] * 2,  # intentional imbalance

    "Target": [0, 1, 0, 1, 0, 1, 0, 1, 0, 1],

}

df = pd.DataFrame(data)

protected_attributes = ["Gender"]

target_col = "Target"

provenance = DataProvenance(

    dataset_name="Demo_Data_v0",

    source="Synthetic example (training only)",

    collection_method="N/A",

    collection_date_range="N/A",

    purpose="Demonstrate end-to-end ethical ML workflow",

    consent_obtained=True,

    consent_scope="Educational demo only",

    pii_present=False,

    retention_policy="Ephemeral / do not persist",

)

print("\n=== STAGE 1: DATA PROVENANCE ===")

print(json.dumps(asdict(provenance), indent=2))

print(f"Dataset fingerprint (sha256): {fingerprint_dataframe(df)}")

if not provenance.consent_obtained:

    raise RuntimeError(

    "[ETHICAL STOP]: Consent not obtained; do not proceed.")

print("\n=== STAGE 1: REPRESENTATIVENESS AUDIT ===")

for attr in protected_attributes:

    rep = audit_protected_attribute_distribution(df, attr)

    print(f"\nDistribution for '{attr}':")

    print(pd.Series(rep["percentages"]).to_string())

    for w in rep["warnings"]:

        print(f"[ETHICAL WARNING]: {w}")

print("\n=== STAGE 1: MISSINGNESS AUDIT ===")

print(audit_missingness(df).to_string())

print("\n=== STAGE 1: TARGET RATE AUDIT BY GROUP ===")

for attr in protected_attributes:

    print(

    audit_target_rate_by_group(df, attr, target_col).to_string(index=False))

    print(

        "[ETHICAL NOTE]: Target-rate gaps can be real signal OR

         label/measurement bias. "

        "Investigate with domain experts."

    )

# Sensitive attribute handling: exclude from model features by default.

FEATURES_TO_EXCLUDE_FROM_MODEL = protected_attributes  

# audit-only by default.

MODEL_FEATURES = [c for c in df.columns if c not in [target_col] + FEATURES_TO_EXCLUDE_FROM_MODEL]

print("\n=== STAGE 1: SENSITIVE ATTRIBUTE MINIMIZATION ===")

print(f"Protected (audit-only): {protected_attributes}")

print(f"Model features: {MODEL_FEATURES}")

# Split with stratification on protected attribute to preserve representation

# across splits.

X = df[MODEL_FEATURES]

y = df[target_col]

audit_df = df[protected_attributes].copy()

stratify_col = df[protected_attributes[0]]

X_train, X_test, y_train, y_test, audit_train, audit_test = train_test_split(

    X, y, audit_df, test_size=0.3, random_state=42, stratify=stratify_col

)

print("\nStage 1 split check:")

print("Train Gender %:")

print((audit_train["Gender"].value_counts(normalize=True) * 100).round(2).to_string())

print("Test Gender %:")

print((audit_test["Gender"].value_counts(normalize=True) * 100).round(2).to_string())

# STAGE 2 — MODEL DEVELOPMENT & TRAINING

def compute_group_weights(series: pd.Series) -> pd.Series:

    probs = series.value_counts(normalize=True)

    w = series.map(lambda g: 1.0 / probs[g])

    return w / w.mean()

# Example mitigation: reweight training samples based on protected group

# frequency.

sample_weight = compute_group_weights(audit_train["Gender"])

numeric_features = ["Feature_A", "Feature_B"]

categorical_features: List[str] = []  

# I excluded Gender; if you have other categoricals, include here.

preprocess = ColumnTransformer(

    transformers=[

        ("num", StandardScaler(), numeric_features),

        ("cat", OneHotEncoder(handle_unknown="ignore"),

        categorical_features),

    ],

    remainder="drop",

)

# Baseline model: Logistic Regression (interpretable-ish, fast,

# good baseline).

model = LogisticRegression(max_iter=1000)

clf = Pipeline(

    steps=[

        ("preprocess", preprocess),

        ("model", model),

    ]

)

print("\n=== STAGE 2: TRAINING (BASELINE + LEAKAGE-SAFE PIPELINE) ===")

clf.fit(X_train, y_train, model__sample_weight=sample_weight)

print(

    "[ETHICAL NOTE]: We used sample reweighting as a mild mitigation

     for representation imbalance. "

    "Upstream data improvement is still the best fix."

)

# STAGE 3 — MODEL EVALUATION & VALIDATION (ETHICAL)

def group_metrics(

    y_true: np.ndarray,

    y_pred: np.ndarray,

    groups: pd.Series,

) -> pd.DataFrame:

    rows = []

    for g in groups.unique():

        idx = groups == g

        yt = y_true[idx]

        yp = y_pred[idx]

        # confusion matrix: [[TN, FP],[FN, TP]]

        tn, fp, fn, tp = confusion_matrix(yt, yp, labels=[0, 1]).ravel()

        # common rates (avoid div-by-zero)

        tpr = tp / (tp + fn) if (tp + fn) else np.nan  # recall / sensitivity

        fpr = fp / (fp + tn) if (fp + tn) else np.nan

        rows.append(

        {

         "group": g,

         "n": int(idx.sum()),

         "accuracy": accuracy_score(yt, yp) if len(yt) else np.nan,

         "precision": precision_score(yt, yp, zero_division=0) if len(yt)

            else np.nan,

         "recall_TPR": tpr,

         "FPR": fpr,

         "f1": f1_score(yt, yp, zero_division=0) if len(yt) else np.nan,

         "TN": int(tn),

         "FP": int(fp),

         "FN": int(fn),

         "TP": int(tp),

        }

      )

    return pd.DataFrame(rows).sort_values("group")

def fairness_gap_report(df_metrics: pd.DataFrame, metric: str,

max_gap: float = 0.20) -> str:

    """

    Simple guardrail:

      If max(metric) - min(metric) > max_gap, flag it.

    """

    vals = df_metrics[metric].dropna()

    if len(vals) < 2:

        return f"[ETHICAL NOTE]: Not enough groups to compute a

        gap for {metric}."

    gap = float(vals.max() - vals.min())

    if gap > max_gap:

        return f"[ETHICAL WARNING]: Large {metric} gap across

        groups: {gap:.3f} (> {max_gap}). Investigate."

    return f"[ETHICAL NOTE]: {metric} gap across groups

        is {gap:.3f} (<= {max_gap})."

print("\n=== STAGE 3: EVALUATION (OVERALL + GROUP SLICES) ===")

y_pred = clf.predict(X_test)

print("Overall metrics:")

print(f"  Accuracy : {accuracy_score(y_test, y_pred):.3f}")

print(f"  Precision: {precision_score(y_test, y_pred, zero_division=0):.3f}")

print(f"  Recall   : {recall_score(y_test, y_pred, zero_division=0):.3f}")

print(f"  F1       : {f1_score(y_test, y_pred, zero_division=0):.3f}")

gm = group_metrics(y_test.to_numpy(), y_pred, audit_test["Gender"])

print("\nGroup-sliced metrics (by Gender):")

print(gm.to_string(index=False))

# Fairness guardrails (example thresholds).

print("\nFairness checks (simple guardrails):")

print(" ", fairness_gap_report(gm, "recall_TPR", max_gap=0.20))

print(" ", fairness_gap_report(gm, "FPR", max_gap=0.20))

print(

    "\n[ETHICAL NOTE]: These are basic fairness checks.

      In real systems, add: "

    "calibration checks, threshold tuning per harm analysis,

      confidence intervals, "

    "and stakeholder review of acceptable trade-offs."

)

# STAGE 4 — DEPLOYMENT & MONITORING (ETHICAL)

@dataclass(frozen=True)

class ModelCard:

    model_name: str

    version: str

    intended_use: str

    out_of_scope_uses: str

    training_data: str

    protected_attributes_audited: List[str]

    key_metrics_overall: Dict[str, float]

    known_limitations: str

    ethical_risks: str

    monitoring_plan: str

model_card = ModelCard(

    model_name="Baseline_LogReg_Demo",

    version="1.0.0",

    intended_use="Educational demo: predict Target from

     Feature_A/B with ethical auditing.",

    out_of_scope_uses="Any real decision impacting

     people (credit, hiring, healthcare) without governance.",

    training_data=f"{provenance.dataset_name} (fingerprint

    {fingerprint_dataframe(df)[:12]}...)",

    protected_attributes_audited=protected_attributes,

    key_metrics_overall={

        "accuracy": float(accuracy_score(y_test, y_pred)),

        "precision": float(precision_score(y_test, y_pred, zero_division=0)),

        "recall": float(recall_score(y_test, y_pred, zero_division=0)),

        "f1": float(f1_score(y_test, y_pred, zero_division=0)),

    },

    known_limitations="Tiny synthetic dataset; fairness metrics

     unreliable; not production-ready.",

    ethical_risks="Representation imbalance; potential subgroup

     error disparities; feedback-loop risk.",

    monitoring_plan=(

        "Track data drift on Feature_A/B; track subgroup TPR/FPR; "

        "alert on metric gaps > threshold; require human review

         on alerts; rollback if needed."

    ),

)

print("\n=== STAGE 4: MODEL CARD (MINIMAL) ===")

print(json.dumps(asdict(model_card), indent=2))

# --- Deployment-like inference wrapper (no PII in logs) ---

def predict_with_monitoring(

    pipeline: Pipeline,

    X_new: pd.DataFrame,

    audit_new: pd.DataFrame | None = None,

) -> Tuple[np.ndarray, Dict[str, object]]:

    """

    In real deployments:

      - log request metadata safely (no PII)

      - log model version + feature summaries

      - optionally log protected attributes ONLY for

        auditing (with proper access controls)

    """

    preds = pipeline.predict(X_new)

    monitoring_payload: Dict[str, object] = {

        "model": model_card.model_name,

        "version": model_card.version,

        "n_requests": int(len(X_new)),

        "feature_summary": {

            "Feature_A_mean": float(X_new["Feature_A"].mean()),

            "Feature_B_mean": float(X_new["Feature_B"].mean()),

        },

    }

    # Optional: subgroup snapshot (ONLY if governed and permitted).

    if audit_new is not None and "Gender" in audit_new.columns:

        dist = (audit_new["Gender"].value_counts(normalize=True) *

         100).round(2).to_dict()

        monitoring_payload["protected_dist_snapshot"] = dist

    return preds, monitoring_payload

# --- Monitoring demo: simulate a small "production batch" with drift +

# subgroup change ---

prod_batch = pd.DataFrame(

{

 "Feature_A": [200, 210, 190, 205],  

  # shifted higher than training => drift signal.

 "Feature_B": [120, 130, 110, 125],

}

)

prod_audit = pd.DataFrame({"Gender": ["Female", "Female", "Female", "Male"]})

preds, payload = predict_with_monitoring(clf, prod_batch, prod_audit)

print("\n=== STAGE 4: INFERENCE + MONITORING HOOK ===")

print("Predictions:", preds.tolist())

print("Monitoring payload:", json.dumps(payload, indent=2))

# --- Simple drift check (mean shift vs training) ---

def simple_mean_drift_check(

    train_df: pd.DataFrame, prod_df: pd.DataFrame, col: str,

    max_rel_change: float = 0.5

) -> str:

    train_mean = float(train_df[col].mean())

    prod_mean = float(prod_df[col].mean())

    if train_mean == 0:

        return f"[ETHICAL NOTE]: Train mean for {col} is 0;

         skip relative drift check."

    rel = abs(prod_mean - train_mean) / abs(train_mean)

    if rel > max_rel_change:

        return (

            f"[ETHICAL WARNING]: Possible drift in '{col}'. "

            f"Train mean={train_mean:.2f}, Prod mean={prod_mean:.2f},

              Rel change={rel:.2f} (> {max_rel_change})."

        )

    return (

        f"[ETHICAL NOTE]: '{col}' drift looks OK. "

        f"Train mean={train_mean:.2f}, Prod mean={prod_mean:.2f},

          Rel change={rel:.2f} (<= {max_rel_change})."

    )

print("\n=== STAGE 4: DRIFT CHECKS ===")

print(simple_mean_drift_check(X_train, prod_batch, "Feature_A", max_rel_change=0.5))

print(simple_mean_drift_check(X_train, prod_batch, "Feature_B", max_rel_change=0.5))

print(

    "\n[ETHICAL NOTE]: In production, monitoring is not just drift.

      You also need: "

    "incident response, audit trails, access control for

     sensitive data, user recourse, "

    "and periodic re-validation with representative samples."

)

Stage 1 code: Data collection and preparation

Now, let's dissect the combined code into its logical parts to understand what each does, starting with Stage 1, data collection and preparation.

So, Stage 1 establishes an ethical foundation for the entire ML pipeline. It aims to ensure the data is appropriate to use, representative, and handled responsibly before any model training occurs.

The imports bring in:

·     Standard utilities (hashlib, json, dataclasses) for metadata tracking and reproducibility.

from __future__ import annotations

import hashlib

import json

from dataclasses import dataclass, asdict

from typing import Dict, List, Tuple

·     Pandas and NumPy for data manipulation.

import numpy as np

import pandas as pd

·     Scikit-learn utilities (only partially used in Stage 1) to prepare for later splitting and modeling.

from sklearn.compose import ColumnTransformer

from sklearn.linear_model import LogisticRegression

from sklearn.metrics import (

    accuracy_score,

    confusion_matrix,

    f1_score,

    precision_score,

    recall_score,

)

from sklearn.model_selection import train_test_split

from sklearn.pipeline import Pipeline

from sklearn.preprocessing import OneHotEncoder, StandardScaler

Even though modeling libraries are imported, Stage 1 does not train a model; it focuses solely on data ethics and preparation.

DataProvenance data class (consent and origin tracking)

The DataProvenance data class captures where the data came from and whether it is ethically viable, including:

·     Source and collection method.

·     Intended purpose of use.

·     Whether consent was obtained.

·     Presence of personally identifiable information (PII).

·     Retention policy.

This reinforces the idea that data legality and consent are first-class requirements, not afterthoughts.

Dataset fingerprinting

This function hashes the dataset contents to produce a stable fingerprint. It allows you to:

·     Prove which exact dataset version was audited.

·     Detect silent data changes later.

·     Reference the dataset safely in documentation or model cards.

def fingerprint_dataframe(df: pd.DataFrame) -> str:

    payload = df.to_csv(index=False).encode("utf-8")

    return hashlib.sha256(payload).hexdigest()

Bias and quality checks

The protected attribute distribution function calculates counts and percentages for a protected attribute, such as gender.

def audit_protected_attribute_distribution(

    df: pd.DataFrame,

    col: str,

    min_pct: float = 30.0,

    min_count: int = 5,

) -> Dict[str, object]:

    counts = df[col].value_counts(dropna=False)

    pct = df[col].value_counts(normalize=True, dropna=False) * 100

    report = {

        "attribute": col,

        "counts": counts.to_dict(),

        "percentages": pct.round(2).to_dict(),

        "warnings": [],

    }

    if float(pct.min()) < min_pct:

        report["warnings"].append(

            f"Imbalance: smallest group is

            {float(pct.min()):.2f}% (< {min_pct}%). "

            "Bias risk for underrepresented groups."

        )

    if int(counts.min()) < min_count:

        report["warnings"].append(

            f"Small-n risk: smallest group has

              {int(counts.min())} samples (< {min_count}). "

            "Metrics will be noisy; consider collecting more data."

        )

    return report

This function also flags ethical risks when a group represents less than 30% of the data (Imbalance) and when a group has very few samples (Small-n risk). This directly addresses representativeness and bias risk.

Missing data audit

This function reports the percentage of missing values per column.

def audit_missingness(df: pd.DataFrame) -> pd.Series:

    return (df.isna().mean() * 100).round(2)

Missing data can disproportionately affect certain groups, so this is an important fairness and data quality check.

Target rate by group

This function examines how often the positive label appears within each protected group.

def audit_target_rate_by_group(df: pd.DataFrame, group_col: str, target_col: str) -> pd.DataFrame:

    out = (

        df.groupby(group_col)[target_col]

        .agg(["count", "mean"])

        .rename(columns={"mean": "target_rate"})

        .reset_index()

    )

    out["target_rate"] = (out["target_rate"] * 100).round(2)

    return out

Significant differences may indicate:

·     Legitimate real-world differences.

·     Label bias, measurement bias, or historical discrimination.

The code explicitly warns that these differences require domain-expert review, not blind acceptance.

Simulated dataset with intentional bias

This dataset intentionally includes an 80/20 gender imbalance.

data = {

    "Feature_A": [10, 20, 30, 40, 50, 60, 70, 80, 90, 100],

    "Feature_B": [5, 15, 25, 35, 45, 55, 65, 75, 85, 95],

    "Gender": ["Male"] * 8 + ["Female"] * 2,  # intentional imbalance

    "Target": [0, 1, 0, 1, 0, 1, 0, 1, 0, 1],

}

df = pd.DataFrame(data)

protected_attributes = ["Gender"]

target_col = "Target"

This is done to demonstrate:

·     How bias appears in real data.

·     How auditing code detects it.

·     Why early warnings matter before modeling.

Provenance validation (ethical gate)

This is a hard ethical stop.

if not provenance.consent_obtained:

    raise RuntimeError(

      "[ETHICAL STOP]: Consent not obtained; do not proceed.")

If consent is missing, the pipeline fails immediately, reinforcing the no-consent, no-model principle.

Representativeness, missingness, and target audits

The script prints:

·     Protected-attribute distributions with warnings.

·     Missing-value percentages.

·     Target rates by group with interpretive guidance.

This makes ethical risks visible and explicit, not hidden in logs.

Sensitive attribute minimization

FEATURES_TO_EXCLUDE_FROM_MODEL = protected_attributes  

# audit-only by default.

MODEL_FEATURES = [c for c in df.columns if c not in [target_col] + FEATURES_TO_EXCLUDE_FROM_MODEL]

Protected attributes (Gender) are:

·     Kept for auditing.

·     Excluded from model features by default.

This follows the ethical principle of data minimization: use sensitive attributes only when clearly justified.

Stratified train/test split

The dataset is split while preserving the protected-attribute distribution in both training and test sets.

X_train, X_test, y_train, y_test, audit_train, audit_test = train_test_split(

    X, y, audit_df, test_size=0.3, random_state=42, stratify=stratify_col

)

This prevents:

·     Minority groups from disappearing in one split.

·     Misleading performance or fairness metrics.

The final display confirms that both splits maintain similar gender proportions.

Stage 2 code: Model development and training

Stage 2 builds a leakage-safe training pipeline and trains a simple baseline model while applying an optional, lightweight fairness mitigation (sample reweighting) to reduce the impact of representation imbalance.

In machine learning, “leakage” usually refers to using information from the test set during training. In practice, this means you must perform the train-test split before normalizing and encoding the training and test data. For example, if you first perform z-score normalization before splitting, you are using information about the entire dataset. Then, when you split, the training data indirectly contains information in the test set.

This is what the demo code does: it correctly first splits, then normalizes and encodes during the preprocessing phase of a pipeline—but it’s not entirely clear that this is what is happening.

Fairness mitigation via reweighting

Here, the code uses the sensitive attribute in the training set (Gender) to compute weights.

def compute_group_weights(series: pd.Series) -> pd.Series:

    probs = series.value_counts(normalize=True)

    w = series.map(lambda g: 1.0 / probs[g])

    return w / w.mean()

sample_weight = compute_group_weights(audit_train["Gender"])

Important nuance:

·     Gender is not used as an input feature.

·     It's used only to reduce imbalance effects during optimization.

So, it's a sensitive attribute used for mitigation, not for prediction.

Define feature types (numeric versus categorical)

This declares which columns should be treated as:

·     Numeric: Scaled for stable learning.

·     Categorical: Would be one-hot encoded (none in this example).

This supports clean, explicit preprocessing decisions.

numeric_features = ["Feature_A", "Feature_B"]

categorical_features: List[str] = []

Leakage-safe preprocessing

preprocess = ColumnTransformer(

    transformers=[

        ("num", StandardScaler(), numeric_features),

        ("cat", OneHotEncoder(handle_unknown="ignore"),

          categorical_features),

    ],

    remainder="drop",

)

This defines the preprocessing steps:

·     StandardScaler() for numeric features (mean/variance scaling, also known as “z-score”).

·     OneHotEncoder(...) placeholder for categoricals.

·     remainder="drop" ensures only the specified columns flow into the model.

Ethical relevance: preprocessing is declared explicitly and can be audited.

Choose a baseline model (logistic regression)

model = LogisticRegression(max_iter=1000)

The code uses logistic regression because it's:

·     A strong baseline.

·     Relatively interpretable.

·     Quick to train and easier to debug than complex models.

This supports the ethical and machine learning good practice of "start simple, document behavior."

Build a pipeline (prevents data leakage)

clf = Pipeline(

    steps=[

        ("preprocess", preprocess),

        ("model", model),

    ]

)

This combines preprocessing plus modeling into one object so that:

·     The scaler/encoder is fit only on training data.

·     Test data is transformed using the training-fitted parameters.

·     Accidental leakage is avoided (such as scaling using the whole dataset).

This is an essential safeguard for correctness and ethics in ML training.

Build a pipeline (prevents data leakage)

clf.fit(X_train, y_train, model__sample_weight=sample_weight)

This trains the pipeline on the training set and passes the weights into the model step:

·     Underrepresented group samples are more affected by the loss.

·     This can reduce the model's tendency to optimize primarily for the majority group.

Ethical note

print(

    "[ETHICAL NOTE]: We used sample reweighting as a mild mitigation for

     representation imbalance. "

    "Upstream data improvement is still the best fix."

)

This print statement reminds the learner that:

·     Reweighting is only a mild, downstream mitigation.

·     The best fix is still upstream data improvement (a more representative collection).

Stage 3 code: Model evaluation and validation

Stage 3 evaluates the trained model in two ways:

·     Overall performance (standard ML metrics).

·     Fairness-aware performance by checking the same metrics per protected group (Gender).

The key ethical idea is that a model can look good overall but still make inaccurate predictions for some subgroups.

Metrics + confusion matrix per group

def group_metrics(

    y_true: np.ndarray,

    y_pred: np.ndarray,

    groups: pd.Series,

) -> pd.DataFrame:

This function computes performance metrics separately for each group in a protected attribute.

How it works:

·     It loops over each unique group value (Male, Female).

·     It builds a Boolean mask idx to select rows for that group.

·     It extracts yt (accurate labels) and yp (predictions) for only that group.

Then it computes a confusion matrix:

tn, fp, fn, tp = confusion_matrix(yt, yp, labels=[0, 1]).ravel()

From that, it derives key error rates:

·     TPR / recall (sensitivity): tp / (tp + fn)
Of the true positives, how many did we correctly catch?

·     FPR: fp / (fp + tn)
Of the true negatives, how many did we incorrectly flag?

It also calculates standard metrics per group:

·     Accuracy, precision, F1 (the harmonic mean/average of precision and recall).

·     It includes TN/FP/FN/TP counts for transparency.

Finally, it returns a DataFrame of results by group.

Why this matters ethically: it exposes where the model fails, not just average performance.

Simple fairness guardrail

def fairness_gap_report(df_metrics: pd.DataFrame, metric: str, max_gap: float = 0.20) -> str:

This function checks whether a metric differs too much across groups. It computes: gap = float(vals.max() - vals.min()):

·     If the gap exceeds a threshold (default 0.20), it prints an ethical warning.

·     If there aren't at least two groups, it notes that a gap can't be computed.

This is a pragmatic red-flag mechanism: it’s not a full fairness framework, but a useful early alert.

Generate predictions on the test set

y_pred = clf.predict(X_test)

This runs the trained pipeline on unseen test data to get predictions for evaluation.

Print overall metrics (standard performance)

print("Overall metrics:")

print(f"  Accuracy : {accuracy_score(y_test, y_pred):.3f}")

print(f"  Precision: {precision_score(y_test, y_pred, zero_division=0):.3f}")

print(f"  Recall   : {recall_score(y_test, y_pred, zero_division=0):.3f}")

print(f"  F1       : {f1_score(y_test, y_pred, zero_division=0):.3f}")

This reports the typical "headline" metrics for the full test set:

·     Accuracy: Overall correctness.

·     Precision: How reliable positive predictions are.

·     Recall: How many true positives are found.

·     F1: Balance between precision and recall.

This tells you how the model performs on average, but not whether it is fair.

Compute and print group-sliced metrics

gm = group_metrics(y_test.to_numpy(), y_pred, audit_test["Gender"])

print("\nGroup-sliced metrics (by Gender):")

print(gm.to_string(index=False))

This produces the subgroup metrics table (by Gender), so you can compare:

·     Performance differences.

·     Error patterns (FP/FN) by group.

·     Whether one group systematically has more false alarms or misses.

Run fairness checks on key error rates

print("\nFairness checks (simple guardrails):")

print(" ", fairness_gap_report(gm, "recall_TPR", max_gap=0.20))

print(" ", fairness_gap_report(gm, "FPR", max_gap=0.20))

These checks specifically focus on:

·     Recall/TPR gap: Are we missing positives more often for one group?

·     FPR gap: Are we falsely accusing/flagging one group more often?

Those differences are often the most ethically important because they signal the possibility of inaccurate predictions and indirect harm.

Stage 4 code: Deployment and monitoring

Stage 4 shows how to "productionize" the model responsibly by:

·     Documenting what the model is for (and not for).

·     Logging safe monitoring signals (without PII).

·     Tracking distribution shifts (drift).

·     Setting up guardrails for escalation (alerts, human review, and rollback).

A model that was fair enough at launch can become unsafe or unfair over time.

ModelCard Dataclass (minimal governance documentation)

@dataclass(frozen=True)

class ModelCard:

This defines a simple structure for model documentation, including:

·     Intended use and out-of-scope uses.

·     Training dataset reference (with fingerprint).

·     Which protected attributes were audited?

·     Key overall metrics.

·     Known limitations and ethical risks.

·     A monitoring plan.

Why this matters: it creates an auditable "contract" for how the model should be used and maintained.

Inference wrapper with safe logging hooks

def predict_with_monitoring(

    pipeline: Pipeline,

    X_new: pd.DataFrame,

    audit_new: pd.DataFrame | None = None,

) -> Tuple[np.ndarray, Dict[str, object]]:

This function behaves like a deployment inference endpoint:

·     Runs pipeline.predict(X_new) to generate predictions.

·     Returns both:

o     Predictions.

o     A monitoring_payload dictionary containing safe telemetry.

The key ethical point is that it logs summaries, not raw user-level details.

Build a monitoring payload (no PII)

monitoring_payload: Dict[str, object] = {

        "model": model_card.model_name,

        "version": model_card.version,

        "n_requests": int(len(X_new)),

        "feature_summary": {

            "Feature_A_mean": float(X_new["Feature_A"].mean()),

            "Feature_B_mean": float(X_new["Feature_B"].mean()),

        },

    }

This records:

·     Model/version produced results.

·     Requests served.

·     Basic feature statistics (means).

These signals help detect drift without storing sensitive data.

Optional protected attribute snapshot (governed use)

if audit_new is not None and "Gender" in audit_new.columns:

   dist = (audit_new["Gender"].value_counts(normalize=True) * 

     100).round(2).to_dict()

   monitoring_payload["protected_dist_snapshot"] = dist

If provided, the function computes the current distribution of the protected attribute in production traffic. It should only be collected when permitted and controlled (through access controls and governance). It's used for monitoring bias risk (such as a sudden demographic shift).

Simulate "production" data with drift

prod_batch = pd.DataFrame(

    {

        "Feature_A": [200, 210, 190, 205],  

        # shifted higher than training => drift signal.

        "Feature_B": [120, 130, 110, 125],

    }

)

prod_audit = pd.DataFrame({"Gender": ["Female", "Female", "Female", "Male"]})

This creates a fake production batch in which feature values are much higher than in the training data.

This demonstrates a real deployment problem:

·     The model starts seeing data from a different regime, where performance and fairness may degrade.

The prod_audit dataframe also simulates a shift in group mix (mostly Female), showing how subgroup representation can change.

Run inference and print monitoring output

preds, payload = predict_with_monitoring(clf, prod_batch, prod_audit)

print("Predictions:", preds.tolist())

print("Monitoring payload:", json.dumps(payload, indent=2))

This shows the deployment flow to:

·     Get predictions.

·     Generate monitoring metadata.

·     Print both so you can see what would be logged or what would be alerted on.

Basic drift detector

def simple_mean_drift_check(

    train_df: pd.DataFrame, prod_df: pd.DataFrame, col: str,

    max_rel_change: float = 0.5

) -> str:

This function compares training versus production:

·     Computes relative change: abs(prod_mean - train_mean) / abs(train_mean).

·     If it exceeds a threshold (50% by default), it raises an ethics warning.

It's intentionally simple, but it illustrates the idea: monitor the input distribution, as drift often precedes failures.

Drift checks for both features

print(simple_mean_drift_check(X_train, prod_batch,

  "Feature_A", max_rel_change=0.5))

print(simple_mean_drift_check(X_train, prod_batch,

  "Feature_B", max_rel_change=0.5))

This runs the drift check across both features and prints warning messages.

Recap

Together, the four stages form a complete ethical ML lifecycle:

Stage 1 establishes trust in the data by documenting provenance and consent, auditing representativeness, detecting imbalance and bias, minimizing the use of sensitive attributes, and creating fair train/test splits.

Stage 2 builds a transparent, leakage-safe training pipeline using a simple baseline model and optional reweighting to reduce the impact of data imbalance while keeping sensitive attributes out of prediction.

Stage 3 evaluates the model not only on overall performance, but also on subgroup-level error patterns, using group-sliced metrics and guardrails to surface potential fairness risks that averages can hide.

Stage 4 operationalizes the model responsibly by documenting intended use and risks, wrapping inference with privacy-aware monitoring, detecting drift and subgroup shifts in production, and defining escalation paths that keep humans accountable after deployment.

Scroll To Top
Disclaimer

DISCLAIMER: Web reader is currently in beta. Please report any issues through our support system. PDF and Kindle format files are also available for download.

Previous

Next



You are one step away from downloading ebooks from the Succinctly® series premier collection!
A confirmation has been sent to your email address. Please check and confirm your email subscription to complete the download.