Source code for ax.modelbridge.transforms.percentile_y

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

import math
from typing import List, Optional, TYPE_CHECKING

from ax.core.observation import ObservationData, ObservationFeatures
from ax.core.search_space import SearchSpace
from ax.modelbridge.transforms.base import Transform
from ax.modelbridge.transforms.utils import get_data
from ax.models.types import TConfig
from ax.utils.common.logger import get_logger
from ax.utils.common.typeutils import checked_cast
from scipy import stats

if TYPE_CHECKING:
    # import as module to make sphinx-autodoc-typehints happy
    from ax import modelbridge as modelbridge_module  # noqa F401  # pragma: no cover


logger = get_logger(__name__)


# TODO(jej): Add OptimizationConfig validation - can't transform outcome constraints.
[docs]class PercentileY(Transform):
    """Map Y values to percentiles based on their empirical CDF."""

    def __init__(
        self,
        search_space: SearchSpace,
        observation_features: List[ObservationFeatures],
        observation_data: List[ObservationData],
        modelbridge: Optional["modelbridge_module.base.ModelBridge"] = None,
        config: Optional[TConfig] = None,
    ) -> None:
        if len(observation_data) == 0:
            raise ValueError(
                "Percentile transform requires non-empty observation data."
            )
        metric_values = get_data(observation_data=observation_data)
        self.percentiles = {
            metric_name: vals for metric_name, vals in metric_values.items()
        }
        if config is not None and "winsorize" in config:
            self.winsorize = checked_cast(bool, (config.get("winsorize") or False))
        else:
            self.winsorize = False

[docs]    def transform_observation_data(
        self,
        observation_data: List[ObservationData],
        observation_features: List[ObservationFeatures],
    ) -> List[ObservationData]:
        """Map observation data to empirical CDF quantiles in place."""
        # TODO (jej): Transform covariances.
        if self.winsorize:
            winsorization_rates = {}
            for metric_name, vals in self.percentiles.items():
                n = len(vals)
                # Calculate winsorization rate based on number of observations
                # using formula from [Salinas, Shen, Perrone 2020]
                # https://arxiv.org/abs/1909.13595
                winsorization_rates[metric_name] = (
                    1.0 / (4 * math.pow(n, 0.25) * math.pow(math.pi * math.log(n), 0.5))
                    if n > 1
                    else 0.25
                )
        else:
            winsorization_rates = {
                metric_name: 0 for metric_name in self.percentiles.keys()
            }
        for obsd in observation_data:
            for idx, metric_name in enumerate(obsd.metric_names):
                if metric_name not in self.percentiles:  # pragma: no cover
                    raise ValueError(
                        f"Cannot map value to percentile"
                        f" for unknown metric {metric_name}"
                    )
                # apply map function
                percentile = self._map(obsd.means[idx], metric_name)
                # apply winsorization. If winsorization_rate is 0, has no effect.
                metric_wr = winsorization_rates[metric_name]
                percentile = max(metric_wr, percentile)
                percentile = min((1 - metric_wr), percentile)
                obsd.means[idx] = percentile
                obsd.covariance.fill(float("nan"))
        return observation_data

    def _map(self, val: float, metric_name: str) -> float:
        vals = self.percentiles[metric_name]
        mapped_val = stats.percentileofscore(vals, val, kind="weak") / 100.0
        return mapped_val