需要源码请点赞关注收藏后评论区留下QQ~~~
一、Sarsa算法简介
Sarsa算法每次更新都需要获取五元组（S,A,R,S',A'）这也是该算法称为Sarsa的原因，每当从非终止状态进行一次转移后，就进行一次更新，但需要注意的是，动作A是情节中实际发生的动作。在更新（S,A）的动作值函数Q（S,A）时，Agent并不实际执行状态S'下的动作A‘。由于采用了贪心策略，Sarsa算法在各时间步都隐式地进行了策略改进，像这种在每个样本更新后都进行策略改进的策略迭代算法，也成为完全乐观策略算法
估算最优策略的Sarsa算法步骤如下
二、风险投资问题实战
在进行投资时，预期收益是一个非常重要的参考指标，现在越来越多的人接收概率的观点，但是收益为正的投资也未必一定理性。
假设一种风险投资，当前本金为S，下一个单位时间有0.5的概率变为原有资产的0.9倍，0.5的概率变为原有资产的1.11倍。经过一个时间单位后预期收益率变为百分之0.5，但是在实际情况中，进行2n各时间步后连续投资预期收益非常低，当N趋近于无穷时，该投资会血本无归，为了使投资更加合理，利用Sarsa算法给出在给定本金情况下的投资方案
该问题的MDP模型如下
1：状态空间
状态为当前资产的数目，因此状态空间为连续的实数空间
2：动作空间
该问题中一共有两个动作，分别为投资和不投资。用0表示不投资，用1表示投资
3：立即奖赏
在这个问题中 如果不投资则立即奖赏为0，如果进行投资则分为四种情况
资产增长且结果大于等于本金 +1奖赏
资产减少 且投资结果小于等于本金 -1奖赏
资产增加 且投资结果小于本金 +0.5奖赏
资产减少 且投资结果大于本金 -0.5奖赏
使用Sarsa算法解决风险投资问题 首先设置本金为10 将动作值函数设置为50×2的数组，且初值为0，初始学习率为0.05，使用贪心策略
当迭代到20000个情节时 迭代已经收敛 导出的Q值迭代过程表如下
结论：当资金小于本金时（10）时，不进行投资，当现有资金大于等于本金(10)时，可以进行投资
部分代码如下
import numpy as np
from invest import InvestEnv
np.random.seed(1)
env = InvestEnv()
def trans_q(Q): # 输出保留3位小数
new_q = []
new_q = [round(x, 3) for x in Q]
return new_q
def Attenuation(epsilon, alpha, episode_sum, episode): # epsilon和alpha衰减函数
epsilon = (float(episode_sum) - float(episode)) / float(episode_sum) * epsilon
alpha = (float(episode_sum) - float(episode)) / float(episode_sum) * alpha
return epsilon, alpha
# 输出函数
def print_ff(list_q, Q, episode_i, epsilon_k, alpha_k):
list_s = range(0,50)
for em in list_q:
if em == episode_i:
print("*******************************情节数:%s*******************************" % (str(em)))
for state in list_s:
print("Q(%d,*)" % (state) + str(trans_q(Q[state])))
prob = [epsilon_k / 2.0, epsilon_k / 2.0]
max_a = np.argmax(Q[state])
prob[max_a] = 1 - (epsilon_k / 2.0)
print('概率值' + str(trans_q(prob)))
print("epsilon_k: {}".format(epsilon_k))
print("alpha_k:{}".format(alpha_k))
# 输出单步计算过程
def print_dd(s, a, R, next_s, next_a, print_len, episode_i, Q, e_k, a_k, P, P_next):
if s == 6 and a == 1:
print("*********************************单步的计算过程************************************")
print(6, 1)
print("alpha:" + str(a_k))
print("epsilon:" + str(e_k))
print("Q_state: {} Q_next: {}".format(Q[s], Q[next_s]))
print("Q[{},{}]: {}".format(s, a, Q[s, a]))
print("Q[{},{}]: {}".format(next_s, next_a, Q[next_s, next_a]))
print("update:" + str(Q[s, a] + a_k * (R + 0.8 * Q[next_s, next_a] - Q[s, a])))
# print(p)
print("************************************************************************************")
def policy_epsilon_greedy(env, s, Q, epsilon):
Q_s = Q[s]
if np.random.rand() < epsilon:
a = np.random.choice(env.action_space)
else:
a = np.argmax(Q_s)
return a
def Sarsa(env, episode_num, alpha, gamma, epsilon):
Q = np.zeros((env.state_space, env.action_space))
epsilon = epsilon
count = 0
list_q = [0,1,2,3,4,9998,9999,10000,10001,10002,19996,19997,19998,19999,20000]
for episode_i in range(episode_num):
env.reset()
S = env.state
epsilon_k, alpha_k = Attenuation(epsilon, alpha, episode_num, episode_i)
A = policy_epsilon_greedy(env, S, Q, epsilon_k)
print_ff(list_q, Q, episode_i, epsilon_k, alpha_k)
if episode_i == 10000:
print("e_k:" + str(epsilon_k) + "a_k" + str(alpha_k))
done = False
P_S = env.getState()
for i in range(1000):
next_S, R = env.step(A)
P_S_next = env.getState()
if next_S > 49:
Q[S, A] = Q[S, A] + alpha_k * (R + gamma * 0.0 - Q[S, A])
break
next_A = policy_epsilon_greedy(env, next_S, Q, epsilon_k)
# 输出某一个
if episode_i in [9999, 10000]:
count += 1
print(count)
return Q
Q = Sarsa(env, 20001, 0.05, 0.8, 0.5)创作不易 觉得有帮助请点赞关注收藏~~~
