DeepMTP package

Subpackages

Submodules

DeepMTP.architecture module

Semantic and terminal-friendly descriptions of DeepMTP architectures.

class DeepMTP.architecture.ArchitectureBranch(name: Literal['instance', 'target'], architecture: str, input_dimension: int | None, output_dimension: int, details: tuple[str, ...] = (), components: tuple[ArchitectureComponent, ...] = ())

Bases: object

A normalized instance or target branch description.

architecture: str
components: tuple[ArchitectureComponent, ...] = ()
details: tuple[str, ...] = ()
input_dimension: int | None
name: Literal['instance', 'target']
output_dimension: int
class DeepMTP.architecture.ArchitectureComponent(name: str, architecture: str, input_dimension: int, output_dimension: int, optional: bool, details: tuple[str, ...] = ())

Bases: object

One independently encoded component inside a composite branch.

architecture: str
details: tuple[str, ...] = ()
input_dimension: int
name: str
optional: bool
output_dimension: int
class DeepMTP.architecture.ArchitectureDescription(validation_setting: str, task: str, loss: str, instance_branch: ArchitectureBranch, target_branch: ArchitectureBranch, fusion: ArchitectureFusion, parameters: ParameterStatistics | None = None, warnings: tuple[str, ...] = ())

Bases: object

Renderer-independent representation of a configured DeepMTP model.

fusion: ArchitectureFusion
instance_branch: ArchitectureBranch
loss: str
parameters: ParameterStatistics | None = None
target_branch: ArchitectureBranch
task: str
to_dict() → dict[str, Any]

Return a JSON-compatible representation for tooling and exports.

validation_setting: str
warnings: tuple[str, ...] = ()
class DeepMTP.architecture.ArchitectureFusion(architecture: str, input_dimension: int, output_dimension: int, details: tuple[str, ...] = ())

Bases: object

The operation that combines encoded instance and target branches.

architecture: str
details: tuple[str, ...] = ()
input_dimension: int
output_dimension: int
class DeepMTP.architecture.ParameterStatistics(total: int, trainable: int, frozen: int)

Bases: object

Exact aggregate parameter counts from an instantiated model.

frozen: int
total: int
trainable: int
property trainable_fraction: float

Fraction of parameters whose gradients are enabled.

DeepMTP.architecture.describe_architecture(config: DeepMTPConfig | Mapping[str, Any], *, model: object | None = None) → ArchitectureDescription

Describe a validated configuration without instantiating its model.

DeepMTP.architecture.export_architecture_svg(config: DeepMTPConfig | Mapping[str, Any] | ArchitectureDescription, destination: str | PathLike[str], *, model: object | None = None, width: int = 140) → Path

Export a Rich architecture rendering as an SVG file.

DeepMTP.architecture.format_architecture(config: DeepMTPConfig | Mapping[str, Any] | ArchitectureDescription, *, model: object | None = None, style: Literal['plain', 'rich'] = 'plain', width: int = 120, color: bool = False) → str

Render an architecture to a string without writing to the terminal.

DeepMTP.architecture.print_architecture(config: DeepMTPConfig | Mapping[str, Any] | ArchitectureDescription, *, model: object | None = None, style: Literal['auto', 'plain', 'rich'] = 'auto', width: int | None = None, file: TextIO | None = None) → ArchitectureDescription

Print a responsive architecture diagram and return its description.

DeepMTP.main module

class DeepMTP.main.DeepMTP(config: DeepMTPConfig | Mapping[str, Any], instance_branch_model: Callable[[Mapping[str, Any]], torch.nn.Module] | None = None, target_branch_model: Callable[[Mapping[str, Any]], torch.nn.Module] | None = None, checkpoint_dir: str | PathLike[str] | None = None, reporter: ExperimentReporter | None = None, progress: ProgressObserver | None = None)

Bases: object

Implements the training and inference logic of the DeepMTP framework.

evaluator

the user can specify wether or not they want to use early stopping. In both cases we create the early stopping object so that we can track the best epoch, model etc.

classmethod from_wandb_artifact(artifact_reference: str, config: DeepMTPConfig | Mapping[str, Any] | None = None, *, download_dir: str | PathLike[str] | None = None, instance_branch_model: Callable[[Mapping[str, Any]], torch.nn.Module] | None = None, target_branch_model: Callable[[Mapping[str, Any]], torch.nn.Module] | None = None, reporter: ExperimentReporter | None = None, progress: ProgressObserver | None = None) → DeepMTP

Download a W&B model artifact and restore its DeepMTP checkpoint.

inference(model: torch.nn.Module, dataloader: Iterable[MTPBatch | Mapping[str, Any]], mode: str, epoch: int = 0, return_predictions: Literal[False] = False, verbose: bool = False) → dict[str, float]
inference(model: torch.nn.Module, dataloader: Iterable[MTPBatch | Mapping[str, Any]], mode: str, epoch: int = 0, return_predictions: Literal[True] = True, verbose: bool = False) → tuple[dict[str, float], DataFrame]
predict(data: Mapping[str, Any], return_predictions: Literal[False] = False, verbose: bool = False) → dict[str, float]
predict(data: Mapping[str, Any], return_predictions: Literal[True] = True, verbose: bool = False) → tuple[dict[str, float], DataFrame]
property preprocessing_state: dict[str, Any] | None

Return a copy of saved data-preparation metadata, when available.

save_model(verbose: bool = False, *, artifact_files: tuple[Path, ...] = ()) → None
train(train_data: Mapping[str, Any], val_data: Mapping[str, Any], test_data: Mapping[str, Any], verbose: bool = False) → dict[str, float]
transform_dense_features(values: object, *, branch: Literal['instance', 'target']) → Any

Replay the saved dense scaler for one feature axis.

DeepMTP.tests module

Module contents

class DeepMTP.ArchitectureBranch(name: Literal['instance', 'target'], architecture: str, input_dimension: int | None, output_dimension: int, details: tuple[str, ...] = (), components: tuple[ArchitectureComponent, ...] = ())

Bases: object

A normalized instance or target branch description.

architecture: str
components: tuple[ArchitectureComponent, ...] = ()
details: tuple[str, ...] = ()
input_dimension: int | None
name: Literal['instance', 'target']
output_dimension: int
class DeepMTP.ArchitectureComponent(name: str, architecture: str, input_dimension: int, output_dimension: int, optional: bool, details: tuple[str, ...] = ())

Bases: object

One independently encoded component inside a composite branch.

architecture: str
details: tuple[str, ...] = ()
input_dimension: int
name: str
optional: bool
output_dimension: int
class DeepMTP.ArchitectureDescription(validation_setting: str, task: str, loss: str, instance_branch: ArchitectureBranch, target_branch: ArchitectureBranch, fusion: ArchitectureFusion, parameters: ParameterStatistics | None = None, warnings: tuple[str, ...] = ())

Bases: object

Renderer-independent representation of a configured DeepMTP model.

fusion: ArchitectureFusion
instance_branch: ArchitectureBranch
loss: str
parameters: ParameterStatistics | None = None
target_branch: ArchitectureBranch
task: str
to_dict() → dict[str, Any]

Return a JSON-compatible representation for tooling and exports.

validation_setting: str
warnings: tuple[str, ...] = ()
class DeepMTP.ArchitectureFusion(architecture: str, input_dimension: int, output_dimension: int, details: tuple[str, ...] = ())

Bases: object

The operation that combines encoded instance and target branches.

architecture: str
details: tuple[str, ...] = ()
input_dimension: int
output_dimension: int
class DeepMTP.CompositeEncoder(*args: Any, **kwargs: Any)

Bases: Module

Fuse standardized outputs from named component encoders.

forward(values: CompositeInput) → torch.Tensor
input_kind: ClassVar[BranchInputKind] = 'composite'
class DeepMTP.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
exception DeepMTP.ConfigDeprecationWarning

Bases: FutureWarning

Warns that a legacy configuration option has no runtime effect.

exception DeepMTP.ConfigNormalizationWarning

Bases: UserWarning

Warns that compatibility configuration input was normalized.

exception DeepMTP.ConfigValidationError

Bases: ValueError

Raised when a DeepMTP configuration is internally inconsistent.

class DeepMTP.ConsoleDataProgressObserver(print_mode: Literal['basic', 'dev'] = 'basic')

Bases: object

Render data progress using the historical basic or developer format.

on_event(event: DataProgressEvent) → None
class DeepMTP.ConsoleEvaluationProgressObserver

Bases: object

Render the historical metric-evaluation console output.

on_event(event: EvaluationProgressEvent) → None
class DeepMTP.ConsoleHPOProgressObserver

Bases: object

Render optimizer events in a terminal.

on_event(event: HPOProgressEvent) → None
class DeepMTP.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.DataLoaderFactory(config: DeepMTPConfig | Mapping[str, Any], device: torch.device | str)

Bases: object

Build configured dataloaders without trainer-specific orchestration.

for_prediction(data: Mapping[str, Any]) → torch.utils.data.DataLoader

Build a deterministic inference dataloader.

for_training(train_data: Mapping[str, Any], validation_data: Mapping[str, Any] | None, test_data: Mapping[str, Any] | None) → TrainingDataLoaders

Build the train, validation, and test dataloaders.

class DeepMTP.DataPreparationState(validation_setting: str, split_method: str, split_ratio: tuple[tuple[str, float], ...], shuffle: bool, seed: int | None, dense: DensePreprocessingState)

Bases: object

Split provenance and fitted dense preprocessing from data_process.

dense: DensePreprocessingState
classmethod from_config(value: object) → DataPreparationState

Validate a saved data-preparation state mapping.

seed: int | None
shuffle: bool
split_method: str
split_ratio: tuple[tuple[str, float], ...]
to_dict() → dict[str, Any]

Return JSON-compatible checkpoint metadata.

validation_setting: str
class DeepMTP.DataProgressEvent(message: str, kind: Literal['message', 'operation_started', 'operation_completed'] = 'message', subject: str | None = None, level: Literal['info', 'warning', 'error'] = 'info', end: str = '\n', value: object | None = None)

Bases: object

One user-facing data preparation or validation message.

end: str = '\n'
kind: Literal['message', 'operation_started', 'operation_completed'] = 'message'
level: Literal['info', 'warning', 'error'] = 'info'
message: str
subject: str | None = None
value: object | None = None
class DeepMTP.DeepMTP(config: DeepMTPConfig | Mapping[str, Any], instance_branch_model: Callable[[Mapping[str, Any]], torch.nn.Module] | None = None, target_branch_model: Callable[[Mapping[str, Any]], torch.nn.Module] | None = None, checkpoint_dir: str | PathLike[str] | None = None, reporter: ExperimentReporter | None = None, progress: ProgressObserver | None = None)

Bases: object

Implements the training and inference logic of the DeepMTP framework.

checkpoint_dict: dict[str, Any] | None
checkpoint_metadata: dict[str, Any] | None
evaluator

the user can specify wether or not they want to use early stopping. In both cases we create the early stopping object so that we can track the best epoch, model etc.

classmethod from_wandb_artifact(artifact_reference: str, config: DeepMTPConfig | Mapping[str, Any] | None = None, *, download_dir: str | PathLike[str] | None = None, instance_branch_model: Callable[[Mapping[str, Any]], torch.nn.Module] | None = None, target_branch_model: Callable[[Mapping[str, Any]], torch.nn.Module] | None = None, reporter: ExperimentReporter | None = None, progress: ProgressObserver | None = None) → DeepMTP

Download a W&B model artifact and restore its DeepMTP checkpoint.

inference(model: torch.nn.Module, dataloader: Iterable[MTPBatch | Mapping[str, Any]], mode: str, epoch: int = 0, return_predictions: bool = False, verbose: bool = False) → dict[str, float] | tuple[dict[str, float], DataFrame]
predict(data: Mapping[str, Any], return_predictions: bool = False, verbose: bool = False) → dict[str, float] | tuple[dict[str, float], DataFrame]
property preprocessing_state: dict[str, Any] | None

Return a copy of saved data-preparation metadata, when available.

reporter: ExperimentReporter | None
save_model(verbose: bool = False, *, artifact_files: tuple[Path, ...] = ()) → None
train(train_data: Mapping[str, Any], val_data: Mapping[str, Any], test_data: Mapping[str, Any], verbose: bool = False) → dict[str, float]
transform_dense_features(values: object, *, branch: Literal['instance', 'target']) → Any

Replay the saved dense scaler for one feature axis.

wandb_artifact: Any
class DeepMTP.DeepMTPConfig(validation_setting: ~typing.Literal['A', 'B', 'C', 'D'] | None = None, general_architecture_version: ~typing.Literal['dot_product', 'mlp', 'kronecker'] | str = 'dot_product', problem_mode: ~typing.Literal['classification', 'regression'] | None = None, loss: ~typing.Literal['binary_cross_entropy', 'binary_cross_entropy_with_logits', 'cross_entropy', 'mean_squared_error', 'mean_absolute_error', 'huber'] | None = None, learning_rate: float = 0.001, decay: float = 0.0, batch_norm: bool = False, dropout_rate: float = 0.0, dropout_rate_instance_branch: float = 0.0, dropout_rate_target_branch: float = 0.0, momentum: float = 0.9, weighted_loss: bool = False, compute_mode: str = 'cuda:0', num_workers: int = 1, train_batchsize: int = 512, val_batchsize: int = 512, num_epochs: int = 100, random_seed: int | None = 2, metrics: list[str] = <factory>, metrics_average: list[str] = <factory>, top_k: int | None = None, metric_to_optimize_early_stopping: str = 'loss', metric_to_optimize_best_epoch_selection: str = 'loss', patience: int = 10, delta: float = 0.0, evaluate_train: bool = False, evaluate_val: bool = False, eval_every_n_epochs: int = 10, use_early_stopping: bool = True, verbose: bool = False, results_verbose: bool = False, eval_instance_verbose: bool = False, eval_target_verbose: bool = False, return_results_per_target: bool = False, use_tensorboard_logger: bool = False, wandb_project_name: str | None = None, wandb_project_entity: str | None = None, wandb_mode: ~typing.Literal['online', 'offline', 'disabled'] | str | None = None, wandb_run_name: str | None = None, wandb_group: str | None = None, wandb_job_type: str = 'train', wandb_tags: list[str] = <factory>, wandb_notes: str | None = None, wandb_watch: ~typing.Literal['gradients', 'parameters', 'all'] | str | None = 'gradients', wandb_watch_log_freq: int = 1000, wandb_log_graph: bool = False, wandb_log_code: bool = False, wandb_log_model_artifact: bool = False, wandb_model_artifact_name: str | None = None, wandb_model_artifact_aliases: list[str] = <factory>, wandb_input_artifacts: list[str] = <factory>, wandb_registry_name: str | None = None, wandb_registry_collection: str | None = None, wandb_log_predictions: bool = False, wandb_prediction_table_max_rows: int = 1000, results_path: str = './results/', experiment_name: str | None = None, experiment_dir: str | None = None, save_model: bool = True, checkpoint_format: ~typing.Literal['full', 'foundation_adapter'] | str = 'full', data_preparation_state: dict[str, ~typing.Any] | None = None, instance_branch_architecture: ~typing.Literal['MLP', 'SPARSE', 'CONV', 'COMPOSITE', 'CUSTOM', 'EMBEDDING', 'GRAPH', 'SEQUENCE', 'TABULAR'] | None = None, use_instance_features: bool = False, instance_branch_input_dim: int | None = None, instance_branch_nodes_reducing_factor: int = 2, instance_branch_nodes_per_layer: list[int] | int = <factory>, instance_branch_layers: int | None = None, instance_branch_conv_architecture: str = 'resnet', instance_branch_conv_architecture_version: str = 'resnet101', instance_branch_conv_architecture_dense_layers: int = 1, instance_branch_conv_architecture_last_layer_trained: str = 'last', instance_branch_conv_pretrained: bool = True, instance_train_transforms: ~typing.Any = None, instance_inference_transforms: ~typing.Any = None, instance_branch_tabular_schema: dict[str, ~typing.Any] | None = None, instance_tabular_preprocessing_state: dict[str, ~typing.Any] | None = None, instance_branch_composite_components: dict[str, dict[str, ~typing.Any]] | None = None, instance_branch_composite_fusion: ~typing.Literal['attention', 'concat', 'gated'] | str = 'concat', instance_branch_composite_attention_dim: int = 64, instance_branch_composite_attention_num_heads: int = 4, instance_branch_composite_modality_dropout: float = 0.0, instance_branch_sequence_encoder: str = 'gru', instance_branch_sequence_embedding_dim: int = 64, instance_branch_sequence_output_dim: int = 64, instance_branch_sequence_conv_kernel_size: int = 3, instance_branch_sequence_transformer_num_heads: int = 4, instance_branch_sequence_transformer_feedforward_dim: int = 256, instance_branch_sequence_num_layers: int = 1, instance_branch_sequence_padding_idx: int = 0, instance_branch_graph_edge_dim: int | None = None, instance_branch_graph_hidden_dim: int = 64, instance_branch_graph_output_dim: int = 64, instance_branch_graph_num_layers: int = 2, instance_branch_graph_pooling: str = 'mean', instance_branch_graph_use_edge_features: bool = True, target_branch_architecture: ~typing.Literal['MLP', 'SPARSE', 'CONV', 'COMPOSITE', 'CUSTOM', 'EMBEDDING', 'GRAPH', 'SEQUENCE', 'TABULAR'] | None = None, use_target_features: bool = False, target_branch_input_dim: int | None = None, target_branch_nodes_reducing_factor: int = 2, target_branch_nodes_per_layer: list[int] | int = <factory>, target_branch_layers: int | None = None, target_branch_conv_architecture: str = 'resnet', target_branch_conv_architecture_version: str = 'resnet101', target_branch_conv_architecture_dense_layers: int = 1, target_branch_conv_architecture_last_layer_trained: str = 'last', target_branch_conv_pretrained: bool = True, target_train_transforms: ~typing.Any = None, target_inference_transforms: ~typing.Any = None, target_branch_tabular_schema: dict[str, ~typing.Any] | None = None, target_tabular_preprocessing_state: dict[str, ~typing.Any] | None = None, target_branch_composite_components: dict[str, dict[str, ~typing.Any]] | None = None, target_branch_composite_fusion: ~typing.Literal['attention', 'concat', 'gated'] | str = 'concat', target_branch_composite_attention_dim: int = 64, target_branch_composite_attention_num_heads: int = 4, target_branch_composite_modality_dropout: float = 0.0, target_branch_sequence_encoder: str = 'gru', target_branch_sequence_embedding_dim: int = 64, target_branch_sequence_output_dim: int = 64, target_branch_sequence_conv_kernel_size: int = 3, target_branch_sequence_transformer_num_heads: int = 4, target_branch_sequence_transformer_feedforward_dim: int = 256, target_branch_sequence_num_layers: int = 1, target_branch_sequence_padding_idx: int = 0, target_branch_graph_edge_dim: int | None = None, target_branch_graph_hidden_dim: int = 64, target_branch_graph_output_dim: int = 64, target_branch_graph_num_layers: int = 2, target_branch_graph_pooling: str = 'mean', target_branch_graph_use_edge_features: bool = True, comb_mlp_nodes_reducing_factor: int = 2, comb_mlp_nodes_per_layer: list[int] | int = <factory>, comb_mlp_layers: int | None = None, embedding_size: int = 100, load_pretrained_model: bool = False, pretrained_model_path: str = '', running_hpo: bool = False, hpo_results_path: str = './', additional_info: dict[str, ~typing.Any] = <factory>, classification_mode: ~typing.Literal['binary', 'multiclass'] | None = None, num_classes: int | None = None, multiclass_average: ~typing.Literal['micro', 'macro', 'weighted'] | None = None, gradient_accumulation_steps: int = 1, gradient_clip_norm: float | None = None, instance_branch_foundation_model: dict[str, ~typing.Any] | None = None, target_branch_foundation_model: dict[str, ~typing.Any] | None = None)

Bases: object

Typed representation of a fully materialized DeepMTP configuration.

generate_config remains the compatibility entry point and continues to return a dictionary. New code can construct this class directly, while old dictionaries can be validated with from_mapping().

additional_info: dict[str, Any]
batch_norm: bool = False
checkpoint_format: Literal['full', 'foundation_adapter'] | str = 'full'
classification_mode: Literal['binary', 'multiclass'] | None = None
comb_mlp_layers: int | None = None
comb_mlp_nodes_per_layer: list[int] | int
comb_mlp_nodes_reducing_factor: int = 2
compute_mode: str = 'cuda:0'
data_preparation_state: dict[str, Any] | None = None
decay: float = 0.0
delta: float = 0.0
dropout_rate: float = 0.0
dropout_rate_instance_branch: float = 0.0
dropout_rate_target_branch: float = 0.0
embedding_size: int = 100
eval_every_n_epochs: int = 10
eval_instance_verbose: bool = False
eval_target_verbose: bool = False
evaluate_train: bool = False
evaluate_val: bool = False
experiment_dir: str | None = None
experiment_name: str | None = None
classmethod from_mapping(config: Mapping[str, Any] | DeepMTPConfig) → DeepMTPConfig

Create a validated typed config without discarding extension keys.

general_architecture_version: Literal['dot_product', 'mlp', 'kronecker'] | str = 'dot_product'
gradient_accumulation_steps: int = 1
gradient_clip_norm: float | None = None
hpo_results_path: str = './'
instance_branch_architecture: Literal['MLP', 'SPARSE', 'CONV', 'COMPOSITE', 'CUSTOM', 'EMBEDDING', 'GRAPH', 'SEQUENCE', 'TABULAR'] | None = None
instance_branch_composite_attention_dim: int = 64
instance_branch_composite_attention_num_heads: int = 4
instance_branch_composite_components: dict[str, dict[str, Any]] | None = None
instance_branch_composite_fusion: Literal['attention', 'concat', 'gated'] | str = 'concat'
instance_branch_composite_modality_dropout: float = 0.0
instance_branch_conv_architecture: str = 'resnet'
instance_branch_conv_architecture_dense_layers: int = 1
instance_branch_conv_architecture_last_layer_trained: str = 'last'
instance_branch_conv_architecture_version: str = 'resnet101'
instance_branch_conv_pretrained: bool = True
instance_branch_foundation_model: dict[str, Any] | None = None
instance_branch_graph_edge_dim: int | None = None
instance_branch_graph_hidden_dim: int = 64
instance_branch_graph_num_layers: int = 2
instance_branch_graph_output_dim: int = 64
instance_branch_graph_pooling: str = 'mean'
instance_branch_graph_use_edge_features: bool = True
instance_branch_input_dim: int | None = None
instance_branch_layers: int | None = None
instance_branch_nodes_per_layer: list[int] | int
instance_branch_nodes_reducing_factor: int = 2
instance_branch_sequence_conv_kernel_size: int = 3
instance_branch_sequence_embedding_dim: int = 64
instance_branch_sequence_encoder: str = 'gru'
instance_branch_sequence_num_layers: int = 1
instance_branch_sequence_output_dim: int = 64
instance_branch_sequence_padding_idx: int = 0
instance_branch_sequence_transformer_feedforward_dim: int = 256
instance_branch_sequence_transformer_num_heads: int = 4
instance_branch_tabular_schema: dict[str, Any] | None = None
instance_inference_transforms: Any = None
instance_tabular_preprocessing_state: dict[str, Any] | None = None
instance_train_transforms: Any = None
learning_rate: float = 0.001
load_pretrained_model: bool = False
loss: Literal['binary_cross_entropy', 'binary_cross_entropy_with_logits', 'cross_entropy', 'mean_squared_error', 'mean_absolute_error', 'huber'] | None = None
metric_to_optimize_best_epoch_selection: str = 'loss'
metric_to_optimize_early_stopping: str = 'loss'
metrics: list[str]
metrics_average: list[str]
momentum: float = 0.9
multiclass_average: Literal['micro', 'macro', 'weighted'] | None = None
num_classes: int | None = None
num_epochs: int = 100
num_workers: int = 1
property output_dimension: int

Number of raw scores produced for each interaction.

patience: int = 10
pretrained_model_path: str = ''
problem_mode: Literal['classification', 'regression'] | None = None
random_seed: int | None = 2
results_path: str = './results/'
results_verbose: bool = False
return_results_per_target: bool = False
running_hpo: bool = False
save_model: bool = True
set_runtime_value(key: str, value: Any) → None

Record metadata added while an experiment is running.

target_branch_architecture: Literal['MLP', 'SPARSE', 'CONV', 'COMPOSITE', 'CUSTOM', 'EMBEDDING', 'GRAPH', 'SEQUENCE', 'TABULAR'] | None = None
target_branch_composite_attention_dim: int = 64
target_branch_composite_attention_num_heads: int = 4
target_branch_composite_components: dict[str, dict[str, Any]] | None = None
target_branch_composite_fusion: Literal['attention', 'concat', 'gated'] | str = 'concat'
target_branch_composite_modality_dropout: float = 0.0
target_branch_conv_architecture: str = 'resnet'
target_branch_conv_architecture_dense_layers: int = 1
target_branch_conv_architecture_last_layer_trained: str = 'last'
target_branch_conv_architecture_version: str = 'resnet101'
target_branch_conv_pretrained: bool = True
target_branch_foundation_model: dict[str, Any] | None = None
target_branch_graph_edge_dim: int | None = None
target_branch_graph_hidden_dim: int = 64
target_branch_graph_num_layers: int = 2
target_branch_graph_output_dim: int = 64
target_branch_graph_pooling: str = 'mean'
target_branch_graph_use_edge_features: bool = True
target_branch_input_dim: int | None = None
target_branch_layers: int | None = None
target_branch_nodes_per_layer: list[int] | int
target_branch_nodes_reducing_factor: int = 2
target_branch_sequence_conv_kernel_size: int = 3
target_branch_sequence_embedding_dim: int = 64
target_branch_sequence_encoder: str = 'gru'
target_branch_sequence_num_layers: int = 1
target_branch_sequence_output_dim: int = 64
target_branch_sequence_padding_idx: int = 0
target_branch_sequence_transformer_feedforward_dim: int = 256
target_branch_sequence_transformer_num_heads: int = 4
target_branch_tabular_schema: dict[str, Any] | None = None
target_inference_transforms: Any = None
target_tabular_preprocessing_state: dict[str, Any] | None = None
target_train_transforms: Any = None
to_dict(*, include_defaults: bool = False) → dict[str, Any]

Return a compatibility dictionary, including extension metadata.

include_defaults materializes every normalized public field for internal runtime consumers. The default preserves the historical round-trip behavior for partial legacy mappings.

top_k: int | None = None
train_batchsize: int = 512
use_early_stopping: bool = True
use_instance_features: bool = False
use_target_features: bool = False
use_tensorboard_logger: bool = False
val_batchsize: int = 512
validate() → None

Validate values needed by model construction and training.

validation_setting: Literal['A', 'B', 'C', 'D'] | None = None
verbose: bool = False
wandb_group: str | None = None
wandb_input_artifacts: list[str]
wandb_job_type: str = 'train'
wandb_log_code: bool = False
wandb_log_graph: bool = False
wandb_log_model_artifact: bool = False
wandb_log_predictions: bool = False
wandb_mode: Literal['online', 'offline', 'disabled'] | str | None = None
wandb_model_artifact_aliases: list[str]
wandb_model_artifact_name: str | None = None
wandb_notes: str | None = None
wandb_prediction_table_max_rows: int = 1000
wandb_project_entity: str | None = None
wandb_project_name: str | None = None
wandb_registry_collection: str | None = None
wandb_registry_name: str | None = None
wandb_run_name: str | None = None
wandb_tags: list[str]
wandb_watch: Literal['gradients', 'parameters', 'all'] | str | None = 'gradients'
wandb_watch_log_freq: int = 1000
weighted_loss: bool = False
class DeepMTP.DensePreprocessingState(instance: DenseScalingState | None = None, target: DenseScalingState | None = None)

Bases: object

Dense scaling state for the instance and target feature axes.

for_branch(branch: str) → DenseScalingState | None

Return the scaler state for one entity axis.

classmethod from_config(value: object) → DensePreprocessingState

Validate a combined dense-preprocessing state mapping.

instance: DenseScalingState | None = None
target: DenseScalingState | None = None
to_dict() → dict[str, Any]

Return JSON-compatible checkpoint metadata.

class DeepMTP.DenseScalingState(method: Literal['MinMax', 'Standard'], feature_count: int, offsets: tuple[float, ...], scales: tuple[float, ...], data_minimums: tuple[float, ...] = (), data_maximums: tuple[float, ...] = (), feature_range: tuple[float, float] = (0.0, 1.0))

Bases: object

Parameters needed to replay one fitted dense-feature transformation.

data_maximums: tuple[float, ...] = ()
data_minimums: tuple[float, ...] = ()
feature_count: int
feature_range: tuple[float, float] = (0.0, 1.0)
classmethod from_config(value: object) → DenseScalingState

Validate and restore a JSON-compatible scaling-state mapping.

classmethod from_fitted_scaler(scaler: object, *, method: Literal['MinMax', 'Standard']) → DenseScalingState

Capture explicit numeric state from a fitted scikit-learn scaler.

method: Literal['MinMax', 'Standard']
offsets: tuple[float, ...]
scales: tuple[float, ...]
to_dict() → dict[str, Any]

Return JSON-compatible checkpoint metadata.

transform(values: object) → ndarray

Transform one vector or matrix using the saved training statistics.

class DeepMTP.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.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.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.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.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.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.FoundationCacheSpec(directory: str | None = None, memory_capacity: int = 1024)

Bases: object

Serializable settings for pooled frozen-backbone output caching.

Supplying this specification enables caching. directory=None creates a process-local memory cache. A directory adds persistent entries that can be shared by later runs with exactly the same foundation-model identity.

directory: str | None = None
classmethod from_config(value: FoundationCacheSpec | Mapping[str, Any]) → FoundationCacheSpec

Validate a public mapping as cache settings.

memory_capacity: int = 1024
to_dict() → dict[str, Any]

Return JSON-compatible cache settings.

class DeepMTP.FoundationLoRASpec(target_modules: tuple[str, ...], rank: int = 8, alpha: int = 16, dropout: float = 0.0)

Bases: object

Serializable LoRA settings supported by the foundation adapter.

alpha: int = 16
dropout: float = 0.0
classmethod from_config(value: FoundationLoRASpec | Mapping[str, Any]) → FoundationLoRASpec

Validate a public mapping as a LoRA specification.

rank: int = 8
target_modules: tuple[str, ...]
to_dict() → dict[str, Any]

Return JSON-compatible LoRA settings.

class DeepMTP.FoundationModelSpec(model_name: str, revision: str | None = None, tokenizer_name: str | None = None, tokenizer_revision: str | None = None, pooling: Literal['cls', 'masked_mean', 'pooler'] | str = 'masked_mean', train_mode: Literal['frozen', 'full', 'lora'] | str = 'frozen', trust_remote_code: bool = False, local_files_only: bool = False, max_length: int | None = None, truncation: bool = False, exclude_special_tokens_from_pooling: bool = True, pooling_excluded_token_ids: tuple[int, ...] = (), resolved_model_revision: str | None = None, resolved_tokenizer_revision: str | None = None, lora: FoundationLoRASpec | Mapping[str, Any] | None = None, cache: FoundationCacheSpec | Mapping[str, Any] | None = None)

Bases: object

Serializable model, tokenizer, pooling, and training settings.

cache: FoundationCacheSpec | Mapping[str, Any] | None = None
exclude_special_tokens_from_pooling: bool = True
classmethod from_config(value: FoundationModelSpec | Mapping[str, Any] | None) → FoundationModelSpec

Validate a public mapping as a foundation-model specification.

local_files_only: bool = False
lora: FoundationLoRASpec | Mapping[str, Any] | None = None
max_length: int | None = None
property model_load_revision: str | None

Immutable resolved revision when known, otherwise the requested one.

model_name: str
pooling: Literal['cls', 'masked_mean', 'pooler'] | str = 'masked_mean'
pooling_excluded_token_ids: tuple[int, ...] = ()
resolved_model_revision: str | None = None
resolved_tokenizer_revision: str | None = None
revision: str | None = None
to_dict() → dict[str, Any]

Return JSON-compatible settings for config, W&B, and checkpoints.

property tokenizer_load_revision: str | None

Immutable tokenizer revision when known, otherwise the requested one.

tokenizer_name: str | None = None
tokenizer_revision: str | None = None
train_mode: Literal['frozen', 'full', 'lora'] | str = 'frozen'
truncation: bool = False
trust_remote_code: bool = False
class DeepMTP.GraphEncoder(*args: Any, **kwargs: Any)

Bases: Module

Encode homogeneous graphs with GIN/GINE message passing and pooling.

forward(values: GraphInput) → torch.Tensor
input_kind: ClassVar[BranchInputKind] = 'graph'
class DeepMTP.HPOProgressEvent(kind: Literal['schedule_calculated', 'worker_configured', 'optimizer_started', 'bracket_started', 'iteration_started', 'configuration_started', 'configuration_completed', 'optimizer_completed'], current: int | None = None, total: int | None = None, bracket: int | None = None, iteration: int | None = None, config: Any = None, score: float | None = None, smax: int | None = None, eta: int | None = None, max_budget: int | None = None, total_budget: int | None = None)

Bases: object

One optimizer lifecycle event.

bracket: int | None = None
config: Any = None
current: int | None = None
eta: int | None = None
iteration: int | None = None
kind: Literal['schedule_calculated', 'worker_configured', 'optimizer_started', 'bracket_started', 'iteration_started', 'configuration_started', 'configuration_completed', 'optimizer_completed']
max_budget: int | None = None
score: float | None = None
smax: int | None = None
total: int | None = None
total_budget: int | None = None
class DeepMTP.HPOProgressObserver(*args, **kwargs)

Bases: Protocol

Receives optimizer lifecycle events.

on_event(event: HPOProgressEvent) → None

Handle an optimizer event.

class DeepMTP.HuggingFaceSequenceEncoder(*args: Any, **kwargs: Any)

Bases: Module

Pool and project one Hugging Face encoder as a DeepMTP branch.

property cache_stats: dict[str, Any]

JSON-compatible cache statistics for diagnostics and reporting.

clear_cache(*, memory: bool = True, disk: bool = False) → None

Clear this encoder’s derived memory and optionally disk entries.

forward(values: SequenceInput) → torch.Tensor
input_kind: ClassVar[BranchInputKind] = 'sequence'
train(mode: bool = True) → HuggingFaceSequenceEncoder

Keep a frozen backbone deterministic while training its projection.

class DeepMTP.IDEmbedding(*args: Any, **kwargs: Any)

Bases: Module

Map zero-based entity IDs to trainable dense representations.

forward(entity_ids: torch.Tensor) → torch.Tensor
input_kind: ClassVar[BranchInputKind] = 'id'
class DeepMTP.ModelBundle(instance_branch: torch.nn.Module, target_branch: torch.nn.Module, model: torch.nn.Module)

Bases: object

The two branches and their combined model.

instance_branch: torch.nn.Module
model: torch.nn.Module
target_branch: torch.nn.Module
class DeepMTP.ModelFactory(config: DeepMTPConfig | Mapping[str, Any])

Bases: object

Construct a complete DeepMTP model from validated configuration.

build(*, instance_branch_factory: Callable[[Mapping[str, Any]], torch.nn.Module] | None = None, target_branch_factory: Callable[[Mapping[str, Any]], torch.nn.Module] | None = None) → ModelBundle

Build branches and their configured combination model.

class DeepMTP.NullDataProgressObserver

Bases: object

No-op observer used when data progress is disabled.

on_event(event: DataProgressEvent) → None
class DeepMTP.NullEvaluationProgressObserver

Bases: object

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

on_event(event: EvaluationProgressEvent) → None
class DeepMTP.NullHPOProgressObserver

Bases: object

No-op optimizer progress observer.

on_event(event: HPOProgressEvent) → None
class DeepMTP.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.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.OptionalIntegrationError

Bases: ImportError

Raised when a requested optional reporting dependency is unavailable.

class DeepMTP.ParameterStatistics(total: int, trainable: int, frozen: int)

Bases: object

Exact aggregate parameter counts from an instantiated model.

frozen: int
total: int
trainable: int
property trainable_fraction: float

Fraction of parameters whose gradients are enabled.

class DeepMTP.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.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.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.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.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.RunSummaries(train: str, validation: str, test: str)

Bases: object

Rendered train, validation, and test histories.

test: str
train: str
validation: str
class DeepMTP.SequenceConv1DEncoder(*args: Any, **kwargs: Any)

Bases: Module

Encode padded token sequences with masked temporal convolutions.

forward(values: SequenceInput) → torch.Tensor
input_kind: ClassVar[BranchInputKind] = 'sequence'
class DeepMTP.SequenceGRUEncoder(*args: Any, **kwargs: Any)

Bases: Module

Encode padded token sequences with an embedding layer and GRU.

forward(values: SequenceInput) → torch.Tensor
input_kind: ClassVar[BranchInputKind] = 'sequence'
class DeepMTP.SequenceTransformerEncoder(*args: Any, **kwargs: Any)

Bases: Module

Encode padded token sequences with masked self-attention.

forward(values: SequenceInput) → torch.Tensor
input_kind: ClassVar[BranchInputKind] = 'sequence'
class DeepMTP.SparseMLP(*args: Any, **kwargs: Any)

Bases: Module

Project sparse inputs before applying a dense MLP tail.

forward(values: torch.Tensor) → torch.Tensor
input_kind: ClassVar[BranchInputKind] = 'sparse'
class DeepMTP.TabularEncoder(*args: Any, **kwargs: Any)

Bases: Module

Encode normalized numeric values and embedded categorical columns.

forward(values: TabularInput) → torch.Tensor
input_kind: ClassVar[BranchInputKind] = 'tabular'
class DeepMTP.TabularSchema(numeric_columns: tuple[str, ...], categorical_columns: tuple[CategoricalColumn, ...], numeric_normalization: Literal['none', 'standard', 'minmax'] = 'standard', numeric_missing: Literal['error', 'mean', 'zero'] = 'mean', categorical_missing: Literal['error', 'unknown'] = 'unknown', categorical_unknown: Literal['error', 'unknown'] = 'unknown', feature_gating: bool = False)

Bases: object

Explicit numeric and categorical layout for one entity branch.

categorical_columns: tuple[CategoricalColumn, ...]
categorical_missing: Literal['error', 'unknown'] = 'unknown'
property categorical_names: tuple[str, ...]

Categorical column names in their stable tensor order.

categorical_unknown: Literal['error', 'unknown'] = 'unknown'
property encoded_width: int

Width after concatenating numeric values and category embeddings.

feature_gating: bool = False
classmethod from_config(value: object, *, branch: str) → TabularSchema

Validate a public tabular schema mapping.

numeric_columns: tuple[str, ...]
numeric_missing: Literal['error', 'mean', 'zero'] = 'mean'
numeric_normalization: Literal['none', 'standard', 'minmax'] = 'standard'
to_dict() → dict[str, Any]

Return a JSON-compatible public configuration mapping.

class DeepMTP.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.TokenizedFoundationSequences(rows: tuple[ndarray, ...], vocabulary_size: int, padding_idx: int, spec: FoundationModelSpec)

Bases: Sequence[ndarray]

Unpadded token rows plus their reproducible tokenizer settings.

padding_idx: int
rows: tuple[ndarray, ...]
spec: FoundationModelSpec
vocabulary_size: int
class DeepMTP.TrainingDataLoaders(train: torch.utils.data.DataLoader, validation: torch.utils.data.DataLoader, test: torch.utils.data.DataLoader)

Bases: object

Dataloaders required by one training run.

test: torch.utils.data.DataLoader
train: torch.utils.data.DataLoader
validation: torch.utils.data.DataLoader
class DeepMTP.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.TwoBranchDotProductModel(*args: Any, **kwargs: Any)

Bases: Module

Combine equal-sized branch embeddings with a dot product.

forward(instance_features: torch.Tensor | CompositeInput | GraphInput | SequenceInput | TabularInput, target_features: torch.Tensor | CompositeInput | GraphInput | SequenceInput | TabularInput) → torch.Tensor
class DeepMTP.TwoBranchKroneckerModel(*args: Any, **kwargs: Any)

Bases: Module

Combine branch embeddings using batched Kronecker products.

forward(instance_features: torch.Tensor | CompositeInput | GraphInput | SequenceInput | TabularInput, target_features: torch.Tensor | CompositeInput | GraphInput | SequenceInput | TabularInput) → torch.Tensor
class DeepMTP.TwoBranchMLPModel(*args: Any, **kwargs: Any)

Bases: Module

Combine branch embeddings using a multilayer perceptron.

forward(instance_features: torch.Tensor | CompositeInput | GraphInput | SequenceInput | TabularInput, target_features: torch.Tensor | CompositeInput | GraphInput | SequenceInput | TabularInput) → torch.Tensor
class DeepMTP.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.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.data_process(data: Mapping[str, Any], validation_setting: str | None = None, split_method: str = 'random', ratio: object = None, shuffle: bool = True, seed: int | None = 42, verbose: bool = False, print_mode: str = 'basic', scale_instance_features: str | None = None, scale_target_features: str | None = None, *, classification_mode: str | None = None, num_classes: int | None = None, instance_feature_kind: Literal['graph', 'sequence'] | None | Mapping[str, Literal['graph', 'sequence'] | None | CompositeFeatureSpec] = None, target_feature_kind: Literal['graph', 'sequence'] | None | Mapping[str, Literal['graph', 'sequence'] | None | CompositeFeatureSpec] = None, progress: DataProgressObserver | None = None, dense_preprocessing_state: Mapping[str, Any] | DataPreparationState | DensePreprocessingState | None = None) → tuple[dict[str, Any], dict[str, Any], dict[str, Any], DataInfo]

The main function that handles all the preprocessing steps and checks needed to prepare the dataset to be used by the model.

Parameters:

  • data (dict) – The dictionary that store all possible data available in a multi-target prediction dataset.

  • validation_setting (str, optional) – The validation setting of the current problem. Defaults to None.

  • split_method (str, optional) – The splitting method used. The current implementation only supports the ‘random split’ using a specific seed but a future goal is to also offer a stratified option. Defaults to ‘random’.

  • ratio (dict, optional) – The train, val and test ratios used to split the data. Defaults to {‘train’: 0.7, ‘test’: 0.2, ‘val’: 0.1}.

  • shuffle (bool, optional) – Whether or not the dataset will be shuffled before the split. If is set to False, the seed value is not used.

  • seed (int, optional) – The seed used to initiate the randomized split. Defaults to 42.

  • verbose (bool, optional) – Whether or not to print usefull info in the terminal. Defaults to False.

  • print_mode (str, optional) – The mode of printing. Two values are possible. If ‘basic’, the prints are just regural python prints. If ‘dev’ then a prefix is used so that the streamlit application can print more usefull messages. Defaults to ‘basic’.

  • scale_instance_features (str, optional) – The scaler used for the instance features. Possible values are ‘MinMax’ for the MinMax scaler and ‘Standard’ for the standard scaler. Defaults to None.

  • scale_target_features (str, optional) – The scaler used for the target features. Possible values are ‘MinMax’ for the MinMax scaler and ‘Standard’ for the standard scaler. Defaults to None.

  • classification_mode (str, optional) – Set to multiclass to interpret zero-based integer scores as class IDs. Binary scores continue to be detected automatically. Defaults to None.

  • num_classes (int, optional) – Declared multiclass count. When omitted in multiclass mode, it is inferred from all supplied score splits.

  • instance_feature_kind (str, optional) – Set to sequence when instance features are already-tokenized integer sequences. For composite features, pass a mapping from component name to sequence, None, or a mapping with kind and optional fields. Defaults to None.

  • target_feature_kind (str, optional) – Set to sequence when target features are already-tokenized integer sequences. For composite features, pass a mapping from component name to sequence, None, or a mapping with kind and optional fields. Defaults to None.

  • progress (DataProgressObserver | None, optional) – Receives verbose messages from the complete data-processing pipeline. Defaults to None.

  • dense_preprocessing_state (mapping, optional) – Previously saved dense scaler parameters. Accepts either the dense section or the full data_preparation_state returned in data_info. When supplied, the saved training statistics are replayed instead of fitting new scalers. Defaults to None.

Returns:

Four different dictionaries containing:

  • Train processed data

  • Validation processed data

  • Test processed data

  • general information about the datasets

Return type:

dict, dict, dict, dict

DeepMTP.describe_architecture(config: DeepMTPConfig | Mapping[str, Any], *, model: object | None = None) → ArchitectureDescription

Describe a validated configuration without instantiating its model.

DeepMTP.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.export_architecture_svg(config: DeepMTPConfig | Mapping[str, Any] | ArchitectureDescription, destination: str | PathLike[str], *, model: object | None = None, width: int = 140) → Path

Export a Rich architecture rendering as an SVG file.

DeepMTP.format_architecture(config: DeepMTPConfig | Mapping[str, Any] | ArchitectureDescription, *, model: object | None = None, style: Literal['plain', 'rich'] = 'plain', width: int = 120, color: bool = False) → str

Render an architecture to a string without writing to the terminal.

DeepMTP.print_architecture(config: DeepMTPConfig | Mapping[str, Any] | ArchitectureDescription, *, model: object | None = None, style: Literal['auto', 'plain', 'rich'] = 'auto', width: int | None = None, file: TextIO | None = None) → ArchitectureDescription

Print a responsive architecture diagram and return its description.

DeepMTP.tokenize_foundation_sequences(values: Sequence[str], spec: FoundationModelSpec | Mapping[str, Any]) → TokenizedFoundationSequences

Tokenize raw protein, SMILES, or text rows without batch padding.

DeepMTP.transform_dense_features(values: object, state: Mapping[str, Any] | DenseScalingState) → ndarray

Replay a saved dense-feature transformation.