# How to design an observation(state) space for a simple Rock-Paper-Scissor game?

For weeks I've been working with this toy game of Rock-Paper-Scissor. I want to use a PPO agent learn to beat a computer opponent whose logic is defined as the code bellow.

For short, this computer opponent, named abbey, uses a strategy that tracks all two consecutive plays of the agent, and gives the opposite play of the most likely guess of the agent's next play, according to the agent's last play.

I design the agent(using gym env) to have an internal state, keeping track of all counts of its two consecutive plays, in a 3x3 matrix. And then I normalized each row of the matrix to be an observation of the agent, representing the probabilities of the second play given the previous one. So the agent will get the same knowledge as what abbey knows.

Then I copied an PPO network algorithm from some RL book, which works well with CartPole. Then I did some minor changes which are commented in the code bellow.

But the algorithm does not converge even a little, and abbey always wins the agent about 60% of the time from first run to the last.

I doubt the state and observation space I designed is the reason why it does not converge. All I get is that the agent maybe should find something from the histories of its own successes and fails, and find its way out.

Can you give me some advice for the designing of a state space? Thank you very much.

### define a Rock-Paper-Scissor opponent

abbey_state = []
play_order=[{
"RR": 0,
"RP": 0,
"RS": 0,
"PR": 0,
"PP": 0,
"PS": 0,
"SR": 0,
"SP": 0,
"SS": 0,
}]
def abbey(prev_opponent_play,
re_init=False):
if not prev_opponent_play:
prev_opponent_play = 'R'
global abbey_state, play_order
if re_init:
abbey_state = []
play_order=[{
"RR": 0,
"RP": 0,
"RS": 0,
"PR": 0,
"PP": 0,
"PS": 0,
"SR": 0,
"SP": 0,
"SS": 0,
}]
abbey_state.append(prev_opponent_play)
last_two = "".join(abbey_state[-2:])
if len(last_two) == 2:
play_order[last_two] += 1
potential_plays = [
prev_opponent_play + "R",
prev_opponent_play + "P",
prev_opponent_play + "S",
]
sub_order = {
k: play_order[k]
for k in potential_plays if k in play_order
}
prediction = max(sub_order, key=sub_order.get)[-1:]
ideal_response = {'P': 'S', 'R': 'P', 'S': 'R'}
return ideal_response[prediction]

### define the gym env
import gym
from gym import spaces
from collections import defaultdict
import numpy as np

ACTIONS = ["R", "P", "S"]
games = 1000

class RockPaperScissorsEnv(gym.Env):

def __init__(self):
super(RockPaperScissorsEnv, self).__init__()
self.action_space = spaces.Discrete(3)
self.observation_space = spaces.Box(low=0.0, high=1.0,
shape=(3,3), dtype=float)
self.reset()

def step(self, actions):
assert actions == 0 or actions == 1 or actions == 2
opponent_play = self.opponent_play()

self.prev_plays[self.prev_actions * 3 + actions] += 1
reward = self.calc_reward(actions, opponent_play)
terminal = False

self.calc_state(self.timestep, opponent_play, actions)
self.prev_actions = actions
self.prev_opponent_play = opponent_play
self.timestep += 1
return self.get_ob(), reward, terminal, None

def reset(self):
self.opponent = abbey
self.timestep = 0
self.prev_opponent_play = 0
self.prev_actions = 0
self.prev_plays = defaultdict(int)
self.init_state = np.zeros((3,3), dtype=int)
# the internal state
self.state = np.copy(self.init_state)
self.results = {"win": 0, "lose": 0, "tie": 0}
return self.get_ob()

def render(self, mode='human'):
pass

def close (self):
pass

def calc_reward(self, actions, play):
if self.timestep % games == games - 1:
pass
if actions == play:
self.results['tie'] += 1
return 0
elif actions == 0 and play == 1:
self.results['lose'] += 1
return -0.3
elif actions == 1 and play == 2:
self.results['lose'] += 1
return -0.3
elif actions == 2 and play == 0:
self.results['lose'] += 1
return -0.3
elif (actions == 1 and play == 0) or (actions == 2 and play == 1) or (actions == 0 and play == 2):
self.results['win'] += 1
return 0.3
else:
raise NotImplementedError('calc_reward something get wrong')

def opponent_play(self):
re_init = (self.timestep == 0)
opp_play = self.opponent(ACTIONS[self.prev_actions], re_init=re_init)
return ACTIONS.index(opp_play)

def calc_state(self, timestep, opponent_play, actions):
self.state[self.prev_actions][actions] += 1

def get_ob(self):
'''return observations'''
state0 = self.state
sum0 = state0.sum()
state1 = self.state
sum1 = state1.sum()
state2 = self.state
sum2 = state2.sum()
init = np.ones(3, dtype=float) / 3.0
ob = np.array([
state0 / sum0 if sum0 else init,
state1 / sum1 if sum1 else init,
state2 / sum2 if sum2 else init,
])
# print(ob)
return ob

### Learning Algo copied from some book

import  matplotlib
from    matplotlib import pyplot as plt
matplotlib.rcParams['font.size'] = 18
matplotlib.rcParams['figure.titlesize'] = 18
matplotlib.rcParams['figure.figsize'] = [9, 7]
matplotlib.rcParams['axes.unicode_minus']=False

plt.figure()

import  gym,os
import  numpy as np
import  tensorflow as tf
from    tensorflow import keras
from    tensorflow.keras import layers,optimizers,losses
from    collections import namedtuple
env = RockPaperScissorsEnv()
env.seed(2222)
tf.random.set_seed(2222)
np.random.seed(2222)
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
assert tf.__version__.startswith('2.')

gamma = 0.98
epsilon = 0.2
batch_size = 32

Transition = namedtuple('Transition', ['state', 'action', 'a_log_prob', 'reward', 'next_state'])

class Actor(keras.Model):
def __init__(self):
super(Actor, self).__init__()
self.fc1 = layers.Dense(18, kernel_initializer='he_normal') # I changed 100 to 18
self.fc2 = layers.Dense(3, kernel_initializer='he_normal') # I changed 4 to 3

def call(self, inputs):
x = tf.nn.relu(self.fc1(inputs))
x = self.fc2(x)
x = tf.nn.softmax(x, axis=1)
return x

class Critic(keras.Model):
def __init__(self):
super(Critic, self).__init__()
self.fc1 = layers.Dense(18, kernel_initializer='he_normal') # I changed 100 to 18
self.fc2 = layers.Dense(1, kernel_initializer='he_normal')

def call(self, inputs):
x = tf.nn.relu(self.fc1(inputs))
x = self.fc2(x)
return x

class PPO():
def __init__(self):
super(PPO, self).__init__()
self.actor = Actor()
self.critic = Critic()
self.buffer = []

def select_action(self, s):
s = tf.constant(s, dtype=tf.float32)
# s = tf.expand_dims(s, 0)   # I removed this line, otherwise we will get a (1,3,3) tensor and later we will get an error
prob = self.actor(s)
a = tf.random.categorical(tf.math.log(prob), 1)
a = int(a)
return a, float(prob[a])

def get_value(self, s):
s = tf.constant(s, dtype=tf.float32)
s = tf.expand_dims(s, axis=0)
v = self.critic(s)
return float(v)

def store_transition(self, transition):
self.buffer.append(transition)

def optimize(self):
state = tf.constant([t.state for t in self.buffer], dtype=tf.float32)
action = tf.constant([t.action for t in self.buffer], dtype=tf.int32)
action = tf.reshape(action,[-1,1])
reward = [t.reward for t in self.buffer]
old_action_log_prob = tf.constant([t.a_log_prob for t in self.buffer], dtype=tf.float32)
old_action_log_prob = tf.reshape(old_action_log_prob, [-1,1])

R = 0
Rs = []
for r in reward[::-1]:
R = r + gamma * R
Rs.insert(0, R)
Rs = tf.constant(Rs, dtype=tf.float32)

for _ in range(round(10*len(self.buffer)/batch_size)):

index = np.random.choice(np.arange(len(self.buffer)), batch_size, replace=False)

v_target = tf.expand_dims(tf.gather(Rs, index, axis=0), axis=1)

v = self.critic(tf.gather(state, index, axis=0))
delta = v_target - v
a = tf.gather(action, index, axis=0)
pi = self.actor(tf.gather(state, index, axis=0))
indices = tf.expand_dims(tf.range(a.shape), axis=1)
indices = tf.concat([indices, a], axis=1)
pi_a = tf.gather_nd(pi, indices)
pi_a = tf.expand_dims(pi_a, axis=1)
# Importance Sampling
ratio = (pi_a / tf.gather(old_action_log_prob, index, axis=0))
surr2 = tf.clip_by_value(ratio, 1 - epsilon, 1 + epsilon) * advantage
policy_loss = -tf.reduce_mean(tf.minimum(surr1, surr2))
value_loss = losses.MSE(v_target, v)

self.buffer = []

def main():
agent = PPO()
returns = []
total = 0
for i_epoch in range(500):
state = env.reset()
for t in range(games):
action, action_prob = agent.select_action(state)
if t == 999:
print(action, action_prob)
next_state, reward, done, _ = env.step(action)
# print(next_state, reward, done, action)
trans = Transition(state, action, action_prob, reward, next_state)
agent.store_transition(trans)
state = next_state
total += reward
if done:
if len(agent.buffer) >= batch_size:
agent.optimize()
break
print(env.results)

if i_epoch % 20 == 0:
returns.append(total/20)
total = 0
print(i_epoch, returns[-1])

print(np.array(returns))
plt.figure()
plt.plot(np.arange(len(returns))*20, np.array(returns))
plt.plot(np.arange(len(returns))*20, np.array(returns), 's')
plt.xlabel('epochs')
plt.ylabel('total return')
plt.savefig('ppo-tf.svg')

if __name__ == '__main__':
main()
print("end")

$$$$
`