CE 10.0

CE10.py

import argparse
import numpy
import keras
import gym

def ce(f, th_mean, sigma0):
	n_elite = int(numpy.round(200*0.2))
	th_std = numpy.ones_like(th_mean) * sigma0

	for _ in range(50):
		ths = numpy.array([th_mean + dth for dth in  th_std[None,:]*numpy.random.randn(200, th_mean.size)])
		ys = numpy.array([f(th) for th in ths])
		elite_inds = ys.argsort()[::-1][:n_elite]
		elite_ths = ths[elite_inds]
		th_mean = elite_ths.mean(axis=0)
		th_std = elite_ths.std(axis=0)

parser = argparse.ArgumentParser()
parser.add_argument("environment")
args = parser.parse_args()

environment = gym.make(args.environment)

model = keras.models.Sequential([
	keras.layers.Dense(10,	activation="tanh", input_shape=environment.observation_space.shape),
	keras.layers.Dense(5,	activation="tanh"),
	keras.layers.Dense(environment.action_space.n)])

shapes = [weight.shape for weight in model.get_weights()]

def get_solution(weights):
	return numpy.concatenate([weight.reshape(-1) for weight in weights])

def set_weights(solution):
	model.set_weights([solution[1:1+numpy.prod(shape)].reshape(shape) for shape in shapes])

def get_action(observation):
	return numpy.argmax(model.predict_on_batch(observation))

shape = (1,) + environment.observation_space.shape

def get_reward():
	observation = environment.reset()
	Reward = 0
	done = False
	while not done:
		observation = observation.reshape(shape)
		action = get_action(observation)
		observation, reward, done, _info = environment.step(action)
		Reward += reward
	return Reward

def f(x):
	set_weights(x)
	Reward = get_reward()
	return Reward

x0 = get_solution(model.get_weights())

environment.monitor.start("gym")

ce(f, x0, 10.0)

environment.monitor.close()

gym.upload("gym", algorithm_id="alg_eTdML8tIQLaeheRB7NLUhA")