맨위로버튼이미지

import random
from collections import deque, namedtuple
from typing import Dict, List, NamedTuple, Optional, Text, Tuple, Union
from IPython.display import display, Math
from Account import Account
import math
from itertools import count

import os.path as path
import numpy as np
import plotly as plt
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
import torchvision.transforms as T
from Market import Market
from PIL import Image
from plotly import express as px
import torchvision
import time
import joblib
import sys,os

action_kind = 4
max_episode = 5000
screen_height = 50
screen_width  = 70
visit_cnt = [0] * action_kind
# replay_buffer = deque()
epsilon = 0.3
dis = 0.9
data_size = 600

BATCH_SIZE = 8
GAMMA = 0.999
EPS_START = 0.9
EPS_END = 0.05
EPS_DECAY = 200
TARGET_UPDATE = 10
steps_done = 0
loss = any

# for i in range(torch.cuda.device_count()):
#     print(torch.cuda.get_device_name(i))

# if gpu is to be used
# device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# device = torch.device("cuda")
# device_str = "cpu"
device_str = "cuda"
device = torch.device(device_str)

from dqn import DQN
    
converter = torchvision.transforms.ToTensor()
market = Market()
policy_net = DQN(device, screen_height, screen_width, action_kind, 32).to(device)
policy_net = nn.DataParallel(policy_net, device_ids=[0,1]).to(device)

def get_chart(market, idx, max_data):
    # img = Image.fromarray(np.uint8(cm.gist_earth(plt.io.to_image(fig, format='png')*255)))
    # im = Image.fromarray(img, bytes=True)
    # im = Image.fromarray(np.uint8(cm.gist_earth(img))/255)
    # im = Image.fromarray(np.uint8(img)/255)
    # img = img.resize((700, 500), resample=Image.BICUBIC)
    # display(img)
    # img = Image.fromarray(cm.gist_earth(plt.io.to_image(fig, format='png'), bytes=True))
    img = market.get_chart(idx, max_data=max_data)
    img.convert("RGB")
    # print(img)
    img.thumbnail((screen_high, screen_width), Image.ANTIALIAS)
    return chart

def plot_durations(last_chart, curr_chart):
    plt.figure()
    # plt.subplot(1,2,1)
    img = plt.imshow(last_chart.cpu().squeeze(0).permute(1, 2, 0).numpy(), interpolation='none')
    plt.title('Example extracted screen')
    plt.figure(2)
    # plt.subplot(1,2,2)
    plt.clf()
    # durations_t = torch.tensor(episode_durations, dtype=torch.float)
    plt.title('Training...')
    plt.xlabel('Episode')
    plt.ylabel('Duration')
    # plt.plot(durations_t.numpy())
    # plt.show()
    # Take 100 episode averages and plot them too
    # if len(durations_t) >= 100:
    #     means = durations_t.unfold(0, 100, 1).mean(1).view(-1)
    #     means = torch.cat((torch.zeros(99), means))
    #     plt.plot(means.numpy())

    img.set_data(curr_chart.cpu().squeeze(0).permute(1, 2, 0).numpy())
    plt.pause(0.01)  # pause a bit so that plots are updated
#     if is_ipython:
    display.clear_output(wait=True)
    display.display(plt.gcf())

def select_action(df, idx):
    return ((df.loc[idx, "close"] - df.loc[idx, "closemin"]) * (action_kind-1) // (df.loc[idx, "closemax"] - df.loc[idx, "closemin"]))
            
def main():
    if path.exists("pt/train_net_{}.pt".format(device_str)):
        policy_net.load_state_dict(torch.load("pt/train_net_{}.pt".format(device_str)))
    policy_net.eval()
    
    for epoch in range(max_episode):
        df = market.get_data()
        account = Account(df, 50000000)
        # account.back_up()
        account.reset()
        count = 0
        correct = 0
        last_chart = None
        while last_chart == None:
            last_chart = get_chart(market, data_size, data_size)

        for idx, _ in enumerate(df.index, start=data_size + 1):
            try:
                since = time.time()
                curr_chart = get_chart(market, idx, data_size)
                if curr_chart is None:
                    continue

                state = curr_chart - last_chart
                curr_chart = last_chart
                dqn_action = policy_net(curr_chart)
                _,dqn_action = torch.max(dqn_action,1)
                dqn_action = dqn_action.cpu().numpy()[0]
                plc_action = select_action(df, idx)
                count += 1
                correct += 1 if dqn_action == plc_action else 0
                
                print("idx:%d==>dqn_action:%d:%d, price:%.2f, match:%.2f"%(idx, dqn_action, plc_action, df.loc[idx, 'close'], correct * 100/count))

                reward, real_action = account.exec_action(2 if dqn_action == (action_kind - 1) else (1 if dqn_action == 0 else 0) , idx)
                reward = torch.tensor([reward], device=device) 
                real_action = torch.tensor([[real_action]], device=device, dtype=torch.long)
                last_chart = curr_chart
                
                if account.is_bankrupt():
                    break
                spend = time.time() - since
                print("idx:%d price [%.4f] unit[%.4f] used time[%.2f] agent rate:%.05f remind money:%.02f" 
                        % (idx, df.loc[idx, 'close'], account.unit, spend, account.rate, account.balance + account.unit * df.loc[idx, 'close']))
            except Exception as ex:
                exc_type, exc_obj, exc_tb = sys.exc_info()
                fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
                print("`epoch loop -> exception! %s : %s %d" % (str(ex) , fname, exc_tb.tb_lineno))

        # torch.save(policy_net.state_dict(),"pt/policy_net.pt")
        # print("epoch[%d] epsode is next loss[%.05f]" % (epoch, loss.item()))
        if path.exists("pt/train_net_{}.pt".format(device_str)) and epoch % 100:
            policy_net.load_state_dict(torch.load("pt/train_net_{}.pt".format(device_str)))
            policy_net.eval()

        # if loss.item() < 0.00001 : 
        #     break
    
    print('Complete')

if __name__ == "__main__":
    main()

오늘은 강화학습으로 pt파일이 만들어졌다면 그것을 이용하여 실제 매매이전에 back test를 진행하는 main_agent.py에 대하여 말씀 드리겠습니다.
get_chart() 함수를 이용하여 torch.tensor로 변환된 PIL 이미지를 받습니다.
최초 실행시 0번 이미지를 미리 받아 두고 0번 이미지와 1번 이미지의 차를 구합니다.
이것이 강화학습에서 말하는 상태 S1이 됩니다.
시간 t1에 대한 S1이 생성되면 이것을 DQN에 넣고 행동 a1을 받습니다.
a1은 0,1,2 중 하나의 값입니다.
0은 관망 1은 매수 2는 매도를 하게 됩니다.
앞 장에서 다룬 내용중 Markget.get_data()함수의 데이터 갯수를 줄이거나 늘려서 매매 횟수를 변경할 수 있으며 epsilone 값을 변경 하면 매매프로그램이 얼마나 자주 모험적인 선택을 할지를 결정 할 수 있습니다.

학습에 사용된 pt/train_net.pt 파일은 학습 및 응용과 지금 main_agent.py에서도 공유됩니다.
공유된 DQN의 하이퍼 파라메트에 의해서 main_agent.py는 훈련 없이 바로 detection이 가능합니다.
실제 실행 되는 상태를 보고 어느 정도의 수익이 발생하는 지를 과거의 데이터로 예상할 수 있습니다.
여기 발견된 오 작동이나 문제점은 main_buroto.py 파일의 recall_history()함수에서 반영되어 새로운 history Series 를 생성하여야 합니다.
새로 생성된 history를 이용하여 학습을 진행 하여 main_agent.py를 실행하면 학습된 내용은 자동으로 반영이 되겠죠.

학습 데이터는 작은 량에서 점점 큰 량으로 변화 시키면서 과적합이 발생하는지 확인 합니다.
학습이 정상적으로 이루어 지면 다시 main_agent.py를 실행하여 detection을 진행한다.
과거의 데이터를 이용하지만 DQN은 학습된 Q(s,a)함수를 실행하는 것이지 과거의 학습된 내용을 바탕으로 한 데이터를 따라 가는 것이 아니다.
실제 history내용과 비교 해 보면 그것을 알 수 있다.
아래 그림은 왼쪽은 과거 데이터를 이용하여 학습을 진행하고 학습된 모델을 이용하여 agent가 실행되는 모습을 담은 사진입니다.
agent가 6%대의 수익을 내고 있으며 학습하는 프로세스가 loss가 적어질 수 록 agent의 수익률은 조금식 증가합니다.
다음 시간에는 실전 매매 프로그램에 DQN 학습 모델을 연동하여 실시간에서는 어떤 동작을 하는지 알아 보겠습니다.

현재 학습이 진행되어 loss가 0.0221정도로 줄어든 로그입니다.

main_agent.py를 실행하면 학습된 training_cuda.pt파일을 읽어서 매매를 실행하여 테스트를 진행하는 사진입니다.
수익률은 44.81%정도 됩니다.
차트를 그리는 시간이 2초 이상걸려서 그리 빠르지는 않습니다.
하루에 약30000건을 테스트 중입니다.
개인 PC사양에 영향을 받을 것 같습니다.

반응형
LIST

맨위로버튼이미지

현재 강화학습 모델은 과거 데이터로 히스토리를 만든 다음 주가차트와 보조지표가 다 있는 이미지로 env의 상태 tensor를 만들어 프로세싱한 모델이 강제로 3가지로 라벨링을 하기 때문에 비슷한 결과에도 어떤때는 보유 또는 관망이 되고 어떤때는 매매가 되므로 optimizer가 하이퍼 파라메트를 결정하기가 쉽지 않게 됩니다. 그래서 loss가 충분히 소멸하지 않고 계속 널뛰기를 하는 현상이 있습니다. DQN 내부의 CNN의 outputs값은 분류 총 값이지만 실제 비지도 학습으로 분류된 값은 약 500~800개 정도로 넓은 범위를 가지고 있습니다. 이것을 최종 nn.Linear함수를 거치면서 모든 값이 소멸하고 3가지로 축소 됩니다. 만약에 차트가 조금 상승할때 팔고 조금하락할때 사고를 반복하면 수익률은 점점 가파르게 감소하여 잔고가 0이 되게 됩니다. 실제 히스토리 함수가 매매를 자주 하지 않는 이유는 작은 변동에 매매가 발생할 경우 수익도 크지 않을뿐더러 낙폭이 크지는 경우 큰 손실이 발생한 다음은 손실을 복구하지 못하는 상황이 발생하게 됩니다. 그래서 매매를 자주 하지 않게 설계되어 있습니다. 이 부분은 실제로는 수학적 손실,손익과는 논리적으로 맞지 않기 때문에 이상적인 목표를 가진 CNN의 입장에서는 모두 loss로 인식이 되게 됩니다. 이 문제를 극복하기 위해서는 CNN자체를 upgrade하는 방법과 CNN과 RNN을 병합하는 방법을 연구해야 합니다. 저도 어느게 맞다고 말할 수 없습니다. 아직도 강화학습은 연구 대상이지 실용단계로 보기는 힘듭니다. RNN이란 말을 들어시고 뭔가 이상함을 느꼈다면 제가 앞에서 이야기한 인간이 개입한 데이터는 시계열 데이터로 보기 힘들다는 이야기 때문일 겁니다. 그러나 RNN은 주가분석같은 시계열 데이터 분석뿐만 아니라 자연어 학습에도 사용됩니다. 우리가 연구 중인 주식이나 코인의 매매 차트를 분류하면 조금 전에 말씀드린것처럼 약 800개의 패턴이 발생합니다. 그 이상을 넘어 가지는 않습니다. RNN 자연어 분석 예제를 보시면 입력값이 입력 글짜 수(여기에 차트의 비지도 학습 분류 패턴 수를 입력),hidden layer 갯수, 출력 글짜 수(여기에 매매프로그램이 사용할 행동갯수를 입력) 입니다. 이 부분을 잘 응용하면 CNN의 출력수(비지도 분류값 약 800개)를 RNN 의 입력값으로 hidden값은 init_hidden함수로 자동 계산되고 출력값은 우리가 구하려는 3가지(매도,매수,관망)일겁니다. 여기에 임베딩(GRU를 사용할 경우 encoding을 사용하여 자연어 분석시 학습률을 높일 수 있음)을 더하면 조금 더 복잡해지겠지만 단순히 차트만으로 분류하는게 아닌 순서적 흐름을 이용한 분류가 가능해 질것 입니다. 대부분의 값은 관망값을 가져야 합니다. 그렇다면 히든값은 무엇을 의미할까요? RNN에 들어갈 첫번째 값이 있다면 이 값은 시간적인 값t0의 값을 가질겁니다. 히든값은 그 다음에 올 수 있는 수 많은 가능성의 t1값의 tensor일 것입니다. 두 값의 연산에 의해서 t1이 결정되면 hidden은 가능성의값 t2의 텐서가 될것이고 그로 인해 실제 t2가 결정될것 입니다. 학습이 진행되면 정확도는 증가할 것으로 예상됩니다.

import torch
import torch.nn as nn
import torch.nn.functional as F

class CNNRNN(nn.Module):
    def __init__(self, device, h, w, outputs, hdnsize):
        super(CNNRNN, self).__init__()
        self.device = device
        self.hidden_size = hdnsize
        self.conv1 = nn.Conv2d(4, h*w, kernel_size=5, stride=2)
        self.bn1 = nn.BatchNorm2d(h*w)
        self.conv2 = nn.Conv2d(h*w, hdnsize, kernel_size=5, stride=2)
        self.bn2 = nn.BatchNorm2d(hdnsize)
        self.conv3 = nn.Conv2d(hdnsize, hdnsize, kernel_size=5, stride=2)
        self.bn3 = nn.BatchNorm2d(hdnsize)
        
        self.i2h = nn.Linear(54 * hdnsize, hdnsize)
        self.h2h = nn.Linear(hdnsize, hdnsize)
        self.i2o = nn.Linear(hdnsize, outputs)
        self.act_fn = nn.Tanh()

    def init_hidden(self):
        return torch.zeros(1, self.hidden_size)
    
    def forward(self, x):
        x = x.to(self.device)
        x = F.relu(self.bn1(self.conv1(x)))
        x = F.relu(self.bn2(self.conv2(x)))
        x = F.relu(self.bn3(self.conv3(x)))
        hidden = self.init_hidden().to(self.device)
        x= self.i2h(x.view(x.size(0), -1))
        hidden = self.h2h(hidden)
        hidden = self.act_fn(x + hidden)
        return self.i2o(hidden)

cnnrnn.py 입니다.

import random
from collections import deque, namedtuple
from IPython.display import display, Math
from Account import Account
import math
from itertools import count
import os

import sys
import os.path as path
import plotly as plt
import torch
import torch.nn as nn
import torch.optim as optim
from Market import Market
import torchvision
import time
from cnnrnn import CNNRNN
from memory import ReplayMemory, Experience

action_kind = 3
max_episode = 5000
screen_height = 100
screen_width  = 140
data_size = 250
visit_cnt = [0] * action_kind
# replay_buffer = deque()
epsilon = 0.3
dis = 0.9

BATCH_SIZE = 8
# GAMMA = 0.999
GAMMA = 0.99
EPS_START = 0.9
EPS_END = 0.05
EPS_DECAY = 200
TARGET_UPDATE = 10
steps_done = 0
loss = any
WINDOW_START = 0
WINDOW_SIZE  = 1500

# for i in range(torch.cuda.device_count()):
#     print(torch.cuda.get_device_name(i))
device_str = "cuda"
device = torch.device(device_str)
    
memory = ReplayMemory(BATCH_SIZE)
converter = torchvision.transforms.ToTensor()
market = Market()
train_net = CNNRNN(device, screen_height, screen_width, action_kind, 32).to(device)
train_net = nn.DataParallel(train_net, device_ids=[0,1]).to(device)

episode_durations = []
    
optimizer = optim.RMSprop(train_net.parameters())

def optimize_action(memory):
    if len(memory) < BATCH_SIZE:
        return None
    epsode = memory.pop(BATCH_SIZE)
    batch = Experience(*zip(*epsode))
    state_batch = torch.cat(batch.state)
    action_batch = torch.cat(batch.action)
    train_net.train()
    state_action_values = train_net(state_batch).gather(1, action_batch)
    
    criterion = nn.SmoothL1Loss()
    loss = criterion(state_action_values, action_batch)

    # Optimize the model
    optimizer.zero_grad()
    loss.backward()
    for param in train_net.parameters():
        param.grad.data.clamp_(-1, 1)
    optimizer.step()
    return loss

def select_action(df, idx):
    try:
        global steps_done
        sample = random.random()
        eps_threshold = EPS_END + (EPS_START - EPS_END) * math.exp(-1. * steps_done / EPS_DECAY)
        steps_done += 1
        if sample > eps_threshold:
            if df.loc[idx, "closemax"] == df.loc[idx, "close"]:
                action = 2
                return  action
            elif df.loc[idx, "closemin"] == df.loc[idx, "close"]:
                action = 1
                return  action
            else:
                return 0
        else:
            action = random.randrange(action_kind)
            return  action
    except Exception as ex:
        exc_type, exc_obj, exc_tb = sys.exc_info()
        fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
        print("`select_action -> exception! %s : %s %d" % (str(ex) , fname, exc_tb.tb_lineno))
        return 0
        
def get_chart(market, idx, max_data):
    img = market.get_chart(idx, max_data=max_data)
    if img is None:
        return None
    # img = Image.fromarray(np.uint8(cm.gist_earth(plt.io.to_image(fig, format='png')*255)))
    # im = Image.fromarray(img, bytes=True)
    # im = Image.fromarray(np.uint8(cm.gist_earth(img))/255)
    # im = Image.fromarray(np.uint8(img)/255)
    # img = img.resize((700, 500), resample=Image.BICUBIC)
    # img = Image.fromarray(cm.gist_earth(plt.io.to_image(fig, format='png'), bytes=True))
    # display(img)
    chart = converter(img).unsqueeze(0).to(device)
    return chart

def plot_durations(last_chart, curr_chart):
    plt.figure()
    # plt.subplot(1,2,1)
    img = plt.imshow(last_chart.cpu().squeeze(0).permute(1, 2, 0).numpy(), interpolation='none')
    plt.title('Example extracted screen')
    plt.figure(2)
    # plt.subplot(1,2,2)
    plt.clf()
    durations_t = torch.tensor(episode_durations, dtype=torch.float)
    plt.title('Training...')
    plt.xlabel('Episode')
    plt.ylabel('Duration')
    plt.plot(durations_t.numpy())
    # plt.show()
    # Take 100 episode averages and plot them too
    if len(durations_t) >= 100:
        means = durations_t.unfold(0, 100, 1).mean(1).view(-1)
        means = torch.cat((torch.zeros(99), means))
        plt.plot(means.numpy())

    img.set_data(curr_chart.cpu().squeeze(0).permute(1, 2, 0).numpy())
    plt.pause(0.01)  # pause a bit so that plots are updated
    display.clear_output(wait=True)
    display.display(plt.gcf())
            
def main():
    if path.exists("pt/train_cnnrnn_{}.pt".format(device_str)):
        train_net.load_state_dict(torch.load("pt/train_cnnrnn_{}.pt".format(device_str)))
        
    for _ in range(10):    
        df = market.get_data()
        # df = df.head(WINDOW_SIZE)

        for epoch in range(max_episode):
            account = Account(df, 50000000)
            account.reset()

            last_chart = get_chart(market, data_size, data_size)
            account.reset()

            for idx,_ in enumerate(df.index, start=(data_size + 1)):
                try:
                    since = time.time()
                    curr_chart = get_chart(market, idx, data_size)
                    if curr_chart is not None:
                        state = curr_chart - last_chart
                    else:
                        continue
                    last_chart = curr_chart
                    reward = 0
                    num_action = select_action(df, idx)
                    reward, real_action = account.exec_action(num_action, idx)
                    print("idx:%d==>action:%d, price:%.2f"%(idx, num_action, df.loc[idx, 'close']))
                    reward = torch.tensor([reward], device=device)
                    action = torch.tensor([[num_action]], device=device, dtype=torch.int64)
                                    
                    memory.push(state, action, reward)
                    while len(memory) >= BATCH_SIZE:
                        loss = optimize_action(memory)
                        if loss is not None:
                            print("epoch[%d:%d] epsode is next loss[%.10f]" % (epoch, idx, loss.item()))

                    if idx % TARGET_UPDATE == 0:
                        torch.save(train_net.state_dict(),"pt/train_cnnrnn_{}.pt".format(device_str))
                    
                    spend = time.time() - since
                    print("idx:%d price [%.4f] unit[%.4f] used time[%.2f] agent rate:%.05f remind money:%.02f" 
                            % (idx, df.loc[idx, 'close'], account.unit, spend, account.rate, account.balance + account.unit * df.loc[idx, 'close']))
                    if account.is_bankrupt():
                        break
                    if idx == df.index.max():
                        break
                except Exception as ex:
                    exc_type, exc_obj, exc_tb = sys.exc_info()
                    fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
                    print("`recall_training -> exception! %s : %s %d" % (str(ex) , fname, exc_tb.tb_lineno))

        print("end training DQN")
    
    print('Complete Training')

if __name__ == "__main__":
    main()

main_cnnrnn.py 입니다.
CNN 단독 모델과 CNN + RNN 모델의 학습 증가율을 비교해 보시기 바랍니다.

대표이미지출처:https://blog.kakaocdn.net/dna/IFp6A/btqASygMYCA/AAAAAAAAAAAAAAAAAAAAAHPRKM8z8_i1rJvLfrqQ8vMKOITsjcC7Q0zi2JxkpFQQ/img.png?credential=yqXZFxpELC7KVnFOS48ylbz2pIh7yKj8&expires=1759244399&allow_ip=&allow_referer=&signature=mFEIH8jDzqhtbOVg9uF%2FFSmH2NY%3D
RNN 소스코드 출처 : 책 파이토치 첫걸음에서 발췌 및 응용

 

반응형
LIST

맨위로버튼이미지

import random
from collections import deque, namedtuple
from typing import Dict, List, NamedTuple, Optional, Text, Tuple, Union
# from IPython.display import display, Math
from Account import Account
import math
from itertools import count
import os

import sys
import os.path as path
import numpy as np
import plotly as plt
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
import torchvision.transforms as T
from Market import Market
from PIL import Image
from plotly import express as px
import torchvision
import time
import joblib
import pandas as pd
from dqn import DQN
from memory import ReplayMemory, Experience

action_kind = 21
max_episode = 5000
screen_height = 100
screen_width  = 140
data_size = 600
epsilon = 0.3
dis = 0.9

BATCH_SIZE = 8
# GAMMA = 0.999
GAMMA = 0.99
EPS_START = 0.9
EPS_END = 0.05
EPS_DECAY = 200
TARGET_UPDATE = 10
steps_done = 0
loss = any
WINDOW_START = 0
WINDOW_SIZE  = 1500

# for i in range(torch.cuda.device_count()):
#     print(torch.cuda.get_device_name(i))

# if gpu is to be used
# device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# device = torch.device("cuda")
# device_str = "cpu"
device_str = "cuda"
device = torch.device(device_str)
    
memory = ReplayMemory(BATCH_SIZE)
converter = torchvision.transforms.ToTensor()
market = Market()
train_net = DQN(device, screen_height, screen_width, action_kind, 32).to(device)
train_net = nn.DataParallel(train_net, device_ids=[0,1]).to(device)

episode_durations = []
    
optimizer = optim.RMSprop(train_net.parameters(), lr=0.000001)

# def select_action(df, idx):
#     try:
#         global steps_done
#         sample = random.random()
#         eps_threshold = EPS_END + (EPS_START - EPS_END) * math.exp(-1. * steps_done / EPS_DECAY)
#         steps_done += 1
#         clslvl = ((df.loc[idx, "close"] - df.loc[idx, "closemin"]) * 40 // (df.loc[idx, "closemax"] - df.loc[idx, "closemin"]))
#         if sample > eps_threshold:
#             if df.loc[idx, "closemax"] == df.loc[idx, "close"]:
#                 action = clslvl
#                 return  action
#             elif df.loc[idx, "closemin"] == df.loc[idx, "close"]:
#                 action = clslvl
#                 return  action
#             else:
#                 return clslvl
#         else:
#             action = random.randrange(action_kind)
#             return  action
#     except Exception as ex:
#         exc_type, exc_obj, exc_tb = sys.exc_info()
#         fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
#         print("`select_action -> exception! %s : %s %d" % (str(ex) , fname, exc_tb.tb_lineno))
#         return 0

def select_action(df, idx):
    return ((df.loc[idx, "close"] - df.loc[idx, "closemin"]) * (action_kind-1) // (df.loc[idx, "closemax"] - df.loc[idx, "closemin"]))
        
def get_chart(market, idx, max_data):
    img = market.get_chart(idx, max_data=max_data)
    if img is None:
        return None
    # if  idx > (df.index.max() - data_size - 1):
    #     return None
    # img = Image.fromarray(np.uint8(cm.gist_earth(plt.io.to_image(fig, format='png')*255)))
    # im = Image.fromarray(img, bytes=True)
    # im = Image.fromarray(np.uint8(cm.gist_earth(img))/255)
    # im = Image.fromarray(np.uint8(img)/255)
    # img = img.resize((700, 500), resample=Image.BICUBIC)
    # display(img)
    # img = Image.fromarray(cm.gist_earth(plt.io.to_image(fig, format='png'), bytes=True))
    chart = converter(img).unsqueeze(0)
    return chart

def plot_durations(last_chart, curr_chart):
    plt.figure()
    # plt.subplot(1,2,1)
    img = plt.imshow(last_chart.cpu().squeeze(0).permute(1, 2, 0).numpy(), interpolation='none')
    plt.title('Example extracted screen')
    plt.figure(2)
    # plt.subplot(1,2,2)
    plt.clf()
    durations_t = torch.tensor(episode_durations, dtype=torch.float)
    plt.title('Training...')
    plt.xlabel('Episode')
    plt.ylabel('Duration')
    plt.plot(durations_t.numpy())
    # plt.show()
    # Take 100 episode averages and plot them too
    if len(durations_t) >= 100:
        means = durations_t.unfold(0, 100, 1).mean(1).view(-1)
        means = torch.cat((torch.zeros(99), means))
        plt.plot(means.numpy())

    img.set_data(curr_chart.cpu().squeeze(0).permute(1, 2, 0).numpy())
    plt.pause(0.01)  # pause a bit so that plots are updated
    # display.clear_output(wait=True)
    # display.display(plt.gcf())
            
def main():
    if path.exists("pt/train_net_{}.pt".format(device_str)):
        train_net.load_state_dict(torch.load("pt/train_net_{}.pt".format(device_str)))
        
    for _ in range(10):    
        df = market.get_data()
        # df = df.head(WINDOW_SIZE)

        for epoch in range(max_episode):
            account = Account(df, 50000000)
            account.reset()

            last_chart = get_chart(market, data_size, data_size)
            account.reset()

            for idx,_ in enumerate(df.index, start=(data_size + 1)):
                try:
                    since = time.time()
                    curr_chart = get_chart(market, idx, data_size)
                    if curr_chart is None or last_chart is None:
                        continue
                    else:
                        state = curr_chart - last_chart
                        
                    last_chart = curr_chart
                    reward = 0
                    num_action = select_action(df, idx)
                    reward, real_action = account.exec_action(2 if num_action == (action_kind - 1) else (1 if num_action == 0 else 0) , idx)
                    print("idx:%d==>action:%d,real_action==>%d price:%.2f"%(idx, num_action, real_action, df.loc[idx, 'close']))
                    # reward = torch.tensor([reward], device=device)
                    num_action = torch.tensor([[num_action]], device=device, dtype=torch.int64)
                                    
                    memory.push(curr_chart, num_action, reward)
                    if len(memory) >= BATCH_SIZE:
                        epsode = memory.pop(BATCH_SIZE)
                        batch = Experience(*zip(*epsode))
                        # non_final_mask = torch.tensor(tuple(map(lambda s: s is not None,
                        #                                       batch.next_state)), device=device, dtype=torch.bool)
                        # non_final_next_states = torch.cat([s for s in batch.next_state
                        #                                             if s is not None])
                        state_batch = torch.cat(batch.state)
                        action_batch = torch.cat(batch.action)
                        # reward_batch = torch.cat(batch.reward)
                        train_net.train()
                        state_action_values = train_net(state_batch).gather(1, action_batch)
                        # next_state_values = torch.zeros(BATCH_SIZE, device=device)
                        # next_state_values[non_final_mask] = train_net(non_final_next_states).max(1)[0].detach()
                        # expected_state_action_values = (next_state_values * GAMMA) + reward_batch
                        # print(action_batch)
                        # print(state_action_values)
                        optimizer.zero_grad()
                        criterion = nn.SmoothL1Loss()
                        loss = criterion(state_action_values, action_batch)
                        # print(loss)
                        # loss = criterion(state_action_values, action_batch.unsqueeze(1))
                        # loss = criterion(market_status, train_status)

                        # Optimize the model
                        loss.backward()
                        for param in train_net.parameters():
                            param.grad.data.clamp_(-1, 1)
                        optimizer.step()
                        print("epoch[%d:%d] epsode is next loss[%.10f]" % (epoch, idx, loss.item()))

                    if idx % TARGET_UPDATE == 0:
                        torch.save(train_net.state_dict(),"pt/train_net_{}.pt".format(device_str))
                    
                    spend = time.time() - since
                    print("idx:%d price [%.4f] unit[%.4f] used time[%.2f] agent rate:%.05f remind money:%.02f" 
                            % (idx, df.loc[idx, 'close'], account.unit, spend, account.rate, account.balance + account.unit * df.loc[idx, 'close']))
                    if account.is_bankrupt():
                        break
                    if idx == df.index.max():
                        break
                        # train_net.load_state_dict(train_net.state_dict())
                    # save_file_model(degreeQ)                
                except Exception as ex:
                    exc_type, exc_obj, exc_tb = sys.exc_info()
                    fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
                    print("`recall_training -> exception! %s : %s %d" % (str(ex) , fname, exc_tb.tb_lineno))

        print("end training DQN")
    
    print('Complete Training')

if __name__ == "__main__":
    main()

main_buroto.py 파일입니다.

import random
from collections import deque, namedtuple

Experience = namedtuple(
    'Experience',
    ('state', 'action', 'reward')
)

class ReplayMemory(object):
    def __init__(self, capacity):
        self.memory = deque([],maxlen=capacity)

    def push(self, *args):
        """Save a transition"""
        self.memory.append(Experience(*args))

    def pop(self, batch_size):
        arr = []
        for _ in range(batch_size):
            arr.append(self.memory.popleft())
        return arr
    
    def sample(self, batch_size):
        return random.sample(self.memory, batch_size)

    def __len__(self):
        return len(self.memory)

memory.py 부분을 분리했습니다.

import torch.nn as nn
import torch.nn.functional as F

class DQN(nn.Module):
    def __init__(self, device, h, w, outputs, qsize):
        super(DQN, self).__init__()
        self.conv1 = nn.Conv2d(4, h*w, kernel_size=5, stride=2)
        self.bn1 = nn.BatchNorm2d(h*w)
        self.conv2 = nn.Conv2d(h*w, qsize, kernel_size=5, stride=2)
        self.bn2 = nn.BatchNorm2d(qsize)
        self.device = device
        self.conv3 = nn.Conv2d(qsize, qsize, kernel_size=5, stride=2)
        self.bn3 = nn.BatchNorm2d(qsize)

        # Number of Linear input connections depends on output of conv2d layers
        # and therefore the input image size, so compute it.
        def conv2d_size_out(size, kernel_size = 5, stride = 2):
            return (size - (kernel_size - 1) - 1) // stride  + 1
        convw = conv2d_size_out(conv2d_size_out(conv2d_size_out(w)))
        convh = conv2d_size_out(conv2d_size_out(conv2d_size_out(h)))
        # print("convw[%d]  convh[%d]" % (convw, convh))
        linear_input_size = 54 * qsize
        self.head = nn.Linear(linear_input_size, outputs)

    # Called with either one element to determine next action, or a batch
    # during optimization. Returns tensor([[left0exp,right0exp]...]).
    def forward(self, x):
        x = x.to(self.device)
        x = F.relu(self.bn1(self.conv1(x)))
        x = F.relu(self.bn2(self.conv2(x)))
        x = F.relu(self.bn3(self.conv3(x)))
        return self.head(x.view(x.size(0), -1))

dqn.py를 분리했습니다.

main함수부터 설명 드리면 가장 이상적인 매매 프로세스인 저점에서 매수하고 고점에서 매도하는 방식을 rolling함수를 이용하여 구하였습니다.
그리고 이것을 전체 데이터에 적용을 하여 history에 저장합니다.
매매 길이에 따라서 수천프로이상의 수익이 발생할 것입니다.
이것은 말 그대로 이상적인 매매 프로세스 입니다.
백프로 매치되는 것은 아니지만 이상적인 grid world의 마르코프 실행 프로세서(MDP) 처럼은 우리도 이것을 MDP로 가정합니다.
이제 실행된 결과를 DQN(CNN)에 학습합니다.
optimize함수가 그 역할을 합니다.
loss함수는 nn.SmoothL1Loss를 사용하여 해당 S의 action a와 기대 이상 행동 a'를 구하여 둘의 차이를 구하여 계산합니다.
아쉽지만 windows에서는 NCCL이 적용되지 않아 pytorch함수에서 에러가 발생합니다.
NCCL은 nvidia의 GPU를 서로 연동하여 최상의 성능을 내는 Api lib입니다.
현재 ubuntu와 red hat에서만 지원하고 있습니다.
그래서 지금 저는 WSL2를 이용한 ubuntu LTS(최종안정화버젼)인 20.04.5버젼으로 이전을 하려고 하는데 windows10과 wsl2의 병행 운영은 엄청난 cpu 제원을 요구 하네요.
어쩔 수 없이 서버를 ubuntu로 다시 설치하고 노트북으로 접속하여 이용하는 방법을 실행하려고 하고 있습니다.
불행히도 제 노트북이 현재 아작이 난 상태라 알리에서 검색해서 LCD판넬을 주문한 상태입니다만 가장 빠른 날짜가 12월 15일 도착이라고 하네요.
Nvidia 홈페이지 상으로는 현재로써는 windows에 NCCL이 적용이 안되어서 여러장의 GPU카드가 있어도 한장의 효과만 있다는 거죠.
현재 소스에서 보시면 nn.DataParallel이 DQN을 감싸고 있습니다.
실제 성능탭으로 보면 그래도 nn.DataParallel을 적용하기 전에는 한개의 GPU만 사용하게 작동이 되었으나 지금은 어느 정도 반반씩 부담하여 사용하는게 보입니다.
중요한 점은

recall_history()함수에서 저장된 데이터를 recall_training 함수에서 DQN을 학습하게 됩니다. 학습된 결과는 pt/training_net.pt파일로 저장을 하게 됩니다. 저의 경우는 데이터가 너무 많은 관계로 과적합 상태를 보이나 로스차이가 그렇게 크지는 않습니다. 적당한 데이터의 크기는 5000~10000개 정도의 데이터를 반복 학습하면서 loss를 최소화 한다음 다시 데이터를 샘플링하고 적용하고 하는 반복 작업을 통하여 실제 DQN이 적은 loss로 동작하는지가 중요합니다. 

이 작업이 어느 정도 끝나면 다음은 main_agent.py 파일을 작성하여 실제 상황에서 detection을 발생하여 거래를 하게 하여 어느 정도의 수익률이 나오는지 분석을 해 보도록 하겠습니다.
저도 처음에는 강화학습이 데이터가 많이 적용되면 더욱 딕텍션이 안정될 거라 생각했지만 실제로는 데이터가 많을 수록 ai는 거래를 하지 않으려하네요.
프로그램을 조금 수정해서 pt파일을 여러 버젼으로 저장하는게 어떨까 제안 합니다.
다음 차에서 agent파일을 돌려 보면 알겠지만 DQN은 가장 안정화된 방법을 선택하려 합니다.
즉 매매를 하지 않아 수익도 손실도 없는 상태를 만들려고 합니다.
그렇지만 어떤 경우 훈련이 잘 진행되어 100%이상의 수익을 내기도 합니다.
아무런 if문 없이 수익을 낸다는 게 신기합니다만 retro.gym을 경험해 보신분들은 아실 겁니다.
사실 retro.gym을 훈련하는 과정에도 인간의 생각 판단 따위는 중요하지 않습니다.
훈련은 우연의 연속입니다.
다만 그 우연들 중에서 가장 스코어가 높은 쪽으로만 AI는 움직이려 합니다.
그렇게 하는것은 인간의 계산에서 epsilon값을 어떻게 판단하는냐에 따라 결과가 달라 질 겁니다.
epsilon을 0.5 정도로 하면 절반은 모험값으로 채워질 겁니다.
그리고 결과를 수익률 또는 남은 돈으로 정할 때 그 값이 최고가 되는 모델들만 history에 저장하고 학습을 돌린다면 엔진은 공격적인 매매를 해서 높은 수익률을 올릴 것입니다.
다음 편 실 agent 시뮬레이션 편을 기대 해 주세요.
 

 

반응형
LIST

맨위로버튼이미지

우리가 우리 생의 모든 부분을 태어날 때부터 죽을 때까지를 다 알고 있다면 우리는 과연 어떤 모습으로 살고 있을 까요?
타임루프 물 애니나 영화에서는 그런 장면이 많이 나옵니다. 자신의 절대 위기나 문제를 이미 경험해 보았기 때문에 모든 문제를 피해 갑니다. 니콜라스 케이지의 2007년 작 넥스트라는 작품이 있습니다. 실제 이 영화는 톰 크루저 주연의 마이너리티 리포트와 같은 소설을 원작으로 하고 있지만 내용은 완전히 다릅니다. 니콜라스 케이지는 한 여자를 운명적으로 좋아하지만 그녀를 만나게 되면 세상은 적어도 미국은 핵폭발로 멸망하게 됩니다. 영화는 니콜라스 케이지의 3분 전을 내다보는 예지 능력의 이야기를 다루지만 2시간 동안의 이야기의 내용은 그래서 아무 일도 안 하고 안 일어났다는 거구 니콜라스 케이지는 몇 번이나 한 여자를 만나서 사랑을 하지만 그녀는 한 번도 니콜라스를 만나지 않았다는 내용의 끝은 좀 허무한 그런 내용입니다. 단 3분 앞을 내다보지만 그 3분만으로 니콜라스 케이지는 아주 쉽게 돈을 벌고 아주 쉽게 FBI를 따 돌리고 세상을 구하지만 결말은 항상 핵폭발이라 그냥 포기하고 혼자 외로운 인생을 산다는 그런 이야기입니다. 우리도 니콜라스 케이지 정도의 예지력은 아니지만 저는 upbit의 이더리움 데이터 1년 6개월치를 가지고 있습니다. Market class의 get_data함수를 보면 아래 부분이 추가되어 있습니다. 주식이던 코인이던 가장 많은 돈을 버는 방법은 가장 고점에서 팔고 가장 저점에서 사고를 계속하면 무조건 돈을 번다는 거죠. python의 rolling함수를 이용하여 앞으로 50 뒤로 50 범위의 수중 가장 높은 값을 미리 구하고 index를 계속 돌리면서 사고팔고를 계속하고 그 중간에는 관망을 하면 어떤 일이 있을 가요? 거의 천문학적인 돈을 벌 수 있습니다. 이런 일이 안 일어나는 이유는 50개의 데이터가 예지의 영역에 있기에 실제로는 가져올 수 없는 값이라는 거죠.

        self.df['closemax'] = self.df.close.rolling(window=100, center=True).max()
        self.df['closemin'] = self.df.close.rolling(window=100, center=True).min()​

위의 내용에서 closemax와 closemin을 미리 계산합니다. 아래 select_action은 제가 말씀 드린 내용으로 매매를 하지만 epsilon값을 구하여 그 값의 하향선에서만 정책적인 매매를 하고 그 이상이 되면 random행동을 한다는 거죠. 예를 들면
섬에 표류한 한 사람이 그 섬에 있는 5가지 열매 중 가장 첫번째 먹어서 죽지 않은 음식을 구출될 때까지 계속 먹게 된다는 이야기와 같은 원리로 머쉰 러닝의 결과도 한 가지 결과 중에서 최고값이 나오면 더 이상은 다른 값을 찾아가지 않습니다. 그래서 프로그래머는 강제로 모험을 하는 확률을 넣어서 모험을 실행하는 계수만큼 모험을 하게 만들어 진짜 최적의 값이 지금 현재 머쉰 러닝이 찾아낸 값인지 계속 질문을 합니다. 아래 함수는 실제 거래에서는 불가능하지만 우리는 이미 거래의 결과를 다 알고 있기 때문에 아래의 방식으로 최고 수익을 내는 방법을 history Series에 저장할 겁니다. 아래는 그 소스입니다. 아래의 행동을 policy action(정책적인 행동)이라고 합니다. 이 부분은 DQN과는 아무 관련이 없으며 인간이 최적의 경로를 찾아가는 MDP(마르코프 실행 프로세스)를 주식이나 코인의 차트에 대입해 본 것입니다. 자 이제는 우리는 최적의 방법으로 미친 수익률을 내는 history를 찾을 수 있습니다. 현실에서는 불가능 하지만 몇천 배의 수익을 발생하게 됩니다. 그야말로 이대로만 되며 누구나 부자가 될 수 있게 죠.

def select_action(df, idx):
    try:
        global steps_done
        sample = random.random()
        eps_threshold = EPS_END + (EPS_START - EPS_END) * math.exp(-1. * steps_done / EPS_DECAY)
        steps_done += 1
        if sample > eps_threshold:
            if df.loc[idx, "closemax"] == df.loc[idx, "close"]:
                action = 2
                return  action
            elif df.loc[idx, "closemin"] == df.loc[idx, "close"]:
                action = 1
                return  action
            else:
                return 0
        else:
            action = random.randrange(action_kind)
            return  action
    except Exception as ex:
        exc_type, exc_obj, exc_tb = sys.exc_info()
        fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
        print("`recall_training -> exception! %s : %s %d" % (str(ex) , fname, exc_tb.tb_lineno))
        return 0
  
  def get_chart(market, df, idx, max_data):
    img = market.get_chart(idx, max_data=max_data)
    if img is None:
        return None
    if  idx > (df.index.max() - 299):
        return None
    # img = Image.fromarray(np.uint8(cm.gist_earth(plt.io.to_image(fig, format='png')*255)))
    # im = Image.fromarray(img, bytes=True)
    # im = Image.fromarray(np.uint8(cm.gist_earth(img))/255)
    # im = Image.fromarray(np.uint8(img)/255)
    # img = img.resize((700, 500), resample=Image.BICUBIC)
    # display(img)
    # img = Image.fromarray(cm.gist_earth(plt.io.to_image(fig, format='png'), bytes=True))
    chart = converter(img).unsqueeze(0).to(device)
    return chart
  
 def recall_history(df, account, idx, re_action):
    max_price = -np.inf
    min_price = np.inf
    max_reward = -np.inf
    max_rew_step = 0
    min_rew_step = 0
    save_point_idx = 0
    account.reset()
        
    for _, idx in enumerate(df.index, start=idx):
        try:
            max_step = idx
            since = time.time()
            if re_action != 0:
                action = re_action
                re_action = 0
            else:
                action = select_action(df, idx)
            
            print("idx:%d==>policy_action:%d"%(idx, action))

            reward, real_action = account.exec_action(action, idx)
            history.loc[idx, "action"] = real_action
            max_reward = reward if reward > max_reward else max_reward
            history.loc[idx, 'max_rew'] = max_reward

            if df.loc[idx, 'close'] > max_price:
                max_price = df.loc[idx, 'close']  
                account.back_up()
                max_rew_step = 0
                save_point_idx = idx
            else:
                max_rew_step += 1
            
            if df.loc[idx, 'close'] < min_price:
                min_price = df.loc[idx, 'close']
                account.back_up()
                min_rew_step = 0
                save_point_idx = idx
            else:
                min_rew_step += 1
            
            # if account.unit > 0 and max_rew_step > 50:
            #     return max_step, True, save_point_idx, 2
            # elif account.unit == 0 and min_rew_step > 50:
            #     return max_step, True, save_point_idx, 1
            
            if account.is_bankrupt():
                break
            spend = time.time() - since
            print("idx:%d used time[%.2f] price[%.2f] unit[%.2f] agent rate:%.05f remind money:%.02f" 
                    % (idx, spend, df.loc[idx, 'close'], account.unit, account.rate, account.balance + account.unit * df.loc[idx, 'close']))
            # if epoch % TARGET_UPDATE == 0:
            #     target_net.load_state_dict(train_net.state_dict())
            # torch.save(train_net.state_dict(),"pt/train_net.pt")
            # save_file_model(degreeQ)                
        except Exception as ex:
            exc_type, exc_obj, exc_tb = sys.exc_info()
            fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
            print("`recall_history -> exception! %s : %s %d" % (str(ex) , fname, exc_tb.tb_lineno))
    return max_step, False, 0, 0

다음 차수에는 드디어 DQN(CNN) 학습으로 Q(s, a) 함수를 구현해 보도록 하겠습니다. 이전 차수의 상태 값 Sn이 바로 env입니다. 이미 우리는 env를 구했습니다. 그다음은 env에 해당 행동을 적용 했을때 기대값과 실행 후의 상태 Sn을 구해야 하는 문제가 남게 되는데 그것은 Account 클래스를 통해서 구현 해 보도록 하겠습니다. 프로세스는 2개가 동시에 실행되어야 합니다. 지금 우리가 만들고 있는 프로그램은 제 기준으로 1주일에 1 epoch를 도는 굉장히 느린 프로그램입니다. 돈이 좀 더 있다면 DQN 계산 능력을 향상 시키겠지만 현재로는 한 스텝에 평균 8초 정도 걸립니다. 이 시간은 차트 Cn - C(n-1)을 해서 상태 Sn을 구하고 이것을 CNN으로 학습하고 loss를 구하여 optimize까지 실행한 결과 입니다.
방식은 위의 이상적인 history를 먼저 실행하여 실행 행동의 행렬을 구하고 그 다음은 DQN학습을 진행하는 방식입니다.
이렇게 학습된 모델은 어떤 특정 디렉터리(pt/training_net.pt) 파일로 저장이 되고 학습을 전혀 하지 않고 DQN으로 행동만을 받아서 실행하는 또 다른 프로세스에서 모델 파일을 읽어서 셋업을 한 다음 실 데이터로 거래를 하는 형식으로 구성이 될 겁니다. 처음에는 진짜 리얼 데이터가 아니라 학습 데이터로 실행을 할 것이고 이후 학습이 되는대로 실 데이터로 실전을 할 것입니다.

 

 

반응형
LIST

맨위로버튼이미지

주식이나 코인 매매 프로그램을 너무 쉽게 생각하시면 큰일 납니다. 제 경험이기도 합니다. 일단 머쉰 러닝을 배우신 분들 중에는 주식차트를 시계열 데이터로 착각하시는 분들이 많습니다. 물론 시계열 데이터이기도 합니다. 시계열 데이터의 가장 큰 특징은 무엇일까요? 정답을 아시는 분들도 있겠지만 바로 계절성입니다. 예를 들면 따릉이의 이용률을 데이터로 변환한다면 봄가을에는 많은 사람들이 이용해서 자전거가 부족하지만 여름이나 겨울에 그 수가 확실히 감소합니다. 특히 겨울에는 이용률이 크게 감소하죠. 따릉이 자전거를 자동 배치하는 AI를 개발한다면 크게 어렵지 않게 개발이 가능하겠죠. 그렇다면 시계열 데이터 방식으로 예를 들면 LSTM 방식으로 삼성전자의 주가를 예측해보신 분들은 나름 데이터가 맞는 것 같아서 신기해합니다. 하지만 그것은 예측된 데이터로 보기엔 문제가 있습니다. 인간이 개입하는 어떤 데이터도 시계열 데이터의 특징인 주기성을 보이지는 않습니다. 그저 서로를 속이는 눈속임일 뿐입니다. 아쉽게도 LSTM이나 RNN, GRU 방식 등의 방식으로 주가를 예측하는 것은 애초에 불가능 합니다.https://m.blog.naver.com/dsz08082/222042702104
위 url의 내용을 한번 참고해 보세요. 앞에서도 말씀드렸듯이 주식, 코인을 자동으로 매매하는 프로그램은 정말 쉽지 않습니다. 인터넷에 통장까지 공개하시는 분들이 많은데 이 번의 사태를 견뎠을까요? 주식 또한 마찬가지로 동일본 지진이나 리먼사태 등을 예측할 수는 없었습니다. 그렇다면 강화 학습으로 그런 문제를 극복할 수 있을까요? 사실 저도 자신이 없네요. 그래서 브레이크 장치가 필요합니다. 그래도 주식은 제도권 안에 있기에 법적인 구제 장치가 많지만 코인 시장은 그렇지 않습니다. 혹시나 인터넷이나 유튜브를 떠도는 프로그램들에는 눈도 돌리지 마시기 바랍니다. 제가 쓴 글들 중에도 몇몇 것은 참담한 실패를 맛봤습니다. 가끔 사람은 세상을 너무 만만하게 보거나 짧은 생각으로 글을 쓸 수 있습니다. 지금 우리가 하는 길이 만만하지 않음을 꼭 아셨으면 합니다. 오늘은 CNN과 상태(S)와의 관계에 대하여 생각해보겠습니다. 코인 차트는 시간(t)에 따라 C0
, C1, C2, ... , Ct-1, Ct, ... 이렇게 생성이 될 겁니다. pytorch 의 tensor데이터는 다차원의 행렬로 이루어져 있습니다. PIL이미지를 torch.tensor로 변환하는 함수를 지원합니다. 차트 이미지를 torch.tensor로 변경했다면 C2 - C1을 하면 t1의 상태(S)가 됩니다. 원래는 RGB데이터로 이미지를 저장하고자 했지만 저도 이미지 쪽은 잘 몰라 ARGB방식 즉 4차원 torch.tensor로 저장이 되었습니다. CNN은 합성곱 인공신경망의 약자입니다. 이미지의 특징을 검색하기 위해 bmp방식으로 저장하게 되면 메모리 overflow로 프로그램은 더 이상 진행하지 않고 런타임 에러가 나거나 엄청난 컴퓨터 비용을 지불해야 할 것입니다. 그래서 화면의 일부를 차원곱을 사용하여 이미지를 쭉 따라서 검색하는 방식으로 이미지를 분석합니다. 인터넷에 있는 개와 고양이를 구분하는 예제는 고양이 코 모양과 개의 코 모양이 완전히 다르다는 것을 신경망 학습이 비지도 학습으로 알아내고 클래스(분류 가짓수)를 2라고 한정했기 때문에 고양이의 코 모양을 0으로 했다면 개의 코 모양은 1로 해서 분류를 하고 학습을 반복할수록 지도 학습으로 정확도를 높이게 됩니다. 학습이 끝나면 거의 100%의 정확도로 고양이와 개를 구분합니다. 즉 이미지의 모든 부분을 이용하는 것이 아니라 특징이 다른 부분을 합성곱으로 빨리 탐색하고 분류하는 것이죠. 다시 주식차트로 오면 C2 - C1을 하게 되면 C2와 C1의 동일한 부분은 모두 0이 될 겁니다. 행렬의 차는 같은 위치의 값을 서로 빼는 것이니 당연히 같은 좌표에 있는 이미지 중에 같은 값을 가진 것들은 전부 0으로 바뀔 겁니다. 그럼 C2 - C1을 S1이라고 하면 대분은 0인 이미지가 나오겠죠. 그리고 변화가 생긴 부분만 남게 됩니다. 그래서 우리가 만드는 차트에 라벨링은 의미가 없습니다. 같은 위치에 반복적으로 쓰이는 값은 삭제될 뿐입니다. 그다음 S1을 합성곱으로 검색해서 지나가는 부분은 비지도 학습에 의해 가짓수를 CNN이 결정합니다. 다시 우리가 지정한 output개수 즉 class의 가짓수에 의해서 다시 CNN은 지도 학습을 진행하면서 각각의 클래스에 해당하는 숫자를 지정하게 됩니다. 학습을 진행하면 우리 인간은 눈치 못 챈 어떤 특징으로 인해 차트는 우리가 지정한 가지 수만큼의 output값을 지정하게 됩니다. 우리는 이 값을 행동 0,1,2로 지정합니다. 저희 프로그램은 100% 매수 또는 매도 또는 관망의 3가지 행동 뿐입니다. CNN은 우리의 의도대로 차트를 가지고 우리가 실행해야 하는 행동을 구하게 됩니다.
어렵지 않게 설명하려고 많이 노력했는데 괜찮았는지 모르겠네요. 상태 값과 CNN의 관계를 이해하셨다면 마르코프 실행 프로세스에 해당하는 부분을 설명하겠습니다. 혹시 오류를 발견하셨다면 댓글 부탁드립니다.

 

 

반응형
LIST

맨위로버튼이미지

지금은 코인을 중심으로 프로그램을 작성하지만 주식에도 그대로 적용이 가능합니다. 코인 거래소는 upbit를 이용하고 있습니다. 지금은 코인원이 수수료가 가장 싸므로 api를 조금 수정하시면 코인원식으로 변경이 가능 할 겁니다. 강화학습에서 중요한 포인트는 env와 agent입니다. 코인에서는 env는 마켓으로 agent는 account라고 저는 명명을 했습니다. 꼭 그렇게 않해도 됩니다. 여기에 게시되는 소스는 아무런 라이센스 없이 사용하셔도 됩니다. 우리가 사용할 DQN에 대하여 간단하게 설명 드리겠습니다. 그전에 참고할 만한 책을 한권 소개 드립니다. '바닥부터 배우는 강화학습'(https://www.googleadservices.com/pagead/aclk?sa=L&ai=DChcSEwj6gIbC29D7AhVy3EwCHXJfAU4YABAKGgJ0bQ&ohost=www.google.com&cid=CAESbeD2J1z4qWevI6iWwrPeahjlRAd5PHQxq3vxCnxSB-XFcQcQlPCdbI5vS31FJsWDEdYDXr7wQRya94UUUP1TeQ_sv2Gp1efEdpeYMlECKaQ1jwRuNHUqGOxGxPQPMtaqjppqY8HY39-QdAnHBII&sig=AOD64_1r3tIroFWvKdgd3IEQB1BX0mmAdg&ctype=5&q=&ved=2ahUKEwizuv_B29D7AhUVBd4KHTcyApkQ9aACKAB6BAgEEAw&adurl=)

을 한번 읽어 보시기 바랍니다. 아주 기초 이론이고 실제 강화학습은 이 책 내용과는 다를 수 있습니다. 강화학습의 그리드월드 내용을 보시면 약간 지도학습의 모습과 닮았습니다. 모든 강화학습의 방법은 책 내용데로 해서는 결과를 보기는 상당히 힘듧니다. 기본적으로 agent는 일단 인공신경망을 배제하고 env와 agent를 구성할 필요가 있습니다. 그리고 나서 실제 agent가 발전하는게 확인이 되면 그다음은 인공신경망을 이용하여 Q(s,a)함수를 Deep learning 모듈화 하는 것입니다. 사실 이 모든 것은 어떤 수학자가 주장한 모든 문제는 곡선화 할 수 있으며 곡선화를 하면 해당 답을 구할 수 있다는 일반 함수론에서 출발하여 Q-learning을 DQN을 이용하여 함수로 만들 수 있다는 개념으로 출발했습니다. 그러니 Q(s,a)함수를 먼저 구하고 그 다음 그 함수를 DQN모듈로 변환하는 것이 바로 강화학습인것이죠. 물론 DB를 사용할 수도 있을 것입니다. 아주 정밀한 값도 저장할 수 있지만 만약에 데이터가 무한하다면 DB로 그 많은 데이터를 저장할 수 있을 까요? 그래서 강화학습에서는 DQN(CNN)을 이용하여 함수화 하자는 거죠 토치의 저장기능을 사용하면 몇 byte의 값으로도 그 함수를 저장할 수 있으니까요? 다시 시작할 때는 그냥 그 모듈을 다시 읽어서 재실행하면 끝입니다. DB보다는 컴퓨터 비용이 훨씬 저렴하죠 아니 저렴한 정도가 아니라 거의 공짜 수준이라고 할 수 있죠?
일단 Market class 소스를 보시죠

반응형
from matplotlib.backend_bases import FigureCanvasBase
import py
import plotly.graph_objects as go
import plotly.subplots as ms
import plotly.express as px
import plotly as plt
import pymysql
import pandas as pd
import numpy as np
import time
import talib
from PIL import Image

ticker = 'ETH'
class Market():
    def __init__(self) -> None:
        self.df = pd.DataFrame()
    
    def add_bbands(self):
        try:
            print("볼리져 밴드 구하기:%s"  % time.ctime(time.time()))
            self.df['ma20'] = talib.SMA(np.asarray(self.df['close']), 20)
            self.df['stddev'] = self.df['close'].rolling(window=20).std() # 20일 이동표준편차
            upper, middle, lower = talib.BBANDS(np.asarray(self.df['close']), timeperiod=40, nbdevup=2.3, nbdevdn=2.3, matype=0)
            self.df['lower']=lower
            self.df['middle']=middle
            self.df['upper']=upper
            # self.df['bbands_sub']=self.df.upper - self.df.lower
            # self.df['bbands_submax'] =  self.df.bbands_sub.rolling(window=840, min_periods=10).max()
            # self.df['bbands_submin'] =  self.df.bbands_sub.rolling(window=840, min_periods=10).min()
            # self.df['bbs_deg'] = (self.df.mavg - self.df.mavg.shift()).apply(lambda x: math.degrees(math.atan2(x, 40))) 

            # self.df['pct8']=(self.df.close - self.df.lower)/(self.df.upper - self.df.lower)
            # self.df['pct8vsclose']=self.df.close - self.df.pct8
        except Exception as ex:
            print("`add_bbands -> exception! %s `" % str(ex))

    def add_relative_strength(self):
        try:
            print("상대 강도 지수 구하기:%s"  % time.ctime(time.time()))
            rsi14 = talib.RSI(np.asarray(self.df['close']), 14)
            self.df['rsi14'] = rsi14
        except Exception as ex:
            print("`add_relative_strength -> exception! %s `" % str(ex))
            
    def add_iip(self):
        try:
            #역추세전략을 위한 IIP계산
            self.df['II'] = (2*self.df['close']-self.df['high']-self.df['low'])/(self.df['high']-self.df['low'])*self.df['volume']
            self.df['IIP21'] = self.df['II'].rolling(window=21).sum()/self.df['volume'].rolling(window=21).sum()*100            
        except Exception as ex:
            print("`add_iip -> exception! %s `" % str(ex))
            
    def add_stock_cast(self):
        try:
            # 스토캐스틱 구하기
            self.df['ndays_high'] = self.df['high'].rolling(window=14, min_periods=1).max()    # 14일 중 최고가
            self.df['ndays_low'] = self.df['low'].rolling(window=14, min_periods=1).min()      # 14일 중 최저가
            self.df['fast_k'] = (self.df['close'] - self.df['ndays_low']) / (self.df['ndays_high'] - self.df['ndays_low']) * 100  # Fast %K 구하기
            self.df['slow_d'] = self.df['fast_k'].rolling(window=3).mean()    # Slow %D 구하기        except Exception as ex:
        except Exception as ex:
            print("`add_stock_cast -> exception! %s `" % str(ex))
    
    def add_mfi(self):
        try:
            # MFI 구하기
            self.df['PB'] = (self.df['close'] - self.df['lower']) / (self.df['upper'] - self.df['lower'])
            self.df['TP'] = (self.df['high'] + self.df['low'] + self.df['close']) / 3
            self.df['PMF'] = 0
            self.df['NMF'] = 0
            for i in range(len(self.df.close)-1):
                if self.df.TP.values[i] < self.df.TP.values[i+1]:
                    self.df.PMF.values[i+1] = self.df.TP.values[i+1] * self.df.volume.values[i+1]
                    self.df.NMF.values[i+1] = 0
                else:
                    self.df.NMF.values[i+1] = self.df.TP.values[i+1] * self.df.volume.values[i+1]
                    self.df.PMF.values[i+1] = 0
            self.df['MFR'] = (self.df.PMF.rolling(window=10).sum() / self.df.NMF.rolling(window=10).sum())
            self.df['MFI10'] = 100 - 100 / (1 + self.df['MFR'])    
        except Exception as ex:
            print("`add_mfi -> exception! %s `" % str(ex))

    def add_macd(self, sort, long, sig):
        try:
            print("MACD 구하기:%s, sort:%d, long:%d, sig:%d"  % (time.ctime(time.time()), sort, long, sig))
            macd, macdsignal, macdhist = talib.MACD(np.asarray(self.df['close']), sort, long, sig) 
            self.df['macd'] = macd
            # self.df['macdmax'] = self.df.macd.rolling(window=840, min_periods=100).max()
            # self.df['macdmin'] = self.df.macd.rolling(window=840, min_periods=100).min()
            # self.df['macd_max_rate'] = self.df.apply(lambda x: (x['macdmax'] - x['macd']) * 100 / (x['macdmax'] - x['macdmin']), axis=1)
            # self.df['macd_min_rate'] = self.df.apply(lambda x: (x['macd'] - x['macdmin']) * 100 / (x['macdmax'] - x['macdmin']), axis=1)
            # self.df['prv_macd_degrees'] = self.df.macd_degrees.shift()
            self.df['signal'] = macdsignal
            self.df['flag'] = macdhist
            # self.df['prv_osc_degrees'] = self.df.osc_degrees.shift()
        except Exception as ex:
            print("`add_macd -> exception! %s `" % str(ex))

    def get_data(self):
        table_name="TB_ETH_TRADE"
        sqlcon = pymysql.connect(host='192.168.xx.xx', user='xxxx', password='xxxx', db='yubank2')
        cursor = sqlcon.cursor(pymysql.cursors.DictCursor)
        str_query = """select 
            'time'
            ,`start` as open
            , high
            , low
            , close
            , volume
            , macd
            , macdmax
            , macdmin
            , `signal`
            , `flag`
            , osc_degrees
            , sma1200
            , sma1200_degrees
            , wma1200
            , wma1200_degrees
            from %s""" % table_name
        cursor.execute(str_query)
        data = cursor.fetchall()
        self.df = pd.DataFrame(data)
        sqlcon.close()
        self.df['sma3'] = talib.SMA(np.asarray(self.df['close']), 3)
        self.df['closemax'] = self.df.close.rolling(window=100, center=True).max()
        self.df['closemin'] = self.df.close.rolling(window=100, center=True).min()
        self.add_macd(12,26,9)
        self.add_bbands()
        self.add_relative_strength()
        self.add_stock_cast()
        self.add_iip()
        self.add_mfi()
        self.df.dropna(inplace=True)
        self.df.reset_index(drop=True,inplace=True)
        return self.df

    def get_test_data(self):
        table_name="TB_ETH_TRADE"
        sqlcon = pymysql.connect(host='192.168.xx.xx', user='xxxx', password='xxxx', db='yubank2')
        cursor = sqlcon.cursor(pymysql.cursors.DictCursor)
        str_query = """select 
            `start` as open
            , high
            , low
            , close
            , volume
            , macd
            , macdmax
            , macdmin
            , `signal`
            , `flag`
            , osc_degrees
            , sma1200
            , sma1200_degrees
            , wma1200
            , wma1200_degrees
            from %s""" % table_name
        cursor.execute(str_query)
        data = cursor.fetchall()
        self.df = pd.DataFrame(data)
        sqlcon.close()
        self.add_bbands()
        self.add_relative_strength()
        self.df['clsmin'] = self.df.close.rolling(window=600, min_periods=600, center=False).min()
        self.df['clsmax'] = self.df.close.rolling(window=600, min_periods=600, center=False).max()
        self.df['action'] = self.df.apply(lambda x: 1 if x['close'] > x['up'] else( 1 if x['close'] < x['dn'] else 0), axis=1)
        # self.df.drop(self.df[self.df['action'] == 0].index, inplace=True)
        self.df.dropna(inplace=True)
        self.df.reset_index(drop=True,inplace=True)
        return self.df

    def get_lstm_data(self):
        table_name="TB_ETH_TRADE"
        sqlcon = pymysql.connect(host='192.168.xx.xx', user='xxxx', password='xxxx', db='yubank2')
        cursor = sqlcon.cursor(pymysql.cursors.DictCursor)
        str_query = """select 
            `date`
            ,`start` as open
            , high
            , low
            , close
            from %s""" % table_name
        cursor.execute(str_query)
        data = cursor.fetchall()
        self.df = pd.DataFrame(data)
        sqlcon.close()
        self.df['fclose'] = self.df.close.shift(-1)
        self.df.dropna(inplace=True)
        self.df.reset_index(drop=True,inplace=True)
        return self.df

    def get_chart(self, idx, max_data:int=300):
        try:
            df = self.df.head(idx + max_data).tail(max_data)
            df.reset_index(drop=True, inplace=True)
            if df.index.max() < 299:
                return None
            
            candle = go.Candlestick(x=df.index,open=df['open'],high=df['high'],low=df['low'],close=df['close'], increasing_line_color = 'red',decreasing_line_color = 'blue', showlegend=False)
            upper = go.Scatter(x=df.index, y=df['upper'], line=dict(color='red', width=2), name='upper', showlegend=False)
            ma20 = go.Scatter(x=df.index, y=df['ma20'], line=dict(color='black', width=2), name='ma20', showlegend=False)
            lower = go.Scatter(x=df.index, y=df['lower'], line=dict(color='blue', width=2), name='lower', showlegend=False)

            volume = go.Bar(x=df.index, y=df['volume'], marker_color='red', name='volume', showlegend=False)

            MACD = go.Scatter(x=df.index, y=df['macd'], line=dict(color='blue', width=2), name='MACD', legendgroup='group2', legendgrouptitle_text='MACD')
            MACD_Signal = go.Scatter(x=df.index, y=df['signal'], line=dict(dash='dashdot', color='green', width=2), name='MACD_Signal')
            MACD_Oscil = go.Bar(x=df.index, y=df['flag'], marker_color='purple', name='MACD_Oscil')

            fast_k = go.Scatter(x=df.index, y=df['fast_k'], line=dict(color='skyblue', width=2), name='fast_k', legendgroup='group3', legendgrouptitle_text='%K %D')
            slow_d = go.Scatter(x=df.index, y=df['slow_d'], line=dict(dash='dashdot', color='black', width=2), name='slow_d')

            PB = go.Scatter(x=df.index, y=df['PB']*100, line=dict(color='blue', width=2), name='PB', legendgroup='group4', legendgrouptitle_text='PB, MFI')
            MFI10 = go.Scatter(x=df.index, y=df['MFI10'], line=dict(dash='dashdot', color='green', width=2), name='MFI10')

            RSI = go.Scatter(x=df.index, y=df['rsi14'], line=dict(color='red', width=2), name='RSI', legendgroup='group5', legendgrouptitle_text='RSI')
            
            # 스타일
            fig = ms.make_subplots(rows=5, cols=2, specs=[[{'rowspan':4},{}],[None,{}],[None,{}],[None,{}],[{},{}]], shared_xaxes=True, horizontal_spacing=0.03, vertical_spacing=0.01)

            fig.add_trace(candle,row=1,col=1)
            fig.add_trace(upper,row=1,col=1)
            fig.add_trace(ma20,row=1,col=1)
            fig.add_trace(lower,row=1,col=1)

            fig.add_trace(volume,row=5,col=1)

            fig.add_trace(candle,row=1,col=2)
            fig.add_trace(upper,row=1,col=2)
            fig.add_trace(ma20,row=1,col=2)
            fig.add_trace(lower,row=1,col=2)

            fig.add_trace(MACD,row=2,col=2)
            fig.add_trace(MACD_Signal,row=2,col=2)
            fig.add_trace(MACD_Oscil,row=2,col=2)

            fig.add_trace(fast_k,row=3,col=2)
            fig.add_trace(slow_d,row=3,col=2)

            fig.add_trace(PB,row=4,col=2)
            fig.add_trace(MFI10,row=4,col=2)

            fig.add_trace(RSI,row=5,col=2)

            # 추세추종
            # trend_fol = 0
            # trend_refol = 0
            # for i in df.index:
            #     if df['PB'][i] > 0.8 and df['MFI10'][i] > 80:
            #         trend_fol = go.Scatter(x=[df.index[i]], y=[df['close'][i]], marker_color='orange', marker_size=20, marker_symbol='triangle-up', opacity=0.7, showlegend=False)
            #         fig.add_trace(trend_fol,row=1,col=1)
            #     elif df['PB'][i] < 0.2 and df['MFI10'][i] < 20:
            #         trend_fol = go.Scatter(x=[df.index[i]], y=[df['close'][i]], marker_color='darkblue', marker_size=20, marker_symbol='triangle-down', opacity=0.7, showlegend=False)
            #         fig.add_trace(trend_fol,row=1,col=1)

            # 역추세추종
            # for i in df.index:
            #     if df['PB'][i] < 0.05 and df['IIP21'][i] > 0:
            #         trend_refol = go.Scatter(x=[df.index[i]], y=[df['close'][i]], marker_color='purple', marker_size=20, marker_symbol='triangle-up', opacity=0.7, showlegend=False)  #보라
            #         fig.add_trace(trend_refol,row=1,col=1)
            #     elif df['PB'][i] > 0.95 and df['IIP21'][i] < 0:
            #         trend_refol = go.Scatter(x=[df.index[i]], y=[df['close'][i]], marker_color='skyblue', marker_size=20, marker_symbol='triangle-down', opacity=0.7, showlegend=False)  #하늘
            #         fig.add_trace(trend_refol,row=1,col=1)    

            # fig.add_trace(trend_fol,row=1,col=1)
            # 추세추총전략을 통해 캔들차트에 표시합니다.

            # fig.add_trace(trend_refol,row=1,col=1)
            # 역추세 전략을 통해 캔들차트에 표시합니다.
            
            # fig.update_layout(autosize=True, xaxis1_rangeslider_visible=False, xaxis2_rangeslider_visible=False, margin=dict(l=50,r=50,t=50,b=50), template='seaborn', title=f'({ticker})의 날짜: ETH [추세추종전략:오↑파↓] [역추세전략:보↑하↓]')
            # fig.update_xaxes(tickformat='%y년%m월%d일', zeroline=True, zerolinewidth=1, zerolinecolor='black', showgrid=True, gridwidth=2, gridcolor='lightgray', showline=True,linewidth=2, linecolor='black', mirror=True)
            # fig.update_yaxes(tickformat=',d', zeroline=True, zerolinewidth=1, zerolinecolor='black', showgrid=True, gridwidth=2, gridcolor='lightgray',showline=True,linewidth=2, linecolor='black', mirror=True)
            # fig.update_traces(xhoverformat='%y년%m월%d일')
            # size = len(img)
            # img = plt.io.to_image(fig, format='png')
            # canvas = FigureCanvasBase(fig)
            # img = Image.frombytes(mode='RGB', size=(700, 500), data=fig.to_image().to, decoder_name='raw')
            # img = Image.frombytes('RGBA', (700, 500), plt.io.to_image(fig, format='png'), 'raw')
            # img = Image.fromarray('RGBA', (700, 500), np.array(plt.io.to_image(fig, format='png')), 'raw')
            # img = Image.fromarray(np.array(plt.io.to_image(fig, format='png')), 'RGB')
            import io
            # img = Image.fromarray(np.array(plt.io.to_image(fig, format='png')), 'L')
            img = Image.open(io.BytesIO(plt.io.to_image(fig, format='png', width=140, height=100)))
            img.convert("RGB")
            img.thumbnail((100, 140), Image.ANTIALIAS)
            # img = Image.Image(fig.to_image(), 'RGB')
            # img.show()
            return img
        except Exception as ex:
            print("`get_chart -> exception! %s `" % str(ex))
            return None

차트 그리기는 https://sjblog1.tistory.com/m/45 의 내용을 참고했습니다. rolling함수를 사용할 경우 속도 문제가 있어서 ta-lib로 변경했습니다. ta-lib의 경우 거의 모든 주식차트 함수를 포함하고 있으며 c++로 개발이 되어 있어서 속도면에서 훨씬 빠릅니다.
클래스를 선언하고 get_chart함수를 호출하면 chart가 bmp 타입(png)의 이미지 데이터로 리턴이 됩니다. 데이터는 저의 경우 DB에 3분 단위로 매일 데이터를 저장하고 있고 그 기간이 벌써 1년반정도 되었씁니다. 꼭 DB를 사용하지 않아도 pyupbit로 데이터를 가져 오면 아마 1주일치 정도의 데이터를 가져 올 수 있습니다. 물론 더 많은 데이터가 있으면 좋겠지만 일주일 정도로도 훈련이 가능합니다. 다음 차에서는 상태값 S를 구하는 것에 대하여 설명 드리겠습니다.

 

 

반응형
LIST

맨위로버튼이미지

들어가기전에
매몰비용이라는 말이 있습니다. 경제학에서는 '콩코드 효과'라는 말이 있습니다. 미국의 보잉사는 1970년대에도 항공기 제조사중 최고였다고 합니다. 이에 영국과 프랑스는 오로지 미국 보잉을 이기기 위한 싸움을 시작합니다. 그리고 태어난 비행기가 콩코드기 였습니다. 마하2의 속도를 최초로 돌파한 민간 항공기 였을 겁니다. 하지만 앞에서 보는 것보다 뒤로 깨지는 돈이 상당한 그런 사업이었나 봅니다. 사업이 계속 될 수록 투자자들에 얼굴에는 검은 먹구름이 끼고 있었죠. 그리고 콩코드 비행기에 사고가 발생합니다. 콩코드는 보잉의 탑승료보다 훨씬 비싼 비행기여서 일반은 잘 안탑니다. 사고로 유명인사들이 많이 죽게 됩니다. 이로 인해 콩코드 프로젝트는 끝이 납니다. 그 사고가 없었다면 더많은 사람이 죽었을지도 모릅니다.

일단 만약 이 글을 읽는 분이 이 프로젝트를 시작하시고자 한다면 위의 콩코드 효과를 걱정하셔야 할겁니다. 그러나 시작했다면 뒤는 돌아 보시면 안됩니다.
오늘은 개발 환경을 셋팅하도록 하겠습니다. 일단 그래픽카드가 하나 있어야 합니다. 더이상 CUDA는 RTX 10xx대 보더 미만을 지원하지 않습니다. 최소 RTX 1030보더 이상은 구매를 해야한다는 말입니다. 다행히 저가 컴구매자를 위한 저가 모델이 아직 생산되고 있습니다. 최소한 보드의 메모리는 6GB 이상이 필요합니다. 그렇지 않으면 DQN(CNN)모듈이 구동하는 순간 RUNTIME 오류가 발생하고 멈추게 될겁니다. 일단 저희 Lib는 차트를 그릴겁니다. 메모리상으로만 그리고 실제 화면으로 출력 되지는 않을 겁니다. 이 화면의 사이즈는 500 x 700 입니다만 이 사이즈의 메모리도 Pytorch는 처리가 불가합니다. 정확하게는 cuda lib입니다만 바로 memory allocation runtime error가 발생합니다. 내용을 자세히 읽어보시면 gpu 카드의 실메모리량이 부족하여 데이타를 처리할 수 없다는 내용입니다. 그 이유로 차트를 다시 100 x 140으로 축소를 합니다. 나중에는 사이즈를 키울거지만 지금은 이게 최선인것 같습니다. 먼저 ndividia에 회원 가입이 필요합니다. 구글아이디로 가입하는 것이 가장 좋을것 같습니다. ndividia 홈페이지에서 download center로 이동하여 cuda 최신 버전을 다운 받습니다. 13이나 14던 상관은 없습니다. cuda의 설치가 완료되었다면 python 버젼을 선택해야 합니다. 혹시 openAi사의 retro gym을 경험하고자 한다면 python 3.7버젼을 추천합니다. 일단 retro gym을 경험해 보시면 프로그램을 어떻게 해야하는지 감이 옵니다. 어떤 책에서는 강화학습을 env(환경)과 agent(게임을 실행하는 프로그램)으로 나누는 데 retro gym을 보면 python프로그램 자체가 agent가 되고 env는 openAi에서 제공해 준다고 보시면 됩니다. agent를 만들어 게임을 깨는 내 프로그램을 보면 대단하다고 혼자 기쁘하실지 모르지만 진짜는 현실의 상황을 env로 만드는게 진짜구나 하는 생각을 하게 됩니다. 즉 retro gym을 경험하시면 진짜는 DQN이 아니라 env를 어떻게 설계하고 만들지 입니다. 유튜브에 보시면 카이스트 출신의 혁펜다임 채널을 보시면 이론은 쉽게 설명하는 영상을 볼 수 있습니다. 보고 나시면 현실을 강화학습으로 바꾸는 과정의 진짜 중요한 부분은 env라는것을 금방 눈치 챌 수 있습니다. 그러나 다행히 의외로 주식 프로그램은 env 제작이 그렇게 어렵지 않습니다. 앞으로 돌아가서 python 3.9 나 3.10을 사용하면 안되는 이유는 retro gym은 3.9이상에는 실행조차도 안됩니다. 책은 Do it시리즈 중 '강화학습 입문'을 추천 드립니다.( https://ebook-product.kyobobook.co.kr/dig/epd/ebook/4801163032527) 3장의 Buruto force 방법을 보시면 됩니다. 일단은 agent의 개념을 이해하는게 중요합니다. cuda가 다 설치 되었다면 pip와 numpy를 업그레이드합니다

python -m pip install -U pip
python -m pip install -U numpy

torch, torchvision만 있어도 충분하지만 필요하다면 torchaudio를 설치합니다. (https://pytorch.org/get-started/previous-versions/) 이때 torch repository site에 가시면 더 많은 whl(pip install file)이 존재하는것을 보실 수 있습니다. pip에 권한 문제가 생기거나 위치 문제 또는 lib 위치 문제가 발생한다면 python -m pip로 설치 해보시기 바랍니다. 그럼 보통은 대부분 해결이 됩니다. 그리고 pip를 바로 실행할 경우 경로가 꼬여 다시 설치해야 하는 경우가 생깁니다. 그런 경우는 python -m pip install - U {모듈명}를 사용하시면 됩니다. 그리고 whl파일을 설치하는 경우 repository url을 알고 있다면 pip install {url}이렇게 해도 됩니다. torch의 경우 내가 설치한 cuda에 맞는 버젼을 찾아서 설치해야해서 꼭 repository url을 참조해서 설치 하시기 바랍니다. 다음은 ta-lib를 설치해야 합니다 windows 의 경우 구글을 검색하여 whl파일을 python 버젼과 일치하는 파일을 찾아서 설치하셔야 합니다. linux 버젼을 깔 경우는 레드햇 계열의 경우 yum(ubuntu는 apt 또는 apt-get)으로 python버젼과 동일한 python-devel 을 설치해야 하고 gmake와 gcc, g++을 설치 해주셔야 합니다. 설치할 python의 whl파일을 검색하여 버젼과 맞는 repository url을 복사하여 pip문 끝에 붙여 줍니다. 기타 pandas나 다른 util은 쉽게 설치가 되므로 python을 실행해보고 에러유형을 보고 추가하면 됩니다. 다음 장에서는 데이터 수집과 마켓 클래스에 대하여 소개해 드리겠습니다. env의 첫 관문입니다. 강화학습이 궁금하신 분은 retro gym의 예제 하나를 실행해 보고 오시죠. OpenAi는 일론 머스크 돈으로 우리같은 사람을 위해서 만든 오픈소스 플랫폼이니 많이 이용하세요.

 

==> WSL 환경으로 변경하였으니 8장 WSL환경 이용바랍니다.

반응형
LIST

맨위로버튼이미지

강화학습 매매 프로그램을 만들면서 프로그래머로써 가장 큰 도전을 견뎌야 했습니다. 학습량이랑 소스, 디버깅 량 시스템 투자 모두 역대 최대였습니다. 중간쯤 진행하다보니 포기할 수가 없었습니다. 이 글은 다른 도전자에게 대한 삽질의 최소화를 도모하고자 작성하게 되었습니다. 부디 이 글을 읽으시는 분들은 좀 더 들 고생하였으면 합니다. 개발 환경은 win10(wsl 개발환경으로 변경)에 메모리 48G CPU는 제온 2650 V3에 RTX 3060 OG 12G 2장과 화면 디스플레이용 RTX 1060한장이 들어간 시스템을 사용했습니다. 꼭 이렇게 할 필요는 없지만 DQN은 CNN을 베이스로 하므로 GPU 메모리를 굉장히 많이 사용합니다.

딥런닝 엔진으로 PYTORCH를 사용했지만 tensoflow로 변경은 그리 어렵지 않을 것으로 생각됩니다. 먼저 머쉰런닝을 공부한 경험이 있는 사람은 지도 학습과 비지도 학습에 대한 내용은 어느 정도 알고 있을 것으로 생각합니다. 지도 학습은 기존에 정재된 데이터가 이미 존재하고 데이터에 대한 결과가 분명할 경우 학습을 통해 새로운 데이타를 분류하거나 값을 예측하는 것을 말합니다. 예를 들면 한 고등학교 학생을 대상으로 방과후 학습시간을 설문 조사한 후 3번의 시험을 치고 나서 4번째 시험 결과를 예측하는 경우가 지도학습과정이라 할 수 있습니다. 비지도 학습은 데이터는 존재하지만 분류기준등이 모호할 경우 머쉰런닝을 통해서 분류값을 얻고 데이터를 분류하는 경우입니다. 예를들면 어느 들판에 있는 들꽃을 조사했는데 같은 종류지만 키가 차이가 나는것을 발견하고 머쉰런닝을 돌려서 같아보이지만 다른 종류의 꽃이 몇가지가 존재하는지 예측하는 프로그램을 만들었다면 그것은 비지도 학습입니다. 그렇다면 다음은 어느경우에 해당할까요? 한국어를 학습하는 머쉰런닝 프로그램이 있다면 프로그램은 무작위로 학습해서 주어와 동사가 있다는 것을 학습한 후 언어에는 순서가 있다는것을 배우고 그후에는 사람과 실시간 채팅을 학습합니다. 사견이지만 지도학습과 비지도학습은 머쉰런닝을 설명하기 위한 기준이지 강화학습으로 들어가면 분류 자체가 무의미해지는 것 같습니다. 강화학습을 가장 쉽게 설명하는 것은 영화일것 입니다. 그것도 타임 루프물 영화들이 그렇습니다. 톰 크루즈가 출연한 작품 중 '엣지 오브 투모로우'라는 영화가 있습니다. 이 영화는 외계 생맹체로 부터 지구가 침공 당해 거의 전멸하기 직전의 상황에서 톰형이 무슨일인지 장교에서 신분강등이 되어 전투지휘부로 끌러 옵니다. 이후 전쟁에 참여해서 외계인의 분대장격인 알파를 죽이고 그 피를 받으면서 죽으면 하루전으로 가는 능력을 가지게 되는게 이야기의 시작입니다.

처음 도입부는 약간 지겹게 느껴질 수 있습니다. 그는 계속 억울함을 주장하고 전투에 출전하고 죽고 다시 깨어나고의 반복입니다. 그 사이 톰은 무기의 사용법과 전투방법과 미래의 일어날 일을 외우고 자기 부대원에게 전투방법을 가르치며 조금씩 발전을 합니다. 그는 전쟁에 승리하기 위한 강화학습을 진행하고 있는것입니다. 그리고 그로인해 전쟁의 양상은 바뀌게 됩니다.

우리가 하고자 하는 봐도 바로 이것입니다. 우리는 쉽게 주식이나 코인의 거래데이타를 받을 수 있습니다. 그 데이터를 이용하여 수천번의 시뮬레이션을 실시해서 최고의 성적을 내는 시나리오를 얻고 이것을 CNN에 하이퍼 파라메트로 학습하여 이것을 이용하여 실전의 행동과 예상 reward 를 얻는 방식으로 진화할 수 있습니다.

CNN은 머쉰런닝 분야에서 가장 발전한 형태일 것입니다. 처음 시작은 팩스로 전송된 손글씨를 인식하여 데이터화 하는것이었습니다. 그 다음은 개와 고양이 사진을 보고 개와 고양이를 구분하는 논문이 발표됩니다. 그 후 Open AI라는 회사는 DQN이라는 인공신경망을 개발하고 아케이드 게임을 인간보다 잘하는 인공지능을 개발합니다. 이후 레트로 짐 이라는 인공지능 학습환경을 개발하여 일반에게 오픈했습니다.

우리는 그 방법들 중 한가지를 사용하여 DQN을 학습하여 그 DQN이 얼마나 사람보다 주식이나 코인 트레이딩을 더 잘하는지 아님 더 잘할때까지 계속 학습하는 방법을 의논할 것입니다. 저의 글을 보시고 다른 의견이 있으시면 답글 부탁드립니다.

 

 

반응형
LIST

맨위로버튼이미지

1. 왜 scikit-learn인가?

scikit-learn을 제가 처음 접한 때는 아직 정식버전을 내놓지 않은 0.x 버전이었습니다. 또한 서점에 나와 있는 책이나 블로그의 내용들도 그러한 이유로 0.x버전의 책이었습니다. 현시점에 scikit-learn은 1.x 버전을 내놓은 상태입니다. 먼저 scikit-learn은 tensflow와 자신의 영역을 분명히 하고 있으며 가는 방향도 다르다고 선언을 했습니다. tensflow가 고성능의 고용량의 데이터에 대한 데이터 분석 및 학습을 목표로 한다면 tensflow를 사용할 수 없는 또는 시스템이 지원하지 않는 영역에서의 분석 및 학습이 scikit-learn의 영역임을 분명히 했습니다. 제가 개발하려는 환경 또한 위의 환경입니다. 저의 개발 및 운영 시스템으로 오라클 클라우드에서 제공받고 있는 평생 무료 서버를 사용하고 있습니다. 그 상황에서 데이터를 얻어서 TA-Lib로 매매데이터를 분석하고 매수 목표나 매도 목표를 찾는 시스템을 운영하고 있습니다. 따라서 당연히 GPU는 사용할 수 없으며 tensflow나 keras의 고급 머쉰 러닝은 사용할 수 없습니다. 하지만 이러한 제약으로 인하여 scikit-learn을 사용하려는 것은 아닙니다. scikit-learn의 구조는 매우 단순한 구조를 가집니다. 데이터의 획득, 학습 데이터와 테스트 데이터의 분류, 학습(fit), 학습 데이터 평가(score), 예측(predict)의 구조가 전부이며 각각의 구조가 scikit-learn의 함수들 각 하나로 이루어져 있습니다. 제가 초기에 scikit-learn을 배우고 적용하는 것이 힘들었던 이유가 바로 위의 내용이기도 합니다. 머쉰 러닝이 간단해도 이렇게 간단할 수 있을까? 하는 의심이 학습을 방해한 이유 중 하나였으며 너무 간단하다 보니 사실 책을 쓰시는 분이나 블로그들의 글도 많지 않을뿐더러 서점의 책도 별로 없는 상태였습니다. 그리고 머쉰 러닝 하면 그 당시 대부분이 tensflow에 대하여만 생각하는 분위기라 scikit-learn을 아는 사람도 별로 없었죠.

2. 목표를 분명하게

제가 하려는 건 분명합니다. 기존에 수집 및 TA분석중인 데이터를 scikit-learn로 재분석하여 가장 데이터에 맞는 scikit-learn의 분석모델을 찾는 것이며 데이터를 최적화하는 것입니다. 위의 왜 scikit-learn인가에서도 아셨겠지만 scikit-learn을 배우는 것은 어렵지 않습니다. 공식 사이트의 연습 문제만 풀어도 내용을 알 수 있지만 제가 중요하고 어렵다고 본 것은 scikit-learn이 아닙니다. 기존의 데이터를 scikit-learn의 데이터로 변환한 다음 모델의 정확도를 지속적으로 높일 수 있으면서 성능이 뛰어나며 실제 매매 엔진을 방해하지 않는 것입니다. 보통 매매 엔진은 1분 데이터를 수집하는 경우 데이터 수집과 매수, 보유, 매도를 결정하고 실행하는 데까지 모두 1분 안에 끝내야 합니다. 이 1분이 초과된다면 그 다음은 3분 데이터를 사용한 매매 엔진을 다시 개발해야겠죠. 각각의 분석의 정확성에 따라 기존의 매매전략을 다시 만들어야 할 수도 있으며 완전히 새로운 방식을 고민해야 할 수도 있을 것입니다.

3. scikit-learn의 학습이란?

scikit-learn의 학습에 대하여 생각하기전에 먼저 일반적인 딮런닝을 생각해 봐야겠네요.
y = a1 * x1 + a2 * x2 + a3 * x3 + ... + an * xn + b
위의 식의 y는 우리가 딮런닝 예측으로 구하려고 하는 값입니다. 딮런닝의 예측에는 분류(Classify)와 회귀(Regress)가 있습니다. 우리가 해결해야 하는 문제가 분류인지 회귀인지를 스스로 판단하고 알아야 합니다. 분류는 붓꼿의 품종을 예측하는 것처럼 집합의 범위가 적으며 정해져 있는 경우일 겁니다. 각 투수별 구종을 학습하여 야구 선수가 다음번에 어떤 공을 떤질지를 딮런닝으로 학습한다면 이런 경우는 범위는 조금 많지만 분류에 해당한다고 할 수 있습니다. 회귀는 무엇일까요? 주가를 예측하는 딮런닝 모델이 있다고 한다면 주가와 주변의 환경변수를 이용하여 주가를 예측할 수 있을 것입니다. 주가를 예측하기 전에 하나의 곡선(또는 다차원 평면)을 그릴 것입니다. 그리고 그 곡선에 학습에 사용한 환경변수를 대입하여 하나의 값을 구할 수 있을 것입니다. 위의 곡선을 구하는 것을 회귀라 하며 주가의 예측은 회귀에 의해서 그려진 곡선에 각각의 좌표를 이루는 값을 대입한 결과가 예측한 주가가 될 것입니다. 그런데 이 곡선은 우리가 수학 시간에 배웠던 2차원 직선이나 3차원의 곡선이 아닙니다. 환경변수의 수가 n이라고 하면 n차식의 곡선인 거죠 그리고 이 n차식에 들어가는 변수가 우리가 학습에 이용한 환경변수( ※ DB 테이블로 치면 컬럼에 해당할 것입니다.)입니다. 위의 식이 바로 이러한 딮런닝을 표현한 식입니다. 아마도 이 글을 읽는 순간 갑자기 어려워하시는 분들이 있을 수 있으나 우리가 해야 하는 일은 우리가 구하려는 값이 분류에 있는지 회귀에 있는지 잠깐 생각하는 게 전부입니다. 주가나 필라델피아 주택 가격이나 금의 시세나 또는 달러의 시세를 예측하는 프로그램이 있다면 이 것은 회귀일 가능성이 높습니다. 그렇다면 이런 문제는 어떨까요? 근 미래의 주가가 오를지 내릴지를 맞추는 딮런닝 프로그램이 있다면 이것은 분류일까요? 회귀일까요? 당신은 이 문제를 어떻게 접근하실 건가요?

4. 데이터 수집

한국투자신탁의 뱅키스API나 upbit API를 이용하여 1분 데이터를 수집하는 프로그램을 작성할 수 있을 것입니다. 사실 꼭 1분일 필요는 없습니다. 30분 데이터로도 훌륭한 데이터 수집 및 분석 프로그램을 만들 수 있습니다. 사실 시간은 의미 없는 눈속임일 수 있습니다. 저는 이렇게 가져온 데이터를 TA-lib(https://mrjbq7.github.io/ta-lib/install.html)를 사용하여 분석후 이용하고 있습니다. 하지만 한 가지 문제를 가지고 있습니다. TA(기술적 분석)은 사실문제를 가지고 있습니다. 그것은 바로 분석된 데이터의 후행성입니다. 예를 들면 60분 ma(이동평균선)을 보면 7~10분 정도 먼저 주가가 올라가고 나서 이동평균선이 상향으로 움직인다는 것을 알 수 있습니다. 반대로 주가가 빠질 때는 주가가 먼저 빠진 후 이동평균선이 하향으로 움직이는 것을 볼 수 있습니다. 이것은 TA 데이터가 거의 대부분이 과거 데이터의 평균값을 이용하고 있으며 미래의 예측이 아니기 때문에 발생하는 문제입니다. 주가 차트의 가장 저점은 사실 그 지점에서 반등한다면 최고의 이익을 선사할지도 모르지만 그 지점마저 무너진다면 우리에게 지옥의 맛을 보여줄 자리 일 수도 있습니다. 더는 내려갈 곳이 없다고 생각하는 것은 인간의 어리석음일 수 있습니다. 가치 투자자들은 자신의 판단에 맡겨 이 정도 값이면 더 내려도 괜찮아라고 생각하겠지만 매매 프로그램 자체가 이미 가치 투자가 아니므로 이런 매매 프로그램에게 그러한 자리는 아주 치명적일 수 있습니다. 모든 상품 및 현물은 성장기 확장기 감쇠기 쇠퇴 기를 반복하지만 거짓된 상품은 확장기에서도 갑자기 가치가 0에 수렴할 수도 있습니다. 그런 경우까지 매매 프로그램이 감지하는 경지까지 아직 딮런닝 프로그램들이 발전하지 않았습니다. 딮런닝 중의 한 부야에는 제한적 학습이라는 분야가 있습니다. 제한적 학습은 많은 경험과 함께 그런 제한적 학습을 어떻게 적용할 것인지를 알아야 될 뿐 아니라 scikit-learn의 학습으로는 불가능할 수 있습니다. scikit-learn은 0.x 버전 때보다는 지금은 엄청나게 많이 발전했지만 scikit-learn은 한 가지 단점이 있습니다. 다른 딮런닝 모듈보다는 확실히 사용하기 쉽지만 다른 일부 엔진처럼 각각의 모듈을 customize 하기에는 제한적이라고 할 수 있습니다. 그런 customize가 가능하려면 여러분 들은 모든 분류엔진과 회귀 엔진을 직접 구현할 수 있을 정도의 실력이 필요하지 않을까요?

async def read_all(sqlcon) -> pd.DataFrame:
    try:
        cursor = await sqlcon.cursor(aiomysql.cursors.DictCursor)
        str_query = """select 
            `time`, start, close, high, low, volume, sma1200, sma1200_degrees, wma1200, wma1200_degrees
            from %s""" % table_name
        await cursor.execute(str_query)
        data = await cursor.fetchall()
        datadf = pd.DataFrame(data)
        return datadf
    except Exception as ex:
        exc_type, exc_obj, exc_tb = sys.exc_info()
        fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
        logger.error("read from db exception! %s ` : %s %d" % (str(ex) , fname, exc_tb.tb_lineno))
    finally:
        await cursor.close()

async def trade_proc():
	pool = await aiomysql.create_pool(host="호스트ip 또는 이름", user="root", password='pass', db='dbname')
	async with pool.acquire() as sqlcon:
		df = await read_all(sqlcon)
		sqlcon.close()
        df['futher'] = df.close.shift(-5)
        df['futher_mid'] = df.close.shift(-5) - df.close
        df['predict'] = (df.close.shift(-5) - df.close).apply(lambda x: (-1 if x < 0 else (0 if x == 0 else 1))) 
        df = df.dropna()

        await regressor(df)  #실제 학습 함수
async def regressor(df):
    X_data = df.drop(['futher', 'futher_mid', 'predict'], axis=1)
    X_new = df.drop(['futher', 'futher_mid', 'predict'], axis=1).tail(5).head(1)
    X_clct = df.tail(5).head(1)
    y_data = df['futher']

    X_train, X_test, y_train, y_test = train_test_split(X_data, y_data, random_state=0)
    mlp = MLPRegressor(solver='lbfgs', max_iter=500, random_state=0)
    mlp.fit(X_train, y_train)  # 학습
    
    logger.info("MLP 훈련 세트의 정확도: {:.2f}".format(mlp.score(X_train, y_train)))
    logger.info("MLP 테스트 세트의 정확도: {:.2f}".format(mlp.score(X_test, y_test)))
    joblib.dump(mlp, 'mlp.jdmp')

if __name__ == '__main__':
    asyncio.run(trade_proc())

기존 DB의 데이터로만은 scikit-learn의 학습이 불가능할 수 있습니다. 학습에는 x1...xn에 들어가는 기본 필드도 필요하지만 답에 해당하는 y_data도 필요합니다. 이런 데이터는 pandas를 이용하면 쉽게 구할 수 있습니다. pandas는 기존의 관계형 DB의 속성과 함께 nosql의 속성 및 큐빅 DB에서 볼 수 있는 DB의 세로(칼럼) 간 연산도 가능합니다. pandas의 이러한 기능은 scikit-learn의 학습 데이터를 아주 쉽게 만들 수 있습니다. 위의 예제 소스는 기존의 DB를 이용하여 scikit-learn 학습 데이트를 만드는 예제입니다. regressor함수는 회귀 학습을 위한 예제입니다. 현재의 데이터로 미래의 값을 예측하기 위해서 과거의 데이터의 과거 시점 5분 후의 데이터를 이용하여 further와 further의 평균, 그리고 예측값을 만들어 y_data를 만든 다음 X_data에서는 해당 값을 제거하여 학습하는 것을 볼 수 있습니다. 여기서 주의할 점은 scikit-learn의 학습능력은 생각보다 정확하여 예측에 사용된 값을 그대로 넣고 학습을 할 경우 학습률 100%의 데이터를 생성하게 됩니다. 물론 실전능력은 떨어지게 됩니다. 학습 시 해당 예측값까지 이용하여 예측값 공식을 만들기 때문에 해당 필드의 가중치가 최고가 되어 결국은 학습이 의미가 없어지게 됩니다.

5. 지속 가능한 학습이란?

지속 가능한 학습을 위해서는 자료 수집과 학습 및 실전예측 프로그램의 설계가 중요합니다. 자료 수집 및 학습이 배치(BATCH) 작업화 되어야 하고 학습된 모델은 파일로 저장되어 실측 프로그램에서 사용되어야 합니다. 자료 수집 및 학습은 매우 큰 시간이 필요하므로 자료 수집과 학습을 실측 프로그램과 분리하여 운영해야 합니다. 아래 섹션 6에서는 모델을 저장하여 이용하는 방법을 소개하고 있습니다. scikit-learn은 모델을 학습하고 학습된 모델을 파일로 저장하고 저장된 파일 자체를 로딩하여 파이썬의 함수로 사용할 수 있는 모듈을 제공하고 있습니다.

6. 모델의 저장 및 학습모델로 예측하기

scikit-learn은 학습한 모델을 python liberary로 분석하여 파일로 저장하는 기능과 그 파일을 불러와서 예측을 하는 함수를 제공합니다. 많은 양의 데이터를 학습하면 모델의 정확도는 당연히 증가할 겁니다. 하지만 학습에 소요되는 시간은 엄청나게 길어집니다. 만약 매번 학습을 돌리고 그 학습 데이터를 이용하여 예측을 진행한다면 제가 2번 섹션에서 이야기한 1분의 시간은 넘어가 버릴 겁니다. 그렇다면 scikit-learn의 학습은 아무런 소용이 없습니다. 하지만 scikit-learn은 joblib.dump라는 함수를 제공합니다.

from sklearn.model_selection import train_test_split
from sklearn.neighbors import KNeighborsClassifier
import joblib

X_train, X_test, y_train, y_test = train_test_split(X_data, y_data, random_state=0)
knn = KNeighborsClassifier(n_neighbors=3)
knn.fit(X_train, y_train)   #학습
logger.info("KNN 훈련 세트의 정확도: {:.2f}".format(knn.score(X_train, y_train)))
logger.info("KNN 테스트 세트의 정확도: {:.2f}".format(knn.score(X_test, y_test)))
    
prediction = knn.predict(X_new)
logger.info("KNN 예측 결과:%d", prediction)
joblib.dump(knn, 'knn.jdmp') #모델 저장

위의 소스코드가 사실은 scikit-learn의 전부 입니다. tensflow의 keras 보다는 월등히 쉬운 코딩이 가능합니다. 6라인을 보시면 아시겠지만 scikit-learn의 모든 모델은 클래스 화해서 모듈화 되어 있습니다. 위에 이야기한 것처럼 scikit-learn이 문제가 아니라는 것을 여러분은 느끼실 겁니다. scikit-learn의 소스는 5라인부터 끝까지가 전부입니다. 모든 클래스는 분류이면 Classifier가 붙을 것이고 회귀이면 Regressor이 붙을 것입니다.

mlp = MLPRegressor(solver='lbfgs', max_iter=500, random_state=0)
mlp.fit(X_train, y_train)  # 학습
    
logger.info("MLP 훈련 세트의 정확도: {:.2f}".format(mlp.score(X_train, y_train)))
logger.info("MLP 테스트 세트의 정확도: {:.2f}".format(mlp.score(X_test, y_test)))

joblib.dump(mlp, 'mlp.jdmp') #모델 저장

위의 소스는 다층신경망 회귀를 이용하여 학습하고 그 학습된 모델을 모듈화 하여 mlp.jdmp파일로 저장하는 소스입니다.
여기서 중요한 부분은 그냥 단순히 학습된 모델을 파일로 저장하는 것이 아니라는 것이다. 저장된 파일을 로딩하면 바로 파이썬 함수로 사용이 가능합니다. 많은 데이터를 학습하려면 몇분에서 많게는 몇시간의 시간이 필요할 수도 있지만 학습이 끝나면 실제 데이터로 예측을 수행하는데는 그리 길지 않은 시간이면 됩니다.

mlp_from_model = joblib.load('mlp.jdmp') #파일로 저장된 모델을 파이썬 함수로 로딩
prct = mlp_from_model.predict(X_new) #로딩된 모델을 실제 값으로 예측
print("예측값:" + prct)   #예측 값은 prct변수에 들어 간다. prct값은 int나 double이 아닌 Dict type입니다

실제 운영되는 실행파일에 위의 내용이 있으면 될 것입니다. 학습에 오랜시간이 걸리겠지만 predict함수가 실행되는 데는 단 몇 초면 끝입니다. 위의 예제는 실제 사용 가능한 다층 신경망 회귀와 근접 이웃 분류의 소스 코드입니다.

7. 글을 마치며

프로그래머가 위의 소스를 보고도 딮러닝 프로그램을 작성하지 못하는 이유는 뭘까요? 사실 위의 소스는 scikit-learn학습을 구현한 소스 전체나 마찬가지 입니다. 차라리 수학, 통계학, 데이터 과학을 전공한 분들은 위의 소스를 보면 바로 scikit-learn 프로그램을 뚝딱 개발하실 수 있을 것입니다. 하지만 이게 다가 아닙니다. 위의 소스가 살아서 움직이려면 기초 데이터의 수집, Database의 연동, 그리고 기초 데이터를 다시 분류하고 스프레드 하는 작업이 필요할 것입니다. scikit-learn프로그램을 운영하려면 적어도 2개 이상의 프로그램이 필요함을 단번에 알 수 있습니다. 하나는 모델의 학습을 하나는 모델을 불러와서 실제 데이터에 적용하는 프로그램이 될 것입니다.

반응형
LIST

'python > 머쉰러닝' 카테고리의 다른 글

구글 colab에서 학습하기  (0) 2023.01.20

맨위로버튼이미지

1. 들어가기 전에

  오늘은 자동매매 프로그램 구현에서 중요하지만 주제와는 약간 빗겨간 DB 내용중 Upsert(Update or Insert)에 대하여 이야기 하고자 합니다. pandas의 DataFrame은 Nosql적 성격을 가진 아주 휼륭한 Lib입니다. 사실 pandas가 없다면 지금의 자동매매 프로그램의 구현은 힘들다고 해도 이견이 없을 것입니다. 하지만 pandas DataFrame을 사용하다보면 계산이 다 끝나서 이제는 실제 DB로 옮겨야 하는데 기존 제공하는 to_sql은 뭔가 약간은 부족합니다. to_sql함수는 중복이 있으면 insert 실패 처리 하거나 table 전체를 replace하거나 또는 그냥 append하는 것 입니다. 저는 지금은 pandas와 mysql을 사용하고 있지만 초창기는 sqlite3을 사용하였고 mysql로 DB를 변경하면서 문제가 발생했습니다. to_sql의 replace기능이 DB를 삭제하는 과정에 프로그램이 먹통이 되고 이때까지 모아온 소중한 데이터가 다 날아 가는 거죠. sqlite사용중에도 프로그램을 중간에 내릴때 타이밍이 안좋으면 table을 그대로 날려 먹는 경우가 생깁니다. 그래서 현재 인터넷에서는 pandas의 to_sql에 새로운 옵션인 upsert가 화재 입니다. 일부 개발 git에는 upsert가 구현되어 올라와 있지만 pandas의 to_sql을 ctrl + link로 따라 가보면 근본적인 sql lib에 있는 기능을 사용하고 있어 수정하는 것이 대공사라는 것을 금방 알 수 있습니다. 오늘은 제가 구현한 Upsert에 대하여 내용을 공유하고자 합니다. 

 

2. Upsert구현

  먼저 아래 소스를 보시죠.

def trade_data_update(df, dbcon, tb_name, if_exists: str='ignore'):
    try : 
        cols = [column for column in list(df.columns)]

        str_query = ""
        if if_exists == 'fail':
            str_query += "INSERT INTO %s (" % tb_name
        elif if_exists == 'replace':
            str_query += "REPLACE INTO %s (" % tb_name
        else:
            str_query += "INSERT IGNORE INTO %s (" % tb_name

        for col in cols:
            str_query += "`" + col + "`,"
        str_query = str_query[:-1]
        str_query += ")"

        rows_val = []
        for index in df.index:
            rowval = []
            for column in df.columns:
                rowval.append(df.loc[index, column])
            rows_val.append(rowval)
        
        str_query += " values "
        for rows in rows_val:
            str_query += "("
            for row in rows:
                if str(type(row)) == "<class 'str'>":
                    str_query += "'" + str(row) + "',"
                elif str(row) == "nan":
                    str_query += "NULL,"
                else:
                    str_query += str(row) + ","
            str_query = str_query[:-1]
            str_query += "),"
        str_query = str_query[:-1]

        cursor = dbcon.cursor()
        cnt = cursor.execute(str_query)
        # logger.info("apply row cnt:%d", cnt)
        cursor.close()

    except Exception as ex:
        exc_type, exc_obj, exc_tb = sys.exc_info()
        fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
        logger.error("trade_data_update -> exception! %s  : %s %d" % (str(ex) , fname, exc_tb.tb_lineno))

mysql기준 쿼리 입니다. mysql 에 INSERT에는 총 4가지의 방법이 있습니다. INSERT ON DUPLICATE KEY와 REPLACE INTO, INSERT IGNORE INTO, 그리고 그냥 INSERT입니다. 당연히 그냥 INSERT를 사용하면 INSERT과정에 Dup key가 발견되면 그 부분에서 바로 에러 후 다음으로 넘어 가겠죠. 나머지 3가지는 이렇게 Dup key가 발견되었을때의 예외 처리 방법을 기술한 것 입니다. 저는 보통 회사에서 프로그램할때는 INSERT INTO 보다는 REPLACE INTO를 많이 사용하는 편인데 지금 자동 매매에서는 INSERT IGNORE INTO를 많이 사용하고 있습니다. 

import pymysql
from lib.Upsert import trade_data_update

sqlcon = pymysql.connect(host='127.0.0.1', user='root', password='1234', db='main')

df = con.get_dataframe()
trade_data_update(df, sqlcon, "tb_training_data")
sqlcon.commit()

처럼 활용이 가능 합니다. 

 

3. 마치며

  다음 회차에서는 이렇게 모인 데이터를 활용하여 실제 training을 공유할까 합니다. 위의 Upsert를 이용하는 이유는 1분 데이터이다 보니 가끔 데이터 유실이 발생합니다. api나 프로그램의 문제가 아니라 upgrade를 위해서 잠시 프로그램을 내려 놓을때 수신하지 못한 데이터가 발생하는 거죠. 1회 수신 데이터를 100개 정도만 받아도 위의 방법을 사용하면 유실 없이 DB에 데이터를 저장할 수 있습니다. 

반응형
LIST

+ Recent posts