不完全信息博弈：Khun牌与CFR及其简单变体

				
					import random
from collections import defaultdict

import kuhn
import time

import pickle

# 信息集类，用于表示玩家在游戏中的特定状态（包括动作历史和手牌）
class InformationSet:
    def __init__(self, action_history, hand_card):
        # 存储动作历史
        self.action_history = action_history
        # 存储玩家手牌
        self.hand_card = hand_card

    # 重写哈希函数，根据动作历史（转换为元组）和手牌计算哈希值
    def __hash__(self):
        return hash((tuple(self.action_history.raw), self.hand_card))

    # 重写相等比较函数，比较两个信息集的动作历史和手牌是否相等
    def __eq__(self, other):
        return (self.action_history.raw, self.hand_card) == (other.action_history.raw, other.hand_card)

    # 重写对象的字符串表示形式，便于打印和调试
    def __repr__(self):
        return f"{self.hand_card} {self.action_history.raw} "


# CFR（Counterfactual Regret Minimization）算法中的节点类
class CfrNode:
    def __init__(self, epsilon):
        # 累计遗憾值列表，用于每个可能的动作（这里是库恩扑克中的过牌和下注）
        self.cum_regrets = [0.0] * kuhn.Action.num()
        # 累计策略列表，用于每个可能的动作
        self.cum_strategy = [0.0] * kuhn.Action.num()
        # 探索率，用于平衡探索新策略和利用已有策略
        self.epsilon = epsilon

    # 根据给定的概率分布采样一个动作索引
    def sample(self, probs):
        return random.choices(range(len(probs)), weights=probs, k=1)[0]

    # 根据累计遗憾值选择一个动作 这里没有被运行？
    def get_action(self):
        # 如果所有累积遗憾值都为零或负数，则随机选择一个动作
        if all(regret <= 0 for regret in self.cum_regrets):
            return kuhn.Action.random()

        # 将负遗憾值转换为0，仅保留正遗憾值作为权重
        weights = [max(0, regret) for regret in self.cum_regrets]
        # 使用随机权重采样
        chosen_index = random.choices(range(len(weights)), weights=weights, k=1)[0]

        return kuhn.Action.from_int(chosen_index)

    # 获取每个动作的概率，考虑了遗憾值和探索率 对于CFR，就是0.5 0.5的平均探索
    def get_action_probability(self):
        regret_sum = sum(max(0, regret) for regret in self.cum_regrets)
        # 如果总遗憾值为0，则每个动作概率均等
        if regret_sum > 0:
            result = [(max(0, regret) / regret_sum) for regret in self.cum_regrets]
        else:
            result = [1.0 / kuhn.Action.num()] * kuhn.Action.num()

        explore_prob = self.epsilon / kuhn.Action.num()
        # 将探索概率和策略概率融合
        result = [explore_prob + prob * (1 - self.epsilon) for prob in result]

        return result

    # 重写对象的字符串表示形式，显示过牌和下注的比例
    def __repr__(self):
        check_ratio = self.cum_strategy[0] / (self.cum_strategy[0] + self.cum_strategy[1])
        bet_ratio = 1.0 - check_ratio
        return f"[Check: {check_ratio}, Bet: {bet_ratio}]"


# 使用CFR算法的玩家类
class CfrPlayer:
    def __init__(self,mode="Random"):
        # 玩家ID，初始化为 -1
        self.player_id = -1
        # 玩家手牌，初始化为 -1
        self.hand_card = -1
        # 玩家的资金
        self.money = 0
        # 存储信息集到CFR节点的映射 Information-set:CFR-Node
        self.cfr_info = {}

        self.mode = mode

    # 玩家注册方法，设置玩家ID
    def on_register(self, player_id):
        self.player_id = player_id

    # 游戏开始时，设置玩家手牌
    def on_start(self, card):
        self.hand_card = card

    # 根据动作历史决定玩家的行动
    def decide_action(self, action_history):
        if self.mode == "Random":
            return kuhn.Action.random()
        info_set = InformationSet(action_history, self.hand_card)
        if info_set not in self.cfr_info.keys():
            # print(info_set,"------118-----")
            self.cfr_info[info_set] = CfrNode(0.0)
            return kuhn.Action.random()
        # print(self.cfr_info[info_set].cum_regrets)
        return self.cfr_info[info_set].get_action()

    # 处理游戏结果，更新玩家资金并打印相关信息
    def handle_result(self, action_history, payoff):
        self.money += payoff
        info_set = InformationSet(action_history, self.hand_card)
        # print(f"player {self.player_id} get payoff {payoff} with info set {str(info_set)}")

    # 训练玩家策略，进行多次迭代
    def train(self, iteration):
        total_payoff = 0.0
        episode_payoff = 0.0
        # start = time.time()
        for round in range(iteration):
            cards = [1, 2, 3]
            action_history = kuhn.ActionHistory([])
            random.shuffle(cards)
            history_probs = {0: 1.0, 1: 1.0}
            round_payoff = 0.0
            if self.mode == "CFR":
                round_payoff = self.cfr(action_history, cards, history_probs)
            elif self.mode == "MCCFR":
                round_payoff = self.mccfr(action_history, cards, history_probs)
            elif self.mode == "OOS":
                _, _, round_payoff = self.oos(action_history, cards,{0: 1.0, 1: 1.0}, {0: 1.0, 1: 1.0})
            total_payoff += round_payoff
            episode_payoff += round_payoff
            if (round+1)%1000==0:
                print(f"round {round}, round payoff {round_payoff},1000-average payoff {episode_payoff / 1000},average payoff {total_payoff / round}")
                episode_payoff = 0.0
            # end = time.time()
            # if end - start > 60:
            #     print(f"round {round}, average payoff {total_payoff / iteration}")
            #     break

    # CFR算法的核心递归方法，计算玩家在当前状态下的价值
    def cfr(self, history, cards, reach_probs):
        player_id = len(history) % 2

        # 检查游戏是否结束，如果结束则返回相应的收益
        maybe_payoff = kuhn.get_payoff(history, cards)
        if maybe_payoff is not None:
            payoff = maybe_payoff if player_id == 0 else -maybe_payoff
            return float(payoff)

        # 处理信息集，如果信息集不存在则创建新的CFR节点
        info_set = InformationSet(history, cards[player_id])
        if info_set not in self.cfr_info.keys():
            self.cfr_info[info_set] = CfrNode(0.0)

        # 获取当前信息集下的行动概率
        action_probs = self.cfr_info[info_set].get_action_probability()
        # 存储每个行动的收益
        action_payoffs = []
        node_value = 0.0

        # 遍历每个可能的行动
        for action_id, action_prob in enumerate(action_probs):
            # 创建下一个动作历史的副本
            next_history = history.clone()
            # 将当前行动添加到动作历史中
            next_history.raw.append(kuhn.Action.from_int(action_id))
            # 复制到达当前状态的概率
            next_reach_probs = reach_probs.copy()
            # 更新到达下一个状态的概率
            next_reach_probs[player_id] *= action_prob
            # 递归调用cfr方法计算下一个状态的收益，并取负（因为是对手的视角）
            action_payoffs.append(-self.cfr(next_history, cards, next_reach_probs))
            node_value += action_prob * action_payoffs[action_id]

        assert len(action_payoffs) == 2
        # 根据计算得到的收益更新节点的累计遗憾值
        node = self.cfr_info[info_set]
        # print(node.cum_strategy) #0.0,0.0
        for action_id, payoff in enumerate(action_payoffs):
            regret = payoff - node_value #动作的反事实值 - 策略的反事实值
            opponent = 1 - player_id
            node.cum_regrets[action_id] += reach_probs[opponent] * regret #计算这个节点的遗憾值

        # 根据行动概率更新节点的累计策略,这个变量用于打印策略信息，不影响
        for action_id, action_prob in enumerate(action_probs):
            node.cum_strategy[action_id] += reach_probs[player_id] * action_prob


        return node_value