cursus.steps.scripts.bspline_calibration

Monotone B-Spline Calibration Script for SageMaker Processing.

main(input_paths, output_paths, environ_vars, job_args) contract, same I/O channels (input evaluation_data -> outputs calibration_output / metrics_output / calibrated_data), and the same env-var surface (it sets CALIBRATION_METHOD=bspline where model_calibration.py defaults to gam, plus the B-spline knobs MONOTONIC_CONSTRAINT / DEGREE / N_KNOTS / ERROR_THRESHOLD).

It therefore has NO .step.yaml of its own (and no registry entry) by design — it satisfies the existing model_calibration.step.yaml interface. To use the B-spline method instead of GAM for the ModelCalibration step, point the step’s contract.entry_point (or the project’s source_dir runner) at this script. See model_calibration.py for the GAM/isotonic/Platt sibling. ==============================================================================

Python port of COSA’s generic_rfuge.r — fits a monotone logistic B-spline to calibrate raw model scores to probabilities. Monotonicity is enforced via quadratic programming (hard constraint: delta[j+1] >= delta[j]).

Algorithm:

  1. Validate inputs (score/tag ranges, minimum records/fraud)

  2. Adaptive knot placement (quantile-based, dense near 1.0)

  3. Aggregate by score (group → freq, nbad)

  4. Fit monotone B-spline via IRLS + constrained QP

  5. Output bspline_parameters.json

Dependencies:

  • numpy, scipy (B-spline basis, initial GLM)

  • quadprog (constrained QP solver — pip install quadprog)

  • pandas (data aggregation)

Replaces: generic_rfuge.r (R script requiring R container + 17 R packages)

bspline_basis(x, knots, degree=3)[source]

Construct B-spline basis matrix with intercept.

Equivalent to R’s bs(x, knots=interior, degree=degree, Boundary.knots=c(0,1), intercept=TRUE).

Parameters:

  • x (ndarray) – Evaluation points (N,)

  • knots (ndarray) – Full knot sequence including boundary (K,)

  • degree (int) – Spline degree (default: 3 = cubic)

Returns:

Basis matrix (N, n_basis) where n_basis = len(knots) + degree - 1

Return type:

B

compute_adaptive_knots(x, n_knots=50, min_count_per_interval=10)[source]

Compute adaptive knots based on data quantiles with dense placement near 1.0.

Exact port of R script’s knot placement logic: 1. Build quantile grid with extra density near 1.0 2. Compute quantiles of x at grid points 3. Add uniform grid seq(0,1,11) 4. Filter: drop intervals with <= min_count_per_interval data points

Parameters:

  • x (ndarray) – Score values (N,)

  • n_knots (int) – Base number of knots

  • min_count_per_interval (int) – Minimum data points per interval

Returns:

Filtered knot positions including 0 and 1

Return type:

knots

monotone_spline(y, x, size, degree=3, boundary=(0.0, 1.0), knots=None, tolerance=1e-06, max_iteration=1000, lam=1e-10)[source]

Fit monotone B-spline via IRLS + constrained QP.

Exact port of R’s monotone_spline() function. At each IRLS iteration, solves a QP with monotonicity constraints delta[j+1] >= delta[j].

Parameters:

  • y (ndarray) – Number of positives per group (N,)

  • x (ndarray) – Score values per group (N,)

  • size (ndarray) – Group sizes (N,)

  • degree (int) – B-spline degree

  • boundary (Tuple[float, float]) – Score range

  • knots (ndarray) – Knot positions (including boundaries)

  • tolerance (float) – Convergence tolerance

  • max_iteration (int) – Maximum IRLS iterations

  • lam (float) – P-spline penalty strength

Returns:

delta, fitted_values, knots, degree, converged

Return type:

Dict with keys

validate_inputs(tags, scores)[source]

Validate inputs (matches R script checks exactly).

validate_result(coefficients, fitted_values, tpr)[source]

Validate calibration result (matches R script checks).

run_bspline_calibration(tags, scores, degree=3, n_knots=None)[source]

Run full B-spline calibration pipeline.

Parameters:

  • tags (ndarray) – Binary labels (N,)

  • scores (ndarray) – Raw model scores (N,)

  • degree (int) – B-spline degree (default: 3)

  • n_knots (int | None) – Number of knots (auto if None)

Returns:

Dict with coefficients, augmented knots, degree, status

Return type:

Dict[str, Any]

main(input_paths, output_paths, environ_vars, job_args=None)[source]

Main entry point for the B-spline calibration script.

Drop-in replacement for model_calibration.py when CALIBRATION_METHOD=bspline. Same input/output contract: reads evaluation data with label + score columns, outputs calibration_summary.json, calibration_metrics.json, calibration_model.pkl, and calibrated data.

Parameters:

  • input_paths (dict) – Dictionary of input paths with logical names - “evaluation_data”: Path to eval data (CSV/TSV/Parquet with label + score columns)

  • output_paths (dict) – Dictionary of output paths with logical names - “calibration_output”: Path for calibration artifacts - “metrics_output”: Path for metrics JSON - “calibrated_data”: Path for calibrated predictions

  • environ_vars (dict) – Dictionary of environment variables - “CALIBRATION_METHOD”: Must be “bspline” (default) - “LABEL_FIELD”: Label column name (default: “label”) - “SCORE_FIELD”: Score column name (default: “prob_class_1”) - “SCORE_FIELDS”: Multi-task comma-separated score fields (optional) - “TASK_LABEL_NAMES”: Multi-task comma-separated label fields (optional) - “IS_BINARY”: “True”/”False” (default: “True”) - “MONOTONIC_CONSTRAINT”: “True”/”False” (default: “True”) - “CALIBRATION_SAMPLE_POINTS”: Lookup table size (default: “1000”) - “DEGREE”: B-spline degree (default: “3”) - “N_KNOTS”: Number of knots (default: auto) - “ERROR_THRESHOLD”: Min improvement threshold (default: “0.05”)

  • job_args (Namespace) – Command line arguments (optional)

Returns:

Dictionary with status and results (same contract as model_calibration.py)

Return type:

dict