CNN预测股票走势基于Tensorflow(思路+程序)

我们希望找出跟随价格上涨的模式。通过每日收盘价，MA，KD，RSI，yearAvgPrice 本次推文研究只是展示深入学习的一个例子。 结果估计不是很好。希望抛砖引玉，给大家带来更多的思考。策略使用的数据从雅虎财务获取。

• 什么时候要买或者卖

• 股票走势预测

这是这个策略的核心思路。我们将数据（年线，月线，收盘价）作为图像，并使用CNN来识别其模式。

• 将图像转换为数组：

• 交易策略：达到+10%或者-5%时卖出。

下面来介绍一下DQN

（参考：知乎 - 智能单元 - Flood Sung）

编写基本DQN类的结构

class DQN():
#初始化
  def __init__(self, env): 
#创建Q网络
  def create_Q_network(self): 
#创建训练方法
  def create_training_method(self): 
#感知存储信息
  def perceive(self,state,action,reward,next_state,done):
#训练网络
  def train_Q_network(self):
#输出带随机的动作
  def egreedy_action(self,state):
#输出动作
  def action(self,state):

我们知道，我们的DQN一个很重要的功能就是要能存储数据，然后在训练的时候minibatch出来。所以，我们需要构造一个存储机制。这里使用deque来实现。

self.replay_buffer = deque()

初始化

def __init__(self, env):
    # init experience replay
    self.replay_buffer = deque()    
    # init some parameters
    self.time_step = 0
    self.epsilon = INITIAL_EPSILON
    self.state_dim = env.observation_space.shape[0]
    self.action_dim = env.action_space.n

    self.create_Q_network()
    self.create_training_method()    # Init session
    self.session = tf.InteractiveSession()
    self.session.run(tf.initialize_all_variables())

这里要注意一点就是egreedy的epsilon是不断变小的，也就是随机性不断变小。怎么理解呢？就是一开始需要更多的探索，所以动作偏随机，慢慢的我们需要动作能够有效，因此减少随机。

创建Q网络

我们这里创建最基本的MLP，中间层设置为20：

def create_Q_network(self):
    # network weights
    W1 = self.weight_variable([self.state_dim,20])
    b1 = self.bias_variable([20])
    W2 = self.weight_variable([20,self.action_dim])
    b2 = self.bias_variable([self.action_dim])    
    # input layer
    self.state_input = tf.placeholder("float",[None,self.state_dim])    
    # hidden layers
    h_layer = tf.nn.relu(tf.matmul(self.state_input,W1) + b1)    
    # Q Value layer
    self.Q_value = tf.matmul(h_layer,W2) + b2  
def weight_variable(self,shape):
    initial = tf.truncated_normal(shape)    
    return tf.Variable(initial)  
def bias_variable(self,shape):
    initial = tf.constant(0.01, shape = shape)    
    return tf.Variable(initial)

只有一个隐层，然后使用relu非线性单元。相信对MLP有了解的知友看上面的代码很easy！要注意的是我们state 输入的格式，因为使用minibatch，所以格式是[None,state_dim]

编写perceive函数

def perceive(self,state,action,reward,next_state,done):
    one_hot_action = np.zeros(self.action_dim)
    one_hot_action[action] = 1
     self.replay_buffer.append((state,one_hot_action,reward,next_state,done))    
    if len(self.replay_buffer) > REPLAY_SIZE:
      self.replay_buffer.popleft()    
    if len(self.replay_buffer) > BATCH_SIZE:
      self.train_Q_network()

这里需要注意的一点就是动作格式的转换。我们在神经网络中使用的是one hot key的形式，而在OpenAI Gym中则使用单值。什么意思呢？比如我们输出动作是1，那么对应的one hot形式就是[0,1]，如果输出动作是0，那么one hot 形式就是[1,0]。这样做的目的是为了之后更好的进行计算。

在perceive中一个最主要的事情就是存储。然后根据情况进行train。这里我们要求只要存储的数据大于Batch的大小就开始训练。

编写action输出函数

def egreedy_action(self,state):
    Q_value = self.Q_value.eval(feed_dict = {
      self.state_input:[state]
      })[0]    
    if random.random() <= self.epsilon:      
       return random.randint(0,self.action_dim - 1)    
    else:      
       return np.argmax(Q_value)
    self.epsilon -= (INITIAL_EPSILON - FINAL_EPSILON)/10000

def action(self,state):
    return np.argmax(self.Q_value.eval(feed_dict = {
      self.state_input:[state]
      })[0])

区别之前已经说过，一个是根据情况输出随机动作，一个是根据神经网络输出。由于神经网络输出的是每一个动作的Q值，因此我们选择最大的那个Q值对应的动作输出。

编写training method函数

  def create_training_method(self):
    self.action_input = tf.placeholder("float",[None,self.action_dim]) # one hot presentation
    self.y_input = tf.placeholder("float",[None])
    Q_action = tf.reduce_sum(tf.mul(self.Q_value,self.action_input),reduction_indices = 1)
    self.cost = tf.reduce_mean(tf.square(self.y_input - Q_action))
    self.optimizer = tf.train.AdamOptimizer(0.0001).minimize(self.cost)

这里的y_input就是target Q值。我们这里采用Adam优化器，其实随便选择一个必然SGD，RMSProp都是可以的。可能比较不好理解的就是Q值的计算。这里大家记住动作输入是one hot key的形式，因此将Q_value和action_input向量相乘得到的就是这个动作对应的Q_value。然后用reduce_sum将数据维度压成一维。

编写training函数

def train_Q_network(self):
    self.time_step += 1
    # Step 1: obtain random minibatch from replay memory
    minibatch = random.sample(self.replay_buffer,BATCH_SIZE)
    state_batch = [data[0] for data in minibatch]
    action_batch = [data[1] for data in minibatch]
    reward_batch = [data[2] for data in minibatch]
    next_state_batch = [data[3] for data in minibatch]    # Step 2: calculate y
    y_batch = []
    Q_value_batch = self.Q_value.eval(feed_dict={self.state_input:next_state_batch})    for i in range(0,BATCH_SIZE):
      done = minibatch[i][4]      if done:
        y_batch.append(reward_batch[i])      else :
        y_batch.append(reward_batch[i] + GAMMA * np.max(Q_value_batch[i]))

    self.optimizer.run(feed_dict={
      self.y_input:y_batch,
      self.action_input:action_batch,
      self.state_input:state_batch
      })

首先就是进行minibatch的工作，然后根据batch计算y_batch。最后就是用optimizer进行优化。

分布式训练

• 需要考虑传输时间

• 比如

  
  

• 实验5层CNN模型

分布式训练在TensorFlow

• 定义一个工作集群{“ps”，“worker”}

• 设置任务（这里我们做数据并行，所以每个任务是一样的）

• 连接时将任务分配给服务器

对于MNIST分布式培训样本，参照以下代码（给出部分）：

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import math
import sys
import re
import tempfile
import timeimport tensorflow as tf
from tensorflow.examples.tutorials.mnist import input_data
from tensorflow.contrib import layers
import tensorflow.contrib.learn as skflow

flags = tf.app.flags
flags.DEFINE_string("data_dir", "./MNIST_data",                    "Directory for storing mnist data")
flags.DEFINE_boolean("download_only", False,                     "Only perform downloading of data; Do not proceed to "
                     "session preparation, model definition or training")#airflags.DEFINE_string("job_name", "","One of 'ps' or 'worker'")'''flags.DEFINE_string("ps_hosts", "10.81.103.124:7777","List of hostname:port for ps jobs."
            "This string should be the same on every host!!")
flags.DEFINE_string("worker_hosts", "10.81.103.124:2224,10.81.103.122:2222,10.81.103.118:2218,10.81.103.119:2219,10.81.103.120:2220,10.81.103.121:2221","List of hostname:port for worker jobs."
            "This string should be the same on every host!!")'''flags.DEFINE_string("ps_hosts", "10.81.103.122:7777","List of hostname:port for ps jobs."
            "This string should be the same on every host!!")
flags.DEFINE_string("worker_hosts", "10.81.103.122:2222,  10.81.103.122:2226","List of hostname:port for worker jobs."
            "This string should be the same on every host!!")

例子1：

• 训练：2011〜2014 15-day 图像只有每日收盘价。

• 测试：2016/01〜2016/08

• 交易策略：回报 =（明天的收盘价） - （今天收盘价）如果预测买入。 反之。

• 表现不佳

例子2：

交易策略：达到+10%或者-5%时卖出。