Environments

Environment dynamics declare the reward domains they support. Use ProbabilityDrift for Bernoulli means; additive GradualDrift and AbruptShift are restricted to unbounded real-valued means. Use LogisticContextualEnvironment when binary contextual rewards follow a logit link.

Custom reward models, dynamics, and stateful context providers must implement a clone method that returns independent state for each replicate. The default base implementations use deepcopy. Plain context callables are wrapped as stateless providers and must not close over mutable state; subclass ContextProvider when state is required.

Bandit environments and support-aware non-stationary dynamics.

class AbruptShift[source]

Bases: EnvironmentDynamics

Periodic Gaussian shifts for unbounded real-valued means.

frequency: int = 100

std: float = 0.5

shift_at_step_zero: bool = False

supported_domains: frozenset[RewardDomain] = frozenset({RewardDomain.REAL})

apply(means, *, step, rng)[source]

Return the arm means for the supplied step.

Parameters:

means (ndarray[tuple[Any, ...], dtype[float64]])
step (int)
rng (Generator)

Return type:

ndarray[tuple[Any, …], dtype[float64]]

__init__(frequency=100, std=0.5, shift_at_step_zero=False)

Parameters:

frequency (int)
std (float)
shift_at_step_zero (bool)

Return type:

None

clone()

Return independent dynamics state for one replicate.

Return type:: EnvironmentDynamics

supports(domain)

Parameters:: domain (RewardDomain)
Return type:: bool

class BanditEnvironment[source]

Bases: object

Classic K-armed simulator with support-aware rewards and dynamics.

means: ndarray[tuple[Any, ...], dtype[float64]]

reward_model: RewardModel

dynamics: EnvironmentDynamics

classmethod from_prior(*, n_arms, prior, reward_model=None, dynamics=None, rng)[source]

Parameters:

n_arms (int)
prior (ArmPrior)
reward_model (RewardModel | None)
dynamics (EnvironmentDynamics | None)
rng (Generator)

Return type:

BanditEnvironment

property n_arms: int

property reward_domain: RewardDomain

property contextual: bool

expected_rewards()[source]

Return type:: ndarray[tuple[Any, …], dtype[float64]]

sample_rewards(*, rng)[source]

Parameters:: rng (Generator)
Return type:: ndarray[tuple[Any, …], dtype[float64]]

advance(*, step, rng)[source]

Parameters:

step (int)
rng (Generator)

Return type:

None

clone()[source]

Return type:: BanditEnvironment

__init__(means, reward_model=<factory>, dynamics=<factory>)

Parameters:

means (ndarray[tuple[Any, ...], dtype[float64]])
reward_model (RewardModel)
dynamics (EnvironmentDynamics)

Return type:

None

class CallableContextProvider[source]

Bases: ContextProvider

Stateless adapter for a context-producing callable.

function: Callable[[Generator], ndarray[tuple[Any, ...], dtype[float64]]]

sample(rng)[source]

Return context features for one decision.

Parameters:: rng (Generator)
Return type:: ndarray[tuple[Any, …], dtype[float64]]

clone()[source]

Return independent provider state for one replicate.

Return type:: CallableContextProvider

__init__(function)

Parameters:: function (Callable[[Generator], ndarray[tuple[Any, ...], dtype[float64]]])
Return type:: None

class ClassicEnvironment[source]

Bases: Protocol

Structural contract for a classic non-contextual environment.

property n_arms: int

property reward_model: RewardModel

property reward_domain: RewardDomain

property contextual: bool

clone()[source]

Return type:: ClassicEnvironment

advance(*, step, rng)[source]

Parameters:

step (int)
rng (Generator)

Return type:

None

expected_rewards()[source]

Return type:: ndarray[tuple[Any, …], dtype[float64]]

__init__(*args, **kwargs)

class ContextProvider[source]

Bases: ABC

Cloneable source of contextual feature matrices.

abstractmethod sample(rng)[source]

Return context features for one decision.

Parameters:: rng (Generator)
Return type:: ndarray[tuple[Any, …], dtype[float64]]

clone()[source]

Return independent provider state for one replicate.

Return type:: ContextProvider

class ContextualEnvironment[source]

Bases: Protocol

Structural contract for a contextual environment.

property n_arms: int

property reward_model: RewardModel

property n_features: int

property reward_domain: RewardDomain

property contextual: bool

clone()[source]

Return type:: ContextualEnvironment

context(rng)[source]

Parameters:: rng (Generator)
Return type:: ndarray[tuple[Any, …], dtype[float64]]

expected_rewards(context)[source]

Parameters:: context (ndarray[tuple[Any, ...], dtype[float64]])
Return type:: ndarray[tuple[Any, …], dtype[float64]]

__init__(*args, **kwargs)

class EnvironmentDynamics[source]

Bases: ABC

Strategy for evolving true arm means.

supported_domains: frozenset[RewardDomain]

abstractmethod apply(means, *, step, rng)[source]

Return the arm means for the supplied step.

Parameters:

means (ndarray[tuple[Any, ...], dtype[float64]])
step (int)
rng (Generator)

Return type:

ndarray[tuple[Any, …], dtype[float64]]

supports(domain)[source]

Parameters:: domain (RewardDomain)
Return type:: bool

clone()[source]

Return independent dynamics state for one replicate.

Return type:: EnvironmentDynamics

class FixedContextProvider[source]

Bases: ContextProvider

Deterministic native-compatible source of contextual features.

value: ndarray[tuple[Any, ...], dtype[float64]]

sample(rng)[source]

Return context features for one decision.

Parameters:: rng (Generator)
Return type:: ndarray[tuple[Any, …], dtype[float64]]

clone()[source]

Return independent provider state for one replicate.

Return type:: FixedContextProvider

__init__(value)

Parameters:: value (ndarray[tuple[Any, ...], dtype[float64]])
Return type:: None

class GaussianContextProvider[source]

Bases: ContextProvider

Native-compatible source of independent Gaussian context matrices.

n_arms: int

n_features: int

mean: float = 0.0

std: float = 1.0

sample(rng)[source]

Return context features for one decision.

Parameters:: rng (Generator)
Return type:: ndarray[tuple[Any, …], dtype[float64]]

clone()[source]

Return independent provider state for one replicate.

Return type:: GaussianContextProvider

__init__(n_arms, n_features, mean=0.0, std=1.0)

Parameters:

n_arms (int)
n_features (int)
mean (float)
std (float)

Return type:

None

class GradualDrift[source]

Bases: EnvironmentDynamics

Gaussian random-walk drift for unbounded real-valued means.

std: float = 0.01

supported_domains: frozenset[RewardDomain] = frozenset({RewardDomain.REAL})

apply(means, *, step, rng)[source]

Return the arm means for the supplied step.

Parameters:

means (ndarray[tuple[Any, ...], dtype[float64]])
step (int)
rng (Generator)

Return type:

ndarray[tuple[Any, …], dtype[float64]]

__init__(std=0.01)

Parameters:: std (float)
Return type:: None

clone()

Return independent dynamics state for one replicate.

Return type:: EnvironmentDynamics

supports(domain)

Parameters:: domain (RewardDomain)
Return type:: bool

class LinearContextualEnvironment[source]

Bases: object

Contextual simulator whose expected rewards are linear in features.

theta: ndarray[tuple[Any, ...], dtype[float64]]

context_provider: ContextProvider | Callable[[Generator], ndarray[tuple[Any, ...], dtype[float64]]]

reward_model: RewardModel

property n_arms: int

property n_features: int

property reward_domain: RewardDomain

property contextual: bool

context(rng)[source]

Parameters:: rng (Generator)
Return type:: ndarray[tuple[Any, …], dtype[float64]]

expected_rewards(context)[source]

Parameters:: context (ndarray[tuple[Any, ...], dtype[float64]])
Return type:: ndarray[tuple[Any, …], dtype[float64]]

sample_rewards(*, context, rng)[source]

Parameters:

context (ndarray[tuple[Any, ...], dtype[float64]])
rng (Generator)

Return type:

ndarray[tuple[Any, …], dtype[float64]]

clone()[source]

Return type:: LinearContextualEnvironment

__init__(theta, context_provider, reward_model=<factory>)

Parameters:

theta (ndarray[tuple[Any, ...], dtype[float64]])
context_provider (ContextProvider | Callable[[Generator], ndarray[tuple[Any, ...], dtype[float64]]])
reward_model (RewardModel)

Return type:

None

class LogisticContextualEnvironment[source]

Bases: LinearContextualEnvironment

Contextual Bernoulli simulator using a logistic link.

reward_model: RewardModel

expected_rewards(context)[source]

Parameters:: context (ndarray[tuple[Any, ...], dtype[float64]])
Return type:: ndarray[tuple[Any, …], dtype[float64]]

__init__(theta, context_provider, reward_model=<factory>)

Parameters:

theta (ndarray[tuple[Any, ...], dtype[float64]])
context_provider (ContextProvider | Callable[[Generator], ndarray[tuple[Any, ...], dtype[float64]]])
reward_model (RewardModel)

Return type:

None

clone()

Return type:: LinearContextualEnvironment

context(rng)

Parameters:: rng (Generator)
Return type:: ndarray[tuple[Any, …], dtype[float64]]

property contextual: bool

property n_arms: int

property n_features: int

property reward_domain: RewardDomain

sample_rewards(*, context, rng)

Parameters:

context (ndarray[tuple[Any, ...], dtype[float64]])
rng (Generator)

Return type:

ndarray[tuple[Any, …], dtype[float64]]

theta: ndarray[tuple[Any, ...], dtype[float64]]

context_provider: ContextProvider | Callable[[Generator], ndarray[tuple[Any, ...], dtype[float64]]]

class ProbabilityDrift[source]

Bases: EnvironmentDynamics

Gaussian random walk in log-odds space for probability means.

logit_std: float = 0.05

epsilon: float = 1e-09

supported_domains: frozenset[RewardDomain] = frozenset({RewardDomain.BINARY, RewardDomain.UNIT_INTERVAL})

apply(means, *, step, rng)[source]

Return the arm means for the supplied step.

Parameters:

means (ndarray[tuple[Any, ...], dtype[float64]])
step (int)
rng (Generator)

Return type:

ndarray[tuple[Any, …], dtype[float64]]

__init__(logit_std=0.05, epsilon=1e-09)

Parameters:

logit_std (float)
epsilon (float)

Return type:

None

clone()

Return independent dynamics state for one replicate.

Return type:: EnvironmentDynamics

supports(domain)

Parameters:: domain (RewardDomain)
Return type:: bool

class RandomArmSwap[source]

Bases: EnvironmentDynamics

RandomArmSwap(probability: ‘float’ = 0.2)

probability: float = 0.2

supported_domains: frozenset[RewardDomain] = frozenset({RewardDomain.BINARY, RewardDomain.REAL, RewardDomain.UNIT_INTERVAL})

apply(means, *, step, rng)[source]

Return the arm means for the supplied step.

Parameters:

means (ndarray[tuple[Any, ...], dtype[float64]])
step (int)
rng (Generator)

Return type:

ndarray[tuple[Any, …], dtype[float64]]

__init__(probability=0.2)

Parameters:: probability (float)
Return type:: None

clone()

Return independent dynamics state for one replicate.

Return type:: EnvironmentDynamics

supports(domain)

Parameters:: domain (RewardDomain)
Return type:: bool

class StationaryDynamics[source]

Bases: EnvironmentDynamics

StationaryDynamics()

supported_domains: frozenset[RewardDomain] = frozenset({RewardDomain.BINARY, RewardDomain.REAL, RewardDomain.UNIT_INTERVAL})

apply(means, *, step, rng)[source]

Return the arm means for the supplied step.

Parameters:

means (ndarray[tuple[Any, ...], dtype[float64]])
step (int)
rng (Generator)

Return type:

ndarray[tuple[Any, …], dtype[float64]]

__init__()

Return type:: None

clone()

Return independent dynamics state for one replicate.

Return type:: EnvironmentDynamics

supports(domain)

Parameters:: domain (RewardDomain)
Return type:: bool