-
Notifications
You must be signed in to change notification settings - Fork 5
/
DuelingDDQN_Agent.py
111 lines (91 loc) · 4.27 KB
/
DuelingDDQN_Agent.py
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
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
# -*- coding: utf-8 -*-
"""
Created on Sat Jul 9 12:17:31 2022
@author: Abhilash
"""
import random
from collections import deque
import numpy as np
from tensorflow.keras import Sequential
from tensorflow.keras.callbacks import TensorBoard
from tensorflow.keras.layers import Dense
from tensorflow.keras.models import load_model
from tensorflow.keras.optimizers import Adam
import tensorflow as tf
from utils import Portfolio
class DuelingDDQN_Agent(Portfolio):
def __init__(self, state_dim, balance, is_eval=False):
super().__init__(balance=balance)
self.model_type = 'DDQN'
self.state_dim = state_dim
self.action_dim = 3 # hold, buy, sell
self.memory = deque(maxlen=100)
self.buffer_size = 60
self.tau = 0.0001
self.gamma = 0.95
self.epsilon = 1.0 # initial exploration rate
self.epsilon_min = 0.01 # minimum exploration rate
self.epsilon_decay = 0.995 # decrease exploration rate as the agent becomes good at trading
self.is_eval = is_eval
self.model = self.model()
self.model_target =self.model
self.model_target.set_weights(self.model.get_weights()) # hard copy model parameters to target model parameters
self.tensorboard = TensorBoard(log_dir='./logs/DuelingDDQN_tensorboard', update_freq=90)
self.tensorboard.set_model(self.model)
def _huber_loss(self, y_true, y_pred, clip_delta=1.0):
error = y_true - y_pred
cond = tf.keras.backend.abs(error) <= clip_delta
squared_loss = 0.5 * tf.keras.backend.square(error)
quadratic_loss = 0.5 * tf.keras.backend.square(clip_delta) + clip_delta * (tf.keras.backend.abs(error) - clip_delta)
return tf.keras.backend.mean(tf.where(cond, squared_loss, quadratic_loss))
def update_model_target(self):
model_weights = self.model.get_weights()
model_target_weights = self.model_target.get_weights()
for i in range(len(model_weights)):
model_target_weights[i] = self.tau * model_weights[i] + (1 - self.tau) * model_target_weights[i]
self.model_target.set_weights(model_target_weights)
def model(self):
backbone = tf.keras.Sequential([
tf.keras.layers.Input((self.state_dim,)),
Dense(32, activation='relu'),
Dense(16, activation='relu')
])
state_input = tf.keras.layers.Input((self.state_dim,))
backbone_1 = Dense(32, activation='relu')(state_input)
backbone_2 = Dense(16, activation='relu')(backbone_1)
value_output = Dense(1)(backbone_2)
advantage_output = Dense(self.action_dim)(backbone_2)
output = tf.keras.layers.Add()([value_output, advantage_output])
model = tf.keras.Model(state_input, output)
#model.compile(loss='mse', optimizer=Adam(lr=0.001))
model.compile(loss=self._huber_loss, optimizer=Adam(lr=0.001))
return model
def reset(self):
self.reset_portfolio()
self.epsilon = 1.0
def remember(self, state, actions, reward, next_state, done):
print("Append state to memory ")
self.memory.append((state, actions, reward, next_state, done))
def act(self, state):
if not self.is_eval and np.random.rand() <= self.epsilon:
print("random value is less than epsilon")
return random.randrange(self.action_dim)
options = self.model.predict(state)
return np.argmax(options[0])
def experience_replay(self):
print("Select online from buffer")
mini_batch = random.sample(self.memory, self.buffer_size)
for state, actions, reward, next_state, done in mini_batch:
if not done:
print("Target Q value not attained")
Q_value = reward + (1 - done) * self.gamma * np.amax(self.model_target.predict(next_state)[0])
else:
print("Target Q value attained")
Q_value=reward
next_actions = self.model.predict(state)
next_actions[0][np.argmax(actions)] = Q_value
history = self.model.fit(state, next_actions, epochs=1, verbose=2)
self.update_model_target()
if self.epsilon > self.epsilon_min:
self.epsilon *= self.epsilon_decay
return history.history['loss'][0]