Ethics, Governance, and Compliance for Federal AI

import pandas as pd
import numpy as np
from sklearn.metrics import confusion_matrix

def bias_audit_report(
    y_true: np.ndarray,
    y_pred: np.ndarray,
    protected_col: pd.Series,
    threshold: float = 0.5,
    reference_group: str = None,
) -> pd.DataFrame:
    """
    Compute demographic parity and equalized odds metrics.
    Returns a DataFrame with per-group statistics.
    The dp_ratio < 0.80 threshold is the EEOC 4/5ths rule.
    """
    groups = protected_col.unique()
    results = []

    for group in groups:
        mask = (protected_col == group).values
        n    = mask.sum()
        if n < 20:
            continue

        tn, fp, fn, tp = confusion_matrix(y_true[mask], y_pred[mask]).ravel()
        positive_rate  = (tp + fp) / n
        tpr = tp / (tp + fn) if (tp + fn) > 0 else 0.0
        fpr = fp / (fp + tn) if (fp + tn) > 0 else 0.0

        results.append({
            "group":         group,
            "n":             n,
            "positive_rate": round(float(positive_rate), 3),
            "tpr":           round(float(tpr), 3),
            "fpr":           round(float(fpr), 3),
        })

    result = pd.DataFrame(results).sort_values("positive_rate")

    if reference_group is None:
        reference_group = result.iloc[-1]["group"]

    ref_row = result[result["group"] == reference_group].iloc[0]
    result["dp_ratio"]  = (result["positive_rate"] / ref_row["positive_rate"]).round(3)
    result["fpr_ratio"] = (result["fpr"] / ref_row["fpr"]).round(3)

    print(f"\nBias Audit  |  Threshold: {threshold}  |  Reference: {reference_group}")
    print("=" * 70)
    print(result.to_string(index=False))

    flagged = result[result["dp_ratio"] < 0.80]
    if len(flagged) > 0:
        print(f"\nPOTENTIAL DISPARATE IMPACT (dp_ratio < 0.80):")
        for _, row in flagged.iterrows():
            print(f"  Group '{row['group']}': dp_ratio={row['dp_ratio']:.3f}")
        print(f"\nAction required: Review features, adjust threshold,")
        print(f"or apply post-processing fairness constraint before deployment.")
    else:
        print(f"\nNo groups below 4/5ths threshold.")

    return result

from scipy.stats import pointbiserialr, chi2_contingency
import itertools

def proxy_correlation_scan(
    X: pd.DataFrame,
    protected_cols: list,
    feature_cols: list,
    correlation_threshold: float = 0.15,
) -> pd.DataFrame:
    """
    Identify features with potentially concerning correlations to protected attributes.
    Flagged features require documentation and monitoring — not automatic exclusion.
    """
    results = []

    for feat_col, prot_col in itertools.product(feature_cols, protected_cols):
        if X[feat_col].nunique() < 2 or X[prot_col].nunique() < 2:
            continue

        if X[feat_col].dtype in [np.float64, np.float32, np.int64, np.int32]:
            if X[prot_col].nunique() == 2:
                corr, _ = pointbiserialr(
                    X[prot_col].astype(float), X[feat_col].astype(float)
                )
                abs_corr = abs(corr)
            else:
                from sklearn.feature_selection import f_classif
                f_stat, _ = f_classif(X[[feat_col]].fillna(0),
                                       X[prot_col].astype("category").cat.codes)
                abs_corr = min(float(f_stat[0]) / (float(f_stat[0]) + len(X)), 1.0)
        else:
            contingency = pd.crosstab(X[feat_col], X[prot_col])
            chi2, _, _, _ = chi2_contingency(contingency)
            n = contingency.sum().sum()
            k = min(contingency.shape) - 1
            abs_corr = float(np.sqrt(chi2 / (n * k))) if (n * k) > 0 else 0.0

        if abs_corr >= correlation_threshold:
            results.append({
                "feature":             feat_col,
                "protected_attribute": prot_col,
                "correlation":         round(abs_corr, 4),
            })

    report = pd.DataFrame(results).sort_values("correlation", ascending=False)
    if len(report) > 0:
        print(f"Proxy correlation scan — {len(report)} feature-attribute pairs above threshold:")
        print(report.to_string(index=False))
    return report

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.catalog import LineageDirection

w = WorkspaceClient()

def get_table_lineage(table_full_name: str) -> dict:
    """
    Retrieve upstream lineage for a Delta table using Unity Catalog API.
    Use this to document training data provenance in a model card or RAI assessment.
    """
    lineage = w.lineage_tracking.table_lineage(
        table_name=table_full_name,
        direction=LineageDirection.UPSTREAM,
    )

    upstream_tables = []
    written_by      = []

    for node in lineage.upstreams or []:
        if hasattr(node, "table_info") and node.table_info:
            upstream_tables.append(node.table_info.full_name)
        if hasattr(node, "notebook_info") and node.notebook_info:
            written_by.append({"type": "notebook", "path": node.notebook_info.path})

    result = {"table": table_full_name, "upstream_tables": upstream_tables,
              "written_by": written_by}

    print(f"Lineage for {table_full_name}:")
    for t in upstream_tables:
        print(f"  Upstream: {t}")

    return result


# Document training data provenance for model card
lineage = get_table_lineage("jupiter_catalog.silver.maintenance_features")