用Python玩转赌徒问题：手把手教你实现MDP的两种经典算法（附完整代码）

用Python玩转赌徒问题手把手教你实现MDP的两种经典算法附完整代码马尔科夫决策过程MDP是强化学习的基础框架之一而赌徒问题则是理解MDP的绝佳案例。本文将带你从零开始用Python实现策略迭代和值迭代这两种经典算法并通过可视化分析不同参数下的策略变化。无论你是想巩固理论知识还是希望获得可复用的代码模板这篇文章都能满足你的需求。1. 环境准备与问题建模在开始编码前我们需要明确赌徒问题的数学模型。假设一个赌徒初始有s美元1≤s≤99每次可以选择下注1到min(s,100-s)美元。硬币正面朝上的概率为ph获胜则获得下注金额失败则失去下注金额。游戏在达到100美元或破产时结束。首先安装必要的库pip install numpy matplotlib seaborn定义问题参数GOAL 100 # 目标金额 STATES np.arange(GOAL 1) # 所有可能状态(0到100) ph 0.4 # 硬币正面概率 gamma 1 # 折扣因子状态值函数初始化时只有达到目标状态(100)才有奖励1state_values np.zeros(GOAL 1) state_values[GOAL] 1.02. 策略迭代算法实现策略迭代分为两个交替进行的阶段策略评估和策略改进。我们先来看完整的类实现class PolicyIteration: def __init__(self, goal100, proba_h0.4, theta1e-9, gamma1): self.ph proba_h self.gamma gamma self.goal goal self.theta theta self.states np.arange(goal 1) self.state_values np.zeros(goal 1) self.state_values[goal] 1.0 self.policy np.zeros(goal 1) # 初始策略全0 self.sweeps_history [] # 记录每次迭代的值函数 def policy_evaluation(self): while True: old_values self.state_values.copy() self.sweeps_history.append(old_values) for s in self.states[1:self.goal]: actions np.arange(min(s, self.goal - s)) 1 action_returns [] for a in actions: ret self.ph * (self.gamma * self.state_values[s a]) \ (1 - self.ph) * (self.gamma * self.state_values[s - a]) action_returns.append(ret) # 使用当前策略选择动作 current_a int(self.policy[s]) if current_a 0 and s self.goal: # 初始策略为0需要处理 current_a actions[0] self.policy[s] current_a self.state_values[s] action_returns[actions.tolist().index(current_a)] delta np.abs(self.state_values - old_values).max() if delta self.theta: break def policy_improvement(self): policy_stable True for s in self.states[1:self.goal]: old_a self.policy[s] actions np.arange(min(s, self.goal - s)) 1 action_returns [] for a in actions: ret self.ph * (self.gamma * self.state_values[s a]) \ (1 - self.ph) * (self.gamma * self.state_values[s - a]) action_returns.append(ret) # 选择回报最大的动作 max_a actions[np.argmax(np.round(action_returns, 5))] self.policy[s] max_a if old_a ! max_a: policy_stable False return policy_stable def solve(self): while True: self.policy_evaluation() if self.policy_improvement(): break关键点说明policy_evaluation通过迭代更新状态值函数直到变化小于阈值thetapolicy_improvement根据当前值函数选择最优动作solve方法交替执行上述两个步骤直到策略稳定3. 值迭代算法实现值迭代将策略评估和策略改进合并为一个步骤直接更新最优值函数class ValueIteration: def __init__(self, goal100, proba_h0.4, theta1e-9, gamma1): self.ph proba_h self.gamma gamma self.goal goal self.theta theta self.states np.arange(goal 1) self.state_values np.zeros(goal 1) self.state_values[goal] 1.0 self.policy np.zeros(goal 1) self.sweeps_history [] def value_iteration(self): while True: old_values self.state_values.copy() self.sweeps_history.append(old_values) for s in self.states[1:self.goal]: actions np.arange(min(s, self.goal - s)) 1 action_returns [] for a in actions: ret self.ph * (self.gamma * self.state_values[s a]) \ (1 - self.ph) * (self.gamma * self.state_values[s - a]) action_returns.append(ret) # 直接取最大值作为新状态值 self.state_values[s] np.max(action_returns) delta np.abs(self.state_values - old_values).max() if delta self.theta: break def derive_policy(self): for s in self.states[1:self.goal]: actions np.arange(min(s, self.goal - s)) 1 action_returns [] for a in actions: ret self.ph * (self.gamma * self.state_values[s a]) \ (1 - self.ph) * (self.gamma * self.state_values[s - a]) action_returns.append(ret) # 选择最优动作 self.policy[s] actions[np.argmax(np.round(action_returns, 5))] def solve(self): self.value_iteration() self.derive_policy()与策略迭代的主要区别每次直接更新为最优值取max而不是当前策略下的期望值值收敛后才一次性推导出策略4. 结果分析与可视化实现算法后我们比较ph0.4和ph0.55两种情况下的策略差异def plot_results(ph, title): # 策略迭代 pi PolicyIteration(proba_hph) pi.solve() # 值迭代 vi ValueIteration(proba_hph) vi.solve() plt.figure(figsize(12, 8)) # 绘制策略 plt.subplot(2, 2, 1) plt.step(pi.states, pi.policy, wherepost) plt.title(fPolicy Iteration (ph{ph})) plt.xlabel(Capital) plt.ylabel(Optimal stake) plt.subplot(2, 2, 2) plt.step(vi.states, vi.policy, wherepost) plt.title(fValue Iteration (ph{ph})) plt.xlabel(Capital) plt.ylabel(Optimal stake) # 绘制值函数 plt.subplot(2, 2, 3) plt.plot(pi.states, pi.state_values) plt.title(State Values (PI)) plt.xlabel(Capital) plt.ylabel(Value estimate) plt.subplot(2, 2, 4) plt.plot(vi.states, vi.state_values) plt.title(State Values (VI)) plt.xlabel(Capital) plt.ylabel(Value estimate) plt.tight_layout() plt.show() plot_results(0.4, ph0.4) plot_results(0.55, ph0.55)关键发现当ph0.4劣势赌局时两种算法都建议保守策略只在特定资本时下注较大金额当ph0.55优势赌局时最优策略变得更激进建议更大胆的下注值迭代收敛更快但策略迭代的策略变化过程更平滑5. 算法对比与工程实践在实际应用中两种算法各有优劣特性策略迭代值迭代收敛速度较慢较快每次迭代计算量较大较小中间结果可用性每次迭代都有完整策略只有最终策略实现复杂度较高较低适合场景需要中间策略/策略变化平缓只需最终结果/快速原型开发工程优化建议向量化计算将内部循环改为矩阵运算# 替代原来的for循环 returns ph * values[s actions] (1 - ph) * values[s - actions]并行化使用多进程处理状态更新早期终止检测策略是否早停滞日志记录保存每次迭代变化用于调试常见问题解决振荡问题适当减小学习率或增加theta值收敛慢检查奖励设置和折扣因子内存不足使用稀疏矩阵表示大状态空间# 示例带收敛诊断的改进版值迭代 def value_iteration_enhanced(max_iter1000): for i in range(max_iter): old_values values.copy() for s in states[1:GOAL]: # ... 更新逻辑 ... delta np.abs(values - old_values).max() if delta theta: print(fConverged at iteration {i}) break elif i % 10 0: print(fIter {i}, delta{delta:.4f})通过这个完整的实现案例我们不仅掌握了MDP两种基本算法的编程技巧还深入理解了它们在策略形成上的差异。建议读者尝试修改参数如ph、gamma或奖励函数观察策略如何随之变化这是巩固MDP概念的最佳方式。