-
Notifications
You must be signed in to change notification settings - Fork 103
/
grid_game_with_average_policy.py
82 lines (73 loc) · 2.47 KB
/
grid_game_with_average_policy.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# author : zlq16
# date : 2018/4/9
'''
这里面存在几个问题,按照书作者的意思是
作用的动作如果不可以移动的话,那么
用于迭代的 v(s') = v(s)
再用于计算 v(s) = Σπ*(r + γ*v(s'))
我感觉这个是不对
如果没有移动应该就没有 r 和 折扣这一说
但是这里但是姑且用这么书中的方式
之后的策略迭代和值迭代都是修改过的
'''
import pandas as pd
import numpy as np
class GridMDP:
def __init__(self, **kwargs):
for k, v in kwargs.items():
setattr(self, k, v)
self.__action_dir = pd.Series(
data=[np.array((-1, 0)),
np.array((1, 0)),
np.array((0, -1)),
np.array((0, 1))],
index=self.action_space)
self.terminal_states = [(0, 0), (3, 3)]
def transform(self, state, action):
dir = self.__action_dir[action]
state_ = np.array(state) + dir
if (state_ >= 0).all() and (state_ < 4).all():
state_ = tuple(state_)
else:
state_ = state
return state_
def average_policy(mdp, v_s, policy):
state_space = mdp.state_space
action_space = mdp.action_space
reward = mdp.reward
gamma = mdp.gamma
while True:
print(v_s)
v_s_ = v_s.copy()
for state in state_space:
v_s_a = pd.Series()
for action in action_space:
state_ = mdp.transform(state,action)
if state_ in mdp.terminal_states:
v_s_a[action] = 0
elif state_ != state:
v_s_a[action] = v_s_[state_]
else:
v_s_a[action] = v_s_[state]
v_s[state] = sum([policy[action] * (reward + gamma * v_s_a[action]) for action in action_space])
if (np.abs(v_s_ - v_s) < 1e-8).all():
break
return v_s
def main():
state_space = [(i, j) for i in range(4) for j in range(4)]
state_space.remove((0, 0))
state_space.remove((3, 3))
mdp = GridMDP(
state_space=state_space,
action_space=["n", "s", "w", "e"],
reward=-1,
gamma=1)
v_s = pd.Series(np.zeros((len(state_space))),index=state_space)
policy = pd.Series(data=0.25 * np.ones(shape=(4)), index=mdp.action_space)
v_s = average_policy(mdp, v_s, policy)
print("convergence valuation of __state is:")
print(v_s)
if __name__ == '__main__':
main()