Contextual Bayesian Optimization¶

This example compares standard Bayesian optimization against contextual Bayesian optimization on a drifting objective.

The optimization variable is x1, while context is observed after each evaluation but cannot be controlled. In the contextual setup, define context as a ContextualVariable in VOCS.

ContextualVariable() defaults to an unbounded domain (-inf, inf). When finite bounds are needed (for example, model input bounds or visualization meshes), Xopt infers them from current data. If you provide an explicit finite domain, that explicit domain takes precedence over inferred bounds.

The Bayesian generator conditions on the latest observed context value when optimizing the acquisition function.

In [1]:

import os
import warnings
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

from xopt import Evaluator
from xopt.vocs import VOCS, ContextualVariable

warnings.filterwarnings(
    "ignore", message="Data \(outcome observations\) is not standardized.*"
)

SMOKE_TEST = os.environ.get("SMOKE_TEST", "").lower() in ["1", "true", "yes"]
N_INITIAL = 3 if SMOKE_TEST else 8
N_STEPS = 8 if SMOKE_TEST else 50

import os import warnings import numpy as np import pandas as pd import matplotlib.pyplot as plt from xopt import Evaluator from xopt.vocs import VOCS, ContextualVariable warnings.filterwarnings( "ignore", message="Data \(outcome observations\) is not standardized.*" ) SMOKE_TEST = os.environ.get("SMOKE_TEST", "").lower() in ["1", "true", "yes"] N_INITIAL = 3 if SMOKE_TEST else 8 N_STEPS = 8 if SMOKE_TEST else 50

<>:11: SyntaxWarning: invalid escape sequence '\('
<>:11: SyntaxWarning: invalid escape sequence '\('
/tmp/ipykernel_3416/1574807644.py:11: SyntaxWarning: invalid escape sequence '\('
  "ignore", message="Data \(outcome observations\) is not standardized.*"

Drifting Objective¶

We construct a simple non-stationary objective where the optimum moves over time. The evaluator reports both the objective value y and the observed context value context at each iteration.

In [2]:

class DriftingObjective:
    def __init__(self, period: int = 20):
        self.period = period
        self.iteration = 0

    def __call__(self, inputs):
        x = float(inputs["x1"])
        context = 0.5 * (1.0 + np.sin(2.0 * np.pi * self.iteration / self.period))

        # Peak value is 1.0 when x1 equals the current context.
        y = -((x - context) ** 2)

        self.iteration += 1
        return {
            "y": float(y),
            "context": float(context),
            "tracking_error": float(abs(x - context)),
        }

class DriftingObjective: def __init__(self, period: int = 20): self.period = period self.iteration = 0 def __call__(self, inputs): x = float(inputs["x1"]) context = 0.5 * (1.0 + np.sin(2.0 * np.pi * self.iteration / self.period)) # Peak value is 1.0 when x1 equals the current context. y = -((x - context) ** 2) self.iteration += 1 return { "y": float(y), "context": float(context), "tracking_error": float(abs(x - context)), }

In [3]:

from xopt.base import Xopt
from xopt.generators.bayesian.upper_confidence_bound import (
    UpperConfidenceBoundGenerator,
)


def run_optimization(use_context: bool):
    objective = DriftingObjective(25)
    evaluator = Evaluator(function=objective)

    if use_context:
        vocs = VOCS(
            # Default ContextualVariable() uses unbounded domain and infers finite
            # bounds from observed data when needed. To force fixed bounds, use:
            # ContextualVariable(domain=[low, high]).
            variables={"x1": [0.0, 1.0], "context": ContextualVariable()},
            objectives={"y": "MAXIMIZE"},
            observables=["context", "tracking_error"],
        )
    else:
        vocs = VOCS(
            variables={"x1": [0.0, 1.0]},
            objectives={"y": "MAXIMIZE"},
            observables=["context", "tracking_error"],
        )

    generator = UpperConfidenceBoundGenerator(vocs=vocs)
    generator.gp_constructor.use_low_noise_prior = True

    X = Xopt(generator=generator, evaluator=evaluator)

    # Build initial data manually since contextual variables are observed, not sampled.
    initial_inputs = pd.DataFrame({"x1": np.linspace(0.05, 0.95, N_INITIAL)})
    X.evaluate_data(initial_inputs)

    for _ in range(N_STEPS):
        X.step()

    data = X.data
    data["iteration"] = np.arange(len(data))

    return X.generator, data

from xopt.base import Xopt from xopt.generators.bayesian.upper_confidence_bound import ( UpperConfidenceBoundGenerator, ) def run_optimization(use_context: bool): objective = DriftingObjective(25) evaluator = Evaluator(function=objective) if use_context: vocs = VOCS( # Default ContextualVariable() uses unbounded domain and infers finite # bounds from observed data when needed. To force fixed bounds, use: # ContextualVariable(domain=[low, high]). variables={"x1": [0.0, 1.0], "context": ContextualVariable()}, objectives={"y": "MAXIMIZE"}, observables=["context", "tracking_error"], ) else: vocs = VOCS( variables={"x1": [0.0, 1.0]}, objectives={"y": "MAXIMIZE"}, observables=["context", "tracking_error"], ) generator = UpperConfidenceBoundGenerator(vocs=vocs) generator.gp_constructor.use_low_noise_prior = True X = Xopt(generator=generator, evaluator=evaluator) # Build initial data manually since contextual variables are observed, not sampled. initial_inputs = pd.DataFrame({"x1": np.linspace(0.05, 0.95, N_INITIAL)}) X.evaluate_data(initial_inputs) for _ in range(N_STEPS): X.step() data = X.data data["iteration"] = np.arange(len(data)) return X.generator, data

In [4]:

np.random.seed(1)

plain_generator, plain_data = run_optimization(use_context=False)
contextual_generator, contextual_data = run_optimization(use_context=True)

print(f"Standard BO evaluations: {len(plain_data)}")
print(f"Contextual BO evaluations: {len(contextual_data)}")

np.random.seed(1) plain_generator, plain_data = run_optimization(use_context=False) contextual_generator, contextual_data = run_optimization(use_context=True) print(f"Standard BO evaluations: {len(plain_data)}") print(f"Contextual BO evaluations: {len(contextual_data)}")

Standard BO evaluations: 58
Contextual BO evaluations: 58

Compare Standard BO and Contextual BO¶

In the first panel, we compare selected x1 values against the drifting context. In the second panel, we compare rolling objective values. In the third panel, we compare rolling tracking error (lower is better).

In [5]:

window = 1 if not SMOKE_TEST else 3

fig, axes = plt.subplots(1, 3, figsize=(16, 4))

axes[0].plot(plain_data["iteration"], plain_data["x1"], label="standard BO", lw=1.8)
axes[0].plot(
    contextual_data["iteration"],
    contextual_data["x1"],
    label="contextual BO",
    lw=1.8,
)
axes[0].plot(
    plain_data["iteration"],
    plain_data["context"],
    "k--",
    label="observed context",
    alpha=0.8,
)
axes[0].set_title("Chosen x1 vs Context")
axes[0].set_xlabel("iteration")
axes[0].set_ylabel("value")
axes[0].legend(loc="best")

axes[1].plot(
    plain_data["iteration"],
    plain_data["y"].rolling(window, min_periods=1).mean(),
    label="standard BO",
)
axes[1].plot(
    contextual_data["iteration"],
    contextual_data["y"].rolling(window, min_periods=1).mean(),
    label="contextual BO",
)
axes[1].set_title(f"Rolling mean objective (window={window})")
axes[1].set_xlabel("iteration")
axes[1].set_ylabel("y")
axes[1].legend(loc="best")

axes[2].plot(
    plain_data["iteration"],
    plain_data["tracking_error"].rolling(window, min_periods=1).mean(),
    label="standard BO",
)
axes[2].plot(
    contextual_data["iteration"],
    contextual_data["tracking_error"].rolling(window, min_periods=1).mean(),
    label="contextual BO",
)
axes[2].set_title(f"Rolling mean tracking error (window={window})")
axes[2].set_xlabel("iteration")
axes[2].set_ylabel("|x1 - context|")
axes[2].legend(loc="best")

plt.tight_layout()

window = 1 if not SMOKE_TEST else 3 fig, axes = plt.subplots(1, 3, figsize=(16, 4)) axes[0].plot(plain_data["iteration"], plain_data["x1"], label="standard BO", lw=1.8) axes[0].plot( contextual_data["iteration"], contextual_data["x1"], label="contextual BO", lw=1.8, ) axes[0].plot( plain_data["iteration"], plain_data["context"], "k--", label="observed context", alpha=0.8, ) axes[0].set_title("Chosen x1 vs Context") axes[0].set_xlabel("iteration") axes[0].set_ylabel("value") axes[0].legend(loc="best") axes[1].plot( plain_data["iteration"], plain_data["y"].rolling(window, min_periods=1).mean(), label="standard BO", ) axes[1].plot( contextual_data["iteration"], contextual_data["y"].rolling(window, min_periods=1).mean(), label="contextual BO", ) axes[1].set_title(f"Rolling mean objective (window={window})") axes[1].set_xlabel("iteration") axes[1].set_ylabel("y") axes[1].legend(loc="best") axes[2].plot( plain_data["iteration"], plain_data["tracking_error"].rolling(window, min_periods=1).mean(), label="standard BO", ) axes[2].plot( contextual_data["iteration"], contextual_data["tracking_error"].rolling(window, min_periods=1).mean(), label="contextual BO", ) axes[2].set_title(f"Rolling mean tracking error (window={window})") axes[2].set_xlabel("iteration") axes[2].set_ylabel("|x1 - context|") axes[2].legend(loc="best") plt.tight_layout()

In [6]:

comparison = pd.DataFrame(
    {
        "mean_y": [plain_data["y"].mean(), contextual_data["y"].mean()],
        "mean_tracking_error": [
            plain_data["tracking_error"].mean(),
            contextual_data["tracking_error"].mean(),
        ],
    },
    index=["standard_bo", "contextual_bo"],
)

comparison

comparison = pd.DataFrame( { "mean_y": [plain_data["y"].mean(), contextual_data["y"].mean()], "mean_tracking_error": [ plain_data["tracking_error"].mean(), contextual_data["tracking_error"].mean(), ], }, index=["standard_bo", "contextual_bo"], ) comparison

Out[6]:

	mean_y	mean_tracking_error
standard_bo	-0.180131	0.358430
contextual_bo	-0.029644	0.128718

In [7]:

if (
    comparison.loc["contextual_bo", "mean_tracking_error"]
    < comparison.loc["standard_bo", "mean_tracking_error"]
):
    print("Contextual BO tracks the drifting optimum better in this run.")
else:
    print(
        "In this run, contextual BO did not improve tracking error. Try changing the period or number of steps."
    )

if ( comparison.loc["contextual_bo", "mean_tracking_error"] < comparison.loc["standard_bo", "mean_tracking_error"] ): print("Contextual BO tracks the drifting optimum better in this run.") else: print( "In this run, contextual BO did not improve tracking error. Try changing the period or number of steps." )

Contextual BO tracks the drifting optimum better in this run.

Visualizing contextual models and acquisition functions¶

These plots inspect the surrogate models learned by the contextual and standard generators. For the contextual generator, the model depends on both x1 and the observed context value, so visualization can show either the full model or a slice over selected variables.

Note that we cannot visualize the acquisition function in the 2D plots shown below because it is conditioned on the context value, however we can visualize it in 1D because the context value is given by the reference point.

In [8]:

contextual_generator.visualize_model()

contextual_generator.visualize_model()

/home/runner/work/Xopt/Xopt/xopt/generators/bayesian/visualize.py:376: RuntimeWarning: Acquisition plot unavailable: conditioned on contextual variables and defined over controllable dimensions only.
  plot_acquisition_function(

Out[8]:

(<Figure size 800x1320 with 14 Axes>,
 array([[<Axes: title={'center': 'Posterior Mean [y]'}, xlabel='x1', ylabel='context'>,
         <Axes: title={'center': 'Posterior SD [y]'}, xlabel='x1', ylabel='context'>],
        [<Axes: title={'center': 'Posterior Mean [context]'}, xlabel='x1', ylabel='context'>,
         <Axes: title={'center': 'Posterior SD [context]'}, xlabel='x1', ylabel='context'>],
        [<Axes: title={'center': 'Posterior Mean [tracking_error]'}, xlabel='x1', ylabel='context'>,
         <Axes: title={'center': 'Posterior SD [tracking_error]'}, xlabel='x1', ylabel='context'>],
        [<Axes: title={'center': 'Acquisition Function'}>, <Axes: >]],
       dtype=object))

In [9]:

contextual_generator.visualize_model(
    variable_names=["x1"],
    show_samples=False,
)

contextual_generator.visualize_model( variable_names=["x1"], show_samples=False, )

Out[9]:

(<Figure size 600x800 with 4 Axes>,
 array([[<Axes: ylabel='y'>],
        [<Axes: ylabel='context'>],
        [<Axes: ylabel='tracking_error'>],
        [<Axes: xlabel='x1', ylabel='$\\exp[ \\alpha]$'>]], dtype=object))

In [10]:

plain_generator.visualize_model()

plain_generator.visualize_model()

Out[10]:

(<Figure size 600x800 with 4 Axes>,
 array([[<Axes: ylabel='y'>],
        [<Axes: ylabel='context'>],
        [<Axes: ylabel='tracking_error'>],
        [<Axes: xlabel='x1', ylabel='$\\exp[ \\alpha]$'>]], dtype=object))

In [ ]: