# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

# # Warsztaty Python w Data Science
#
# ---
#
# ## Machine Learning - część 5 z 5. Reinforcement Learning
#
# - ### Gra w kółko i krzyżyk
# - ### Q-Learning i wzór Bellmana
# - ### Testy A/B
# - ### Problem wielorękiego bandyty
# - ### Wyżarzanie
#
# ---

#
# ## Gra w kółko i krzyżyk

def cell(x):
    try:
        if x.lower()=="x":
            return " X "
        if x.lower()=="o":
            return " O "
    except:
        pass
    return "   "


def print_state(state):
    sep = "_"*11+"\n"
    ret = []
    n=0
    for row in state:
        ret.append("|".join(map(cell,row)))
        n+=1
        if n<3:
                ret.append(sep)
    for line in ret:
        print(line)


state = (('x','o',0.0),
         ('o','x',0.0),
         (0.0,'o','x'))
print_state(state)


def has_won(state):
    players = ['x', 'o']
    for i in [0,1]:
        for row in state:
            if row==tuple(players[i]*3): # _
                return i, True
        for cols in [0, 1, 2]:
            if state[0][cols]==state[1][cols] and state[2][cols]==state[0][cols] and state[0][cols]==players[i]: # |
                return i, True
        if state[0][0]==state[1][1] and state[0][0]==state[2][2] and state[0][0]==players[i]:   # \
                return i, True
        if state[2][0]==state[1][1] and state[2][0]==state[0][2] and state[0][2]==players[i]:   # /
                return i, True
            
    return -1, False


has_won(state)

state = (('o','o','o'),
         ('o','x',0.0),
         (0.33,'o','x'))
print_state(state)
has_won(state)

state = (('o','x','o'),
         ('o','x',0.0),
         ('o','o','x'))
print_state(state)
has_won(state)

state = (('o','x','x'),
         ('o','x',0.0),
         ('x','o','x'))
print_state(state)
has_won(state)

state = (('o','x','o'),
         ('o','x',0.0),
         ('x','o','x'))
print_state(state)
has_won(state)


def is_full(state):
    for row in state:
        for v in row:
            if v!="x" and v!="o":
                return False
    return True


is_full((('o','x','o'),
         ('o','x',0.0),
         ('o','o','x')))

is_full((('o','x','o'),
         ('o','x','x'),
         ('o','o','x')))


def is_valid_move(state, x, y):
    row = state[x]
    if row[y]!="x" and row[y]!="o":
        return True
    return False


def new_state(state, x, y, character):
    l = []
    for row in state:
        l.append(list(row))
    l[x][y] = character
    return (tuple(l[0]), tuple(l[1]), tuple(l[2]))


state = (('o','x','o'),
         ('o','x',0.0),
         ('o','o','x'))
print_state(new_state(state, 1, 2, "x"))


class Player:
    def move(state):
        return state
    def learn(self, won):
        pass


class HumanPlayer(Player):
    def __init__(self, character):
        self.character=character
        
    def move(self, state):
        print_state(state)
        flag = False
        while not flag:
            print()
            x = int(input(f"({self.character}) row: "))
            y = int(input(f"({self.character}) col: "))
            flag=is_valid_move(state, x, y)
        print()
        
        return new_state(state, x, y, self.character)


h = HumanPlayer("x")
print_state(
    h.move((('o','x','o'),
            ('o','x',0.0),
            ('o','o','x')))
)


def do_game(player0, player1, stats, verbose=False):
    state = ((0.0, 0.0, 0.0), (0.0, 0.0, 0.0), (0.0, 0.0, 0.0))
    ended = False
    full = False
    won_won = -1
    
    while not ended and not full:
        state = player0.move(state)
        who_won, ended = has_won(state)
        full = is_full(state)        
        if not ended and not full:
            state = player1.move(state)
            who_won, ended = has_won(state)
            full = is_full(state)
    if ended:
        stats.append("WIN 0" if who_won==0 else "WIN 1")
        player0.learn(1.0 if who_won==0 else 0.0)
        player1.learn(1.0 if who_won==1 else 0.0)
    if full and not ended:
        stats.append("DRAW")    
        player0.learn(0.5)
        player1.learn(0.5)
    if verbose:
        print("DRAW" if full and not ended else f"{who_won} has WON!")
        print()
        print_state(state)


# +
import random

print([random.randrange(3) for x in range(100)])


# -

class RandomPlayer(Player):
    def __init__(self, character):
        self.character=character
        
    def move(self, state):
        flag = False
        while not flag:
            x = random.randrange(3)
            y = random.randrange(3)
            flag=is_valid_move(state, x, y)
       
        return new_state(state, x, y, self.character)


do_game(HumanPlayer('x'), RandomPlayer('o'), [], True)

# +
from collections import Counter

stats = []

for i in range(1_000):
    do_game(RandomPlayer('x'), RandomPlayer('o'), stats, False)
    
print(Counter(stats))
# -

1_000_000 == 1000000

1_000.0 == 1000.0

# +
# %matplotlib inline
import matplotlib.pyplot as plt
import numpy as np
plt.style.use("dark_background")

def plot_games(qstats, label0, label1):
    games = range(40, len(qstats), 20)
    won0 = [ Counter(qstats[:x])['WIN 0']/x for x in games]
    won1 = [ Counter(qstats[:x])['WIN 1']/x for x in games]
    draw = [ Counter(qstats[:x])['DRAW']/x for x in games]


    fig = plt.figure();
    plt.figure(figsize=(20,10));
    plt.title('Results ratio');
    player0, = plt.plot(games, won0, label=label0);
    player1, = plt.plot(games, won1, label=label1);
    draws, = plt.plot(games, draw, label="Draws");
    plt.legend(handles=[player0, player1,  draws], frameon=True, loc="best",  fontsize=16);
    plt.show();

plot_games(stats, "Random Player 0 Won", "Random Player 1 Won");
# -

# ---
# # _*Q-Learning*_

# https://en.wikipedia.org/wiki/Markov_chain

# https://en.wikipedia.org/wiki/Q-learning

# $
# Q: S \times A \rightarrow I\!R
# $
# ---
#
# ## W danym kroku $s$ podejmujemy akcję $a$ która maksymalizuje $Q$
#
# ---

#
# ## $Q$ Może zostać stabularyzowana (tzn. wsadzona do słownika)
# ## Kluczami są pary `(stan, akcja)` a wartością wynik $Q$
#

# ---
# # Uczymy się $Q$ korzytając ze __*Wzoru Bellmana*__
#
#
# $
# Q^{new}(s_t, a_t) = Q(s_t, a_t) + \alpha \cdot ( r_t +  \gamma \cdot   \max_{a}Q(s_{t+1}, a) - Q(s_t, a_t) )
# $
# ---
# Upraszczając:
#
# $
# Q^{new}(s_t, a_t) = (1- \alpha) \cdot Q(s_t, a_t) + \alpha \cdot  \gamma \cdot   \max_{a}Q(s_{t+1}, a)
# $
# ---
#
#
#
# $\alpha$ - Tempo uczenia się
#
# $\gamma$ - Degradacja wagi z odległością od bodźca
#
# $r_t$ - nagroda w danym stanie

# ---

α = 0.15 # Tempo uczenia się
β = 0.60 # Wartość początkowa `Q` - 'agresja'
γ = 0.99 # Degradacja wagi z odległością od bodźca 

# ---

# +
from random import sample


class QPlayer(Player):
    def __init__(self, character):
        self.character=character
        #################################################                     
        ## stabularyzowane Q  postaci:                 ##
        ## { stan1:                                    ##   
        ##     { nowy_stan1: q1, nowy_stan2: q2, ...}, ##
        ##   stan2:                                    ##
        ##     { nowy_stan3: q3, nowy_stan4: q4, ...}, ##
        ##  ... }                                      ##
        #################################################                     
        self.q_table={}
        
        self.previous_state=None
        self.current_state=((0.0, 0.0, 0.0), (0.0, 0.0, 0.0), (0.0, 0.0, 0.0))
        
    def initialize_q_table(self, state):    
        actions = {}   
        for x in range(3):
            for y in range(3):
                if is_valid_move(state, x, y):
                    ##################################
                    ## Inicjalizacja przez β (beta) ##
                    ##################################
                    actions[
                        new_state(state, x, y, self.character)
                    ] = β
        self.q_table[state] = actions
        
        return actions
        
    def move(self, state):
        
        ##############################################################
        ## Tworzę powiązanie z ruchem drugiego gracza,              ##
        ## Aby w Q-table był ciągły łańcuch od poprzedniego ruchu   ##
        ##############################################################
        
        if self.previous_state:
            actions = self.q_table.get(self.current_state, {})
            if state not in actions.keys():
                actions[state] = β
                self.q_table[self.current_state] = actions
            
        ##############################################################
        ## Jeśli w Q-table nie ma akcji powiązanych z obecnym       ##
        ## stanem, dopisuje wszystkie możliwe akcje z wagą β (beta) ##
        ##############################################################
        actions = self.q_table.get(state)
        if not actions:
            actions = self.initialize_q_table(state)

        ##############################################################
        ## Biorę najlepsze Q dla akcji w tym stanie                 ##
        ##############################################################
        best_q = max(actions.values())
        
        ##############################################################
        ## Losuję akcję pośród tych co mają najlepsze Q             ##
        ##############################################################
        best_actions = [ action for action, q in actions.items() if q==best_q ]
 
        self.previous_state = state
        self.current_state = sample(best_actions, 1)[0]

        return self.current_state
    
    def learn(self, learned_weight):
        
        self.q_table[self.previous_state][self.current_state] = learned_weight
        self.previous_state=None
        
        for state, actions in self.q_table.items():
            for next_move, q in actions.items():
                next_move_actions = self.q_table.get(next_move)
                if next_move_actions:
                    best_next_q = max(next_move_actions.values())
                   
                    #########################
                    ##    <NAME>    ##
                    #########################
                    actions[next_move] = (1-α)*q + γ*α*best_next_q 
                    
                    self.q_table[state] = actions



# -

qstats = []
qplayer = QPlayer('o')

# +
from collections import Counter

for i in range(1):
    do_game(RandomPlayer('x'), qplayer, qstats, False)
    
print(Counter(qstats))

# -

from pprint import pprint 
pprint(qplayer.q_table)

# +
from collections import Counter

for i in range(1):
    do_game(RandomPlayer('x'), qplayer, qstats, False)
    
print(Counter(qstats))


# -

from pprint import pprint 
pprint(qplayer.q_table)

# +
for i in range(2_000):
    do_game(RandomPlayer('x'), qplayer,  qstats, False)
    
print(Counter(qstats))
plot_games(qstats, "Random Player 0 Won", "Q Player 1 Won")

# +
for i in range(2_000):
    do_game(RandomPlayer('x'), qplayer,  qstats, False)
    
print(Counter(qstats))
plot_games(qstats, "Random Player 0 Won", "Q Player 1 Won")


# +
def plot_2_game(qstats, stats, labelA, labelB):
    qgames = range(40, len(qstats), 20)
    qwon0 = [ Counter(qstats[:x])['WIN 0']/x for x in qgames]
    qwon1 = [ Counter(qstats[:x])['WIN 1']/x for x in qgames]
    qdraw = [ Counter(qstats[:x])['DRAW']/x for x in qgames]
    games = range(40, len(stats), 20)
    won0 = [ Counter(stats[:x])['WIN 0']/x for x in games]
    won1 = [ Counter(stats[:x])['WIN 1']/x for x in games]
    draw = [ Counter(stats[:x])['DRAW']/x for x in games]


    plt.rcParams['figure.figsize'] = [20,10]
 
    fig = plt.figure();
 
    l = fig.add_subplot(121)   
    plt.ylim([0.0, 0.9])
    player0, = l.plot(qgames, qwon0, label="Random Player 0");
    player1, = l.plot(qgames, qwon1, label=labelA);
    draws, = l.plot(qgames, qdraw, label="Draws");
    l.legend(handles=[player0, player1,  draws], frameon=True, loc="best", fontsize=16);

    r = fig.add_subplot(122)  
    plt.ylim([0.0, 0.9])
    player0, = r.plot(games, won0, label="Random Player 0");
    player1, = r.plot(games, won1, label=labelB);
    draws, = r.plot(games, draw, label="Draws");
    r.legend(handles=[player0, player1,  draws], frameon=True, loc="best", fontsize=16);
    plt.show();

plot_2_game(qstats, stats, "Q-Player 1", "Random Player 1")

# +
qstats2 = []
for i in range(1_000):
    do_game(RandomPlayer('x'), qplayer,  qstats2, False)
    
print(Counter(qstats2))
plot_games(qstats2, "Random Player 0 Won", "Q Player 1 Won")
# -

# ___

# # __*A/B testing*__

# ## Za optimizely:
#
# ### __*A/B testing*__ (also known as split testing or bucket testing) is a method of __*comparing two versions*__ of a webpage or app against each other to determine which one performs __*better*__.
#

# #### Co to znaczy __*better*__
# - Bounce rate
# - Conversion
# - Click-Through Rate
# - Etc. 
#

# ---
# ## A/B Testing ma zazwyczaj dwie fazy
# - ### __*Eksploracji*__ - sprawdzamy która lepsza jest lepsza
# - ### __*Eksploatacji*__ - "lepsza" wersja działa jako jedyna

# ## Wady typowego podejścia
# - Nie ma płynnego przejścia z 2 wersji na 1
# - W czasie eksploracji sprawdzamy wiele dużo wersji gorszych od wersji kontrolnej
# - __*Eksploracja w praktyce nigdy się nie kończy*__

# ---
# # Problem Wielorękiego Bandyty
#
# ![](img\Las_Vegas_slot_machines.jpg)
#
# https://en.wikipedia.org/wiki/Multi-armed_bandit
#
# Mamy budżet aby jak najwięcej "ugrać". Poszczególni jednoręcy bandyci różnią się między sobą szansą na wygraną. 
#
# Ile spędzić budzetu na eksplorację najlepszej maszyny, a ile na osiągnięcie wygranej ?
#
# Istnieje olbrzymi katalog algorytmów (ang. __bandit algorithms__) które na to odpowiadają.
# - ε-greedy
# - soft-max
# - soft-max z wyżarzaniem
# - Upper Confidence Bound
# - Exp3
# - etc. etc.
#
# ![](img\banditalgos.jpg)
#
# ---
# ## Algorytm ε-zachłanny - ε-greedy
#
# - Z prawdopodobieństwem `1-ε`  wybieramy strategię kontrolną `A` do eksploatacji. 
#
# - Z prawdopodobieństwem `ε` (np. 20%) wybieramy strategię `B`  do eksploracji. 
#
#

# ![](img\epsilongreedy.png)

# ---
# ## Algorytm Soft-max
#
# $r_A$ - proporcja sukcesów w gałęzi `A`
#
# $r_B$ - proporcja sukcesów w gałęzi `B`
#
#
# - Z prawdopodobieństwem $$\frac{r_A}{r_A+r_B}$$ wybieramy strategię kontrolną `A` do eksploatacji. 
#
# - Z prawdopodobieństwem $$\frac{r_B}{r_A+r_B}$$ wybieramy strategię `B`  do eksploracji. 

# ---
# ## Algorytm Soft-max z wyżarzaniem
#
# $r_A$ - proporcja sukcesów w gałęzi `A`
#
# $r_B$ - proporcja sukcesów w gałęzi `B`
#
# $\tau$ - "temperatura"
#
#
# - Z prawdopodobieństwem $$\frac{exp(r_A/\tau)}{exp(r_A/\tau)+exp(r_B/\tau)}$$ wybieramy strategię kontrolną `A` do eksploatacji. 
#
# - Z prawdopodobieństwem $$\frac{exp(r_B/\tau)}{exp(r_A/\tau)+exp(r_B/\tau)}$$ wybieramy strategię `B`  do eksploracji. 
#
# ---
# `t = sum(self.counts) + 1`
#
# `tau = 1 / math.log(t + 0.0000001)`
#
#
# ---
# ## __*Wyżarzanie*__
#
# ### __*Wyżarzanie*__ - to obniżanie "temperatury" czyli zmienności układu z czasem
#
# ### Dzięki __*wyżarzaniu*__ - system z czasem coraz rzadziej eksploruje.
#
# #### __*Hartowanie*__ to zwiększanie temperatury raz na jakiś czas aby "wybić" system z minimów lokalnych 
#

# ---
# ## Q-Player z __*Wyżarzaniem*__
#
# Zmniejszamy `α` przez mnożenie przez `γ` co krok - przez co spada zmienność układu

# +
α = 0.99  # Tempo uczenia się - "temperatura"
β = 0.80  # Wartość początkowa `Q` - 'agresja'
γ = 0.975 # Degradacja wagi z odległością od bodźca oraz tempo spadku "temperatury"

class AnnealingQPlayer(QPlayer):
    def __init__(self, character):
        super().__init__(character)
        
    def learn(self, learned_weight):
        global α
        
        ################
        ## Wyżarzanie ##
        ################
        α *= γ
        
        super().learn(learned_weight)



# -

annealing_qplayer = AnnealingQPlayer('o')

# +
aqstats = []
for i in range(2_500):
    do_game(RandomPlayer('x'), annealing_qplayer,  aqstats, False)
    
print(Counter(aqstats))
plot_games(aqstats, "Random Player 0 Won", "Annealing Q Player 1 Won")

# +
for i in range(2_500):
    do_game(RandomPlayer('x'), annealing_qplayer,  aqstats, False)
    
print(Counter(aqstats))
plot_games(aqstats, "Random Player 0 Won", "Annealing Q Player 1 Won")
# -

plot_2_game(qstats, aqstats, "Q-Player 1", "Annealing Q-Player 1")

# +
aqstats2 = []
for i in range(1000):
    do_game(RandomPlayer('x'), annealing_qplayer,  aqstats2, False)
    
print(Counter(aqstats2))


plot_2_game(qstats2, aqstats2, "Q-Player 1", "Annealing Q-Player 1")
# -

# ---
# # Zadanie 1
#
# Zagrajcie w kółko i krzyżyk

StarcoderdataJupyterScriptsDedupFiltered