nirs4all.pipeline.dagml.conformal_contracts module

Internal conformal calibration contracts for the native DAG-ML lane.

This is the Python-side substrate for the planned nirs4all.calibrate() surface. It intentionally works on already replayed point predictions and targets; it does not fit, select, refit, or call public prediction APIs.

class nirs4all.pipeline.dagml.conformal_contracts.CalibratedPredictionBlock(y_pred: ndarray, intervals: dict[float, ConformalIntervalBlock], method: Literal['split_absolute_residual'] = 'split_absolute_residual', unit: str = 'physical_sample', group_keys: tuple[str, ...] = ())

Bases: object

Point predictions and calibrated intervals for one prediction cohort.

property coverages: tuple[float, ...]

Return materialized coverages in deterministic order.

group_keys: tuple[str, ...] = ()

interval(coverage: float) → ConformalIntervalBlock

Return the interval block for an already materialized coverage.

intervals: dict[float, ConformalIntervalBlock]

method: Literal['split_absolute_residual'] = 'split_absolute_residual'

unit: str = 'physical_sample'

y_pred: ndarray

class nirs4all.pipeline.dagml.conformal_contracts.CalibratedRunResult(artifact: ConformalCalibrationArtifact, prediction: CalibratedPredictionBlock, sample_ids: tuple[str, ...]=(), metadata: Mapping[str, ~typing.Any]=<factory>, version: int = 1)

Bases: object

Serializable internal result for one calibrated prediction cohort.

artifact: ConformalCalibrationArtifact

property calibration_replay_source: dict[str, Any] | None

Return persisted calibration replay provenance, when present.

property conformal_guarantee_status: dict[str, Any] | None

Return persisted guarantee metadata, when present.

property fingerprint: str

TCV1 fingerprint of the calibrated result summary.

classmethod from_dict(payload: Mapping[str, Any]) → CalibratedRunResult

Parse a serialized calibrated result and verify its fingerprint.

classmethod from_json(payload: str) → CalibratedRunResult

Parse a JSON calibrated result and verify its fingerprint.

classmethod load_json(path: str | Path) → CalibratedRunResult

Load a persisted calibrated result and verify its fingerprint.

metadata: Mapping[str, Any]

metrics(y_true: Any) → dict[float, ConformalMetricSet]

Evaluate materialized conformal intervals against observed targets.

prediction: CalibratedPredictionBlock

robustness(*, y_true: Any, X: Any | None = None, predictor: Any | None = None, predictor_bundle: str | Path | None = None, mode: str = 'clean_frozen', scenarios: Sequence[Any] | None = None, slice_by: Sequence[str] | None = None, metadata: Mapping[str, Any] | Sequence[Mapping[str, Any]] | None = None, seed: int | None = None, workspace_path: str | Path | None = None, workspace_name: str = '', workspace_robustness_id: str | None = None, workspace_metadata: Mapping[str, Any] | None = None) → RobustnessReport

Compute an audit-only robustness report for this calibrated result.

sample_ids: tuple[str, ...] = ()

save_json(path: str | Path) → Path

Persist the calibrated result as deterministic verified JSON.

to_dict() → dict[str, Any]

Return the deterministic JSON-like calibrated result form.

to_json(*, indent: int | None = 2) → str

Return deterministic JSON with a trailing newline.

to_predict_result()

Convert this calibrated result to the public PredictResult surface.

property tuning_calibration_source: dict[str, Any] | None

Return persisted native tuning calibration provenance, when present.

version: int = 1

class nirs4all.pipeline.dagml.conformal_contracts.ConformalCalibrationArtifact(spec: ConformalCalibrationSpec, calibrator: SplitConformalCalibrator, calibration_size: int, target_name: str | None = None, predictor_fingerprint: str | None = None, calibration_data_fingerprint: str | None = None, calibration_cohort: ConformalCalibrationCohortManifest | None = None, group_calibrators: Mapping[str, ~nirs4all.pipeline.dagml.conformal_contracts.SplitConformalCalibrator]=<factory>, version: int = 1)

Bases: object

Serializable fitted conformal calibrator artifact.

calibration_cohort: ConformalCalibrationCohortManifest | None = None

calibration_data_fingerprint: str | None = None

calibration_size: int

calibrator: SplitConformalCalibrator

property fingerprint: str

TCV1 fingerprint of the artifact summary.

classmethod from_dict(payload: Mapping[str, Any]) → ConformalCalibrationArtifact

Parse a serialized conformal artifact and verify its fingerprint.

classmethod from_json(payload: str) → ConformalCalibrationArtifact

Parse a JSON conformal artifact and verify its fingerprint.

group_calibrators: Mapping[str, SplitConformalCalibrator]

classmethod load_json(path: str | Path) → ConformalCalibrationArtifact

Load a persisted conformal artifact and verify its fingerprint.

predictor_fingerprint: str | None = None

save_json(path: str | Path) → Path

Persist the conformal artifact as deterministic verified JSON.

spec: ConformalCalibrationSpec

target_name: str | None = None

to_dict() → dict[str, Any]

Return the deterministic JSON-like artifact form.

to_json(*, indent: int | None = 2) → str

Return deterministic JSON with a trailing newline.

version: int = 1

class nirs4all.pipeline.dagml.conformal_contracts.ConformalCalibrationCohortManifest(rows: tuple[ConformalCalibrationCohortRow, ...], unit: str = 'physical_sample', version: int = 1)

Bases: object

Physical-sample calibration cohort identity manifest.

property fingerprint: str

TCV1 fingerprint of the calibration cohort manifest.

classmethod from_dict(payload: Mapping[str, Any]) → ConformalCalibrationCohortManifest

Parse a serialized cohort manifest and verify its fingerprint.

property groups: tuple[str | None, ...]

Calibration group labels in row order.

property n_samples: int

Number of physical calibration samples.

rows: tuple[ConformalCalibrationCohortRow, ...]

property sample_ids: tuple[str, ...]

Calibration physical sample ids in row order.

to_dict() → dict[str, Any]

Return the deterministic JSON-like manifest form.

unit: str = 'physical_sample'

version: int = 1

class nirs4all.pipeline.dagml.conformal_contracts.ConformalCalibrationCohortRow(row_index: int, sample_id: str, role: str = 'calibration', group: str | None = None, metadata: Mapping[str, ~typing.Any]=<factory>)

Bases: object

One physical sample row used for conformal calibration.

classmethod from_dict(payload: Mapping[str, Any]) → ConformalCalibrationCohortRow

Parse one calibration cohort row.

group: str | None = None

metadata: Mapping[str, Any]

role: str = 'calibration'

row_index: int

sample_id: str

to_dict() → dict[str, Any]

Return a deterministic JSON-like row form.

class nirs4all.pipeline.dagml.conformal_contracts.ConformalCalibrationSpec(coverage: tuple[float, ...], method: Literal['split_absolute_residual'] = 'split_absolute_residual', unit: str = 'physical_sample', group_by: tuple[str, ...] = (), multi_target: Literal['marginal', 'joint_max'] = 'marginal')

Bases: object

Normalized planned nirs4all.calibrate(...) calibration request.

coverage: tuple[float, ...]

property fingerprint: str

TCV1 fingerprint of the normalized conformal calibration contract.

group_by: tuple[str, ...] = ()

method: Literal['split_absolute_residual'] = 'split_absolute_residual'

multi_target: Literal['marginal', 'joint_max'] = 'marginal'

to_dict() → dict[str, Any]

Return the deterministic JSON-like contract form.

unit: str = 'physical_sample'

class nirs4all.pipeline.dagml.conformal_contracts.ConformalIntervalBlock(coverage: float, qhat: float | ndarray, lower: ndarray, upper: ndarray)

Bases: object

Prediction intervals for one coverage level.

coverage: float

lower: ndarray

qhat: float | ndarray

upper: ndarray

class nirs4all.pipeline.dagml.conformal_contracts.ConformalMetricSet(coverage: float, observed_coverage: float, coverage_gap: float, mean_width: float, median_width: float, mean_interval_score: float, n_samples: int, n_covered: int, n_missed_below: int, n_missed_above: int, unit: str = 'physical_sample', version: int = 1)

Bases: object

Deterministic conformal diagnostics for one materialized coverage.

coverage: float

coverage_gap: float

property fingerprint: str

TCV1 fingerprint of the conformal metric summary.

mean_interval_score: float

mean_width: float

median_width: float

n_covered: int

n_missed_above: int

n_missed_below: int

n_samples: int

observed_coverage: float

to_dict() → dict[str, Any]

Return a deterministic JSON-like metric form.

unit: str = 'physical_sample'

version: int = 1

class nirs4all.pipeline.dagml.conformal_contracts.SplitConformalCalibrator(scores: ndarray, coverages: tuple[float, ...], qhat_by_coverage: dict[float, float], method: Literal['split_absolute_residual'] = 'split_absolute_residual', unit: str = 'physical_sample')

Bases: object

Fitted split-absolute-residual calibrator for one target.

apply(y_pred: Any) → dict[float, ConformalIntervalBlock]

Apply calibrated intervals to point predictions.

apply_block(y_pred: Any) → CalibratedPredictionBlock

Apply calibrated intervals and keep point predictions with them.

coverages: tuple[float, ...]

method: Literal['split_absolute_residual'] = 'split_absolute_residual'

qhat_by_coverage: dict[float, float]

scores: ndarray

unit: str = 'physical_sample'

nirs4all.pipeline.dagml.conformal_contracts.apply_conformal_artifact_to_predictions(artifact: ConformalCalibrationArtifact, y_pred: Any, *, group_keys: Sequence[str] = ()) → CalibratedPredictionBlock

Apply a conformal artifact to predictions, including grouped calibrators.

nirs4all.pipeline.dagml.conformal_contracts.apply_split_conformal_calibrator(calibrator: SplitConformalCalibrator, y_pred: Any) → CalibratedPredictionBlock

Apply a split conformal calibrator and return a typed prediction block.

nirs4all.pipeline.dagml.conformal_contracts.calibrate_replayed_predictions(*, y_true_calibration: Any, y_pred_calibration: Any, y_pred: Any, spec: ConformalCalibrationSpec, calibration_sample_ids: Any, prediction_sample_ids: Any, calibration_groups: Any | None = None, calibration_metadata: Any | None = None, prediction_groups: Any | None = None, prediction_metadata: Any | None = None, result_metadata: Mapping[str, Any] | None = None, target_name: str | None = None, predictor_fingerprint: str | None = None, calibration_replay_source: Mapping[str, Any] | None = None) → CalibratedRunResult

Calibrate already replayed predictions against an explicit calibration cohort.

nirs4all.pipeline.dagml.conformal_contracts.conformal_finite_sample_quantile(scores: Any, coverage: float) → float

Return finite-sample split conformal quantile.

Uses k = ceil((n + 1) * coverage) over sorted nonconformity scores. If k > n, the finite-sample conformal interval is unbounded.

nirs4all.pipeline.dagml.conformal_contracts.conformal_guarantee_status(artifact: ConformalCalibrationArtifact, *, effective_engine: str, requested_engine: str = 'nirs4all.conformal.v1', calibration_replay_source: Mapping[str, Any] | None = None, selected_coverages: Any | None = None, source_calibrated_result_fingerprint: str | None = None, invalidation_reasons: tuple[str, ...] = ()) → dict[str, Any]

Return fail-loud metadata for the statistical guarantee actually carried.

The payload is intentionally descriptive and JSON-compatible. It is not a new proof by itself; it records the method, engine, unit, selected coverages, and invalidation state that downstream APIs and Studio can show without inferring guarantees from interval arrays or empirical metrics.

nirs4all.pipeline.dagml.conformal_contracts.evaluate_conformal_prediction(*, y_true: Any, prediction: CalibratedPredictionBlock) → dict[float, ConformalMetricSet]

Compute deterministic diagnostics for materialized conformal intervals.

nirs4all.pipeline.dagml.conformal_contracts.fit_conformal_calibration_artifact(y_true: Any, y_pred: Any, *, spec: ConformalCalibrationSpec, calibration_cohort: ConformalCalibrationCohortManifest | None = None, target_name: str | None = None, predictor_fingerprint: str | None = None, calibration_data_fingerprint: str | None = None) → ConformalCalibrationArtifact

Fit the current V1 conformal artifact from replayed calibration arrays.

nirs4all.pipeline.dagml.conformal_contracts.fit_split_absolute_residual_calibrator(y_true: Any, y_pred: Any, *, coverage: float | list[float] | tuple[float, ...], unit: str = 'physical_sample') → SplitConformalCalibrator

Fit split conformal intervals from calibration targets and predictions.

nirs4all.pipeline.dagml.conformal_contracts.normalize_conformal_calibration_cohort(y_true: Any, y_pred: Any, *, sample_ids: Any, groups: Any | None = None, metadata: Any | None = None, unit: str = 'physical_sample', role: str = 'calibration') → ConformalCalibrationCohortManifest

Normalize row-aligned calibration arrays into a physical-sample manifest.

nirs4all.pipeline.dagml.conformal_contracts.normalize_coverages(coverage: float | list[float] | tuple[float, ...]) → tuple[float, ...]

Normalize coverage scalar/list to a sorted unique tuple.

nirs4all.pipeline.dagml.conformal_contracts.parse_conformal_calibration_spec(calibration: Mapping[str, Any], *, context: str = 'nirs4all.calibrate(...)', _allow_normalized_empty_group_by: bool = False) → ConformalCalibrationSpec

Validate and normalize a planned conformal calibration mapping.