08 - 统计学习理论视角¶

import numpy as np
from typing import Callable, List, Tuple

class RegularizedValueLearning:
    """带正则化的值函数学习"""

    def __init__(self, n_states: int, n_actions: int, gamma: float = 0.9,
                 lambda_reg: float = 0.01):
        self.n_states = n_states
        self.n_actions = n_actions
        self.d = n_states * n_actions  # 特征维度（one-hot编码）
        self.gamma = gamma
        self.lambda_reg = lambda_reg
        self.theta = np.zeros(self.d)

    def phi(self, state: int, action: int) -> np.ndarray:
        """特征映射（示例：one-hot编码）"""
        features = np.zeros(self.d)
        idx = state * self.n_actions + action
        if idx < self.d:
            features[idx] = 1.0
        return features

    def q_value(self, state: int, action: int) -> float:
        """计算Q值"""
        return np.dot(self.theta, self.phi(state, action))  # np.dot矩阵/向量点乘

    def bellman_error(self, samples: List[Tuple],
                      regularization: str = 'l2') -> float:
        """
        计算经验贝尔曼误差

        参数:
            samples: [(s, a, r, s'), ...]
            regularization: 'l2' 或 'none'
        返回:
            带正则化的误差
        """
        total_error = 0.0

        for s, a, r, s_next in samples:
            # 当前Q值估计
            q_sa = self.q_value(s, a)

            # 目标值（使用当前参数）
            q_targets = [self.q_value(s_next, a_next)
                        for a_next in range(self.n_actions)]
            target = r + self.gamma * max(q_targets)

            # TD误差
            total_error += (q_sa - target) ** 2

        empirical_error = total_error / len(samples)

        # 添加正则化项
        if regularization == 'l2':
            reg_term = self.lambda_reg * np.sum(self.theta ** 2)
        else:
            reg_term = 0.0

        return empirical_error + reg_term

    def fit(self, samples: List[Tuple], num_iterations: int = 1000,
            learning_rate: float = 0.01) -> np.ndarray:
        """
        使用梯度下降拟合值函数

        参数:
            samples: 训练样本
            num_iterations: 迭代次数
            learning_rate: 学习率
        返回:
            theta: 学习后的参数
        """
        for _ in range(num_iterations):
            grad = np.zeros(self.d)

            for s, a, r, s_next in samples:
                phi_sa = self.phi(s, a)
                q_sa = np.dot(self.theta, phi_sa)

                # 计算目标
                q_targets = [self.q_value(s_next, a_next)
                            for a_next in range(self.n_actions)]
                target = r + self.gamma * max(q_targets)

                # 梯度
                grad += 2 * (q_sa - target) * phi_sa

            grad /= len(samples)

            # L2正则化梯度
            grad += 2 * self.lambda_reg * self.theta

            # 更新
            self.theta -= learning_rate * grad

        return self.theta

from typing import List, Tuple
import numpy as np

class BatchQLearning:
    """批量Q-Learning（带悲观修正）"""

    def __init__(self, n_states: int, n_actions: int,
                 gamma: float = 0.9, pessimism: float = 0.1):
        self.n_states = n_states
        self.n_actions = n_actions
        self.gamma = gamma
        self.pessimism = pessimism
        self.Q = np.zeros((n_states, n_actions))

    def fit(self, dataset: List[Tuple], num_iterations: int = 1000):
        """
        从批量数据学习Q函数

        参数:
            dataset: [(s, a, r, s', done), ...]
            num_iterations: 拟合迭代次数
        """
        # 统计访问次数（用于不确定性估计）
        visit_counts = np.zeros((self.n_states, self.n_actions))
        for s, a, r, s_next, done in dataset:
            visit_counts[s, a] += 1

        # 计算悲观惩罚
        uncertainty = self.pessimism / np.sqrt(visit_counts + 1)

        for _ in range(num_iterations):
            Q_new = self.Q.copy()

            for s, a, r, s_next, done in dataset:
                if done:
                    target = r
                else:
                    target = r + self.gamma * np.max(self.Q[s_next])

                # 悲观修正：减少不确定性高的Q值
                Q_new[s, a] += (target - uncertainty[s, a] - self.Q[s, a]) / len(dataset)

            self.Q = Q_new

    def get_policy(self) -> np.ndarray:
        """提取贪婪策略"""
        return np.argmax(self.Q, axis=1)

    def evaluate_coverage(self, dataset: List[Tuple]) -> dict:
        """
        评估数据集对状态-动作空间的覆盖程度

        返回:
            coverage_metrics: 覆盖率指标
        """
        visited_sa = set()
        state_visits = np.zeros(self.n_states)

        for s, a, r, s_next, done in dataset:
            visited_sa.add((s, a))
            state_visits[s] += 1

        total_sa = self.n_states * self.n_actions
        coverage_metrics = {
            'sa_coverage': len(visited_sa) / total_sa,
            'state_coverage': np.sum(state_visits > 0) / self.n_states,
            'min_state_visits': np.min(state_visits[state_visits > 0]) if np.any(state_visits > 0) else 0,  # any()任一为True则返回True
            'max_state_visits': np.max(state_visits),
            'visit_entropy': -np.sum((state_visits / np.sum(state_visits)) *
                                     np.log(state_visits / np.sum(state_visits) + 1e-10))
        }

        return coverage_metrics