利用强化学习进行量化投资的尝试

选题及概述

代码将会上传到我的GitHub

写在前面

选题概述

本文希望通过机器学习算法，设立相应的环境，选取真实的数据，使得程序能够预判风险进行投资，使得投资能够最大化。

本文的经济学部分参考了论文：ML-TEA 一套基于机器学习和技术分析的量化投资算法 李斌，主要通过其经济学方面的数据，筛选出了主要的4种状态变量以及经济指标。

创新部分

根据股票投资制度，本文创建了简易的投资模型以及相应的环境，设计并实现了相应的算法库。

在传统的经济量化领域，或者说股票投资和机器学习的交叉领域，程序常常被用来实现对股票趋势的预测，而得知股票趋势概率后，投资的具体行为却是由人自己控制的，这便意味着量化过程并非全部由程序实现，在面对概率等因素下，人所固有的情感不仅会影响对股票趋势的判断，也会影响在判断完趋势和概率后，对股票持有情况的选择。因此，本文不同于其它希望预测股票趋势，然后简单地涨买跌卖的论文，而是着重于将整个过程作为一种“游戏”全部交给程序执行，可能在最开始，程序将不能学会如何购买股票，但经过长时间大量的训练后，它的表现可能优于人类。

比如说，假如程序预测一支股票涨的概率是0.6，跌的概率是0.4，那么最后对股票的决策仍然是由人来完成的。人需要结合程序预测的准确度，历史信息，还有涨跌预测的幅度来判断信息，并尽力让自己最后赚钱。那何不让机器来完成这项工作呢——即让机器作为行动方，独立完成股票的预测和买卖，并最终通过训练，让自己有着赚钱的能力。

程序思想

机器学习部分（程序主体）

程序采用ActorCritic的思路，将整个股票交易视为一场游戏，程序在“买入”，“不动”，和“卖出”三种动作中选择，并得到系统给予的reward，根据reward，训练C网络（评价网络），然后用TD-error训练A网络（动作网络）。

选择AC的原因在于，该程序在预测股票趋势的基础上增加了选择的过程，但这样的选择不一定马上能改变当前的reward，因此使用动作-评委网络，可以改善程序歪打正着的情况，以及平衡概率和选择的矛盾（冒险与否）。

相关程序被集成为ACSDK.py下的Actor类和Critic类，并由主程序main.py调用。

相关代码如下：

class Actor(object):
    def __init__(self, sess, n_features, n_actions, lr=0.001):
        self.sess = sess

        self.s = tf.compat.v1.placeholder(tf.float32, [1, n_features], "state")
        self.a = tf.compat.v1.placeholder(tf.int32, None, "act")
        self.td_error = tf.compat.v1.placeholder(tf.float32, None, "td_error")  # TD_error

        with tf.compat.v1.variable_scope('Actor'):
            l1 = tf.compat.v1.layers.dense(
                inputs=self.s,
                units=20,  # number of hidden units
                activation=tf.nn.relu,
                kernel_initializer=tf.random_normal_initializer(0., .1),  # weights
                bias_initializer=tf.constant_initializer(0.1),  # biases
                name='l1'
            )

            self.acts_prob = tf.compat.v1.layers.dense(
                inputs=l1,
                units=n_actions,  # output units
                activation=tf.nn.softmax,  # get action probabilities
                kernel_initializer=tf.random_normal_initializer(0., .1),  # weights
                bias_initializer=tf.constant_initializer(0.1),  # biases
                name='acts_prob'
            )

        with tf.compat.v1.variable_scope('exp_v'):
            log_prob = tf.compat.v1.log(self.acts_prob[0, self.a])
            self.exp_v = tf.reduce_mean(log_prob * self.td_error)  # advantage (TD_error) guided loss

        with tf.compat.v1.variable_scope('train'):
            self.train_op = tf.compat.v1.train.AdamOptimizer(lr).minimize(-self.exp_v)  # minimize(-exp_v) = maximize(exp_v)

    def learn(self, s, a, td):
        s = s[np.newaxis, :]
        feed_dict = {self.s: s, self.a: a, self.td_error: td}
        _, exp_v = self.sess.run([self.train_op, self.exp_v], feed_dict)
        return exp_v

    def choose_action(self, s):
        s = s[np.newaxis, :]
        probs = self.sess.run(self.acts_prob, {self.s: s}) 
        return np.random.choice(np.arange(probs.shape[1]), p=probs.ravel())  # return a int

class Critic(object):
    def __init__(self, sess, n_features, lr=0.01):
        self.sess = sess

        self.s = tf.compat.v1.placeholder(tf.float32, [1, n_features], "state")
        self.v_ = tf.compat.v1.placeholder(tf.float32, [1, 1], "v_next")
        self.r = tf.compat.v1.placeholder(tf.float32, None, 'r')

        with tf.compat.v1.variable_scope('Critic'):
            l1 = tf.compat.v1.layers.dense(
                inputs=self.s,
                units=20,  
                activation=tf.nn.relu,  
                kernel_initializer=tf.random_normal_initializer(0., .1),  # weights
                bias_initializer=tf.constant_initializer(0.1),  # biases
                name='l1'
            )

            self.v = tf.compat.v1.layers.dense(
                inputs=l1,
                units=1,  # output units
                activation=None,
                kernel_initializer=tf.random_normal_initializer(0., .1),  # weights
                bias_initializer=tf.constant_initializer(0.1),  # biases
                name='V'
            )

        with tf.compat.v1.variable_scope('squared_TD_error'):
            self.td_error = self.r + GAMMA * self.v_ - self.v
            self.loss = tf.compat.v1.square(self.td_error)  # TD_error = (r+gamma*V_next) - V_eval
        with tf.compat.v1.variable_scope('train'):
            self.train_op = tf.compat.v1.train.AdamOptimizer(lr).minimize(self.loss)

    def learn(self, s, r, s_):
        s, s_ = s[np.newaxis, :], s_[np.newaxis, :]

        v_ = self.sess.run(self.v, {self.s: s_})
        td_error, _ = self.sess.run([self.td_error, self.train_op],
                                    {self.s: s, self.v_: v_, self.r: r})
        return td_error,v_

以及主函数：

from turtle import color
from ACSDK import Actor,Critic 
import matplotlib.pyplot as plt
import numpy as np
import tensorflow as tf

import Environment
import getShock

GAMMA = 0.1  # 衰变值
LR_A = 0.0001  # Actor学习率
LR_C = 0.5  # Critic学习率

N_F = 4  # 状态空间
N_A = 3  # 动作空间

def main():
    # 数据设定
    stock_code = '000037'
    trade_cost = 15/10000

    #getShock.get(stock_code)

    sess = tf.compat.v1.Session()
    actor = Actor(sess, n_features=N_F, n_actions=N_A, lr=LR_A)  # 初始化Actor
    critic = Critic(sess, n_features=N_F, lr=LR_C)  # 初始化Critic
    sess.run(tf.compat.v1.global_variables_initializer())  # 初始化参数

    env = Environment.Env(stock_code, trade_cost)
    obv, reward, done = env.init_module()
    rewards = []
    td_errors = []
    vs = []
    print(obv)
    while(True):
        action = actor.choose_action(obv)
        obv_, reward, done = env.step(action)
        rewards.append(reward)
        td_error,v_1 = critic.learn(obv, reward, obv_)
        td_errors.append(td_error[0])
        vs.append(v_1[0])
        actor.learn(obv, action, td_error)
        obv = obv_
        #print(reward)
        if (done):break
    
    x = np.linspace(0,env.readLines()-1,env.readLines())
    plt.plot(x,np.array(rewards),color = 'blue')
    plt.plot(x,np.array(vs),color = 'black')
    plt.plot(x,100 * np.array(env.readData())[1:],color = 'red')
    plt.plot(x,100 * np.array(env.readStocks())[0:],color = 'yellow')
    plt.show()



if __name__ == "__main__":
    main()

环境编写

为该人工智能配置一个具有真实股票环境的反应系统，主要负责接收动作，读取当前股价，根据规则返回对应的 reward ，一次迭代时间为一天。

关于激励reward的设计，采用赚取的总金额作为激励，同时为避免程序采取消极措施，reward在刚开始加上了负偏置。即从负值开始，每天可以选择“买进”一单位股票，“卖出”一单位股票或者不动，然后以当天的收盘价扣除现金或获得现金，并以第二天的开盘价格算出现金加上股票的总价作为激励，即：$$ reward = stock * price + cash$$

在状态的设计上，本实验参考了论文：ML-TEA 一套基于机器学习和技术分析的量化投资算法 李斌，其中提出了17种观测指标，但这些指标均由4个状态量得出，即：“开盘价格”，“收盘价格”，“最高价”和“最低价”。为了保持程序的简介易收敛，考虑到它们的关系可以通过神经网络进行拟合，所以可以只采用这四个指标构成状态。

同时，为了模拟真实的股票交易市场，每次对股票的转换（买进或者卖出），都会收取万分之十五的手续费用，即转移部分价值减少万分之十五。此值与真实的股票交易情况类似但略大，可以起到避免系统无限制买卖的作用。

每次迭代，环境都会接收一次程序的动作，然后获取并输出当前的状态变量和计算出的reward，然后进入下一次迭代。

相关程序被集成为Environment.py下的Env类，通过调用自定义的函数实现读取。

相关代码如下：

# 环境搭建
class Env:
    def __init__(self,stock_code, tc):
        self.code = stock_code
        self.data = pd.read_csv('D:\\'+ stock_code + '.csv',encoding="gb2312")
        #self.data = pd.read_csv('D:\\'+ stock_code + '.csv',encoding="utf-8")
        # print(self.data)
        self.lines = 0
        self.tc = tc
        self.maxPrice = 0
        self.money = 0
        self.stocks = []
        self.stock = 0
    
    # 重新开始
    def init_module(self):
        if (self.lines != 0):
            for i in range(self.lines - 1):
                obv = self.observation(i)
                obv = obv[np.newaxis, :]
        obv = self.observation(self.lines)
        rew = 0
        done = 0
        return obv, rew ,done

    # 计算读取各种经济数据
    def observation(self,lines):
        o = self.data['开盘'].values[lines]
        c = self.data['收盘'].values[lines]
        l = self.data['最低'].values[lines]
        h = self.data['最高'].values[lines]
        self.maxPrice = max(c,self.maxPrice)
        return np.array((o, c, l, h))

    # 为限定模型，设置每次只能买或卖，且一次只能买卖一股,0买入,1卖出,2不动
    # 决定买入时，将会以当天的收盘价买入，reward以第二天开盘价计算
    def step(self, action):
        if (action == 0):
            self.stock += 1
            self.money -= self.data['收盘'].values[self.lines]*(1+self.tc)
        elif (action == 1):
            self.stock -= 1
            self.money += self.data['收盘'].values[self.lines]*(1-self.tc)
        self.lines += 1
        self.stocks.append(self.stock)
        rew = self.stock * self.data['开盘'].values[self.lines] + self.money
        #rew = rew - self.maxPrice * self.lines
        obv = self.observation(self.lines)
        if (self.lines >= len(self.data['开盘'].values) - 1): done = 1
        else: done = 0
        return obv, rew ,done
    
    def readLines(self):
        return self.lines

    def readStocks(self):
        return self.stocks

    def readData(self):
        return self.data['开盘'].values

股票获取

股票信息采用akshare开放端口获取，其具体步骤被集合为文件getShock下的get函数，运行该过程将在D盘创建相应股票代号历史信息的.csv文件，主程序中也需要通过运行此类读取股票数据。

在正常的股票函数之外，还有一个特殊的股票生成文件specialShock.py，用于生成特殊的股票函数，方便调试和测验，如生成代号为0的直线上涨的函数，每次运行时同样在D盘生成目标文件。

需要注意的是，由于格式问题，读取时特殊股票和正常股票所需要的文本格式分别是utf-8和gb2312，这需要在Environment.py下的Env类的init函数下修改。

相关代码如下：

正常股票：

# 用于拉取股票历史数据

import pandas as pd
import akshare as ak

def get(stock_code):
    # 参数调整
    start_date = '20190101'
    end_date = '20210101'
    period = 'daily'
    adj = 'hfq'

    # 获取数据

    data = ak.stock_zh_a_hist(symbol=stock_code,period = period, start_date=start_date, end_date=end_date, adjust = adj)

    data.to_csv('D:\\'+ stock_code + '.csv',index = False, mode = 'w', encoding = 'gbk')

    print("-------- Get Data Done -------")

特殊股票：

# 特殊数据集验证
import pandas as pd
import numpy as np

def make():
    # 参数调整
    stock_code = '0'

    # 获取数据

    data = []
    for i in range(100):
        data.append([i,i+1,i,i+1])
    data = np.array(data)
    name = ['开盘','收盘','最低','最高']
    files = pd.DataFrame(data,columns = name)
    files.to_csv('D:\\'+ stock_code + '.csv')

    print("-------- make Data Done -------")

if __name__ == "__main__":
    make()

结论与展望

可以看到，赋予程序买卖的权力，而非只让它们预测股票的思路有一定的实际价值和应用前景，但它还需要更为精确的数据和改进，鉴于时间关系，本文只能对已经存在的问题和解决办法进行展望而无法实操。

该系统的问题首先在于C网络和实际reward的收敛性并不好，而当前系统的不稳定几乎全是因为C网络收敛失败造成的。本质原因在于股票的难以预测上，即：C网络本质上是在预测股票并评定价值，所以，在C网络种加入RNN乃至LSTM结构是必要的，增强它对连续数据的预测能力才能增强C网络预测reward的能力。

其次，在环境的设定上，每次只能买卖一股是一种理想而保守的情况，因此，将动作的选取从3选1调整为一个正态分布的随机抽取是合理的，而A输出的值决定着这个正态分布的参数，而通过正态分布抽取到的值将被作为买进或卖出的股票份额是更加符合实际的选择。

最后，在价值的评判上，单纯的以赚取的数目作为reward可能不是一个最好的选择，因为在一支一直上升的股票中，即使系统只买入了一点，其reward也会一直上升，也就是说，当系统选择保守但没有风险的策略时，也会一直受到鼓励，但过于保守将无钱可赚：系统可能选择“躺平”。因此，我们可以借鉴之前偏置的思路，使用一个参数来调整reward的阈值，使得没有赚到某个数目或者赚到能赚到的最高数目的一定比例时，激励为负。

使用“游戏”的方式研究股票的操作是一种罕见的视角，因为它不满足于传统的对股票的预测，而是加入对概率的考量，采用更为科学的视角观察股票。正如打牌时，仅仅知道各种情况的概率远远不够，能够根据这些条件做出判断更为重要。随着数据的积累和模型的优化，笔者认为利用机器学习完成股票预测将不是问题。