您应当知道的 MQL5 向导技术（第 62 部分）：结合 ADX 与 CCI 形态的强化学习 TRPO

概述

我们继续考察如何在机器学习中，与跟踪价格不同部分的技术指标进行配对。在上一篇文章中，我们见识了在多层感知器（MLP）中的监督学习如何为预测价格走势奠定基础。我们引用 MLP 的输入当作特征，其预测输出当作状态。我们在上一篇文章中定义特征的方式，与第 57 - 60 部分的文章略有不同，我们的靶向是采用更连续的输入向量，而这与我们之前曾用过离散选项对立。从离散数据和分类转向连续数据和回归，如果我们考察人工智能的趋势，或能更好地论证。

以往用法，为了提示计算机程序有可用或实际响应，必须将这个响应手写进程序当中。本质上，if-从句是大多数程序编程的核心。若您思考过有关内容，依赖 if-从句意味着用户输入数据、或由程序处理的数据必须属于某些类别。必须是离散的。因此，对于大部分而论，我们在响应中开发和使用离散数据是对编程约束，而与数据或正在解决的问题无关。

然后，2023 年秋季 OpenAI 推出了他们的首个公开 GPT，一切都随之变化。变换器网络和 GPT 的发展并非一夜发生，在于第一批感知器是在 60 年代末才被开发出来的，但可以确凿地说，ChatGPT 的推出是一个重要的里程碑。随着大语言模型的广泛运用，日益清楚，词元化、单词嵌入，当然还有自我注意力，已成为模型能够扩充其可处理内容的关键组件。没有更多 if-从句。配合词元化、以及单词嵌入，令网络输入尽可能连续的背景下，我们也令监督学习 MLP 的输入“更加连续”。

为了描绘这一点，我们的第二个特征-1，以 Python 按如下 MLP 表示：

def feature_1(adx_df, cci_df):
    """
    Creates a modified 3D signal array with:
    1. ADX > 25 (1 when above 25, else 0)
    2. CCI crosses from below 0 to above +50 (1 when condition met, else 0)
    3. CCI crosses from above 0 to below -50 (1 when condition met, else 0)
    """
    # Initialize empty array with 3 dimensions
    feature = np.zeros((len(adx_df), 5))
    
    # Dimension 1: ADX above 25 (continuous, not just crossover)
    feature[:, 0] = (adx_df['adx'] > 25).astype(int)
    
    # Dimension 2: CCI crosses from <0 to >+50
    feature[:, 1] = (cci_df['cci'] > 50).astype(int)
    feature[:, 2] = (cci_df['cci'].shift(1) < 0).astype(int)
    
    # Dimension 3: CCI crosses from >0 to <-50
    feature[:, 3] = (cci_df['cci'] < -50).astype(int)
    feature[:, 4] = (cci_df['cci'].shift(1) > 0).astype(int)
    
    # Set first row to 0 (no previous values to compare)
    feature[0, :] = 0
    
    return feature

如果我们坚持之前 #57 到 #60 文章里采用的方法，那么它会被如下处理：

def feature_1(adx_df, cci_df):
    """
    """
    # Initialize empty array with 3 dimensions and same length as input
    feature = np.zeros((len(dem_df), 3))
    
    # Dimension 1:
    feature[:, 0] = (adx_df['adx'] > 25).astype(int)
    feature[:, 1] = ((cci_df['cci'] > 50) &
                     (cci_df['cci'].shift(1) < 0)).astype(int)
    feature[:, 2] = ((cci_df['cci'] < -50) &
                     (cci_df['cci'].shift(1) > 0)).astype(int)
    
    # Set first row to 0 (no previous values to compare)
    feature[0, :] = 0
    
    return feature

这种方式倾向于把信号按照预期的典型看涨和看跌形态列队分类，因为输出向量中的第二个项仅捕捉看涨信号的特质。第三项仅捕捉看跌特质。坚守已定义的看涨或看跌形态，该方式更偏向分类，因此是离散的。话虽如此，我们的测试结果显示，测试的 10 个形态中只有 3 个能够从 2024.01.01 至 2025.01.01 通过前向漫游测试，而这些形态曾据 2020.01.01到 2024.01.01 测试/训练。所用品种是 EURUSD，时间帧是日线时间帧。

如是所见，我们所用的较长时间帧、及相对较长的训练窗口，对于我们的初始 MLP，坚持采用离散的输入数据可能更具可信度。如果我们考察 LLM 的输入，也可能就此产生一个深入案例。是的，词元化和单词嵌入令输入数据更连续，不过大语言模型的自注意力“秘制酱汁”本质上是离散的。这是因为它搜寻提示输入中提供的每个单词的相对重要性权重。

我们并未做类似的事情，因此这可能是一种解释。因此，读者可自由修改和测试不同的输入格式，因为所有 MQL5 源代码都附带于后。就我们而言，我们会坚持该方式，看看强化学习会带来什么结果。

强化学习

我们在上一篇文章构建了监督学习模型，引入了动作和奖励。回忆一下，我们有特征作为 MLP 的输入，以及状态（预测的价格变化）作为输出。该阶段的动作代表的是，当我们知道 MLP 正在预测什么时，我们需要如何行事。例如，如果预测价格下跌，我们就能执行限价卖出、或破位卖出、或即时市价卖出。开发和训练政策网络有助于磨练这一决策。

通常在上述我们执行不同卖单类型示例的给定场景中，政策网络输出向量的大小会与可能动作的数量相匹配。在该情况下，应该是 3 个选项：限价单、止损单、和市价单。遵照这些设定进行训练应当存在性能差异，欢迎读者去探索。就我们而言，我们坚持认为动作是一个单维向量，本质上是我们状态输出向量的复制品，即上一篇文章中 MLP 的状态输出向量。那么这样做的目的是什么？它权当依据监督学习网络来确认看涨或看跌预测。

此外，奖励还被用来调整每笔交易的盈利额度大小。奖励是价值网络的输出，尽管我们再次以 1-维向量来衡量它们，但它们也可以是多维的。这是因为后期交易分析不仅能参考盈亏，还包括短途走势。这些分为有利和有害，因此奖励向量也可以是 3-维大小，包含有害短途走势、有利短途走势、和净收益。

信任区域政策优化

信任区域政策优化（TRPO）是一种强化学习算法，全与改进政策有关。它迭代更新政策网络的权重和偏差，同时将它们保持在当前政策的“信任区域”之内。

实现中所涉及的关键组件是：政策网络；信任区域；以及 KL-发散度。政策网络是一个神经网络，将状态映射到可能动作的概率分布来表示动作选择。信任区域是一个约束，限制政策每次迭代的变化。它确保新政策不会偏离旧政策太多，从而避免不稳定。最后，KL-发散度衡量预测概率分布与信任概率分布之间的差异。它实质上定义信任区域约束。

训练过程涉及：在使用现行政策的前提下，理想情况下按轨迹批量收集数据；针对数据中的状态-动作对估算优势函数，以便获取最佳动作较之平均动作的优劣程度；优化问题的公式化，以便寻找合适的政策和价值网络的权重和偏差，从而在 KL-发散度约束下最大化奖励，即新旧政策之间的发散度保持在指定阈值之内；通过梯度下降等技术求解优化问题；最后更新政策和价值网络的权重。

TRPO 的关键优势：单调性改进，其中政策改进有保障；稳定性，因为信任区域预防政策的不必要大幅更新，这会造成不稳定；以及效率，因为 TRPO 比其它政策梯度方法更侧重依据少量样本来有效学习。总之，TRPO 的核心理念是最大化新政策相较于旧政策的预期优势，意在约束政策变动的允许幅度。它由以下方程捕捕捉：

其中：

• θ：新政策参数。
• θold：更新前的旧策略参数。
• πθ(a∣s)：新策略 πθ 下行动动作 a 的概率
• πθold(a∣s)：旧政策 πθold 下行动的概率
• Aπθold(s,a)：优势函数，估算 a 在状态 s 下优于平均动作的程度。
• ρθold (s)：旧政策下的状态探视分布。
• DKL：库尔巴克-莱布勒（Kullback-Leibler）发散度，衡量旧政策与新政策的差异。
• δ：信任区域约束（较小的正数值）。

政策网络

我们以 Python 如下实现政策和价值网络：

class PolicyNetwork(nn.Module):
    def __init__(self, state_dim, action_dim, hidden_size=64, discrete=False):
        super(PolicyNetwork, self).__init__()
        self.discrete = discrete
        
        self.fc1 = nn.Linear(state_dim, hidden_size)
        self.fc2 = nn.Linear(hidden_size, hidden_size)

        self.export_mode = False
        
        if self.discrete:
            self.fc3 = nn.Linear(hidden_size, action_dim)
        else:
            self.mean = nn.Linear(hidden_size, action_dim)
            self.log_std = nn.Parameter(torch.zeros(action_dim))
    
    def forward(self, x):
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        
        if self.discrete:
            action_probs = F.softmax(self.fc3(x), dim=-1)
            dist = Categorical(action_probs)
        else:
            mean = self.mean(x)
            std = torch.exp(self.log_std)
            
            if self.export_mode:
                return mean, std  # return raw tensors
            
            cov_mat = torch.diag_embed(std).unsqueeze(dim=0)
    
            dist = MultivariateNormal(mean, cov_mat)
        
        return dist

政策网络继承自 nn.module，其为一个 PyTorch 网络模块。它取状态维度的参数，即输入状态空间的大小；动作维度，即动作空间的大小；隐藏层的尺寸，我们默认为 64。我们还取一个布尔标志标志，作为输入的离散标签，即设定动作空间是离散的、亦或连续的。这个离散标志最终决定了输出层结构、及所用的分布类型。

这种设定令我们的网络在强化环境中，离散和连续动作空间两者都能处理。故此，像 CartPole 这样的游戏可用离散值，等于 true 选项，而我们的交易案例、甚至机器人案例则会将这个选项设置为 false。在 TRPO 中，政策网络定义智代的政策，即状态与动作的映射。因此，处理不同类型作用空间的灵活性在普适性中非常重要。至于交易者，我们可将动作约束在上述三种订单类型：限价、破位、和市价单，且在这种情况下离散会被赋予 true 值。 

当实现这一点时，重点要确保状态维度和动作维度与环境或所用数据集中的规格集相匹配。离散标志还应与环境的动作空间类型一致。隐藏大小 64 是一个可调节的超参数，根据环境或数据集的复杂度，可能要增加，甚至增加更多隐藏层。

网络架构具有两个全连通线性层，其中 fc1 将输入状态映射到隐藏层，fc2 映射到另一个同样大小的隐藏层。“export_mode” 是一个标志，用来控制网络是返回未加工张量以供导出，还是返回分布以便训练/抽样。 

这些层最终搭建起政策网络的骨干，因为它们的效果是将未加工状态输入转化为适合动作选择的高层次表示。采用简单的、激活 ReLU 的两层架构，我们获得足够的表现力，又保持模型轻量。在 TRPO 中，政策网络必须是可微的。这对于计算政策更新的梯度非常重要。结合这两个线性系统，满足了这一需求。 

选择两个大小为 64 的隐藏层作为默认设置是合理的，但随着环境或测试数据集变得更复杂或庞大，往往需要调整。若为获得特征形态而配对的指标超过两个，或向政策网络输入更精细的状态，那该数字就需要放大。 

最后，根据我们的网络是否运行在离散模式，最终输出层会包含单个或两个网络。对于离散空间，fc3 线性层的输出，是每个动作的 logits 表示。另一方面，离散若设置为 false，则我们是与连续空间打交道，此情况下两个网络各自输出单独的向量。第一个是每个动作维度的高斯分布平均值。其二，我们得到每个动作维度的高斯分布的对数标准差向量。

这很重要，因为分岔允许按不同动作空间类型建模。离散（开）选项输出一个涵盖预设动作数量的概率分布。连续或离散（关）备案输出多元高斯分布。简单说就是两个向量，其一是均值，为每个动作提供了指标性均值和权重；而另一个对数分布向量则为每个平均预测提供对数分布、或置信度量值。

在 TRPO 中，政策分布用于抽样动作，并计算政策梯度更新的对数概率。因此，选择离散和连续，会影响所需的计算次数和网络效率。对数概率的标准差是一个可学习参数，令网络在训练过程中能够调整探索水平（方差）。这对于平衡探索和开发非常重要。

对于离散动作，动作维度应与可能动作的数量相匹配。我们坚持单维，因为这是一个连续变量，但我们会在后续文章中重新讨论离散动作选项。尽然，对于我们当前的连续动作，始终需要谨慎初始化 log_std。从 Torch.zeros(action_dim) 开始意味着初始标准差为 exp(0)=1，其或许过宽或过窄，这取决于动作尺度。应置办针对环境的缩放方法。

在 TRPO 中，政策的对数概率也用于目标函数，和 kl_divergence 约束。这意味着它对于确保分布内的数值稳定性至关重要。最后，在环境采用边界动作的情况下，网络输出有时必然会超界。因此，这需要对平均输出进行裁剪或缩放，以确保它们符合预期范围。

政策网络的前向通验，执行通用过程，其中输入状态 x 经由 fc1 和 fc2 通验，并应用 ReLU 激活，以增加非线性。在动作是离散的情况，那么 fc3 的输出会经由 softmax，以产生动作概率。如果它是连续的，则均值通过均值层计算；标准差按 exp(log_std) 计算，以确保其为正数值；如果 export_mode 设为 true，则返回未加工的标准差张量。如果 export_mode 为 false，则将构造对角协方差矩阵（cov_mat），并为抽样对数创建一个多元正态分布。

前向通验定义了政策措施，即如何将状态映射到动作分布，这也是强化学习智代的决策指定核心。在 TRPO 中，政策分布：在环境交互过程中抽样动作，计算政策梯度目标的对数概率，评估信任区域约束。使用类别分布和多元正态分布，可确保与标准强化学习库（如 PyTorch 的 torch.distributions）兼容。export_mode 选项允许实际部署，因为所用是未加工输出，其能够按需后期处理。

在离散情况下 softmax 确保概率总和为 1。数值在 logits 内部频频偏向不稳定，因此应监测 NaN，并按照需要使用 Torch.Clamp。对于连续动作，对角 cov-mat 矩阵假设动作维数独立。若另一方面动作相关，则应当应用完整的协方差矩阵。这会增加计算成本。在 TRPO 中，政策对数概率应计算高效、且准确，因为它们在共轭梯度和线搜索步骤还会被用到。

价值网络

我们如下实现价值网络：

class ValueNetwork(nn.Module):
    def __init__(self, state_dim, hidden_size=64):
        super(ValueNetwork, self).__init__()
        self.fc1 = nn.Linear(state_dim, hidden_size)
        self.fc2 = nn.Linear(hidden_size, hidden_size)
        self.fc3 = nn.Linear(hidden_size, 1)
    
    def forward(self, x):
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        x = self.fc3(x)
        return x

价值网络的设计和实现，与政策网络有相当多的重叠。因此我将略过大部分内容。然而，理论上，价值网络估算状态值，是出于优势（或奖励）计算和方差降低。我们还采用了类似政策网络的简单架构，我们的输出是一个单标量。对于在 TRPO 中经由准确估算奖励，稳定更新政策，该网络至关重要。

TRPO 智代

我们如下以 Python 实现 TRPO 智代类：

class TRPO_Agent:
    def __init__(self, state_dim, action_dim, discrete=False, 
                 hidden_size=64, lr_v=0.001, gamma=0.99, 
                 delta=0.01, lambda_=0.97, max_kl=0.01, cg_damping=0.1, 
                 cg_iters=10, device='cpu'):
        
        self.policy = PolicyNetwork(state_dim, action_dim, hidden_size, discrete).to(device)
        self.value_net = ValueNetwork(state_dim, hidden_size).to(device)
        self.value_optimizer = optim.Adam(self.value_net.parameters(), lr=lr_v)
        
        self.gamma = gamma
        self.delta = delta
        self.lambda_ = lambda_
        self.max_kl = max_kl
        self.cg_damping = cg_damping
        self.cg_iters = cg_iters
        
        self.discrete = discrete
        self.device = device
        self.state_dim = state_dim
    
    def get_action(self, state):
        # Convert state to tensor and add batch dimension
        state = torch.FloatTensor(state).unsqueeze(0).to(self.device)
        
        # Get action distribution from policy
        dist = self.policy(state)
        
        # Sample action from distribution
        action = dist.sample()
        
        # Get log probability BEFORE converting to numpy/item
        log_prob = dist.log_prob(action)
        
        # Convert action to appropriate format
        if self.discrete:
            action = action.item()  # For discrete actions
        else:
            action = action.detach().cpu().numpy()[0]  # For continuous actions
        
        # Clip continuous actions to [-1, 1] range (optional for discrete)
        if not self.discrete:
            action = np.clip(action, -1, 1)
        
        return action, log_prob
        
    def update_value_net(self, states, targets):
        # Convert inputs to proper tensor format
        if torch.is_tensor(states):
            states = states.detach().cpu().numpy()
        if torch.is_tensor(targets):
            targets = targets.detach().cpu().numpy()
        
        states = np.array(states, dtype=np.float32)
        targets = np.array(targets, dtype=np.float32)
        
        # Ensure proper shapes
        if len(states.shape) == 1:
            states = np.expand_dims(states, 0)
        if len(targets.shape) == 0:
            targets = np.expand_dims(targets, 0)
        
        states_tensor = torch.FloatTensor(states).to(self.device)
        targets_tensor = torch.FloatTensor(targets).to(self.device)
        
        # Forward pass
        self.value_optimizer.zero_grad()
        values = self.value_net(states_tensor)
        
        # Ensure matching shapes for loss calculation
        values = values.view(-1)
        targets_tensor = targets_tensor.view(-1)
        
        loss = F.mse_loss(values, targets_tensor)
        loss.backward()
        self.value_optimizer.step()

    def update_policy(self, states, actions, old_log_probs, advantages):
        # Handle tensor conversion safely
        def safe_convert(x):
            if torch.is_tensor(x):
                return x.detach().cpu().numpy()
            return np.array(x, dtype=np.float32)
        
        states = safe_convert(states)
        actions = safe_convert(actions)
        old_log_probs = safe_convert(old_log_probs)
        advantages = safe_convert(advantages)
        
        # Convert to tensors with proper shapes
        states_tensor = torch.FloatTensor(states).to(self.device)
        actions_tensor = torch.FloatTensor(actions).to(self.device)
        old_log_probs_tensor = torch.FloatTensor(old_log_probs).to(self.device)
        advantages_tensor = torch.FloatTensor(advantages).to(self.device)
        
        # Get old distribution
        with torch.no_grad():
            old_dist = self.policy(states_tensor)
        
        # Compute gradient of surrogate loss
        def get_loss():
            dist = self.policy(states_tensor)
            if self.discrete:
                log_probs = dist.log_prob(actions_tensor.long())
            else:
                log_probs = dist.log_prob(actions_tensor)
            return -self.surrogate_loss(log_probs, old_log_probs_tensor, advantages_tensor)
        
        # Rest of the TRPO update remains the same...
        loss = get_loss()
        grads = torch.autograd.grad(loss, self.policy.parameters(), create_graph=True)
        flat_grad = torch.cat([grad.view(-1) for grad in grads]).detach()
        
        step_dir = self.conjugate_gradient(states_tensor, old_dist, flat_grad, nsteps=self.cg_iters)
        
        shs = 0.5 * torch.dot(step_dir, self.hessian_vector_product(states_tensor, old_dist, step_dir))
        step_size = torch.sqrt(self.max_kl / (shs + 1e-8))
        full_step = step_size * step_dir
        
        old_params = torch.cat([param.view(-1) for param in self.policy.parameters()])
        
        def line_search():
            for alpha in [0.5**x for x in range(10)]:
                new_params = old_params + alpha * full_step
                self.set_policy_params(new_params)
                
                with torch.no_grad():
                    new_dist = self.policy(states_tensor)
                    new_loss = get_loss()
                    kl = self.kl_divergence(old_dist, new_dist)
                
                if kl <= self.max_kl and new_loss < loss:
                    return True
            return False
        
        if not line_search():
            self.set_policy_params(old_params)
    
    def set_policy_params(self, flat_params):
        prev_idx = 0
        for param in self.policy.parameters():
            flat_size = param.numel()
            param.data.copy_(flat_params[prev_idx:prev_idx + flat_size].view(param.size()))
            prev_idx += flat_size
    
    def compute_advantages(self, rewards, values, dones):
        advantages = np.zeros_like(rewards)
        last_advantage = 0
        
        for t in reversed(range(len(rewards))):
            if dones[t]:
                delta = rewards[t] - values[t]
                last_advantage = delta
            else:
                delta = rewards[t] + self.gamma * values[t+1] - values[t]
                last_advantage = delta + self.gamma * self.lambda_ * last_advantage
            advantages[t] = last_advantage
        
        advantages = (advantages - advantages.mean()) / (advantages.std() + 1e-8)
        return advantages
    
    def surrogate_loss(self, new_probs, old_probs, advantages):
        ratio = torch.exp(new_probs - old_probs)
        return torch.mean(ratio * advantages)
    
    def kl_divergence(self, old_dist, new_dist):
        if self.discrete:
            return torch.mean(torch.sum(old_dist.probs * (torch.log(old_dist.probs) - torch.log(new_dist.probs)), dim=1))
        else:
            return torch.distributions.kl.kl_divergence(old_dist, new_dist).mean()
    
    def hessian_vector_product(self, states, old_dist, vector):
        kl = self.kl_divergence(old_dist, self.policy(states))
        
        # First compute gradient of KL
        grads = torch.autograd.grad(kl, self.policy.parameters(), create_graph=True)
        flat_grad_kl = torch.cat([grad.view(-1) for grad in grads])
        
        # Compute gradient of (grad_KL * vector)
        grad_vector_product = torch.sum(flat_grad_kl * vector)
        grad_grad = torch.autograd.grad(grad_vector_product, self.policy.parameters(), retain_graph=True)
        flat_grad_grad = torch.cat([grad.contiguous().view(-1) for grad in grad_grad])
        
        return flat_grad_grad + self.cg_damping * vector
    
    def conjugate_gradient(self, states, old_dist, b, nsteps=10, residual_tol=1e-10):
        x = torch.zeros_like(b)
        r = b.clone()
        p = b.clone()
        rdotr = torch.dot(r, r)
        
        for i in range(nsteps):
            Avp = self.hessian_vector_product(states, old_dist, p)
            alpha = rdotr / torch.dot(p, Avp)
            x += alpha * p
            r -= alpha * Avp
            new_rdotr = torch.dot(r, r)
            if new_rdotr < residual_tol:
                break
            beta = new_rdotr / rdotr
            p = r + beta * p
            rdotr = new_rdotr
        
        return x

TRPO 智代类适用于连续和离散动作空间两者，它使用政策网络来选择动作，而价值网络估算每种状态的奖励。于此，TRPO 价值网络的一个关键区别在于，输入仅为状态，不包含动作，这在其它强化学习算法中常见。TRPO 以最大化代表目标来优化政策，同时约束政策更新保持在由 kl_divergence 定义的信任区域内。该类包括选择动作的方法、价值函数更新、政策优化、奖励估算、及其它计算。

def __init__() 函数启动 TRPO 智代，配以政策和价值网络、优化器、以及 TRPO 信任区域优化的超参数。输入包括超参数，其中一些是 TRPO 特有的，比如 “max_kl” & “cg_damping”，还有一些是强化学习特有的，譬如 gamma & lambda。在调整这些超参数时，默认值 max_kl=0.01、cg_damping=0.1、以及 lambda=0.97 是合理的，但取决于环境或数据集的特殊性。

对于更复杂的环境，譬如高维数据集，对于更严格的约束，较小的 max_kl 约为 0.005；或较大的 cg_iters 约为 20，可能更适合共轭梯度收敛。针对价值网络的 Adam 的优化器选择是标准的，但政策网络依赖于 TRPO 的自定义更新，无需优化器。同时，确保价值网络学习率足够小，从而网络学习稳定性更佳，也是个好主意。

GetAction 函数将输入状态转换为 PyTorch 张量。状态经由政策网络处理，得到动作分布，从分布中抽样一个动作，计算其对数概率，最后将动作转换为环境兼容格式。该格式是针对离散动作的标量，而针对连续动作则是裁剪的 NumPy 数组。 

该函数代表智代与环境的交互，因为它支持基于政略选择动作。对数概率对于 TRPO 的梯度计算至关重要，因为它用在 surrogate_loss，评估政策性能。将连续动作裁剪到 [-1,1] 范围，能够确保政策网络的输出与环境有界动作的期望兼容。就连续动作，标准差监测应避免退化分布。所用的批处理方法假设单一状态输入，不过在向量化、或多维状态下，这能加以扩展，更高效地处理状态。

价值网络更新将输入状态转换为政策网络预测的 Q-值，即奖励。它计算与目标相悖的均方误差损失，并通过 Adam 优化器的反向传播更新价值网络。准确的价值估算降低了政策梯度的方差，从而提升了 TRPO 的稳定性。MSE 损失确保价值网络学会预测预期贴现回报，从而与强化学习的目标保持一致。我们在计算目标时使用时态差分目标来训练值网络。这些都需要准确计算，因为不准确的估值可能导致不稳定的政策更新。

损失函数是标准的 MSE，不过也可以考虑 Huber-Loss，以便在高方差环境/数据集中对异常值更健壮。形状校正逻辑似乎也很擅长，不过在大数据状况下或许会是一种挑战。这或许需要对输入形状进行预优化，确保它们按正确的形状预处理。此外，梯度剪裁还能协同 Torch.nn.utils.clip_grd_norm_ 等模块加入，以便限制超规格更新，从而稳定网络训练。

政策更新函数将状态、动作、旧的对数概率、以及奖励转换为符合形状的张量。它还能依据旧的政策分布执行 kl_divergence 计算，并定义了 surrogate_loss 函数，这是一种衡量新政策下预期回报的方式，即新、旧政策的对比。此外，政策更新函数计算政策梯度；使用共轭梯度来寻找搜索方向；并根据信任区域约束 “max-kl” 确定步长。它执行线搜索，确保新策略满足 kl_divergence 约束，同时改善 surrogate_loss，若搜索失败，则恢复到旧参数。 

在很多方面，这正是 TRPO 的核心，因为它实现了信任区域优化，在政策改进与稳定性之间取得了平衡。surrogate_loss 估算政策梯度目标，而 kl_divergence 约束确保不会有大幅度的变更政策，否则可能降低性能。换言之，共轭梯度方法高效求解搜索方向，而线搜索则确保了更新的健壮。

在 TRPO 中，max-kl 参数至关重要。太小的数值，譬如低于 0.005，或许会过度约束更新，导致学习速度非常缓慢。相较之，数值太大，譬如高于 0.05，或许导致更新不稳定，这正是 TRPO 试图缓解的问题。参数 “cg-iters”（共轭梯度迭代）应具有足够的大小，以便得以收敛。还应监测残差，从而验证解的准确性。

set_policy_params 函数复制平滑向量的数值，来更新政策网络的参数，然后重塑它们，以匹配每个参数的大小。这允许 TRPO 自定义参数更新，在共轭梯度线搜索期间，作为平滑向量加以计算。因此，该函数确保每次更新后政策网络都能反映已优化的参数。

奖励的计算，其在代码中称为 compute_advantages，通过广义优势估算或 GAE 评定奖励。这涉及计算每个时间步的时太差值误差。TD 误差和 lambda_ 的结合有助于平衡偏差和方差。它还会在每局次结束时依据 dones[t] 参数重置奖励/优势，并将这些奖励归一化为零均值、和单位方差。

surrogate_loss 函数计算代理损失，按照 πnew(a∣s)/πold(a∣s) 乘以 advantages，作为预期概率比率。surrogate_loss 通过衡量政策变动对预期回报的影响，来估算政策梯度目标。在 TRPO 中，在信任区域约束内，该损失被最大化，且因此在 get_loss 中被抵消。

kl_divergence 函数决定了新、旧政策分布之间的远近。当动作是离散时，用到一个类别分布的解析公式。对于连续动作，PyTorch 使用多元正态分布。这些测量有助于强制 TRPO 的信任区域约束。

黑森（Hessian）向量积函数，正如其名，计算在共轭梯度方法中用于 kl-divergence 的黑森积。计算方法是获得 KL-divergence 的梯度，将其与输入向量相乘，然后计算二阶梯度。它加入了阻尼，从而提升数值稳定性。通过近似费舍尔（Fisher）信息矩阵在向量上的动作，我们能在 TRPO 中高效计算搜索方向。阻尼项确保黑森为正定，且共轭梯度收敛。

最后，conjugate_gradient 函数实现了求解 Hx = g 的方法，其中 H 是 hessian_vector_product 函数近似的费舍尔矩阵，g 是政策梯度。它迭代细化解 x 或搜索方向，直至收敛，或完成 n 步迭代。

测试运行

如果我们只取上一篇文章中能够通过前向漫游测试的 3 个特征形态进行前向漫游测试，即特征-2、3 和 4，我们会得到以下报告。我们正在测试 EURUSD 货币对，区间为 2020.01.01 至 2025.01.01。训练依据该区间 80% 的数据，即 2020.01.01 至 2024.01.01 区间，利用 Python 执行。 

如果我们研究仅据 2024 年区间的前向漫游测试，那么似乎只有形态-2 和形态-3 能够走通。一如既往，许多因素在起作用，建议在使用本文中分享的任何代码/素材之前进行独立、尽职的调查。为了汇编和使用上述测试中的智能系统，需用到附带代码的文件和 MQL5 向导。对于新读者，这里和这里有关于如何操作的指导。

结束语

我们之前发表了一篇关于的监督学习模型如何取用 ADX 与 CCI 结合形态作为输入，开发智能系统的文章，以及发布其它一些文章。本文采用相同的指标，但用在强化学习当中。强化学习旨在通过谨慎延长其学习窗口，令早期开发的智能系统更加强大。

随后要撰写一篇总结文章，针对这些指标形态，意在考察推理。我们在此运用推理来总结和“归档”监督学习和强化学习中所学的内容。该方式的概括在该篇文章。不过，我们将把推理留给读者，因为我们将回归更简单的文章格式，交替包含一些机器学习思路。 

名称	描述
wz_62.mq5	向导汇编的智能系统，其头文件显示包含的文件。
SignalWZ_62.mqh	自定义信号类文件
61_2.onnx	特征-2 ONNX 监督学习模型
61_3.onnx	特征-3 ONNX 监督学习模型
61_4.onnx	特征-4 ONNX 监督学习模型
62_policy_2.onnx	特征-2 强化学习参与者
62_policy_3.onnx	特征-3 强化学习参与者
62_policy_4.onnx	特征-4 强化学习参与者
62_value_2.onnx	特征-2 强化学习评论者
62_value_3.onnx	特征-3 强化学习评论者
62_value_4.onnx	特征-4 强化学习评论者