rl_blackjack.py

# -*- coding: utf-8 -*-
"""RL_BlackJack.ipynb

Automatically generated by Colab.

Original file is located at
    https://colab.research.google.com/drive/1HGS7_jyffKojNLWeRaD6mssy5_OsVR58

Blackjack RL Agent — Q-Learning vs Monte Carlo Train two RL agents to play Blackjack and compare their performance.

Actions: H = Hit  |  S = Stand  |  D = Double Down
Reward: +1 win, -1 loss, 0 draw (doubled for double down)

libaries and globle variables
"""

import numpy as np
import random
import os
import csv
import json
import matplotlib.pyplot as plt
import seaborn as sns
import pandas as pd

SEED = 42
random.seed(SEED)
np.random.seed(SEED)

NUM_DECKS   = 8        # Number of decks in the shoe
EPISODES    = 200000   # Training episodes per agent
EVAL_GAMES  = 50000    # Evaluation games
MIN_EPSILON = 0.05     # Minimum exploration rate
DECAY       = 0.99999  # Epsilon decay per episode

# Q-Learning hyperparameters
QL_ALPHA    = 0.01     # Learning rate
QL_GAMMA    = 0.95     # Discount factor

# Monte Carlo hyperparameters
MC_ALPHA    = 0.02     # Learning rate
MC_GAMMA    = 1.0      # Discount factor

# Output Directory
# All CSVs, JSONs, and charts save here
LOG_DIR = "experiment_logs"
os.makedirs(LOG_DIR, exist_ok=True)

"""Card & Deck Functions"""

# Map card ranks to numeric values
CARD_VALUE = {str(i): i for i in range(2, 11)}
CARD_VALUE.update({'jack': 10, 'queen': 10, 'king': 10, 'ace': 11})

def make_deck():
    #Create and shuffle a multi-deck shoe.
    suits = ['hearts', 'diamonds', 'clubs', 'spades']
    ranks = [str(i) for i in range(2, 11)] + ['jack', 'queen', 'king', 'ace']
    deck  = [(rank, suit) for rank in ranks for suit in suits] * NUM_DECKS
    random.shuffle(deck)
    return deck

def hand_value(hand):
    #Calculate best hand value, handling soft aces.
    total = 0
    aces  = 0
    for rank, suit in hand:
        if rank == 'ace':
            total += 11
            aces  += 1
        else:
            total += CARD_VALUE[rank]
    # Convert aces from 11 -> 1 to avoid busting
    while total > 21 and aces > 0:
        total -= 10
        aces  -= 1
    return total

def is_soft(hand):
    #True if hand has an ace counted as 11 (soft hand).
    return any(r == 'ace' for r, s in hand) and hand_value(hand) <= 21

"""## 🎰 Game Logic"""

# Shared deck (reshuffle automatically when low)
deck = make_deck()

def draw():
    #Draw one card, reshuffling the shoe if needed.
    global deck
    if len(deck) < 20:
        deck = make_deck()
    return deck.pop()

def new_game():
    #Deal starting hands and return the initial state.
    global player, dealer, doubled, done
    player  = [draw(), draw()]
    dealer  = [draw(), draw()]
    doubled = False
    done    = False
    return get_state()

def get_state():
    """
    State = (player total, dealer upcard value, is soft hand, has doubled)
    This is what the agent 'sees' when making a decision.
    """
    return (hand_value(player), CARD_VALUE[dealer[0][0]], is_soft(player), doubled)

def step(action):
    """
    Take an action and return (next_state, reward, done).
    action: 'h' = hit, 's' = stand, 'd' = double down
    """
    global player, dealer, doubled, done

    # HIT
    if action == 'h':
        player.append(draw())
        if hand_value(player) > 21:   # bust
            done = True
            return 'terminal', -1, True
        return get_state(), 0, False

    #STAND
    if action == 's':
        while hand_value(dealer) < 17:   # dealer hits until 17+
            dealer.append(draw())
        done = True
        p, d = hand_value(player), hand_value(dealer)
        if d > 21 or p > d:  return 'terminal',  1, True   # player wins
        if p < d:            return 'terminal', -1, True   # dealer wins
        return 'terminal', 0, True                         # push

    # DOUBLE DOWN
    if action == 'd' and len(player) == 2:
        doubled = True
        player.append(draw())
        if hand_value(player) > 21:
            done = True
            return 'terminal', -2, True
        while hand_value(dealer) < 17:
            dealer.append(draw())
        done = True
        p, d = hand_value(player), hand_value(dealer)
        if d > 21 or p > d:  return 'terminal',  2, True
        if p < d:            return 'terminal', -2, True
        return 'terminal', 0, True

    # Fallback (invalid action)
    return get_state(), 0, False

def get_actions():
    """Available actions — double only allowed on first two cards."""
    return ['h', 's', 'd'] if len(player) == 2 else ['h', 's']

"""Q-Learning Agent
Updates the Q-table after every single step
"""

ql_table   = {}   # Q-table: {(state, action): q_value}
ql_epsilon = 1.0  # Start fully random, decay over time

def ql_get_q(state, action):
    return ql_table.get((state, action), 0.0)

def ql_choose(state, actions):
    """Epsilon-greedy: explore randomly or exploit best known action."""
    if random.random() < ql_epsilon:
        return random.choice(actions)
    return max(actions, key=lambda a: ql_get_q(state, a))

def ql_update(state, action, reward, next_state, done):
    """Bellman update."""
    old_q    = ql_get_q(state, action)
    next_max = 0 if done else max(ql_get_q(next_state, a) for a in ['h', 's', 'd'])
    new_q    = old_q + QL_ALPHA * (reward + QL_GAMMA * next_max - old_q)
    ql_table[(state, action)] = new_q

#Train
ql_rewards = []
ql_log     = []

print("Training Q-Learning agent...")
for ep in range(EPISODES):
    state = new_game()
    total = 0
    while True:
        actions         = get_actions()
        action          = ql_choose(state, actions)
        next_s, r, done = step(action)
        ql_update(state, action, r, next_s, done)
        state = next_s
        total += r
        if done:
            break
    ql_rewards.append(total)
    if ql_epsilon > MIN_EPSILON:
        ql_epsilon *= DECAY

    # Save a log entry every 1,000 episodes
    if (ep + 1) % 1000 == 0:
        avg = np.mean(ql_rewards[-1000:])
        ql_log.append({
            'episode':      ep + 1,
            'avg_reward':   round(avg, 4),
            'epsilon':      round(ql_epsilon, 4),
            'q_table_size': len(ql_table)
        })

    if (ep + 1) % 50000 == 0:
        avg = np.mean(ql_rewards[-10000:])
        print(f"  Ep {ep+1:,} | ε={ql_epsilon:.3f} | "
              f"avg_reward={avg:.4f} | Q-table={len(ql_table):,} entries")

print(f"Done! Q-table has {len(ql_table)} entries.")

# Save log to CSV
with open(f"{LOG_DIR}/ql_training_log.csv", 'w', newline='') as f:
    w = csv.DictWriter(f, fieldnames=['episode', 'avg_reward', 'epsilon', 'q_table_size'])
    w.writeheader(); w.writerows(ql_log)
print(f"  Saved: {LOG_DIR}/ql_training_log.csv")

"""Monte Carlo Agent
Waits until the end of a full episode, then updates every state visited using the actual return G:
"""

mc_table   = {}   # Q-table: {(state, action): q_value}
mc_epsilon = 1.0
mc_log     = []

def mc_get_q(state, action):
    return mc_table.get((state, action), 0.0)

def mc_choose(state, actions):
    if random.random() < mc_epsilon:
        return random.choice(actions)
    return max(actions, key=lambda a: mc_get_q(state, a))

# Train
mc_rewards = []

print("Training Monte Carlo agent...")
for ep in range(EPISODES):
    # Step 1: Play a full episode and record every (state, action, reward)
    episode = []
    state   = new_game()
    while True:
        actions         = get_actions()
        action          = mc_choose(state, actions)
        next_s, r, done = step(action)
        episode.append((state, action, r))
        state = next_s
        if done:
            break

    # Step 2: Work backwards, compute return G, update Q (first-visit only)
    G       = 0
    visited = set()
    for state, action, reward in reversed(episode):
        G = MC_GAMMA * G + reward
        if (state, action) not in visited:
            visited.add((state, action))
            old_q = mc_get_q(state, action)
            mc_table[(state, action)] = old_q + MC_ALPHA * (G - old_q)

    mc_rewards.append(G)
    if mc_epsilon > MIN_EPSILON:
        mc_epsilon *= DECAY

    if (ep + 1) % 1000 == 0:
        avg = np.mean(mc_rewards[-1000:])
        mc_log.append({
            'episode':      ep + 1,
            'avg_reward':   round(avg, 4),
            'epsilon':      round(mc_epsilon, 4),
            'q_table_size': len(mc_table)
        })

    if (ep + 1) % 50000 == 0:
        avg = np.mean(mc_rewards[-10000:])
        print(f"  Ep {ep+1:,} | ε={mc_epsilon:.3f} | "
              f"avg_reward={avg:.4f} | Q-table={len(mc_table):,} entries")

print(f"Done! MC Q-table has {len(mc_table)} entries.")

with open(f"{LOG_DIR}/mc_training_log.csv", 'w', newline='') as f:
    w = csv.DictWriter(f, fieldnames=['episode', 'avg_reward', 'epsilon', 'q_table_size'])
    w.writeheader(); w.writerows(mc_log)
print(f"  Saved: {LOG_DIR}/mc_training_log.csv")

"""Hyperparameters Experiments"""

HYPERPARAM_EPISODES = 100000   # Shorter run for quick comparison
ALPHA_VARIANTS      = [0.001, 0.01, 0.1]   # Learning rates to test

hyperparam_results = {}   # Store curves for each alpha value

print("Running Hyperparameter Experiments (Alpha variants)")

for alpha in ALPHA_VARIANTS:
    temp_table   = {}
    temp_epsilon = 1.0
    temp_rewards = []

    for ep in range(HYPERPARAM_EPISODES):
        state = new_game()
        total = 0
        while True:
            actions = get_actions()
            # Epsilon-greedy
            if random.random() < temp_epsilon:
                action = random.choice(actions)
            else:
                action = max(actions, key=lambda a: temp_table.get((state, a), 0.0))
            next_s, r, done = step(action)
            # Bellman update with this alpha
            old_q    = temp_table.get((state, action), 0.0)
            next_max = 0 if done else max(temp_table.get((next_s, a), 0.0) for a in ['h','s','d'])
            temp_table[(state, action)] = old_q + alpha * (r + QL_GAMMA * next_max - old_q)
            state = next_s
            total += r
            if done:
                break
        temp_rewards.append(total)
        if temp_epsilon > MIN_EPSILON:
            temp_epsilon *= DECAY

    hyperparam_results[alpha] = temp_rewards
    final_avg = np.mean(temp_rewards[-5000:])
    print(f"  alpha={alpha:.3f} | Final avg reward (last 5k): {final_avg:.4f}")

# Save hyperparameter CSV
hp_rows = []
for alpha, rewards in hyperparam_results.items():
    for i, r in enumerate(rewards):
        if (i + 1) % 500 == 0:
            hp_rows.append({'alpha': alpha, 'episode': i+1,
                            'avg_reward': round(np.mean(rewards[max(0,i-500):i+1]), 4)})

with open(f"{LOG_DIR}/hyperparameter_experiment.csv", 'w', newline='') as f:
    writer = csv.DictWriter(f, fieldnames=['alpha', 'episode', 'avg_reward'])
    writer.writeheader()
    writer.writerows(hp_rows)
print(f"Hyperparameter log saved to {LOG_DIR}/hyperparameter_experiment.csv\n")

"""Convergence Analysis"""

WINDOW_SIZE       = 10000    # Check every 10,000 episodes
CONVERGENCE_DELTA = 0.005    # If change < 0.5%, consider converged


def check_convergence(rewards, window=WINDOW_SIZE, delta=CONVERGENCE_DELTA):

    window_avgs     = []
    converged_at    = None
    stable_count    = 0

    for i in range(0, len(rewards) - window, window // 2):
        chunk = rewards[i: i + window]
        avg   = np.mean(chunk)
        window_avgs.append((i + window, avg))

        if len(window_avgs) > 1:
            change = abs(window_avgs[-1][1] - window_avgs[-2][1])
            if change < delta:
                stable_count += 1
            else:
                stable_count = 0  # Reset if it jumps again

        if stable_count >= 3 and converged_at is None:
            converged_at = i + window

    return converged_at, window_avgs


ql_converge_ep, ql_window_avgs = check_convergence(ql_rewards)
mc_converge_ep, mc_window_avgs = check_convergence(mc_rewards)

print("Convergence Analysis")

if ql_converge_ep:
    print(f"  Q-Learning converged at episode ~{ql_converge_ep:,}")
else:
    print("  Q-Learning: convergence not clearly detected")

if mc_converge_ep:
    print(f"  Monte Carlo converged at episode ~{mc_converge_ep:,}")
else:
    print("  Monte Carlo: convergence not clearly detected")

# Save convergence data
conv_data = {
    'ql_converged_at': ql_converge_ep,
    'mc_converged_at': mc_converge_ep,
    'ql_final_avg':    round(np.mean(ql_rewards[-10000:]), 4),
    'mc_final_avg':    round(np.mean(mc_rewards[-10000:]), 4),
}
with open(f"{LOG_DIR}/convergence_analysis.json", 'w') as f:
    json.dump(conv_data, f, indent=2)
print(f"Convergence data saved to {LOG_DIR}/convergence_analysis.json\n")

"""Evaluate Both Agents"""

def evaluate(q_table, label, games=EVAL_GAMES):

    """Run greedy policy (no exploration) and count wins/losses/draws."""

    wins = losses = draws = 0

    for _ in range(games):
        state = new_game()
        while True:
            actions         = get_actions()
            action          = max(actions, key=lambda a: q_table.get((state, a), 0.0))
            next_s, r, done = step(action)
            state = next_s
            if done:
                break
        if   r > 0: wins   += 1
        elif r < 0: losses += 1
        else:       draws  += 1

    win_pct  = wins   / games * 100
    loss_pct = losses / games * 100
    draw_pct = draws  / games * 100
    edge     = (wins - losses) / games * 100

    print(f"  {label} - Evaluation ({games:,} games)")
    print(f"  Win  %      : {win_pct:.1f}%")
    print(f"  Loss %      : {loss_pct:.1f}%")
    print(f"  Draw %      : {draw_pct:.1f}%")
    print(f"  Player Edge : {edge:+.2f}%  (positive = agent is profitable)")
    return wins, losses, draws


print("Evaluating Both Agents (Greedy Policy, No Exploration)")
ql_w, ql_l, ql_d = evaluate(ql_table, "Q-LEARNING")
mc_w, mc_l, mc_d = evaluate(mc_table, "MONTE CARLO")

# Save evaluation results
eval_data = {
    'Q-Learning': {
        'wins': ql_w, 'losses': ql_l, 'draws': ql_d,
        'win_pct': round(ql_w/EVAL_GAMES*100, 2),
        'loss_pct': round(ql_l/EVAL_GAMES*100, 2),
        'draw_pct': round(ql_d/EVAL_GAMES*100, 2),
        'edge': round((ql_w-ql_l)/EVAL_GAMES*100, 2)
    },
    'Monte Carlo': {
        'wins': mc_w, 'losses': mc_l, 'draws': mc_d,
        'win_pct': round(mc_w/EVAL_GAMES*100, 2),
        'loss_pct': round(mc_l/EVAL_GAMES*100, 2),
        'draw_pct': round(mc_d/EVAL_GAMES*100, 2),
        'edge': round((mc_w-mc_l)/EVAL_GAMES*100, 2)
    }
}
with open(f"{LOG_DIR}/evaluation_results.json", 'w') as f:
    json.dump(eval_data, f, indent=2)
print(f"\nEvaluation results saved to {LOG_DIR}/evaluation_results.json")

"""## 📈 Visualize Results"""

WINDOW = 2000   # Rolling average window for smooth curves

fig, axes = plt.subplots(3, 3, figsize=(20, 18))
fig.suptitle("Blackjack RL: Q-Learning vs Monte Carlo\nFull Analysis",
             fontsize=16, fontweight='bold', y=1.01)

# Chart 1: Learning Curves
ax = axes[0, 0]
ax.plot(pd.Series(ql_rewards).rolling(WINDOW).mean(),
        color='steelblue', label='Q-Learning', linewidth=1.2)
ax.plot(pd.Series(mc_rewards).rolling(WINDOW).mean(),
        color='darkorange', label='Monte Carlo', linewidth=1.2)
ax.axhline(0, color='gray', linestyle='--', alpha=0.5)
# Mark convergence points
if ql_converge_ep:
    ax.axvline(ql_converge_ep, color='steelblue', linestyle=':', alpha=0.7,
               label=f'QL converge ~{ql_converge_ep:,}')
if mc_converge_ep:
    ax.axvline(mc_converge_ep, color='darkorange', linestyle=':', alpha=0.7,
               label=f'MC converge ~{mc_converge_ep:,}')
ax.set_title('Learning Curves (Rolling Avg Reward)')
ax.set_xlabel('Episode'); ax.set_ylabel('Avg Reward')
ax.legend(fontsize=8); ax.grid(alpha=0.3)

# Chart 2: Win/Loss/Draw Bar Chart
ax = axes[0, 1]
cats = ['Win %', 'Loss %', 'Draw %']
ql_v = [ql_w/EVAL_GAMES*100, ql_l/EVAL_GAMES*100, ql_d/EVAL_GAMES*100]
mc_v = [mc_w/EVAL_GAMES*100, mc_l/EVAL_GAMES*100, mc_d/EVAL_GAMES*100]
x = np.arange(3)
b1 = ax.bar(x - 0.2, ql_v, 0.4, label='Q-Learning',  color='steelblue',  alpha=0.85)
b2 = ax.bar(x + 0.2, mc_v, 0.4, label='Monte Carlo', color='darkorange', alpha=0.85)
ax.set_title('Win / Loss / Draw Comparison')
ax.set_xticks(x); ax.set_xticklabels(cats); ax.set_ylabel('%')
ax.legend(); ax.grid(axis='y', alpha=0.3)
for b in list(b1) + list(b2):
    ax.text(b.get_x()+b.get_width()/2, b.get_height()+0.2,
            f'{b.get_height():.1f}%', ha='center', fontsize=8)

# Chart 3: Player Edge
ax = axes[0, 2]
edges  = [(ql_w-ql_l)/EVAL_GAMES*100, (mc_w-mc_l)/EVAL_GAMES*100]
colors = ['steelblue' if e >= 0 else 'tomato' for e in edges]
bars   = ax.bar(['Q-Learning', 'Monte Carlo'], edges, color=colors, alpha=0.85, width=0.4)
ax.axhline(0, color='black', linewidth=1)
ax.set_title('Player Edge (Win% − Loss%)')
ax.set_ylabel('Edge %'); ax.grid(axis='y', alpha=0.3)
for bar, e in zip(bars, edges):
    ax.text(bar.get_x()+bar.get_width()/2, e + (0.1 if e >= 0 else -0.4),
            f'{e:+.2f}%', ha='center', fontweight='bold')

# Chart 4: Convergence Window Plot
ax = axes[1, 0]
ql_x, ql_y = zip(*ql_window_avgs) if ql_window_avgs else ([], [])
mc_x, mc_y = zip(*mc_window_avgs) if mc_window_avgs else ([], [])
ax.plot(ql_x, ql_y, 'o-', color='steelblue',  label='Q-Learning',  markersize=4)
ax.plot(mc_x, mc_y, 's-', color='darkorange', label='Monte Carlo', markersize=4)
ax.axhline(0, color='gray', linestyle='--', alpha=0.5)
ax.set_title('Convergence Analysis\n(Window Avg Reward)')
ax.set_xlabel('Episode'); ax.set_ylabel('Avg Reward per Window')
ax.legend(); ax.grid(alpha=0.3)

# Chart 5: Hyperparameter Experiment (Alpha comparison)
ax = axes[1, 1]
colors_hp = ['#1f77b4', '#2ca02c', '#d62728', '#9467bd']
for (alpha, rewards), col in zip(hyperparam_results.items(), colors_hp):
    smooth = pd.Series(rewards).rolling(3000).mean()
    ax.plot(smooth, label=f'α={alpha}', color=col, linewidth=1.2)
ax.axhline(0, color='gray', linestyle='--', alpha=0.5)
ax.set_title('Q-Learning: Hyperparameter Experiment\n(Different Learning Rates α)')
ax.set_xlabel('Episode'); ax.set_ylabel('Avg Reward')
ax.legend(fontsize=9); ax.grid(alpha=0.3)

# Chart 6: Q-Table Size Comparison
ax = axes[1, 2]
sizes = [len(ql_table), len(mc_table)]
bars  = ax.bar(['Q-Learning', 'Monte Carlo'], sizes,
               color=['steelblue', 'darkorange'], alpha=0.85, width=0.4)
ax.set_title('Q-Table Size\n(Number of States Explored)')
ax.set_ylabel('Entries'); ax.grid(axis='y', alpha=0.3)
for i, v in enumerate(sizes):
    ax.text(i, v + 5, f'{v:,}', ha='center', fontweight='bold')

# Chart 7 & 8: Policy Heatmaps
def policy_pivot(q_table):
    """Build a grid: Player Total × Dealer Upcard → Best Action"""
    rows = []
    for pv in range(4, 22):
        for du in range(2, 12):
            state = (pv, du, False, False)
            best  = max(['h', 's', 'd'], key=lambda a: q_table.get((state, a), 0.0))
            rows.append({'Player Total': pv, 'Dealer Upcard': du, 'Action': best})
    df = pd.DataFrame(rows)
    return df.pivot(index='Player Total', columns='Dealer Upcard', values='Action')

a2n  = {'h': 0, 's': 1, 'd': 2}
cmap = plt.colormaps.get_cmap('RdYlGn').resampled(3)

for ax, q_table, title in [
    (axes[2, 0], ql_table, 'Q-Learning Policy'),
    (axes[2, 1], mc_table, 'Monte Carlo Policy'),
]:
    pivot = policy_pivot(q_table)
    nums  = pivot.map(lambda x: a2n.get(x, -1))
    sns.heatmap(nums, ax=ax, cmap=cmap, annot=pivot, fmt='',
                linewidths=0.3, cbar_kws={'ticks': [0, 1, 2]}, vmin=0, vmax=2)
    ax.set_title(f'{title} (Hard Hands)\nGreen=Stand  Yellow=Double  Red=Hit',
                 fontweight='bold')
    ax.invert_yaxis()

# Chart 9: Reward Distribution Histogram
ax = axes[2, 2]
ax.hist(ql_rewards[-50000:], bins=[-1.5, -0.5, 0.5, 1.5, 2.5],
        alpha=0.6, color='steelblue', label='Q-Learning', density=True)
ax.hist(mc_rewards[-50000:], bins=[-2.5, -1.5, -0.5, 0.5, 1.5, 2.5],
        alpha=0.6, color='darkorange', label='Monte Carlo', density=True)
ax.set_title('Reward Distribution\n(Last 50k Episodes)')
ax.set_xlabel('Reward'); ax.set_ylabel('Density')
ax.set_xticks([-2, -1, 0, 1, 2])
ax.set_xticklabels(['-2\n(DD Loss)', '-1\n(Loss)', '0\n(Draw)', '+1\n(Win)', '+2\n(DD Win)'])
ax.legend(); ax.grid(alpha=0.3)

plt.tight_layout()
plt.savefig(f"{LOG_DIR}/full_analysis.png", dpi=150, bbox_inches='tight')
plt.show()
print(f"\nFull analysis chart saved to {LOG_DIR}/full_analysis.png")

"""Inspect Q-Tables
View the learned Q-values as a table (sorted by player total).
"""

def show_q_table(q_table, label, rows=20):
    """Print the top N rows of a Q-table sorted by player total."""
    data = []
    for (state, action), q_val in q_table.items():
        pv, du, soft, dd = state
        data.append({
            'Player Total': pv,
            'Dealer Up':    du,
            'Soft Hand':    soft,
            'Doubled':      dd,
            'Action':       action,
            'Q-Value':      round(q_val, 4)
        })
    df = pd.DataFrame(data).sort_values(['Player Total', 'Dealer Up', 'Action'])
    print(f"{label} Q-Table ({len(df)} total entries)")
    print(df.head(rows).to_string(index=False))
    # Save full Q-table
    df.to_csv(f"{LOG_DIR}/{label.lower().replace(' ', '_')}_q_table.csv", index=False)
    print(f"Full Q-table saved to {LOG_DIR}/")

show_q_table(ql_table, "Q-Learning")
show_q_table(mc_table, "Monte Carlo")

"""Save Files"""

from google.colab import files
import shutil

# Zip the entire experiment_logs folder
shutil.make_archive('experiment_logs', 'zip', 'experiment_logs')

# Download it to computer
files.download('experiment_logs.zip')