coding utf8 references sutton barto 2017 reinforcement learning an introduction mit press cambridge ma deep q learning implementation parameters minepsilon float minimal value for epsilon epsilon float greedy value decay float epsilon decay rate memorylimit int limit of experience replay memory experience replay exploration exploitation run one timestep of the environment sample batch from experience replay construct training data train deep learning model exit from current episode remove old entries from replay memory coding utf 8 references sutton barto 2017 reinforcement learning an introduction mit press cambridge ma deep q learning implementation parameters min_epsilon float minimal value for epsilon epsilon float ε greedy value decay float epsilon decay rate memory_limit int limit of experience replay memory experience replay exploration exploitation run one timestep of the environment sample batch from experience replay construct training data train deep learning model exit from current episode remove old entries from replay memory
import logging
import random

import gym
import numpy as np
from gym import wrappers

np.random.seed(9999)

logger = logging.getLogger()
logger.setLevel(logging.INFO)




class DQN(object):
    def __init__(
        self, n_episodes=500, gamma=0.99, batch_size=32, epsilon=1.0, decay=0.005, min_epsilon=0.1, memory_limit=500
    ):
        

        self.memory_limit = memory_limit
        self.min_epsilon = min_epsilon
        self.gamma = gamma
        self.epsilon = epsilon
        self.n_episodes = n_episodes
        self.batch_size = batch_size
        self.decay = decay

    def init_environment(self, name="CartPole-v0", monitor=False):
        self.env = gym.make(name)
        if monitor:
            self.env = wrappers.Monitor(self.env, name, force=True, video_callable=False)

        self.n_states = self.env.observation_space.shape[0]
        self.n_actions = self.env.action_space.n

        
        self.replay = []

    def init_model(self, model):
        self.model = model(self.n_actions, self.batch_size)

    def train(self, render=False):
        max_reward = 0

        for ep in range(self.n_episodes):
            state = self.env.reset()

            total_reward = 0

            while True:
                if render:
                    self.env.render()

                if np.random.rand() <= self.epsilon:
                    
                    action = np.random.randint(self.n_actions)
                else:
                    
                    action = np.argmax(self.model.predict(state[np.newaxis, :])[0])

                
                new_state, reward, done, _ = self.env.step(action)
                self.replay.append([state, action, reward, new_state, done])

                
                batch_size = min(len(self.replay), self.batch_size)
                batch = random.sample(self.replay, batch_size)

                X = np.zeros((batch_size, self.n_states))
                y = np.zeros((batch_size, self.n_actions))

                states = np.array([b[0] for b in batch])
                new_states = np.array([b[3] for b in batch])

                Q = self.model.predict(states)
                new_Q = self.model.predict(new_states)

                
                for i in range(batch_size):
                    state_r, action_r, reward_r, new_state_r, done_r = batch[i]
                    target = Q[i]

                    if done_r:
                        target[action_r] = reward_r
                    else:
                        target[action_r] = reward_r + self.gamma * np.amax(new_Q[i])

                    X[i, :] = state_r
                    y[i, :] = target

                
                self.model.fit(X, y)

                total_reward += reward
                state = new_state

                if done:
                    
                    break

            
            while len(self.replay) > self.memory_limit:
                self.replay.pop(0)

            self.epsilon = self.min_epsilon + (1.0 - self.min_epsilon) * np.exp(-self.decay * ep)

            max_reward = max(max_reward, total_reward)
            logger.info(
                "Episode: %s, reward %s,  epsilon %s, max reward %s" % (ep, total_reward, self.epsilon, max_reward)
            )
        logging.info("Training finished.")

    def play(self, episodes):
        for i in range(episodes):
            state = self.env.reset()
            total_reward = 0

            while True:
                self.env.render()
                action = np.argmax(self.model.predict(state[np.newaxis, :])[0])
                state, reward, done, _ = self.env.step(action)
                total_reward += reward
                if done:
                    break
            logger.info("Episode: %s, reward %s" % (i, total_reward))
        self.env.close()