Chapter 56: Reinforcement Learning for Game Competitions

import random
from functools import lru_cache

import numpy as np

from kaggle_companion.activities._common import clean

SEED = 56
CELLS, MINE, STEPS = 14, 3, 30    # corridor of 14 cells, a resource cell at 3, the goal at the far end, 30 steps to win
SLIP, GAMMA = 0.1, 0.97           # a move slips (reverses) 10% of the time; discount for the learner
EPISODES, LEARNERS, EVAL_GAMES = 800, 8, 200
ALPHA, EPSILON = 0.2, 0.1
OPTIMISM = 1.0                    # optimistic initial action values make the learner try every action before settling


def move(state, action, rng):
    """One step of the corridor game: action 1 is right, 0 is left; the move slips with probability SLIP."""
    if rng.random() < SLIP:
        action = 1 - action
    return min(CELLS - 1, max(0, state + (1 if action else -1)))


def reward(state, nxt, bonus, kind):
    """True reward (+1 for winning) plus shaping. 'bonus' pays for each fresh arrival at the resource cell;
    'potential' pays the change in a progress potential (gamma * phi(next) - phi(state)), which cannot change the best policy."""
    done = nxt == CELLS - 1
    r = 1.0 if done else 0.0
    if kind == "bonus" and nxt == MINE and state != MINE:
        r += bonus
    if kind == "potential":
        phi = lambda s: bonus * s / (CELLS - 1)
        r += GAMMA * (0.0 if done else phi(nxt)) - phi(state)
    return r


def train(bonus, kind, seed):
    """Tabular Q-learning with epsilon-greedy exploration; returns the greedy policy (one action per cell)."""
    rng = random.Random(seed)
    q = [[OPTIMISM, OPTIMISM] for _ in range(CELLS)]
    for _ in range(EPISODES):
        state = 0
        for _ in range(STEPS):
            if rng.random() < EPSILON:
                action = rng.randrange(2)
            else:
                action = 0 if q[state][0] > q[state][1] else 1 if q[state][1] > q[state][0] else rng.randrange(2)
            nxt = move(state, action, rng)
            done = nxt == CELLS - 1
            target = reward(state, nxt, bonus, kind) + (0.0 if done else GAMMA * max(q[nxt]))
            q[state][action] += ALPHA * (target - q[state][action])
            state = nxt
            if done:
                break
    return [0 if row[0] > row[1] else 1 for row in q]


def optimal_policy(bonus):
    """Value iteration on the same game with the bonus reward: the best any learner could settle on."""
    q = np.zeros((CELLS, 2))
    for _ in range(600):
        for s in range(CELLS - 1):
            for a in (0, 1):
                total = 0.0
                for actual, p in ((a, 1 - SLIP), (1 - a, SLIP)):
                    nxt = min(CELLS - 1, max(0, s + (1 if actual else -1)))
                    done = nxt == CELLS - 1
                    total += p * (reward(s, nxt, bonus, "bonus") + (0.0 if done else GAMMA * q[nxt].max()))
                q[s, a] = total
    return [int(q[s, 1] >= q[s, 0]) for s in range(CELLS)]


def play(policy, bonus, rng):
    """Mean win rate and mean shaped return (win plus bonus per resource arrival) of a policy over fresh games."""
    wins, shaped = 0, 0.0
    for _ in range(EVAL_GAMES):
        state = 0
        for _ in range(STEPS):
            nxt = move(state, policy[state], rng)
            shaped += reward(state, nxt, bonus, "bonus")
            state = nxt
            if state == CELLS - 1:
                wins += 1
                break
    return wins / EVAL_GAMES, shaped / EVAL_GAMES


@lru_cache(maxsize=None)
def sparse_policies(seed):
    """The learners trained on the true reward only; they do not depend on the bonus, so they are trained once."""
    return [train(0.0, "bonus", seed * 100 + i) for i in range(LEARNERS)]


def run(bonus):
    rng = random.Random(SEED)
    sparse = [play(p, bonus, rng) for p in sparse_policies(SEED)]
    shaped = [play(train(bonus, "bonus", SEED * 100 + 50 + i), bonus, rng) for i in range(LEARNERS)]
    potential = [play(train(bonus, "potential", SEED * 100 + 80 + i), bonus, rng) for i in range(LEARNERS)]
    best = play(optimal_policy(bonus), bonus, rng)
    column = lambda rows, k: [row[k] for row in rows]
    return clean({
        "bonus": bonus,
        "win": {"sparse": float(np.mean(column(sparse, 0))), "bonus": float(np.mean(column(shaped, 0))),
                "potential": float(np.mean(column(potential, 0))), "optimal": best[0]},
        "shaped_return": {"sparse": float(np.mean(column(sparse, 1))), "bonus": float(np.mean(column(shaped, 1))),
                          "optimal": best[1]},
        "per_learner_win": {"sparse": column(sparse, 0), "bonus": column(shaped, 0), "potential": column(potential, 0)},
        "learners": LEARNERS, "games": EVAL_GAMES,
    })

import pandas as pd
from kaggle_companion.activities.ch56 import explain

VALUES = [0.02, 0.05, 0.1, 0.5]
results = {value: run(value) for value in VALUES}
table = pd.DataFrame({str(v): explain(results[v], v)['metrics'] for v in VALUES})
print(table.to_string())

import matplotlib.pyplot as plt
from kaggle_companion.activities.ch56 import draw, NCOLS, HEIGHT

DEFAULT = 0.1
fig, axes = plt.subplots(1, NCOLS, figsize=(6.6 if NCOLS == 1 else 10.4, HEIGHT), layout='constrained')
draw(axes, results[DEFAULT], DEFAULT)
plt.show()