<jupyter_start><jupyter_text>Multi-arm bandits - banners with bayes agents (native)<jupyter_code>from scipy.stats import beta
from typing import Dict, List, Tuple

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import seaborn as sns

# Apply the default theme
sns.set_theme()<jupyter_output><empty_output><jupyter_text>Environment<jupyter_code>class BannerEnvironment(object):
  def __init__(self, ctrs: List[int]):
    self.ctrs = ctrs
    self._observe()

  def reset(self):
    return self._observe()

  def _observe(self) -> List[float]:
    self._observation = np.random.rand(1)
    return self._observation

  def step(self, action: int) -> Tuple[int, float]:
    self._observe()
    #print(f' - observation={self._observation[0]}, ctr={self.ctrs[action]}')
    return (
        int(self.ctrs[action] > self._observation[0]), 
        self._observation[0])
    
    
env = BannerEnvironment([0.25, 0.4, 0.67])

observation = env.reset()
print(f'observation: {observation}')

action =  2#@param
print(f'action: {action}')

ret = env.step(action)
print(f'reward: {ret[0]}')<jupyter_output>observation: [0.8981948]
action: 2
reward: 1<jupyter_text>Policy<jupyter_code>class BetaPolicy(object):
  def __init__(self, alpha: List[int], beta: List[int]):
    self._alpha = alpha
    self._beta = beta
  
  def action(self) -> int:
    sample = np.random.beta(self._alpha, self._beta)
    #print(' - sample', sample, self._alpha, self._beta)
    return np.argmax(sample)
pol = BetaPolicy([1, 10, 5], [1, 1, 1])
action = pol.action()
print(f'action: {action}')<jupyter_output>action: 1<jupyter_text>Agent<jupyter_code>class BannerAgent(object):
  def __init__(self):
    self._alpha = [1, 1, 1]
    self._beta = [1, 1, 1]

    self.policy = BetaPolicy(self._alpha, self._beta)

  def train(self, experience: Dict[str, int]):
    action = experience['action']
    reward = experience['reward']

    if reward == 1:
      self._alpha[action] += 1
    else:
      self._beta[action] += 1<jupyter_output><empty_output><jupyter_text>Utils<jupyter_code>def plot_pdf(a: List[int], b: List[int]):
  plt.figure(figsize=(10, 7))
  x = np.linspace(beta.ppf(0.01, a, b),
                  beta.ppf(0.99, a, b), 100)

  lines = plt.plot(x, beta.pdf(x, a, b), '-', lw=1, alpha=0.6, label='beta pdf')

  #lines = ax.plot(x,y.T[:,:])
  plt.legend(lines, [f'action={j}' for j in range(len(lines))])  

  plt.xlabel('x')
  plt.ylabel('PDF')
  plt.show()
  

def plot_cumsum(df: pd.DataFrame, show_actions=False):
  plt.figure(figsize=(10, 7))

  if show_actions:
    df_actions = df.groupby(['trial', 'action']).mean().reset_index()

    for action in sorted(df.action.unique()):
      plt.plot(
          df_actions[df_actions['action']==action].trial, 
          df_actions[df_actions['action']==action].action_cum_sum,
          label=f'action={int(action)} cumulative reward'
      )

  df_all = df.groupby('trial')['cum_sum'].mean().reset_index()
  plt.plot(df_all.trial, df_all.cum_sum, label='agent cumulative reward')

  plt.xlim(0)
  plt.xlabel('Number of trials')
  plt.ylabel('Cumulative Reward')
  plt.legend()
  #plt.title(f'Cumulative Reward of the {algorithm_name} Algorithm')
  plt.show()
np.random.random_sample(3)<jupyter_output><empty_output><jupyter_text>Experiment<jupyter_code>def run(environment: object, agent: object, trials=100):
  columns = ['trial', 'action', 'observation', 'reward']
  df_trajectory = pd.DataFrame(columns=columns, dtype=float)

  experience = {'trial': 0, 'action': -1, 'observation': 0, 'reward': 0}

  for i in range(trials):
    experience['trial'] = i + 1
    experience['action'] = agent.policy.action()
    experience['reward'], experience['observation'] = environment.step(experience['action'])

    agent.train(experience)

    df_trajectory = df_trajectory.append(experience, ignore_index=True)

  df_cumsum = df_trajectory.groupby('action')['reward'].cumsum()
  df_trajectory = df_trajectory.assign(cum_sum=df_trajectory['reward'].cumsum())
  df_trajectory = df_trajectory.assign(action_cum_sum=df_cumsum)

  return df_trajectory
environment = BannerEnvironment([0.25, 0.4, 0.67])
step = environment.reset()

agent = BannerAgent()

df_trajectory = run(environment, agent, trials=1000)
plot_pdf(np.array(params['alpha']).mean(axis=0), np.array(params['beta']).mean(axis=0))
plot_cumsum(df_trajectory, show_actions=True)<jupyter_output><empty_output><jupyter_text>Epochs<jupyter_code>def experiment(epochs: int = 1, trials: int = 10):
  columns = ['trial', 'action', 'observation', 'reward']
  df_trajectory = pd.DataFrame(columns=columns, dtype=float)

  params = {
    'alpha': [],
    'beta': []  
  }

  for epoch in range(epochs):
    environment = BannerEnvironment([0.25, 0.4, 0.67])
    step = environment.reset()
    agent = BannerAgent()

    df = run(environment, agent, trials=trials)

    df_trajectory = df_trajectory.append(df, ignore_index=True)

    params['alpha'].append(agent._alpha)
    params['beta'].append(agent._beta)

  return df_trajectory, params
df_trajectory, params = experiment(epochs=100, trials=1000)
plot_pdf(np.array(params['alpha']).mean(axis=0), np.array(params['beta']).mean(axis=0))
plot_cumsum(df_trajectory, show_actions=True)<jupyter_output><empty_output><jupyter_text>Change of the environment<jupyter_code>environment = BannerEnvironment([0.25, 0.4, 0.67])
step = environment.reset()

agent = BannerAgent()

df_trajectory = run(environment, agent, trials=100)

print(f'\nalpha={agent._alpha}, beta={agent._beta}')
print(f'mu={beta.stats(agent._alpha, agent._beta)[0]}')
plot_pdf(agent._alpha, agent._beta)
environment = BannerEnvironment([0.85, 0.4, 0.15])
step = environment.reset()

#agent = BannerAgent()

df_trajectory = run(environment, agent, trials=100)

print(f'\nalpha={agent._alpha}, beta={agent._beta}')
print(f'mu={beta.stats(agent._alpha, agent._beta)[0]}')
plot_pdf(agent._alpha, agent._beta)<jupyter_output><empty_output><jupyter_text>**Úkol**: Vyřešit, aby při změně environmentu agent rychleji reagoval.<jupyter_code><jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.163013021262568, 'fraction_numerical': 0.020932915773858578, 'mean_word_length': 5.166666666666667, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 39, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '91095', 'license': 'Apache-2.0', 'n_tokens_mistral': 2334, 'n_tokens_neox': 2202, 'n_words': 392, 'path': '01-banners_bayes.ipynb', 'repo_name': 'pstanisl/UnifiedPreference', 'size': '6067'}