AI量化（代码）：深度强化学习DRL应用于金融量化

原创文章第93篇，专注“个人成长与财富自由、世界运作的逻辑， AI量化投资”。

今天要说说强化学习。

强化学习个人认为，是最契合金融投资的范式。它其实不是一个具体的算法，而是一种范式。

与监督学习不同，监督学习是“一次性”的，有明确的标签。而类比人类的学习，我们不会拿1万张猫的图片让孩子学习什么是猫，一个小朋友也许只见过几只猫，他几乎可以认出各种图片的，抽象的，卡通的猫。人脑的学习是激励-反馈机制。你做一个动作，得到一个正反馈就可以强化这个动作，如果得到负反馈，则弱化这个动作。

我们学骑自行车，并不会用物理学的公式，去计算，人体和地面的角度，手臂要如何发力，而是骑就是了，自行车和地面给你反馈，你随之调整。

简言之，智能体通过与环境交互，得到即时反馈，来调整自己的动作，直到学会某个技能。——这样的学习范式的通过的。所以有了alpha go， alpha master和alpha zero这样的围棋大师，然后它们的技能迁移的打战略游戏上，也是一等一的好手，都拜强化学习所赐。

昨天我们讲的DNN的模式，我们是让模型学习过去的因子“预测”当天的收盘价的“涨跌”。但现实交易过程中，我们就不会因为判断今天会涨就买入，而明天会跌就卖出吧。

投资是一个系列的决策，寻找是一段时间内的收益率，而不是某一天或某几天的涨与跌。这种多轮博弈，而且带有成本的多轮博弈，使用强化学习再适合不过了。

01 强化学习

导入一般常规配置：

import os

import math
import random
import numpy as np
import pandas as pd
from pylab import plt, mpl
plt.style.use('seaborn')
mpl.rcParams['savefig.dpi'] = 300
mpl.rcParams['font.family'] = 'serif'
np.set_printoptions(precision=4, suppress=True)
os.environ['PYTHONHASHSEED'] = '0'

实现一个观察空间类，这是一个N维向量，n的维度等于我们因子的个数。

class observation_space:
    def __init__(self, n):
        self.shape = (n,)

动作空间，这里动作空间是2（买，卖）

class action_space:
    def __init__(self, n):
        self.n = n
    def seed(self, seed):
        pass
    def sample(self):
        return random.randint(0, self.n - 1)

模似交易环境：

交易环境除了准备交易数据之外，最重要的就是step函数，我们这里的reward仍然与昨天的DNN类似，智能体猜中一次，则奖励加1分。

准确率低于47.5%时退出，或者一个episode结束时退出。

相当于我们做一个交易游戏，一个智能体进来“猜”，猜中加1分，计算准确率，低于47.5%直接出局。

class Finance:
    url = 'http://hilpisch.com/aiif_eikon_eod_data.csv'
    def __init__(self, symbol, features):
        self.symbol = symbol
        self.features = features
        self.observation_space = observation_space(4)
        self.osn = self.observation_space.shape[0]
        self.action_space = action_space(2)
        self.min_accuracy = 0.475
        self._get_data()
        self._prepare_data()
    def _get_data(self):
        self.raw = pd.read_csv(self.url, index_col=0,
                               parse_dates=True).dropna()
    def _prepare_data(self):
        self.data = pd.DataFrame(self.raw[self.symbol])
        self.data['r'] = np.log(self.data / self.data.shift(1))
        self.data.dropna(inplace=True)
        self.data = (self.data - self.data.mean()) / self.data.std()
        self.data['d'] = np.where(self.data['r'] > 0, 1, 0)
    def _get_state(self):
        return self.data[self.features].iloc[
            self.bar - self.osn:self.bar].values
    def seed(self, seed=None):
        pass
    def reset(self):
        self.treward = 0
        self.accuracy = 0
        self.bar = self.osn
        state = self.data[self.features].iloc[
            self.bar - self.osn:self.bar]
        return state.values
    def step(self, action):
        correct = action == self.data['d'].iloc[self.bar]
        reward = 1 if correct else 0
        self.treward += reward
        self.bar += 1
        self.accuracy = self.treward / (self.bar - self.osn)
        if self.bar >= len(self.data):
            done = True
        elif reward == 1:
            done = False
        elif (self.accuracy < self.min_accuracy and
              self.bar > self.osn + 10):
            done = True
        else:
            done = False
        state = self._get_state()
        info = {}
        return state, reward, done, info

智能体的实现DRL:

def set_seeds(seed=100):
    random.seed(seed)
    np.random.seed(seed)
    env.seed(seed)


set_seeds(100)

from collections import deque
from keras.optimizers import Adam, RMSprop
from keras.layers import Dense, Dropout
from keras.models import Sequential
from keras.optimizers import Adam, RMSprop
from sklearn.metrics import accuracy_score
from tensorflow.python.framework.ops import disable_eager_execution

disable_eager_execution()


class DQLAgent:
    def __init__(self, gamma=0.95, hu=24, opt=Adam,
                 lr=0.001, finish=False):
        self.finish = finish
        self.epsilon = 1.0
        self.epsilon_min = 0.01
        self.epsilon_decay = 0.995
        self.gamma = gamma
        self.batch_size = 32
        self.max_treward = 0
        self.averages = list()
        self.memory = deque(maxlen=2000)
        self.osn = env.observation_space.shape[0]
        self.model = self._build_model(hu, opt, lr)

    def _build_model(self, hu, opt, lr):
        model = Sequential()
        model.add(Dense(hu, input_dim=self.osn,
                        activation='relu'))
        model.add(Dense(hu, activation='relu'))
        model.add(Dense(env.action_space.n, activation='linear'))
        model.compile(loss='mse', optimizer=opt(lr=lr))
        return model

    def act(self, state):
        if random.random() <= self.epsilon:
            return env.action_space.sample()
        action = self.model.predict(state)[0]
        return np.argmax(action)

    def replay(self):
        batch = random.sample(self.memory, self.batch_size)
        for state, action, reward, next_state, done in batch:
            if not done:
                reward += self.gamma * np.amax(
                    self.model.predict(next_state)[0])
            target = self.model.predict(state)
            target[0, action] = reward
            self.model.fit(state, target, epochs=1,
                           verbose=False)
        if self.epsilon > self.epsilon_min:
            self.epsilon *= self.epsilon_decay

    def learn(self, episodes):
        trewards = []
        for e in range(1, episodes + 1):
            state = env.reset()
            state = np.reshape(state, [1, self.osn])
            for _ in range(5000):
                action = self.act(state)
                next_state, reward, done, info = env.step(action)
                next_state = np.reshape(next_state,
                                        [1, self.osn])
                self.memory.append([state, action, reward,
                                    next_state, done])
                state = next_state
                if done:
                    treward = _ + 1
                    trewards.append(treward)
                    av = sum(trewards[-25:]) / 25
                    self.averages.append(av)
                    self.max_treward = max(self.max_treward, treward)
                    templ = 'episode: {:4d}/{} | treward: {:4d} | '
                    templ += 'av: {:6.1f} | max: {:4d}'
                    print(templ.format(e, episodes, treward, av,
                                       self.max_treward), end='\r')
                    break
            if av > 195 and self.finish:
                print()
                break
            if len(self.memory) > self.batch_size:
                self.replay()

    def test(self, episodes):
        trewards = []
        for e in range(1, episodes + 1):
            state = env.reset()
            for _ in range(5001):
                state = np.reshape(state, [1, self.osn])
                action = np.argmax(self.model.predict(state)[0])
                next_state, reward, done, info = env.step(action)
                state = next_state
                if done:
                    treward = _ + 1
                    trewards.append(treward)
                    print('episode: {:4d}/{} | treward: {:4d}'
                          .format(e, episodes, treward), end='\r')
                    break
        return trewards

重点关注learn函数：

每个episode跑5000次，遇到失败会退出。DRL有重放机制，这个我们系统讲强化学习算法的时候再说。

训练这个模型花了22分钟。

02 DRL应用于投资实战

样本内的表现还是不错的，我们再来看样本外的表现情况：

还可以，注意这里未考虑手续费。

小结：

本文重在演示强化学习如何作用于金融时间序列数据，这个方向一定是可行，而且应该深入探讨的。

AI量化实证：使用DNN(深度学习模型)在指数上获得超额收益

极简代码演示“向量化”回测——双均线多空策略

知识星球