最近重看了一遍大作手回忆录，就想能否让机器识别一下 M 顶、 W 底这种相当主观的技术分析。

假设各种技术分析图形可以看作是各种技术分析因子在一段时间内的相对比例变化，

这样可以认为将技术分析因子进行标准化之后不会损失太多的信息。

将因子随交易时间推进的变换看作时间序列 in_length 表示时间序列的长度，

in_width 看作时间序列的宽度也就是并列的多种技术分析因子数值。

不考虑时间推进因素的影响可以使用 CNN 等非时间序列 DNN 处理。

考虑时间推进对因子变化有影响，此时刻因子变化和股价涨跌受到前面状态的影响的时间序列可以使用 HMM 或 RNN 处理。

HMM 常见的假设为状态 i_t 只受到状态 i_{t-1}时刻的影响， RNN 则泛泛认为可以使用训练集来学习到时间序列的非线性关系。

本帖对于 RNN 结构进行简单探索。

定长时间序列

训练 RNN 模型定长序列，假设在 T_i 交易日可以通过前 m 交易日技术分析走势预判后第 n 交易日股价涨跌幅度，并且 RNN 模型可以自动从 T_{i-m}到 T_i 时间序列学习到这种预测关系。

输入数据格式[批次，步长，多因子] 其中步长表示从 T_{i-m}到 T_i 时间序列

class.fit(trainX, trainY)训练模型

clf.pred_prob(trainX) 预测返回概率矩阵

clf.pred_signal(trainX) 预测返回标签

trainX 输入格式 [row, in_length, in_width]

trainY 输入格式 [row]

batch_size=128 喂入批次大小

display_step=5 显示步长

layer_units_num=2000 隐藏层单元数目

training_epoch=100 训练次数

class test_1(object):
    def __init__(self,
                batch_size = 128,
                learning_rate = 0.001,
                training_epoch = 10,
                display_step = 5,
                layer_units_num = 100):
        self.batch_size = batch_size
        self.learning_rate = learning_rate
        self.training_epoch = training_epoch
        self.display_step = display_step
        self.layer_units_num = layer_units_num
        
    def dense_to_one_hot(self,labels_dense):
        """标签 转换 one hot 编码
        输入 labels_dense 必须为非负数
        2016-11-21
        """
        num_classes = len(np.unique(labels_dense)) # np.unique 去掉重复函数
        raws_labels = labels_dense.shape[0]
        index_offset = np.arange(raws_labels) * num_classes
        labels_one_hot = np.zeros((raws_labels, num_classes))
        labels_one_hot.flat[index_offset + labels_dense.ravel()] = 1
        return labels_one_hot  
        
    def Preprocessing(self, trainX, trainY, seed=False):
        trainY = self.dense_to_one_hot(trainY)
        self.in_length= in_length= trainX.shape[1]
        self.in_width= in_width= trainX.shape[2]
        self.out_classes= out_classes= trainY.shape[1]
        
        if seed:
            tf.set_random_seed(20170204)
        weights = {
            'out': tf.Variable(tf.truncated_normal(shape=[self.layer_units_num, out_classes], 
                                                mean=0., stddev=1., seed=None, dtype=tf.float32),
                               trainable=True, name='Weight_full_out')
        }
        
        biases = {
            'out': tf.Variable(tf.truncated_normal([out_classes]), trainable=True, name= 'Biases_full_out')
        }    
        
        self.weights = weights
        self.biases = biases        
        
        X = tf.placeholder(dtype=tf.float32, shape=[None, in_length, in_width], name='trainX') # 批次，时间序列，多因子
        Y = tf.placeholder(dtype= tf.float32, shape=[None, out_classes], name='trainY') 
        keep_prob = tf.placeholder(dtype= tf.float32)
        self.X = X
        self.Y = Y
        self.keep_prob = keep_prob  
        
    def Network(self):
        
        with tf.name_scope('layer_1'):            
            monolayer_1 = tf.nn.rnn_cell.BasicLSTMCell(num_units= self.layer_units_num, 
                                                       forget_bias=1., state_is_tuple=True, activation=tf.tanh)            
            monolayer_1 = tf.nn.rnn_cell.DropoutWrapper(cell=monolayer_1, output_keep_prob= keep_prob)

        
        with tf.name_scope('layer_2'):        
            monolayer_2 = tf.nn.rnn_cell.BasicLSTMCell(num_units= self.layer_units_num,
                                                       forget_bias=1., state_is_tuple=True, activation=tf.tanh)
            monolayer_2 = tf.nn.rnn_cell.DropoutWrapper(cell=monolayer_2, output_keep_prob= keep_prob)
        
        with tf.name_scope('layer_3'):                  
            monolayer_3 = tf.nn.rnn_cell.BasicLSTMCell(num_units= self.layer_units_num,
                                                       forget_bias=1., state_is_tuple=True, activation=tf.tanh)
            monolayer_3 = tf.nn.rnn_cell.DropoutWrapper(cell=monolayer_3, output_keep_prob= keep_prob)
        
        with tf.name_scope('layer_Final'):
            monolayer_final = tf.nn.rnn_cell.BasicLSTMCell(num_units=self.layer_units_num, 
                                                       forget_bias=1., state_is_tuple=True, activation=tf.tanh)
        
        with tf.name_scope('Layer_Structure_Combination'):
            layer_units_num = self.layer_units_num
            Layers = tf.nn.rnn_cell.MultiRNNCell(cells=[monolayer_1,monolayer_2,monolayer_3,monolayer_final],
                                                state_is_tuple = True)    
        self.Layers = Layers
        return Layers
        
    def Model(self):
        X = self.X
        keep_prob = self.keep_prob
        
        X = tf.transpose(X, [1, 0, 2])
        X = tf.reshape(X, [-1,self.in_width])
        X = tf.split(split_dim=0, num_split=self.in_length, value=X)
        
        with tf.name_scope('layer_1'):            
            monolayer_1 = tf.nn.rnn_cell.BasicLSTMCell(num_units= self.layer_units_num, 
                                                       forget_bias=1., state_is_tuple=True, activation=tf.tanh)            
            monolayer_1 = tf.nn.rnn_cell.DropoutWrapper(cell=monolayer_1, output_keep_prob= keep_prob)

        
        with tf.name_scope('layer_2'):        
            monolayer_2 = tf.nn.rnn_cell.BasicLSTMCell(num_units= self.layer_units_num,
                                                       forget_bias=1., state_is_tuple=True, activation=tf.tanh)
            monolayer_2 = tf.nn.rnn_cell.DropoutWrapper(cell=monolayer_2, output_keep_prob= keep_prob)
        
        with tf.name_scope('layer_3'):                  
            monolayer_3 = tf.nn.rnn_cell.BasicLSTMCell(num_units= self.layer_units_num,
                                                       forget_bias=1., state_is_tuple=True, activation=tf.tanh)
            monolayer_3 = tf.nn.rnn_cell.DropoutWrapper(cell=monolayer_3, output_keep_prob= keep_prob)
        
        with tf.name_scope('layer_Final'):
            monolayer_final = tf.nn.rnn_cell.BasicLSTMCell(num_units=self.layer_units_num, 
                                                       forget_bias=1., state_is_tuple=True, activation=tf.tanh)
        
        with tf.name_scope('Layer_Structure_Combination'):
            layer_units_num = self.layer_units_num
            Layers = tf.nn.rnn_cell.MultiRNNCell(cells=[monolayer_1,monolayer_2,monolayer_3,monolayer_final],
                                                state_is_tuple = True)    
        
        outputs,_ = tf.nn.rnn(cell=monolayer_final, inputs=X, dtype=tf.float32)
        output = outputs[-1]

        return tf.nn.bias_add(value= tf.matmul(output, self.weights['out']), bias= self.biases['out'])  
        
    def train(self, trainX, trainY, seed=False):
        self.sess = tf.InteractiveSession()
        
        self.Preprocessing(trainX, trainY, seed)
        
        tmp = self.Model()
        
        self.predict = tf.nn.softmax(tmp)
        
        self.cost = tf.reduce_sum(tf.nn.softmax_cross_entropy_with_logits(tmp, self.Y))
        
        optimizer = tf.train.AdamOptimizer(learning_rate= self.learning_rate) # 0 设置训练器
        grads_and_vars = optimizer.compute_gradients(self.cost)
        for i, (grid, var) in enumerate(grads_and_vars):
            if grid != None:
                grid = tf.clip_by_value(grid, -1., 1.)
                grads_and_vars[i] = (grid, var)
        optimizer = optimizer.apply_gradients(grads_and_vars)
        self.optimizer = optimizer
        
        self.correct_pred = tf.equal(tf.argmax(tmp,1), tf.argmax(self.Y,1))
        accuracy = tf.reduce_mean(tf.cast(self.correct_pred, tf.float32))
        self.accuracy = accuracy
        
        #self.init = tf.global_variables_initializer()   
        self.init = tf.initialize_all_variables()
    def fit(self,trainX, trainY, dropout = 0.3, seed=True):
        self.train(trainX, trainY, seed=True)
        sess = self.sess
        sess.run(self.init)
        batch_size = self.batch_size
        trainY = self.dense_to_one_hot(trainY)
        for ep in range(self.training_epoch):
            for i in range(int(len(trainX)/batch_size)+1):
                if i < int(len(trainX)/batch_size)+1:
                    batch_x = trainX[i*batch_size : (i+1)*batch_size]
                    batch_y = trainY[i*batch_size : (i+1)*batch_size]
                elif i== int(len(trainX)/batch_size)+1:
                    batch_x = trainX[-batch_size:]
                    batch_y = trainY[-batch_size:]
                sess.run(self.optimizer, feed_dict={self.X:batch_x, self.Y:batch_y, self.keep_prob:(1.-dropout)})
            if ep%self.display_step==0:                
                loss, acc = sess.run([self.cost,self.accuracy], feed_dict={self.X:trainX, self.Y:trainY, self.keep_prob:1.})
                print (str(ep)+"th "+'Epoch Loss = {:.5f}'.format(loss)+" Training Accuracy={:.5f}".format(acc))
        self.sess= sess
        print("Optimization Finished!") 
    
    def pred_prob(self, testX):
        sess = self.sess
        batch_size = self.batch_size
        trainX = testX
        predict_output = np.zeros([1,self.out_classes])
        for i in range(int(len(trainX)/batch_size)+1):
            if i < int(len(trainX)/batch_size)+1:
                batch_x = trainX[i*batch_size : (i+1)*batch_size]
                batch_y = trainY[i*batch_size : (i+1)*batch_size]
            elif i== int(len(trainX)/batch_size)+1:
                batch_x = trainX[-batch_size:]
                batch_y = trainY[-batch_size:]
            tp = sess.run(self.predict,feed_dict={self.X:batch_x, self.keep_prob:1.})
            predict_output = np.row_stack([predict_output, tp])
        predict_output = np.delete(predict_output, obj=0, axis=0)
        return predict_output
    
    def pred(self, testX):
        pred_prob = self.pred_prob(testX)
        return np.argmax(pred_prob, axis=1)

变长时间序列

训练 RNN 模型定长序列，假设在 T_i 交易日可以通过前 m 交易日技术分析走势预判后第 n 交易日股价涨跌幅度， RNN 模型可以从 T_{i-m}到 T_i 时间序列学习到这种预测关系，但是无法训练得到准确的稀疏权重。对应 T 交易日进行预判的时候考虑前 2 周走势或者前 2 个月的走势，这个时候模型无法隐式学习，需要指定训练集为变长序列对应涨跌标签。

输入 对于变长时间序列 设格式为

[batch_size, real_length, in_width]

其中 real_length 表示变长时间序列输入步长，为一个变化值，使用 LSTM 进行预测的时候将输入数据修改格式为

batch_size, in_length, in_width

其中 in_length 使用 real_length 中的最大步长值，空余部分使用 0 填充

修改原始稠密矩阵为稀疏矩阵统一格式进行输入。

输出裁剪 由于输入 tensor 为稀疏矩阵，则对应的 RNN 网络计算得到的矩阵为稀疏矩阵（对于全为 0 的填充稀疏部分进行数值优化的时候梯度为 0 实际不变化）

将拓扑结构图得到的矩阵进行裁剪，使得输出 tensor 格式从

[batch_size, in_length, layer_units_num]

转换为

[batch_size, 1, layer_units_num]

softmax [batch_size, out_classes]

这里 1 表示这里对输入步长取实际输入步长最后一步。

import functools
from functools import reduce  
import numpy as np
import tensorflow as tf

def lazy_property(function):
    attribute = '_' + function.__name__

    @property
    @functools.wraps(function)
    def wrapper(self):
        if not hasattr(self, attribute):
            setattr(self, attribute, function(self))
        return getattr(self, attribute)
    return wrapper

class test(object):
    def __init__(self,
                batch_size = 128,
                learning_rate = 0.001,
                error = .01,
                display_step = 5,
                layer_units_num = 200):
        self.batch_size = batch_size
        self.learning_rate = learning_rate
        self.error = error
        self.display_step = display_step
        self.layer_units_num = layer_units_num
        
    def dense_to_one_hot(self,labels_dense):
        """标签 转换 one hot 编码
        输入 labels_dense 必须为非负数
        2016-11-21
        """
        num_classes = len(np.unique(labels_dense)) # np.unique 去掉重复函数
        raws_labels = labels_dense.shape[0]
        index_offset = np.arange(raws_labels) * num_classes
        labels_one_hot = np.zeros((raws_labels, num_classes))
        labels_one_hot.flat[index_offset + labels_dense.ravel()] = 1
        return labels_one_hot  
    
    # 获得权重和偏置
    @staticmethod
    def _weight_and_bias(in_size, out_size):
        weight = tf.truncated_normal([in_size, out_size], stddev=0.01)
        bias = tf.constant(0.1, shape=[out_size])
        return tf.Variable(weight), tf.Variable(bias)
    
    @lazy_property
    def length(self):
        dense_sign = tf.sign(tf.reduce_max(tf.abs(self.X),reduction_indices=2))
        length = tf.reduce_sum(input_tensor=dense_sign, reduction_indices=1)
        length = tf.cast(length, tf.int32)
        return length
    
    @staticmethod
    def _final_relevant(output, length):
        # length 输入时间序列的实际长度
        # in_length 表示输入时间序列长度
        # max_length 表示最大时间序列长度,也就是稀疏矩阵 最大时间序列长度
        batch_size = tf.shape(output)[0]
        max_length = int(output.get_shape()[1])
        output_size = int(output.get_shape()[2])
        index = tf.range(start=0, limit=batch_size)*max_length + (length-1) # 这里使用 max_length 开创间隔，使用 length-1 表示实际位置，最后一个输出的位置
        flat = tf.reshape(output, [-1,output_size]) # 将输出展平， batch_size*length in_width
        relevant = tf.gather(flat, index) # 根据实际长度选出最后一个输出 output 状态使用
        return relevant  
    
    def Preprocessing(self, trainX, trainY):
        self.in_length= in_length= trainX.shape[1]
        self.in_width= in_width= trainX.shape[2]
        self.out_classes= out_classes= trainY.shape[1]
        
        self.X = tf.placeholder(dtype=tf.float32, shape=[None, in_length, in_width], name='trainX') # 批次，时间序列，多因子
        self.Y = tf.placeholder(dtype= tf.float32, shape=[None, out_classes], name='trainY') 
        self.keep_prob = tf.placeholder(dtype= tf.float32)
    
    def str2float(self,s):  
        def char2num(s):  
            return {'0': 0, '1': 1, '2': 2, '3': 3, '4': 4, '5': 5, '6': 6, '7': 7, '8': 8, '9': 9}[s]  
        n = s.index('.')  
        return reduce(lambda x,y:x*10+y,map(char2num,s[:n]+s[n+1:]))/(10**n)  
    
    def Interface(self):        
        # 4 层 GRU 结构描述
        monolayer = tf.nn.rnn_cell.GRUCell(num_units= self.layer_units_num)
        monolayer = tf.nn.rnn_cell.DropoutWrapper(cell=monolayer, output_keep_prob=self.keep_prob)
        monolayer_final = tf.nn.rnn_cell.GRUCell(num_units= self.layer_units_num)
        layers = tf.nn.rnn_cell.MultiRNNCell([monolayer]*3+[monolayer_final])
        # 激活 注意 in_length 表示输入序列步长， length 表示实际步长
        output,_ = tf.nn.dynamic_rnn(cell= layers, inputs= self.X, dtype= tf.float32, sequence_length= self.length)        
        output = self._final_relevant(output, self.length)
        
        weights, biases = self._weight_and_bias(self.layer_units_num, self.out_classes)        
        Prediction = tf.nn.bias_add(tf.matmul(output, weights),biases)
        return Prediction
    
    def Graph(self, trainX, trainY):
        try:
            tf.InteractiveSession.close()
        except:
            pass
        self.sess = tf.InteractiveSession()
        tf.get_default_session()
        self.Preprocessing(trainX, trainY)
        tmp = self.Interface()
        
        self.pred = tf.nn.softmax(tmp)
        self.cost = tf.reduce_sum(tf.nn.softmax_cross_entropy_with_logits(tmp, self.Y))
        
        optimizer = tf.train.AdamOptimizer(learning_rate= self.learning_rate) # 0 设置训练器
        grads_and_vars = optimizer.compute_gradients(self.cost)
        for i, (grid, var) in enumerate(grads_and_vars):
            if grid != None:
                grid = tf.clip_by_value(grid, -1., 1.)
                grads_and_vars[i] = (grid, var)
        optimizer = optimizer.apply_gradients(grads_and_vars)
        self.optimizer = optimizer
        
        self.correct_pred = tf.equal(tf.argmax(tmp,1), tf.argmax(self.Y,1))
        self.accuracy = tf.reduce_mean(tf.cast(self.correct_pred, tf.float32))
        #self.init = tf.global_variables_initializer()
        self.init = tf.initialize_all_variables()
    def fit(self, trainX, trainY, dropout= 0.618):
        # 对标签 one_hot 编码
        trainY = self.dense_to_one_hot(trainY)
        
        self.Graph(trainX, trainY)
        self.sess.run(self.init)
        batch_size = self.batch_size
        sig =10.
        ep = 0
        while (sig > self.error):
            for i in range(int(len(trainX)/batch_size)+1):
                if i < int(len(trainX)/batch_size)+1:
                    batch_x = trainX[i*batch_size : (i+1)*batch_size]
                    batch_y = trainY[i*batch_size : (i+1)*batch_size]
                elif i== int(len(trainX)/batch_size)+1:
                    batch_x = trainX[-batch_size:]
                    batch_y = trainY[-batch_size:]
                self.sess.run(self.optimizer,feed_dict={self.X:batch_x, self.Y:batch_y, self.keep_prob:(1.-dropout)})
            sig = self.sess.run(self.accuracy, feed_dict={self.X:trainX, self.Y:trainY, self.keep_prob:1.})
            if ep%self.display_step==0:                
                loss = self.sess.run(self.cost, feed_dict={self.X:trainX, self.Y:trainY, self.keep_prob:1.})
                print (str(ep)+"th "+'Epoch Loss = {:.5f}'.format(loss)+" Training Accuracy={:.5f}".format(sig))
            ep += 1
        print("Optimization Finished!")                
        
    def pred_prob(self, testX):
        batch_size = self.batch_size
        trainX = testX
        predict_output = np.zeros([1,self.out_classes])
        for i in range(int(len(trainX)/batch_size)+1):
            if i < int(len(trainX)/batch_size)+1:
                batch_x = trainX[i*batch_size : (i+1)*batch_size]
                batch_y = trainY[i*batch_size : (i+1)*batch_size]
            elif i== int(len(trainX)/batch_size)+1:
                batch_x = trainX[-batch_size:]
                batch_y = trainY[-batch_size:]
            tp = self.sess.run(self.pred, feed_dict={self.X:batch_x, self.keep_prob:1.})
            predict_output = np.row_stack([predict_output, tp])
        predict_output = np.delete(predict_output, obj=0, axis=0)
        return predict_output
    
    def pred_signal(self, testX):
        pred_prob = self.pred_prob(testX)
        return np.argmax(pred_prob, axis=1)

因限制原因，图片与源代码戳这里： https://uqer.io/community/share/589d3cc2c1e3cc00567fdbea