21、强化学习信号：DQN、PPO在交易信号中的应用

说实话，刚接触强化学习做交易那会儿，我也挺懵的。传统的技术指标吧，你给它一个价格序列，它给你一个信号，逻辑是确定的。但强化学习不一样，它是个「试错」的过程——让智能体自己去跟市场交互，亏了钱就长记性，赚了钱就强化行为。说白了，就是让机器自己学会怎么交易。

我个人习惯把强化学习交易系统拆成三块：状态（市场长什么样）、动作（我能干什么）、奖励（干得好不好）。今天咱们重点聊两个主流算法：DQN 和 PPO。它们在高频信号提取里各有各的脾气。

21.1 从 Q-Learning 到 DQN：离散动作的利器

先说说 DQN。它的前身是 Q-Learning，核心思想是维护一张 Q 表，记录「在某个状态下，做某个动作能赚多少钱」。但高频数据的状态空间太大了——价格、成交量、订单簿深度、波动率……你根本没法用表格存。

DQN 的聪明之处在于：用神经网络来近似 Q 函数。输入是市场状态，输出是每个动作的 Q 值。我选 Q 值最大的那个动作去执行。

嗯，这里要注意：DQN 适合离散动作。比如「买入」、「卖出」、「持有」三个动作。如果你要做连续仓位控制（比如 0.3 手、0.7 手），那 DQN 就不太行了。

实战代码：用 DQN 生成买卖信号

import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
from collections import deque
import random

# 定义 DQN 网络
class DQN(nn.Module):
    def __init__(self, state_dim, action_dim):
        super().__init__()
        self.fc1 = nn.Linear(state_dim, 128)
        self.fc2 = nn.Linear(128, 64)
        self.fc3 = nn.Linear(64, action_dim)
    
    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        return self.fc3(x)

# 经验回放缓冲区
class ReplayBuffer:
    def __init__(self, capacity=10000):
        self.buffer = deque(maxlen=capacity)
    
    def push(self, state, action, reward, next_state, done):
        self.buffer.append((state, action, reward, next_state, done))
    
    def sample(self, batch_size):
        batch = random.sample(self.buffer, batch_size)
        states, actions, rewards, next_states, dones = zip(*batch)
        return (np.array(states), np.array(actions), 
                np.array(rewards), np.array(next_states), np.array(dones))

# 训练 DQN 智能体
class DQNAgent:
    def __init__(self, state_dim, action_dim, lr=1e-3, gamma=0.99):
        self.q_net = DQN(state_dim, action_dim)
        self.target_net = DQN(state_dim, action_dim)
        self.target_net.load_state_dict(self.q_net.state_dict())
        self.optimizer = optim.Adam(self.q_net.parameters(), lr=lr)
        self.gamma = gamma
        self.buffer = ReplayBuffer()
        self.action_dim = action_dim
    
    def act(self, state, epsilon=0.1):
        if np.random.random() < epsilon:
            return np.random.randint(self.action_dim)
        state_tensor = torch.FloatTensor(state).unsqueeze(0)
        q_values = self.q_net(state_tensor)
        return torch.argmax(q_values).item()
    
    def update(self, batch_size=64):
        if len(self.buffer.buffer) < batch_size:
            return
        states, actions, rewards, next_states, dones = self.buffer.sample(batch_size)
        
        states = torch.FloatTensor(states)
        actions = torch.LongTensor(actions).unsqueeze(1)
        rewards = torch.FloatTensor(rewards).unsqueeze(1)
        next_states = torch.FloatTensor(next_states)
        dones = torch.FloatTensor(dones).unsqueeze(1)
        
        # 当前 Q 值
        current_q = self.q_net(states).gather(1, actions)
        # 目标 Q 值
        with torch.no_grad():
            max_next_q = self.target_net(next_states).max(1, keepdim=True)[0]
            target_q = rewards + self.gamma * max_next_q * (1 - dones)
        
        loss = nn.MSELoss()(current_q, target_q)
        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()
    
    def update_target(self):
        self.target_net.load_state_dict(self.q_net.state_dict())

# 使用示例：生成交易信号
def generate_dqn_signals(prices, window=20):
    # 构造状态：过去 window 个价格 + 技术指标
    states = []
    for i in range(window, len(prices)):
        price_window = prices[i-window:i]
        returns = np.diff(price_window) / price_window[:-1]
        state = np.concatenate([
            price_window / price_window[-1] - 1,  # 归一化价格
            returns,
            [np.std(returns)],  # 波动率
            [np.mean(returns)]
        ])
        states.append(state)
    
    agent = DQNAgent(state_dim=len(states[0]), action_dim=3)
    # 实际应用中需要先训练，这里仅展示信号生成逻辑
    signals = [agent.act(s, epsilon=0.0) for s in states]
    # 0=持有, 1=买入, 2=卖出
    return signals

21.2 PPO：连续动作与稳定训练

DQN 有个毛病：训练不稳定。稍微调一下超参数，Q 值就炸了。后来我转用 PPO，发现它更适合高频交易场景。为什么？因为 PPO 是策略梯度方法，直接优化策略网络，而且通过「裁剪」机制保证每次更新不会太激进。

PPO 的核心思想很简单：让新策略不要偏离旧策略太远。它用重要性采样比率来约束更新步长，公式长这样：

其中 r_t(θ) 是新旧策略的概率比，Â_t 是优势函数。说白了就是：如果新策略比旧策略好太多，我就限制你；如果差太多，我也限制你。稳稳当当往前走。

实战代码：PPO 连续仓位控制

import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np

class ActorCritic(nn.Module):
    def __init__(self, state_dim, action_dim):
        super().__init__()
        self.actor = nn.Sequential(
            nn.Linear(state_dim, 128),
            nn.ReLU(),
            nn.Linear(128, 64),
            nn.ReLU(),
            nn.Linear(64, action_dim),
            nn.Tanh()  # 输出在 [-1, 1] 之间
        )
        self.critic = nn.Sequential(
            nn.Linear(state_dim, 128),
            nn.ReLU(),
            nn.Linear(128, 64),
            nn.ReLU(),
            nn.Linear(64, 1)
        )
    
    def forward(self, state):
        action_mean = self.actor(state)
        state_value = self.critic(state)
        return action_mean, state_value

class PPOAgent:
    def __init__(self, state_dim, action_dim, lr=3e-4, gamma=0.99, clip_epsilon=0.2):
        self.net = ActorCritic(state_dim, action_dim)
        self.optimizer = optim.Adam(self.net.parameters(), lr=lr)
        self.gamma = gamma
        self.clip_epsilon = clip_epsilon
    
    def get_action(self, state):
        state_tensor = torch.FloatTensor(state).unsqueeze(0)
        action_mean, _ = self.net(state_tensor)
        # 添加高斯噪声探索
        action = action_mean + torch.randn_like(action_mean) * 0.1
        return torch.clamp(action, -1.0, 1.0).detach().numpy()[0]
    
    def update(self, trajectories):
        # trajectories: list of (state, action, reward, next_state, done)
        states, actions, rewards, next_states, dones = zip(*trajectories)
        
        states = torch.FloatTensor(np.array(states))
        actions = torch.FloatTensor(np.array(actions))
        rewards = torch.FloatTensor(np.array(rewards)).unsqueeze(1)
        next_states = torch.FloatTensor(np.array(next_states))
        dones = torch.FloatTensor(np.array(dones)).unsqueeze(1)
        
        # 计算优势函数（GAE 简化版）
        with torch.no_grad():
            _, next_values = self.net(next_states)
            _, values = self.net(states)
            advantages = rewards + self.gamma * next_values * (1 - dones) - values
        
        # PPO 裁剪更新
        for _ in range(10):  # 多轮更新
            action_means, values = self.net(states)
            # 假设动作服从高斯分布
            dist = torch.distributions.Normal(action_means, 0.1)
            log_probs = dist.log_prob(actions).sum(dim=-1, keepdim=True)
            
            # 旧策略的 log_prob（这里简化处理，实际需存储旧策略）
            old_log_probs = log_probs.detach()
            ratio = torch.exp(log_probs - old_log_probs)
            
            surr1 = ratio * advantages
            surr2 = torch.clamp(ratio, 1 - self.clip_epsilon, 
                                1 + self.clip_epsilon) * advantages
            actor_loss = -torch.min(surr1, surr2).mean()
            critic_loss = nn.MSELoss()(values, rewards + self.gamma * next_values * (1 - dones))
            
            loss = actor_loss + 0.5 * critic_loss
            self.optimizer.zero_grad()
            loss.backward()
            self.optimizer.step()

# 使用示例：生成连续仓位信号
def generate_ppo_signals(prices, window=20):
    states = []
    for i in range(window, len(prices)):
        price_window = prices[i-window:i]
        returns = np.diff(price_window) / price_window[:-1]
        state = np.concatenate([
            price_window / price_window[-1] - 1,
            returns,
            [np.std(returns)],
            [np.mean(returns)]
        ])
        states.append(state)
    
    agent = PPOAgent(state_dim=len(states[0]), action_dim=1)
    # 实际应用中需先训练
    positions = [agent.get_action(s)[0] for s in states]
    # positions 在 [-1, 1] 之间，-1 全仓做空，1 全仓做多
    return positions

21.3 DQN vs PPO：怎么选？

你可能会问：到底用 DQN 还是 PPO？我个人的经验是：

说白了，如果你只需要「买、卖、持有」三个离散信号，DQN 够用。但如果你要做精细化仓位管理——比如根据波动率动态调整杠杆——那 PPO 更合适。

21.4 奖励函数设计：比算法更重要

嗯，这里我要多说一句。很多人把精力花在调网络结构上，却忽略了奖励函数。其实在强化学习里，奖励函数决定了智能体学什么。

我踩过一个大坑：直接用「PnL」做奖励。结果智能体学会了「扛单」——浮亏时死扛，偶尔回本就平仓。虽然最终 PnL 是正的，但最大回撤高达 40%。

后来我改成这样的奖励函数：

def reward_function(pnl, sharpe, max_drawdown, holding_cost):
    # pnl: 本次交易的盈亏
    # sharpe: 滚动夏普比率
    # max_drawdown: 当前最大回撤
    # holding_cost: 持仓成本（滑点+手续费）
    
    reward = pnl - holding_cost  # 基础奖励
    reward += 0.1 * sharpe       # 鼓励高夏普
    reward -= 0.5 * max_drawdown # 惩罚回撤
    return reward

你看，加了夏普和回撤惩罚后，智能体自然就学会了「赚稳钱」，而不是「赌一把」。

21.5 知识体系总览

最后，我用一张图总结一下强化学习信号的整体逻辑：

从这张图能看出来，强化学习交易系统本质上是个闭环：市场数据进来，编码成状态，智能体输出动作，环境反馈奖励，然后更新策略。DQN 和 PPO 只是这个闭环里「策略更新」环节的不同实现方式。

好了，关于 DQN 和 PPO 在交易信号中的应用，今天就聊到这儿。记住：算法是工具，奖励函数才是灵魂。把奖励设计好了，哪怕用最简单的 DQN，也能跑出不错的信号。