Thompson sampling

Nomenclature

Consider:

Rationale

Definition

For each iteration tt: - A “player” obtains a context x∈Xx\in \mathcal{X} - Plays an action a∈Aa\in \mathcal{A} - Receives a reward r∈Rr\in \mathcal{R} - This rewards is distributed according to the context and the resulting action - The player’s goal is to execute actions that maximize the cumulative rewards.

Implementation

The implementation will focus on these concepts:

Thompson sampling consists in playing the action a∗∈Aa^{\ast }\in {\mathcal {A}} according to the probability that it maximizes the expected reward, i.e. action a∗a^{\ast } is chosen with probability

 0 

where I\mathbb {I} is the indicator function.

In practice, the rule is implemented by sampling. In each round, parameters θ∗\theta^\ast are sampled from the posterior P(θ∣D)P(\theta |{\mathcal {D}}), and an action a∗a^{\ast } chosen that maximizes E[r∣θ∗,a∗,x]{\mathbb {E}}[r|\theta ^{\ast },a^{\ast },x], i.e. the expected reward given the sampled parameters, the action, and the current context. Conceptually, this means that the player instantiates their beliefs randomly in each round according to the posterior distribution, and then acts optimally according to them. In most practical applications, it is computationally onerous to maintain and sample from a posterior distribution over models. As such, Thompson sampling is often used in conjunction with approximate sampling techniques.

Example

N_TRIALS = 2000
N_ARMS = 16
N_FEATURES = 5
BEST_ARMS = [3, 7, 9, 15]

We now define a function to generate context vectors for all arms for each of the trial. We need:

This function will return a matrix of size NT×NA×NfN_{T} \times N_{A} \times N_{f}

def make_design_matrix(n_trials: int, 
                       n_arms: int, 
                       n_features: int) -> np.ndarray:
    available_arms = np.arange(n_arms)
    X = np.array([[np.random.uniform(0, 1, size = n_features) 
                   for _ in np.arange(n_arms)] 
                   for _ in np.arange(n_trials)])
    return X
X = make_design_matrix(n_trials=N_TRIALS, 
                       n_arms=N_ARMS, 
                       n_features=N_FEATURES)

This will have the shape

![[Thompson sampling trials.excalidraw.svg]]

The following function will generate the true Θ=θ1,…,θn\Theta = {\theta_1,\dots,\theta_n} for testing purposes. We provide:

A matrix of size NA×NfN_{A} \times N_{f}, each value is a random value with μ=0\mu = 0 and standard deviation of 14\frac{1}{4}. However, for the best arms, we will add the bias.

![[Thompson sampling thetas.excalidraw.svg]]

def make_theta(n_arms: int, 
               n_features: int, best_arms, bias = 1):
    true_theta = np.array(
        [np.random.normal(size=n_features, scale=1.0/4.0) 
        for _ in range(n_arms)])
    true_theta[best_arms] += bias
    return true_theta
true_theta = make_theta(
    n_arms=N_ARMS, 
    n_features=N_FEATURES, 
    best_arms=BEST_ARMS)

A function is also available to generate rewards. It creates rewards for each arm, given a context.

We provide:

This will return the estimated score for the arm (with the arm index and the context observed corresponding to the given theta).

def generate_reward(arm, x, theta, scale_noise = 1.0/10.0):
    signal = theta[arm].dot(x)
    noise = np.random.normal(scale=scale_noise)
    return signal + noise
random_payoffs = np.array(
    [generate_reward(
        arm=np.random.choice(N_ARMS), 
        x=X[t, np.random.choice(N_ARMS)], 
        theta=true_theta) 
        for t in range(N_TRIALS)])

# Defining oracle (best payoffs based on the true_theta)
oracles = np.array(
    [np.max(
        [generate_reward(
            arm=arm,
            x=X[t, arm],
            theta=true_theta) 
        for arm in range(N_ARMS)]) 
    for t in range(N_TRIALS)])

We also create a function to generate the cumulative regret over time.

We provide:

And we get an array of the cumulative sum over time (of size TT).

def make_regret(payoffs: np.ndarray, 
                oracles: np.ndarray) -> np.ndarray:
    return np.cumsum(oracles - payoffs)
payoffs = [
    [generate_reward(
        arm=arm, 
        x=X[t, arm], 
        theta=true_theta) 
        for arm in np.arange(N_ARMS)] 
    for t in np.arange(N_TRIALS)]

ave_rewards = np.mean(payoffs, axis=0)

The actual sampling

The method to perform the actual sampling is next. We provide:

delta=0.5
R = 0.01
epsilon=0.5

We use r_payoffs to store the payoff for each trial (the payoff for the selected arm based on the true_theta). As such, we initialise a zero array of size n_trials.

r_payoffs = np.zeros(N_TRIALS)
v = R * np.sqrt(24 / epsilon * N_FEATURES * np.log(1 / delta))

Model initialisation:

B = np.identity(N_FEATURES) 
mu_hat = np.zeros(shape=(N_FEATURES, 1))
f = np.zeros(shape=(N_FEATURES,1))  
for t in range(N_TRIALS):
        context = X[t]
        mu_tilde = np.random.multivariate_normal(mu_hat.flat, v**2 * np.linalg.inv(B))[..., np.newaxis]
        score_array = context.dot(mu_tilde)
        chosen_arm = np.argmax(score_array)
        context_t = context[chosen_arm]
        reward = generate_reward(arm=chosen_arm, x=context_t, theta=true_theta)
        r_payoffs[t] = reward
        context_t = np.reshape(context_t, (-1, 1))
        B += context_t.dot(context_t.T)
        f += reward*context_t
        mu_hat = np.linalg.inv(B).dot(f)