Any model — PyTorch, Keras-shaped, remote endpoints¶

Nothing in bdp-model-gate imports a deep-learning framework. Instead of requiring an object of a particular shape, the gate accepts plain functions, and the boundary is deliberately narrow: DataFrame in, array out. Your function owns tensor conversion, device placement, batching and auth.

That matters because the scikit-learn conventions are not universal:

Model	.predict()	.predict_proba()	Wants a DataFrame
scikit-learn	yes	yes, (n, 2)	yes
Keras	yes	no — sigmoid gives (n, 1)	yes
PyTorch nn.Module	no — it's callable	no	no — wants a tensor
Remote endpoint	no object at all

Gradient-boosting libraries with a native API — XGBoost's Booster, which wants a DMatrix — are covered in 05_boosters_and_cli.ipynb. They are a separate notebook for a practical reason: on macOS, XGBoost and PyTorch link different OpenMP runtimes and segfault when used in the same process, so combining them would ship a notebook that crashes for many readers.

Three optional context fields cover all of it:

Field	Signature	Unlocks
predict_fn	fn(DataFrame) -> array	everything; takes precedence over model
predict_proba_fn	fn(DataFrame) -> array	CounterfactualFlipCheck
gradient_fn	fn(DataFrame) -> (n_rows, n_features)	a real targeted adversarial attack

New to the library? 01_binary_classification_sklearn.ipynb covers contexts, reports and verdicts. This notebook assumes those.

Extra dependency: torch. TensorFlow is not required — the Keras case is shown with a numpy model emitting Keras's output shape, which is the part the adapter actually has to handle.

In [1]:

# %pip install -q "bdp-model-gate[structured]" torch
import logging

import numpy as np
import pandas as pd
import torch

import bdp_model_gate

logging.basicConfig(level=logging.INFO, format="%(levelname)-8s %(name)s: %(message)s")
logging.getLogger("bdp_model_gate").setLevel(logging.WARNING)
torch.manual_seed(0)
pd.set_option("display.width", 130)

print("bdp-model-gate", bdp_model_gate.__version__)
print("torch", torch.__version__)

# %pip install -q "bdp-model-gate[structured]" torch import logging import numpy as np import pandas as pd import torch import bdp_model_gate logging.basicConfig(level=logging.INFO, format="%(levelname)-8s %(name)s: %(message)s") logging.getLogger("bdp_model_gate").setLevel(logging.WARNING) torch.manual_seed(0) pd.set_option("display.width", 130) print("bdp-model-gate", bdp_model_gate.__version__) print("torch", torch.__version__)

bdp-model-gate 0.4.1
torch 2.13.0

1. A fraud-detection problem¶

Binary, heavily imbalanced — the usual shape for fraud, and a case where average_precision says more than roc_auc.

In [2]:

from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler

rng = np.random.default_rng(11)
N = 4000

X = pd.DataFrame(
    {
        "amount_ngn": rng.lognormal(10.5, 1.1, N).round(2),
        "hour_of_day": rng.integers(0, 24, N).astype(float),
        "days_since_signup": rng.exponential(180, N).round(1),
        "device_changes_30d": rng.poisson(0.4, N).astype(float),
        "prior_disputes": rng.poisson(0.15, N).astype(float),
    }
)
channel = rng.choice(["app", "ussd", "web"], N, p=[0.55, 0.30, 0.15])

risk = (
    -3.6
    + 0.55 * np.log1p(X["amount_ngn"] / 20_000)
    + 0.9 * X["device_changes_30d"]
    + 1.3 * X["prior_disputes"]
    - 0.004 * X["days_since_signup"]
    + 0.6 * (channel == "ussd")
)
is_fraud = rng.binomial(1, 1 / (1 + np.exp(-risk)))
protected_df = pd.DataFrame({"channel": channel})

X_train, X_val, y_train, y_val, prot_train, prot_val = train_test_split(
    X, is_fraud, protected_df, test_size=0.3, random_state=11, stratify=is_fraud
)

# Networks want scaled inputs; the scaler lives inside our predict_fn later.
scaler = StandardScaler().fit(X_train)

print(f"train {len(X_train)} | validation {len(X_val)} | fraud rate {is_fraud.mean():.2%}")

from sklearn.model_selection import train_test_split from sklearn.preprocessing import StandardScaler rng = np.random.default_rng(11) N = 4000 X = pd.DataFrame( { "amount_ngn": rng.lognormal(10.5, 1.1, N).round(2), "hour_of_day": rng.integers(0, 24, N).astype(float), "days_since_signup": rng.exponential(180, N).round(1), "device_changes_30d": rng.poisson(0.4, N).astype(float), "prior_disputes": rng.poisson(0.15, N).astype(float), } ) channel = rng.choice(["app", "ussd", "web"], N, p=[0.55, 0.30, 0.15]) risk = ( -3.6 + 0.55 * np.log1p(X["amount_ngn"] / 20_000) + 0.9 * X["device_changes_30d"] + 1.3 * X["prior_disputes"] - 0.004 * X["days_since_signup"] + 0.6 * (channel == "ussd") ) is_fraud = rng.binomial(1, 1 / (1 + np.exp(-risk))) protected_df = pd.DataFrame({"channel": channel}) X_train, X_val, y_train, y_val, prot_train, prot_val = train_test_split( X, is_fraud, protected_df, test_size=0.3, random_state=11, stratify=is_fraud ) # Networks want scaled inputs; the scaler lives inside our predict_fn later. scaler = StandardScaler().fit(X_train) print(f"train {len(X_train)} | validation {len(X_val)} | fraud rate {is_fraud.mean():.2%}")

train 2800 | validation 1200 | fraud rate 7.58%

2. PyTorch — a model with no .predict() at all¶

An nn.Module is callable, wants a tensor, and returns a tensor. None of that matches the scikit-learn contract, and before 0.3.1 it simply could not be gated.

In [3]:

import torch.nn as nn

net = nn.Sequential(
    nn.Linear(X.shape[1], 24), nn.ReLU(),
    nn.Linear(24, 12), nn.ReLU(),
    nn.Linear(12, 1),
)

Xtr_t = torch.tensor(scaler.transform(X_train), dtype=torch.float32)
ytr_t = torch.tensor(y_train, dtype=torch.float32).unsqueeze(1)

optimiser = torch.optim.Adam(net.parameters(), lr=0.01)
loss_fn = nn.BCEWithLogitsLoss(pos_weight=torch.tensor([(1 - y_train.mean()) / y_train.mean()]))

net.train()
for _epoch in range(220):
    optimiser.zero_grad()
    loss = loss_fn(net(Xtr_t), ytr_t)
    loss.backward()
    optimiser.step()
net.eval()

print(f"final training loss {loss.item():.4f}")

import torch.nn as nn net = nn.Sequential( nn.Linear(X.shape[1], 24), nn.ReLU(), nn.Linear(24, 12), nn.ReLU(), nn.Linear(12, 1), ) Xtr_t = torch.tensor(scaler.transform(X_train), dtype=torch.float32) ytr_t = torch.tensor(y_train, dtype=torch.float32).unsqueeze(1) optimiser = torch.optim.Adam(net.parameters(), lr=0.01) loss_fn = nn.BCEWithLogitsLoss(pos_weight=torch.tensor([(1 - y_train.mean()) / y_train.mean()])) net.train() for _epoch in range(220): optimiser.zero_grad() loss = loss_fn(net(Xtr_t), ytr_t) loss.backward() optimiser.step() net.eval() print(f"final training loss {loss.item():.4f}")

final training loss 0.8965

The three functions¶

Each is a small adapter you write once. Note they take a DataFrame and return a numpy array — the library never sees a tensor.

In [4]:

def torch_proba(df: pd.DataFrame) -> np.ndarray:
    """P(fraud). Keras-style (n, 1) output — the adapter flattens it."""
    with torch.no_grad():
        logits = net(torch.tensor(scaler.transform(df), dtype=torch.float32))
        return torch.sigmoid(logits).numpy()


def torch_predict(df: pd.DataFrame) -> np.ndarray:
    return (torch_proba(df).ravel() >= 0.5).astype(int)


def torch_gradients(df: pd.DataFrame) -> np.ndarray:
    """d(output)/d(input), one row per sample — a real attack direction."""
    tensor = torch.tensor(scaler.transform(df), dtype=torch.float32, requires_grad=True)
    torch.sigmoid(net(tensor)).sum().backward()
    # Chain back through the scaler so gradients are in the units of X.
    return tensor.grad.numpy() / scaler.scale_


y_pred_torch = torch_proba(X_val).ravel()
print("proba shape from the network:", torch_proba(X_val).shape, "-> flattened by the adapter")
print("gradient shape:", torch_gradients(X_val.head()).shape, "(rows, features)")

def torch_proba(df: pd.DataFrame) -> np.ndarray: """P(fraud). Keras-style (n, 1) output — the adapter flattens it.""" with torch.no_grad(): logits = net(torch.tensor(scaler.transform(df), dtype=torch.float32)) return torch.sigmoid(logits).numpy() def torch_predict(df: pd.DataFrame) -> np.ndarray: return (torch_proba(df).ravel() >= 0.5).astype(int) def torch_gradients(df: pd.DataFrame) -> np.ndarray: """d(output)/d(input), one row per sample — a real attack direction.""" tensor = torch.tensor(scaler.transform(df), dtype=torch.float32, requires_grad=True) torch.sigmoid(net(tensor)).sum().backward() # Chain back through the scaler so gradients are in the units of X. return tensor.grad.numpy() / scaler.scale_ y_pred_torch = torch_proba(X_val).ravel() print("proba shape from the network:", torch_proba(X_val).shape, "-> flattened by the adapter") print("gradient shape:", torch_gradients(X_val.head()).shape, "(rows, features)")

proba shape from the network: (1200, 1) -> flattened by the adapter
gradient shape: (5, 5) (rows, features)

In [5]:

from bdp_model_gate import GateConfig, ModelGate, StructuredGateContext
from bdp_model_gate.structured import default_structured_checks

model_card = {
    "model_name": "fraud-net",
    "version": "1.0.0",
    "use_case": "claims_decisioning",
    "legal_basis": "Legitimate interest — fraud prevention (NDPA 2023, s.25(1)(f))",
    "data_minimization_justification": "Transaction and device signals only.",
    "training_data_source": "Payments ledger 2023-2025",
    "dpia_completed": True,
    "influences_decision_about_person": True,
    "explainability_method": "Input gradients retained per decision",
}

torch_context = StructuredGateContext(
    # No `model=` at all — the functions are the whole interface.
    X=X_val,
    y_true=y_val,
    y_pred=y_pred_torch,
    protected_df=prot_val,
    model_card=model_card,
    predict_fn=torch_predict,
    predict_proba_fn=torch_proba,
    gradient_fn=torch_gradients,
    task="binary",
)

config = GateConfig()
config.performance.metric = "average_precision"   # better than AUC when 5% positive
config.performance.min_score = 0.20

report = ModelGate(checks=default_structured_checks(config, include_plugins=False)).run(
    torch_context
)
print(report.summary())
print("\ncontext.model is:", torch_context.model)

from bdp_model_gate import GateConfig, ModelGate, StructuredGateContext from bdp_model_gate.structured import default_structured_checks model_card = { "model_name": "fraud-net", "version": "1.0.0", "use_case": "claims_decisioning", "legal_basis": "Legitimate interest — fraud prevention (NDPA 2023, s.25(1)(f))", "data_minimization_justification": "Transaction and device signals only.", "training_data_source": "Payments ledger 2023-2025", "dpia_completed": True, "influences_decision_about_person": True, "explainability_method": "Input gradients retained per decision", } torch_context = StructuredGateContext( # No `model=` at all — the functions are the whole interface. X=X_val, y_true=y_val, y_pred=y_pred_torch, protected_df=prot_val, model_card=model_card, predict_fn=torch_predict, predict_proba_fn=torch_proba, gradient_fn=torch_gradients, task="binary", ) config = GateConfig() config.performance.metric = "average_precision" # better than AUC when 5% positive config.performance.min_score = 0.20 report = ModelGate(checks=default_structured_checks(config, include_plugins=False)).run( torch_context ) print(report.summary()) print("\ncontext.model is:", torch_context.model)

Gate status: PASS (4857ms, binary)
  average_precision: 0.2297
  performance: 0 flag(s)
  compliance: 0 flag(s)
  security: 0 flag(s)
  fairness: 0 flag(s)

context.model is: None

What gradient_fn buys you¶

AdversarialRobustnessCheck picks its attack direction in this order:

1. true per-row gradients — if gradient_fn is supplied
2. linear coefficients — for models exposing coef_
3. isotropic random noise — otherwise

For a neural network only the first is meaningful. The step is sign-of-gradient at full epsilon — an FGSM-style attack — tried in both directions, so it can flip rows on either side of the boundary.

In [6]:

from bdp_model_gate.structured.security import AdversarialRobustnessCheck

from bdp_model_gate import SecurityConfig

no_grad_context = StructuredGateContext(
    X=X_val, y_true=y_val, y_pred=y_pred_torch, predict_fn=torch_predict, task="binary",
)

rows = []
for epsilon in (0.02, 0.05, 0.10):
    cfg = SecurityConfig(adversarial_epsilon=epsilon)
    rows.append(
        {
            "epsilon": epsilon,
            "gradient-directed": AdversarialRobustnessCheck(cfg)
            .run(torch_context)[0].metadata["flip_rate"],
            "random noise": AdversarialRobustnessCheck(cfg)
            .run(no_grad_context)[0].metadata["flip_rate"],
        }
    )
display(pd.DataFrame(rows).set_index("epsilon"))

from bdp_model_gate.structured.security import AdversarialRobustnessCheck from bdp_model_gate import SecurityConfig no_grad_context = StructuredGateContext( X=X_val, y_true=y_val, y_pred=y_pred_torch, predict_fn=torch_predict, task="binary", ) rows = [] for epsilon in (0.02, 0.05, 0.10): cfg = SecurityConfig(adversarial_epsilon=epsilon) rows.append( { "epsilon": epsilon, "gradient-directed": AdversarialRobustnessCheck(cfg) .run(torch_context)[0].metadata["flip_rate"], "random noise": AdversarialRobustnessCheck(cfg) .run(no_grad_context)[0].metadata["flip_rate"], } ) display(pd.DataFrame(rows).set_index("epsilon"))

	gradient-directed	random noise
epsilon
0.02	0.045	0.000
0.05	0.125	0.055
0.10	0.215	0.110

At the default epsilon of 0.02, random noise finds nothing while the gradient attack already finds flips. Perturbing features at random mostly moves samples parallel to the decision boundary; the gradient points straight at it.

That matters for how you read a ROBUSTNESS_RISK flag. A clean result from the random path means "we did not find a weakness with a scattergun", which is much weaker evidence than a clean result from a gradient attack. If your model is differentiable, supplying gradient_fn is what makes the check worth trusting.

And predict_proba_fn¶

CounterfactualFlipCheck needs probabilities. It used to require a model with a .predict_proba() method, so it was NOT_APPLICABLE for every Keras and PyTorch model. Supplying the function is enough now.

In [7]:

from bdp_model_gate.structured.fairness import CounterfactualFlipCheck

X_val_channel = X_val.assign(is_ussd=(prot_val["channel"] == "ussd").astype(float))


def proba_with_channel(df):
    base = df.drop(columns=["is_ussd"])
    return torch_proba(base).ravel() * (1 + 0.15 * df["is_ussd"].to_numpy())


cf_context = StructuredGateContext(
    X=X_val_channel, y_true=y_val, y_pred=y_pred_torch,
    protected_df=pd.DataFrame({"is_ussd": X_val_channel["is_ussd"].to_numpy()}),
    predict_fn=lambda df: (proba_with_channel(df) >= 0.5).astype(int),
    predict_proba_fn=proba_with_channel,
    task="binary",
)
for r in CounterfactualFlipCheck().run(cf_context):
    print(f"{r.flag:22} {r.detail}")

from bdp_model_gate.structured.fairness import CounterfactualFlipCheck X_val_channel = X_val.assign(is_ussd=(prot_val["channel"] == "ussd").astype(float)) def proba_with_channel(df): base = df.drop(columns=["is_ussd"]) return torch_proba(base).ravel() * (1 + 0.15 * df["is_ussd"].to_numpy()) cf_context = StructuredGateContext( X=X_val_channel, y_true=y_val, y_pred=y_pred_torch, protected_df=pd.DataFrame({"is_ussd": X_val_channel["is_ussd"].to_numpy()}), predict_fn=lambda df: (proba_with_channel(df) >= 0.5).astype(int), predict_proba_fn=proba_with_channel, task="binary", ) for r in CounterfactualFlipCheck().run(cf_context): print(f"{r.flag:22} {r.detail}")

OK                     flipping is_ussd to np.float64(1.0) shifts predictions by 0.0373 on average
OK                     flipping is_ussd to np.float64(0.0) shifts predictions by 0.0198 on average

3. Keras-shaped output¶

A Keras binary classifier ends in a single sigmoid unit, so predict() returns (n, 1) — not scikit-learn's (n, 2). Both mean the same thing, and the adapter reduces either to one positive-class vector, so you never have to know which the library expects.

Below is a numpy model emitting that shape. (Real TensorFlow would behave identically here; the shape is the part that matters, and skipping the dependency keeps this notebook installable everywhere.)

In [8]:

from bdp_model_gate.model import ModelAdapter


class KerasShaped:
    """Emits (n, 1) from a sigmoid, exactly as a Keras Sequential would."""

    def __init__(self, weights, bias):
        self.weights, self.bias = np.asarray(weights, float), float(bias)

    def predict(self, df, verbose=0):
        z = scaler.transform(df) @ self.weights + self.bias
        return (1 / (1 + np.exp(-z))).reshape(-1, 1)


keras_like = KerasShaped(rng.normal(0, 0.6, X.shape[1]), -2.4)

def _flat(df):
    return torch_proba(df).ravel()


shapes = {
    "scikit-learn (n, 2)": lambda df: np.column_stack([1 - _flat(df), _flat(df)]),
    "Keras (n, 1)": lambda df: keras_like.predict(df),
    "bare (n,)": _flat,
}
for label, fn in shapes.items():
    out = np.asarray(fn(X_val.head(4)))
    reduced = ModelAdapter(predict_proba_fn=fn).predict_positive_proba(X_val.head(4))
    print(f"{label:22} raw {str(out.shape):8} -> {reduced.round(3)}")

from bdp_model_gate.model import ModelAdapter class KerasShaped: """Emits (n, 1) from a sigmoid, exactly as a Keras Sequential would.""" def __init__(self, weights, bias): self.weights, self.bias = np.asarray(weights, float), float(bias) def predict(self, df, verbose=0): z = scaler.transform(df) @ self.weights + self.bias return (1 / (1 + np.exp(-z))).reshape(-1, 1) keras_like = KerasShaped(rng.normal(0, 0.6, X.shape[1]), -2.4) def _flat(df): return torch_proba(df).ravel() shapes = { "scikit-learn (n, 2)": lambda df: np.column_stack([1 - _flat(df), _flat(df)]), "Keras (n, 1)": lambda df: keras_like.predict(df), "bare (n,)": _flat, } for label, fn in shapes.items(): out = np.asarray(fn(X_val.head(4))) reduced = ModelAdapter(predict_proba_fn=fn).predict_positive_proba(X_val.head(4)) print(f"{label:22} raw {str(out.shape):8} -> {reduced.round(3)}")

scikit-learn (n, 2)    raw (4, 2)   -> [0.526 0.197 0.076 0.488]
Keras (n, 1)           raw (4, 1)   -> [0.041 0.107 0.221 0.102]
bare (n,)              raw (4,)     -> [0.526 0.197 0.076 0.488]

A genuinely multiclass (n, k) output is refused rather than silently sliced — taking column 1 of a three-class model would yield a real number that means nothing.

In [9]:

from bdp_model_gate.exceptions import GateConfigurationError

try:
    ModelAdapter(
        predict_proba_fn=lambda df: np.tile([0.2, 0.3, 0.5], (len(df), 1))
    ).predict_positive_proba(X_val.head())
except GateConfigurationError as exc:
    print("GateConfigurationError:", exc)

from bdp_model_gate.exceptions import GateConfigurationError try: ModelAdapter( predict_proba_fn=lambda df: np.tile([0.2, 0.3, 0.5], (len(df), 1)) ).predict_positive_proba(X_val.head()) except GateConfigurationError as exc: print("GateConfigurationError:", exc)

GateConfigurationError: predict_proba returned 3 columns, so this is not a binary classifier — there is no single positive class to take

4. A remote scoring endpoint¶

There is no model object to pass at all — just a function that happens to make a network call. Batching, retries and auth live in your code, where they belong.

In [10]:

def score_remotely(df: pd.DataFrame) -> np.ndarray:
    """Stands in for an HTTP call to a model-serving endpoint.

    A real one would batch, retry and authenticate here; the gate neither
    knows nor cares.
    """
    payload = df.to_dict(orient="records")           # what you would POST
    # response = requests.post(URL, json=payload, timeout=30).json()["scores"]
    response = torch_proba(pd.DataFrame(payload)).ravel()   # simulated response
    return np.asarray(response)


remote_context = StructuredGateContext(
    X=X_val, y_true=y_val, y_pred=score_remotely(X_val),
    protected_df=prot_val, model_card=model_card,
    predict_fn=lambda df: (score_remotely(df) >= 0.5).astype(int),
    predict_proba_fn=score_remotely,
    task="binary",
)
remote_report = ModelGate(
    checks=default_structured_checks(config, include_plugins=False)
).run(remote_context)
print(remote_report.summary())
print("\nmodel object:", remote_context.model)

def score_remotely(df: pd.DataFrame) -> np.ndarray: """Stands in for an HTTP call to a model-serving endpoint. A real one would batch, retry and authenticate here; the gate neither knows nor cares. """ payload = df.to_dict(orient="records") # what you would POST # response = requests.post(URL, json=payload, timeout=30).json()["scores"] response = torch_proba(pd.DataFrame(payload)).ravel() # simulated response return np.asarray(response) remote_context = StructuredGateContext( X=X_val, y_true=y_val, y_pred=score_remotely(X_val), protected_df=prot_val, model_card=model_card, predict_fn=lambda df: (score_remotely(df) >= 0.5).astype(int), predict_proba_fn=score_remotely, task="binary", ) remote_report = ModelGate( checks=default_structured_checks(config, include_plugins=False) ).run(remote_context) print(remote_report.summary()) print("\nmodel object:", remote_context.model)

Gate status: PASS (3835ms, binary)
  average_precision: 0.2297
  performance: 0 flag(s)
  compliance: 0 flag(s)
  security: 0 flag(s)
  fairness: 0 flag(s)

model object: None

One caveat worth planning for: several checks re-score the model — AdversarialRobustnessCheck and CounterfactualFlipCheck each call it hundreds of times. Against a rate-limited or paid endpoint, either budget for that or drop those checks from the suite.

In [11]:

from bdp_model_gate.structured.compliance import ComplianceMappingCheck
from bdp_model_gate.structured.performance import PerformanceThresholdCheck
from bdp_model_gate.structured.security import PIILeakageCheck

# A suite that scores the model exactly once.
cheap = ModelGate(
    checks=[
        PerformanceThresholdCheck(config.performance),
        ComplianceMappingCheck(config.compliance),
        PIILeakageCheck(config.security),
    ]
).run(remote_context)
print(cheap.summary())

from bdp_model_gate.structured.compliance import ComplianceMappingCheck from bdp_model_gate.structured.performance import PerformanceThresholdCheck from bdp_model_gate.structured.security import PIILeakageCheck # A suite that scores the model exactly once. cheap = ModelGate( checks=[ PerformanceThresholdCheck(config.performance), ComplianceMappingCheck(config.compliance), PIILeakageCheck(config.security), ] ).run(remote_context) print(cheap.summary())

Gate status: PASS (1ms, binary)
  average_precision: 0.2297
  performance: 0 flag(s)
  compliance: 0 flag(s)
  security: 0 flag(s)
  fairness: 0 flag(s)

5. Three models, one gate¶

In [12]:

rows = []
for label, ctx in (
    ("PyTorch nn.Module", torch_context),
    ("Keras-shaped (untrained)", StructuredGateContext(
        X=X_val, y_true=y_val, y_pred=keras_like.predict(X_val).ravel(),
        protected_df=prot_val, model_card=model_card,
        predict_fn=lambda df: (keras_like.predict(df).ravel() >= 0.5).astype(int),
        predict_proba_fn=keras_like.predict, task="binary",
    )),
    ("Remote endpoint", remote_context),
):
    rep = ModelGate(checks=default_structured_checks(config, include_plugins=False)).run(ctx)
    rows.append(
        {
            "model": label,
            "verdict": rep.gate_status,
            "metric": rep.model_metric,
            "score": round(rep.model_score, 4),
            "flags": len(rep.flags),
            "check_errors": sum(r.flag == "CHECK_ERROR" for r in rep.results),
        }
    )
display(pd.DataFrame(rows).set_index("model"))

rows = [] for label, ctx in ( ("PyTorch nn.Module", torch_context), ("Keras-shaped (untrained)", StructuredGateContext( X=X_val, y_true=y_val, y_pred=keras_like.predict(X_val).ravel(), protected_df=prot_val, model_card=model_card, predict_fn=lambda df: (keras_like.predict(df).ravel() >= 0.5).astype(int), predict_proba_fn=keras_like.predict, task="binary", )), ("Remote endpoint", remote_context), ): rep = ModelGate(checks=default_structured_checks(config, include_plugins=False)).run(ctx) rows.append( { "model": label, "verdict": rep.gate_status, "metric": rep.model_metric, "score": round(rep.model_score, 4), "flags": len(rep.flags), "check_errors": sum(r.flag == "CHECK_ERROR" for r in rep.results), } ) display(pd.DataFrame(rows).set_index("model"))

	verdict	metric	score	flags	check_errors
model
PyTorch nn.Module	PASS	average_precision	0.2297	0	0
Keras-shaped (untrained)	BLOCKED	average_precision	0.0579	1	0
Remote endpoint	PASS	average_precision	0.2297	0	0

6. From the CLI — --model-loader¶

joblib only reads pickles, which rules out .pt checkpoints, Keras SavedModel directories, ONNX graphs and endpoints. --model-loader names a package.module:factory function that returns a model or a scoring callable. Your loader does the framework import, so this package needs no deep-learning dependency.

In [13]:

import subprocess
import sys
from pathlib import Path

workdir = Path("cli_framework")
workdir.mkdir(exist_ok=True)

# Save the network the way you actually would.
torch.save(net.state_dict(), workdir / "fraud_net.pt")
np.save(workdir / "scaler_mean.npy", scaler.mean_)
np.save(workdir / "scaler_scale.npy", scaler.scale_)

# A loader module, shipped alongside your training code.
(workdir / "serving.py").write_text(
    '''
import numpy as np
import torch
import torch.nn as nn


def load_scorer():
    """Returns fn(DataFrame) -> array. The framework import lives here,
    not in bdp-model-gate."""
    net = nn.Sequential(
        nn.Linear(5, 24), nn.ReLU(), nn.Linear(24, 12), nn.ReLU(), nn.Linear(12, 1)
    )
    net.load_state_dict(torch.load("fraud_net.pt"))
    net.eval()
    mean = np.load("scaler_mean.npy")
    scale = np.load("scaler_scale.npy")

    def score(df):
        with torch.no_grad():
            scaled = (df.to_numpy(dtype=float) - mean) / scale
            logits = net(torch.tensor(scaled, dtype=torch.float32))
            return torch.sigmoid(logits).numpy().ravel()

    return score
'''.lstrip()
)

frame = X_val.copy()
frame["is_fraud"] = y_val
frame.to_csv(workdir / "validation.csv", index=False)
prot_val.to_csv(workdir / "protected.csv", index=False)

proc = subprocess.run(
    [
        sys.executable, "-m", "bdp_model_gate.cli",
        "--model-loader", "serving:load_scorer",
        "--data", "validation.csv",
        "--target-col", "is_fraud",
        "--protected", "protected.csv",
        "--task", "binary",
        "--metric", "average_precision",
        "--min-score", "0.20",
        "--output", "gate_report.json",
    ],
    capture_output=True, text=True, cwd=workdir,
)
print(proc.stdout or proc.stderr)
print("exit", proc.returncode, "->",
      {0: "PASS", 1: "BLOCKED", 2: "NEEDS_REVIEW"}.get(proc.returncode))

import subprocess import sys from pathlib import Path workdir = Path("cli_framework") workdir.mkdir(exist_ok=True) # Save the network the way you actually would. torch.save(net.state_dict(), workdir / "fraud_net.pt") np.save(workdir / "scaler_mean.npy", scaler.mean_) np.save(workdir / "scaler_scale.npy", scaler.scale_) # A loader module, shipped alongside your training code. (workdir / "serving.py").write_text( ''' import numpy as np import torch import torch.nn as nn def load_scorer(): """Returns fn(DataFrame) -> array. The framework import lives here, not in bdp-model-gate.""" net = nn.Sequential( nn.Linear(5, 24), nn.ReLU(), nn.Linear(24, 12), nn.ReLU(), nn.Linear(12, 1) ) net.load_state_dict(torch.load("fraud_net.pt")) net.eval() mean = np.load("scaler_mean.npy") scale = np.load("scaler_scale.npy") def score(df): with torch.no_grad(): scaled = (df.to_numpy(dtype=float) - mean) / scale logits = net(torch.tensor(scaled, dtype=torch.float32)) return torch.sigmoid(logits).numpy().ravel() return score '''.lstrip() ) frame = X_val.copy() frame["is_fraud"] = y_val frame.to_csv(workdir / "validation.csv", index=False) prot_val.to_csv(workdir / "protected.csv", index=False) proc = subprocess.run( [ sys.executable, "-m", "bdp_model_gate.cli", "--model-loader", "serving:load_scorer", "--data", "validation.csv", "--target-col", "is_fraud", "--protected", "protected.csv", "--task", "binary", "--metric", "average_precision", "--min-score", "0.20", "--output", "gate_report.json", ], capture_output=True, text=True, cwd=workdir, ) print(proc.stdout or proc.stderr) print("exit", proc.returncode, "->", {0: "PASS", 1: "BLOCKED", 2: "NEEDS_REVIEW"}.get(proc.returncode))

Gate status: BLOCKED (5007ms, binary)
  average_precision: 0.2297
  performance: 0 flag(s)
  compliance: 0 flag(s)
  security: 1 flag(s)
  fairness: 0 flag(s)
Full report written to gate_report.json

exit 1 -> BLOCKED

In [14]:

import shutil

shutil.rmtree(workdir, ignore_errors=True)
print("cleaned up")

import shutil shutil.rmtree(workdir, ignore_errors=True) print("cleaned up")

cleaned up

Summary¶

Your model	What to pass
scikit-learn / LightGBM / XGBoost sklearn API	model=estimator — unchanged
Keras	model=keras_model (it has .predict()); add predict_proba_fn for the (n, 1) output
PyTorch nn.Module	predict_fn, plus predict_proba_fn and gradient_fn
XGBoost Booster	see 05
Remote endpoint	predict_fn only — no model at all
From CI	--model-loader "package.module:factory"

Three things worth carrying away:

1. The boundary is DataFrame in, array out. Everything framework-specific lives in your function, which is why this package depends on no ML framework and never guesses at a dtype or a device.
2. Probability shapes are normalised. (n, 2), (n, 1) and (n,) all reduce to one positive-class vector; a real multiclass (n, k) is refused rather than sliced.
3. gradient_fn is worth supplying for anything differentiable. It turns the robustness check from random noise into a genuine targeted attack.

Next: 05 boosters and the CLI loader · back to 01 binary · 02 multiclass · 03 regression