Algorithm API

Textbook-oriented reinforcement-learning algorithms and policy utilities.

The available families are grouped into submodules:

• dynamic_programming: model-based policy evaluation, value iteration, and policy iteration.
• monte_carlo_methods: prediction and exploring-starts control using sampled episodes.
• temporal_difference_learning: tabular Sarsa(0) and Q-learning.
• linear_approximation: semi-gradient TD(0) prediction and Sarsa(0) control with feature-based value functions.
• eligibility_traces: semi-gradient Sarsa(lambda) control.
• policy_gradient_methods: REINFORCE and actor-critic with parameterized policies.

The policy and schedules submodules provide policy helpers and schedules for exploration rates and step sizes. Import functions from the specific submodule that defines them, for example::

from gym_classics2.algorithms.dynamic_programming import value_iteration

Dynamic programming

gym_classics2.algorithms.dynamic_programming

This file implements dynamic programming algorithms for solving Markov Decision Processes (MDPs) in tabular gym-classics2 environments with model access. The algorithms include value iteration and policy iteration, which are fundamental methods in reinforcement learning for computing optimal policies and value functions.

backup

backup(env, discount, V, state, action)

Computes a Bellman backup for a given state, action, and value function. This is the building block for the other algorithms.

Parameters:

Name	Type	Description	Default
env		A gym-classics environment with model access.	required
discount		The discount factor.	required
V		The current value function.	required
state		The current state.	required
action		The action to evaluate.	required

Returns: The computed Q-value for the given state and action.

Source code in gym_classics2/algorithms/dynamic_programming.py

def backup(env, discount, V, state, action):
    """Computes a Bellman backup for a given state, action, and value function. This is the building block
    for the other algorithms.

    Args:
        env: A gym-classics environment with model access.
        discount: The discount factor.
        V: The current value function.
        state: The current state.
        action: The action to evaluate.
    Returns:
        The computed Q-value for the given state and action.
    """

    V = np.array(V)

    next_states, rewards, terminals, probs = env.model(state, action)
    bootstraps = (1.0 - terminals) * V[next_states]
    return np.sum(probs * (rewards + discount * bootstraps))

value_iteration

value_iteration(env, discount, precision=0.001, history=False, verbose=False)

Performs value iteration to calculate the optimal value function.

Parameters:

Name	Type	Description	Default
env		A gym-classics environment with model access.	required
discount		The discount factor (0 <= discount <= 1).	required
precision		The precision for convergence (default: 1e-3).	0.001
history		If True, returns a list of intermediate value functions.	False
verbose		If True, prints progress information.	False

Returns:

Type	Description
	The optimal value function V. If history is True, returns a list of intermediate value functions.

Source code in gym_classics2/algorithms/dynamic_programming.py

def value_iteration(env, discount, precision=1e-3, history = False, verbose = False):
    """Performs value iteration to calculate the optimal value function.

    Args:
        env: A gym-classics environment with model access.
        discount: The discount factor (0 <= discount <= 1).
        precision: The precision for convergence (default: 1e-3).
        history: If True, returns a list of intermediate value functions.
        verbose: If True, prints progress information.

    Returns:
        The optimal value function V. If history is True, returns a list of intermediate value functions.
    """

    assert isinstance(env, GymClassicsBaseEnv), "Value iteration requires a gym-classics environment with model access to the environment." 
    assert 0.0 <= discount <= 1.0
    assert precision > 0.0

    V = np.zeros(len(env.states()), dtype=np.float64)  
    if history:
        V_list = []
        V_list.append(V.copy())

    sweeps = 0
    progress = tqdm(total=None, desc="Value Iteration", disable=verbose)
    while True:
        progress.update()
        if verbose:
            print('.', end = '')
            sweeps += 1

        V_old = V.copy()

        for s in env.states():
            Q_values = [backup(env, discount, V, s, a) for a in range(len(env.actions()))]
            V[s] = np.max(Q_values)

        if history:
            V_list.append(V.copy())

        if np.abs(V - V_old).max() <= precision:
            break

    if verbose:
        print(f'\nConverged after {sweeps} sweeps.')

    if history:
        return V_list 

    return V

policy_evaluation

policy_evaluation(env, discount, policy, precision=0.001, max_backups=1000, verbose=False)

Performs (modified) policy evaluation to approximate the given policy's value function. Regular policy evaluation does not limit the number of backups and converges to the true value funciton. Modified policy evaluation limits the number of backups and produces an estimate of the true value function. This has several advantages: It is faster and it prevents infinite loops for undiscounted problems where the evaluated policy does not reach a terminal state.

Policy iteration using modified policy evaluation still converges. Note: Value iteration is a special case of policy iteration where policy evaluation is limited to a single backup.

Parameters:

Name	Type	Description	Default
env		A gym-classics environment with model access.	required
discount		The discount factor (0 <= discount <= 1).	required
policy		The policy to evaluate.	required
precision		The precision for convergence (default: 1e-3).	0.001
max_backups		Maximum number of sweeps over all states (default: 1000). Limits computation and prevents infinite loops for undiscounted problems.	1000
verbose		If True, prints progress information.	False

Returns:

Type	Description
	The value function for the given policy.

Source code in gym_classics2/algorithms/dynamic_programming.py

def policy_evaluation(env, discount, policy, precision=1e-3, max_backups=1000, verbose=False):
    """Performs (modified) policy evaluation to approximate the given policy's value function.
    Regular policy evaluation does not limit the number of backups and converges to the true value funciton. 
    Modified policy evaluation limits the number of backups and produces an estimate of the true value function. 
    This has several advantages: It is faster and it
    prevents infinite loops for undiscounted problems where the evaluated policy does not reach a terminal state. 

    Policy iteration using modified policy evaluation still converges. Note: Value iteration is a special case of 
    policy iteration where policy evaluation is limited to a single backup.

    Args:
        env: A gym-classics environment with model access.
        discount: The discount factor (0 <= discount <= 1).
        policy: The policy to evaluate.
        precision: The precision for convergence (default: 1e-3).
        max_backups: Maximum number of sweeps over all states (default: 1000).
            Limits computation and prevents infinite loops for undiscounted problems.
        verbose: If True, prints progress information.

    Returns:
        The value function for the given policy.
    """

    assert isinstance(env, GymClassicsBaseEnv), "Value iteration requires a gym-classics environment with model access to the environment." 
    assert 0.0 <= discount <= 1.0
    assert precision > 0.0

    V = np.zeros(len(policy), dtype=np.float64)

    i = 0
    while i < max_backups:
        V_old = V.copy()

        for s in env.states():
            V[s] = backup(env, discount, V, s, policy[s])

        if np.abs(V - V_old).max() <= precision:
            break

        i += 1

    if verbose:
        if i < max_backups:
            print(f'\nConverged after {i} sweeps.')
        else:       
            print(f'\nReached maximum number of backups ({max_backups}) without convergence.')

    return V

policy_improvement

policy_improvement(policy, V_policy, env, discount, precision=0.001)

Improves the policy based on the given value function. It does this by calculating a Q-value function for all states and actions using Bellman updates and then extracts a greedy policy.

Policy evaluation adn policy improvement are the steps of policy iteration.

Parameters:

Name	Type	Description	Default
policy		The current policy to improve.	required
V_policy		The value function of the current policy.	required
env		A gym-classics environment with model access.	required
discount		The discount factor (0 <= discount <= 1).	required
precision		The precision for determining stability (default: 1e-3).	0.001

Returns:

Type	Description
	A tuple (improved_policy, stable) where stable is True if the policy did not change.

Source code in gym_classics2/algorithms/dynamic_programming.py

def policy_improvement(policy, V_policy, env, discount, precision=1e-3):
    """Improves the policy based on the given value function. It does this by 
    calculating a Q-value function for all states and actions using Bellman updates 
    and then extracts a greedy policy.

    Policy evaluation adn policy improvement are the steps of policy iteration.

    Args:
        policy: The current policy to improve.
        V_policy: The value function of the current policy.
        env: A gym-classics environment with model access.
        discount: The discount factor (0 <= discount <= 1).
        precision: The precision for determining stability (default: 1e-3).

    Returns:
        A tuple (improved_policy, stable) where stable is True if the policy did not change.
    """

    policy_old = policy.copy()
    V_old = V_policy.copy()

    for s in env.states():
        Q_values = [backup(env, discount, V_policy, s, a) for a in env.actions()]
        policy[s] = np.argmax(Q_values)
        V_policy[s] = max(Q_values)

    stable = np.logical_or(
        policy == policy_old,
        np.abs(V_policy - V_old).max() <= precision,
    ).all()

    return policy, stable

policy_iteration

policy_iteration(env, discount, precision=0.001, max_backups=1000, history=False, verbose=False, rng=None)

Performs policy iteration to find the optimal policy. It repeats modified policy evaluation followed by policy improvement till it converges to the optimal policy.

Using modified policy evaluation has several advantages: It is faster and it prevents infinite loops for undiscounted problems where the evaluated policy does not reach a terminal state. Policy iteration using modified policy evaluation still converges. Note: Value iteration is a special case of policy iteration where policy evaluation is limited to a single backup.

Parameters:

Name	Type	Description	Default
env		A gym-classics environment with model access.	required
discount		The discount factor (0 <= discount <= 1).	required
precision		The precision for convergence (default: 1e-3).	0.001
max_backups		Maximum number of iterations used in modified policy evaluation.	1000
history		If True, returns lists of intermediate policies and value functions.	False
verbose		If True, prints progress information.	False
rng		NumPy generator or integer seed used to initialize the policy.	None

Returns:

Type	Description
	The optimal policy. If history is True, returns a tuple (policy_list, V_list) containing lists of intermediate policies and value functions.

Source code in gym_classics2/algorithms/dynamic_programming.py

def policy_iteration(env, discount, precision=1e-3, max_backups=1000, history=False, verbose=False, rng=None):
    """Performs policy iteration to find the optimal policy. It repeats modified policy evaluation followed by
    policy improvement till it converges to the optimal policy.

    Using modified policy evaluation has several advantages: It is faster and it
    prevents infinite loops for undiscounted problems where the evaluated policy does not reach a terminal state.
    Policy iteration using modified policy evaluation still converges. 
    Note: Value iteration is a special case of policy iteration where policy evaluation is limited to a 
    single backup.  

    Args:
        env: A gym-classics environment with model access.
        discount: The discount factor (0 <= discount <= 1).
        precision: The precision for convergence (default: 1e-3).
        max_backups: Maximum number of iterations used in modified policy evaluation. 
        history: If True, returns lists of intermediate policies and value functions.
        verbose: If True, prints progress information.
        rng: NumPy generator or integer seed used to initialize the policy.

    Returns:
        The optimal policy. If history is True, returns a tuple (policy_list, V_list) containing lists of intermediate policies and value functions.
    """

    assert isinstance(env, GymClassicsBaseEnv), "Value iteration requires a gym-classics environment with model access to the environment." 
    assert 0.0 <= discount <= 1.0
    assert precision > 0.0

    policy = random_policy(env, rng=rng)

    if history:
        pol_list = []
        pol_list.append(policy.copy())
        V_list = []

    iterations = 0
    progress = tqdm(total=None, desc="Policy Iteration", disable=verbose)
    while True:
        progress.update()
        if verbose:
            print('.', end = '')
            iterations += 1

        V_policy = policy_evaluation(env, discount, policy, precision, max_backups)
        if history:
            V_list.append(V_policy.copy())

        policy, stable = policy_improvement(policy, V_policy, env, discount, precision)

        if stable:
            break

        if history:
            pol_list.append(policy.copy())

    if verbose:
        print(f'\nConverged after {iterations} iterations.')

    if history:
        return pol_list, V_list

    return policy

Policy helpers

gym_classics2.algorithms.policy

Create, convert, and select actions from tabular policies.

A deterministic tabular policy is a vector of integer action IDs indexed by state ID. For gym-classics2 environments with MultiDiscrete observations, policies can instead be dictionaries mapping coordinate tuples to action IDs.

Greedy policies can be derived from state values using model access or directly from action values. epsilon_greedy_action samples from action values rather than from a deterministic policy vector. Random choices accept a NumPy generator, an integer seed, or None through rng.

random_policy

random_policy(env, rng=None)

Create a deterministic policy by sampling one action per state.

Actions are drawn uniformly and independently when the policy is created. Looking up a state in the resulting policy always gives the same action.

Parameters:

Name	Type	Description	Default
env		Environment with a discrete action space. Observations must be Discrete, or MultiDiscrete in a gym-classics2 environment.	required
rng		NumPy generator, integer seed, or None for sampling actions.	None

Returns:

Type	Description
	NumPy array of action IDs indexed by state ID for Discrete observations,
	or a dictionary mapping coordinate tuples to action IDs for
	MultiDiscrete observations.

Source code in gym_classics2/algorithms/policy.py

def random_policy(env, rng=None):
    """Create a deterministic policy by sampling one action per state.

    Actions are drawn uniformly and independently when the policy is created.
    Looking up a state in the resulting policy always gives the same action.

    Args:
        env: Environment with a discrete action space. Observations must be
            Discrete, or MultiDiscrete in a gym-classics2 environment.
        rng: NumPy generator, integer seed, or None for sampling actions.

    Returns:
        NumPy array of action IDs indexed by state ID for Discrete observations,
        or a dictionary mapping coordinate tuples to action IDs for
        MultiDiscrete observations.
    """

    rng = get_rng(rng)

    if isinstance(env.observation_space, gym.spaces.Discrete):
        return rng.integers(env.action_space.n, size=env.observation_space.n)
    else:
        assert isinstance(env, GymClassicsBaseEnv), "Only gym-classics environments are supported for random policies with multi-discrete state spaces."
        return make_multidiscrete_policy(rng.integers(env.action_space.n, size=len(env.states())), env)

encode_policy

encode_policy(policy, env, type='text')

Convert a sequence of action IDs to labels for display.

Parameters:

Name	Type	Description	Default
policy		Deterministic policy vector containing integer action IDs.	required
env		gym-classics2 environment providing id2action.	required
type		Label format passed to env.id2action; defaults to "text". Available formats depend on the environment.	'text'

Returns:

Type	Description
	List of action labels in the same order as the input policy.

Source code in gym_classics2/algorithms/policy.py

def encode_policy(policy, env, type = "text"):
    """Convert a sequence of action IDs to labels for display.

    Args:
        policy: Deterministic policy vector containing integer action IDs.
        env: gym-classics2 environment providing ``id2action``.
        type: Label format passed to ``env.id2action``; defaults to ``"text"``.
            Available formats depend on the environment.

    Returns:
        List of action labels in the same order as the input policy.
    """

    assert isinstance(env, GymClassicsBaseEnv)

    return [env.unwrapped.id2action(a, type = type) for a in policy]

greedy_policy

greedy_policy(V, env, discount=1, rng=None)

Create a greedy policy by one-step look-ahead from state values.

For each state, select an action maximizing the expected reward plus the discounted next-state value. Terminal transitions do not bootstrap. Ties are broken uniformly at random when the policy is created.

Parameters:

Name	Type	Description	Default
V		One-dimensional state-value array indexed by state ID.	required
env		gym-classics2 environment with Discrete observation and action spaces and access to the transition and reward model.	required
discount		Discount factor in [0, 1].	1
rng		NumPy generator, integer seed, or None for random tie-breaking.	None

Returns:

Type	Description
	NumPy array of integer action IDs indexed by state ID.

Source code in gym_classics2/algorithms/policy.py

def greedy_policy(V, env, discount=1, rng=None):
    """Create a greedy policy by one-step look-ahead from state values.

    For each state, select an action maximizing the expected reward plus the
    discounted next-state value. Terminal transitions do not bootstrap.
    Ties are broken uniformly at random when the policy is created.

    Args:
        V: One-dimensional state-value array indexed by state ID.
        env: gym-classics2 environment with Discrete observation and action
            spaces and access to the transition and reward model.
        discount: Discount factor in ``[0, 1]``.
        rng: NumPy generator, integer seed, or None for random tie-breaking.

    Returns:
        NumPy array of integer action IDs indexed by state ID.
    """
    assert isinstance(env, GymClassicsBaseEnv), "greedy_policy requires a gym-classics environment with discrete state space."
    assert isinstance(env.action_space, gym.spaces.Discrete)
    assert isinstance(env.observation_space, gym.spaces.Discrete)
    assert 0.0 <= discount <= 1.0

    policy = np.zeros(len(env.states()), dtype=np.int64)
    rng = get_rng(rng)

    env = env.unwrapped

    for s in env.states():
        Q_values = [_backup(env, discount, V, s, a) for a in range(env.action_space.n)]
        policy[s] = random_argmax(Q_values, rng=rng)

    return policy

greedy_policy_Q

greedy_policy_Q(Q, env, discount=1, rng=None)

Create a greedy policy by maximizing action values in each state.

Ties are broken uniformly at random when the policy is created.

Parameters:

Name	Type	Description	Default
Q		Two-dimensional action-value array indexed by state and action.	required
env		Environment with discrete observation and action spaces.	required
discount		Must be in [0, 1] but does not affect action selection; retained for API compatibility with greedy_policy.	1
rng		NumPy generator, integer seed, or None for random tie-breaking.	None

Returns:

Type	Description
	NumPy array of integer action IDs indexed by state ID.

Source code in gym_classics2/algorithms/policy.py

def greedy_policy_Q(Q, env, discount=1, rng=None):
    """Create a greedy policy by maximizing action values in each state.

    Ties are broken uniformly at random when the policy is created.

    Args:
        Q: Two-dimensional action-value array indexed by state and action.
        env: Environment with discrete observation and action spaces.
        discount: Must be in ``[0, 1]`` but does not affect action selection;
            retained for API compatibility with ``greedy_policy``.
        rng: NumPy generator, integer seed, or None for random tie-breaking.

    Returns:
        NumPy array of integer action IDs indexed by state ID.
    """

    assert isinstance(env.action_space, gym.spaces.Discrete)
    assert isinstance(env.observation_space, gym.spaces.Discrete)
    assert 0.0 <= discount <= 1.0

    return random_argmax(Q, axis=1, rng=rng)

epsilon_greedy_action

epsilon_greedy_action(policy, state=None, epsilon=0, rng=None)

Select an epsilon-greedy action from tabular action values.

With probability epsilon, sample uniformly from all actions, including greedy ones. Otherwise, maximize the action values and break ties uniformly at random.

Parameters:

Name	Type	Description	Default
policy		One-dimensional action-value vector, or a table indexed by state and then action. Despite the name, this argument contains action values, not the action IDs of a deterministic policy.	required
state		Current state index. If omitted, policy is treated as the action-value vector for the current state.	None
epsilon		Exploration probability in [0, 1].	0
rng		NumPy generator, integer seed, or None for action sampling and tie-breaking.	None

Returns:

Type	Description
	Selected integer action ID.

Source code in gym_classics2/algorithms/policy.py

def epsilon_greedy_action(policy, state=None, epsilon=0, rng=None):
    """Select an epsilon-greedy action from tabular action values.

    With probability ``epsilon``, sample uniformly from all actions, including
    greedy ones. Otherwise, maximize the action values and break ties
    uniformly at random.

    Args:
        policy: One-dimensional action-value vector, or a table indexed by
            state and then action. Despite the name, this argument contains
            action values, not the action IDs of a deterministic policy.
        state: Current state index. If omitted, ``policy`` is treated as the
            action-value vector for the current state.
        epsilon: Exploration probability in ``[0, 1]``.
        rng: NumPy generator, integer seed, or None for action sampling
            and tie-breaking.

    Returns:
        Selected integer action ID.
    """

    rng = get_rng(rng)

    if epsilon > 0 and rng.random() < epsilon:
        return rng.integers(len(policy)) if state is None else rng.integers(len(policy[state]))

    if state is None:
        return random_argmax(policy, rng=rng)
    else:
        return random_argmax(policy[state], rng=rng)

make_multidiscrete_policy

make_multidiscrete_policy(policy, env)

Convert a policy indexed by state ID to a coordinate-keyed dictionary.

Parameters:

Name	Type	Description	Default
policy		Deterministic policy vector indexed by state ID, or an existing dictionary mapping coordinate tuples to action IDs.	required
env		Environment with MultiDiscrete observations. Converting a vector requires a gym-classics2 environment providing states and id2state.	required

Returns:

Type	Description
	Dictionary mapping coordinate tuples to integer action IDs. An input
	dictionary is returned unchanged, without copying or validating entries.

Raises:

Type	Description
AssertionError	If the observation space is not MultiDiscrete.
ValueError	If converting a vector for an unsupported environment type.

Source code in gym_classics2/algorithms/policy.py

def make_multidiscrete_policy(policy, env):
    """Convert a policy indexed by state ID to a coordinate-keyed dictionary.

    Args:
        policy: Deterministic policy vector indexed by state ID, or an existing
            dictionary mapping coordinate tuples to action IDs.
        env: Environment with MultiDiscrete observations. Converting a vector
            requires a gym-classics2 environment providing ``states`` and
            ``id2state``.

    Returns:
        Dictionary mapping coordinate tuples to integer action IDs. An input
        dictionary is returned unchanged, without copying or validating entries.

    Raises:
        AssertionError: If the observation space is not MultiDiscrete.
        ValueError: If converting a vector for an unsupported environment type.
    """
    assert isinstance(env.observation_space, gym.spaces.MultiDiscrete), "Requires an environment with a multi-discrete state spaces."

    if isinstance(policy, dict):
        return policy

    if isinstance(env, GymClassicsBaseEnv):
        return dict(zip(env.id2state(list(env.states())), [policy[s] for s in env.states()]))
    else:
        raise ValueError("Unsupported environment type")

Monte Carlo methods

gym_classics2.algorithms.monte_carlo_methods

This file implements tabular Monte Carlo methods for policy evaluation and control for environments with discrete state spaces.

sample_episode

sample_episode(env, policy=None, start_state=None, start_action=None, epsilon=0, max_len=1000, verbose=False, rng=None)

Samples an episode from the environment using the given policy and starting conditions.

Parameters:

Name	Type	Description	Default
env		The environment to sample from.	required
policy		A mapping from states to actions. For discrete state spaces, use an action list in state order. If None, a random policy is used.	None
start_state		The state to start the episode from (if None, the environment's default starting state will be used).	None
start_action		The action to take in the first step of the episode (if None, the action will be chosen according to the policy or randomly if no policy is given).	None
epsilon		The probability of taking a random action instead of the policy's action at each step (for epsilon-greedy exploration).	0
max_len		The maximum length of the episode to prevent infinite loops.	1000
verbose		If True, prints the state transitions and rewards for each step in the episode.	False
rng		NumPy generator or integer seed for policy and exploration choices.	None

Returns:

Type	Description
	A list of (state, action, reward, next_state) tuples representing the episode.
	The episode ends when a terminal state is reached or when max_len steps have been taken.
	Note that the last tuple in the episode will have a next_state that is either terminal or the
	state at which the episode was truncated due to max_len.

Source code in gym_classics2/algorithms/monte_carlo_methods.py

def sample_episode(env, policy=None, start_state=None, start_action=None, epsilon=0,
                   max_len=1000, verbose=False, rng=None):
    """
    Samples an episode from the environment using the given policy and starting conditions.

    Args: 
        env: The environment to sample from.
        policy: A mapping from states to actions. For discrete state spaces, use
            an action list in state order. If ``None``, a random policy is used.
        start_state: The state to start the episode from (if None, the environment's default starting state will be used).
        start_action: The action to take in the first step of the episode (if None, the action will be chosen according to the policy or randomly if no policy is given).
        epsilon: The probability of taking a random action instead of the policy's action at each step (for epsilon-greedy exploration).
        max_len: The maximum length of the episode to prevent infinite loops.
        verbose: If True, prints the state transitions and rewards for each step in the episode.
        rng: NumPy generator or integer seed for policy and exploration choices.

    Returns:
        A list of (state, action, reward, next_state) tuples representing the episode. 
        The episode ends when a terminal state is reached or when max_len steps have been taken.    
        Note that the last tuple in the episode will have a next_state that is either terminal or the 
        state at which the episode was truncated due to max_len.  
    """

    assert isinstance(env.observation_space, gym.spaces.Discrete) or isinstance(env.observation_space, gym.spaces.MultiDiscrete), "Tabular methods require discrete state space."  
    assert 0.0 <= epsilon <= 1.0
    assert max_len > 0

    rng = get_rng(rng)

    if policy is None:
        policy = random_policy(env, rng=rng)
        epsilon = 1.0
        if verbose:
            print("*** No policy given, sampling using random actions!")

    if isinstance(env.observation_space, gym.spaces.MultiDiscrete):
        policy = make_multidiscrete_policy(policy, env)

    episode = []
    s, r = env.reset()

    # force custom start state could be a state or an state index.
    if not start_state is None:
        if isinstance(env, BaseEnv) and env.tabular:
            env.state = env.id2state(start_state)
        elif isinstance(env.observation_space, gym.spaces.Discrete):
            env.state = int(start_state)
        else:
            raise ValueError("Custom start state is only supported for tabular gym-classics environments and environments with a MultiDiscrete observation space.")
        s = start_state

    step = 0
    done = False
    while not done and step < max_len:
        a = policy[s]

        # epsilon greedy choice?
        if epsilon > 0 and rng.random() < epsilon:
            a = rng.integers(env.action_space.n)

        # for exploring starts
        if step == 0 and not start_action is None:
            a = start_action

        sp, reward, done, _, _ = env.step(a)

        episode.append((s,a,reward,sp))

        if verbose:
            print ("Step", step, "(s,a,r,s')=",s,a,reward,sp)

        s = sp
        step += 1

    return(episode)

on_policy_state_distribution

on_policy_state_distribution(policy, env, discount=1, epsilon=0, n=100, verbose=False, rng=None)

Estimate a policy's state distribution using rng for exploration.

Source code in gym_classics2/algorithms/monte_carlo_methods.py

def on_policy_state_distribution(policy, env, discount=1, epsilon=0, n=100,
                                 verbose=False, rng=None):
    """Estimate a policy's state distribution using ``rng`` for exploration."""

    assert isinstance(env.observation_space, gym.spaces.Discrete), "Tabular methods require discrete state space."   
    assert 0.0 <= epsilon <= 1.0
    assert 0.0 < discount <= 1.0
    assert n > 0

    rng = get_rng(rng)
    state_counts = np.zeros(env.observation_space.n)

    for _ in tqdm(range(n), desc="Sampling Episodes", disable=verbose):
        episode = np.array(sample_episode(env, policy=policy, epsilon=epsilon, rng=rng))
        states = np.append(episode[:,0], episode[-1,3])
        discounts = np.array([discount**i for i in range(len(states))])
        for s, d in zip(states, discounts):
            state_counts[int(s)] += d

    state_prob = state_counts / sum(state_counts)
    return state_prob

MC_prediction

MC_prediction(policy, env, discount, n=100, max_episode_len=100, verbose=False, rng=None)

Estimate a policy's state values with first-visit Monte Carlo prediction.

Parameters:

Name	Type	Description	Default
policy		Array-like mapping from state IDs to action IDs.	required
env		Gymnasium environment with a discrete observation space.	required
discount		Reward discount factor.	required
n		Number of episodes to sample.	100
max_episode_len		Maximum sampled steps per episode.	100
verbose		Print episode progress when true.	False
rng		NumPy generator or integer seed for episode sampling.	None

Returns:

Type	Description
	A NumPy value array with one entry per state. Unvisited states contain
	numpy.nan.

Source code in gym_classics2/algorithms/monte_carlo_methods.py

def MC_prediction(policy, env, discount, n=100, max_episode_len=100,
                  verbose=False, rng=None):
    """Estimate a policy's state values with first-visit Monte Carlo prediction.

    Args:
        policy: Array-like mapping from state IDs to action IDs.
        env: Gymnasium environment with a discrete observation space.
        discount: Reward discount factor.
        n: Number of episodes to sample.
        max_episode_len: Maximum sampled steps per episode.
        verbose: Print episode progress when true.
        rng: NumPy generator or integer seed for episode sampling.

    Returns:
        A NumPy value array with one entry per state. Unvisited states contain
        ``numpy.nan``.
    """
    assert isinstance(env.observation_space, gym.spaces.Discrete), "Tabular methods require discrete state space."  
    assert n > 0
    assert max_episode_len > 0

    rng = get_rng(rng)
    Returns = defaultdict(list) # a list for each s

    for i in tqdm(range(n), desc="MC Prediction", disable=verbose):
        if verbose:
            print("episode", i," of ", n)

        episode = sample_episode(env, policy, max_len=max_episode_len, rng=rng)
        G = 0

        # for first visit check for s
        visited = [s for s,a,r,sp in episode]

        # process episode in reverse order
        for t in range(len(episode)-1, -1, -1):
            s,a,r,sp = episode[t]
            G = discount * G + r

            # use first visit of s only
            if not s in visited[0:t]:
                Returns[s].append(G)

    Vs = np.array([np.mean(Returns[s]) if len(Returns[s]) else np.nan for s in range(env.observation_space.n)])
    return Vs

MC_control_ES_textbook

MC_control_ES_textbook(env, discount, n=100, Q=None, max_episode_len=100, history=False, verbose=False, rng=None)

Monte Carlo control with exploring starts and stored sample returns.

This direct textbook implementation stores every first-visit return. Prefer :func:MC_control_ES for larger experiments because its running-average update uses less memory.

Parameters:

Name	Type	Description	Default
env		A tabular gym_classics2 environment.	required
discount		Reward discount factor.	required
n		Number of episodes to sample.	100
Q		Optional initial action-value array.	None
max_episode_len		Maximum sampled steps per episode.	100
history		Retain intermediate policies, Q arrays, episodes, and returns.	False
verbose		Print episode details; values greater than one print transitions.	False
rng		NumPy generator or integer seed for all algorithm choices.	None

Returns:

Type	Description
	(policy, Q). If history=True, a third item contains the history dictionary with policies, Q_values, episodes, and returns.

Source code in gym_classics2/algorithms/monte_carlo_methods.py

def MC_control_ES_textbook(env, discount, n=100, Q=None, max_episode_len=100,
                           history=False, verbose=False, rng=None):
    """Monte Carlo control with exploring starts and stored sample returns.

    This direct textbook implementation stores every first-visit return. Prefer
    :func:`MC_control_ES` for larger experiments because its running-average
    update uses less memory.

    Args:
        env: A tabular ``gym_classics2`` environment.
        discount: Reward discount factor.
        n: Number of episodes to sample.
        Q: Optional initial action-value array.
        max_episode_len: Maximum sampled steps per episode.
        history: Retain intermediate policies, Q arrays, episodes, and returns.
        verbose: Print episode details; values greater than one print transitions.
        rng: NumPy generator or integer seed for all algorithm choices.

    Returns:
        ``(policy, Q)``. If ``history=True``, a third item contains the history dictionary with ``policies``, ``Q_values``, ``episodes``, and ``returns``.
    """
    assert isinstance(env.observation_space, gym.spaces.Discrete), "Tabular methods require discrete state space."  
    assert n > 0
    assert max_episode_len > 0
    assert Q is None or Q.shape == (len(env.states()), env.action_space.n)

    rng = get_rng(rng)
    policy = random_policy(env, rng=rng)

    if Q is None:
        Q = np.zeros((len(env.states()), env.action_space.n))

    # lists that grow are very slow. We should use a running average instead. But this is easier to read and understand.
    Returns = defaultdict(list) # a list for each (s,a)

    if history:
        Q_list = []
        Q_list.append(Q.copy())
        pol_list = []
        pol_list.append(policy.copy())
        ep_list = []
        return_list = []


    for i in tqdm(range(n), desc="MC Control", disable=verbose):
        if verbose:
            print("episode", i," of ", n)

        # Sample starting (s,a)
        s = rng.integers(env.observation_space.n)
        a = rng.integers(env.action_space.n)

        episode = sample_episode(env, policy, start_state=s, start_action=a,
                                 max_len=max_episode_len, verbose=verbose > 1,
                                 rng=rng)
        G = 0

        # for first visit check for (s,a)
        visited = [(s,a) for s,a,r,sp in episode]

        # process episode in reverse order
        for t in range(len(episode)-1, -1, -1):
            s,a,r,sp = episode[t]
            G = discount * G + r

            # use first visit of (s,a) only
            if not (s,a) in visited[0:t]:
                Returns[(s,a)].append(G)

                # update policy
                Q[s,a] = np.mean(Returns[(s,a)])
                policy[s] = random_argmax(Q[s, :], rng=rng)

        if history:
            pol_list.append(policy.copy())
            Q_list.append(Q.copy())
            ep_list.append(episode.copy())
            return_list.append(G)

    if history:
        return policy, Q, {'policies': pol_list, 'Q_values': Q_list, 'episodes': ep_list, 'returns': return_list}

    return policy, Q

MC_control_ES

MC_control_ES(env, discount, n=100, Q=None, max_episode_len=100, history=False, verbose=False, rng=None)

Monte Carlo Control with Exploring Starts (incremental version). This algorithm estimates the optimal action-value function Q and the corresponding greedy policy by sampling episodes with exploring starts. It uses incremental updates to compute the average returns for each (s,a) pair, which is more memory efficient than storing all returns.

Parameters:

Name	Type	Description	Default
env		A tabular gym_classics2 environment.	required
discount		Reward discount factor.	required
n		Number of episodes to sample.	100
Q		Optional initial action-value array.	None
max_episode_len		Maximum sampled steps per episode.	100
history		Retain intermediate policies, Q arrays, episodes, and returns. Note: retaining the history may require a lot of memory.	False
verbose		Print episode details; values greater than one print transitions.	False
rng		NumPy generator or integer seed for all algorithm choices.	None

Returns:

Type	Description
	(policy, Q). If history=True, a third item contains the history dictionary with policies, Q_values, episodes, and returns.

Source code in gym_classics2/algorithms/monte_carlo_methods.py

def MC_control_ES(env, discount, n=100, Q=None, max_episode_len=100,
                  history=False, verbose=False, rng=None):
    """Monte Carlo Control with Exploring Starts (incremental version).
    This algorithm estimates the optimal action-value function Q and the corresponding greedy policy by sampling episodes with exploring starts. It uses incremental updates to compute the average returns for each (s,a) pair, which is more memory efficient than storing all returns.

    Args:
        env: A tabular ``gym_classics2`` environment.
        discount: Reward discount factor.
        n: Number of episodes to sample.
        Q: Optional initial action-value array.
        max_episode_len: Maximum sampled steps per episode.
        history: Retain intermediate policies, Q arrays, episodes, and returns. Note: retaining the history may require a lot of memory.
        verbose: Print episode details; values greater than one print transitions.
        rng: NumPy generator or integer seed for all algorithm choices.

    Returns:
        ``(policy, Q)``. If ``history=True``, a third item contains the history dictionary with ``policies``, ``Q_values``, ``episodes``, and ``returns``.
    """

    assert isinstance(env.observation_space, gym.spaces.Discrete), "Tabular methods require discrete state space."  
    assert n > 0
    assert max_episode_len > 0
    assert Q is None or Q.shape == (env.observation_space.n, env.action_space.n)

    rng = get_rng(rng)
    policy = random_policy(env, rng=rng)

    if Q is None:
        Q = np.zeros((env.observation_space.n, env.action_space.n))

    # Count how many first-visit returns have been used for each (s, a)
    N = np.zeros((env.observation_space.n, env.action_space.n), dtype=int)

    if history:
        Q_list = []
        Q_list.append(Q.copy())
        pol_list = []
        pol_list.append(policy.copy())
        ep_list = []
        return_list = []

    for i in tqdm(range(n), desc="MC Control (Incremental)", disable=verbose):
        if verbose:
            print("episode", i, "of", n)

        # Exploring starts
        s = rng.integers(env.observation_space.n)
        a = rng.integers(env.action_space.n)

        episode = sample_episode(
            env,
            policy,
            start_state=s,
            start_action=a,
            max_len=max_episode_len,
            verbose=verbose > 1,
            rng=rng,
        )

        G = 0

        # Process episode backward
        visited = set()
        for t in range(len(episode) - 1, -1, -1):
            s, a, r, sp = episode[t]
            G = discount * G + r

            # First-visit MC: only update the first occurrence from the start
            if (s, a) not in visited:
                visited.add((s, a))

                N[s, a] += 1
                Q[s, a] += (G - Q[s, a]) / N[s, a]

                # Improve policy greedily
                policy[s] = random_argmax(Q[s, :], rng=rng)

        if history:
            pol_list.append(policy.copy())
            Q_list.append(Q.copy())
            ep_list.append(episode.copy())
            return_list.append(G)

    if history:
        return policy, Q, {'policies': pol_list, 'Q_values': Q_list, 'episodes': ep_list, 'returns': return_list}

    return policy, Q

Temporal-difference learning

gym_classics2.algorithms.temporal_difference_learning

This file implements tabular temporal difference learning algorithms for policy evaluation and control for discrete environments. The algorithms include Sarsa(0) and Q-learning.

Sarsa_0

Sarsa_0(env, discount, alpha, epsilon, Q=None, n=100, verbose=False, history=False, rng=None)

Learn action values with one-step on-policy Sarsa.

Parameters:

Name	Type	Description	Default
env		Gymnasium environment with discrete observation and action spaces.	required
discount		Reward discount factor in [0, 1].	required
alpha		Scalar step size or Schedule evaluated once per episode.	required
epsilon		Scalar exploration probability or Schedule evaluated once per episode.	required
Q		Optional initial action-value array shaped (env.observation_space.n, env.action_space.n). The array is updated in place.	None
n		Number of training episodes.	100
verbose		Print individual updates when true.	False
history		Retain Q arrays, discounted episode returns, and episode lengths.	False
rng		NumPy generator or integer seed for exploration and tie-breaking.	None

Returns:

Type	Description
	The learned Q array. If history=True, returns (Q, history_dict);
	the dictionary contains Qs, returns, and ep_lens.

Raises:

Type	Description
AssertionError	If the observation space is not discrete.

Source code in gym_classics2/algorithms/temporal_difference_learning.py

def Sarsa_0(env, discount, alpha, epsilon, Q=None, n=100, verbose=False,
            history=False, rng=None):
    """Learn action values with one-step on-policy Sarsa.

    Args:
        env: Gymnasium environment with discrete observation and action spaces.
        discount: Reward discount factor in ``[0, 1]``.
        alpha: Scalar step size or ``Schedule`` evaluated once per episode.
        epsilon: Scalar exploration probability or ``Schedule`` evaluated once
            per episode.
        Q: Optional initial action-value array shaped
            ``(env.observation_space.n, env.action_space.n)``. The array is updated
            in place.
        n: Number of training episodes.
        verbose: Print individual updates when true.
        history: Retain Q arrays, discounted episode returns, and episode lengths.
        rng: NumPy generator or integer seed for exploration and tie-breaking.

    Returns:
        The learned Q array. If ``history=True``, returns ``(Q, history_dict)``;
        the dictionary contains ``Qs``, ``returns``, and ``ep_lens``.

    Raises:
        AssertionError: If the observation space is not discrete.
    """
    assert isinstance(env.observation_space, gym.spaces.Discrete), "Tabular methods require discrete state space."  

    rng = get_rng(rng)

    if not isinstance(alpha, Schedule):
        alpha = ConstantSchedule(alpha)
    if not isinstance(epsilon, Schedule):
        epsilon = ConstantSchedule(epsilon)

    if Q is None:
        Q = np.zeros((env.observation_space.n, env.action_space.n))

    if history:
        Q_list = []
        Q_list.append(Q.copy())
        return_list = []
        ep_len_list = []


    for i in tqdm(range(n), desc="Sarsa", disable=verbose):
        s, r = env.reset()

        if verbose:
            print(f"--- Episode {i} ---")      

        if rng.random() > epsilon(i):
            a = random_argmax(Q[s, :], rng=rng)
        else:
            a = rng.integers(env.action_space.n)

        t = 0
        done = False
        G = 0
        while not done:            
            sp, r, done, _, _ = env.step(a)
            G += r * pow(discount, t)
            t += 1


            if rng.random() > epsilon(i):
                ap = random_argmax(Q[sp, :], rng=rng)
            else:
                ap = rng.integers(env.action_space.n)

            Q[s,a] = Q[s,a] + alpha(i) * (r + discount * Q[sp,ap] - Q[s,a])

            if verbose:
                print(f"{t} - sarsa: {s},{a},{r},{sp},{ap}, - new Q(s,a): {Q[s,a]}")
                if done:
                    print("Total return:", G)

            s = sp
            a = ap

        if history:
            Q_list.append(Q.copy())
            return_list.append(G)   
            ep_len_list.append(t)

    if history:
        return Q, {'Qs': Q_list, 'returns': return_list, 'ep_lens': ep_len_list}

    return Q

Q_learning

Q_learning(env, discount, alpha, epsilon, Q=None, n=100, verbose=False, history=False, rng=None)

Learn action values with one-step off-policy Q-learning.

Parameters:

Name	Type	Description	Default
env		Gymnasium environment with discrete observation and action spaces.	required
discount		Reward discount factor in [0, 1].	required
alpha		Scalar step size or Schedule evaluated once per episode.	required
epsilon		Scalar exploration probability or Schedule evaluated once per episode.	required
Q		Optional initial action-value array shaped (env.observation_space.n, env.action_space.n). The array is updated in place.	None
n		Number of training episodes.	100
verbose		Print progress information when true.	False
history		Retain Q arrays, discounted episode returns, episode lengths, and state-visit counts.	False
rng		NumPy generator or integer seed for exploration and tie-breaking.	None

Returns:

Type	Description
	The learned Q array. If history=True, returns (Q, history_dict);
	the dictionary contains Qs, returns, ep_lens, and
	state_visits.

Raises:

Type	Description
AssertionError	If the observation space is not discrete.

Source code in gym_classics2/algorithms/temporal_difference_learning.py

def Q_learning(env, discount, alpha, epsilon, Q=None, n=100, verbose=False,
               history=False, rng=None):
    """Learn action values with one-step off-policy Q-learning.

    Args:
        env: Gymnasium environment with discrete observation and action spaces.
        discount: Reward discount factor in ``[0, 1]``.
        alpha: Scalar step size or ``Schedule`` evaluated once per episode.
        epsilon: Scalar exploration probability or ``Schedule`` evaluated once
            per episode.
        Q: Optional initial action-value array shaped
            ``(env.observation_space.n, env.action_space.n)``. The array is updated
            in place.
        n: Number of training episodes.
        verbose: Print progress information when true.
        history: Retain Q arrays, discounted episode returns, episode lengths, and
            state-visit counts.
        rng: NumPy generator or integer seed for exploration and tie-breaking.

    Returns:
        The learned Q array. If ``history=True``, returns ``(Q, history_dict)``;
        the dictionary contains ``Qs``, ``returns``, ``ep_lens``, and
        ``state_visits``.

    Raises:
        AssertionError: If the observation space is not discrete.
    """
    assert isinstance(env.observation_space, gym.spaces.Discrete), "Tabular methods require discrete state space."  

    rng = get_rng(rng)

    if not isinstance(alpha, Schedule):
        alpha = ConstantSchedule(alpha)
    if not isinstance(epsilon, Schedule):
        epsilon = ConstantSchedule(epsilon)

    if Q is None:
        Q = np.zeros((env.observation_space.n, env.action_space.n))

    if history:
        Q_list = []
        Q_list.append(Q.copy())
        return_list = []
        ep_len_list = []
        #state_visits = np.zeros(env.observation_space.n, dtype=int)
        # Note we use float so visualization works better
        state_visits = np.zeros(env.observation_space.n, dtype=float)

    for i in tqdm(range(n), desc="Q-Learning", disable=verbose):
        s, r = env.reset()

        if history:
            state_visits[s] += 1

        done = False
        G = 0
        t = 0   
        while not done:
            # epsilon-greedy choice w.r.t. Q 
            if rng.random() > epsilon(i):
                a = random_argmax(Q[s, :], rng=rng)
            else:
                a = rng.integers(env.action_space.n)

            sp, r, done, _, _ = env.step(a)

            if history:
                state_visits[sp] += 1

            if history:
                G += r * pow(discount, t)
                t += 1

            Q[s,a] = Q[s,a] + alpha(i) * (r + discount * np.max(Q[sp,:]) - Q[s,a])

            s = sp

        if history:
            Q_list.append(Q.copy())
            return_list.append(G)
            ep_len_list.append(t)

    if history:
        return Q, {'Qs': Q_list, 'returns': return_list, 'ep_lens': ep_len_list, 'state_visits': state_visits}

    return Q

Linear function approximation

gym_classics2.algorithms.linear_approximation

Linear function approximation algorithms for policy evaluation and control.

Implements linear v and q-value function approximation, prediction with semi-gradient TD(0), control with semi-gradient Sarsa(0) and feature construction with Fourier basis functions.

The algorithms do not require discrete state spaces. Callers have to provide a state_features(state, env) function that converts states to state feature vectors x(s).

For an environment where the state is a set of numbers, e.g., a x- and y-coordinate. The leading 1 is the intercept.

def state_features(s, env):
    return np.concatenate([np.array([1]), np.array(s)])

state_action_features

state_action_features(s, a, env, state_features)

Construct the state-action feature vector x(s, a).

The state feature vector is expected to have the form [1, x1, ..., xd], with a leading intercept. The returned vector keeps one s hared intercept and has a separate block of the remaining state features for each action. Only the intercept and the weights for a are selected; all other elements are set to zero.

Parameters:

Name	Type	Description	Default
s		State to represent.	required
a		Integer ID of the action to represent.	required
env		Environment providing the discrete action space.	required
state_features		Callable that returns [1, x1, ..., xd] for (s, env).	required

Returns:

Type	Description
	Block-coded NumPy feature vector for the state-action pair (s, a).

Source code in gym_classics2/algorithms/linear_approximation.py

def state_action_features(s, a, env, state_features):
    """Construct the state-action feature vector ``x(s, a)``.

    The state feature vector is expected to have the form
    ``[1, x1, ..., xd]``, with a leading intercept. The returned vector keeps
    one s   hared intercept and has a separate block of the remaining state
    features for each action. Only the intercept and the weights for ``a`` are
    selected; all other elements are set to zero.

    Args:
        s: State to represent.
        a: Integer ID of the action to represent.
        env: Environment providing the discrete action space.
        state_features: Callable that returns ``[1, x1, ..., xd]`` for
            ``(s, env)``.

    Returns:
        Block-coded NumPy feature vector for the state-action pair ``(s, a)``.
    """
    def active_weights(a, sf_len):
        return [0] + list(range(a*sf_len+1, a*sf_len+sf_len+1))

    s = state_features(s, env)
    x = np.zeros(1+len(s)*env.action_space.n)
    x[active_weights(a, len(s)-1)] = s
    return x

v_hat

v_hat(s, w, env, state_features)

Compute the linear state-value approximation v_hat(s, w).

Approximates the state value as w^T x(s), the weighted sum of the components in the state feature vector x(s). The weight and feature vectors must have the same length.

Parameters:

Name	Type	Description	Default
s		State to evaluate.	required
w		Weight vector of the linear approximator.	required
env		Environment containing the state.	required
state_features		Callable returning the feature vector x(s) for (s, env).	required

Returns:

Type	Description
	Scalar estimate of the expected return from s.

Source code in gym_classics2/algorithms/linear_approximation.py

def v_hat(s, w, env, state_features):
    """Compute the linear state-value approximation ``v_hat(s, w)``.

    Approximates the state value as ``w^T x(s)``, the weighted sum of the components in the
    state feature vector ``x(s)``. The weight and feature vectors must have the
    same length.

    Args:
        s: State to evaluate.
        w: Weight vector of the linear approximator.
        env: Environment containing the state.
        state_features: Callable returning the feature vector ``x(s)`` for
            ``(s, env)``.

    Returns:
        Scalar estimate of the expected return from ``s``.
    """
    return np.dot(w, state_features(s, env))

q_hat

q_hat(s, a, w, env, state_features)

Compute the linear action-value approximation q_hat(s, a, w).

Estimates the q-value as w^T x(s, a) using the block-coded state-action features produced by state_action_features(). The weight vector must have the same length as that feature vector.

Parameters:

Name	Type	Description	Default
s		State to evaluate.	required
a		Integer ID of the action to evaluate.	required
w		Weight vector of the linear approximator.	required
env		Environment providing the discrete action space.	required
state_features		Callable returning a state feature vector with a leading intercept for (s, env).	required

Returns:

Type	Description
	Scalar estimate of the expected return from taking a in s.

Source code in gym_classics2/algorithms/linear_approximation.py

def q_hat(s, a, w, env, state_features):
    """Compute the linear action-value approximation ``q_hat(s, a, w)``.

    Estimates the q-value as ``w^T x(s, a)`` using the block-coded state-action
    features produced by `state_action_features()`. The weight vector must
    have the same length as that feature vector.

    Args:
        s: State to evaluate.
        a: Integer ID of the action to evaluate.
        w: Weight vector of the linear approximator.
        env: Environment providing the discrete action space.
        state_features: Callable returning a state feature vector with a
            leading intercept for ``(s, env)``.

    Returns:
        Scalar estimate of the expected return from taking ``a`` in ``s``.
    """    
    x = state_action_features(s, a, env, state_features)
    return np.dot(w, x)

epsilon_greedy_action_w

epsilon_greedy_action_w(s, w, env, state_features, epsilon=0, rng=None)

Select an epsilon-greedy action from an approximate action-values function.

Parameters:

Name	Type	Description	Default
s		Current state.	required
w		Weight vector for the action-value approximator.	required
env		Environment providing the discrete action space.	required
state_features		Callable converting (state, env) to a feature vector.	required
epsilon		Probability of selecting a uniformly random action.	0
rng		NumPy generator or integer seed.	None

Returns:

Type	Description
	Selected integer action ID.

Source code in gym_classics2/algorithms/linear_approximation.py

def epsilon_greedy_action_w(
    s, w, env, state_features, epsilon=0, rng=None
):
    """Select an epsilon-greedy action from an approximate action-values function.

    Args:
        s: Current state.
        w: Weight vector for the action-value approximator.
        env: Environment providing the discrete action space.
        state_features: Callable converting ``(state, env)`` to a feature vector.
        epsilon: Probability of selecting a uniformly random action.
        rng: NumPy generator or integer seed.

    Returns:
        Selected integer action ID.
    """

    rng = get_rng(rng)

    if epsilon > 0 and rng.random() < epsilon:
        return rng.integers(env.action_space.n)

    return random_argmax(
        [
            q_hat(s, a, w, env, state_features)
            for a in range(env.action_space.n)
        ],
        rng=rng,
    )

MSVE

MSVE(V, V_true, weight=None)

Calculate the weighted mean squared value error.

Parameters:

Name	Type	Description	Default
V		Estimated value for each state.	required
V_true		Reference value for each state.	required
weight		Weight for each state, typically its stationary visitation probability. If omitted, use unit weights.	None

Returns:

Type	Description
	Weighted sum of squared value errors.

Source code in gym_classics2/algorithms/linear_approximation.py

def MSVE(V, V_true, weight=None):
    """Calculate the weighted mean squared value error.

    Args:
        V: Estimated value for each state.
        V_true: Reference value for each state.
        weight: Weight for each state, typically its stationary visitation
            probability. If omitted, use unit weights.

    Returns:
        Weighted sum of squared value errors.
    """
    if weight is None:
        weight = np.ones(len(V))

    return np.sum(weight * (V - V_true)**2)

semi_gradient_TD0

semi_gradient_TD0(env, state_features, policy, n, alpha, gamma, max_episode_length=1000, verbose=False)

Estimate state values with semi-gradient TD(0).

This function runs TD(0) learning with function approximation over multiple episodes generated from a given policy and environment. Updates are performed using the semi-gradient of the value function approximation.

Parameters:

Name	Type	Description	Default
env		Episodic Gymnasium environment used to generate experience.	required
state_features		Callable converting (state, env) to a feature vector.	required
policy		Deterministic policy indexed by state. For multidiscrete gym-classics environments, a policy array in state-ID order is converted to a dictionary indexed by coordinate states.	required
n		Number of training episodes.	required
alpha		Step size or schedule.	required
gamma		Discount factor in [0, 1].	required
max_episode_length		Maximum number of steps per episode.	1000
verbose		Whether to print step-by-step diagnostics.	False

Returns:

Type	Description
	Learned weight vector for the approximate value function.

Source code in gym_classics2/algorithms/linear_approximation.py

def semi_gradient_TD0(
    env,
    state_features,
    policy,
    n,
    alpha,
    gamma,
    max_episode_length=1000,
    verbose=False,
):
    """Estimate state values with semi-gradient TD(0).

    This function runs TD(0) learning with function approximation over multiple
    episodes generated from a given policy and environment. Updates are performed
    using the semi-gradient of the value function approximation.

    Args:
        env: Episodic Gymnasium environment used to generate experience.
        state_features: Callable converting ``(state, env)`` to a feature vector.
        policy: Deterministic policy indexed by state. For multidiscrete
            gym-classics environments, a policy array in state-ID order is
            converted to a dictionary indexed by coordinate states.
        n: Number of training episodes.
        alpha: Step size or schedule.
        gamma: Discount factor in ``[0, 1]``.
        max_episode_length: Maximum number of steps per episode.
        verbose: Whether to print step-by-step diagnostics.

    Returns:
        Learned weight vector for the approximate value function.
    """
    assert gamma >= 0 and gamma <= 1, "Gamma must be in [0,1]"
    assert n > 0, "Number of episodes must be positive"
    assert max_episode_length > 0, "Max episode length must be positive"

    if isinstance(env.observation_space, gym.spaces.Discrete):
        warnings.warn("The environment has a discrete state space. Consider using a tabular method instead of function approximation.")

    if isinstance(env.observation_space, gym.spaces.MultiDiscrete):
        policy = make_multidiscrete_policy(policy, env)

    if not isinstance(alpha, Schedule):
        alpha = ConstantSchedule(alpha)

    state, _ = env.reset()
    w = np.zeros(len(state_features(state, env)))  # Initialize weights (intercept + x and y)

    for episode in tqdm(range(n), desc="Semi-Gradient TD(0)", disable=verbose):
        state, _ = env.reset()
        done = False

        i = 0
        while not done and i < max_episode_length:
            action = policy[state]  # follow policy
            next_state, reward, terminated, truncated, _ = env.step(action)
            done = terminated or truncated

            # Semi-gradient TD(0) update
            # Note: v_hat(terminal, w) needs to be 0
            if terminated:
                w += alpha(episode) * (reward - v_hat(state, w, env, state_features)) * state_features(state, env)
            else: 
                w += alpha(episode) * (reward + gamma * v_hat(next_state, w, env, state_features) - v_hat(state, w, env, state_features)) * state_features(state, env)

            if verbose:
                print (f"Episode {episode+1}, Step {i+1}: S={state}, A={action}, R={reward}, S'={next_state}, w={w}")

            state = next_state
            i += 1

    return w

semi_gradient_Sarsa0

semi_gradient_Sarsa0(env, state_features, n, epsilon, alpha, gamma, w=None, max_episode_length=1000, verbose=False, history=False, rng=None)

Run semi-gradient Sarsa(0) with function approximation for control.

Implements the semi-gradient Sarsa(0) algorithm for estimating the optimal action-value function q_*(s, a) using a differentiable function approximator q̂(s, a, w). Actions are selected according to an ε-greedy policy derived from the current action-value estimate.

Episodes are truncated after max_episode_length time steps.

Parameters:

Name	Type	Description	Default
env		Episodic environment used to generate experience.	required
state_features		Callable converting (state, env) to a feature vector.	required
n		Number of training episodes.	required
epsilon		Exploration rate or schedule for the epsilon-greedy policy.	required
alpha		Step size or schedule.	required
gamma		Discount factor in [0, 1].	required
w		Initial action-value weight vector. If omitted, initialize it to zeros.	None
max_episode_length		Maximum number of steps per episode.	1000
verbose		Whether to print step-by-step diagnostics.	False
history		Whether to return weights, returns, and episode lengths collected during training.	False
rng		NumPy generator or integer seed for exploration and tie-breaking.	None

Returns:

Type	Description
	Learned weight vector. If history is true, returns (w, history).

Source code in gym_classics2/algorithms/linear_approximation.py

def semi_gradient_Sarsa0(env, state_features, n, epsilon, alpha, gamma, w=None,
                          max_episode_length=1000, verbose=False,
                          history=False, rng=None):
    """Run semi-gradient Sarsa(0) with function approximation for control.

    Implements the **semi-gradient Sarsa(0)** algorithm for estimating the optimal
    action-value function q_*(s, a) using a differentiable function approximator
    q̂(s, a, w). Actions are selected according to an ε-greedy policy derived
    from the current action-value estimate.

    Episodes are truncated after `max_episode_length` time steps.

    Args:
        env: Episodic environment used to generate experience.
        state_features: Callable converting ``(state, env)`` to a feature vector.
        n: Number of training episodes.
        epsilon: Exploration rate or schedule for the epsilon-greedy policy.
        alpha: Step size or schedule.
        gamma: Discount factor in ``[0, 1]``.
        w: Initial action-value weight vector. If omitted, initialize it to zeros.
        max_episode_length: Maximum number of steps per episode.
        verbose: Whether to print step-by-step diagnostics.
        history: Whether to return weights, returns, and episode lengths collected
            during training.
        rng: NumPy generator or integer seed for exploration and tie-breaking.

    Returns:
        Learned weight vector. If ``history`` is true, returns ``(w, history)``.
    """

    assert gamma >= 0 and gamma <= 1, "Gamma must be in [0,1]"
    assert n > 0, "Number of episodes must be positive"
    assert max_episode_length > 0, "Max episode length must be positive"

    rng = get_rng(rng)

    if isinstance(env.observation_space, gym.spaces.Discrete):
        warnings.warn("The environment has a discrete state space. Consider using a tabular method instead of function approximation.")

    if not isinstance(alpha, Schedule):
        alpha = ConstantSchedule(alpha)
    if not isinstance(epsilon, Schedule):
        epsilon = ConstantSchedule(epsilon)

    if w is None:
        state, _ = env.reset()
        w = np.zeros(len(state_action_features(state, 0, env, state_features)))

    if history:
        ws = []
        ws.append(w.copy())
        returns = []
        ep_lens = []

    for episode in tqdm(range(n), desc="Semi-Gradient SARSA(0)", disable=verbose):
        state, _ = env.reset()
        action = epsilon_greedy_action_w(
            state, w, env, state_features, epsilon(episode), rng=rng
        )
        done = False

        i = 0
        if history:
            G = 0
        while not done and i < max_episode_length:


            next_state, reward, terminated, truncated, _ = env.step(action)
            done = terminated or truncated

            x = state_action_features(state, action, env, state_features)

            if terminated:
                next_action = None
                w += alpha(episode) * (reward - q_hat(state, action, w, env, state_features)) * x

            else:
                next_action = epsilon_greedy_action_w(
                    next_state, w, env, state_features, epsilon(episode), rng=rng
                )
                w += alpha(episode) * (reward + gamma * q_hat(next_state, next_action, w, env, state_features) - q_hat(state, action, w, env, state_features)) * x

            if verbose:
                print (f"Episode {episode+1}, Step {i+1}: S={state}, A={action}, R={reward}, S'={next_state}, w={w}")

            state = next_state
            action = next_action
            i += 1

            if history:
                G += reward * (gamma ** (i-1))

        if history:
            ws.append(w.copy())
            returns.append(G)
            ep_lens.append(i)

    if history:
        return w, {'ws': ws, 'returns': returns, 'ep_lens': ep_lens}

    return w  

create_fourier_basis_coefs

create_fourier_basis_coefs(dim, order)

Create Fourier basis coefficient vectors.

Parameters:

Name	Type	Description	Default
dim		Number of state-feature dimensions.	required
order		Maximum coefficient in each dimension.	required

Returns:

Type	Description
	Array containing the Cartesian product of coefficients from zero through
	order in each dimension.

Source code in gym_classics2/algorithms/linear_approximation.py

def create_fourier_basis_coefs(dim, order): 
    """Create Fourier basis coefficient vectors.

    Args:
        dim: Number of state-feature dimensions.
        order: Maximum coefficient in each dimension.

    Returns:
        Array containing the Cartesian product of coefficients from zero through
        ``order`` in each dimension.
    """
    return np.array(list(product(range(order+1), repeat=dim)))

transformation_fourier_basis

transformation_fourier_basis(min, max, order)

Create a Fourier basis feature transformation.

Parameters:

Name	Type	Description	Default
min		Minimum state value in each dimension.	required
max		Maximum state value in each dimension.	required
order		Maximum Fourier coefficient in each dimension.	required

Returns:

Type	Description
	Callable that normalizes a state to the unit hypercube and returns its
	Fourier basis features.

Example

transform = transformation_fourier_basis([0, 0], [1, 1], order=3)

def state_features(state, env):
    return transform(state)

Source code in gym_classics2/algorithms/linear_approximation.py

def transformation_fourier_basis(min, max, order):
    """Create a Fourier basis feature transformation.

    Args:
        min: Minimum state value in each dimension.
        max: Maximum state value in each dimension.
        order: Maximum Fourier coefficient in each dimension.

    Returns:
        Callable that normalizes a state to the unit hypercube and returns its
        Fourier basis features.

    Example:
        ```python
        transform = transformation_fourier_basis([0, 0], [1, 1], order=3)

        def state_features(state, env):
            return transform(state)
        ```
    """

    min = np.array(min)
    max = np.array(max)
    coefs = create_fourier_basis_coefs(len(min), order)

    def fourier_basis(s):
        # normalize state to [0,1]
        s = np.array(s)
        assert s.shape == min.shape, "State dimension does not match Fourier basis dimension"

        norm_s = (s - min) / (max - min)
        return np.cos(np.pi * np.dot(coefs, norm_s))

    return fourier_basis

Eligibility traces

gym_classics2.algorithms.eligibility_traces

Semi-gradient Sarsa(lambda) with linear function approximation.

Callers provide a state_features(state, env) function that converts states to feature vectors.

semi_gradient_Sarsa_lambda

semi_gradient_Sarsa_lambda(env, state_features, n, epsilon, alpha, gamma, lam, w=None, max_episode_length=1000, verbose=False, history=False, rng=None)

Run semi-gradient Sarsa(lambda) with linear function approximation.

Parameters:

Name	Type	Description	Default
env		Episodic environment used to generate experience.	required
state_features		Callable converting (state, env) to a feature vector.	required
n		Number of episodes.	required
epsilon		Exploration rate or schedule for the epsilon-greedy policy.	required
alpha		Step size or schedule.	required
gamma		Discount factor in [0, 1].	required
lam		Trace-decay parameter lambda in [0, 1].	required
w		Initial weight vector. If omitted, initialize it to zeros.	None
max_episode_length		Maximum number of steps per episode.	1000
verbose		Whether to print step-by-step diagnostics.	False
history		Whether to return weights, returns, and episode lengths collected during training.	False
rng		NumPy generator or integer seed for exploration and tie-breaking.	None

Returns:

Type	Description
	Learned weight vector. If history is true, returns (w, history).

Source code in gym_classics2/algorithms/eligibility_traces.py

def semi_gradient_Sarsa_lambda(
    env,
    state_features,
    n,
    epsilon,
    alpha,
    gamma,
    lam,
    w=None,
    max_episode_length=1000,
    verbose=False,
    history=False,
    rng=None,
):
    """Run semi-gradient Sarsa(lambda) with linear function approximation.

    Args:
        env: Episodic environment used to generate experience.
        state_features: Callable converting ``(state, env)`` to a feature vector.
        n: Number of episodes.
        epsilon: Exploration rate or schedule for the epsilon-greedy policy.
        alpha: Step size or schedule.
        gamma: Discount factor in ``[0, 1]``.
        lam: Trace-decay parameter lambda in ``[0, 1]``.
        w: Initial weight vector. If omitted, initialize it to zeros.
        max_episode_length: Maximum number of steps per episode.
        verbose: Whether to print step-by-step diagnostics.
        history: Whether to return weights, returns, and episode lengths collected
            during training.
        rng: NumPy generator or integer seed for exploration and tie-breaking.

    Returns:
        Learned weight vector. If ``history`` is true, returns ``(w, history)``.
    """

    assert gamma >= 0 and gamma <= 1, "gamma must be in [0,1]"
    assert lam >= 0 and lam <= 1, "lambda must be in [0,1]"
    assert n > 0, "number of episodes must be positive"
    assert max_episode_length > 0, "max episode length must be positive"

    rng = get_rng(rng)

    if not isinstance(alpha, Schedule):
        alpha = ConstantSchedule(alpha)
    if not isinstance(epsilon, Schedule):
        epsilon = ConstantSchedule(epsilon)

    if w is None:
        state, _ = env.reset()
        w = np.zeros(len(state_action_features(state, 0, env, state_features)))

    if history:
        ws = []
        ws.append(w.copy())
        returns = []
        ep_lens = []


    for episode in tqdm(range(n), desc="Semi-Gradient SARSA(lambda)", disable=verbose):
        state, _ = env.reset()
        action = epsilon_greedy_action_w(
            state, w, env, state_features, epsilon(episode), rng=rng
        )

        # eligibility trace vector, same size as w
        z = np.zeros_like(w)
        Q_old = 0

        done = False
        i = 0

        G = 0  # for tracking returns if history is enabled

        while not done and i < max_episode_length:
            next_state, reward, terminated, truncated, _ = env.step(action)
            done = terminated or truncated

            G += reward * (gamma ** i)  # accumulate return if history is enabled

            # current feature vector for (state, action)
            x = state_action_features(state, action, env, state_features)

            # update trace
            z = gamma * lam * z + (1 - alpha(episode) * gamma * lam * np.dot(z, x)) * x

            if terminated:
                delta = reward - q_hat(state, action, w, env, state_features)
            else:
                next_action = epsilon_greedy_action_w(
                    next_state, w, env, state_features, epsilon(episode), rng=rng
                )
                delta = reward + gamma * q_hat(next_state, next_action, w, env, state_features) - q_hat(state, action, w, env, state_features)

            # semi-gradient weight update
            Q = q_hat(state, action, w, env, state_features)
            Q_prime = q_hat(next_state, next_action, w, env, state_features) if not terminated else 0
            w += alpha(episode) * (delta + Q - Q_old) * z - alpha(episode) * (Q - Q_old) * x

            Q_old = Q_prime

            if verbose:
                if terminated:
                    print(
                        f"Episode {episode+1}, Step {i+1}: "
                        f"S={state}, A={action}, R={reward}, S'={next_state}, "
                        f"delta={delta}, z={z}, w={w}"
                    )
                else:
                    print(
                        f"Episode {episode+1}, Step {i+1}: "
                        f"S={state}, A={action}, R={reward}, S'={next_state}, A'={next_action}, "
                        f"delta={delta}, z={z}, w={w}"
                    )

            if done:
                break

            state = next_state
            action = next_action
            i += 1

        if history:
            returns.append(G)
            ws.append(w.copy())
            ep_lens.append(i)


    if history:        
        return w, {'ws': ws, 'returns': returns, 'ep_lens': ep_lens}

    return w

Policy-gradient methods

gym_classics2.algorithms.policy_gradient_methods

Policy-gradient control with linear function approximation.

REINFORCE uses sampled returns; one-step actor-critic (AC) uses a learned state-value function to estimate the policy update. Both learn a stochastic softmax policy over linear action preferences for a discrete action space.

Callers supply state_features(state, env), returning a fixed-length NumPy vector with a leading intercept, such as [1, x, y]. Policy features are constructed by state_action_features; AC also uses the state features directly for its value approximation.

h

h(s, a, theta, env, state_features)

Compute the linear action preference h(s, a) = theta^T x(s, a).

Parameters:

Name	Type	Description	Default
s		State to evaluate.	required
a		Integer action ID in range(env.action_space.n).	required
theta		Policy parameter vector matching state_action_features.	required
env		Environment providing the discrete action space.	required
state_features		Callable returning state features with a leading intercept for (s, env).	required

Returns:

Type	Description
	Scalar preference used as the action's softmax logit.

Source code in gym_classics2/algorithms/policy_gradient_methods.py

def h(s, a, theta, env, state_features):
    """Compute the linear action preference ``h(s, a) = theta^T x(s, a)``.

    Args:
        s: State to evaluate.
        a: Integer action ID in ``range(env.action_space.n)``.
        theta: Policy parameter vector matching ``state_action_features``.
        env: Environment providing the discrete action space.
        state_features: Callable returning state features with a leading
            intercept for ``(s, env)``.

    Returns:
        Scalar preference used as the action's softmax logit.
    """
    return np.dot(theta, state_action_features(s, a, env, state_features))

pi

pi(s, theta, env, state_features)

Compute the softmax policy pi(a | s) over linear preferences.

Parameters:

Name	Type	Description	Default
s		State to evaluate.	required
theta		Policy parameter vector matching state_action_features.	required
env		Environment providing the discrete action space.	required
state_features		Callable returning state features with a leading intercept for (s, env).	required

Returns:

Type	Description
	NumPy vector of length env.action_space.n, indexed by action ID,
	with probabilities proportional to exp(h(s, a)).

Source code in gym_classics2/algorithms/policy_gradient_methods.py

def pi(s, theta, env, state_features):
    """Compute the softmax policy ``pi(a | s)`` over linear preferences.

    Args:
        s: State to evaluate.
        theta: Policy parameter vector matching ``state_action_features``.
        env: Environment providing the discrete action space.
        state_features: Callable returning state features with a leading
            intercept for ``(s, env)``.

    Returns:
        NumPy vector of length ``env.action_space.n``, indexed by action ID,
        with probabilities proportional to ``exp(h(s, a))``.
    """
    hs = np.array(
        [h(s, a, theta, env, state_features) for a in range(env.action_space.n)]
    )
    exp_hs = np.exp(hs)
    return exp_hs / np.sum(exp_hs)

sample_episode_approx_policy

sample_episode_approx_policy(env, pi, theta, state_features, max_episode_length=1000, rng=None)

Sample an episode from a parameterized policy.

Parameters:

Name	Type	Description	Default
env		Episodic Gymnasium environment with a discrete action space.	required
pi		Callable pi(state, theta, env, state_features) returning an action-probability vector indexed by action ID.	required
theta		Policy parameter vector.	required
state_features		Feature callable passed through to pi.	required
max_episode_length		Maximum number of transitions to sample.	1000
rng		NumPy generator, integer seed, or None for action sampling. Does not seed the environment.	None

Returns:

Type	Description
	List of (state, action, reward, next_state) transitions, beginning
	with a fresh environment reset. Stops on termination or after
	max_episode_length transitions; truncation is currently ignored.

Source code in gym_classics2/algorithms/policy_gradient_methods.py

def sample_episode_approx_policy(
    env, pi, theta, state_features, max_episode_length=1000, rng=None
):
    """Sample an episode from a parameterized policy.

    Args:
        env: Episodic Gymnasium environment with a discrete action space.
        pi: Callable ``pi(state, theta, env, state_features)`` returning an
            action-probability vector indexed by action ID.
        theta: Policy parameter vector.
        state_features: Feature callable passed through to ``pi``.
        max_episode_length: Maximum number of transitions to sample.
        rng: NumPy generator, integer seed, or None for action sampling.
            Does not seed the environment.

    Returns:
        List of ``(state, action, reward, next_state)`` transitions, beginning
        with a fresh environment reset. Stops on termination or after
        ``max_episode_length`` transitions; truncation is currently ignored.
    """

    rng = get_rng(rng)
    s, _ = env.reset()
    episode_data = []

    for t in range(max_episode_length):
        a = rng.choice(
            env.action_space.n, p=pi(s, theta, env, state_features)
        )
        next_s, r, done, _, _ = env.step(a)
        episode_data.append((s, a, r, next_s))
        s = next_s

        if done:
            break

    return episode_data

choose_action_w

choose_action_w(env, pi, theta, state, state_features, rng=None)

Sample an action from a parameterized policy.

Parameters:

Name	Type	Description	Default
env		Environment providing the discrete action space.	required
pi		Callable pi(state, theta, env, state_features) returning an action-probability vector indexed by action ID.	required
theta		Policy parameter vector.	required
state		Current state.	required
state_features		Feature callable passed through to pi.	required
rng		NumPy generator, integer seed, or None for action sampling.	None

Returns:

Type	Description
	Selected integer action ID.

Source code in gym_classics2/algorithms/policy_gradient_methods.py

def choose_action_w(env, pi, theta, state, state_features, rng=None):
    """Sample an action from a parameterized policy.

    Args:
        env: Environment providing the discrete action space.
        pi: Callable ``pi(state, theta, env, state_features)`` returning an
            action-probability vector indexed by action ID.
        theta: Policy parameter vector.
        state: Current state.
        state_features: Feature callable passed through to ``pi``.
        rng: NumPy generator, integer seed, or None for action sampling.

    Returns:
        Selected integer action ID.
    """
    rng = get_rng(rng)
    return rng.choice(
        env.action_space.n, p=pi(state, theta, env, state_features)
    )

REINFORCE

REINFORCE(env, state_features, n, alpha, gamma, theta=None, max_episode_length=1000, verbose=False, history=False, rng=None)

Learn a linear softmax policy with episodic REINFORCE (no baseline).

Sample an episode, then update theta at each visited state using alpha(episode) * gamma**t * G_t * grad_log_pi, where G_t is the discounted return from step t to the end of the sampled trajectory.

Parameters:

Name	Type	Description	Default
env		Episodic Gymnasium environment with a discrete action space.	required
state_features		Callable returning a fixed-length NumPy feature vector with a leading intercept for (state, env).	required
n		Positive number of training episodes.	required
alpha		Float step size or callable of the zero-based episode index.	required
gamma		Discount factor in [0, 1].	required
theta		Initial NumPy policy parameter vector, updated in place. Its length must match state_action_features. Defaults to zeros, giving a uniform initial policy.	None
max_episode_length		Positive maximum number of steps per episode.	1000
verbose		Print step-by-step diagnostics instead of a progress bar.	False
history		Whether to return training history alongside the parameters.	False
rng		NumPy generator, integer seed, or None for action sampling. Does not seed the environment.	None

Returns:

Type	Description
	Learned parameter vector theta, or (theta, history) when
	history=True. The history dictionary contains thetas (copies
	before training and after each episode), returns (discounted
	episode returns), and ep_lens (episode lengths in steps).

Notes

Sampling ignores environment truncation. If the step limit cuts an episode short, returns use only the sampled rewards, without bootstrapping a value for the final state.

Source code in gym_classics2/algorithms/policy_gradient_methods.py

def REINFORCE(
    env,
    state_features,
    n,
    alpha,
    gamma,
    theta = None,
    max_episode_length=1000,
    verbose=False,
    history=False,
    rng=None,
    ):
    """Learn a linear softmax policy with episodic REINFORCE (no baseline).

    Sample an episode, then update ``theta`` at each visited state using
    ``alpha(episode) * gamma**t * G_t * grad_log_pi``, where ``G_t`` is the
    discounted return from step ``t`` to the end of the sampled trajectory.

    Args:
        env: Episodic Gymnasium environment with a discrete action space.
        state_features: Callable returning a fixed-length NumPy feature vector
            with a leading intercept for ``(state, env)``.
        n: Positive number of training episodes.
        alpha: Float step size or callable of the zero-based episode index.
        gamma: Discount factor in ``[0, 1]``.
        theta: Initial NumPy policy parameter vector, updated in place. Its
            length must match ``state_action_features``. Defaults to zeros,
            giving a uniform initial policy.
        max_episode_length: Positive maximum number of steps per episode.
        verbose: Print step-by-step diagnostics instead of a progress bar.
        history: Whether to return training history alongside the parameters.
        rng: NumPy generator, integer seed, or None for action sampling.
            Does not seed the environment.

    Returns:
        Learned parameter vector ``theta``, or ``(theta, history)`` when
        ``history=True``. The history dictionary contains ``thetas`` (copies
        before training and after each episode), ``returns`` (discounted
        episode returns), and ``ep_lens`` (episode lengths in steps).

    Notes:
        Sampling ignores environment truncation. If the step limit cuts an
        episode short, returns use only the sampled rewards, without
        bootstrapping a value for the final state.
    """

    assert gamma >= 0 and gamma <= 1, "gamma must be in [0  ,1]"
    assert n > 0, "number of episodes must be positive"
    assert max_episode_length > 0, "max episode length must be positive"

    rng = get_rng(rng)

    if isinstance(env.observation_space, gym.spaces.Discrete):
        warnings.warn("The environment has a discrete state space. Consider using a tabular method instead of function approximation.")

    if isinstance(alpha, float):
        alpha = ConstantSchedule(alpha)

    if theta is None:
        state, _ = env.reset()
        theta = np.zeros(
            len(state_action_features(state, 0, env, state_features))
        )

    if history:
        returns = []        
        ep_lens = []
        thetas = []
        thetas.append(theta.copy())

    for episode in tqdm(range(n), desc="Episodes", disable=verbose):
        if verbose:
            print(f"Episode {episode+1}/{n}")

        # sample complete episode using pi (this is a MC method)
        episode_data = sample_episode_approx_policy(
            env, pi, theta, state_features, max_episode_length, rng=rng
        )

        for t in range(len(episode_data)):
            # update policy for each step in the episode using the return observed from that step on
            #print(episode_data[t])

            G = np.sum([e[2] for e in episode_data[t:]] * (gamma ** np.arange(len(episode_data[t:]))))
            if history and t == 0:
                returns.append(G)   

            s,a,r,next_s = episode_data[t]

            # ln policy gradient= x(s,a)- sum_b pi(b|s,theta) x(s,b)
            grad_log_pi = state_action_features(s, a, env, state_features) - sum(
                pi(s, theta, env, state_features)[b]
                * state_action_features(s, b, env, state_features)
                for b in range(env.action_space.n)
            )

            if verbose: 
                print (f"t: {t}, G: {G:.2f}, grad_log_pi: {grad_log_pi}")

            theta += alpha(episode) * (gamma**t) * G * grad_log_pi

        if history:
            thetas.append(theta.copy())
            ep_lens.append(len(episode_data))

    if history:
        return theta, {'thetas': thetas, 'returns': returns, 'ep_lens': ep_lens}

    return theta

AC

AC(env, state_features, n, alpha_policy, alpha_value, gamma, max_episode_length=1000, verbose=False, history=False, rng=None)

Learn a linear softmax policy with one-step actor-critic.

The critic approximates v(s) = w^T x(s) and learns from the TD error delta = r + gamma * v(next_state) - v(state). The actor updates its policy using alpha_policy(episode) * gamma**t * delta * grad_log_pi. Both parameter vectors start at zero, giving a uniform initial policy.

Parameters:

Name	Type	Description	Default
env		Episodic Gymnasium environment with a discrete action space.	required
state_features		Callable returning a fixed-length NumPy feature vector with a leading intercept for (state, env). Used by the critic directly and by the actor through state_action_features.	required
n		Positive number of training episodes.	required
alpha_policy		Float policy step size or callable of the zero-based episode index.	required
alpha_value		Float value-function step size or callable of the zero-based episode index.	required
gamma		Discount factor in [0, 1].	required
max_episode_length		Positive maximum number of steps per episode.	1000
verbose		Print episode numbers instead of a progress bar.	False
history		Whether to return training history alongside the parameters.	False
rng		NumPy generator, integer seed, or None for action sampling. Does not seed the environment.	None

Returns:

Type	Description
	Tuple (theta, w) containing the learned policy and value parameter
	vectors, or (theta, w, history) when history=True. The history
	dictionary contains thetas (copies before training and after each
	episode), returns (discounted episode returns), and ep_lens
	(episode lengths in steps).

Notes

This implementation ignores environment truncation and bootstraps from the next state's value even on a terminal transition. For an episodic task, terminal-state features should therefore yield a value of zero.

Source code in gym_classics2/algorithms/policy_gradient_methods.py

def AC(
    env,
    state_features,
    n,
    alpha_policy,
    alpha_value,
    gamma,
    max_episode_length=1000,
    verbose=False,
    history=False,
    rng=None,
    ):
    """Learn a linear softmax policy with one-step actor-critic.

    The critic approximates ``v(s) = w^T x(s)`` and learns from the TD error
    ``delta = r + gamma * v(next_state) - v(state)``. The actor updates its
    policy using ``alpha_policy(episode) * gamma**t * delta * grad_log_pi``.
    Both parameter vectors start at zero, giving a uniform initial policy.

    Args:
        env: Episodic Gymnasium environment with a discrete action space.
        state_features: Callable returning a fixed-length NumPy feature vector
            with a leading intercept for ``(state, env)``. Used by the critic
            directly and by the actor through ``state_action_features``.
        n: Positive number of training episodes.
        alpha_policy: Float policy step size or callable of the zero-based
            episode index.
        alpha_value: Float value-function step size or callable of the
            zero-based episode index.
        gamma: Discount factor in ``[0, 1]``.
        max_episode_length: Positive maximum number of steps per episode.
        verbose: Print episode numbers instead of a progress bar.
        history: Whether to return training history alongside the parameters.
        rng: NumPy generator, integer seed, or None for action sampling.
            Does not seed the environment.

    Returns:
        Tuple ``(theta, w)`` containing the learned policy and value parameter
        vectors, or ``(theta, w, history)`` when ``history=True``. The history
        dictionary contains ``thetas`` (copies before training and after each
        episode), ``returns`` (discounted episode returns), and ``ep_lens``
        (episode lengths in steps).

    Notes:
        This implementation ignores environment truncation and bootstraps
        from the next state's value even on a terminal transition. For an
        episodic task, terminal-state features should therefore yield a
        value of zero.
    """

    assert gamma >= 0 and gamma <= 1, "gamma must be in [0  ,1]"
    assert n > 0, "number of episodes must be positive"
    assert max_episode_length > 0, "max episode length must be positive"

    rng = get_rng(rng)

    if isinstance(env.observation_space, gym.spaces.Discrete):
        warnings.warn("The environment has a discrete state space. Consider using a tabular method instead of function approximation.")

    if isinstance(alpha_policy, float):
        alpha_policy = ConstantSchedule(alpha_policy)
    if isinstance(alpha_value, float):
        alpha_value = ConstantSchedule(alpha_value) 

    state, _ = env.reset()

    # for simplicity we use the same features for the value function and the policy approximation
    # value function weights
    w = np.zeros(len(state_features(state, env)))
    # policy weights
    theta = np.zeros(len(state_action_features(state, 0, env, state_features)))

    if history:
        returns = []        
        ep_lens = []
        ws = []
        ws.append(w.copy())
        thetas = []
        thetas.append(theta.copy())

    for episode in tqdm(range(n), desc="Episodes", disable=verbose):
        if verbose:
            print(f"Episode {episode+1}/{n}")

        disc_factor = 1.0
        state, _ = env.reset()
        done = False
        i = 0
        G = 0.0

        while not done and  i < max_episode_length:
            # use actor to determine next action
            a = rng.choice(
                env.action_space.n,
                p=pi(state, theta, env, state_features),
            )

            # execute action
            next_state, r, done, _, _ = env.step(a)

            # use critic to calculate  TD error
            td_error = r + gamma * np.dot(w, state_features(next_state, env)) - np.dot(w, state_features(state, env))

            # update critic
            w += alpha_value(episode) * td_error * state_features(state, env)

            # update actor
            grad_log_pi = state_action_features(
                state, a, env, state_features
            ) - sum(
                pi(state, theta, env, state_features)[b]
                * state_action_features(state, b, env, state_features)
                for b in range(env.action_space.n)
            )
            theta += alpha_policy(episode) * disc_factor * td_error * grad_log_pi

            G += disc_factor * r
            disc_factor *= gamma      
            state = next_state
            i += 1

        if history:
            thetas.append(theta.copy())
            ws.append(w.copy())
            returns.append(G)
            ep_lens.append(i)

    if history:
        return theta, w, {'thetas': thetas, 'returns': returns, 'ep_lens': ep_lens}

    return theta, w

Schedules

gym_classics2.algorithms.schedules

Scalar schedules for reinforcement-learning hyperparameters.

A schedule is a callable that maps a nonnegative episode or step index t to a floating-point value. The included algorithms use schedules to vary parameters such as the step size (alpha) and exploration rate (epsilon) during training.

For example, a linear schedule can decrease epsilon from 1.0 to 0.1 over 10 steps:

Example

```python schedule = LinearDecaySchedule(1.0, min_value=0.1, decay_steps=10) for t in (0, 5, 10, 15): print(t, schedule(t))

```

Schedule

Interface for a scalar value indexed by episode or time step.

Source code in gym_classics2/algorithms/schedules.py

class Schedule:
    """Interface for a scalar value indexed by episode or time step."""

    def __call__(self, t):
        """Evaluate the schedule.

        Args:
            t: Nonnegative episode or step index.

        Returns:
            Scheduled scalar value at ``t``.
        """
        raise NotImplementedError

    def __repr__(self):
        return f"{type(self).__name__}(" + ", ".join([f"{k}={v}" for k, v in self.__dict__.items()]) + ")"

__call__

__call__(t)

Evaluate the schedule.

Parameters:

Name	Type	Description	Default
t		Nonnegative episode or step index.	required

Returns:

Type	Description
	Scheduled scalar value at t.

Source code in gym_classics2/algorithms/schedules.py

def __call__(self, t):
    """Evaluate the schedule.

    Args:
        t: Nonnegative episode or step index.

    Returns:
        Scheduled scalar value at ``t``.
    """
    raise NotImplementedError

ConstantSchedule

Bases: Schedule

Return the same value for every index.

Parameters:

Name	Type	Description	Default
value		Constant value returned by the schedule.	required

Source code in gym_classics2/algorithms/schedules.py

class ConstantSchedule(Schedule):
    """Return the same value for every index.

    Args:
        value: Constant value returned by the schedule.
    """
    def __init__(self, value):
        self.value = float(value)

    def __call__(self, t):
        return self.value

StepSchedule

Bases: Schedule

Switch once from a high value to a low value.

Typical use: Only when you have observed in experiments when training performance plateaus.

The schedule returns high_value while t < steps and low_value from t == steps onward.

Parameters:

Name	Type	Description	Default
high_value		Value before the switch.	required
low_value		Value at and after the switch.	required
steps		Index at which to switch to low_value.	required

Source code in gym_classics2/algorithms/schedules.py

class StepSchedule(Schedule):
    """Switch once from a high value to a low value.

    Typical use: Only when you have observed in experiments when training performance plateaus. 

    The schedule returns ``high_value`` while ``t < steps`` and ``low_value``
    from ``t == steps`` onward.

    Args:
        high_value: Value before the switch.
        low_value: Value at and after the switch.
        steps: Index at which to switch to ``low_value``.
    """
    def __init__(self, high_value, low_value, steps):
        self.high_value = float(high_value)
        self.low_value = float(low_value)
        self.steps = int(steps)

    def __call__(self, t):
        if t < self.steps:
            return self.high_value
        return self.low_value

LinearDecaySchedule

Bases: Schedule

Interpolate linearly from an initial value to a final value.

Typical use: Transformers, NLP fine-tuning

The schedule returns initial_value at t = 0, changes linearly through decay_steps, and returns min_value thereafter.

Parameters:

Name	Type	Description	Default
initial_value		Value at index zero.	required
min_value		Final value and lower endpoint of a decreasing schedule.	required
decay_steps		Number of indices over which to interpolate.	required

Source code in gym_classics2/algorithms/schedules.py

class LinearDecaySchedule(Schedule):
    """Interpolate linearly from an initial value to a final value.

    Typical use: Transformers, NLP fine-tuning

    The schedule returns ``initial_value`` at ``t = 0``, changes linearly through
    ``decay_steps``, and returns ``min_value`` thereafter.

    Args:
        initial_value: Value at index zero.
        min_value: Final value and lower endpoint of a decreasing schedule.
        decay_steps: Number of indices over which to interpolate.
    """
    def __init__(self, initial_value, min_value, decay_steps):
        self.initial_value = float(initial_value)
        self.min_value = float(min_value)
        self.decay_steps = int(decay_steps)

    def __call__(self, t):
        fraction = min(float(t) / max(1, self.decay_steps), 1.0)
        return self.initial_value + fraction * (self.min_value - self.initial_value)

ExponentialDecaySchedule

Bases: Schedule

Decay geometrically to a minimum value.

Deterministic or stationary environments. Rapidly scaling down works

well with algorithms that start with no information (e.g., all Q-values are randomly initialized for tabular methods).

Do not use this for complex, multi-stage RL tasks.

At index t, the value is max(min_value, initial_value * decay_rate**t).

Parameters:

Name	Type	Description	Default
initial_value		Unclipped value at index zero.	required
min_value		Lower bound for the returned value.	required
decay_rate		Multiplicative factor applied at each successive index.	required

Source code in gym_classics2/algorithms/schedules.py

class ExponentialDecaySchedule(Schedule):
    """Decay geometrically to a minimum value.

    Typical use: Deterministic or stationary environments. Rapidly scaling down works
        well with algorithms that start with no information (e.g., all Q-values are randomly initialized
        for tabular methods).

    Do not use this for complex, multi-stage RL tasks.

    At index ``t``, the value is ``max(min_value, initial_value * decay_rate**t)``.

    Args:
        initial_value: Unclipped value at index zero.
        min_value: Lower bound for the returned value.
        decay_rate: Multiplicative factor applied at each successive index.
    """
    def __init__(self, initial_value, min_value, decay_rate):
        self.initial_value = float(initial_value)
        self.min_value = float(min_value)
        self.decay_rate = float(decay_rate)

    def __call__(self, t):
        return max(self.min_value, self.initial_value * (self.decay_rate ** t))

InverseDecaySchedule

Bases: Schedule

Decay in inverse proportion to the index.

The schedule returns initial_value at t = 0. For t > 0, it returns max(min_value, initial_value / t).

Parameters:

Name	Type	Description	Default
initial_value		Value at index zero and numerator of the inverse schedule.	required
min_value		Lower bound for the returned value.	0.0

Source code in gym_classics2/algorithms/schedules.py

class InverseDecaySchedule(Schedule):
    """Decay in inverse proportion to the index.

    The schedule returns ``initial_value`` at ``t = 0``. For ``t > 0``, it
    returns ``max(min_value, initial_value / t)``.

    Args:
        initial_value: Value at index zero and numerator of the inverse schedule.
        min_value: Lower bound for the returned value.
    """
    def __init__(self, initial_value, min_value=0.0):
        self.initial_value = float(initial_value)
        self.min_value = float(min_value)

    def __call__(self, t):
        if t == 0:
            return self.initial_value
        return max(self.min_value, self.initial_value / t)

CosineAnnealingSchedule

Bases: Schedule

Anneal from an initial value to a final value along a half-cosine curve.

It starts with lots of exploration and then switches to to fine grained-exploration.

Typical use: PPO and deep RL

The schedule returns initial_value at t = 0, reaches min_value at decay_steps, and stays there without restarting.

Parameters:

Name	Type	Description	Default
initial_value		Value at index zero.	required
min_value		Final value and lower endpoint of a decreasing schedule.	required
decay_steps		Number of indices over which to anneal.	required

Source code in gym_classics2/algorithms/schedules.py

class CosineAnnealingSchedule(Schedule):
    """Anneal from an initial value to a final value along a half-cosine curve.

    It starts with lots of exploration and then switches to to fine grained-exploration.

    Typical use: PPO and deep RL

    The schedule returns ``initial_value`` at ``t = 0``, reaches ``min_value``
    at ``decay_steps``, and stays there without restarting.

    Args:
        initial_value: Value at index zero.
        min_value: Final value and lower endpoint of a decreasing schedule.
        decay_steps: Number of indices over which to anneal.
    """
    def __init__(self, initial_value, min_value, decay_steps, warm_restarts = False):
        self.initial_value = float(initial_value)
        self.min_value = float(min_value)
        self.decay_steps = int(decay_steps)
        self.warm_restarts = bool(warm_restarts)

    def __call__(self, t):
        if (not self.warm_restarts):
            fraction = min(float(t) / max(1, self.decay_steps), 1.0)
        else:
            fraction = min(float(t % self.decay_steps) / max(1, self.decay_steps), 1.0)
        return self.min_value + 0.5 * (self.initial_value - self.min_value) * (
            1.0 + np.cos(np.pi * fraction)
        )

plot_schedule

plot_schedule(schedule, steps=1000, title=None)

Plot scheduled values for indices 0 through steps - 1.

Example

from gym_classics2.algorithms.schedules import (
    LinearDecaySchedule,
    plot_schedule,
)

epsilon = LinearDecaySchedule(1.0, min_value=0.1, decay_steps=1_000)
plot_schedule(epsilon, steps=1_000)

Parameters:

Name	Type	Description	Default
schedule		Callable that accepts an integer index and returns a scalar.	required
steps		Number of scheduled values to plot.	1000

Source code in gym_classics2/algorithms/schedules.py

def plot_schedule(schedule, steps=1000, title = None):
    """Plot scheduled values for indices ``0`` through ``steps - 1``.

    Example:
        ```python
        from gym_classics2.algorithms.schedules import (
            LinearDecaySchedule,
            plot_schedule,
        )

        epsilon = LinearDecaySchedule(1.0, min_value=0.1, decay_steps=1_000)
        plot_schedule(epsilon, steps=1_000)
        ```

    Args:
        schedule: Callable that accepts an integer index and returns a scalar.
        steps: Number of scheduled values to plot.
    """
    import matplotlib.pyplot as plt

    values = [schedule(t) for t in range(steps)]

    plt.figure(figsize=(8, 5))
    plt.plot(range(steps), values, linewidth=2)
    plt.xlabel('Step')
    plt.ylabel('Schedule Value')
    plt.title(title if title is not None else repr(schedule))
    plt.grid(True)
    plt.show()