Membership Inference: Audit a Model for Training-Data Leakage Before It Ships

"""Harbor Health: a privacy audit before a readmission model ships. Membership inference asks one question:
given a patient's record, can someone tell whether it was in the training set? For a hospital model, a yes
reveals that the person was a patient. You measure that leak, then remove it."""
import csv

import numpy as np
from sklearn.ensemble import RandomForestClassifier

COLS = ["age", "bmi", "systolic_bp", "glucose", "prior_visits", "length_of_stay", "medications", "a1c"]


def load(path):
    """(X, y) from one of the record files."""
    X, y = [], []
    with open(path) as f:
        for r in csv.DictReader(f):
            X.append([float(r[c]) for c in COLS])
            y.append(int(r["readmitted"]))
    return np.array(X), np.array(y)


def load_all():
    """members = the training set; nonmembers = same population, never trained on; test = for accuracy."""
    return {"members": load("members.csv"), "nonmembers": load("nonmembers.csv"), "test": load("test.csv")}


def train_overfit(X, y):
    """The model the team was about to ship: a random forest grown to full depth."""
    return RandomForestClassifier(n_estimators=100, random_state=0, n_jobs=1).fit(X, y)


class _ScoresHidden:
    def __init__(self, model):
        self.model = model

    def predict(self, X):
        return self.model.predict(X)

    def predict_proba(self, X):
        return np.full((len(X), 2), 0.5)

    def score(self, X, y):
        return float((self.predict(X) == y).mean())


def hide_scores(model):
    """The shortcut someone will propose: serve the same model, but return only labels. Every confidence the
    API exposes becomes 0.5."""
    return _ScoresHidden(model)


def true_label_confidence(model, X, y):
    """The probability the model gives to each record's true label."""
    # TODO (Step 1): predict_proba, picking each row's true-label column.
    raise NotImplementedError("Step 1: write true_label_confidence()")


def confidence_gap(model, data):
    """Mean true-label confidence on members minus on nonmembers. Near 0 means the model treats records it
    trained on like any other record; a large gap is what an attacker exploits."""
    # TODO (Step 1): mean true-label confidence on members minus on nonmembers.
    raise NotImplementedError("Step 1: write confidence_gap()")


def loss(model, X, y):
    """Per-record loss: -log of the true-label confidence, clipped at 1e-6 so a zero stays finite."""
    raise NotImplementedError("loss() arrives in Step 2")


def attack_auc(member_losses, nonmember_losses):
    """AUC of the loss-threshold attack: the chance a random member has a lower loss than a random nonmember,
    counting ties as half. 0.5 means the attack learns nothing."""
    raise NotImplementedError("attack_auc() arrives in Step 2")


def tpr_at_fpr(member_losses, nonmember_losses, fpr=0.01):
    """Flag a record as a member when its loss is at or below a threshold. Over every threshold that flags at
    most `fpr` of nonmembers, return the largest share of members flagged."""
    raise NotImplementedError("tpr_at_fpr() arrives in Step 3")


def advantage(member_losses, nonmember_losses):
    """Membership advantage: the best TPR - FPR over all thresholds. 0 means no leak."""
    raise NotImplementedError("advantage() arrives in Step 3")


def audit(model, data):
    """One report: train and test accuracy, attack AUC, TPR at 1% FPR, and membership advantage."""
    raise NotImplementedError("audit() arrives in Step 4")


def train_private(X, y):
    """Your defence: a model that generalises instead of memorising, at about the same test accuracy."""
    raise NotImplementedError("train_private() arrives in Step 4")


def label_only_auc(model, data):
    """The attack when the API returns only the predicted label: a member is guessed when the prediction is
    correct. Hiding confidences does not help a model that memorised its training set."""
    raise NotImplementedError("label_only_auc() arrives in Step 5")


def release_gate(model, data, max_auc=0.6):
    """Block the release if either attack beats max_auc. Returns {"release", "attack_auc", "label_auc"}."""
    raise NotImplementedError("release_gate() arrives in Step 5")