DeepMTP.hpo package

The canonical hyperparameter-optimization API is available directly from DeepMTP.hpo:

from DeepMTP.hpo import BaseWorker, HyperBand, RandomSearch

Worker

class DeepMTP.hpo.worker.BaseWorker(train: Any, val: Any, test: Any, data_info: Any, base_config: Mapping[str, Any], metric_to_optimize: str, mode: str = 'standard', progress: HPOProgressObserver | None = None)

Bases: object

Implements a basic worker that can be used by HPO methods. The basic idea is that an HPO methods just has to pass a config and then it gets back the performance of the best epoch on the validation set

compute(budget: int | float, config: Mapping[str, Any]) → HPOWorkerResult

The input parameter ‘config’ (dictionary) contains the sampled configurations passed by the bohb optimizer

Progress

Progress events shared by hyperparameter optimizers and UI adapters.

class DeepMTP.hpo.progress.ConsoleHPOProgressObserver

Bases: object

Render optimizer events in a terminal.

on_event(event: HPOProgressEvent) → None
class DeepMTP.hpo.progress.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.hpo.progress.HPOProgressObserver(*args, **kwargs)

Bases: Protocol

Receives optimizer lifecycle events.

on_event(event: HPOProgressEvent) → None

Handle an optimizer event.

class DeepMTP.hpo.progress.NullHPOProgressObserver

Bases: object

No-op optimizer progress observer.

on_event(event: HPOProgressEvent) → None
DeepMTP.hpo.progress.build_hpo_progress_observer(verbose: bool) → HPOProgressObserver

Types

Shared structural types for hyperparameter optimizers.

class DeepMTP.hpo.types.ConfigurationSpace(*args, **kwargs)

Bases: Protocol

Configuration-space surface consumed by the bundled optimizers.

sample_configuration() → Mapping[str, Any]

Sample one candidate configuration.

class DeepMTP.hpo.types.HPOWorker(*args, **kwargs)

Bases: Protocol

Worker surface consumed by the bundled optimizers.

compute(budget: int | float, config: Mapping[str, Any]) → HPOWorkerResult

Evaluate one configuration at the requested budget.

class DeepMTP.hpo.types.HPOWorkerResult

Bases: TypedDict

Stable result returned by an HPO worker.

info: dict[str, Any]
loss: float

Package contents

Hyperparameter optimization workers, optimizers, types, and progress events.

class DeepMTP.hpo.BaseWorker(train: Any, val: Any, test: Any, data_info: Any, base_config: Mapping[str, Any], metric_to_optimize: str, mode: str = 'standard', progress: HPOProgressObserver | None = None)

Bases: object

Implements a basic worker that can be used by HPO methods. The basic idea is that an HPO methods just has to pass a config and then it gets back the performance of the best epoch on the validation set

compute(budget: int | float, config: Mapping[str, Any]) → HPOWorkerResult

The input parameter ‘config’ (dictionary) contains the sampled configurations passed by the bohb optimizer

class DeepMTP.hpo.ConfigurationSpace(*args, **kwargs)

Bases: Protocol

Configuration-space surface consumed by the bundled optimizers.

sample_configuration() → Mapping[str, Any]

Sample one candidate configuration.

class DeepMTP.hpo.ConsoleHPOProgressObserver

Bases: object

Render optimizer events in a terminal.

on_event(event: HPOProgressEvent) → None
class DeepMTP.hpo.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.hpo.HPOProgressObserver(*args, **kwargs)

Bases: Protocol

Receives optimizer lifecycle events.

on_event(event: HPOProgressEvent) → None

Handle an optimizer event.

class DeepMTP.hpo.HPOWorker(*args, **kwargs)

Bases: Protocol

Worker surface consumed by the bundled optimizers.

compute(budget: int | float, config: Mapping[str, Any]) → HPOWorkerResult

Evaluate one configuration at the requested budget.

class DeepMTP.hpo.HPOWorkerResult

Bases: TypedDict

Stable result returned by an HPO worker.

info: dict[str, Any]
loss: float
class DeepMTP.hpo.HyperBand(base_worker: HPOWorker, configspace: ConfigurationSpace, eta: int = 3, max_budget: int = 1, direction: str = 'min', verbose: bool = False, progress: HPOProgressObserver | None = None)

Bases: object

Implements a basic version of the Hyperband HPO method. One cool thing about it is that I reduced the training time by continuing to train later configurations instead of starting from scratch each time.

calculate_hyperband_iters(R: int, eta: int, verbose: bool = False) → dict[int, HyperbandBracket]
get_run_summary() → dict[int, dict[int, list[BaseExperimentInfo]]]
run_optimizer() → BaseExperimentInfo
class DeepMTP.hpo.HyperbandBracket

Bases: TypedDict

Configuration counts and budgets for one Hyperband bracket.

n_i: list[int]
num_iters: int
r_i: list[int]
class DeepMTP.hpo.NullHPOProgressObserver

Bases: object

No-op optimizer progress observer.

on_event(event: HPOProgressEvent) → None
class DeepMTP.hpo.RandomSearch(base_worker: HPOWorker, configspace: ConfigurationSpace, budget: int = 1, max_num_epochs: int = 100, direction: str = 'min', verbose: bool = False, progress: HPOProgressObserver | None = None)

Bases: object

Implements the basic Random search HPO method. Nothing fancy, just a for loop over randomly generated configurations.

get_run_summary() → dict[int, BaseExperimentInfo]
run_optimizer() → BaseExperimentInfo
DeepMTP.hpo.build_hpo_progress_observer(verbose: bool) → HPOProgressObserver