Reinforcement Learning Basics

import numpy as np
import random

# Simple 4x4 Grid World
# S = Start (0,0), G = Goal (3,3)
# Agent can move: 0=Up, 1=Right, 2=Down, 3=Left

class GridWorld:
    def __init__(self, size=4):
        self.size = size
        self.state = (0, 0)
        self.goal = (size-1, size-1)
    
    def reset(self):
        self.state = (0, 0)
        return self.state
    
    def step(self, action):
        x, y = self.state
        if action == 0 and x > 0: x -= 1           # Up
        elif action == 1 and y < self.size-1: y += 1  # Right
        elif action == 2 and x < self.size-1: x += 1  # Down
        elif action == 3 and y > 0: y -= 1           # Left
        
        self.state = (x, y)
        
        if self.state == self.goal:
            return self.state, 10.0, True   # Big reward for reaching goal
        return self.state, -0.1, False       # Small penalty for each step

# Q-Learning
env = GridWorld(4)
Q = {}  # Q-table: {state: [Q-values for each action]}
alpha = 0.1    # Learning rate
gamma = 0.95   # Discount factor
epsilon = 1.0  # Exploration rate
epsilon_decay = 0.995
episodes = 1000

for ep in range(episodes):
    state = env.reset()
    if state not in Q:
        Q[state] = [0.0] * 4
    
    total_reward = 0
    for step in range(100):  # Max steps per episode
        # Epsilon-greedy action selection
        if random.random() < epsilon:
            action = random.randint(0, 3)
        else:
            action = np.argmax(Q[state])
        
        next_state, reward, done = env.step(action)
        total_reward += reward
        
        if next_state not in Q:
            Q[next_state] = [0.0] * 4
        
        # Q-Learning update (Bellman equation)
        best_next = max(Q[next_state])
        Q[state][action] += alpha * (reward + gamma * best_next - Q[state][action])
        
        state = next_state
        if done:
            break
    
    epsilon = max(0.01, epsilon * epsilon_decay)
    
    if ep % 200 == 0:
        print(f"Episode {ep:4d} | Reward: {total_reward:7.2f} | Epsilon: {epsilon:.3f}")

# Show learned policy
print("\nLearned Policy (best action per state):")
actions = ["Up", "Right", "Down", "Left"]
for x in range(4):
    for y in range(4):
        if (x, y) in Q:
            best = actions[np.argmax(Q[(x, y)])]
            print(f"  ({x},{y}): {best:5s}", end="")
    print()

import gymnasium as gym
import numpy as np
import random

# Create CartPole environment
env = gym.make("CartPole-v1")

# Discretize continuous state space into bins
def discretize_state(state):
    bins = [
        np.linspace(-2.4, 2.4, 10),    # Cart position
        np.linspace(-3.0, 3.0, 10),    # Cart velocity
        np.linspace(-0.21, 0.21, 10),  # Pole angle
        np.linspace(-3.0, 3.0, 10),    # Pole angular velocity
    ]
    discrete = []
    for i, val in enumerate(state):
        discrete.append(np.digitize(val, bins[i]))
    return tuple(discrete)

# Q-Learning for CartPole
Q = {}
alpha = 0.1
gamma = 0.99
epsilon = 1.0
epsilon_decay = 0.9995
episodes = 2000
reward_history = []

for ep in range(episodes):
    state, _ = env.reset()
    state = discretize_state(state)
    if state not in Q:
        Q[state] = [0.0, 0.0]  # 2 actions: left, right
    
    total_reward = 0
    done = False
    
    while not done:
        # Epsilon-greedy
        if random.random() < epsilon:
            action = env.action_space.sample()
        else:
            action = np.argmax(Q[state])
        
        next_raw, reward, terminated, truncated, _ = env.step(action)
        done = terminated or truncated
        next_state = discretize_state(next_raw)
        total_reward += reward
        
        if next_state not in Q:
            Q[next_state] = [0.0, 0.0]
        
        # Q-Learning update
        best_next = max(Q[next_state]) if not done else 0
        Q[state][action] += alpha * (reward + gamma * best_next - Q[state][action])
        
        state = next_state
    
    epsilon = max(0.01, epsilon * epsilon_decay)
    reward_history.append(total_reward)
    
    if ep % 400 == 0:
        avg = np.mean(reward_history[-100:])
        print(f"Episode {ep:4d} | Avg Reward (100): {avg:7.1f} | Epsilon: {epsilon:.3f} | Q-states: {len(Q)}")

env.close()
final_avg = np.mean(reward_history[-100:])
print(f"\nFinal average reward (last 100): {final_avg:.1f}")
print(f"Solved (>= 475)? {final_avg >= 475}")

import gymnasium as gym
import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
from collections import deque
import random

# DQN Neural Network
class DQN(nn.Module):
    def __init__(self, state_dim, action_dim):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(state_dim, 128),
            nn.ReLU(),
            nn.Linear(128, 128),
            nn.ReLU(),
            nn.Linear(128, action_dim)
        )
    
    def forward(self, x):
        return self.net(x)

# Experience Replay Buffer
class ReplayBuffer:
    def __init__(self, capacity=10000):
        self.buffer = deque(maxlen=capacity)
    
    def push(self, state, action, reward, next_state, done):
        self.buffer.append((state, action, reward, next_state, done))
    
    def sample(self, batch_size):
        batch = random.sample(self.buffer, batch_size)
        states, actions, rewards, next_states, dones = zip(*batch)
        return (np.array(states), np.array(actions), np.array(rewards),
                np.array(next_states), np.array(dones))
    
    def __len__(self):
        return len(self.buffer)

# Training setup
env = gym.make("CartPole-v1")
state_dim = env.observation_space.shape[0]  # 4
action_dim = env.action_space.n              # 2

q_net = DQN(state_dim, action_dim)
target_net = DQN(state_dim, action_dim)
target_net.load_state_dict(q_net.state_dict())  # Copy weights

optimizer = optim.Adam(q_net.parameters(), lr=1e-3)
buffer = ReplayBuffer(10000)

# Hyperparameters
gamma = 0.99
epsilon = 1.0
epsilon_min = 0.01
epsilon_decay = 0.995
batch_size = 64
target_update_freq = 10
episodes = 500

for ep in range(episodes):
    state, _ = env.reset()
    total_reward = 0
    done = False
    
    while not done:
        # Epsilon-greedy action
        if random.random() < epsilon:
            action = env.action_space.sample()
        else:
            with torch.no_grad():
                q_values = q_net(torch.FloatTensor(state))
                action = q_values.argmax().item()
        
        next_state, reward, terminated, truncated, _ = env.step(action)
        done = terminated or truncated
        buffer.push(state, action, reward, next_state, done)
        total_reward += reward
        state = next_state
        
        # Train on batch from replay buffer
        if len(buffer) >= batch_size:
            s, a, r, ns, d = buffer.sample(batch_size)
            s = torch.FloatTensor(s)
            a = torch.LongTensor(a)
            r = torch.FloatTensor(r)
            ns = torch.FloatTensor(ns)
            d = torch.FloatTensor(d)
            
            # Current Q-values
            current_q = q_net(s).gather(1, a.unsqueeze(1)).squeeze(1)
            
            # Target Q-values (from target network)
            with torch.no_grad():
                next_q = target_net(ns).max(1)[0]
                target_q = r + gamma * next_q * (1 - d)
            
            # Update Q-network
            loss = nn.MSELoss()(current_q, target_q)
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
    
    epsilon = max(epsilon_min, epsilon * epsilon_decay)
    
    # Update target network periodically
    if ep % target_update_freq == 0:
        target_net.load_state_dict(q_net.state_dict())
    
    if ep % 100 == 0:
        print(f"Episode {ep:3d} | Reward: {total_reward:6.0f} | Epsilon: {epsilon:.3f}")

env.close()
print(f"\nDQN parameters: {sum(p.numel() for p in q_net.parameters()):,}")

import numpy as np

def linear_decay(episode, total_episodes, start=1.0, end=0.01):
    return max(end, start - (start - end) * episode / total_episodes)

def exponential_decay(episode, start=1.0, decay_rate=0.995, min_epsilon=0.01):
    return max(min_epsilon, start * (decay_rate ** episode))

def step_decay(episode, start=1.0, drop_rate=0.5, drop_every=200, min_epsilon=0.01):
    return max(min_epsilon, start * (drop_rate ** (episode // drop_every)))

episodes = 1000
print(f"{'Episode':>8} | {'Linear':>8} | {'Exponential':>12} | {'Step':>8}")
print("-" * 48)

for ep in [0, 100, 200, 400, 600, 800, 999]:
    lin = linear_decay(ep, episodes)
    exp = exponential_decay(ep)
    stp = step_decay(ep)
    print(f"{ep:8d} | {lin:8.4f} | {exp:12.4f} | {stp:8.4f}")

print("\nLinear: Steady decrease. Simple and predictable.")
print("Exponential: Fast initial decay, slow later. Most popular.")
print("Step: Sudden drops at fixed intervals. Good for staged learning.")

import gymnasium as gym
import numpy as np

# Explore the CartPole environment
env = gym.make("CartPole-v1")

print("=== CartPole-v1 Environment ===")
print(f"Observation space: {env.observation_space}")
print(f"Observation shape: {env.observation_space.shape}")
print(f"Observation low:  {env.observation_space.low}")
print(f"Observation high: {env.observation_space.high}")
print(f"Action space: {env.action_space}")
print(f"Number of actions: {env.action_space.n}")
print(f"Actions: 0=Push Left, 1=Push Right")
print()

# Run one episode with random actions
state, info = env.reset(seed=42)
print("=== Random Episode ===")
print(f"Initial state: {np.round(state, 3)}")
print(f"State meaning: [cart_pos, cart_vel, pole_angle, pole_angular_vel]")

total_reward = 0
steps = 0
done = False

while not done:
    action = env.action_space.sample()  # Random action
    state, reward, terminated, truncated, info = env.step(action)
    total_reward += reward
    steps += 1
    done = terminated or truncated
    
    if steps <= 3 or done:
        print(f"  Step {steps:3d}: action={action}, reward={reward}, "
              f"state={np.round(state, 3)}, done={done}")
    elif steps == 4:
        print(f"  ... (continuing) ...")

print(f"\nEpisode ended after {steps} steps")
print(f"Total reward: {total_reward}")
print(f"Why it ended: {'Pole fell' if terminated else 'Max steps reached'}")

env.close()

import numpy as np

def compute_returns(rewards, gamma):
    """Compute discounted returns for each timestep."""
    returns = []
    G = 0
    for r in reversed(rewards):
        G = r + gamma * G
        returns.insert(0, G)
    return returns

# Example: 5-step episode with constant reward of 1.0
rewards = [1.0, 1.0, 1.0, 1.0, 1.0]

print("Effect of Discount Factor (gamma):")
print(f"Rewards per step: {rewards}")
print()

for gamma in [0.0, 0.5, 0.9, 0.99, 1.0]:
    returns = compute_returns(rewards, gamma)
    print(f"gamma={gamma:.2f}: Returns = {[round(r, 3) for r in returns]}")

print()
print("gamma=0.0: Agent is myopic -- only cares about immediate reward")
print("gamma=0.5: Agent moderately values future rewards")
print("gamma=0.9: Agent significantly values future rewards")
print("gamma=0.99: Agent nearly equally values all future rewards")
print("gamma=1.0: Agent values all rewards equally (may diverge in infinite tasks)")
print()

# Reward shaping example
print("\nReward Shaping Example (reach goal faster):")
print("Sparse reward:  [0, 0, 0, 0, 10]  -- only reward at goal")
print("Shaped reward:  [-1, -1, -1, -1, 10]  -- penalty per step + goal reward")
print("Distance-based: [1, 2, 3, 4, 10]  -- reward increases near goal")

for name, rewards in [("Sparse", [0,0,0,0,10]), ("Step penalty", [-1,-1,-1,-1,10]), ("Distance", [1,2,3,4,10])]:
    ret = compute_returns(rewards, 0.9)
    print(f"  {name:12s}: G_0 = {ret[0]:.2f}")