RLTrader의 학습기 모듈 개발
2020-04-02 • rltrader • stock, 주식투자, reinforcement learning, rl, 강화학습, learner, 학습기 • 21 min read
학습기 모듈(learners.py)은 다양한 강화학습 방식을 수행하기 위한 학습기 클래스들을 가지는 모듈입니다. 각 학습기 클래스의 속성과 함수를 살펴보고 파이썬으로 구현한 소스 코드를 상세히 확인합니다.
학습기 모듈의 주요 속성과 함수
학습기 모듈은 ReinforcementLearner 클래스를 기본으로 다양한 강화학습 방법을 구현한 클래스를 가집니다. ReinforcementLearner 클래스의 주요 속성 및 함수는 다음과 같습니다.
속성
- stock_code: 강화학습 대상 주식 종목 코드
- chart_data: 주식 종목의 차트 데이터
- environment: 강화학습 환경 객체
- agent: 강화학습 에이전트 객체
- training_data: 학습 데이터
- value_network: 가치 신경망
- policy_network: 정책 신경망
함수
- init_value_network(): 가치 신경망 생성 함수
- init_policy_network(): 정책 신경망 생성 함수
- build_sample(): 환경 객체에서 샘플을 획득하는 함수
- get_batch(): 배치 학습 데이터 생성 함수
- update_networks(): 가치 신경망 및 정책 신경망 학습 함수
- fit(): 가치 신경망 및 정책 신경망 학습 요청 함수
- visualize(): 에포크 정보 가시화 함수
- run(): 강화학습 수행 함수
- save_models(): 가치 신경망 및 정책 신경망 저장 함수
코드 조각 1: 학습기 모듈의 의존성 임포트
다음은 학습기 모듈의 의존성 임포트 부분을 보여줍니다.
learners 모듈의 의존 패키지, 모듈, 클래스 임포트 부분
import os
import logging
import abc
import collections
import threading
import time
import numpy as np
from utils import sigmoid
from environment import Environment
from agent import Agent
from networks import Network, DNN, LSTMNetwork, CNN
from visualizer import Visualizer
파이썬 기본 모듈인 os, logging, abc, collections, threading, time을 임포트합니다. os는 폴더 생성이나 파일 경로 준비 등을 위해 사용합니다. logging은 학습 과정 중에 정보를 기록하기 위해 사용합니다. abc는 추상 클래스를 정의하기 위해 사용합니다. time은 학습 시간을 측정하기 위해 사용합니다.
파이썬 팁: abc는 추상 베이스 클래스(abstract base class)의 약자로 추상 클래스 정의를 도와줍니다.
@abstractmethod데코레이터를 사용해 추상 메서드를 선언할 수 있습니다. abc에 대한 더 자세한 설명은 https://docs.python.org/ko/3/library/abc.html을 참고하기 바랍니다.
배열 자료구조를 조작하기 위해 파이썬 라이브러리인 NumPy를 임포트합니다. 그리고 이어서 RLTrader의 모듈들을 임포트합니다. 이전 장에서 다룬 environment, agent, networks, visualizer 모듈을 모두 임포트합니다. 정책 신경망 학습 레이블을 생성하기 위해 utils에 정의한 sigmoid 함수를 임포트합니다.
파이썬 팁: 임포트에 사용하는 키워드는 import, from, as입니다. import는 패키지, 모듈, 클래스, 함수를 임포트하기 위한 키워드입니다. from은 임포트할 모듈의 상위 패키지나 임포트할 클래스의 상위 모듈, 임포트할 함수의 상위 모듈을 지정하기 위한 키워드입니다. as 키워드는 임포트한 패키지, 모듈, 클래스, 함수를 다른 이름으로 사용하기 위한 키워드입니다.
코드 조각 2: 학습기 클래스의 생성자
다음은 ReinforcementLearner 클래스의 생성자 초반부를 보여줍니다. ReinforcemetLearner 클래스는 DQNLearner, PolicyGradientLearner, ActorCriticLearner, A2CLearner 클래스가 상속하는 상위 클래스입니다. ReinforcementLearner 클래스 생성자는 강화학습에 필요한 환경, 에이전트, 신경망 인스턴스들과 학습 데이터를 속성으로 가집니다.
ReinforcementLearner 클래스: 생성자 (1)
class ReinforcementLearner:
__metaclass__ = abc.ABCMeta
lock = threading.Lock()
def __init__(self, rl_method='rl', stock_code=None,
chart_data=None, training_data=None,
min_trading_unit=1, max_trading_unit=2,
delayed_reward_threshold=.05,
net='dnn', num_steps=1, lr=0.001,
value_network=None, policy_network=None,
output_path='', reuse_models=True):
# 인자 확인
assert min_trading_unit > 0
assert max_trading_unit > 0
assert max_trading_unit >= min_trading_unit
assert num_steps > 0
assert lr > 0
# 강화학습 기법 설정
self.rl_method = rl_method
# 환경 설정
self.stock_code = stock_code
self.chart_data = chart_data
self.environment = Environment(chart_data)
# 에이전트 설정
self.agent = Agent(self.environment,
min_trading_unit=min_trading_unit,
max_trading_unit=max_trading_unit,
delayed_reward_threshold=delayed_reward_threshold)
rl_method는 강화학습 기법을 의미합니다. 이 값은 하위 클래스에 따라 달라집니다. DQNLearner는 "dqn", PolictGradientLearner는 "pg", ActorCriticLearner는 "ac", A2CLearner는 "a2c", A3CLearner는 "a3c"로 정해집니다. stock_code는 학습을 진행하는 주식 종목 코드입니다. rl_method와 stock_code는 학습 과정에 꼭 필요한 것은 아니며, 로그 기록에 주로 쓰입니다.
chart_data는 강화학습의 환경에 해당하는 주식 일봉 차트 데이터입니다. training_data는 학습을 위한 전처리된 학습 데이터입니다. min_trading_unit과 max_trading_unit은 각각 투자 최소 및 최대 단위입니다. 주식 종목마다 주가의 스케일이 달라 투자 주식 수 단위가 다르기 때문에 적절한 투자 단위를 설정하는 것이 중요합니다.
delayed_reward_threshold는 지연 보상 임곗값입니다. 수익률이나 손실률이 이 임곗값보다 클 경우 지연 보상이 발생해 이전 행동들에 대한 학습이 진행됩니다.
mini_batch_size는 미니 배치 학습을 위한 크기입니다. mini_batch_size만큼 데이터가 쌓이는 데도 포트폴리오 가치가 크게 변하지 않아서 지연 보상이 발생하지 않을 때 바로 학습을 수행합니다.
신경망 종류는 net 인자로 받습니다. net 인자는 "dnn", "lstm", "cnn" 등의 값이 될 수 있으며 이 값에 따라서 가치 신경망과 정책 신경망으로 사용할 신경망 클래스가 달라집니다.
n_steps는 LSTM, CNN 신경망에서 사용하는 샘플 묶음의 크기입니다. lr은 학습 속도로, 이 값이 너무 크면 학습이 제대로 진행되지 않으며 너무 작으면 학습이 너무 오래 걸립니다.
value_network와 policy_network가 인자로 들어오면 이 모델들을 가치 신경망과 정책 신경망으로 사용합니다. 학습 과정에서 발생하는 로그, 가시화 결과 및 학습 종료 후 저장되는 신경망 모델 파일은 output_path에 지정된 경로에 저장됩니다.
강화학습 환경을 생성하기 위해 차트 데이터 chart_data를 인자로 하여 Environment 클래스의 인스턴스를 생성합니다. 환경은 차트 데이터를 순서대로 읽으면서 주가, 거래량 등의 데이터를 제공합니다. 그리고 이 강화학습 환경을 인자로 Agent 클래스의 인스턴스를 생성합니다. 주식투자 단위와 지연 보상 임곗값도 에이전트 생성 인자로 넣어줍니다.
파이썬 팁: assert는 가정 설정문입니다. 다음과 같이 assert 뒤에 조건을 넣어줍니다.
assert <condition>
condition이 만족하지 않으면AssertionError예외를 발생시킵니다.
다음 코드조각에서는 ReinforcementLearner 클래스의 생성자 소스 코드를 살펴봅니다.
ReinforcementLearner 클래스: 생성자 (2)
# 학습 데이터
self.training_data = training_data
self.sample = None
self.training_data_idx = -1
# 벡터 크기 = 학습 데이터 벡터 크기 + 에이전트 상태 크기
self.num_features = self.agent.STATE_DIM
if self.training_data is not None:
self.num_features += self.training_data.shape[1]
# 신경망 설정
self.net = net
self.num_steps = num_steps
self.lr = lr
self.value_network = value_network
self.policy_network = policy_network
self.reuse_models = reuse_models
# 가시화 모듈
self.visualizer = Visualizer()
학습 데이터는 학습에 사용할 특징(Feature)들을 포함합니다. 이 특징들은 4.2.1절에서 상세하게 다뤘습니다. 학습에 사용할 최종 특징 개수는 학습 데이터에 포함된 26개의 특징과 에이전트의 상태인 2개 특징을 더해서 28개입니다. ReinforcementLearner 클래스는 이 특징들을 학습하기 위한 신경망 클래스 객체를 가지고 있습니다. 신경망 클래스 객체는 ReinforcementLearner 클래스의 하위 클래스들이 생성합니다. reuse_models가 True이고 모델이 이미 존재하는 경우 재활용합니다.
파이썬 팁: NumPy 배열은 배열의 모양을 의미하는 shape 변수를 가집니다. N차원 배열이면 shape 변수는 N차원 튜플입니다. 예를 들어 1차원 배열의 shape는 1차원 튜플, 2차원 배열의 shape는 2차원 튜플입니다. [ [1, 2, 3], [4, 5, 6] ]의 shape는 (2, 3)입니다.
학습 과정을 가시화하기 위해 Visualizer 클래스 객체를 생성합니다. 이 객체를 통해 에포크마다 투자 결과를 가시화합니다.
다음은 ReinforcementLearner 클래스 생성자의 마지막 부분입니다.
ReinforcementLearner 클래스: 생성자 (3)
# 메모리
self.memory_sample = []
self.memory_action = []
self.memory_reward = []
self.memory_value = []
self.memory_policy = []
self.memory_pv = []
self.memory_num_stocks = []
self.memory_exp_idx = []
self.memory_learning_idx = []
# 에포크 관련 정보
self.loss = 0.
self.itr_cnt = 0
self.exploration_cnt = 0
self.batch_size = 0
self.learning_cnt = 0
# 로그 등 출력 경로
self.output_path = output_path
강화학습 과정에서 발생하는 각종 데이터를 쌓아두기 위해 momory_*라는 이름의 변수들을 가지고 있습니다. 학습 데이터 샘플, 수행한 행동, 획득한 보상, 행동의 예측 가치, 행동의 예측 확률, 포트폴리오 가치, 보유 주식 수, 탐험 위치, 학습 위치 등을 저장합니다. 이렇게 저장한 샘플, 보상 등의 데이터로 신경망 학습을 진행합니다.
강화학습을 진행하면서 에포크 관련 정보도 쌓입니다. 에포크 동안 학습에서 발생한 손실, 수익 발생 횟수, 탐험 횟수, 학습 횟수 등을 기록하고 로그로 남깁니다. 로그, 가시화, 학습 모델 등은 output_path로 지정된 경로 하위에 저장됩니다.
코드 조각 3: 가치 신경망 생성 함수
init_value_network() 함수는 net에 지정된 신경망 종류에 맞게 가치 신경망을 생성합니다. 다음 코드조각은 init_value_network() 함수의 소스 코드를 보여줍니다.
ReinforcementLearner 클래스: 가치 신경망 생성 함수
def init_value_network(self, shared_network=None,
activation='linear', loss='mse'):
if self.net == 'dnn':
self.value_network = DNN(
input_dim=self.num_features,
output_dim=self.agent.NUM_ACTIONS,
lr=self.lr, shared_network=shared_network,
activation=activation, loss=loss)
elif self.net == 'lstm':
self.value_network = LSTMNetwork(
input_dim=self.num_features,
output_dim=self.agent.NUM_ACTIONS,
lr=self.lr, num_steps=self.num_steps,
shared_network=shared_network,
activation=activation, loss=loss)
elif self.net == 'cnn':
self.value_network = CNN(
input_dim=self.num_features,
output_dim=self.agent.NUM_ACTIONS,
lr=self.lr, num_steps=self.num_steps,
shared_network=shared_network,
activation=activation, loss=loss)
if self.reuse_models and \
os.path.exists(self.value_network_path):
self.value_network.load_model(
model_path=self.value_network_path)
net이 'dnn'이면 DNN 클래스로 가치 신경망을 생성하고, 'lstm'이면 LSTMNetwork 클래스로, 'cnn'이면 CNN 클래스로 가치 신경망을 생성합니다. 이 클래스들은 Network 클래스를 상속하므로 Network 클래스의 함수를 모두 가지고 있습니다.
가치 신경망은 손익률을 회귀분석하는 모델로 보면 됩니다. 그래서 activation은 선형 함수로, 손실 함수는 MSE로 설정했습니다.
reuse_models가 True이고 value_network_path에 지정된 경로에 신경망 모델 파일이 존재하면 신경망 모델 파일을 불러옵니다.
코드 조각 4: 정책 신경망 생성 함수
다음은 신경망 종류에 따라 정책 신경망을 생성하는 소스 코드입니다.
ReinforcementLearner 클래스: 정책 신경망 생성 함수
def init_policy_network(self, shared_network=None,
activation='sigmoid', loss='mse'):
if self.net == 'dnn':
self.policy_network = DNN(
input_dim=self.num_features,
output_dim=self.agent.NUM_ACTIONS,
lr=self.lr, shared_network=shared_network,
activation=activation, loss=loss)
elif self.net == 'lstm':
self.policy_network = LSTMNetwork(
input_dim=self.num_features,
output_dim=self.agent.NUM_ACTIONS,
lr=self.lr, num_steps=self.num_steps,
shared_network=shared_network,
activation=activation, loss=loss)
elif self.net == 'cnn':
self.policy_network = CNN(
input_dim=self.num_features,
output_dim=self.agent.NUM_ACTIONS,
lr=self.lr, num_steps=self.num_steps,
shared_network=shared_network,
activation=activation, loss=loss)
if self.reuse_models and \
os.path.exists(self.policy_network_path):
self.policy_network.load_model(
model_path=self.policy_network_path)
가치 신경망을 생성하는 init_value_network() 함수와 매우 유사한 함수입니다. 차이는 활성화 함수 activation 인자로 가치 신경망은 'linear'를 쓰는 반면, 정책 신경망은 'sigmoid'를 쓴다는 점입니다. 정책 신경망은 샘플에 대해서 PV를 높이기 위해 취하기 좋은 행동에 대한 분류 모델로 볼 수 있습니다. 활성화 함수로 시그모이드를 써서 결괏값이 0과 1 사이로 나오게 해서 확률로 사용할 수 있게 했습니다.
reuse_models가 True이고 policy_network_path 변수로 저장된 경로에 신경망 모델 파일이 존재하면 이 모델 파일을 불러옵니다.
코드 조각 5: 에포크 초기화 함수
다음 코드조각은 에포크마다 새로 데이터가 쌓이는 변수들을 초기화하는 reset() 함수를 보여줍니다.
ReinforcementLearner 클래스: 초기화 함수
def reset(self):
self.sample = None
self.training_data_idx = -1
# 환경 초기화
self.environment.reset()
# 에이전트 초기화
self.agent.reset()
# 가시화 초기화
self.visualizer.clear([0, len(self.chart_data)])
# 메모리 초기화
self.memory_sample = []
self.memory_action = []
self.memory_reward = []
self.memory_value = []
self.memory_policy = []
self.memory_pv = []
self.memory_num_stocks = []
self.memory_exp_idx = []
self.memory_learning_idx = []
# 에포크 관련 정보 초기화
self.loss = 0.
self.itr_cnt = 0
self.exploration_cnt = 0
self.batch_size = 0
self.learning_cnt = 0
이 함수는 학습 데이터를 다시 처음부터 읽기 위해 training_data_idx를 -1로 재설정합니다. 이 값은 학습 데이터를 읽어가면서 1씩 증가합니다. 읽어온 데이터는 sample에 저장되는데, 초기화 단계에는 읽어온 학습 데이터가 없기 때문에 None으로 할당합니다.
그 외에 환경, 에이전트, 가시화기, 메모리, 에포크 관련 정보를 모두 초기화합니다. environment, agent의 reset() 함수와 visualizer의 clear() 함수를 호출합니다. 그리고 memory_* 리스트도 비워줍니다.
에포크 관련 정보를 초기화합니다. loss는 신경망의 결과가 학습 데이터와 얼마나 차이가 있는지를 저장하는 변수입니다. loss 값은 학습이 진행되면서 줄어드는 것이 좋습니다.
itr_cnt 변수는 수행한 에포크 수를 저장합니다. exploration_cnt 변수는 무작위 투자를 수행한 횟수를 저장합니다. epsilon이 0.1이고 100번의 투자 결정이 있다고 한다면 약 10번의 무작위 투자를 할 것입니다.
학습할 미니 배치 크기는 batch_size에 저장되며 learning_cnt에는 한 에포크 동안 수행한 미니 배치 학습 횟수를 저장합니다.
코드 조각 6: 가치 신경망 및 정책 신경망 학습
다음은 학습 데이터를 구성하는 샘플 하나를 생성하는 build_sample() 함수를 보여줍니다.
ReinforcementLearner 클래스의 build_sample() 함수
def build_sample(self):
self.environment.observe()
if len(self.training_data) > self.training_data_idx + 1:
self.training_data_idx += 1
self.sample = self.training_data.iloc[
self.training_data_idx].tolist()
self.sample.extend(self.agent.get_states())
return self.sample
return None
환경 객체의 observe() 함수를 호출해 차트 데이터의 현재 인덱스에서 다음 인덱스 데이터를 읽게 합니다. 그리고 학습 데이터의 다음 인덱스가 존재하는지 확인합니다.
학습 데이터에 다음 인덱스 데이터가 존재하면 training_data_idx 변수를 1만큼 증가시키고 training_data 배열에서 training_data_idx 인덱스의 데이터를 받아와서 sample로 저장합니다. 현재까지는 sample 데이터는 26개의 값으로 구성돼 있습니다. 다음으로 sample에 에이전트 상태를 추가해 sample을 28개의 값으로 구성합니다.
예제 5.41은 신경망을 학습하기 위해 배치 학습 데이터를 생성하는 get_batch() 함수와 신경망을 학습하는 update_networks() 함수를 보여줍니다.
예제 5.41 ReinforcementLearner 클래스의 get_batch(), update_networks() 함수
@abc.abstractmethod
def get_batch(self, batch_size, delayed_reward, discount_factor):
pass
def update_networks(self,
batch_size, delayed_reward, discount_factor):
# 배치 학습 데이터 생성
x, y_value, y_policy = self.get_batch(
batch_size, delayed_reward, discount_factor)
if len(x) > 0:
loss = 0
if y_value is not None:
# 가치 신경망 갱신
loss += self.value_network.train_on_batch(x, y_value)
if y_policy is not None:
# 정책 신경망 갱신
loss += self.policy_network.train_on_batch(x, y_policy)
return loss
return None
get_batch() 함수는 추상 메서드로서 ReinforcementLearner 클래스의 하위 클래스들은 반드시 이 함수를 구현해야 합니다. ReinforcementLearner 클래스를 상속하고도 이 추상 메서드를 구현하지 않으면 NotImplemented 예외가 발생합니다.
update_networks() 함수는 get_batch() 함수를 호출해 배치 학습 데이터를 생성하고 가치 신경망과 정책 신경망을 학습하기 위해 신경망 클래스의 train_on_batch() 함수를 호출합니다. 가치 신경망은 DQNLearner, ActorCriticLearner, A2CLearner에서 학습하고 정책 신경망은 PolicyGradientLearner, ActorCriticLearner, A2CLearner에서 학습합니다.
학습 후 발생하는 손실(loss)를 반환합니다. 가치 신경망과 정책 신경망을 모두 학습하는 경우 두 학습 손실을 합산해 반환합니다.
다음 코드조각은 ReinforcementLearner 클래스의 fit() 함수를 보여줍니다.
ReinforcementLearner 클래스의 fit() 함수
def fit(self, delayed_reward, discount_factor):
# 배치 학습 데이터 생성 및 신경망 갱신
if self.batch_size > 0:
_loss = self.update_networks(
self.batch_size, delayed_reward, discount_factor)
if _loss is not None:
self.loss += abs(_loss)
self.learning_cnt += 1
self.memory_learning_idx.append(self.training_data_idx)
self.batch_size = 0
fit() 함수는 배치 학습 데이터의 크기를 정하고 update_networks() 함수를 호출합니다. 그리고 반환받은 학습 손실 값인 _loss를 loss에 더합니다. loss는 에포크 동안의 총 학습 손실을 가지게 됩니다. learning_cnt에 학습 횟수를 저장하고 나중에 loss를 learning_cnt로 나누어 에포크의 학습 손실로 여깁니다. 그리고 memory_learning_idx에 학습 위치를 저장합니다.
에포크 동안 쌓은 모든 데이터를 학습에 사용할지 말지를 full 인자로 받으며 full이 True이면 전체 데이터에 대해 학습을 수행합니다. 이는 에포크가 종료됐을 때 가치 신경망을 추가로 학습하는 용도로 사용합니다.
코드 조각 7: 에포크 결과 가시화
다음 코드조각은 하나의 에포크가 완료되어 에포크 관련 정보를 가시화하는 부분입니다.
예제 5.43 ReinforcementLearner 클래스: 가시화 함수 (1)
def visualize(self, epoch_str, num_epoches, epsilon):
self.memory_action = [Agent.ACTION_HOLD] \
* (self.num_steps - 1) + self.memory_action
self.memory_num_stocks = [0] * (self.num_steps - 1) \
+ self.memory_num_stocks
if self.value_network is not None:
self.memory_value = [np.array([np.nan] \
* len(Agent.ACTIONS))] * (self.num_steps - 1) \
+ self.memory_value
if self.policy_network is not None:
self.memory_policy = [np.array([np.nan] \
* len(Agent.ACTIONS))] * (self.num_steps - 1) \
+ self.memory_policy
self.memory_pv = [self.agent.initial_balance] \
* (self.num_steps - 1) + self.memory_pv
가시화하는 대상은 에이전트의 행동, 보유 주식 수, 가치 신경망 출력, 정책 신경망 출력, 포트폴리오 가치, 탐험 위치, 학습 위치 등입니다.
LSTM 신경망과 CNN 신경망을 사용하는 경우 에이전트 행동, 보유 주식 수, 가치 신경망 출력, 정책 신경망 출력, 포트폴리오 가치는 환경의 일봉 수보다 num_steps – 1만큼 부족하기 때문에 num_steps – 1만큼 의미 없는 값을 첫 부분에 채워줍니다.
파이썬 팁: 파이썬에서는 리스트에 곱하기를 하면 똑같은 리스트를 뒤에 붙여줍니다. 예를 들어,
[1, 2, 3] * 3은[1, 2, 3, 1, 2, 3, 1, 2, 3]이 됩니다.
다음은 본격적으로 가시화를 시작하는 부분입니다.
ReinforcementLearner 클래스: 가시화 함수 (2)
self.visualizer.plot(
epoch_str=epoch_str, num_epoches=num_epoches,
epsilon=epsilon, action_list=Agent.ACTIONS,
actions=self.memory_action,
num_stocks=self.memory_num_stocks,
outvals_value=self.memory_value,
outvals_policy=self.memory_policy,
exps=self.memory_exp_idx,
learning_idxes=self.memory_learning_idx,
initial_balance=self.agent.initial_balance,
pvs=self.memory_pv,
)
self.visualizer.save(os.path.join(
self.epoch_summary_dir,
'epoch_summary_{}.png'.format(epoch_str))
)
객체 visualizer의 plot() 함수를 호출합니다. 그리고 생성된 에포크 결과 그림을 PNG 그림 파일로 저장합니다.
코드 조각 8: 강화학습 실행 함수
run() 함수는 ReinforcementLearner 클래스의 핵심 함수이며 그 길이 역시 상대적으로 깁니다. 그래서 여러 부분으로 나누어 살펴보고자 합니다. 다음은 run() 함수의 선언부를 보여줍니다.
ReinforcementLearner 클래스: 강화학습 실행 함수 (1)
def run(
self, num_epoches=100, balance=10000000,
discount_factor=0.9, start_epsilon=0.5, learning=True):
info = "[{code}] RL:{rl} Net:{net} LR:{lr} " \
"DF:{discount_factor} TU:[{min_trading_unit}," \
"{max_trading_unit}] DRT:{delayed_reward_threshold}".format(
code=self.stock_code, rl=self.rl_method, net=self.net,
lr=self.lr, discount_factor=discount_factor,
min_trading_unit=self.agent.min_trading_unit,
max_trading_unit=self.agent.max_trading_unit,
delayed_reward_threshold=self.agent.delayed_reward_threshold
)
with self.lock:
logging.info(info)
# 시작 시간
time_start = time.time()
num_epoches는 총 수행할 반복 학습 횟수입니다. 반복 학습을 거치면서 가치 신경망과 정책 신경망이 점점 포트폴리오 가치를 높이는 방향으로 갱신되기 때문에 충분한 반복 횟수를 정해 줘야 합니다. 그러나 num_epoches를 너무 크게 잡으면 학습에 소요되는 시간이 너무 길어지므로 적절하게 정해야 합니다. 얼마나 많은 양의 데이터를 학습하는가에 따라 다르지만, 여기서는 기본값을 100번으로 정합니다.
balance는 에이전트의 초기 투자 자본금을 정하기 위한 인자입니다. RLTrader에서는 신용 거래와 같이 보유 현금을 넘어서는 투자는 고려하지 않습니다. 보유 현금이 부족하면 정책 신경망 결과 매수가 좋아도 관망합니다.
discount_factor는 상태-행동 가치를 구할 때 적용할 할인율입니다. 보상이 발생했을 때 그 이전 보상이 발생한 시점과 현재 보상이 발생한 시점 사이에서 수행한 행동 전체에 현재의 보상이 영향을 미칩니다. 이때, 과거로 갈수록 현재 보상을 적용할 판단 근거가 흐려지기 때문에 먼 과거의 행동일수록 현재의 보상을 약하게 적용합니다.
start_epsilon은 초기 탐험 비율을 의미합니다. 전혀 학습되지 않은 초기에는 탐험 비율 크게 해서 더 많은 탐험, 즉 무작위 투자를 수행하게 해야 합니다. 탐험을 통해 특정 상황에서 좋은 행동과 그렇지 않은 행동을 결정하기 위한 경험을 쌓습니다.
learning은 학습 유무를 정하는 불리언(Boolean) 값입니다. 불리언 값이란 참(True) 또는 거짓(False)을 가지는 이진 값을 말합니다. 학습을 마치면 학습된 가치 신경망 모델과 정책 신경망 모델이 만들어집니다. 이렇게 학습을 해서 신경망 모델을 만들고자 한다면 learning을 True로, 학습된 모델을 가지고 투자 시뮬레이션만 하려 한다면 learning을 False로 줍니다.
run() 함수에 들어오면 강화 학습 설정을 로그로 기록합니다. 그리고 학습 시작 시간을 저장해 둡니다. 학습 종료 후의 시간과의 차이를 학습 시간으로 기록하기 위해서입니다.
다음은 run() 함수에서 가시화를 준비하는 부분입니다.
ReinforcementLearner 클래스: 강화학습 실행 함수 (2)
# 가시화 준비
# 차트 데이터는 변하지 않으므로 미리 가시화
self.visualizer.prepare(self.environment.chart_data, info)
# 가시화 결과 저장할 폴더 준비
self.epoch_summary_dir = os.path.join(
self.output_path, 'epoch_summary_{}'.format(
self.stock_code))
if not os.path.isdir(self.epoch_summary_dir):
os.makedirs(self.epoch_summary_dir)
else:
for f in os.listdir(self.epoch_summary_dir):
os.remove(os.path.join(self.epoch_summary_dir, f))
# 에이전트 초기 자본금 설정
self.agent.set_balance(balance)
# 학습에 대한 정보 초기화
max_portfolio_value = 0
epoch_win_cnt = 0
가시화 객체 visualizer의 prepare() 함수를 호출해 가시화를 준비합니다. prepare() 함수는 에포크가 진행돼도 변하지 않는 주식투자 환경인 차트 데이터를 미리 가시화합니다.
그리고 가시화 결과를 저장할 경로를 준비합니다. 가시화 결과는 output_path 경로 하위의 epoch_summary_* 폴더에 저장됩니다. 이미 epoch_summary_* 폴더에 저장된 파일이 있을 경우 모두 삭제합니다.
그리고 에이전트의 초기 자본금을 설정합니다. 기본 자본금으로 천만 원이 설정돼 있습니다.
max_portfolio_value 변수에는 수행한 에포크 중에서 가장 높은 포트폴리오 가치가 저장됩니다. epoch_win_cnt 변수에는 수행한 에포크 중에서 수익이 발생한 에포크 수를 저장합니다. 즉, 포트폴리오 가치가 초기 자본금보다 높아진 에포크 수입니다.
다음은 이어서 지정된 에포크 수만큼 주식투자 시뮬레이션을 반복하며 학습하는 반복문 도입부입니다.
ReinforcementLearner 클래스: 강화학습 실행 함수 (3)
# 학습 반복
for epoch in range(num_epoches):
time_start_epoch = time.time()
# step 샘플을 만들기 위한 큐
q_sample = collections.deque(maxlen=self.num_steps)
# 환경, 에이전트, 신경망, 가시화, 메모리 초기화
self.reset()
# 학습을 진행할 수록 탐험 비율 감소
if learning:
epsilon = start_epsilon \
* (1. - float(epoch) / (num_epoches - 1))
self.agent.reset_exploration()
else:
epsilon = start_epsilon
이어서 살펴볼 코드들도 이 반복문 내에 포함돼 있음을 유의하기 바랍니다. 이 반복문을 빠져나오는 곳에서 다시 언급하겠습니다.
파이썬 팁: 파이썬에서 코드 블록은 들여쓰기(Indent)로 구분합니다. 특히,
class,def,if,elif,else,for,while,try,except,final,with등을 사용할 때 들여 쓰기에 유의해야 합니다.
반복문 안으로 들어오면 먼저 에포크의 시작 시간을 기록합니다. 이는 한 에포크를 수행하는 데 시간이 얼마나 걸렸는지 확인하기 위해서입니다.
그리고 num_step만큼 샘플을 담아둘 큐(Queue)를 초기화합니다.
파이썬 팁: 큐(Queue)는 선입선출(First In First Out, FIFO) 자료구조입니다. 데크(Deque)는 양방향 큐로 볼 수 있습니다. 파이썬에서는
collections모듈에deque함수로 양방향 큐 자료구조를 만들 수 있습니다. 이때,maxlen파라미터를 주면 이 양방향 큐의 크기를 제한할 수 있습니다.
그리고 에포크마다 초기화되는 환경, 에이전트, 신경망, 가시화 정보, 메모리들을 초기화하기 위해 reset() 함수를 호출합니다.
에포크가 거듭될수록 epsilon은 start_epsilon에서 점차 줄어들게 됩니다. epsilon 값을 정할 때는 최초 무작위 투자 비율인 start_epsilon 값에 현재 epoch 수에 학습 진행률을 곱해서 정합니다. 예를 들어, start_epsilon이 0.5라면 첫 번째 에포크에서는 30%의 확률로 무작위 투자를 진행합니다. 총 수행할 에포크 수가 100이라고 했을 때 50번째 에포크에서 epsilon은 0.5×(1-49/99)≈0.49이 됩니다.
그리고 agent 객체의 reset_exploration() 함수를 호출해 exploration_base를 새로 정합니다. exploration_base는 무작위로 정해지고 이 값이 클수록 매수 기조의 탐험을 하게 됩니다.
다음은 하나의 에포크를 수행하는 while 문의 초반부를 보여줍니다.
ReinforcementLearner 클래스: 강화학습 실행 함수 (4)
while True:
# 샘플 생성
next_sample = self.build_sample()
if next_sample is None:
break
# num_steps만큼 샘플 저장
q_sample.append(next_sample)
if len(q_sample) < self.num_steps:
continue
build_sample() 함수를 호출해 환경 객체로부터 하나의 샘플을 읽어옵니다. next_sample이 None이라면 마지막까지 데이터를 다 읽은 것이므로 while 반복문을 종료합니다.
num_steps 개수만큼 샘플이 준비돼야 행동을 결정할 수 있기 때문에 샘플 큐에 샘플이 모두 찰 때까지 continue를 통해 이후 로직을 건너뜁니다.
파이썬 팁:
for와while반복문에서는break로 반복문을 끝낼 수 있고continue로 이후의 코드를 건너뛸 수 있습니다.
다음은 가치 신경망과 정책 신경망으로 예측 행동 가치와 예측 행동 확률을 구하는 부분입니다.
ReinforcementLearner 클래스: 강화학습 실행 함수 (5)
# 가치, 정책 신경망 예측
pred_value = None
pred_policy = None
if self.value_network is not None:
pred_value = self.value_network.predict(
list(q_sample))
if self.policy_network is not None:
pred_policy = self.policy_network.predict(
list(q_sample))
# 신경망 또는 탐험에 의한 행동 결정
action, confidence, exploration = \
self.agent.decide_action(
pred_value, pred_policy, epsilon)
# 결정한 행동을 수행하고 즉시 보상과 지연 보상 획득
immediate_reward, delayed_reward = \
self.agent.act(action, confidence)
각 신경망 객체의 predict() 함수를 호출해 예측 행동 가치와 예측 행동 확률을 구합니다. 이렇게 구한 가치와 확률로 행동을 결정합니다.
이렇게 구한 예측 가치와 확률로 투자 행동을 결정합니다. 여기서는 매수와 매도 중 하나를 결정합니다. 이 행동 결정은 무작위 투자 비율인 epsilon 값의 확률로 무작위로 하거나, 그렇지 않은 경우 신경망의 출력을 통해 결정됩니다. 정책 신경망의 출력은 매수를 했을 때와 매도를 했을 때의 포트폴리오 가치를 높일 확률을 의미합니다. 즉, 매수에 대한 정책 신경망 출력이 매도에 대한 출력보다 높으면 매수를, 그 반대의 경우 매도를 선택합니다. 정책 신경망의 출력이 없으면 가치 신경망의 출력값이 높은 행동을 선택합니다. 가치 신경망의 출력은 행동에 대한 예측 가치(손익률)를 의미합니다.
decide_action() 함수가 반환하는 값은 세 가지입니다. 결정한 행동인 action, 결정에 대한 확신도인 confidence, 무작위 투자 유무인 exploration입니다.
결정한 행동을 수행하도록 에이전트의 act() 함수를 호출합니다. act() 함수는 행동을 수행하고 즉시 보상과 지연 보상을 반환합니다.
다음은 수행한 행동과 행동에 대한 결과를 메모리에 저장하고 학습을 수행하는 부분입니다.
ReinforcementLearner 클래스: 강화학습 실행 함수 (6)
# 행동 및 행동에 대한 결과를 기억
self.memory_sample.append(list(q_sample))
self.memory_action.append(action)
self.memory_reward.append(immediate_reward)
if self.value_network is not None:
self.memory_value.append(pred_value)
if self.policy_network is not None:
self.memory_policy.append(pred_policy)
self.memory_pv.append(self.agent.portfolio_value)
self.memory_num_stocks.append(self.agent.num_stocks)
if exploration:
self.memory_exp_idx.append(self.training_data_idx)
# 반복에 대한 정보 갱신
self.batch_size += 1
self.itr_cnt += 1
self.exploration_cnt += 1 if exploration else 0
# 지연 보상 발생된 경우 미니 배치 학습
if learning and (delayed_reward != 0):
self.fit(delayed_reward, discount_factor)
# 에포크 종료 후 학습
if learning:
self.fit(self.agent.profitloss, discount_factor)
행동과 행동에 대한 결과를 memory로 시작하는 변수들에 저장합니다. 이 변수들은 학습 데이터의 샘플, 에이전트 행동, 즉시 보상, 가치 신경망 출력, 정책 신경망 출력, 포트폴리오 가치, 보유 주식 수, 탐험 위치를 저장하는 배열입니다. 메모리 변수들은 두 가지 목적으로 (1) 학습에서 배치 학습 데이터로 사용하고 (2) 가시화기에서 차트를 그릴 때 사용합니다.
배치 크기 batch_size, 반복 카운팅 횟수 itr_cnt, 무작위 투자 횟수 exploration_cnt를 증가시킵니다. exploration_cnt의 경우 탐험한 경우에만 1을 증가시키고 그렇지 않으면 0을 더해서 변화가 없게 합니다.
지연 보상이 발생한 경우 신경망 학습 함수인 fit()을 호출합니다. 지연 보상은 지연 보상 임계치가 넘는 손익률이 발생했을 때 주어집니다.
이렇게 while 블록 안에서 환경으로부터 샘플을 받으며 하나의 에포크를 수행합니다. 더 이상 샘플이 없는 경우 while 블록을 빠져나옵니다. while 블록을 빠져나온 후에 남은 미니 배치를 학습합니다. 이때는 대부분 지연 보상이 발생하지 않은 상태이므로 에이전트 객체가 가지고 있는 손익률을 사용합니다.
파이썬 팁: 파이썬에서는
if else문을 한 줄에 쓸 수 있습니다. 예를 들어x가0보다 크거나 같을 경우에1을 증가시키고0보다 작을 경우에1을 감소시키는 코드를 작성해 보겠습니다. 일반적인 방법으로는 다음과 같이 작성할 수 있습니다.if x >= 0: x += 1 else: x -= 1파이썬에서는 이 코드를 한줄로 다음과 같이 작성할 수 있습니다.
x += 1 if x >= 0 else -1
다음은 하나의 에포크에 대한 정보를 로깅하고 가시화하는 부분입니다.
ReinforcementLearner 클래스: 강화학습 실행 함수 (7)
# 에포크 관련 정보 로그 기록
num_epoches_digit = len(str(num_epoches))
epoch_str = str(epoch + 1).rjust(num_epoches_digit, '0')
time_end_epoch = time.time()
elapsed_time_epoch = time_end_epoch - time_start_epoch
if self.learning_cnt > 0:
self.loss /= self.learning_cnt
logging.info("[{}][Epoch {}/{}] Epsilon:{:.4f} "
"#Expl.:{}/{} #Buy:{} #Sell:{} #Hold:{} "
"#Stocks:{} PV:{:,.0f} "
"LC:{} Loss:{:.6f} ET:{:.4f}".format(
self.stock_code, epoch_str, num_epoches, epsilon,
self.exploration_cnt, self.itr_cnt,
self.agent.num_buy, self.agent.num_sell,
self.agent.num_hold, self.agent.num_stocks,
self.agent.portfolio_value, self.learning_cnt,
self.loss, elapsed_time_epoch))
# 에포크 관련 정보 가시화
self.visualize(epoch_str, num_epoches, epsilon)
# 학습 관련 정보 갱신
max_portfolio_value = max(
max_portfolio_value, self.agent.portfolio_value)
if self.agent.portfolio_value > self.agent.initial_balance:
epoch_win_cnt += 1
하나의 에포크에 대한 로그 기록은 주식 종목 코드, 현재 에포크 번호, 해당 에포크에서의 탐험률, 에포크 동안 매수 행동 수행 횟수, 매도 수행 횟수, 관망 횟수, 보유 주식 수, 달성한 포트폴리오 가치, 미니 배치 학습 수행 횟수, 학습 손실, 에포크 수행 소요 시간을 포함합니다.
총 에포크 수의 문자열 길이를 확인합니다. 총 에포크 수가 1,000이면 길이는 4가 됩니다. 현재 에포크 수를 num_epoches_digit 길이의 문자열로 만들어 epoch_str에 저장합니다. 4자리 문자열을 만든다고 할 때, 첫 번째 에포크의 경우 epoch가 0이기 때문에 1을 더하고 앞에 '0'을 채워서 '0001'로 만듭니다.
그리고 현재 시간인 time_end_epoch에 time_start_epoch를 빼서 에포크 수행 소요 시간 elapsed_time_epoch을 저장합니다. loss 변수는 에포크 동안 수행한 미니 배치들의 학습 손실을 모두 더해놓은 상태입니다. loss를 학습 횟수만큼 나눠서 미니 배치의 평균 학습 손실로 갱신합니다.
파이썬 팁:
rjust()함수는 문자열을 자릿수에 맞게 오른쪽으로 정렬해주는 함수입니다. 예를 들어,"1".rjust(5)를 하면' 1'이 됩니다. 앞에 빈칸 4자리를 채워주고 1을 붙여서 5자리 문자열이 되는 것입니다. 두 번째 인자로 빈칸 대신 채워줄 문자를 정해줄 수도 있습니다."1".rjust(5, '0')은'00001'이 됩니다.비슷한 함수로
ljust()함수가 있습니다. 이 함수는 기존 문자열 앞에 빈칸 또는 특정 문자를 채워줍니다.이렇게 기존 문자 앞 또는 뒤에 어떠한 문자를 채워주는 것을 패딩(Padding)이라고 합니다.
파이썬 팁: 파이썬 문자열
format()함수에서 키워드 명 뒤에 콜론(:)을 붙이고 형식 옵션을 지정할 수 있습니다.{:,.0f}는 천 단위로 콤마(,)를 붙이고 소수점은 표시하지 않겠다는 뜻입니다.
가시화기 객체를 이용해 에포크 정보를 하나의 그림으로 가시화한 후 파일로 저장합니다. 이를 위해 visualize() 함수를 호출합니다.
이제 학습 관련 정보를 갱신합니다. 에포크를 수행하는 동안 최대 포트폴리오 가치를 갱신하고 해당 에포크에서 포트폴리오 가치가 자본금보다 높으면 epoch_win_cnt를 증가시킵니다.
여기까지가 에포크 반복 for 문 블록에 해당되는 부분입니다. 다음 코드조각은 모든 에포크 수행을 마친 후 강화학습 실행 함수의 남은 로직을 보여줍니다.
ReinforcementLearner 클래스: 강화학습 실행 함수 (8)
# 종료 시간
time_end = time.time()
elapsed_time = time_end - time_start
# 학습 관련 정보 로그 기록
with self.lock:
logging.info("[{code}] Elapsed Time:{elapsed_time:.4f} "
"Max PV:{max_pv:,.0f} #Win:{cnt_win}".format(
code=self.stock_code, elapsed_time=elapsed_time,
max_pv=max_portfolio_value, cnt_win=epoch_win_cnt))
모든 에포크를 수행하고 전체 에포크 수행 소요 시간을 기록합니다. 그리고 전체 소요 시간, 최대 포트폴리오 가치, 포트폴리오 가치가 자본금보다 높았던 에포크 수를 로그로 남깁니다.
다음은 학습을 마친 신경망 모델을 저장하는 함수입니다.
ReinforcementLearner 클래스: 신경망 모델 저장 함수
def save_models(self):
if self.value_network is not None and \
self.value_network_path is not None:
self.value_network.save_model(self.value_network_path)
if self.policy_network is not None and \
self.policy_network_path is not None:
self.policy_network.save_model(self.policy_network_path)
가치 신경망이 존재한다면 가치 신경망 모델 파일 경로를 확인해 신경망 클래스의 save_model() 함수를 호출합니다. 마찬가지로 학습한 정책 신경망이 존재하면 파일로 저장합니다.
코드 조각 9: DQN 강화학습 클래스
DQN은 가치 신경망으로만 강화학습을 하는 방식입니다. 다음은 DQNLearner 클래스의 생성자와 배치 학습 데이터 생성 함수를 보여줍니다.
class DQNLearner(ReinforcementLearner):
def __init__(self, *args, value_network_path=None, **kwargs):
super().__init__(*args, **kwargs)
self.value_network_path = value_network_path
self.init_value_network()
def get_batch(self, batch_size, delayed_reward, discount_factor):
memory = zip(
reversed(self.memory_sample[-batch_size:]),
reversed(self.memory_action[-batch_size:]),
reversed(self.memory_value[-batch_size:]),
reversed(self.memory_reward[-batch_size:]),
)
x = np.zeros((batch_size, self.num_steps, self.num_features))
y_value = np.zeros((batch_size, self.agent.NUM_ACTIONS))
value_max_next = 0
reward_next = self.memory_reward[-1]
for i, (sample, action, value, reward) in enumerate(memory):
x[i] = sample
y_value[i] = value
r = (delayed_reward + reward_next - reward * 2) * 100
y_value[i, action] = r + discount_factor * value_max_next
value_max_next = value.max()
reward_next = reward
return x, y_value, None
DQNLearner의 생성자에서 value_network_path를 속성으로 저장하고 가치 신경망 생성을 위해 init_value_network() 함수를 호출합니다. DQNLearner는 ReinforcementLearner 클래스를 상속하므로 ReinforcementLearner의 속성과 함수를 모두 가집니다. ReinforcementLearner를 상속하는 모든 클래스는 ReinforcementLearner 클래스의 추상 메서드인 get_batch() 함수를 구현해야 합니다.
DQNLearner의 get_batch() 함수에서 먼저 메모리 배열을 묶어줍니다. 이때 메모리 배열들을 역으로 묶어줍니다. 그리고 샘플 배열 x와 레이블 배열 y_value를 준비합니다. 배열은 모두 0으로 채워넣습니다.
파이썬 팁: 파이썬에서 리스트를 역으로 뒤집을 수 있습니다. 리스트를 역으로 뒤집는 세 가지 방법을 소개합니다. 리스트 변수의
reverse()함수,reversed()내장 함수,[::-1]슬라이싱 트릭을 사용하면 리스트의 요소들을 역순으로 바꿀 수 있습니다.lst라는 리스트 변수가 있을 때lst.reverse(),reversed(lst),lst[::-1]과 같이 사용할 수 있습니다. 주의할 점은lst.reverse()는lst변수 자체가 역순으로 바뀝니다. 즉,reverse()함수는 제자리에서(in-place) 값을 변경합니다.reversed()내장 함수는lst를 역으로 취한 새로운 리스트를 반환합니다.lst[::-1]슬라이싱 트릭도 마찬가지로 새로운 리스트를 반환합니다.파이썬 팁: NumPy의
zeros()함수는 인자로 배열의 형태인shape를 받습니다. 이shape형태의 배열을 0으로 채워서 반환합니다. 예를 들어,zeros(3, 1)은[0, 0, 0]을,zeros((2, 2))는[ [0, 0], [0, 0] ]을 반환합니다. 주의할 점은 다차원 배열의 경우 그 형태를 튜플로 넘겨줘야 한다는 것입니다.
이제 for 문으로 샘플 배열과 레이블 배열에 값을 채워줍니다. 메모리를 역으로 취했기 때문에 for 문은 배치 학습 데이터의 마지막 부분부터 처리합니다. 먼저 x[i]에 샘플을 채워주고 y_value[i]에 가치 신경망의 출력을 넣어줍니다.
변수 r에 학습에 사용할 보상을 구해 저장합니다. 여기서 delayed_reward는 배치 데이터 내에서의 마지막 손익률이고 reward는 행동을 수행한 시점에서의 손익률입니다. 최종 수익률과 현재 수익률을 빼고 다음 행동 수행 시점과 현재 행동 수행 시점의 손익률을 빼서 더하는 것입니다.
그리고 다음 상태의 최대 가치에 할인율을 적용해 구한 r을 더해줍니다. 이 값을 학습할 상태-행동 가치로 적용합니다. 다음 상태의 최대 가치는 value_max_next 변수에 저장합니다. 그리고 다음 행동 시점에서의 손익률을 next_reward 변수에 저장합니다.
파이썬 팁: NumPy 배열은
min(),max(),mean()등의 기본적인 통계 함수를 가지고 있습니다.np_array.min(),np_array.max(),np_array.mean()과 같이 사용하면 됩니다.
get_batch() 함수는 최종적으로 샘플 배열, 가치 신경망 학습 레이블 배열, 정책 신경망 학습 레이블 배열을 반환합니다. DQNLearner의 경우 정책 신경망을 사용하지 않기 때문에 정책 신경망 학습 레이블 배열 부분은 None으로 처리합니다.
get_batch() 함수를 원하는 대로 수정해 다른 방식으로 배치 학습 데이터를 생성할 수 있습니다. RLTrader 커스터마이징에 관한 내용은 부록 2에서 다룹니다.
코드 조각 10: 정책 경사 강화학습 클래스
정책 경사 강화학습은 정책 신경망으로만 강화학습을 하는 방식입니다. 다음은 정책 경사 강화학습을 위한 PolicyGradientLearner 클래스를 보여줍니다.
class PolicyGradientLearner(ReinforcementLearner):
def __init__(self, *args, policy_network_path=None, **kwargs):
super().__init__(*args, **kwargs)
self.policy_network_path = policy_network_path
self.init_policy_network()
def get_batch(self, batch_size, delayed_reward, discount_factor):
memory = zip(
reversed(self.memory_sample[-batch_size:]),
reversed(self.memory_action[-batch_size:]),
reversed(self.memory_policy[-batch_size:]),
reversed(self.memory_reward[-batch_size:]),
)
x = np.zeros((batch_size, self.num_steps, self.num_features))
y_policy = np.full((batch_size, self.agent.NUM_ACTIONS), .5)
reward_next = self.memory_reward[-1]
for i, (sample, action, policy, reward) in enumerate(memory):
x[i] = sample
y_policy[i] = policy
r = (delayed_reward + reward_next - reward * 2) * 100
y_policy[i, action] = sigmoid(r)
reward_next = reward
return x, None, y_policy
x는 일련의 학습 데이터 및 에이전트 상태로 구성된 샘플 배열입니다. y_policy는 정책 신경망 학습을 위한 레이블 배열입니다. x 배열의 형태는 배치 데이터 크기, 학습 데이터 특징 크기의 2차원으로 구성됩니다. y_policy 배열의 형태는 배치 데이터 크기, 정책 신경망이 결정하는 에이전트 행동의 수로 2차원으로 구성됩니다. y_policy는 0.5로 일괄적으로 채웁니다.
파이썬 팁: NumPy의
full()함수는 첫 번째 인자로 배열의 형태인shape를 받아서 두 번째 인자로 입력된 값으로 채워진 NumPy 배열을 반환합니다. 예를 들어,full(3, 1)은[1, 1, 1]을,full((2, 2), 0.5)는[ [0.5, 0.5], [0.5, 0.5] ]를 반환합니다.
배치 데이터 크기는 지연 보상이 발생할 때 결정되기 때문에 매번 다르고 학습 데이터 특징의 크기와 에이전트 행동 수는 28과 2로 고정되어 있습니다. 물론 학습 데이터의 특징을 다르게 하고 확률을 예측할 행동을 늘리면 이 수는 바뀝니다.
x[i]에 특징벡터를 지정하고 y_policy[i]에 정책 신경망의 출력을 넣어줍니다. 여기에 DQNLearner에서와 마찬가지로 보상을 구합니다. 이 값에 sigmoid 함수를 취해서 정책 신경망 학습 레이블로 정합니다. sigmoid 함수는 utils 모듈에 구현되어 있으며 5.10절에서 다룹니다.
정책 경사 강화학습에서는 가치 신경망이 없기 때문에 PolicyGradientLearner의 get_batch() 함수는 두 번째 반환값을 None으로 넣습니다.
코드 조각 11: Actor-Critic 강화학습 클래스
Actor-Critic 강화학습은 가치 신경망과 정책 신경망 모두를 사용하는 강화학습 방법입니다. 다음은 ActorCriticLearner 클래스의 생성자를 보여줍니다.
class ActorCriticLearner(ReinforcementLearner):
def __init__(self, *args, shared_network=None,
value_network_path=None, policy_network_path=None, **kwargs):
super().__init__(*args, **kwargs)
if shared_network is None:
self.shared_network = Network.get_shared_network(
net=self.net, num_steps=self.num_steps,
input_dim=self.num_features)
else:
self.shared_network = shared_network
self.value_network_path = value_network_path
self.policy_network_path = policy_network_path
if self.value_network is None:
self.init_value_network(shared_network=shared_network)
if self.policy_network is None:
self.init_policy_network(shared_network=shared_network)
여기서는 가치 신경망과 정책 신경망 모두 생성합니다. 이때 두 신경망은 상단부 레이어들을 공유하게 했습니다. 꼭 두 신경망이 레이어를 공유할 필요는 없으니 레이어 공유를 원하지 않으면 이 부분을 수정하면 됩니다.
예제 5.57 ActorCriticLearner 클래스: 액터-크리틱 강화학습 함수
def get_batch(self, batch_size, delayed_reward, discount_factor):
memory = zip(
reversed(self.memory_sample[-batch_size:]),
reversed(self.memory_action[-batch_size:]),
reversed(self.memory_value[-batch_size:]),
reversed(self.memory_policy[-batch_size:]),
reversed(self.memory_reward[-batch_size:]),
)
x = np.zeros((batch_size, self.num_steps, self.num_features))
y_value = np.zeros((batch_size, self.agent.NUM_ACTIONS))
y_policy = np.full((batch_size, self.agent.NUM_ACTIONS), .5)
value_max_next = 0
reward_next = self.memory_reward[-1]
for i, (sample, action, value, policy, reward) \
in enumerate(memory):
x[i] = sample
y_value[i] = value
y_policy[i] = policy
r = (delayed_reward + reward_next - reward * 2) * 100
y_value[i, action] = r + discount_factor * value_max_next
y_policy[i, action] = sigmoid(value[action])
value_max_next = value.max()
reward_next = reward
return x, y_value, y_policy
DQNLearner 클래스와 PolicyGradientLearner 클래스의 get_batch() 함수를 살펴봤으니 이들과 차이 나는 부분만 짚어보겠습니다.
액터-크리틱 강화학습은 가치 신경망과 정책 신경망 둘 다 사용하기 때문에 가치 신경망 학습 레이블 y_value와 정책 신경망 학습 레이블 y_policy를 모두 가집니다. y_value에는 DQN 강화학습과 동일하게 레이블을 넣어줍니다. 가치 신경망의 예측 상태-행동 가치에 sigmoid 함수를 취해서 정책 신경망 학습 레이블로 넣어줍니다.
시그모이드를 취했기 때문에 정책 신경망 학습 레이블은 예측 가치가 양수면 0.5 이상이 되고 음수면 0.5 미만이 될 것이며 값의 범위는 0~1 사이가 됩니다.
ActorCriticLearner 클래스의 get_batch() 함수는 샘플 배열, 가치 신경망 학습 레이블 배열, 정책 신경망 학습 레이블 배열을 반환합니다.
코드 조각 12: A2C 강화학습 클래스
A2C(advantage actor-critic) 강화학습은 액터-크리틱 강화학습과 거의 유사합니다. 다만 정책 신경망을 학습할 때 가치 신경망의 값을 그대로 사용하지 않고 Advantage를 사용합니다. 예제 5.58은 A2CLearner 클래스를 보여줍니다.
class A2CLearner(ActorCriticLearner):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def get_batch(self, batch_size, delayed_reward, discount_factor):
memory = zip(
reversed(self.memory_sample[-batch_size:]),
reversed(self.memory_action[-batch_size:]),
reversed(self.memory_value[-batch_size:]),
reversed(self.memory_policy[-batch_size:]),
reversed(self.memory_reward[-batch_size:]),
)
x = np.zeros((batch_size, self.num_steps, self.num_features))
y_value = np.zeros((batch_size, self.agent.NUM_ACTIONS))
y_policy = np.full((batch_size, self.agent.NUM_ACTIONS), .5)
value_max_next = 0
reward_next = self.memory_reward[-1]
for i, (sample, action, value, policy, reward) \
in enumerate(memory):
x[i] = sample
r = (delayed_reward + reward_next - reward * 2) * 100
y_value[i, action] = r + discount_factor * value_max_next
advantage = value[action] - value.mean()
y_policy[i, action] = sigmoid(advantage)
value_max_next = value.max()
reward_next = reward
return x, y_value, y_policy
A2CLearner 클래스는 ActorCriticLearner 클래스를 상속합니다. 그렇기 때문에 A2CLearner 클래스의 생성자에서는 부모 클래스의 생성자 호출 이외에 별도로 해줄 것이 없습니다.
A2CLearner 클래스의 get_batch() 함수에서는 Advantage로 정책 신경망을 학습합니다. Advantage는 상태-행동 가치에 상태 가치를 뺀 값입니다. Advantage는 어떠한 상태에서 어떠한 행동이 다른 행동보다 얼마나 더 가치가 높은지를 의미합니다. 여기서는 가치 신경망의 예측 상태-행동 가치 값들을 평균 내어 상태 가치로 사용했습니다. 이렇게 구한 Advantage를 시그모이드 함수에 적용해 정책 신경망의 학습 레이블로 적용했습니다.
코드 조각 13: A3C 강화학습 클래스
A3C(asynchronous advantage actor-critic)는 A2C 강화학습을 병렬로 수행하는 강화학습 방법입니다. A3C 역시 가치 신경망과 정책 신경망을 사용합니다. 다음은 A3C 강화학습을 위한 A3CLearner 클래스의 생성자를 보여줍니다.
class A3CLearner(ReinforcementLearner):
def __init__(self, *args, list_stock_code=None,
list_chart_data=None, list_training_data=None,
list_min_trading_unit=None, list_max_trading_unit=None,
value_network_path=None, policy_network_path=None,
**kwargs):
assert len(list_training_data) > 0
super().__init__(*args, **kwargs)
self.num_features += list_training_data[0].shape[1]
# 공유 신경망 생성
self.shared_network = Network.get_shared_network(
net=self.net, num_steps=self.num_steps,
input_dim=self.num_features)
self.value_network_path = value_network_path
self.policy_network_path = policy_network_path
if self.value_network is None:
self.init_value_network(shared_network=self.shared_network)
if self.policy_network is None:
self.init_policy_network(shared_network=self.shared_network)
# A2CLearner 생성
self.learners = []
for (stock_code, chart_data, training_data,
min_trading_unit, max_trading_unit) in zip(
list_stock_code, list_chart_data, list_training_data,
list_min_trading_unit, list_max_trading_unit
):
learner = A2CLearner(*args,
stock_code=stock_code, chart_data=chart_data,
training_data=training_data,
min_trading_unit=min_trading_unit,
max_trading_unit=max_trading_unit,
shared_network=self.shared_network,
value_network=self.value_network,
policy_network=self.policy_network, **kwargs)
self.learners.append(learner)
A3CLearner 클래스는 ReinforcementLearner 클래스를 상속합니다. 생성자의 인자들은 이전에 살펴본 A2C와는 다르게 리스트로 학습할 종목 코드, 차트 데이터, 학습 데이터, 최소 및 최대 투자 단위를 받습니다. 이 리스트들의 크기만큼 A2CLearner 클래스의 객체들을 생성합니다. learners 리스트에 생성한 A2CLearner 클래스 객체들을 저장해 놓습니다. 각 A2CLearner 클래스 객체는 가치 신경망과 정책 신경망을 공유합니다.
다음은 A3CLearner 클래스의 강화학습 수행 함수를 보여줍니다.
def run(
self, num_epoches=100, balance=10000000,
discount_factor=0.9, start_epsilon=0.9, learning=True):
threads = []
for learner in self.learners:
threads.append(threading.Thread(
target=learner.fit, daemon=True, kwargs={
'num_epoches': num_epoches, 'balance': balance,
'discount_factor': discount_factor,
'start_epsilon': start_epsilon,
'learning': learning
}))
for thread in threads:
thread.start()
time.sleep(1)
for thread in threads: thread.join()
A3C는 A2C를 병렬로 동시에 수행합니다. 가치 신경망과 정책 신경망을 공유하면서 이들을 동시에 학습시킵니다. 하나의 A2CLearner 클래스 객체는 하나의 주식 종목 환경에서 탐험하며 손익률을 높이는 방향으로 가치 신경망과 정책 신경망의 학습을 수행합니다.
A3CLearner 클래스의 run() 함수에서는 스레드를 이용해 각 A2CLearner 클래스 객체의 run() 함수를 동시에 수행합니다. 모든 A2C 강화학습을 마칠 때까지 기다리고 최종적으로 A3C 강화학습을 마칩니다.
파이썬 팁: 파이썬에서 스레드를 사용하려면
threading모듈의 Thread 클래스를 사용합니다. Thread 클래스는 수행할 함수를target인자로 받습니다. 이 타깃 함수에 전달할 인자들은args와kwargs인자로 지정할 수 있습니다.args로 가변 개수 인자들을 튜플로 전달합니다.kwargs에 키워드 인자들을 딕셔너리(dictionary) 형태로 전달합니다.파이썬 팁: Thread 클래스의
daemon인자로 데몬 스레드 여부를 지정할 수 있습니다. 데몬 스레드는 메인 스레드가 종료될 때 함께 종료되는 스레드를 의미합니다.
