Ethics & Best Practices

# Technique 1: Balance representation
import numpy as np
import pandas as pd
from scipy.stats import chi2_contingency
from sklearn.utils import resample
from sklearn.utils.class_weight import compute_sample_weight


def balance_dataset(df, protected_attr, target_col, random_state=42):
    """
    Downsample every (group, label) cell to the size of the smallest one.

    Balancing on the protected attribute alone can leave the label skewed
    inside a group, so the cells are (protected_attr, target_col) pairs.
    """
    cells = df.groupby([protected_attr, target_col], observed=True)
    min_size = cells.size().min()

    # replace=False matters: resample() defaults to replace=True, which turns a
    # downsample into a bootstrap and drops distinct rows from the very group
    # the balancing exists to protect.
    balanced_dfs = [
        resample(cell_df, n_samples=min_size, replace=False, random_state=random_state)
        for _, cell_df in cells
    ]

    return pd.concat(balanced_dfs, ignore_index=True)


# Technique 2: Remove biased correlations
def _as_categorical(s, bins=10, max_levels=20):
    """
    Bin a numeric column so it can be crosstabbed against a categorical one.

    Only high-cardinality columns are binned. Running qcut on a 0/1 flag
    collapses it to a single bin, and a one-column table scores 0, so a
    perfect binary proxy would be graded as unrelated.
    """
    if pd.api.types.is_numeric_dtype(s) and s.nunique() > max_levels:
        return pd.qcut(s, q=bins, duplicates='drop')
    return s


def cramers_v(a, b):
    """
    Association between two columns on a 0-1 scale.

    Pearson correlation is undefined for categories, so the chi-squared
    statistic is normalised by the sample size and by the smaller dimension
    of the contingency table.
    """
    table = pd.crosstab(_as_categorical(a), _as_categorical(b))
    if min(table.shape) < 2:
        return 0.0

    chi2 = chi2_contingency(table).statistic
    n = table.to_numpy().sum()

    return float(np.sqrt(chi2 / (n * (min(table.shape) - 1))))


def remove_proxy_features(df, protected_attrs, target_col=None, threshold=0.7):
    """
    Drop the protected attributes and every feature that stands in for one.

    Example: ZIP code often correlates with race, income

    Both halves are needed. Dropping 'race' on its own leaves a ZIP code that
    is determined by race to carry the same information back in, and dropping
    the proxy while keeping the attribute is not a fairness measure at all.

    target_col is excluded from the scan. Left in, the label is itself a proxy
    candidate, so a label that correlates with a protected attribute, which is
    the exact situation under investigation, would be silently deleted.

    Numeric pairs use Pearson correlation; any pair involving a categorical
    column uses Cramer's V. pd.api.types.is_numeric_dtype is the check that
    holds across pandas versions: on pandas 3 a plain text column is
    StringDtype, so a dtype == 'object' test silently misses it and .corr()
    then raises trying to cast 'M' to a float.
    """
    skip = set(protected_attrs) | ({target_col} if target_col else set())
    proxy_features = []

    for col in df.columns:
        if col in skip:
            continue

        for protected in protected_attrs:
            if (pd.api.types.is_numeric_dtype(df[col])
                    and pd.api.types.is_numeric_dtype(df[protected])):
                strength = abs(df[col].corr(df[protected]))
            else:
                strength = cramers_v(df[col], df[protected])

            if strength > threshold:
                proxy_features.append(col)
                break   # one match is enough, and it keeps the list unique

    print(f"Identified proxy features: {proxy_features}")
    return df.drop(columns=[*protected_attrs, *proxy_features])


# Technique 3: Reweighting samples
def compute_fair_weights(y, protected_attr):
    """
    Compute sample weights to balance groups.

    Gives higher weight to underrepresented groups.
    """
    # Combine target and protected attribute
    combined = y.astype(str) + "_" + protected_attr.astype(str)

    weights = compute_sample_weight('balanced', combined)

    return weights

# Use in training:
# model.fit(X, y, sample_weight=compute_fair_weights(y, df['gender']))


# Example usage
if __name__ == "__main__":
    rng = np.random.default_rng(42)
    n_samples = 1000

    race = rng.choice(['White', 'Black', 'Asian', 'Hispanic'], n_samples)

    # ZIP is stored as text and is almost fully determined by race. This is the
    # classic proxy: drop 'race' from the model and the ZIP column carries it
    # straight back in.
    zip_by_race = {'White': '90210', 'Black': '10001',
                   'Asian': '60601', 'Hispanic': '73301'}
    zip_code = np.array([zip_by_race[r] for r in race])

    df = pd.DataFrame({
        'gender': rng.choice(['M', 'F'], n_samples, p=[0.8, 0.2]),
        'race': race,
        'zip_code': zip_code,
        'income': rng.integers(20_000, 150_000, n_samples),
        'approved': rng.integers(0, 2, n_samples),
    })

    print("Association with 'race':")
    for col in ['zip_code', 'income', 'approved']:
        print(f"  {col:10s} Cramer's V = {cramers_v(df[col], df['race']):.3f}")

    print()
    print(f"Before: {df.shape[1]} columns {list(df.columns)}")
    clean = remove_proxy_features(df, ['gender', 'race'], target_col='approved')
    print(f"After:  {clean.shape[1]} columns {list(clean.columns)}")

    print("\nGroup sizes before balancing:")
    print(df.groupby(['gender', 'approved']).size().to_string())

    balanced = balance_dataset(df, 'gender', 'approved')
    print("\nGroup sizes after balancing:")
    print(balanced.groupby(['gender', 'approved']).size().to_string())

    weights = compute_fair_weights(df['approved'], df['gender'])
    print("\nMean sample weight by group:")
    print(pd.DataFrame({'gender': df['gender'], 'weight': weights})
          .groupby('gender')['weight'].mean().round(3).to_string())

import numpy as np
from sklearn.base import BaseEstimator, ClassifierMixin
from sklearn.linear_model import LogisticRegression


class FairClassifier(ClassifierMixin, BaseEstimator):
    """
    Wrap a classifier and train it toward demographic parity.

    In-processing changes the training loop itself. Each round refits the base
    model with new sample weights: the group that gets selected too often has
    its positive examples down-weighted and the other group's are up-weighted,
    until the selection rates converge.

    ClassifierMixin is listed before BaseEstimator, and the order is load
    bearing. Since scikit-learn 1.6 the estimator type is resolved through
    __sklearn_tags__, and BaseEstimator's implementation wins the MRO when it
    comes first, so is_classifier() returns False and cross-validation silently
    falls back to KFold instead of StratifiedKFold.
    """

    def __init__(self, base_model, tolerance=0.05, max_rounds=30, step=0.2):
        self.base_model = base_model
        self.tolerance = tolerance
        self.max_rounds = max_rounds
        self.step = step

    def fit(self, X, y, sensitive_features=None):
        """
        Train under a demographic parity target.

        sensitive_features is a keyword with a default, matching fairlearn, so
        that clone() and get_params() keep working. That is all it buys:
        cross_val_score still fails here, because it calls fit(X, y) and this
        raises. Feeding a third array through cross-validation needs
        scikit-learn metadata routing (set_fit_request), not a signature tweak.
        """
        if sensitive_features is None:
            raise ValueError("sensitive_features is required")

        X, y = np.asarray(X), np.asarray(y)
        groups = np.asarray(sensitive_features)
        self.classes_ = np.unique(y)

        weights = np.ones(len(y), dtype=float)
        self.history_ = []

        for _ in range(self.max_rounds):
            self.base_model.fit(X, y, sample_weight=weights)
            predictions = self.base_model.predict(X)

            rates = {g: predictions[groups == g].mean() for g in np.unique(groups)}
            over = max(rates, key=rates.get)
            under = min(rates, key=rates.get)
            gap = rates[over] - rates[under]

            self.history_.append(gap)
            if gap < self.tolerance:
                break

            # Move weight off the favored group's positives and onto the other's.
            # Only positives are touched: demographic parity is about who gets
            # selected, so the negatives carry no signal about the gap.
            weights = np.where((groups == over) & (y == 1),
                               weights * (1 - self.step), weights)
            weights = np.where((groups == under) & (y == 1),
                               weights / (1 - self.step), weights)

        self.sample_weight_ = weights
        return self

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


def parity_gap(predictions, groups):
    """Largest difference in selection rate between any two groups."""
    rates = [predictions[groups == g].mean() for g in np.unique(groups)]
    return max(rates) - min(rates)


if __name__ == "__main__":
    from sklearn.base import is_classifier

    rng = np.random.default_rng(42)
    n_samples = 4000

    # Group membership genuinely shifts the label, so an unconstrained fit is
    # accurate and unfair at the same time. That is the tension being managed.
    group = rng.integers(0, 2, n_samples)
    features = rng.normal(size=(n_samples, 3))
    y = ((features[:, 0] + 1.5 * group
          + rng.normal(scale=0.5, size=n_samples)) > 0).astype(int)
    X = np.column_stack([features, group])

    baseline = LogisticRegression(max_iter=1000).fit(X, y)
    baseline_pred = baseline.predict(X)
    print(f"Unconstrained: gap={parity_gap(baseline_pred, group):.4f} "
          f"accuracy={(baseline_pred == y).mean():.4f}")

    fair = FairClassifier(LogisticRegression(max_iter=1000), tolerance=0.05)
    print(f"\nis_classifier(fair) = {is_classifier(fair)}")
    fair.fit(X, y, sensitive_features=group)

    print(f"Reweighting rounds: {len(fair.history_)}")
    for round_number, gap in enumerate(fair.history_):
        if round_number % 3 == 0 or round_number == len(fair.history_) - 1:
            print(f"  round {round_number:2d}: gap={gap:.4f}")

    fair_pred = fair.predict(X)
    print(f"\nConstrained:   gap={parity_gap(fair_pred, group):.4f} "
          f"accuracy={(fair_pred == y).mean():.4f}")

# pip install fairlearn==0.14.0

import numpy as np
from fairlearn.metrics import demographic_parity_difference
from fairlearn.reductions import DemographicParity, ExponentiatedGradient
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score
from sklearn.model_selection import train_test_split

rng = np.random.default_rng(42)
n_samples = 4000

group = rng.integers(0, 2, n_samples)
features = rng.normal(size=(n_samples, 3))
y = ((features[:, 0] + 1.5 * group
      + rng.normal(scale=0.5, size=n_samples)) > 0).astype(int)
X = np.column_stack([features, group])

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

baseline = LogisticRegression(max_iter=1000).fit(X_train, y_train)

# ExponentiatedGradient solves the constrained problem properly instead of
# nudging weights: it returns a weighted ensemble of classifiers.
fair_model = ExponentiatedGradient(
    estimator=LogisticRegression(max_iter=1000),
    constraints=DemographicParity()
)
fair_model.fit(X_train, y_train, sensitive_features=g_train)

# predict() picks a predictor from that ensemble per row, so it is randomized
# by design. Pass random_state or the numbers move between runs.
results = {
    "unconstrained": baseline.predict(X_test),
    "DemographicParity": fair_model.predict(X_test, random_state=42),
}

for name, predictions in results.items():
    dp = demographic_parity_difference(
        y_test, predictions, sensitive_features=g_test
    )
    print(f"{name:18s} accuracy={accuracy_score(y_test, predictions):.3f} "
          f"dp_difference={dp:.3f}")

import numpy as np


def calibrate_by_group(model, X, y_true, protected_attr, target_fpr=0.1):
    """
    Pick a decision threshold per group so every group lands on the same
    false positive rate.

    A false positive can only come from a true negative, so the threshold is
    read off the negatives alone and the labels are required. Thresholding on
    every score in the group instead controls the selection rate, which is
    demographic parity: a different criterion that leaves FPR untouched.

    Args:
        model: Trained model with predict_proba
        X: Features
        y_true: True labels, needed to identify the negatives
        protected_attr: Protected attribute values
        target_fpr: False positive rate to hold every group to

    Returns:
        (predictions, thresholds), where thresholds maps group -> cutoff
    """
    probabilities = model.predict_proba(X)[:, 1]
    predictions = np.zeros(len(X), dtype=int)
    thresholds = {}

    for group in np.unique(protected_attr):
        group_mask = protected_attr == group
        negatives = group_mask & (y_true == 0)

        # Cut at the score that leaves target_fpr of the negatives above it.
        thresholds[group] = float(
            np.quantile(probabilities[negatives], 1 - target_fpr)
        )

        predictions[group_mask] = (
            probabilities[group_mask] >= thresholds[group]
        ).astype(int)

    return predictions, thresholds


def false_positive_rates(predictions, y_true, protected_attr):
    """FPR per group: of the true negatives, how many were flagged."""
    return {
        group: predictions[(protected_attr == group) & (y_true == 0)].mean()
        for group in np.unique(protected_attr)
    }


if __name__ == "__main__":
    from sklearn.linear_model import LogisticRegression
    from sklearn.model_selection import train_test_split

    rng = np.random.default_rng(42)
    n_samples = 20000

    group = rng.integers(0, 2, n_samples)
    features = rng.normal(size=(n_samples, 3))
    y = ((features[:, 0] + 1.5 * group
          + rng.normal(scale=0.5, size=n_samples)) > 0).astype(int)
    X = np.column_stack([features, group])

    # One half calibrates the thresholds, the other half never informs them.
    X_cal, X_test, y_cal, y_test, g_cal, g_test = train_test_split(
        X, y, group, test_size=0.5, random_state=42
    )

    model = LogisticRegression(max_iter=1000).fit(X_cal, y_cal)

    calibrated, thresholds = calibrate_by_group(
        model, X_cal, y_cal, g_cal, target_fpr=0.1
    )

    print("Per-group thresholds:")
    for group_value, threshold in thresholds.items():
        print(f"  group {group_value}: {threshold:.4f}")

    print("\nFPR on the calibration half:")
    for group_value, rate in false_positive_rates(calibrated, y_cal, g_cal).items():
        print(f"  group {group_value}: {rate:.3f}")

    # Apply the same thresholds to data they were not fitted on.
    probabilities = model.predict_proba(X_test)[:, 1]
    single = (probabilities >= 0.5).astype(int)
    per_group = np.zeros(len(X_test), dtype=int)
    for group_value, threshold in thresholds.items():
        mask = g_test == group_value
        per_group[mask] = (probabilities[mask] >= threshold).astype(int)

    print("\nFPR on held-out data:")
    for label, predictions in [("single 0.5 threshold", single),
                               ("per-group thresholds", per_group)]:
        rates = false_positive_rates(predictions, y_test, g_test)
        gap = max(rates.values()) - min(rates.values())
        detail = "  ".join(f"group {k}={v:.3f}" for k, v in rates.items())
        print(f"  {label:22s} {detail}  gap={gap:.3f}")

# pip install fairlearn==0.14.0

import numpy as np
from fairlearn.metrics import equalized_odds_difference
from fairlearn.postprocessing import ThresholdOptimizer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score
from sklearn.model_selection import train_test_split

rng = np.random.default_rng(42)
n_samples = 20000

group = rng.integers(0, 2, n_samples)
features = rng.normal(size=(n_samples, 3))
y = ((features[:, 0] + 1.5 * group
      + rng.normal(scale=0.5, size=n_samples)) > 0).astype(int)
X = np.column_stack([features, group])

X_train, X_rest, y_train, y_rest, g_train, g_rest = train_test_split(
    X, y, group, test_size=0.5, random_state=42
)
X_val, X_test, y_val, y_test, g_val, g_test = train_test_split(
    X_rest, y_rest, g_rest, test_size=0.5, random_state=42
)

trained_model = LogisticRegression(max_iter=1000).fit(X_train, y_train)

# prefit=True is what stops ThresholdOptimizer from throwing away the model
# just trained. It defaults to False, which refits the estimator from scratch.
threshold_optimizer = ThresholdOptimizer(
    estimator=trained_model,
    constraints="equalized_odds",
    prefit=True,
)
threshold_optimizer.fit(X_val, y_val, sensitive_features=g_val)

# predict() applies a randomized threshold rule, so pin random_state or the
# reported numbers move between runs.
fair_predictions = threshold_optimizer.predict(
    X_test, sensitive_features=g_test, random_state=42
)
baseline_predictions = trained_model.predict(X_test)

for name, predictions in [("unadjusted", baseline_predictions),
                          ("equalized_odds", fair_predictions)]:
    eo = equalized_odds_difference(
        y_test, predictions, sensitive_features=g_test
    )
    print(f"{name:15s} accuracy={accuracy_score(y_test, predictions):.3f} "
          f"eo_difference={eo:.3f}")

# Bad: Collecting everything
user_data = {
    'name': 'John Doe',
    'ssn': '123-45-6789',
    'address': '123 Main St',
    'phone': '555-1234',
    'email': 'john@example.com',
    'browsing_history': [...],
    'purchase_history': [...],
    'location_history': [...]
}

# Good: Collect only what you need for the task
# For a recommendation system:
user_data = {
    'user_id': 'uuid-123',  # Pseudonymous identifier
    'age_group': '25-34',   # Aggregated, not exact age
    'category_preferences': ['electronics', 'books'],  # Not full history
    'region': 'Northeast'   # Not exact location
}


# Automatically delete old data
import sqlite3
from datetime import datetime, timedelta, timezone


def cleanup_old_data(db, retention_days=90):
    """Delete interaction rows older than the retention period."""
    cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)

    cursor = db.execute(
        "DELETE FROM user_interactions WHERE created_at < ?",
        (cutoff.isoformat(),)
    )
    db.commit()

    return cursor.rowcount


if __name__ == "__main__":
    db = sqlite3.connect(":memory:")
    db.execute("CREATE TABLE user_interactions (user_id TEXT, created_at TEXT)")

    now = datetime.now(timezone.utc)
    db.executemany(
        "INSERT INTO user_interactions VALUES (?, ?)",
        [("u1", (now - timedelta(days=age)).isoformat())
         for age in (5, 45, 100, 200)]
    )
    db.commit()

    before = db.execute("SELECT count(*) FROM user_interactions").fetchone()[0]
    deleted = cleanup_old_data(db, retention_days=90)
    after = db.execute("SELECT count(*) FROM user_interactions").fetchone()[0]

    print(f"rows before:    {before}")
    print(f"rows deleted:   {deleted}")
    print(f"rows remaining: {after}")

import hmac
import os
from hashlib import sha256

import pandas as pd


def pseudonymize_dataframe(df, id_columns, sensitive_columns, secret):
    """
    Replace direct identifiers with keyed pseudonyms and coarsen numeric detail.

    Identifiers are hashed with HMAC rather than a bare digest. A plain
    sha256 of a low-entropy value like 'P001' or 'Alice' is reversible by
    hashing every candidate and comparing, which takes seconds. The secret is
    what makes the mapping one-way to anyone who does not hold it.

    Pseudonymous is not anonymous. The rows still describe individuals and
    remain personal data under GDPR; rotating or destroying the secret is what
    actually severs the link.

    Args:
        df: Original dataframe
        id_columns: Columns with direct identifiers (names, emails, SSN)
        sensitive_columns: Numeric columns to coarsen into ranges
        secret: Key held outside the dataset, from a secrets manager

    Returns:
        Pseudonymized copy, safe to share more widely than the original
    """
    df_anon = df.copy()

    for col in id_columns:
        df_anon[col] = df_anon[col].map(
            lambda value: hmac.new(
                secret, str(value).encode(), sha256
            ).hexdigest()[:16]
        )

    for col in sensitive_columns:
        # is_numeric_dtype rather than a dtype identity test: on pandas 3 a
        # text column is StringDtype, so comparing against np.int64/np.float64
        # misclassifies columns silently.
        if pd.api.types.is_numeric_dtype(df_anon[col]):
            df_anon[col] = pd.cut(
                df_anon[col],
                bins=5,
                labels=['very_low', 'low', 'medium', 'high', 'very_high']
            )

    return df_anon


if __name__ == "__main__":
    # In production this comes from a secrets manager, never from the source.
    secret = os.environ.get("PSEUDONYM_KEY", "demo-key-not-for-production").encode()

    medical_data = pd.DataFrame({
        'patient_id': ['P001', 'P002', 'P003', 'P004', 'P005'],
        'name': ['Alice', 'Bob', 'Charlie', 'Dana', 'Erin'],
        'age': [34, 45, 29, 61, 52],
        'salary': [75000, 120000, 45000, 98000, 61000],
        'diagnosis': ['diabetes', 'hypertension', 'healthy',
                      'diabetes', 'healthy'],
    })

    anonymized = pseudonymize_dataframe(
        medical_data,
        id_columns=['patient_id', 'name'],
        sensitive_columns=['age', 'salary'],
        secret=secret,
    )

    print(anonymized.to_string(index=False))

# Differential privacy: Add noise to protect individuals

import numpy as np


def add_laplace_noise(value, sensitivity, epsilon, rng):
    """
    Add Laplace noise calibrated to sensitivity and the privacy budget.

    Args:
        value: True value
        sensitivity: Maximum change one individual can cause
        epsilon: Privacy parameter (smaller = more privacy, less accuracy)
        rng: numpy Generator, seeded so results are reproducible

    Returns:
        Noisy value
    """
    scale = sensitivity / epsilon

    return value + rng.laplace(0, scale)


def private_mean(data, lower, upper, epsilon=1.0, rng=None):
    """
    Mean with a differential privacy guarantee.

    The bounds are arguments, not measurements. Deriving them with
    data.min()/data.max() would read the private values to decide how much
    noise to add, which leaks the extremes and voids the guarantee: the
    published result would depend on the very outliers it is meant to hide.
    Bounds have to come from outside the data, such as a form that already
    caps what a person can enter.

    Args:
        data: Array of values
        lower, upper: Publicly known range the values must fall in
        epsilon: Privacy budget
        rng: numpy Generator

    Returns:
        Differentially private mean
    """
    rng = np.random.default_rng() if rng is None else rng

    clipped = np.clip(data, lower, upper)
    # Swapping one person's value moves a mean of n clipped values by at
    # most (upper - lower) / n, so that is the sensitivity.
    sensitivity = (upper - lower) / len(data)

    return add_laplace_noise(clipped.mean(), sensitivity, epsilon, rng)


if __name__ == "__main__":
    rng = np.random.default_rng(42)

    ages = np.array([25, 30, 35, 40, 45, 50, 55, 60])

    # Published range for the field, not derived from these eight people.
    AGE_LOWER, AGE_UPPER = 0, 120

    print(f"True mean age: {ages.mean():.2f}\n")
    print("Five independent releases at each privacy budget:")

    for epsilon in (10.0, 1.0, 0.1):
        draws = [private_mean(ages, AGE_LOWER, AGE_UPPER, epsilon, rng)
                 for _ in range(5)]
        formatted = "  ".join(f"{value:7.2f}" for value in draws)
        print(f"  epsilon={epsilon:<5} {formatted}")

    print("\nSmaller epsilon means more privacy and a wider spread.")

import json
import sqlite3
from datetime import datetime, timezone

from cryptography.fernet import Fernet


class SecureDataStore:
    """Store sensitive data encrypted, with an access audit trail."""

    def __init__(self, db, encryption_key):
        # The key is a required argument on purpose. Generating one here when
        # the caller passes nothing produces a key that lives only in this
        # process: restart the service and every row written so far is
        # unreadable forever. Keys belong in a KMS or secrets manager and are
        # loaded at startup.
        self.db = db
        self.cipher = Fernet(encryption_key)

    def encrypt_data(self, data):
        """Encrypt data before storage."""
        return self.cipher.encrypt(json.dumps(data).encode())

    def decrypt_data(self, encrypted_data):
        """Decrypt data when needed."""
        return json.loads(self.cipher.decrypt(encrypted_data).decode())

    def store_user_data(self, user_id, data):
        """Store user data encrypted."""
        self.db.execute(
            "INSERT INTO secure_data (user_id, encrypted_data) VALUES (?, ?)",
            (user_id, self.encrypt_data(data))
        )
        self._log_access(user_id, action="store")
        self.db.commit()

    def load_user_data(self, user_id):
        """Read user data back, logging the access."""
        row = self.db.execute(
            "SELECT encrypted_data FROM secure_data WHERE user_id = ?",
            (user_id,)
        ).fetchone()
        self._log_access(user_id, action="read")
        self.db.commit()

        return self.decrypt_data(row[0])

    def _log_access(self, user_id, action):
        """Log all data access for audit trail."""
        self.db.execute(
            "INSERT INTO audit_log (user_id, action, timestamp) VALUES (?, ?, ?)",
            (user_id, action, datetime.now(timezone.utc).isoformat())
        )


# Usage
if __name__ == "__main__":
    db = sqlite3.connect(":memory:")
    db.execute("CREATE TABLE secure_data (user_id TEXT, encrypted_data BLOB)")
    db.execute("CREATE TABLE audit_log (user_id TEXT, action TEXT, timestamp TEXT)")

    # Generated inline only because this is a demo with nothing to preserve.
    store = SecureDataStore(db, Fernet.generate_key())

    user_data = {
        'medical_history': ['asthma', 'penicillin allergy'],
        'financial_info': {'account': '****4821', 'credit_limit': 5000},
    }

    store.store_user_data('user_123', user_data)

    stored = db.execute("SELECT encrypted_data FROM secure_data").fetchone()[0]
    print(f"Ciphertext on disk: {len(stored)} bytes, unreadable without the key")
    print(f"Decrypted:          {store.load_user_data('user_123')}")

    print("\nAudit log:")
    for user_id, action in db.execute("SELECT user_id, action FROM audit_log"):
        print(f"  {user_id}  {action}")

# Defense: Input validation and adversarial training

import numpy as np


def detect_adversarial_input(model, x, min_confidence=0.7, probes=10,
                             jitter=0.05, max_swing=0.15, rng=None):
    """
    Flag inputs that look engineered rather than natural.

    Two signals: a prediction the model is not confident about, and a
    prediction that moves sharply when the input is nudged.

    min_confidence has to sit above 0.5. For a binary model the winning
    probability is max(p, 1 - p), which is never below 0.5, so a threshold of
    0.5 marks nothing at all.
    """
    rng = np.random.default_rng() if rng is None else rng

    probabilities = model.predict_proba([x])[0]
    if probabilities.max() < min_confidence:
        return True, "Low confidence prediction"

    # Watch the probability under perturbation, not the predicted label.
    # Counting distinct labels cannot measure instability: a binary model
    # only ever emits two of them.
    neighbourhood = x + rng.normal(0, jitter, size=(probes, len(x)))
    probe_probabilities = model.predict_proba(neighbourhood)[:, 1]
    swing = probe_probabilities.max() - probe_probabilities.min()

    if swing > max_swing:
        return True, f"Unstable under perturbation (swing={swing:.3f})"

    return False, "Input appears legitimate"


def fgsm_perturbation(model, X, y, epsilon=0.3):
    """
    Fast gradient sign perturbation against a linear model.

    Random noise is not an adversarial example. It points in an arbitrary
    direction and the model shrugs most of it off. An adversarial example
    moves along the gradient of the loss, the one direction that costs the
    most accuracy per unit of budget. For logistic regression that gradient
    is available in closed form: d(loss)/dx = (p - y) * w.
    """
    probabilities = model.predict_proba(X)[:, 1]
    gradient = (probabilities - y)[:, None] * model.coef_

    return X + epsilon * np.sign(gradient)


def adversarial_training(model, X_train, y_train, epsilon=0.3):
    """
    Train on clean data plus adversarial examples built from it.

    The examples keep their true labels, so the boundary is pushed away from
    the points an attacker would aim for.
    """
    X_adversarial = fgsm_perturbation(model, X_train, y_train, epsilon)

    X_combined = np.vstack([X_train, X_adversarial])
    y_combined = np.concatenate([y_train, y_train])

    model.fit(X_combined, y_combined)

    return model


if __name__ == "__main__":
    from sklearn.linear_model import LogisticRegression
    from sklearn.model_selection import train_test_split

    rng = np.random.default_rng(42)
    n_samples = 4000
    epsilon = 0.3

    features = rng.normal(size=(n_samples, 6))
    weights = np.array([1.5, -1.0, 0.8, -0.6, 0.4, -0.2])
    y = ((features @ weights
          + rng.normal(scale=0.5, size=n_samples)) > 0).astype(int)

    X_train, X_test, y_train, y_test = train_test_split(
        features, y, test_size=0.5, random_state=42
    )

    model = LogisticRegression(max_iter=1000).fit(X_train, y_train)

    noisy = X_test + rng.uniform(-epsilon, epsilon, X_test.shape)
    attacked = fgsm_perturbation(model, X_test, y_test, epsilon)

    print(f"Accuracy at perturbation budget {epsilon}:")
    print(f"  clean input:    {model.score(X_test, y_test):.3f}")
    print(f"  random noise:   {model.score(noisy, y_test):.3f}")
    print(f"  FGSM attack:    {model.score(attacked, y_test):.3f}")

    hardened = adversarial_training(
        LogisticRegression(max_iter=1000).fit(X_train, y_train),
        X_train, y_train, epsilon
    )
    hardened_attack = fgsm_perturbation(hardened, X_test, y_test, epsilon)

    print("\nAfter adversarial training:")
    print(f"  clean input:    {hardened.score(X_test, y_test):.3f}")
    print(f"  FGSM attack:    {hardened.score(hardened_attack, y_test):.3f}")

    detector_rng = np.random.default_rng(0)
    clean_flagged = np.mean([
        detect_adversarial_input(model, x, rng=detector_rng)[0]
        for x in X_test[:500]
    ])
    attacked_flagged = np.mean([
        detect_adversarial_input(model, x, rng=detector_rng)[0]
        for x in attacked[:500]
    ])

    print("\nDetector flag rate over 500 inputs:")
    print(f"  clean:          {clean_flagged:.3f}")
    print(f"  attacked:       {attacked_flagged:.3f}")