5. Reinforcement Learning - Q Learning

DQL Agent

from collections import deque
from tensorflow.keras.optimizers import Adam, RMSprop

class DQLAgent:
    def __init__(self, gamma=0.95, hu=24, opt=Adam, # hu가 노드의 수 opt: optimizer의 약자
           lr=0.001, finish=False): # lr이 머지. . . ? => ask
        self.finish = finish
        # epsilon => 탐색(무작위)을 할 비율. 무작위를 얼마나 할거냐?
        self.epsilon = 1.0  # 1) 초기 탐색(무작위) 비율 
        self.epsilon_min = 0.01 # 2) 최소 탐색 비율 # 무작위가 약간은 있어야함. 0 이면 실행 자체를 안 하기 때문.
        # 0.01이 되면 강화 학습이 끝난 것. 
        # 강화학습은 무작위가 없는 모델 만드는 것. 신경망 통해 학습해서 결론을 예측해야.
        self.epsilon_decay = 0.995 # 3) 탐색 비율에 대한 감소 비율 (무작위를 얼마나 감소시킬거냐) 
        # => 0.5프로씩 줄여나가겠다
        self.gamma = gamma # 4) 지연 보상에 대한 할인율
        self.batch_size = 32 # 5) 재실행 배치 사이즈
        self.max_treward = 0
        self.averages = list() # 평균
        self.memory = deque(maxlen=2000) # 6) 제한된 기록용 데큐
        # que(자료구조 메모리) 
        self.osn = env.observation_space.shape[0] # state
        # 아직 env (cartpole). env의 리셋하면 state가 나오는데 그것이 observation
        # 상태가 4개 들어있음. 
        self.model = self._build_model(hu, opt, lr)
        
    def _build_model(self, hu, opt, lr): # 카트폴 메모리가 4개
        model = Sequential()
        model.add(Dense(hu, input_dim=self.osn,
                        activation='relu'))
        model.add(Dense(hu, activation='relu'))
        model.add(Dense(env.action_space.n, activation='linear')) #action_space.n이 2개. 출력층이 2개 나옴. 선형하게 나옴.
        model.compile(loss='mse', optimizer=opt(lr=lr)) # loss는 mse (회귀로 나옴. 분류 아님.) 
        return model
        
    def act(self, state): 
        if random.random() <= self.epsilon: # random을 epsilon 값만큼 준다(give).
            return env.action_space.sample()
        action = self.model.predict(state)[0]  # 출력층이 2개 나왔으므로 확률이 나옴
        return np.argmax(action) # action에서 2개 값이 나옴. 0 or 1이 나옴. 최대인 index가 나옴. 0 아님 1.
    
    def replay(self): # random 에 sample 메모리 안에서 batch 사이즈의 상태를 다시 가져옴. 
        # 재실행할 땐 이전 것을 다시 가져와야함.
        # replay의 핵심. 원래 메모리에서 가져오는 것. self.batch_size = 32이기때문에 32번 가져옴.
        batch = random.sample(self.memory, self.batch_size) # 7) 재실행을 위한 history batch 무작위 선택
        for state, action, reward, next_state, done in batch:
            if not done:
                reward += self.gamma * np.amax(
                    self.model.predict(next_state)[0]) # 8) 상태-행위값에 대한 Q 값
            target = self.model.predict(state)
            target[0, action] = reward
            self.model.fit(state, target, epochs=1, # 학습 시킴.
                           verbose=False) # 9) 새로운 생태-행위값에 대해 신경망 업데이트
        if self.epsilon > self.epsilon_min: # epsilon min보다 작으면 epsilon decay해서 곱한다.
            self.epsilon *= self.epsilon_decay # 10) 탐색 비율 업데이터
    
    def learn(self, episodes): # episode 만큼 돌린다.
        trewards = []
        for e in range(1, episodes + 1):
            state = env.reset()
            state = np.reshape(state[0], [1, self.osn]) # self.osn은 4번 돌아감.
            for i in range(5000): # step을 5000번 하겠다.
                action = self.act(state)
                next_state, reward, done, _,_ = env.step(action)
                next_state = np.reshape(next_state,
                                        [1, self.osn])
                self.memory.append([state, action, reward,
                                     next_state, done]) # 11) 메모리 append가 끝났는지 안 끝났는지 까지. 
                # 상태들을 기억하고 가겠다. => 상태들을 학습.
                #
                state = next_state
                # state = np.reshape(state, [1, self.osn])
                if done: # 끝났으면 계싼
                    treward = i + 1
                    trewards.append(treward)
                    av = sum(trewards[-25:]) / 25
                    self.averages.append(av)
                    self.max_treward = max(self.max_treward, treward)
                    templ = 'episode: {:4d}/{} | treward: {:4d} | '
                    templ += 'av: {:6.1f} | max: {:4d}'
                    print(templ.format(e, episodes, treward, av,
                                       self.max_treward), end='\r')
                    break
            if av > 195 and self.finish: # av 값이 finish 값보다 큰지
                print()
                break
            if len(self.memory) > self.batch_size: # memory의 크기를 비교해서 replay를 하는 것. replay가 핵심
                self.replay() # 12) 지나간 경험에 기반하여 정책을 업데이트하기 위한 재실행
    def test(self, episodes):
        trewards = []
        for e in range(1, episodes + 1):
            state = env.reset()
            state = np.reshape(state[0], [1, self.osn])
            for i in range(5001):
                action = np.argmax(self.model.predict(state)[0])
                next_state, reward, done, _,_ = env.step(action)
                state = next_state
                state = np.reshape(state, [1, self.osn])
                if done:
                    treward = i + 1
                    trewards.append(treward)
                    print('episode: {:4d}/{} | treward: {:4d}'
                          .format(e, episodes, treward), end='\r')
                    break
        return trewards

env = Finance('EUR=', 'EUR=')

env.reset()

a = env.action_space.sample()
a

env.step(a)

set_seeds(100)
agent = DQLAgent(gamma=0.5, opt=RMSprop)