머신러닝을 통한 주식 추천 시스템 개발 일지 (2) - DB 구성, 데이터 수집

개발 일지 (1)

머신러닝을 통한 주식 추천 시스템 개발 일지 (1) - 계기, 전체적 설계

DB 구성

먼저 DB를 구성해야 한다. RDB Scheme을 어떻게 구성할지에 대해 같이 회의하고 고민한 적은 많아도, 직접 테이블을 만들어보기는 또 처음이었다. 배우면서 해야 했다. 그래도 구글 클라우드 콘솔과 Cloudsql은 이전 회사 때문에 익숙했기 때문에 큰 어려움 없이 DB 인스턴스를 생성했다. (근데 처음에.. 과 스펙의 디비를 설정해놓아서 쌩돈이 날라갔다. 한 10만원 정도.. 다른 분들은 최대한 낮은 스펙의 DB를 사용하시길 추천)

그렇게 해서 줄인 스펙은 이 정도다. 나중에 얘기하겠지만 최저스펙에서는 디비가 너무 구려서 한단계 올렸다.

 

스킴을 디비에 반영하는 것은, 내가 사용하려고 했던 ORM인 TypeORM(js,ts기반)의 sync 기능을 활용하려고 했다. 이 기능을 활용하면 코드 상에서 클래스로 디비 구성을 정의하면, 알아서 해당 코드의 런타임이 디비와 연결될 때마다 디비를 업데이트해준다. (물론 프로덕션 환경에서 이렇게 쓰면 자살 행위다

 

대강 아래와 같은 느낌으로 코드를 구성하면 저 스펙에 맞게 디비를 수정해준다.

데이터 마련 1 - 회사 정보

유명한 라이브러리들 중에서는, 회사 Ticker를 넣으면 알아서 해당 회사 정보를 알려주는 것들이 많다. 하지만 당장 나에게 필요한 것은 회사 전체의 목록이었다. 구글링을 좀 해보니, 할 수 있는 방법들은 꽤 많았는데 내가 선택한 것은 FMP라는 주식 데이터 SASS? 회사에서 제공하는 무료 API였다. 제공하는 API가 매우 다양했고, 거래 가능한 모든 주식 목록을 가져오는 api도 있었다.  자세한 것은 아래 링크에서 참조하시길

https://site.financialmodelingprep.com/

Financial Modeling Prep - FinancialModelingPrep

해당 api를 활용해 작업할 때 있었던 약간의 애로 사항은, 이 api가 매우 많은 나라의 회사 데이터를 리턴하기 때문에, 정보를 보고, 적당히 걸러내야 했다. 나는 미국에 상장된 회사와 국내 상장된 회사만을 필터링해서 디비에 저장했다. (근데 특히 국내 상장사 정보의 경우, 누락 없이 모든 데이터를 주지는 않는 것 같았다. 하지만 어차피 누락되는 회사는 잡주겠거니 하고, 무시하고 진행했다.)

 

모델 훈련 시에 수치 정보 말고도 해당 회사가 어떤 종류의 회사인지도 넣어주면 좋을 것 같아서 추후에 industry, sector 정보를 얻어와 디비에 추가해줬다. 이 정보는 yahooquery라는 파이썬 라이브러리를 사용해 가져왔다.

 

데이터 마련 2 - 주식 거래 정보

회사 수만 몇천개 이상이었기 때문에 과거 어느 시점부터의 주식 거래데이터를 저장할지 결정하는 일이 중요했다. 1년치를 늘릴 때마다 대강 100-200만 행의 데이터가 늘어나는 꼴이었다. 물론 그래봤자 최적화된 형태의 데이터이기 때문에 그렇게 양이 크지는 않다. 하지만 과거 데이터의 경우에는 훈련의 경우에 쓰이기만 할 뿐이기 때문이고 내가 무슨 주식왕이 되고자 하는 것이 아니기 때문에.. 적당히 타협을 봐서 

2022년 1월 1일 이후의 데이터만 확보하기로 했다. 그리고 사실 훈련에 사용할 데이터셋의 관점에서 보자면, 비슷한 시장 상황을 겪고 있는 데이터를 활용하는 것이 좋기 때문에 코로나 이후 회복기라고 할 수 있는 2022년을 기점으로 데이터를 확보하기로 결정했다. 

 

원래는 크롤링을 직접 구현해서, 주식 거래 정보를 가져오려고 했으나 yahoo-finance2라는 npm 라이브러리가 있어서, typescript로 거래 정보를 가져오는 코드를 짰다.

 

데이터 마련 2.1 - 주식 거래 정보 - Cloud Run Job 병렬 태스크 활용

위 라이브러리에서 데이터를 호출할 때, 짧은 시간 안에 너무 많은 데이터를 호출하면 아예 응답을 막아버렸다. 그렇기 때문에 적당히 딜레이를 주며 정보를 가져와야 했다. 하지만 그런 식으로 가져오자니 시간이 너무 많이 걸릴 것 같아서 Google Cloud Run Job에 띄워서, 이 서비스가 지원하는 병렬 태스크 기능을 활용하기로 했다. 그리고 해당 서버리스 코드가 실행될 때, Job이 알아서 태스크 수 정보와 이게 몇번 인덱스의 태스크인지를 환경변수로 넣어준다. 이 정보를 활용해 병렬성을 활용한 형태로 코드를 짜면 된다.

 

(참고로 Cloud Run을 사용하려면 내가 작업한 코드를 Docker 이미지화시킨 후, Google Cloud 상에서 활용할 수 있도록 배포시켜야 한다. 이 번거로운 과정을 통합해주고 생략해주는 여러 툴? 시스템들이 있으나 나는 그냥 순정으로 이미지 빌드하고, Artifact Registry라는 곳에 직접 업로드하는 형태를 택했다. 자세한 것은 구글링 ㄱㄱㄹ)

콘솔에서 job 만들 때 일부 모습 - 이렇게 태스크 수를 설정할 수 있다.

 

 

하지만 이 때부터 뻘짓의 향연이 시작되었다. (물론 덕분에 새로운 경험 많이 했다)

 

[메모리 문제]

라이브러리를 통해 주식 거래 정보를 분할하여 가져오고, 그 과정이 다 끝난 다음에 디비에 저장하는 시퀀스를 밟으려고 했다. 그래서 디비에 데이터를 저장하기 전까지 수집한 데이터를 한 변수에 담고 있어야 하는, 즉 메모리 상에 띄워 놓고 있어야 하는 상황이었다. 그런데 마침 돈을 아끼겠다고 메모리 512mb 스펙의 제일 싼 인스턴스를 사용중이었기 때문에 OutOfMemory로 인스턴스가 터지는 상황이 발생했다. 메모리가 없어서 터지는 걸 실물로 본 건 처음이었다. 해결 방법은 심플했다. 거래 정보 수집과 디비 저장 시퀀스를 통합하고, 일정량의 데이터를 수집한 뒤에는 디비에 저장하고 해당 데이터를 참조하고 있는 어레이 변수를 비워준다. (그냥 빈 리스트로 재 할당해주면 알아서 GarbageCollecter가 슥 치워준다.)

 

 

[DB Connection 개수 문제]

앞서 태스크를 1000개 실행한다고 하지 않았는가? 하지만 무의미한 짓이었다. 내가 세팅한 DB는 값싼 스펙의 DB라서 connection을 그 정도로 동시에 유지하는 것이 불가능하다. 즉, DB가 동시에 커버할 수 있는 Max Connection이 내가 활용할 수 있는 병렬성의 최대 크기였다. 사실 훨씬 비싼 DB를 세팅했어도 1000개의 동시 connection을 커버할 수 있을 정도의 스펙은 쉽게 얻지 못했을 것이다. 

- 심플하게 DB 스펙을 2단계 정도 올려서 최대 50개의 connection을 커버할 수 있도록 했다.

- 병렬 태스크 전체 개수를 N으로 설정하더라도 동시 실행되는 태스크의 개수를 n개로 제한할 수 있는 기능이 있어 활용했다. 

 

[DB 저장 시, 동시 작업 행 수의 제한]

위에서 언급한 메모리 문제 때문에 수집한 정보량이 10000개를 넘어가면 디비에 저장한 뒤 메모리를 비워주고 다시 데이터를 수집하는 형태로 코드를 고쳤다. 하지만 이번엔 SQL 한 구문에서 동시에 작업할 수 있는 행 수의 제한이 있었다. 그래서 이것 또한 1000개 단위로 나눠 INSERT 작업을 했다.

 

[Console.log 비용 폭탄 문제 - Google Cloud Log]

로컬에서 코드 작업하고 테스트할 때 썼던 console.log를 하나도 삭제하지 않고 그대로 배포해 Cloud Run에서 작업해버렸다. 나는 여태까지 구글 콘솔에서 log 보는게 당연히 무료 기능인 줄 알았다. 근데 아니었다. 생각해보니, 어딘가 그 로그 정보도 저장을 해놓고 보여줘야 하는 건데 무제한 무료로 할 수는 없는 것이 당연했다. 무성의하게 곳곳에 박혀 있는 console.log는 심지어 거의 10만, 100만번 이상 실행되고, 심지어 json 자체를 대상으로 하는 것도 있었기 때문에 어마무시하게 로그가 생성되었다. 

대참사의 현장

그랬더니.. 그걸 모르고 거의 2번인가 3번 정도 실행했는데 14만원이 날라갔다. ㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋ 그 때의 충격은 이루 말할 수가 없다. 돈 아낀다고 디비 스펙 낮추고, Cloud Run 인스턴스 스펙 낮추고 쌩쇼를 했는데 결국 별 이상한 이유로 돈을 쓰레기 통에 버린 격이 됐다. 

 

[행 중복 저장 문제 (행 무결성)]

yahoo-finance2 라이브러리 자체적인 문제로 코드에 문제가 없어도 병렬 태스크 중 몇개가 실패하는 일이 발생하기도 했다. 이런 경우 자동으로 재실행되는 인스턴스 때문에, 중복되면 안되는 데이터가 중복되는 경우가 생겼다. 주식 거래 정보이므로 (회사, 거래 일자) 이 순서쌍은 테이블 상에서 유일해야 했으나 그것이 지켜지지 않는 데이터가 생겨나는 것을 나중에야 발견했다. 

 

해결책으로는 2가지를 실행했다.

1. (회사, 거래 일자) 정보를 한번에 Unique Index로 세팅한다. Unique 제약이 있다는 것은 알았지만 여러 컬럼의 정보를 합쳐서 Unique Index를 쓸 수 있다는 것은 몰랐다. 

2. Upsert 기능을 사용한다. Upsert라는 것은 Insert와 Update를 합친 말로 공식 SQL은 아니고 자주 사용되는 기능이라 저렇게 부르는 것 같다. (몇몇 orm 라이브러리에서는 공식적으로 upsert라는 기능을 제공하기도 한다) 즉, 기준에 따라 해당 데이터가 있으면 update하고, 없으면 insert하는 것이다. 이 기능을 활용하기 위해서는 Unique Index 즉, 기존에 데이터가 있는지 없는지 판단할 근거가 있어야 했는데 앞서 index를 세팅했기 때문에 이 기능을 바로 붙일 수 있었다.

 

[특정 쿼리의 압도적 비효율성]

데이터를 쌓아가는 와중에 점점 작업이 느려지는 것을 발견했다. 어디에 병목이 있나 살펴봤는데, 데이터를 수집하고 저장하는 부분이 아니라, 각 회사별 최신 거래 정보의 일자가 언제인지를 가져오는 쿼리가 매우 느렸다. 가만 살펴보니 그럴만도 했다. 회사 별로 최신 날짜가 언제인지를 거의 1000만 행 수준 규모가 된 테이블에서 선형 탐색을 하고 있으니 느릴 수 밖에 없었다. 

const latestMarketDates = await marketDataRepository
    .createQueryBuilder('stockData')
    .select(`to_char(MAX(stockData.date)::DATE, 'YYYY-MM-dd')`, 'latestDate')
    .addSelect('stockData.company_id', 'companyId')
    .where('stockData.company_id IN (:...targetCompanyIds)', {
      targetCompanyIds: filteredCompaniesByTaskNumber.map(
        (company) => company.id,
      ),
    }) // WHERE 조건 추가
    .groupBy('stockData.company_id')
    .getRawMany();

 

대안으로 이것저것 고민을 했었다. 1) MaterializedView(정기적으로 특정 쿼리를 실행해서 데이터를 뽑아내 갖고 있는 기능) 활용, 2) 정규성을 포기하고 회사별 최신 업데이트 일자를 다른 곳에 추가로 저장하기

 

근데 이런 고민을 할 필요가 없었다. 그냥 아래와 같이 특수한 index를 만들어주니 말도 안되게 빨라졌다. 

CREATE INDEX idx_stock_market_data_company_date ON stock_market_data (company_id, date DESC);

company_id와 date에 대해 복합 index를 만들되, date에 대하여 내림차순 세팅을 하면, 자연스레 회사별 최신 date를 빠르게 찾을 수 있는 구조가 되는 것이다. (이런게 가능하다는 것 또한 이번에 작업하면서 처음 알았다... 크흠)

 

 

뭐 이런 뻘짓이 있긴 했지만 최종적으로 3년치의 데이터를 초기 데이터로 확보하고, 주식 시장이 마감된 뒤에 최신 데이터를 수집할 수 있도록 국내 주식과 미국 주식을 분리하여 스케쥴러를 적절히 세팅해주었다~ 완료~

Cloud Run Job 자체를 한국, 미국으로 분리하여 생성하고 각기 다른 스케쥴러를 달아주었다. 코드베이스는 동일하게 가져가되, 인스턴스마다 설정하는 환경변수를 통해 차이를 주었다.