Source code for ax.core.objective

#!/usr/bin/env python3
# Copyright (c) Facebook, Inc. and its affiliates.
#
# This source code is licensed under the MIT license found in the
# LICENSE file in the root directory of this source tree.

from typing import Any, Iterable, List, Optional, Tuple

from ax.core.base import Base
from ax.core.metric import Metric
from ax.utils.common.logger import get_logger


logger = get_logger(__name__)


[docs]class Objective(Base): """Base class for representing an objective. Attributes: minimize: If True, minimize metric. """ def __init__(self, metric: Metric, minimize: bool = False) -> None: """Create a new objective. Args: metric: The metric to be optimized. minimize: If True, minimize metric. """ self._metric = metric self.minimize = minimize @property def metric(self) -> Metric: """Get the objective metric.""" return self._metric @property def metrics(self) -> List[Metric]: """Get a list of objective metrics.""" return [self._metric]
[docs] def clone(self) -> "Objective": """Create a copy of the objective.""" return Objective(self.metric.clone(), self.minimize)
def __repr__(self) -> str: return 'Objective(metric_name="{}", minimize={})'.format( self.metric.name, self.minimize )
# TODO (jej): Support sqa_store encoding. Currenlty only single metric obj supported.
[docs]class MultiObjective(Objective): """Class for an objective composed of a multiple component objectives. The Acquisition function determines how the objectives are weighted. Attributes: metrics: List of metrics. """ weights: List[float] def __init__( self, metrics: List[Metric], minimize: bool = False, **extra_kwargs: Any, # Here to satisfy serialization. ) -> None: """Create a new objective. Args: metrics: The list of metrics to be jointly optimized. minimize: If true, minimize the aggregate of these metrics. """ self._metrics = metrics self.weights = [] for metric in metrics: # Set weights from "lower_is_better" if metric.lower_is_better is None: logger.warning( f"metric {metric.name} has not set `lower_is_better`. " "Treating as `False` (Metric should be maximized)." ) self.weights.append(-1.0 if metric.lower_is_better is True else 1.0) self.minimize = minimize @property def metric_weights(self) -> Iterable[Tuple[Metric, float]]: """Get the objective metrics and weights.""" return zip(self.metrics, self.weights) @property def metric(self) -> Metric: """Override base method to error.""" raise NotImplementedError( f"{type(self).__name__} is composed of multiple metrics" ) @property def metrics(self) -> List[Metric]: """Get the objective metrics.""" return self._metrics
[docs] def clone(self) -> "Objective": """Create a copy of the objective.""" return MultiObjective( metrics=[m.clone() for m in self.metrics], minimize=self.minimize )
def __repr__(self) -> str: return "MultiObjective(metric_names={}, minimize={})".format( [metric.name for metric in self.metrics], self.minimize )
[docs]class ScalarizedObjective(MultiObjective): """Class for an objective composed of a linear scalarization of metrics. Attributes: metrics: List of metrics. weights: Weights for scalarization; default to 1. """ weights: List[float] def __init__( self, metrics: List[Metric], weights: Optional[List[float]] = None, minimize: bool = False, ) -> None: """Create a new objective. Args: metric: The metric to be optimized. weights: The weights for the linear combination of metrics. minimize: If true, minimize the linear combination. """ if weights is None: weights = [1.0 for i in range(len(metrics))] else: if len(weights) != len(metrics): raise ValueError("Length of weights must equal length of metrics") super().__init__(metrics, minimize) self.weights = weights
[docs] def clone(self) -> "Objective": """Create a copy of the objective.""" return ScalarizedObjective( metrics=[m.clone() for m in self.metrics], weights=self.weights.copy(), minimize=self.minimize, )
def __repr__(self) -> str: return "ScalarizedObjective(metric_names={}, weights={}, minimize={})".format( [metric.name for metric in self.metrics], self.weights, self.minimize )