Gradient boosters and the CLI loader¶

Gradient-boosting libraries usually ship two APIs: a scikit-learn-compatible wrapper, and a native one. The wrapper needs no adapter at all. The native one is the interesting case — xgboost.Booster.predict() has the right method name but wants a DMatrix, not a DataFrame.

This notebook covers:

1. XGBClassifier — the sklearn API, model= and nothing else
2. Booster — the native API, via predict_fn
3. --model-loader, for gating a checkpoint joblib cannot unpickle

Why this is separate from 04: on macOS, XGBoost and PyTorch link different OpenMP runtimes and segfault when used in the same process — not an exception, a hard crash. Combining them would ship a notebook that dies for many readers.

Extra dependency: xgboost. On macOS it also needs brew install libomp, which XGBoost requires on that platform regardless of this library.

In [1]:

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

import numpy as np
import pandas as pd
import xgboost as xgb

import bdp_model_gate

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

print("bdp-model-gate", bdp_model_gate.__version__, "| xgboost", xgb.__version__)

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

bdp-model-gate 0.4.0 | xgboost 3.4.1

1. A fraud problem¶

Binary and heavily imbalanced, so average_precision is the metric to gate on — roc_auc looks flattering at a 5% base rate.

In [2]:

from sklearn.model_selection import train_test_split

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)))

X_train, X_val, y_train, y_val, prot_train, prot_val = train_test_split(
    X, is_fraud, pd.DataFrame({"channel": channel}),
    test_size=0.3, random_state=11, stratify=is_fraud,
)
print(f"train {len(X_train)} | validation {len(X_val)} | fraud rate {is_fraud.mean():.2%}")

from sklearn.model_selection import train_test_split 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))) X_train, X_val, y_train, y_val, prot_train, prot_val = train_test_split( X, is_fraud, pd.DataFrame({"channel": channel}), test_size=0.3, random_state=11, stratify=is_fraud, ) 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. XGBClassifier — the sklearn API needs no adapter¶

It exposes .predict() and .predict_proba() returning (n, 2), exactly like a scikit-learn estimator, so it goes straight into model=.

In [3]:

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

model_card = {
    "model_name": "fraud-xgb",
    "version": "1.2.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": "SHAP TreeExplainer",
}

sk_model = xgb.XGBClassifier(
    n_estimators=120, max_depth=4, learning_rate=0.2,
    scale_pos_weight=(1 - y_train.mean()) / y_train.mean(),
    eval_metric="logloss",
).fit(X_train, y_train)

config = GateConfig()
config.performance.metric = "average_precision"
config.performance.min_score = 0.15

sk_context = StructuredGateContext(
    model=sk_model,                       # no adapter needed
    X=X_val, y_true=y_val, y_pred=sk_model.predict_proba(X_val)[:, 1],
    protected_df=prot_val, model_card=model_card, task="binary",
)
sk_report = ModelGate(checks=default_structured_checks(config, include_plugins=False)).run(
    sk_context
)
print(sk_report.summary())

from bdp_model_gate import GateConfig, ModelGate, StructuredGateContext from bdp_model_gate.structured import default_structured_checks model_card = { "model_name": "fraud-xgb", "version": "1.2.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": "SHAP TreeExplainer", } sk_model = xgb.XGBClassifier( n_estimators=120, max_depth=4, learning_rate=0.2, scale_pos_weight=(1 - y_train.mean()) / y_train.mean(), eval_metric="logloss", ).fit(X_train, y_train) config = GateConfig() config.performance.metric = "average_precision" config.performance.min_score = 0.15 sk_context = StructuredGateContext( model=sk_model, # no adapter needed X=X_val, y_true=y_val, y_pred=sk_model.predict_proba(X_val)[:, 1], protected_df=prot_val, model_card=model_card, task="binary", ) sk_report = ModelGate(checks=default_structured_checks(config, include_plugins=False)).run( sk_context ) print(sk_report.summary())

Gate status: BLOCKED (5015ms, binary)
  average_precision: 0.1704
  performance: 0 flag(s)
  compliance: 0 flag(s)
  security: 1 flag(s)
  fairness: 0 flag(s)

Note the coef_ fallback does not apply here — a tree ensemble has no linear coefficients, so AdversarialRobustnessCheck uses random perturbation. There is no gradient to supply either, so for tree models the random path is genuinely the best available. That is worth knowing when you read the flag: for a booster it is a weaker signal than for a network with gradient_fn.

In [4]:

adv = [r for r in sk_report.results if r.check_name == "adversarial_robustness"][0]
print("attack direction used:", adv.metadata["method"])
print(adv.detail)

adv = [r for r in sk_report.results if r.check_name == "adversarial_robustness"][0] print("attack direction used:", adv.metadata["method"]) print(adv.detail)

attack direction used: random
flip rate under random perturbation=0.0950 (max 0.05)

3. Booster — the native API wants a DMatrix¶

The second failure mode: right method name, wrong argument type.

In [5]:

booster = xgb.train(
    {
        "objective": "binary:logistic",
        "max_depth": 4,
        "eta": 0.2,
        "verbosity": 0,
        "scale_pos_weight": float((1 - y_train.mean()) / y_train.mean()),
    },
    xgb.DMatrix(X_train, label=y_train),
    num_boost_round=120,
)

# What happens without an adapter:
try:
    booster.predict(X_val)
except Exception as exc:
    print(f"{type(exc).__name__}: {str(exc)[:120]}")

booster = xgb.train( { "objective": "binary:logistic", "max_depth": 4, "eta": 0.2, "verbosity": 0, "scale_pos_weight": float((1 - y_train.mean()) / y_train.mean()), }, xgb.DMatrix(X_train, label=y_train), num_boost_round=120, ) # What happens without an adapter: try: booster.predict(X_val) except Exception as exc: print(f"{type(exc).__name__}: {str(exc)[:120]}")

TypeError: ('Expecting data to be a DMatrix object, got: ', <class 'pandas.core.frame.DataFrame'>)

predict_fn is where the DMatrix wrapping belongs. Note the output is a bare (n,) vector of positive-class probabilities — neither scikit-learn's (n, 2) nor Keras's (n, 1) — and the adapter accepts all three.

In [6]:

def booster_proba(df: pd.DataFrame) -> np.ndarray:
    return booster.predict(xgb.DMatrix(df))


print("raw output shape:", booster_proba(X_val.head()).shape)

booster_context = StructuredGateContext(
    # No `model=`: predict_fn and predict_proba_fn are the whole interface.
    X=X_val, y_true=y_val, y_pred=booster_proba(X_val),
    protected_df=prot_val, model_card=model_card,
    predict_fn=lambda df: (booster_proba(df) >= 0.5).astype(int),
    predict_proba_fn=booster_proba,
    task="binary",
)
booster_report = ModelGate(
    checks=default_structured_checks(config, include_plugins=False)
).run(booster_context)
print(booster_report.summary())
print("\ncontext.model is:", booster_context.model)

def booster_proba(df: pd.DataFrame) -> np.ndarray: return booster.predict(xgb.DMatrix(df)) print("raw output shape:", booster_proba(X_val.head()).shape) booster_context = StructuredGateContext( # No `model=`: predict_fn and predict_proba_fn are the whole interface. X=X_val, y_true=y_val, y_pred=booster_proba(X_val), protected_df=prot_val, model_card=model_card, predict_fn=lambda df: (booster_proba(df) >= 0.5).astype(int), predict_proba_fn=booster_proba, task="binary", ) booster_report = ModelGate( checks=default_structured_checks(config, include_plugins=False) ).run(booster_context) print(booster_report.summary()) print("\ncontext.model is:", booster_context.model)

raw output shape: (5,)

Gate status: BLOCKED (2166ms, binary)
  average_precision: 0.1704
  performance: 0 flag(s)
  compliance: 0 flag(s)
  security: 1 flag(s)
  fairness: 0 flag(s)

context.model is: None

In [7]:

display(
    pd.DataFrame(
        [
            {
                "api": label,
                "verdict": rep.gate_status,
                "average_precision": round(rep.model_score, 4),
                "flags": len(rep.flags),
                "check_errors": sum(r.flag == "CHECK_ERROR" for r in rep.results),
            }
            for label, rep in (("XGBClassifier (model=)", sk_report),
                               ("Booster (predict_fn)", booster_report))
        ]
    ).set_index("api")
)

display( pd.DataFrame( [ { "api": label, "verdict": rep.gate_status, "average_precision": round(rep.model_score, 4), "flags": len(rep.flags), "check_errors": sum(r.flag == "CHECK_ERROR" for r in rep.results), } for label, rep in (("XGBClassifier (model=)", sk_report), ("Booster (predict_fn)", booster_report)) ] ).set_index("api") )

	verdict	average_precision	flags	check_errors
api
XGBClassifier (model=)	BLOCKED	0.1704	1	0
Booster (predict_fn)	BLOCKED	0.1704	1	0

4. SHAP on a tree booster¶

ShapSubgroupCheck prefers shap.TreeExplainer when it can recognise a tree model, which computes exact contributions. With a predict_fn-only context there is no model object to introspect, so it explains the function as a black box — an approximation instead.

So there is a reason to pass model= as well as predict_fn when you have both: the checks that can introspect the model will.

In [8]:

import time

from bdp_model_gate.structured.fairness import ShapSubgroupCheck

for label, ctx in (("model= (TreeExplainer)", sk_context),
                   ("predict_fn (black box)", booster_context)):
    start = time.perf_counter()
    results = ShapSubgroupCheck().run(ctx)
    elapsed = time.perf_counter() - start
    flags = [r for r in results if not r.is_ok]
    print(f"{label:26} {elapsed:6.2f}s  {len(flags)} flag(s)  first={results[0].flag}")

import time from bdp_model_gate.structured.fairness import ShapSubgroupCheck for label, ctx in (("model= (TreeExplainer)", sk_context), ("predict_fn (black box)", booster_context)): start = time.perf_counter() results = ShapSubgroupCheck().run(ctx) elapsed = time.perf_counter() - start flags = [r for r in results if not r.is_ok] print(f"{label:26} {elapsed:6.2f}s {len(flags)} flag(s) first={results[0].flag}")

model= (TreeExplainer)       1.72s  0 flag(s)  first=OK

predict_fn (black box)       2.15s  0 flag(s)  first=OK

On this dataset the two are close in wall-clock time — the black-box explainer is not always dramatically slower, and on a small validation set the difference is easily lost in noise. The reason to prefer model= is exactness, not speed: the tree path computes contributions analytically while the black-box path samples. Expect the gap to widen with more features and more rows.

5. --model-loader — gating a checkpoint from CI¶

joblib only reads pickles. A Booster saved with save_model() is JSON, so the CLI needs a function that knows how to load it. --model-loader takes a package.module:factory reference; your loader does the framework import, which is why this package needs no XGBoost dependency.

In [9]:

import json
import subprocess
import sys
from pathlib import Path

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

booster.save_model(str(workdir / "fraud.json"))

(workdir / "serving.py").write_text(
    '''
import numpy as np
import xgboost as xgb


def load_scorer():
    """Returns fn(DataFrame) -> array of positive-class probabilities."""
    booster = xgb.Booster()
    booster.load_model("fraud.json")
    return lambda df: booster.predict(xgb.DMatrix(df))
'''.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)
(workdir / "model_card.json").write_text(json.dumps(model_card, indent=2))

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",
        "--model-card", "model_card.json",
        "--task", "binary",
        "--metric", "average_precision",
        "--min-score", "0.15",
        "--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 json import subprocess import sys from pathlib import Path workdir = Path("cli_booster") workdir.mkdir(exist_ok=True) booster.save_model(str(workdir / "fraud.json")) (workdir / "serving.py").write_text( ''' import numpy as np import xgboost as xgb def load_scorer(): """Returns fn(DataFrame) -> array of positive-class probabilities.""" booster = xgb.Booster() booster.load_model("fraud.json") return lambda df: booster.predict(xgb.DMatrix(df)) '''.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) (workdir / "model_card.json").write_text(json.dumps(model_card, indent=2)) 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", "--model-card", "model_card.json", "--task", "binary", "--metric", "average_precision", "--min-score", "0.15", "--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 (5161ms, binary)
  average_precision: 0.1704
  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

--model and --model-loader are mutually exclusive, and a bad loader reference fails with a clear message rather than a traceback.

In [10]:

for bad in ("no_colon_here", "definitely_not_a_module:load"):
    proc = subprocess.run(
        [
            sys.executable, "-m", "bdp_model_gate.cli",
            "--model-loader", bad,
            "--data", "validation.csv", "--target-col", "is_fraud",
            "--output", "out.json",
        ],
        capture_output=True, text=True, cwd=workdir,
    )
    print(f"{bad:32} exit={proc.returncode}  {proc.stderr.strip().splitlines()[-1][:80]}")

for bad in ("no_colon_here", "definitely_not_a_module:load"): proc = subprocess.run( [ sys.executable, "-m", "bdp_model_gate.cli", "--model-loader", bad, "--data", "validation.csv", "--target-col", "is_fraud", "--output", "out.json", ], capture_output=True, text=True, cwd=workdir, ) print(f"{bad:32} exit={proc.returncode} {proc.stderr.strip().splitlines()[-1][:80]}")

no_colon_here                    exit=1  Configuration error: --model-loader must be 'package.module:factory', got 'no_co

definitely_not_a_module:load     exit=1  Configuration error: could not import 'definitely_not_a_module' for --model-load

In [11]:

import shutil

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

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

cleaned up

In CI, the loader lives with your training code:

- name: Model governance gate
  run: |
    bdp-model-gate \
      --model-loader "mypkg.serving:load_scorer" \
      --data validation.csv --target-col is_fraud \
      --task binary --metric average_precision --min-score 0.20 \
      --output gate_report.json

Summary¶

Your booster	What to pass
XGBClassifier / LGBMClassifier (sklearn API)	model=estimator — no adapter
xgboost.Booster (native)	predict_fn=lambda df: booster.predict(xgb.DMatrix(df))
lightgbm.Booster	predict_fn=lambda df: booster.predict(df)
A saved checkpoint, from CI	--model-loader "package.module:factory"

Three things worth carrying away:

1. Pass model= when you have it, even alongside predict_fn. Checks that can introspect the model do — ShapSubgroupCheck computes exact contributions via TreeExplainer rather than sampling a black box.
2. Read robustness flags in context. Tree models have no gradient and no coef_, so the check falls back to random perturbation — a weaker probe than the gradient attack available to a differentiable model (04).
3. average_precision over roc_auc for imbalanced fraud. AUC looks comfortable at a 5% base rate; average precision does not flatter.

Back to 01 binary · 02 multiclass · 03 regression · 04 PyTorch