scripts/bias_audit.py

from __future__ import annotations

from pathlib import Path

import joblib
import numpy as np
import pandas as pd
from sklearn.metrics import roc_auc_score
from sklearn.model_selection import train_test_split

from scripts.train_xgb import TARGET, _read_joined, _select_columns, _freq_encode


def _group_metrics(y: np.ndarray, p: np.ndarray, group: pd.Series) -> pd.DataFrame:
    rows = []
    group = group.fillna("missing").astype(str)
    group = group.reset_index(drop=True)
    for g in group.unique().tolist():
        mask = (group == g).to_numpy()
        n = int(mask.sum())
        if n < 500:
            continue
        yy = y[mask]
        pp = p[mask]
        auc = float("nan")
        if len(np.unique(yy)) == 2:
            auc = float(roc_auc_score(yy, pp))
        rows.append(
            {
                "group": str(g),
                "n": n,
                "fraud_rate": float(np.mean(yy)),
                "mean_score": float(np.mean(pp)),
                "roc_auc": auc,
            }
        )
    return pd.DataFrame(rows).sort_values("n", ascending=False)


def main() -> None:
    repo_root = Path(__file__).resolve().parents[1]
    raw_dir = repo_root / "data" / "raw"
    artifacts_path = repo_root / "artifacts" / "model.joblib"
    out_dir = repo_root / "reports"
    out_dir.mkdir(parents=True, exist_ok=True)

    df = _read_joined(raw_dir, max_rows=200000)
    df, numeric, categorical = _select_columns(df)
    df = df.dropna(subset=[TARGET]).copy()

    y = df[TARGET].astype(int).to_numpy()
    X = df.drop(columns=[TARGET])

    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.25, random_state=42, stratify=y
    )
    X_test = X_test.reset_index(drop=True)

    payload = joblib.load(artifacts_path)
    model = payload["model"]
    imputer = payload["imputer"]
    cat_cols = list(payload.get("categorical_cols") or [])

    X_train_enc, X_test_enc, _ = _freq_encode(X_train, X_test, cat_cols)
    Xte = imputer.transform(X_test_enc)
    p = model.predict_proba(Xte)[:, 1]

    product = X_test.get("ProductCD", pd.Series([None] * len(X_test)))
    device = X_test.get("DeviceType", pd.Series([None] * len(X_test)))

    product_table = _group_metrics(y_test, p, product)
    device_table = _group_metrics(y_test, p, device)

    md = []
    md.append("## Bias audit (limited slices)\n")
    md.append(
        "This audit checks basic performance skews across a couple of available categorical slices. It is not a substitute for a full fairness review.\n"
    )
    md.append("### Slice: ProductCD\n")
    md.append(product_table.to_markdown(index=False) if not product_table.empty else "_Not available._")
    md.append("\n\n### Slice: DeviceType\n")
    md.append(device_table.to_markdown(index=False) if not device_table.empty else "_Not available._")
    md.append("\n")

    out_path = out_dir / "bias_audit.md"
    out_path.write_text("\n".join(md))
    print(f"Wrote {out_path}")


if __name__ == "__main__":
    main()