Source code for ax.modelbridge.transforms.ivw

#!/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.

# pyre-strict

from logging import Logger
from typing import Dict, List

import numpy as np
from ax.core.observation import ObservationData
from ax.modelbridge.transforms.base import Transform
from ax.utils.common.logger import get_logger

logger: Logger = get_logger(__name__)


[docs]def ivw_metric_merge(
    obsd: ObservationData, conflicting_noiseless: str = "warn"
) -> ObservationData:
    """Merge multiple observations of a metric with inverse variance weighting.

    Correctly updates the covariance of the new merged estimates:
    ybar1 = Sum_i w_i * y_i
    ybar2 = Sum_j w_j * y_j
    cov[ybar1, ybar2] = Sum_i Sum_j w_i * w_j * cov[y_i, y_j]

    w_i will be infinity if any variance is 0. If one variance is 0., then
    the IVW estimate is the corresponding mean. If there are multiple
    measurements with 0 variance but means are all the same, then IVW estimate
    is that mean. If there are multiple measurements and means differ, behavior
    depends on argument conflicting_noiseless. "ignore" and "warn" will use
    the first of the measurements as the IVW estimate. "warn" will additionally
    log a warning. "raise" will raise an exception.

    Args:
        obsd: An ObservationData object
        conflicting_noiseless: "warn", "ignore", or "raise"
    """
    if len(obsd.metric_names) == len(set(obsd.metric_names)):
        return obsd
    if conflicting_noiseless not in {"warn", "ignore", "raise"}:
        raise ValueError(
            'conflicting_noiseless should be "warn", "ignore", or "raise".'
        )
    # Get indicies and weights for each metric.
    # weights is a map from metric name to a vector of the weights for each
    # measurement of that metric. indicies gives the corresponding index in
    # obsd.means for each measurement.
    weights: Dict[str, np.ndarray] = {}
    indicies: Dict[str, List[int]] = {}
    for metric_name in set(obsd.metric_names):
        indcs = [i for i, mn in enumerate(obsd.metric_names) if mn == metric_name]
        indicies[metric_name] = indcs
        # Extract variances for observations of this metric
        sigma2s = obsd.covariance[indcs, indcs]
        # Check for noiseless observations
        idx_noiseless = np.where(sigma2s == 0.0)[0]
        if len(idx_noiseless) == 0:
            # Weight is inverse of variance, normalized
            # Expected `np.ndarray` for 3rd anonymous parameter to call
            # `dict.__setitem__` but got `float`.
            # pyre-fixme[6]:
            weights[metric_name] = 1.0 / sigma2s
            weights[metric_name] /= np.sum(weights[metric_name])
        else:
            # Check if there are conflicting means for the noiseless observations
            means_noiseless = obsd.means[indcs][idx_noiseless]
            _check_conflicting_means(
                means_noiseless, metric_name, conflicting_noiseless
            )
            # The first observation gets all the weight.
            weights[metric_name] = np.zeros_like(sigma2s)
            weights[metric_name][idx_noiseless[0]] = 1.0
    # Compute the new values
    metric_names = sorted(set(obsd.metric_names))
    means = np.zeros(len(metric_names))
    covariance = np.zeros((len(metric_names), len(metric_names)))
    for i, metric_name in enumerate(metric_names):
        ys = obsd.means[indicies[metric_name]]
        means[i] = np.sum(weights[metric_name] * ys)
        # Calculate covariances with metric_name
        for j, metric_name2 in enumerate(metric_names[i:], start=i):
            for ii, idx_i in enumerate(indicies[metric_name]):
                for jj, idx_j in enumerate(indicies[metric_name2]):
                    covariance[i, j] += (
                        weights[metric_name][ii]
                        * weights[metric_name2][jj]
                        * obsd.covariance[idx_i, idx_j]
                    )
            covariance[j, i] = covariance[i, j]
    return ObservationData(
        metric_names=metric_names, means=means, covariance=covariance
    )


def _check_conflicting_means(
    means_noiseless: np.ndarray, metric_name: str, conflicting_noiseless: str
) -> None:
    if np.var(means_noiseless) > 0:
        message = f"Conflicting noiseless measurements for {metric_name}."
        if conflicting_noiseless == "warn":
            logger.warning(message)
        elif conflicting_noiseless == "raise":
            raise ValueError(message)


[docs]class IVW(Transform):
    """If an observation data contains multiple observations of a metric, they
    are combined using inverse variance weighting.
    """

    def _transform_observation_data(
        self,
        observation_data: List[ObservationData],
    ) -> List[ObservationData]:
        # pyre: conflicting_noiseless is declared to have type `str` but is
        # pyre-fixme[9]: used as type `typing.Union[float, int, str]`.
        conflicting_noiseless: str = self.config.get("conflicting_noiseless", "warn")
        return [
            ivw_metric_merge(obsd=obsd, conflicting_noiseless=conflicting_noiseless)
            for obsd in observation_data
        ]