DeepMTP.runtime package

The runtime namespace owns numerical training and evaluation, experiment persistence, lifecycle presentation, reporting integrations, and structured evaluation progress:

from DeepMTP.runtime import (
    Evaluator,
    ExperimentStore,
    LOSS_REGISTRY,
    OUTPUT_HEAD_REGISTRY,
    ProgressObserver,
    TensorBoardReporter,
    TrainingEngine,
)

Task components

LOSS_REGISTRY and OUTPUT_HEAD_REGISTRY expose the components selected for classification and regression. Binary classification defaults to raw model scores and BCEWithLogitsLoss; sigmoid is applied only to values collected for metrics and prediction. The explicit binary_cross_entropy compatibility loss applies sigmoid before BCELoss.

Regression supports mean_squared_error, mean_absolute_error, and huber. Every regression loss uses raw model output and an identity output head. build_task_components rejects task/loss combinations that do not share a defined contract, and every component builder returns a fresh PyTorch module for each trainer.

Registries for task-level losses and output heads.

class DeepMTP.runtime.components.TaskComponents(loss: torch.nn.Module, loss_input_transform: torch.nn.Module, output_head: torch.nn.Module)

Bases: object

Loss-input and user-output components selected for one problem mode.

loss: torch.nn.Module

loss_input_transform: torch.nn.Module

output_head: torch.nn.Module

DeepMTP.runtime.components.build_default_loss(problem_mode: str) → torch.nn.Module: Build the default registered loss for a problem mode.

DeepMTP.runtime.components.build_default_loss_input_transform(problem_mode: str) → torch.nn.Module: Build the transform applied to raw model scores before the loss.

DeepMTP.runtime.components.build_default_output_head(problem_mode: str, classification_mode: str | None = None) → torch.nn.Module: Build the default registered output head for a problem mode.

DeepMTP.runtime.components.build_default_task_components(problem_mode: str) → TaskComponents: Build the behavior-preserving task components for a problem mode.

DeepMTP.runtime.components.build_loss(name: str) → torch.nn.Module: Build a registered loss module by name.

DeepMTP.runtime.components.build_loss_input_transform(loss_name: str) → torch.nn.Module: Build the transform applied before a selected registered loss.

DeepMTP.runtime.components.build_output_head(name: str) → torch.nn.Module: Build a registered output-head module by name.

DeepMTP.runtime.components.build_task_components(problem_mode: str, loss_name: str | None = None, classification_mode: str | None = None) → TaskComponents: Build compatible loss and output modules for one problem mode.

Evaluation

Evaluation policy and metric calculation for DeepMTP experiments.

class DeepMTP.runtime.evaluation.EvaluationPolicy(evaluate_train: bool, evaluate_val: bool, eval_every_n_epochs: int, num_epochs: int, metric_to_optimize_early_stopping: str, use_early_stopping: bool = True, metric_to_optimize_best_epoch_selection: str = 'loss')

Bases: object

Decides when prediction-dependent metrics should be calculated.

eval_every_n_epochs: int

evaluate_train: bool

evaluate_val: bool

classmethod from_config(config: DeepMTPConfig) → EvaluationPolicy

metric_to_optimize_best_epoch_selection: str = 'loss'

metric_to_optimize_early_stopping: str

property metric_to_track: str: Return the metric used to select the model retained after training.

num_epochs: int

should_evaluate(mode: str, epoch: int) → bool: Return whether inference metrics are required for this epoch.

should_evaluate_training(epoch: int) → bool: Return whether training metrics are required for this epoch.

use_early_stopping: bool = True

class DeepMTP.runtime.evaluation.Evaluator(config: DeepMTPConfig | Mapping[str, Any], progress: EvaluationProgressObserver | None = None)

Bases: object

Calculates configured metrics from a numerical epoch output.

evaluate(mode: str, epoch: int, output: EpochOutput) → dict[str, float]: Calculate configured performance metrics.

property policy: EvaluationPolicy: Return policy values reflecting the current typed configuration.

static predictions_frame(output: EpochOutput) → DataFrame: Build the stable tabular prediction result.

Training

Numerical training and evaluation primitives.

class DeepMTP.runtime.training.EpochOutput(losses: list[float] = <factory>, true_values: list[~typing.Any] = <factory>, predicted_values: list[~typing.Any] = <factory>, instance_ids: list[~typing.Any] = <factory>, target_ids: list[~typing.Any] = <factory>, loss_weights: list[int] = <factory>, class_probabilities: list[list[float]] = <factory>)

Bases: object

Losses and optional predictions collected during one data pass.

add_loss(loss: float, *, observations: int) → None: Record a batch-mean loss and the observations it represents.

class_probabilities: list[list[float]]

instance_ids: list[Any]

loss_weights: list[int]

losses: list[float]

property mean_loss: float

predicted_values: list[Any]

target_ids: list[Any]

true_values: list[Any]

class DeepMTP.runtime.training.TrainingEngine(model: nn.Module, criterion: nn.Module, device: torch.device, problem_mode: str, optimizer: torch.optim.Optimizer | None = None, output_head: nn.Module | None = None, loss_input_transform: nn.Module | None = None, classification_mode: str | None = None, num_classes: int | None = None, gradient_accumulation_steps: int = 1, gradient_clip_norm: float | None = None)

Bases: object

Run numerical training and evaluation independently of reporting.

output_head transforms raw model scores for metrics and returned predictions. loss_input_transform independently controls the values consumed by criterion. When it is omitted, BCEWithLogitsLoss sees raw scores and other criteria retain the historical behavior of consuming the output-head values.

evaluate_epoch(dataloader: Iterable[MTPBatch | Mapping[str, Any]], *, calculate_loss: bool = True, collect_predictions: bool = True) → EpochOutput: Evaluate the model for one pass without constructing gradients.

replace_model(model: torch.nn.Module) → None: Update the model after best-epoch selection.

train_epoch(dataloader: Iterable[MTPBatch | Mapping[str, Any]], *, collect_predictions: bool = False) → EpochOutput: Optimize the model for one complete pass over dataloader.

Persistence

Experiment directory and artifact persistence.

class DeepMTP.runtime.persistence.ExperimentStore(experiment_dir: Path)

Bases: object

Owns the filesystem side effects associated with one experiment.

classmethod create(results_path: str | PathLike[str], experiment_name: str | None = None, *, timestamp: str | None = None) → ExperimentStore: Create a unique experiment directory below results_path.

experiment_dir: Path

static load_checkpoint(checkpoint_path: str | os.PathLike[str], *, map_location: str | torch.device = 'cpu') → dict[str, Any]: Load a trusted DeepMTP checkpoint onto an explicit device.

save_checkpoint(*, model_state_dict: Mapping[str, Any], optimizer_state_dict: Mapping[str, Any], config: Mapping[str, Any] | DeepMTPConfig, metadata: Mapping[str, Any] | None = None) → Path: Atomically save a checkpoint using the historical payload format.

save_config(config: Mapping[str, Any] | DeepMTPConfig) → Path: Save JSON-compatible configs as JSON and all others as pickle.

save_summary(train_summary: str, validation_summary: str, test_summary: str) → Path

Checkpoint metadata

The versioned manifest records preprocessing, training, environment, source, and tracking provenance while preserving legacy checkpoints without a metadata entry.

Versioned checkpoint provenance and reproducibility metadata.

DeepMTP.runtime.metadata.build_checkpoint_metadata(config: DeepMTPConfig | Mapping[str, Any], *, effective_device: str, training: Mapping[str, Any] | None = None, tracking: Mapping[str, Any] | None = None) → dict[str, Any]: Build one JSON-compatible manifest for local and remote artifacts.

DeepMTP.runtime.metadata.validate_checkpoint_metadata(value: object) → dict[str, Any]: Validate a checkpoint manifest without accepting future formats.

Presentation

Run history formatting and user-facing progress events.

class DeepMTP.runtime.presentation.ConsoleProgressObserver

Bases: object

Renders progress events using the historical console messages.

on_event(event: ProgressEvent | EvaluationProgressEvent) → None

on_summary(summaries: RunSummaries, *, include_train: bool, include_validation: bool) → None

class DeepMTP.runtime.presentation.NullProgressObserver

Bases: object

No-op observer used for non-verbose experiments.

on_event(event: ProgressEvent | EvaluationProgressEvent) → None

on_summary(summaries: RunSummaries, *, include_train: bool, include_validation: bool) → None

class DeepMTP.runtime.presentation.ProgressEvent(kind: Literal['checkpoint_loading_started', 'checkpoint_loading_completed', 'device_selected', 'checkpoint_weights_started', 'checkpoint_weights_completed', 'checkpoint_saving_started', 'checkpoint_saving_completed', 'training_started', 'training_completed', 'training_failed', 'epoch_started', 'epoch_completed', 'validation_started', 'validation_completed', 'metrics_started', 'metrics_completed', 'early_stopping_counter_updated', 'early_stopping_triggered', 'testing_started', 'testing_completed'], epoch: int | None = None, mode: str | None = None, metrics: Mapping[str, float] | None = None, early_stopping_counter: int | None = None, early_stopping_patience: int | None = None, best_epoch: int | None = None, message: str | None = None)

Bases: object

A lifecycle event suitable for console or UI adapters.

best_epoch: int | None = None

early_stopping_counter: int | None = None

early_stopping_patience: int | None = None

epoch: int | None = None

kind: Literal['checkpoint_loading_started', 'checkpoint_loading_completed', 'device_selected', 'checkpoint_weights_started', 'checkpoint_weights_completed', 'checkpoint_saving_started', 'checkpoint_saving_completed', 'training_started', 'training_completed', 'training_failed', 'epoch_started', 'epoch_completed', 'validation_started', 'validation_completed', 'metrics_started', 'metrics_completed', 'early_stopping_counter_updated', 'early_stopping_triggered', 'testing_started', 'testing_completed']

message: str | None = None

metrics: Mapping[str, float] | None = None

mode: str | None = None

class DeepMTP.runtime.presentation.ProgressObserver(*args, **kwargs)

Bases: Protocol

Receives training lifecycle events and final summaries.

on_event(event: ProgressEvent | EvaluationProgressEvent) → None: Handle one lifecycle event.

on_summary(summaries: RunSummaries, *, include_train: bool, include_validation: bool) → None: Handle the final rendered run summaries.

class DeepMTP.runtime.presentation.RunHistory(metrics: Sequence[str], averaging: Sequence[str])

Bases: object

Collects experiment results and formats the legacy summary tables.

add_test(best_epoch: int, results: Mapping[str, float]) → None

add_training(epoch: int, loss: float, results: Mapping[str, float]) → None

add_validation(epoch: int, loss: float, results: Mapping[str, float], *, early_stopping_counter: int, early_stopping_patience: int) → None

render() → RunSummaries

class DeepMTP.runtime.presentation.RunSummaries(train: str, validation: str, test: str)

Bases: object

Rendered train, validation, and test histories.

test: str

train: str

validation: str

DeepMTP.runtime.presentation.build_progress_observer(verbose: bool) → ProgressObserver: Select the default observer for the configured verbosity.

Reporting

Optional experiment-reporting integrations.

class DeepMTP.runtime.reporting.CompositeReporter(reporters: Sequence[ExperimentReporter])

Bases: object

Fan reporting events out to multiple integrations.

close() → None

log_metrics(metrics: Mapping[str, float], *, step: int | None = None) → None

log_model_artifact(artifact: ReporterModelArtifact) → Mapping[str, Any] | None

log_predictions(report: ReporterPredictionTable) → None

log_summary(values: Mapping[str, Any]) → None

mark_failed(error: BaseException) → None

runtime_metadata() → Mapping[str, Any]

start(config: Mapping[str, Any], model: Any) → None

update_config(config: Mapping[str, Any]) → None

class DeepMTP.runtime.reporting.ExperimentReporter(*args, **kwargs)

Bases: Protocol

Backward-compatible lifecycle used to report experiment information.

close() → None: Flush resources and finish the reporting run.

log_metrics(metrics: Mapping[str, float], *, step: int | None = None) → None: Record a group of metrics.

start(config: Mapping[str, Any], model: Any) → None: Initialize the reporter for an experiment.

class DeepMTP.runtime.reporting.NullReporter

Bases: object

No-op reporter used when no integrations are configured.

close() → None

log_metrics(metrics: Mapping[str, float], *, step: int | None = None) → None

log_model_artifact(artifact: ReporterModelArtifact) → Mapping[str, Any] | None

log_predictions(report: ReporterPredictionTable) → None

log_summary(values: Mapping[str, Any]) → None

mark_failed(error: BaseException) → None

runtime_metadata() → Mapping[str, Any]

start(config: Mapping[str, Any], model: Any) → None

update_config(config: Mapping[str, Any]) → None

exception DeepMTP.runtime.reporting.OptionalIntegrationError

Bases: ImportError

Raised when a requested optional reporting dependency is unavailable.

class DeepMTP.runtime.reporting.ReporterModelArtifact(checkpoint_path: Path, files: tuple[Path, ...], metadata: Mapping[str, Any], name: str | None = None, aliases: tuple[str, ...] = ('latest', 'best'), registry_name: str | None = None, registry_collection: str | None = None)

Bases: object

Local files and metadata to publish as one versioned model artifact.

aliases: tuple[str, ...] = ('latest', 'best')

checkpoint_path: Path

files: tuple[Path, ...]

metadata: Mapping[str, Any]

name: str | None = None

registry_collection: str | None = None

registry_name: str | None = None

class DeepMTP.runtime.reporting.ReporterPredictionTable(split: str, frame: DataFrame, problem_mode: str, classification_mode: str | None, max_rows: int, random_seed: int | None)

Bases: object

Bounded prediction data and task semantics for rich reporters.

classification_mode: str | None

frame: DataFrame

max_rows: int

problem_mode: str

random_seed: int | None

split: str

class DeepMTP.runtime.reporting.TensorBoardReporter(log_dir: str | Path)

Bases: object

TensorBoard adapter with a lazy optional dependency import.

close() → None

log_metrics(metrics: Mapping[str, float], *, step: int | None = None) → None

log_summary(values: Mapping[str, Any]) → None

runtime_metadata() → Mapping[str, Any]

start(config: Mapping[str, Any], model: Any) → None

update_config(config: Mapping[str, Any]) → None

class DeepMTP.runtime.reporting.WandBReporter(project_name: str, project_entity: str, *, experiment_dir: str | Path | None = None, mode: str | None = None, run_name: str | None = None, group: str | None = None, job_type: str = 'train', tags: Sequence[str] = (), notes: str | None = None, watch: str | None = 'gradients', watch_log_freq: int = 1000, log_graph: bool = False, log_code: bool = False, input_artifacts: Sequence[str] = ())

Bases: object

Weights & Biases adapter with lazy imports and artifact support.

close() → None

log_metrics(metrics: Mapping[str, float], *, step: int | None = None) → None

log_model_artifact(artifact: ReporterModelArtifact) → Mapping[str, Any] | None

log_predictions(report: ReporterPredictionTable) → None

log_summary(values: Mapping[str, Any]) → None

mark_failed(error: BaseException) → None

runtime_metadata() → Mapping[str, Any]

start(config: Mapping[str, Any], model: Any) → None

update_config(config: Mapping[str, Any]) → None

DeepMTP.runtime.reporting.build_reporter(config: DeepMTPConfig | Mapping[str, Any], experiment_dir: str | Path) → ExperimentReporter: Build configured reporters from typed or legacy configuration.

DeepMTP.runtime.reporting.log_reporter_model_artifact(reporter: ExperimentReporter, artifact: ReporterModelArtifact) → Mapping[str, Any] | None

DeepMTP.runtime.reporting.log_reporter_predictions(reporter: ExperimentReporter, report: ReporterPredictionTable) → None

DeepMTP.runtime.reporting.log_reporter_summary(reporter: ExperimentReporter, values: Mapping[str, Any]) → None

DeepMTP.runtime.reporting.mark_reporter_failed(reporter: ExperimentReporter, error: BaseException) → None

DeepMTP.runtime.reporting.reporter_runtime_metadata(reporter: ExperimentReporter) → dict[str, Any]

DeepMTP.runtime.reporting.update_reporter_config(reporter: ExperimentReporter, config: Mapping[str, Any]) → None

W&B Artifacts

W&B Artifact download helpers kept behind the optional tracking extra.

class DeepMTP.runtime.wandb.WandBCheckpoint(artifact_reference: str, checkpoint_path: Path, download_root: Path, artifact_name: str | None, artifact_version: str | None, artifact_digest: str | None, checkpoint_format: str = 'full')

Bases: object

A downloaded W&B model artifact containing one DeepMTP checkpoint.

artifact_digest: str | None

artifact_name: str | None

artifact_reference: str

artifact_version: str | None

checkpoint_format: str = 'full'

checkpoint_path: Path

download_root: Path

DeepMTP.runtime.wandb.download_wandb_checkpoint(artifact_reference: str, *, download_dir: str | PathLike[str] | None = None) → WandBCheckpoint: Download and validate one DeepMTP model artifact through the Public API.

Evaluation progress

Structured progress events for metric evaluation.

class DeepMTP.runtime.evaluation_progress.ConsoleEvaluationProgressObserver

Bases: object

Render the historical metric-evaluation console output.

on_event(event: EvaluationProgressEvent) → None

class DeepMTP.runtime.evaluation_progress.EvaluationProgressEvent(kind: Literal['input_inspected', 'evaluation_started', 'values_unscaled', 'metric_calculated', 'group_results_calculated', 'aggregate_results_calculated', 'single_class_group', 'evaluation_completed'], mode: str | None = None, nonzero_predictions: int | None = None, true_values_have_nan: bool = False, predicted_values_have_nan: bool = False, values_preview: object | None = None, announce_unscaled_values: bool = False, metric_name: str | None = None, metric_value: float | None = None, averaging: Literal['micro', 'macro', 'instance'] | None = None, group: Literal['target', 'instance'] | None = None, group_id: object | None = None, top_k: int | None = None, results: object | None = None, unique_true_values: int | None = None)

Bases: object

A diagnostic or result produced while calculating metrics.

announce_unscaled_values: bool = False

averaging: Literal['micro', 'macro', 'instance'] | None = None

group: Literal['target', 'instance'] | None = None

group_id: object | None = None

kind: Literal['input_inspected', 'evaluation_started', 'values_unscaled', 'metric_calculated', 'group_results_calculated', 'aggregate_results_calculated', 'single_class_group', 'evaluation_completed']

metric_name: str | None = None

metric_value: float | None = None

mode: str | None = None

nonzero_predictions: int | None = None

predicted_values_have_nan: bool = False

results: object | None = None

top_k: int | None = None

true_values_have_nan: bool = False

unique_true_values: int | None = None

values_preview: object | None = None

class DeepMTP.runtime.evaluation_progress.EvaluationProgressObserver(*args, **kwargs)

Bases: Protocol

Receives detailed metric-evaluation events.

on_event(event: EvaluationProgressEvent) → None: Handle one evaluation event.

class DeepMTP.runtime.evaluation_progress.NullEvaluationProgressObserver

Bases: object

No-op observer used when detailed metric output is disabled.

on_event(event: EvaluationProgressEvent) → None

DeepMTP.runtime.evaluation_progress.build_evaluation_progress_observer(enabled: bool) → EvaluationProgressObserver: Select the default detailed-evaluation observer.

Package contents

Training-runtime services and integration boundaries.

class DeepMTP.runtime.CompositeReporter(reporters: Sequence[ExperimentReporter])

Bases: object

Fan reporting events out to multiple integrations.

close() → None

log_metrics(metrics: Mapping[str, float], *, step: int | None = None) → None

log_model_artifact(artifact: ReporterModelArtifact) → Mapping[str, Any] | None

log_predictions(report: ReporterPredictionTable) → None

log_summary(values: Mapping[str, Any]) → None

mark_failed(error: BaseException) → None

runtime_metadata() → Mapping[str, Any]

start(config: Mapping[str, Any], model: Any) → None

update_config(config: Mapping[str, Any]) → None

class DeepMTP.runtime.ConsoleEvaluationProgressObserver

Bases: object

Render the historical metric-evaluation console output.

on_event(event: EvaluationProgressEvent) → None

class DeepMTP.runtime.ConsoleProgressObserver

Bases: object

Renders progress events using the historical console messages.

on_event(event: ProgressEvent | EvaluationProgressEvent) → None

on_summary(summaries: RunSummaries, *, include_train: bool, include_validation: bool) → None

class DeepMTP.runtime.EpochOutput(losses: list[float] = <factory>, true_values: list[~typing.Any] = <factory>, predicted_values: list[~typing.Any] = <factory>, instance_ids: list[~typing.Any] = <factory>, target_ids: list[~typing.Any] = <factory>, loss_weights: list[int] = <factory>, class_probabilities: list[list[float]] = <factory>)

Bases: object

Losses and optional predictions collected during one data pass.

add_loss(loss: float, *, observations: int) → None: Record a batch-mean loss and the observations it represents.

class_probabilities: list[list[float]]

instance_ids: list[Any]

loss_weights: list[int]

losses: list[float]

property mean_loss: float

predicted_values: list[Any]

target_ids: list[Any]

true_values: list[Any]

class DeepMTP.runtime.EvaluationPolicy(evaluate_train: bool, evaluate_val: bool, eval_every_n_epochs: int, num_epochs: int, metric_to_optimize_early_stopping: str, use_early_stopping: bool = True, metric_to_optimize_best_epoch_selection: str = 'loss')

Bases: object

Decides when prediction-dependent metrics should be calculated.

eval_every_n_epochs: int

evaluate_train: bool

evaluate_val: bool

classmethod from_config(config: DeepMTPConfig) → EvaluationPolicy

metric_to_optimize_best_epoch_selection: str = 'loss'

metric_to_optimize_early_stopping: str

property metric_to_track: str: Return the metric used to select the model retained after training.

num_epochs: int

should_evaluate(mode: str, epoch: int) → bool: Return whether inference metrics are required for this epoch.

should_evaluate_training(epoch: int) → bool: Return whether training metrics are required for this epoch.

use_early_stopping: bool = True

class DeepMTP.runtime.EvaluationProgressEvent(kind: Literal['input_inspected', 'evaluation_started', 'values_unscaled', 'metric_calculated', 'group_results_calculated', 'aggregate_results_calculated', 'single_class_group', 'evaluation_completed'], mode: str | None = None, nonzero_predictions: int | None = None, true_values_have_nan: bool = False, predicted_values_have_nan: bool = False, values_preview: object | None = None, announce_unscaled_values: bool = False, metric_name: str | None = None, metric_value: float | None = None, averaging: Literal['micro', 'macro', 'instance'] | None = None, group: Literal['target', 'instance'] | None = None, group_id: object | None = None, top_k: int | None = None, results: object | None = None, unique_true_values: int | None = None)

Bases: object

A diagnostic or result produced while calculating metrics.

announce_unscaled_values: bool = False

averaging: Literal['micro', 'macro', 'instance'] | None = None

group: Literal['target', 'instance'] | None = None

group_id: object | None = None

kind: Literal['input_inspected', 'evaluation_started', 'values_unscaled', 'metric_calculated', 'group_results_calculated', 'aggregate_results_calculated', 'single_class_group', 'evaluation_completed']

metric_name: str | None = None

metric_value: float | None = None

mode: str | None = None

nonzero_predictions: int | None = None

predicted_values_have_nan: bool = False

results: object | None = None

top_k: int | None = None

true_values_have_nan: bool = False

unique_true_values: int | None = None

values_preview: object | None = None

class DeepMTP.runtime.EvaluationProgressObserver(*args, **kwargs)

Bases: Protocol

Receives detailed metric-evaluation events.

on_event(event: EvaluationProgressEvent) → None: Handle one evaluation event.

class DeepMTP.runtime.Evaluator(config: DeepMTPConfig | Mapping[str, Any], progress: EvaluationProgressObserver | None = None)

Bases: object

Calculates configured metrics from a numerical epoch output.

evaluate(mode: str, epoch: int, output: EpochOutput) → dict[str, float]: Calculate configured performance metrics.

property policy: EvaluationPolicy: Return policy values reflecting the current typed configuration.

static predictions_frame(output: EpochOutput) → DataFrame: Build the stable tabular prediction result.

class DeepMTP.runtime.ExperimentReporter(*args, **kwargs)

Bases: Protocol

Backward-compatible lifecycle used to report experiment information.

close() → None: Flush resources and finish the reporting run.

log_metrics(metrics: Mapping[str, float], *, step: int | None = None) → None: Record a group of metrics.

start(config: Mapping[str, Any], model: Any) → None: Initialize the reporter for an experiment.

class DeepMTP.runtime.ExperimentStore(experiment_dir: Path)

Bases: object

Owns the filesystem side effects associated with one experiment.

classmethod create(results_path: str | PathLike[str], experiment_name: str | None = None, *, timestamp: str | None = None) → ExperimentStore: Create a unique experiment directory below results_path.

experiment_dir: Path

static load_checkpoint(checkpoint_path: str | os.PathLike[str], *, map_location: str | torch.device = 'cpu') → dict[str, Any]: Load a trusted DeepMTP checkpoint onto an explicit device.

save_checkpoint(*, model_state_dict: Mapping[str, Any], optimizer_state_dict: Mapping[str, Any], config: Mapping[str, Any] | DeepMTPConfig, metadata: Mapping[str, Any] | None = None) → Path: Atomically save a checkpoint using the historical payload format.

save_config(config: Mapping[str, Any] | DeepMTPConfig) → Path: Save JSON-compatible configs as JSON and all others as pickle.

save_summary(train_summary: str, validation_summary: str, test_summary: str) → Path

class DeepMTP.runtime.FoundationAdapterCheckpoint(checkpoint_path: Path, config: dict[str, Any], metadata: dict[str, Any], task_state_dict: Mapping[str, Any], components: tuple[_LoadedFoundationComponent, ...], manifest: dict[str, Any])

Bases: object

Validated compact checkpoint ready to restore into a built model.

checkpoint_path: Path

components: tuple[_LoadedFoundationComponent, ...]

config: dict[str, Any]

manifest: dict[str, Any]

metadata: dict[str, Any]

task_state_dict: Mapping[str, Any]

class DeepMTP.runtime.FoundationAdapterExport(checkpoint_path: Path, manifest: dict[str, Any])

Bases: object

A newly written compact checkpoint and its JSON manifest.

property artifact_metadata: dict[str, Any]: Small W&B-safe summary retaining exact foundation identities.

checkpoint_path: Path

manifest: dict[str, Any]

class DeepMTP.runtime.NullEvaluationProgressObserver

Bases: object

No-op observer used when detailed metric output is disabled.

on_event(event: EvaluationProgressEvent) → None

class DeepMTP.runtime.NullProgressObserver

Bases: object

No-op observer used for non-verbose experiments.

on_event(event: ProgressEvent | EvaluationProgressEvent) → None

on_summary(summaries: RunSummaries, *, include_train: bool, include_validation: bool) → None

class DeepMTP.runtime.NullReporter

Bases: object

No-op reporter used when no integrations are configured.

close() → None

log_metrics(metrics: Mapping[str, float], *, step: int | None = None) → None

log_model_artifact(artifact: ReporterModelArtifact) → Mapping[str, Any] | None

log_predictions(report: ReporterPredictionTable) → None

log_summary(values: Mapping[str, Any]) → None

mark_failed(error: BaseException) → None

runtime_metadata() → Mapping[str, Any]

start(config: Mapping[str, Any], model: Any) → None

update_config(config: Mapping[str, Any]) → None

exception DeepMTP.runtime.OptionalIntegrationError

Bases: ImportError

Raised when a requested optional reporting dependency is unavailable.

class DeepMTP.runtime.ProgressEvent(kind: Literal['checkpoint_loading_started', 'checkpoint_loading_completed', 'device_selected', 'checkpoint_weights_started', 'checkpoint_weights_completed', 'checkpoint_saving_started', 'checkpoint_saving_completed', 'training_started', 'training_completed', 'training_failed', 'epoch_started', 'epoch_completed', 'validation_started', 'validation_completed', 'metrics_started', 'metrics_completed', 'early_stopping_counter_updated', 'early_stopping_triggered', 'testing_started', 'testing_completed'], epoch: int | None = None, mode: str | None = None, metrics: Mapping[str, float] | None = None, early_stopping_counter: int | None = None, early_stopping_patience: int | None = None, best_epoch: int | None = None, message: str | None = None)

Bases: object

A lifecycle event suitable for console or UI adapters.

best_epoch: int | None = None

early_stopping_counter: int | None = None

early_stopping_patience: int | None = None

epoch: int | None = None

kind: Literal['checkpoint_loading_started', 'checkpoint_loading_completed', 'device_selected', 'checkpoint_weights_started', 'checkpoint_weights_completed', 'checkpoint_saving_started', 'checkpoint_saving_completed', 'training_started', 'training_completed', 'training_failed', 'epoch_started', 'epoch_completed', 'validation_started', 'validation_completed', 'metrics_started', 'metrics_completed', 'early_stopping_counter_updated', 'early_stopping_triggered', 'testing_started', 'testing_completed']

message: str | None = None

metrics: Mapping[str, float] | None = None

mode: str | None = None

class DeepMTP.runtime.ProgressObserver(*args, **kwargs)

Bases: Protocol

Receives training lifecycle events and final summaries.

on_event(event: ProgressEvent | EvaluationProgressEvent) → None: Handle one lifecycle event.

on_summary(summaries: RunSummaries, *, include_train: bool, include_validation: bool) → None: Handle the final rendered run summaries.

class DeepMTP.runtime.ReporterModelArtifact(checkpoint_path: Path, files: tuple[Path, ...], metadata: Mapping[str, Any], name: str | None = None, aliases: tuple[str, ...] = ('latest', 'best'), registry_name: str | None = None, registry_collection: str | None = None)

Bases: object

Local files and metadata to publish as one versioned model artifact.

aliases: tuple[str, ...] = ('latest', 'best')

checkpoint_path: Path

files: tuple[Path, ...]

metadata: Mapping[str, Any]

name: str | None = None

registry_collection: str | None = None

registry_name: str | None = None

class DeepMTP.runtime.ReporterPredictionTable(split: str, frame: DataFrame, problem_mode: str, classification_mode: str | None, max_rows: int, random_seed: int | None)

Bases: object

Bounded prediction data and task semantics for rich reporters.

classification_mode: str | None

frame: DataFrame

max_rows: int

problem_mode: str

random_seed: int | None

split: str

class DeepMTP.runtime.RunHistory(metrics: Sequence[str], averaging: Sequence[str])

Bases: object

Collects experiment results and formats the legacy summary tables.

add_test(best_epoch: int, results: Mapping[str, float]) → None

add_training(epoch: int, loss: float, results: Mapping[str, float]) → None

add_validation(epoch: int, loss: float, results: Mapping[str, float], *, early_stopping_counter: int, early_stopping_patience: int) → None

render() → RunSummaries

class DeepMTP.runtime.RunSummaries(train: str, validation: str, test: str)

Bases: object

Rendered train, validation, and test histories.

test: str

train: str

validation: str

class DeepMTP.runtime.TaskComponents(loss: torch.nn.Module, loss_input_transform: torch.nn.Module, output_head: torch.nn.Module)

Bases: object

Loss-input and user-output components selected for one problem mode.

loss: torch.nn.Module

loss_input_transform: torch.nn.Module

output_head: torch.nn.Module

class DeepMTP.runtime.TensorBoardReporter(log_dir: str | Path)

Bases: object

TensorBoard adapter with a lazy optional dependency import.

close() → None

log_metrics(metrics: Mapping[str, float], *, step: int | None = None) → None

log_summary(values: Mapping[str, Any]) → None

runtime_metadata() → Mapping[str, Any]

start(config: Mapping[str, Any], model: Any) → None

update_config(config: Mapping[str, Any]) → None

class DeepMTP.runtime.TrainingEngine(model: nn.Module, criterion: nn.Module, device: torch.device, problem_mode: str, optimizer: torch.optim.Optimizer | None = None, output_head: nn.Module | None = None, loss_input_transform: nn.Module | None = None, classification_mode: str | None = None, num_classes: int | None = None, gradient_accumulation_steps: int = 1, gradient_clip_norm: float | None = None)

Bases: object

Run numerical training and evaluation independently of reporting.

output_head transforms raw model scores for metrics and returned predictions. loss_input_transform independently controls the values consumed by criterion. When it is omitted, BCEWithLogitsLoss sees raw scores and other criteria retain the historical behavior of consuming the output-head values.

evaluate_epoch(dataloader: Iterable[MTPBatch | Mapping[str, Any]], *, calculate_loss: bool = True, collect_predictions: bool = True) → EpochOutput: Evaluate the model for one pass without constructing gradients.

replace_model(model: torch.nn.Module) → None: Update the model after best-epoch selection.

train_epoch(dataloader: Iterable[MTPBatch | Mapping[str, Any]], *, collect_predictions: bool = False) → EpochOutput: Optimize the model for one complete pass over dataloader.

class DeepMTP.runtime.WandBCheckpoint(artifact_reference: str, checkpoint_path: Path, download_root: Path, artifact_name: str | None, artifact_version: str | None, artifact_digest: str | None, checkpoint_format: str = 'full')

Bases: object

A downloaded W&B model artifact containing one DeepMTP checkpoint.

artifact_digest: str | None

artifact_name: str | None

artifact_reference: str

artifact_version: str | None

checkpoint_format: str = 'full'

checkpoint_path: Path

download_root: Path

class DeepMTP.runtime.WandBReporter(project_name: str, project_entity: str, *, experiment_dir: str | Path | None = None, mode: str | None = None, run_name: str | None = None, group: str | None = None, job_type: str = 'train', tags: Sequence[str] = (), notes: str | None = None, watch: str | None = 'gradients', watch_log_freq: int = 1000, log_graph: bool = False, log_code: bool = False, input_artifacts: Sequence[str] = ())

Bases: object

Weights & Biases adapter with lazy imports and artifact support.

close() → None

log_metrics(metrics: Mapping[str, float], *, step: int | None = None) → None

log_model_artifact(artifact: ReporterModelArtifact) → Mapping[str, Any] | None

log_predictions(report: ReporterPredictionTable) → None

log_summary(values: Mapping[str, Any]) → None

mark_failed(error: BaseException) → None

runtime_metadata() → Mapping[str, Any]

start(config: Mapping[str, Any], model: Any) → None

update_config(config: Mapping[str, Any]) → None

DeepMTP.runtime.build_checkpoint_metadata(config: DeepMTPConfig | Mapping[str, Any], *, effective_device: str, training: Mapping[str, Any] | None = None, tracking: Mapping[str, Any] | None = None) → dict[str, Any]: Build one JSON-compatible manifest for local and remote artifacts.

DeepMTP.runtime.build_default_loss(problem_mode: str) → torch.nn.Module: Build the default registered loss for a problem mode.

DeepMTP.runtime.build_default_loss_input_transform(problem_mode: str) → torch.nn.Module: Build the transform applied to raw model scores before the loss.

DeepMTP.runtime.build_default_output_head(problem_mode: str, classification_mode: str | None = None) → torch.nn.Module: Build the default registered output head for a problem mode.

DeepMTP.runtime.build_default_task_components(problem_mode: str) → TaskComponents: Build the behavior-preserving task components for a problem mode.

DeepMTP.runtime.build_evaluation_progress_observer(enabled: bool) → EvaluationProgressObserver: Select the default detailed-evaluation observer.

DeepMTP.runtime.build_loss(name: str) → torch.nn.Module: Build a registered loss module by name.

DeepMTP.runtime.build_loss_input_transform(loss_name: str) → torch.nn.Module: Build the transform applied before a selected registered loss.

DeepMTP.runtime.build_output_head(name: str) → torch.nn.Module: Build a registered output-head module by name.

DeepMTP.runtime.build_progress_observer(verbose: bool) → ProgressObserver: Select the default observer for the configured verbosity.

DeepMTP.runtime.build_reporter(config: DeepMTPConfig | Mapping[str, Any], experiment_dir: str | Path) → ExperimentReporter: Build configured reporters from typed or legacy configuration.

DeepMTP.runtime.build_task_components(problem_mode: str, loss_name: str | None = None, classification_mode: str | None = None) → TaskComponents: Build compatible loss and output modules for one problem mode.

DeepMTP.runtime.download_wandb_checkpoint(artifact_reference: str, *, download_dir: str | PathLike[str] | None = None) → WandBCheckpoint: Download and validate one DeepMTP model artifact through the Public API.

DeepMTP.runtime.load_foundation_adapter_checkpoint(checkpoint_path: str | os.PathLike[str], *, map_location: str | torch.device = 'cpu') → FoundationAdapterCheckpoint: Read and validate a compact foundation adapter checkpoint.

DeepMTP.runtime.log_reporter_model_artifact(reporter: ExperimentReporter, artifact: ReporterModelArtifact) → Mapping[str, Any] | None

DeepMTP.runtime.log_reporter_predictions(reporter: ExperimentReporter, report: ReporterPredictionTable) → None

DeepMTP.runtime.log_reporter_summary(reporter: ExperimentReporter, values: Mapping[str, Any]) → None

DeepMTP.runtime.mark_reporter_failed(reporter: ExperimentReporter, error: BaseException) → None

DeepMTP.runtime.reporter_runtime_metadata(reporter: ExperimentReporter) → dict[str, Any]

DeepMTP.runtime.restore_foundation_adapter_checkpoint(model: torch.nn.Module, checkpoint: FoundationAdapterCheckpoint) → None: Restore validated task and LoRA state into a newly built DeepMTP model.

DeepMTP.runtime.save_foundation_adapter_checkpoint(checkpoint_path: str | PathLike[str], *, model: torch.nn.Module, config: DeepMTPConfig | Mapping[str, Any], metadata: Mapping[str, Any]) → FoundationAdapterExport: Save task weights and native PEFT adapters without base-model weights.

DeepMTP.runtime.update_reporter_config(reporter: ExperimentReporter, config: Mapping[str, Any]) → None

DeepMTP.runtime.validate_checkpoint_metadata(value: object) → dict[str, Any]: Validate a checkpoint manifest without accepting future formats.