이전 일지 모음
머신러닝을 통한 주식 추천 시스템 개발 일지 (3) - 모델 훈련 feat.VertexAi활용
머신러닝을 통한 주식 추천 시스템 개발 일지 (2) - DB 구성, 데이터 수집
머신러닝을 통한 주식 추천 시스템 개발 일지 (1) - 계기, 전체적 설계
모델을 어떻게 배포할까?
사실 학교에서나, 혼자서나 머신러닝에 대해 공부할 때 계속 의문이 들었던 것은 이렇게 만든 모델을 어떤 방식으로 실제 프로그램에서 사용하는지였다. 찾아보니 유즈케이스는 크게 2개로 나뉜다.
1. 실시간 서비스에 모델이 활용되는 경우
- 보통 서버의 로컬에 모델을 위치시키고, 그걸 로컬 IO 즉, 메모리 상으로 가져와서 사용하는 듯 하다. 사용할 때마다 로컬에서 불러오는게 아니라 서버 시작할 때 메모리로 가져와서 서버가 실행되고 있는 동안 쭉, 메모리 상에 모델이 존재하게 하는 방법. (쥐피티가 알려준건데 내가 제대로 이해했는지는 모르겠다..;;)
- 근데 모델 사이즈가 엄청 큰 경우에도 이런 방법론으로 가능할지는 모르겠다. 메모리 크기를 그만큼 키운 컴퓨팅을 활용하려나? 친구네 연구실 보니깐 메모리가 120GB인 서버를 사용한다고 하긴 하더라...
2. 주기적이고 Async하게 모델을 실행해 대용량 결과를 한번에 얻는 경우
- 굳이 실시간으로 모델 예측을 사용할 필요가 없고 대용량 데이터를 쌓아뒀다가 한번에 예측한다.
- 한번에 데이터를 처리하니 여러모로 효율이 좋다. 굳이 실시간으로 모델을 사용할 필요가 없는 경우 대부분 이런 유즈케이스일 듯 하다.
- 혹은, 미리 자주 올 수 있는 Input을 캐시할 수 있도록 결과를 미리 저장하는 용도로 사용할 수도 있을 듯 하다.
역시 Vertex AI에서는 2가지 방법에 대한 자체 솔루션이 준비되어 있다. 굳이 이걸 사용할 필요는 없었지만 한번 슥 살펴봤는데 쓰기 나쁘지는 않아보였다. 좀 큰 모델을 관리하고, 결과까지 파이프라인으로서 관리하고 싶은 경우에 사용하는 것이 적절할 것 같다.
나는 굳이 실시간 배포를 할 필요가 없어서 일괄 예측을 사용해볼까 했는데, 파일이나 빅쿼리로 인풋 데이터를 전달해줘야 하는게 여러모로 귀찮고 굳이 싶어서, 그냥 직접 모델 사용 코드를 짜서 Cloud Run에 배포하기로 했다. 그리고 무엇보다 나는 전처리 조건이 조금씩 다른 여러 모델을 동시에 사용한 뒤, 각 분류 결과를 종합 점수로 변환하는 기능도 있어야 했기 때문에 굳이 이 서비스를 사용할 필요가 없었다. 즉, 여기에 모델을 배포해놓고 갖다 쓰더라도 어차피 추가로 번거로운 코드를 짜야 하는 상황이어서 서비스의 효용이 떨어졌다.
다만 모델을 사용하려고 하는데, 현재 회사 메인 서버가 파이썬이 아니라서 파이썬에서 훈련시킨 모델을 사용하기 까다로운 경우에 굳이 서버를 파지 말고, Vertex Ai의 '온라인 예측' 서비스를 이용하는 것은 꽤나 유용해 보인다.
나의 배포 - Cloud Run Job, Cloud Storage 활용
내 모델이 사용되는 것은 평일에 한번씩, 미국 주식시장 개장 전과 한국 주식시장 개장 전이다. 즉, 일주일에 10번만 실행되면 된다. 이런 경우엔 당.연.히. 실시간 서버로 배포해서 컴퓨팅 리소스와 내 돈을 낭비할 필요가 없다. 이럴 땐 역시 Cloud Run Job이다. 그런데 그럼 모델을 어디에 위치시켜야 할까? 앞선 게시물에서 언급했는데, 나는 총 24개의 모델을 활용한다. 용량을 좀 줄였으나 랜덤 포레스트 모델의 경우에는 개당 300MB 정도는 나간다. 도커 이미지 빌드할 때 24개 모델을 다 포함시키는 기행?이 가능한지는 모르겠지만 말도 안되는 것 같아서 애초에 시도하지는 않았고 Cloud Storage에서 모델을 순차적으로 불러오는 것으로 하기로 했다.
[코드의 흐름은 대강 다음과 같다]
1) 회사별 최신 주식 정보를 가져온다(전처리에 50일치가 필요해서 각 회사별 50일치)
2) 전처리를 통해 회사별 하나의 데이터로 압축한다. (50개 -> 1개)
3) 모델 종류, 몇일 뒤의 수익률이냐, 몇개로 분류하는 모델이냐에 따라 a.다른 모델을 불러오고 b.다른 전처리 방법을 사용한다.
4) 회사 별 각 케이스의 결과를 적당한 수식으로 합산해 단기, 중기, 총합 점수를 도출하고 랭킹화한다.
5) 결과를 정리하여 디스코드 웹훅을 통해 채널로 보내준다.
코드 정리를 안 하면서 해서 좀 부끄럽지만 전체적인 흐름이 약간 복잡하기에 코드 자체를 아래 공유한다
import pandas as pd
from src.pre_process import pre_process_all_data
from src.load_model import load_cb_model, load_rf_model
from src.save_csv import save_dataframe_to_gcs
from src.timer import timer
from src.load_data import load_data_from_db
from src.load_model import load_rf_model
from src.send_discord import send_discord_message
from catboost import Pool
from dotenv import load_dotenv
import os
load_dotenv(override=True)
NATION = os.getenv("NATION", "")
if NATION == "":
raise ValueError("환경변수 NATION을 설정해주세요 - KR or US")
N_DAY_RETURN_LIST = [5, 7, 21, 30]
N_CLASS_LIST = [2,3, 4]
MODEL_LIST = ['CB', 'RF'] # CatBoost, RandomForest
def score_by_condition(value, n_class):
coef = 3.5 if n_class == 2 else 3 if n_class == 3 else 2.5
return value * coef
def main():
print("main 함수 시작")
with timer("데이터 불러오기"):
data = load_data_from_db(nation=NATION)
print(f"원본 데이터 형태 {data.shape}")
unique_symbols = data[['symbol', 'name','exchange_short_name']].drop_duplicates(subset=['symbol'])
data.drop(columns=['name'], inplace=True)
print(f"unique_symbols {len(unique_symbols)}")
score_board = unique_symbols.set_index('symbol')
unique_dates = data['date'].unique()
latest_date = unique_dates.max()
for model_name in MODEL_LIST:
for n_day_return in N_DAY_RETURN_LIST:
for n_class in N_CLASS_LIST:
print(f"모델 {model_name} {n_day_return} {n_class} 전처리 시작")
processed_data = pre_process_all_data(
data, n_day_return=n_day_return, n_class=n_class,
no_cat=model_name=='RF', one_hot_encoding=False,
drop_symbol=False, for_prediction=True,internal_normalize=False)
print(f"전처리 데이터 형태 {processed_data.shape}")
symbols = processed_data.pop('symbol')
print(f"모델 {model_name} {n_day_return} {n_class} 모델 예측 시작")
# 모델 가져오기
if model_name == 'RF':
rf = load_rf_model(n_day_return=n_day_return, n_class=n_class, nation=NATION)
# 여기서 모델 예측하고 symbol 별 결과를 변수에 기록
predict_result = rf.predict(processed_data, )
predict_result_with_symbol = pd.DataFrame({'symbol': symbols, 'predict': predict_result})
else:
cb = load_cb_model(n_day_return=n_day_return, n_class=n_class, nation=NATION)
cat_cols = ["sector_key", "industry_key", "exchange_short_name"]
processed_data[cat_cols] = processed_data[cat_cols].fillna("Unknown")
pool = Pool(processed_data, cat_features=cat_cols)
predict_result = cb.predict(pool,)
predict_result = predict_result.reshape(-1)
predict_result_with_symbol = pd.DataFrame({'symbol': symbols, 'predict': predict_result})
# n_class에 따라 score 점수 다르게 부여하여 predict_result에 추가
predict_result_with_symbol['score'] = predict_result_with_symbol['predict'].apply(lambda x: score_by_condition(x, n_class))
symbol_indexed_predict_result = predict_result_with_symbol.set_index('symbol')
# 컬럼 명 지정
column_name = f"{model_name}_{n_day_return}_{n_class}"
# score_board에 예측 결과 추가
score_board[column_name] = symbol_indexed_predict_result['score']
print(f"모델 {model_name} {n_day_return} {n_class} score_board에 추가 완료")
# score_board에 저장된 결과를 symbol 별로 합산
score_board['total_score'] = score_board.drop(columns=['name','exchange_short_name'], errors='ignore').sum(axis=1)
# 단기 점수에 한하여 symbol 별로 합산
score_board['short_term_score'] = score_board[[f'{model_name}_{n_day_return}_{n_class}' for model_name in MODEL_LIST for n_day_return in [5, 7] for n_class in N_CLASS_LIST]].sum(axis=1)
score_board['mid_term_score'] = score_board[[f'{model_name}_{n_day_return}_{n_class}' for model_name in MODEL_LIST for n_day_return in [21, 30] for n_class in N_CLASS_LIST]].sum(axis=1)
# total_score 기준으로 내림차순 정렬
total_score_descending = score_board.sort_values(by='total_score', ascending=False)
save_dataframe_to_gcs(total_score_descending, nation=NATION)
total_tops = total_score_descending.head(6)
# short_term_score 기준으로 내림차순 정렬
short_term_tops = score_board.sort_values(by='short_term_score', ascending=False).head(3)
# mid_term_score 기준으로 내림차순 정렬
mid_term_tops = score_board.sort_values(by='mid_term_score', ascending=False).head(3)
send_discord_message(short_term_tops, "단기(5, 7일) 추천 종목", latest_date,NATION,'🔥' )
send_discord_message(mid_term_tops,"중기(21, 30일) 추천 종목",latest_date,NATION, '🌟' )
send_discord_message(total_tops,"종합 추천 종목", latest_date, NATION,'🚀' )
# now_time = pd.Timestamp.now()
# score_board.to_csv(f'score_board_{now_time}.csv')
if __name__ == "__main__":
main()
최종 점수표는 아래와 같이 생겼다.
여담) 토스 증권 크롤링 - 미국 주식 id
디스코드로 결과를 보내줄 때, 주식 이름을 클릭하면 바로 주식 상세 정보를 보여줄 수 있는 사이트로 이동시켜주고 싶었다. 한국 주식의 경우에는 네이버 주식으로 이동시켜주려 했으나 미국 주식의 경우에는 외국 사이트로 보내주는게 좀 찝찝해서 고민하다가 나도 토스 증권을 사용하고 있기도 하고, 토스 증권 웹 사이트로 보내주는게 여러모로 이쁘고 깔끔하니 좋다고 생각했다.
하지만, 한국 주식의 경우에는 Ticker symbol만 알면 해당 주식의 상세 페이지로 이동시킬 수 있는 반면, 미국 주식의 경우에는 알 수 없는 id를 사용했다. 그래서 그냥 포기할까 하다가 어차피 분석의 대상이 되는 미국 회사도 1000개 미만이니(구린 회사는 다 지워버림) 크롤링으로 고유 id를 가져오는 코드를 짰다. 예외가 약간 계속 있어서 3-4번 뻘짓 했지만 그 이후에는 잘 됐다. 혹시 크롤링할 일이 있는 분은 참조하세용.
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager
import time
import pandas as pd
# 심볼 리스트 예시
raw_data = pd.read_csv('data/us_company_market_data.csv')
symbols = raw_data['symbol'].unique()
print(f"{symbols.shape[0]} 개")
# 셀레니움 브라우저 설정
options = webdriver.ChromeOptions()
# options.add_argument("--headless") # 창 없이 실행하려면
driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()), options=options)
# 사이트 진입
driver.get("https://tossinvest.com/") # 실제 사용하는 주식 정보 사이트 주소로 변경
# 데이터 저장용 리스트
data = []
# 검색창 찾기
search_box = driver.find_element(By.CSS_SELECTOR, "nav button:first-of-type") # nav 밑에 있는 첫 번째 버튼 선택
# 검색창 클릭
search_box.click()
time.sleep(0.5)
for symbol in symbols:
print(f"Searching for: {symbol}")
# 심볼 입력
driver.switch_to.active_element.send_keys(symbol)
time.sleep(1)
# 'data-section-name'이 '종목'인 div 찾기
try:
section_div = driver.find_element(By.CSS_SELECTOR, "div[data-section-name='종목']")
except Exception as e:
data.append({"symbol": symbol, "id": None})
search_box = driver.switch_to.active_element
search_box.clear()
search_box.click()
driver.switch_to.active_element.send_keys("")
print(f"Section not found for {symbol}")
continue
# 해당 div 밑의 첫 번째 버튼 클릭
first_button = section_div.find_element(By.TAG_NAME, "button")
first_button.click()
time.sleep(1.5)
# 현재 URL 가져오기
current_url = driver.current_url
# 고유 ID 추출
if "stocks" in current_url and "order" in current_url:
unique_id = current_url.split("stocks/")[1].split("/order")[0]
print(f"Extracted ID for {symbol}: {unique_id}")
search_button = driver.find_element(By.CSS_SELECTOR, "button[data-section-name='검색']")
print(f"Search button found for {symbol}")
try:
search_button.click()
except Exception as e:
print("Search button click failed")
print("상장폐지된 주식인 듯?")
# 뒤로 가기
driver.back()
time.sleep(1)
# 다시 검색창 찾기
search_button = driver.find_element(By.CSS_SELECTOR, "button[data-section-name='검색']")
search_button.click()
data.append({"symbol": symbol, "id": unique_id})
time.sleep(1)
# 반복문 종료 후 데이터프레임 생성 및 CSV 저장
df = pd.DataFrame(data)
df.to_csv("symbol_ids.csv", index=False)
driver.quit()
프로그램 완성 후기
퇴사 이후에 방학 때부터 시작하긴 했지만 이래저래 다른 일이랑도 겹치고, 개인적인 일도 있고 해서 진도가 매우 느리게 나갔다. 놀랍게도 원래 계획은 한 2-3주만에 끝내는 것이었는데 복한한 이후에서야 프로그램을 완성하게 되었다.
잘하고 익숙한 것(나는 거의 앱이나 웹 개발을 위주로 해왔다)을 벗어난 경험을 좀 하고 싶었는데 나랑 거리가 멀었던 머신러닝과 클라우드, 백엔드 쪽 작업을 쭉 하면서 나름 배운 것도 많고 개인 학습 차원에서는 만족스러운 프로젝트였다.
프로젝트 자체 완성도로 보면 부족한게 많다. 애초에 주식 추천을 정확하게 하는 프로그램을 겨우 나 따위가 잘 작동하도록 만드는 그런 평행 우주가 있다면 일단 나부터 거기로 가고 싶다.... ㅋㅋ. 각 모델의 precision이나 recall이 전체적으로 매우 떨어지기 때문에 제대로된 예측을 하고 있다고 보기 힘들다. 어쩌면 머신이 그냥 동전 던지기를 계속 하는 것일수도 있다. 그러니 대안으로 여러 모델을 활용할 생각을 했던 것인데... 아이디어는 좋았지만 객관적 근거가 있는 솔루션은 아닌 것 같다. 바보 24명의 의견을 모았을 때 더 현명한 결론이 도출되는가...? 맞을 수도 있고 아닐 수도 있다. 다만 모델의 결과가 좋지 않았던 상황에서 그나마 할 만한 선택이었던 것 같다.
이 추천을 곧이 곧대로 믿지 말고, 기술적 지표 상 고려해볼만한 주식을 새롭게 알게 된다는 정도의 효용으로 활용하면 나쁘지는 않은 것 같다. 아버지가 요새 주식 중기 단타?를 좀 치신다고 했는데, 앞으로 여기에서 추천한 주식을 트라이해보시길 권유해드렸다. 한번도 부모님이 사용할 만한 프로덕트를 만들어 본 적이 없는게 마음이 걸렸는데 그래도 아버지가 관심을 보이시는 프로그램을 만들었다는 것이 기분이 좋았다. 저번에 집에 갔을 때 아버지 폰에 손수 디스코드도 설치해서 사용하실 수 있게끔 세팅하고 왔다. ㅋㅋ
슬슬 다음엔 좀 더 제대로 된 걸 만들어야겠다.
'개발 프로젝트' 카테고리의 다른 글
| 머신러닝을 통한 주식 추천 시스템 개발 일지 (3) - 모델 훈련 feat.VertexAi활용 (2) | 2025.03.31 |
|---|---|
| 머신러닝을 통한 주식 추천 시스템 개발 일지 (2) - DB 구성, 데이터 수집 (0) | 2025.03.30 |
| 머신러닝을 통한 주식 추천 시스템 개발 일지 (1) - 계기, 전체적 설계 (1) | 2025.03.30 |