This post illustrates CNN LSTM Models that can be fit to simple time series data.

Simulate Time Series Data

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import numpy as np
import matplotlib.pyplot as plt

def test_set1(seed,n): np.random.seed(seed) X = np.random.normal(size=(n,6)) Y = X[:,0:3].sum(axis=1) return X,Y

def test_set2(seed,n): np.random.seed(seed) X = np.random.normal(size=(n,9)) Y1 = X[:,0:3].sum(axis=1) Y2 = X[:,3:6].sum(axis=1) Y3 = X[:,6:].sum(axis=1) Y = np.c_[Y1,Y2,Y3] return X,Y, Y1, Y2, Y3

def test_set3(seed,n): np.random.seed(seed) X = np.random.normal(size=(n,9,2)) Y = X.sum(axis=1).sum(axis=1) return X,Y

Plain Vanilla LSTM - Multiple Input - Single Output

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
from keras.models import Sequential
from keras.layers import Dense, LSTM
model = Sequential()
model.add(LSTM(50, activation='relu', input_shape=(X.shape[1], X.shape[2])))
model.add(Dense(1))
model.compile(optimizer='adam', loss='mse')
model.summary()
model.fit(X,Y, epochs=100)

X,Y = test_set1(4321,100) X = X.reshape(X.shape[0], X.shape[1], 1) model.evaluate(X,Y)

y_pred = model.predict(X) plt.scatter(y_pred, Y) plt.show() #+end_src>

Stacked LSTM - Multiple Input - Single Output #+begin_src python from keras.models import Sequential from keras.layers import Dense, LSTM model = Sequential() model.add(LSTM(50, return_sequences=True,activation='relu', input_shape=(X.shape[1], X.shape[2]))) model.add(LSTM(50)) model.add(Dense(1)) model.compile(optimizer='adam', loss='mse') model.summary()

X,Y = test_set1(1234,1000) X = X.reshape(X.shape[0], X.shape[1], 1)

model.fit(X,Y, epochs=100)

X,Y = test_set1(4321,100) X = X.reshape(X.shape[0], X.shape[1], 1) model.evaluate(X,Y)

y_pred = model.predict(X) plt.scatter(y_pred, Y) plt.show()

Bidirectional LSTM - Multiple Input - Single Output

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
from keras.models import Sequential
from keras.layers import Dense, LSTM, Bidirectional
model = Sequential()
model.add(Bidirectional(LSTM(50,activation='relu'), input_shape=(X.shape[1], X.shape[2])))
model.add(Dense(1))
model.compile(optimizer='adam', loss='mse')
model.summary()

X,Y = test_set1(1234,1000) X = X.reshape(X.shape[0], X.shape[1], 1) model.fit(X,Y, epochs=100)

X,Y = test_set1(4321,100) X = X.reshape(X.shape[0], X.shape[1], 1) model.evaluate(X,Y)

y_pred = model.predict(X) plt.scatter(y_pred, Y) plt.show()

LSTM Model - Vector Output

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from keras.models import Sequential
from keras.layers import Conv1D, MaxPooling1D, Flatten, Dense
from keras.layers import Dense, LSTM
n = 1000
X,Y,_,_,_        = test_set2(1234,n)
X                = X.reshape(X.shape[0], X.shape[1], 1)

model = Sequential() model.add(LSTM(50, return_sequences=True,activation='relu', input_shape=(X.shape[1], X.shape[2]))) model.add(LSTM(50)) model.add(Dense(3)) model.compile(optimizer='adam', loss='mse') model.summary()

model.fit(X,Y, epochs=100)

X,Y,,,_ = test_set2(4321,100) X = X.reshape(X.shape[0], X.shape[1], 1) model.evaluate(X,Y)

y_pred = model.predict(X) plt.scatter(y_pred, Y) plt.show()

LSTM Model - Encoder - Decoder Output

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
from keras.models import Sequential
from keras.layers import Conv1D, MaxPooling1D, Flatten, Dense, RepeatVector, TimeDistributed
from keras.layers import Dense, LSTM
n = 1000
X,Y,_,_,_        = test_set2(1234,n)
X                = X.reshape(X.shape[0], X.shape[1], 1)
Y = Y.reshape(Y.shape[0],Y.shape[1],1)

from keras.model import Model from keras.layers import Input

model =Sequential() model.add(LSTM(100, activation='relu', input_shape=(X.shape[1], X.shape[2]))) model.add(RepeatVector(3)) model.add(LSTM(100, return_sequences=True,activation='relu')) model.add(TimeDistributed(Dense(1))) model.compile(optimizer='adam', loss='mse') model.summary()

model.fit(X,Y, epochs=100) n = 100 X,Y,,,_ = test_set2(4321,n) X = X.reshape(X.shape[0], X.shape[1], 1) Y = Y.reshape(Y.shape[0],Y.shape[1],1)

model.evaluate(X,Y)

y_pred = model.predict(X) y_pred.shape

plt.scatter(y_pred[:,0,:], Y[:,0,:]) plt.show()

plt.scatter(y_pred[:,1,:], Y[:,1,:]) plt.show()

plt.scatter(y_pred[:,2,:], Y[:,2,:]) plt.show()

Takeaway

Need to use this architecture to replicate Deep Portfolio