-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathlearning_agent.py
More file actions
151 lines (146 loc) · 7.03 KB
/
Copy pathlearning_agent.py
File metadata and controls
151 lines (146 loc) · 7.03 KB
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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
import torchbnn as bnn
import torch
import numpy as np
import torch.nn as nn
import torch.nn.functional as F
import random as rand
class DeepQ:
def __init__(self, agent_id, input_size, output_size, hidden_size, learning_rate, epsilon, decay, gamma, trade_limit = 1000, starting_cash = 10000000, bayesian = True):
self.bayesian = bayesian
self.network = None
if bayesian:
self.network = nn.Sequential(
bnn.BayesLinear(prior_mu=0, prior_sigma=10, in_features=input_size, out_features=hidden_size),
nn.Sigmoid(),
bnn.BayesLinear(prior_mu=0, prior_sigma=10, in_features=hidden_size, out_features=output_size),
)
else:
self.network = nn.Sequential(
nn.Linear(in_features=input_size, out_features=hidden_size),
nn.Sigmoid(),
nn.Linear(in_features=hidden_size, out_features=output_size)
)
self.decay = decay
self.trade_limit = trade_limit
self.agent_id = agent_id
self.epsilon = epsilon
self.gamma = gamma
self.mse_loss = nn.MSELoss()
self.kl_loss = bnn.BKLLoss(reduction='mean', last_layer_only=False)
self.kl_weight = 0.01
self.optimizer = torch.optim.Adam(self.network.parameters(), learning_rate)
self.samples = 500
self.num_experiences = 0
self.experiences = np.array([np.array([0.0000001 for i in range(0, 6)]) for j in range(0, self.samples * 2)])
self.cash = starting_cash
self.previous_action = 0
self.previous_state = np.array([0.0, 0.0])
self.shares = 0
self.holdings = 0
self.rewards = 0
self.gather_states_tensor =torch.tensor(np.array([np.array([0,1]) for i in range(0,self.samples)])).type(torch.int64)
self.gather_actions_tensor = torch.tensor(np.array([np.array([2]) for i in range(0,self.samples)])).type(torch.int64)
self.gather_rewards_tensor = torch.tensor(np.array([np.array([3]) for i in range(0,self.samples)])).type(torch.int64)
self.gather_next_states_tensor =torch.tensor(np.array([np.array([4,5]) for i in range(0,self.samples)])).type(torch.int64)
def update_experiences(self, reward, bid_ask):
experience_tuple = [self.previous_state[0],
self.previous_state[1],
self.previous_action,
reward,
bid_ask,
self.shares]
i = self.num_experiences % (2 * self.samples)
self.experiences[i,] = experience_tuple
if i == 0:
self.num_experiences = 0
self.previous_state = np.array([bid_ask, self.shares])
def train(self):
idx = np.random.randint(self.experiences.shape[0], size=self.samples)
sampled_experiences = self.experiences[idx,:]
sampled_experiences = torch.tensor(sampled_experiences)
states = torch.gather(sampled_experiences, 1, self.gather_states_tensor)
actions = torch.gather(sampled_experiences,1, self.gather_actions_tensor)
rewards = torch.gather(sampled_experiences,1, self.gather_rewards_tensor)
next_states = torch.gather(sampled_experiences,1, self.gather_next_states_tensor)
predictions_s = None
predictions_s_prime = None
if self.bayesian:
for i in range(0, 100):
sample_s = self.network(states.type(torch.float32))
sample_s_prime = (self.network(next_states.type(torch.float32)) * self.gamma + rewards)
if predictions_s == None:
predictions_s = sample_s
else:
predictions_s = predictions_s + sample_s
if predictions_s_prime == None:
predictions_s_prime = sample_s_prime
else:
predictions_s_prime = predictions_s_prime + sample_s_prime
predictions_s = predictions_s / 100
predictions_s_prime = predictions_s_prime / 100
else:
predictions_s = self.network(states.type(torch.float32))
predictions_s_prime = (self.network(next_states.type(torch.float32)) * self.gamma + rewards)
q_actions = torch.gather(predictions_s, 1, actions.type(torch.int64))
q_prime_actions = torch.gather(predictions_s_prime, 1, torch.unsqueeze(predictions_s_prime.argmax(1), 1).type(torch.int64))
self.optimizer.zero_grad()
if self.bayesian:
mse = self.mse_loss(q_actions.float(), q_prime_actions.float())
kl = self.kl_loss(self.network) * self.kl_weight
cost = (mse + kl).float()
cost.backward()
self.optimizer.step()
else:
mse = self.mse_loss(q_actions.float(), q_prime_actions.float())
cost = mse
cost.backward()
self.optimizer.step()
def take_action(self, x, random = False):
if random == True:
explore = rand.uniform(0,1)
if explore > self.epsilon:
self.epsilon *= self.decay
with torch.no_grad():
if self.bayesian:
samples = []
mu = None
sigma = torch.tensor(np.array([0.0, 0.0, 0.0]))
for i in range(0, 100):
sample = self.network(torch.tensor(x).type(torch.float32))
samples.append(sample)
if mu == None:
mu = sample
else:
mu = (mu + sample)
mu = mu / 100
for sample in samples:
sigma += (((mu - sample) ** 2) / 99)
if torch.sum(torch.sqrt(sigma)) >= 80.0:
return rand.randint(0,2), 80
return mu.argmax(), torch.sum(torch.sqrt(sigma))
else:
return self.network(torch.tensor(x).type(torch.float32)).argmax()
else:
self.epsilon *= self.decay
if self.bayesian:
return rand.randint(0,2), 80
else:
return rand.randint(0,2)
else:
if self.bayesian:
samples = []
mu = None
sigma = torch.tensor(np.array([0.0, 0.0, 0.0]))
for i in range(0, 100):
sample = self.network(torch.tensor(x).type(torch.float32))
samples.append(sample)
if mu == None:
mu = sample
else:
mu = (mu + sample)
mu = mu / 100
for sample in samples:
sigma += (((mu - sample) ** 2) / 99)
return mu.argmax(), torch.sum(torch.sqrt(sigma))
else:
return self.network(torch.tensor(x).type(torch.float32)).argmax()