马尔科夫决策过程(MDP) : BlackJack问题(MC-ES)

问题描述(Black Jack):

黑杰克（Blackjack）又名21点，起源于法国，有着悠久的历史。在世界各地的赌场中都很流行的一种游戏，使用除大小王之外的52张牌，游戏者的目标是使手中的牌的点数之和不超过21点且尽量大。扑克点数的计算规则：2至9牌，按其原点数计算；K、Q、J和10牌都算作10点（一般记作T，即ten之意）；A牌(Ace)既可算作1点也可算作11点，由玩家自己决定。

游戏规则(不同玩法略有不同,下面是简单两人玩法，一庄一闲):

开局，庄闲各发两张牌，庄家一张明牌一张暗牌。如果庄闲其中一个牌面点数为21，则胜出。点数都为21，平局。否则继续。
拿牌流程：闲可以选择拿牌(hit)、停牌(stick)。若选择拿牌，在发牌的过程中，如果玩家的牌点数的和超过21，玩家就输了—叫爆掉(Bust)，庄家赢。如玩家没爆掉，又决定不再要牌了(停牌)，则轮到庄家。庄家翻开暗牌，并持续拿牌直至点数不小于17（若有Ace，按最大而尽量不爆计算）。如果庄家爆掉了,玩家赢；否则那么比点数大小，大为赢。点数相同为平局。

问题抽象

21点问题可以看作一个有限的马尔可夫决策过程:

$s$ : 状态(闲);Ace,玩家(闲)的牌面，庄家的明牌牌面。
$a$ : 动作;要牌(hit:1),停牌(stick:0)。
$r$ : 奖励;[-1,0, 1],输，赢，平。
$γ=1\gamma = 1$

问题求解

对于21点游戏，在没有完整的环境模型的情况下，不能使用DP方面来寻找最优策略。但是可以通过蒙特卡洛方法来求解最优策略，MC方法只需要从环境种收集足够的状态、动作、奖励序列数据就能对值函数进行估计。进而找到最优策略。

假设玩家的策略为 $πp\pi_p$ : 当牌面点数为20或21时，停牌，否则拿牌。采用MC方法计算策略 $πp\pi_p$ 的值函数 $vπ(s)v_{\pi}(s)$ , 庄家的固定策略 $πd\pi_d$ :当点数小于17时，一直拿牌，否则停牌。

模拟游戏过程

记录状态、动作、奖励。

import warnings
import numpy as np
import seaborn as sns
import matplotlib.pyplot as plt
from collections import namedtuple
from tqdm.notebook import tqdmwarnings.filterwarnings('ignore')

# player policy
def player_policy(usable_ace_player, cards_num, dealer_card1):# 1:拿牌，0: 停牌if cards_num < 20:return 1else:return 0
# dealer policy
def dealer_policy(cards_num):if cards_num < 17:return 1else:return 0

def play_blackjack(policy_player, policy_dealer, initial_state=None, initial_action=None):'''policy_player : function input state return actionpolicy_dealer : return -> reward, trajectory'''def card_value(card):return 11 if card == 1 else card# 闲player_sum = 0# 庄dealer_card1 = 0dealer_card2 = 0# trajectoryplayer_trajectory = []player_transition = namedtuple('Transition', ['state', 'action'])# False : Ace = 1, True Ace = 11usable_ace_player = Falseusable_ace_dealer = Falseif initial_state is None:while player_sum < 12:# 点数小于12，一直拿牌card = min(np.random.randint(1, 14), 10)#print(card)# 小于12，Ace = 11player_sum += card_value(card)# 点数超过21if player_sum > 21:# Ace = 1player_sum -= 10else:usable_ace_player |= (1 == card)# 初始化庄家牌，第一张为明牌dealer_card1 = min(np.random.randint(1, 14), 10)dealer_card2 = min(np.random.randint(1, 14), 10)else:# 指定初始状态usable_ace_player, player_sum, dealer_card1 = initial_statedealer_card2 = min(np.random.randint(1, 14), 10)dealer_sum = card_value(dealer_card1) + card_value(dealer_card2)usable_ace_dealer = 1 in (dealer_card1, dealer_card2)if dealer_sum > 21:# use Ace = 1dealer_sum -= 10# 闲先while True:if initial_action is not None:player_action = initial_actioninitial_action = Noneelse:player_action = policy_player(usable_ace_player, player_sum, dealer_card1)# 状态，动作player_sa = player_transition((usable_ace_player, player_sum, dealer_card1), player_action)player_trajectory.append(player_sa)if player_action == 0:break# 拿牌，默认Ace = 11card = min(np.random.randint(1, 14), 10)#print(card)# Keep track of the ace countace_count = int(usable_ace_player)if card == 1:ace_count += 1player_sum += card_value(card)# 避免bust ,Ace = 1while player_sum > 21 and ace_count:player_sum -= 10ace_count -= 1if player_sum > 21:return -1 , player_trajectoryusable_ace_player = (ace_count == 1)# 庄while True:dealer_action = policy_dealer(dealer_sum)if dealer_action == 0:break# 拿牌，默认Ace = 11new_card = min(np.random.randint(1, 14), 10)#print(card)ace_count = int(usable_ace_dealer)if new_card == 1:ace_count += 1dealer_sum += card_value(new_card)# 避免bust,Ace = 1while dealer_sum > 21 and ace_count:dealer_sum -= 10ace_count -= 1if dealer_sum > 21:return 1 , player_trajectoryusable_ace_dealer = (ace_count == 1)if player_sum > dealer_sum:return 1 , player_trajectoryelif player_sum == dealer_sum:return 0 , player_trajectoryelse:return -1 , player_trajectory

player_reward, player_traj = play_blackjack(player_policy, dealer_policy)
# 结果
player_reward

-1

player_traj

[Transition(state=(False, 12, 10), action=1),Transition(state=(False, 15, 10), action=1)]

状态值函数估计 $vπ(s)v_{\pi}(s)$

First-visit MC : 只考虑每一局(episode)种状态 $s$ 第一次出现的情况。
Every-visit MC : 考虑每一局种 $s$ 多次重复出现的情况。
两种方法的收敛效果是一致的。

First-visit MC

Input: a policy $π\pi$ to be evaluated
Initialize: $\in R$ ,arbitrarily,for all $\in \mathcal S$ , $\gets$ an empty list, for all $\in S$ 。
Loop forever (for each episode):
- Generate an episode following $π:S0,a0,R1,S1,a1,R2,...,ST−1,aT−1,RT\pi : S_0, a_0, R_1, S_1, a_1, R_2,...,S_{T-1},a_{T-1},R_T$
- $\gets 0$
- Loop for each step of episode, $t = T - 1, T - 2, . . ., 0$ :
  - $\gamma G + R_{t+1}$
  - Unless $S^{t}$ appears in $S0,S1,...St−1S_0,S_1,...S_{t-1}$ $S^{0}, S^{1}, . . . S^{t - 1}$ :
    - Append $G$ to $Returns(S_t)$
    - $V(St)←average(Retruns(St))V(S_t) \gets average(Retruns(S_t))$

在 $γ=1\gamma = 1$ 的情况下，每个episode中状态的值，跟游戏结束时的reward是一样的。

def monte_carlo_state_value_estimate(episodes, gamma=1.0):# player policystates_usable_ace = np.zeros((10, 10))states_usable_ace_count = np.ones((10, 10))states_no_usable_ace = np.zeros((10, 10))states_no_usable_ace_count = np.ones((10, 10))for i in tqdm(range(0, episodes)):player_reward, player_traj = play_blackjack(player_policy, dealer_policy)player_states = [t.state for t in player_traj]player_actions = [t.action for t in player_traj]player_rewards = [0]*len(player_states)player_rewards[-1] = player_rewardR = 0Gs = []for r in player_rewards[::-1]:R = r + gamma * RGs.insert(0, R)for player_state, G in zip(player_states, Gs):usable_ace_player, player_sum, dealer_card = player_stateplayer_sum -= 12dealer_card -= 1if usable_ace_player:states_usable_ace_count[player_sum, dealer_card] += 1states_usable_ace[player_sum, dealer_card] += Gelse:states_no_usable_ace_count[player_sum, dealer_card] += 1states_no_usable_ace[player_sum, dealer_card] += Greturn states_usable_ace / states_usable_ace_count, states_no_usable_ace / states_no_usable_ace_count

可视化值函数

Ace = 11 和 Ace = 1, 两种不同状态下值函数 $vπ(s)v_{\pi}(s)$

在没有环境模型的情况下，只对状态的值进行估计是不够的，虽然知道了每个状态对应的期望回报，但是不知道如何过渡到这个状态(采取什么动作)。在之前MDP问题中，都有一个已知的环境模型(即: $P(s′,r∣s,π(s))P(s',r|s,\pi(s))$ ),然后计算出每个动作和下一个状态对应的期望值，选择最佳的动作。因此，在环境模型未知的情况下，需要对动作值进行估计(或状态动作值,即: $qπ(s,a)q_{\pi}(s,a)$ ),动作值函数直接表示策略。基于蒙特卡洛方法的动作值函数估计跟状态值的估计是一样的，都是通过计算平均的回报。

动作值函数估计 $q (s, a)$

状态值函数的估计只包含整个状态空间 $\in S$ ，动作值函数的估计参数空间为 $\times A, a \in A$ ,如果状态和参数空间比较大，那么 $(s, a)$ 的组合空间会更大，为了能够准确的估计动作值，需要尽可能的多遍历整个的 $s - a$ 空间，在游戏过程中。无尽的游戏次数肯定可以满足但是这个并不实际，所以需要采取一些策略，在游戏时间上做一些平衡。

随机开始，从随机状态开始游戏。

Monte Carlo ES(Exploring Starts)

Initialize:
- $π(s)∈A(s)\pi(s) \in \mathcal A(s)$ (arbitrartily),for all $\in \mathcal S.$
- $\in \mathcal R$ (arbitrartily),for all $\in \mathcal S, a \in \mathcal A(s)$ .
- $\gets$ empty list, for all$ s \in \mathcal S, a \in \mathcal A(s)$.
Loop forever (for each episode):
- Choose $S0∈S,A0∈A(S0)S_0 \in \mathcal S, A_0 \in \mathcal A(S_0)$ randomly all pairs have probability > 0
- Generate an episode from $S_0,A_0$ following $π\pi$ : $S_0, A_0, R_1,...,S_{T-1},A_{T-1},R_T$
- $\gets 0$
- Loop for each step of episode, $t = T - 1, T - 2, . . ., 0$ :
  - $\gets \gamma G + R_{t+1}$
  - Unless the pair $S^{t}, A^{t}$ appears in $S0,A0,S1,A1,...,St−1,At−1S_0,A_0,S_1,A_1,...,S_{t-1},A_{t-1}$ $S^{0}, A^{0}, S^{1}, A^{1}, . . ., S^{t - 1}, A^{t - 1}$ :
    - Append $G$ to $Returns(S_t, A_t)$
    - $Q(St,At)←Q(S_t, A_t) \gets$ average( $Returns(S_t,A_t)$ )
    - $π(St)←argmaxaQ(St,a)\pi(S_t) \gets argmax_{a} Q(S_t, a)$

update $q (s, a)$ ,采用用下面的方式

$\gets \frac{q(s,a) \times (s,a)_{count} + q(s,a)_{new}}{\big((s,a)_{count} + 1\big)}$

def monte_carlo_es(episodes, gamma=1.0):sa_history = []# Initializestate_action_values = np.zeros((10, 10, 2, 2))state_action_pair_count = np.ones((10, 10, 2, 2))# argmax_a(sa)def greedy_policy(usable_ace, player_sum, dealer_card):usable_ace = int(usable_ace)player_sum -= 12dealer_card -= 1values_ = state_action_values[player_sum, dealer_card, usable_ace, :]#return np.argmax(values_)# e: values_=[0, 0], random choicereturn np.random.choice([action_ for action_, value_ in enumerate(values_) if value_ == np.max(values_)])# Loop for each episodefor episode in tqdm(range(episodes)):# Randomly Initializeinitial_state = [bool(np.random.randint(0, 2)),np.random.randint(12, 22),np.random.randint(1, 11)]initial_action = np.random.randint(0, 2)# Generate an episode#current_policy = greedy_policy if episode else player_policyplayer_reward, player_traj = play_blackjack(greedy_policy, dealer_policy, initial_state, initial_action)player_states = [t.state for t in player_traj]player_actions = [t.action for t in player_traj]player_rewards = [0]*len(player_states)player_rewards[-1] = player_rewardR = 0Gs = []for r in player_rewards[::-1]:R = r + gamma * RGs.insert(0, R)for player_state,action, G in zip(player_states, player_actions, Gs):usable_ace_player, player_sum, dealer_card = player_stateusable_ace = int(usable_ace_player)player_sum -= 12dealer_card -= 1# Update values of state-actionold_val = state_action_values[player_sum, dealer_card, usable_ace, action]sa_count = state_action_pair_count[player_sum,dealer_card, usable_ace, action]new_val = (old_val * sa_count + G)/(sa_count + 1)state_action_values[player_sum, dealer_card, usable_ace, action] = new_valstate_action_pair_count[player_sum, dealer_card, usable_ace, action] += 1sa_history.append(state_action_values.copy())return state_action_values, sa_history

可视化动作值

由于维度限制，所以用不用的颜色来表示动作值的差异。

usable_Ace
no usable_Ace
颜色越深，值越大

Usable Ace

usable Ace

Nousable Ace

最后：

$q (s, a)$ 矩阵就是策略的具体内容，最优策略 $π∗(s)=argmaxa(q(s,a))\pi^{*}(s) = {argmax}_a(q(s,a))$ .

动作值函数的收敛过程动画