【原】深度强化学习DDPG在量化投资的应用

AI量化实验室 2023-10-12 发布于北京

展开全文

主动基金被动管，被动基金主动管。

所以，我们的模型主要应用于场内ETF，ETF可以随时交易且手续费相对较低。而且ETF是支持T+0的。

继续强化学习。

今天探讨一下这DDPG:深度确定性策略梯度(Deep Deterministic Policy Gradient，简称DDPG)

先说“确定性”策略，它和随机策略相对而言的，对于某一些动作集合来说，它可能是连续值，或者非常高维的离散值，这样动作的空间维度极大。如果我们使用随机策略，即像DQN一样研究它所有的可能动作的概率，并计算各个可能的动作的价值的话，那需要的样本量是非常大才可行的。于是有人就想出使用确定性策略来简化这个问题。

虽然在同一个状态处，采用的动作概率不同，但是最大概率只有一个，如果我们只取最大概率的动作，去掉概率分布，那么就简单多了。即作为确定性策略，相同的策略，在同一个状态处，动作是唯一确定的，即策略变成

πθ(s)=a

DDPG参考了DDQN的算法思想，通过双网络和经验回放，加一些其他的优化，比较好的解决了Actor-Critic难收敛的问题。因此在实际产品中尤其是自动化相关的产品中用的比较多，是一个比较成熟的Actor-Critic算法。

baselines里的算法做了非常好的封装，有点类似sklearn里的算法集，就是一个learn和一个predict就搞定了。

与传统机器学习不同在于，传统机器学习的输出就是标签数据，输出就是分类或预测，数据相对简单，算法也相对固定。当然参数的意义还是需要自己去深度理解。

而强化学习本质上是一个框架，里边的算法结构，神经网络一直在进化，所以这个框架叫baselines。我们还是从头自己实现一个简化版本的强化学习，可以参考baseline的封装接口。

基本接口就两个，一个是learn，另一个是predict，今天来实现learn，也就是模型训练。强化学习与深度学习训练有一个不同，深度学习里有episode的概念，就是一批素材训练完了，再来一次，但强化学习就是一个环境，甚至每次经历的返回都会不一样，故而只有一个走多少步，而没有走多少轮之说。当然，针对证券市场，训练数据的tick是确实是固定且按序列出的。

learn函数的主体是一个while循环，一共走N走，比如10000步。

在这10000步里，如果判断env返回done即是一个episode。

这个大循环比较清晰，和深度学习的训练没什么差别（在episode上有一点不同）

def learn(self,env,max_timesteps=1e6,random_steps=1e4):
    step = 0
    done = False
    episode_reward = 0
    episode_num = 0
    obs = env.reset()

    #一共走max_timesteps步
    while step < max_timesteps:
        #done为点计为一个episode
        if done:
            obs = env.reset()
            done = False
            episode_reward = 0
        #这里为何要随机走10%
        if step < random_steps:
            action = env.action_space.sample()

        #根据obs选择action，然后走一步
        action = self.model.select_action(np.array(obs))
        new_obs, reward, done, _ = env.step(action)

        #一次episode的奖励累加
        episode_reward += reward
        episode_num += 1
        obs = new_obs
        step += 1

然后最重要就是调用模型本身训练环节。

网络构成：

class Actor(nn.Module):
    def __init__(self, state_dim, action_dim, max_action):
        super(Actor, self).__init__()
        self.l1 = nn.Linear(state_dim, 400)
        self.l2 = nn.Linear(400, 300)
        self.l3 = nn.Linear(300, action_dim)
        self.max_action = max_action

    def forward(self, x):
        x = F.relu(self.l1(x))
        x = F.relu(self.l2(x))
        x = self.max_action * torch.tanh(self.l3(x))
        return x


class Critic(nn.Module):
    def __init__(self, state_dim, action_dim):
        super(Critic, self).__init__()
        self.l1 = nn.Linear(state_dim, 400)
        self.l2 = nn.Linear(400 + action_dim, 300)
        self.l3 = nn.Linear(300, 1)

    def forward(self, x, u):
        x = F.relu(self.l1(x))
        x = F.relu(self.l2(torch.cat([x, u], 1)))
        x = self.l3(x)
        return x