tensorflow 笔记2 双向LSTM
数据文件颇大 有空再传
BasicLSTM¶
%%time
from __future__ import division
from __future__ import print_function
import numpy as np
import pandas as pd
import matplotlib.pylab as plt
%matplotlib inline
import seaborn as sns
import tensorflow as tf
from tensorflow.python.ops import rnn, rnn_cell
fac = np.load('/home/big/Quotes/TensorFlow deal with Uqer/fac16.npy').astype(np.float32)
ret = np.load('/home/big/Quotes/TensorFlow deal with Uqer/ret16.npy').astype(np.float32)
# 数据格式 日期-多因子 例如 （09-01 Ab1 Ab2 Ab3 ）（09-02 Ab1 Ab2 Ab3）
# Parameters
learning_rate = 0.001
batch_size = 1024
training_iters = int(fac.shape[0]/batch_size)
display_step = 10
# Network Parameters
n_input = 17
n_steps = 40
n_hidden = 1024
n_classes = 7
# tf Graph input
x = tf.placeholder('float',[None, n_steps, n_input])
y = tf.placeholder('float',[None, n_classes])
# Define weights
weights = {
'out': tf.Variable(tf.random_normal([n_hidden, n_classes]))
}
biases = {
'out': tf.Variable(tf.random_normal([n_classes]))
}
def BasicLSTM(x, weights, biases):
x = tf.transpose(x, [1, 0, 2])
x = tf.reshape(x, [-1,n_input])
x = tf.split(0, n_steps, x)
# 这一段不用注意，因为使用CNN提取的npy数据这里进行数据处理，转换成格式为
# 日期-一批次数据（多只股票）-多因子数据，相当于将多只股票的多因子数据以时间序列一天一天喂给RNN模型
Basicl_LSTM_cell = tf.nn.rnn_cell.BasicLSTMCell(n_hidden, forget_bias=1.0)
outputs, states = tf.nn.rnn(Basicl_LSTM_cell, x, dtype=tf.float32)
return tf.matmul(outputs[-1], weights['out']) + biases['out']
pred = BasicLSTM(x, weights, biases)
cost = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits(pred, y))
optimizer = tf.train.AdamOptimizer(learning_rate=learning_rate).minimize(cost)
correct_pred = tf.equal(tf.argmax(pred,1), tf.argmax(y,1))
accuracy = tf.reduce_mean(tf.cast(correct_pred, tf.float32))
init = tf.global_variables_initializer()
# Launch the graph
with tf.Session() as sess:
sess.run(init)
step = 1
for step in range(1):
for i in range(int(len(fac)/batch_size)):
batch_x = fac[i*batch_size:(i+1)*batch_size].reshape([batch_size,n_steps,n_input])
batch_y = ret[i*batch_size:(i+1)*batch_size].reshape([batch_size,n_classes])
sess.run(optimizer,feed_dict={x:batch_x,y:batch_y})
if i % display_step ==0:
print(i,'----',(int(len(fac)/batch_size)))
loss, acc = sess.run([cost, accuracy], feed_dict={x: batch_x,y: batch_y})
print("Iter " + str(step*batch_size) + ", Minibatch Loss= " + \
"{:.6f}".format(loss) + ", Training Accuracy= " + \
"{:.5f}".format(acc))
print("Optimization Finished!")
# Calculate accuracy for 128 mnist test images
test_len = 1280
test_data = fac[:test_len].reshape([batch_size,n_steps,n_input])
test_label = ret[:test_len].reshape([batch_size,n_classes])
print("Testing Accuracy:", \
sess.run(accuracy, feed_dict={x: test_data, y: test_label}))
sess.close()
双向LSTM¶
%%time
from __future__ import division
from __future__ import print_function
import numpy as np
import pandas as pd
import matplotlib.pylab as plt
%matplotlib inline
import seaborn as sns
import tensorflow as tf
fac = np.load('/home/big/Quotes/TensorFlow deal with Uqer/fac16.npy').astype(np.float32)
ret = np.load('/home/big/Quotes/TensorFlow deal with Uqer/ret16.npy').astype(np.float32)
# 数据格式 日期-多因子 例如 （09-01 Ab1 Ab2 Ab3 ）（09-02 Ab1 Ab2 Ab3）
# Parameters
learning_rate = 0.001
training_iters = 100000
batch_size = 1280
display_step = 10
# Network Parameters
n_input = 17 # MNIST data input (img shape: 28*28)
n_steps = 40 # timesteps
n_hidden = 128 # hidden layer num of features
n_classes = 7 # MNIST total classes (0-9 digits)
# tf Graph input
x = tf.placeholder("float", [None, n_steps, n_input])
y = tf.placeholder("float", [None, n_classes])
# Define weights
weights = {
# Hidden layer weights => 2*n_hidden because of forward + backward cells
'out': tf.Variable(tf.random_normal([2*n_hidden, n_classes]))
}
biases = {
'out': tf.Variable(tf.random_normal([n_classes]))
}
def BiRNN(x, weights, biases):
# Prepare data shape to match `bidirectional_rnn` function requirements
# Current data input shape: (batch_size, n_steps, n_input)
# Required shape: 'n_steps' tensors list of shape (batch_size, n_input)
# Permuting batch_size and n_steps
x = tf.transpose(x, [1, 0, 2])
# Reshape to (n_steps*batch_size, n_input)
x = tf.reshape(x, [-1, n_input])
# Split to get a list of 'n_steps' tensors of shape (batch_size, n_input)
x = tf.split(0, n_steps, x)
# Define lstm cells with tensorflow
# Forward direction cell
lstm_fw_cell = tf.nn.rnn_cell.BasicLSTMCell(n_hidden, forget_bias=1.0)
# Backward direction cell
lstm_bw_cell = tf.nn.rnn_cell.BasicLSTMCell(n_hidden, forget_bias=1.0)
# Get lstm cell output
try:
outputs, _, _ = tf.nn.bidirectional_rnn(lstm_fw_cell, lstm_bw_cell, x,
dtype=tf.float32)
except Exception: # Old TensorFlow version only returns outputs not states
outputs = tf.nn.bidirectional_rnn(lstm_fw_cell, lstm_bw_cell, x,
dtype=tf.float32)
# Linear activation, using rnn inner loop last output
return tf.matmul(outputs[-1], weights['out']) + biases['out']
pred = BiRNN(x, weights, biases)
# Define loss and optimizer
cost = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits(pred, y))
optimizer = tf.train.AdamOptimizer(learning_rate=learning_rate).minimize(cost)
# Evaluate model
correct_pred = tf.equal(tf.argmax(pred,1), tf.argmax(y,1))
accuracy = tf.reduce_mean(tf.cast(correct_pred, tf.float32))
# Initializing the variables
init = tf.global_variables_initializer()
# Launch the graph
with tf.Session() as sess:
sess.run(init)
step = 1
for step in range(100):
for i in range(int(len(fac)/batch_size)):
batch_x = fac[i*batch_size:(i+1)*batch_size].reshape([batch_size,n_steps,n_input])
batch_y = ret[i*batch_size:(i+1)*batch_size].reshape([batch_size,n_classes])
sess.run(optimizer,feed_dict={x:batch_x,y:batch_y})
if i % display_step ==0:
print(i,'----',(int(len(fac)/batch_size)))
loss, acc = sess.run([cost, accuracy], feed_dict={x: batch_x,y: batch_y})
print("Iter " + str(step*batch_size) + ", Minibatch Loss= " + \
"{:.6f}".format(loss) + ", Training Accuracy= " + \
"{:.5f}".format(acc))
print("Optimization Finished!")
# Calculate accuracy for 128 mnist test images
test_len = 1280
test_data = fac[:test_len].reshape([batch_size,n_steps,n_input])
test_label = ret[:test_len].reshape([batch_size,n_classes])
print("Testing Accuracy:", \
sess.run(accuracy, feed_dict={x: test_data, y: test_label}))
sess.close()
