[gcp] 주식데이터 크롤링 해서 bigquery에 적재하기 - 1

HOU·2023년 3월 23일

gcp로pipline만들기

목록 보기

9/11

자동 적재를 위해 필요한 것

데이터를 가져오기
crontab으로 주기적으로 크롤링 하기
크롤링한 데이터 빅쿼리에 추가하기

데이터 가져오기

yfinance
pykrx
financedatareader
marcap
opendartreader

위와 같은 데이터 가져오는 법에 대해서 이번에 공부할 수 있었다. 위의 내용들은 구글링만 해도 엄청나게 데이터 가져오는 법이 잘 나와있다. 내가 참고한 부분들만 링크를 올려놓겠다.

yfinance, pykrx, financedatareader가 한분의 블로그, marcap과 opendartreader가 하나의 블로그이다. 두 블로그에서 정말 많은 도움을 받을 수 있었다.

사용하는 데이터

marcap 으로 전날의 발생한 종가 데이터를 opendartreader로 원하는 회사의 재무제표를 가지고 오기로 했다.
이 과정에서 문제가 분석가 분들과 여러가지 문제가 발생했는데 extract과정에서 이 데이터가 옳은지 옳지 않은지를 판단하면서 extract + transfer 과정이 한번에 진행되었단 점이다. 다음에는 이렇게 하지 말아야지.

데이터의 특징

재무재표는 1분기, 반기보고서, 3분기보고서, 사업보고서로 나뉘고 나오는 시점이 다 다르다. 이 것도 모르고 그냥 무식하게 크롤링 하다가 왜 데이터가 없지 했다. 물론 데이터가 있는 회사도 존재했다. 그 회사는 다른 회사보다 사업보고서가 빠른 이유는 무엇일까 ? 이래서 도메인지식이 중요하다고 하는 거 같다.

marcap은 실시간으로 데이터가 라이브러리 호출할 때 가져오지 않는다. 파일 형식으로 git에서 pull 받아야 오늘기준으로 어제 종가를 알 수 있다. 항상 사용하기 전에 git pull명령어를 입력해주자!

코드

marcap 으로 매일의 데이터를 가져오는 함수이다.

import pandas as pd
import numpy as np
from datetime import datetime
from datetime import timedelta


STOCK_CODE = ['036570', '251270', '225570', '078340']
STOCK_NAME = ["ncsoft", "netmarble", "nexon", "comtus"]
yesterday = datetime.now() - timedelta(1)

def marcap_data(start, end=None, code=None):
    '''
    지정한 기간 데이터 가져오기
    :param datetime start: 시작일
    :param datetime end: 종료일 (지정하지 않으면 시작일과 동일)
    :param str code: 종목코드 (지정하지 않으면 모든 종목)
    :return: DataFrame
    '''
    print("start time");
    start = pd.to_datetime(start)
    end = start if end==None else pd.to_datetime(end)
    df_list = []

    dtypes={'Code':str, 'Name':str, 
            'Open':int, 'High':int, 'Low':int, 'Close':int, 'Volume':int, 'Amount':float,
            'Changes':int, 'ChangeCode':str, 'ChagesRatio':float, 'Marcap':float, 'Stocks':int,
            'MarketId':str, 'Market':str, 'Dept':str,
            'Rank':int}
    
    print("반복문 시작")
    for year in range(start.year, end.year + 1):
        print(year)
        try:
            csv_file = './marcap/data/marcap-%s.csv.gz' % (year)
            print(csv_file)
            df = pd.read_csv(csv_file, dtype=dtypes, parse_dates=['Date'])
            df_list.append(df)
            print("df_list")
        except Exception as e:
            print(e)
            pass
    df_merged = pd.concat(df_list)
    print("df_merged")
    df_merged = df_merged[(start <= df_merged['Date']) & (df_merged['Date'] <= end)] 
    print("조건문 이후 df_merged")
    df_merged = df_merged.sort_values(['Date','Rank'])
    print("if문 시작")
    if code:
        print("들어왓니? code야?")
        df_merged = df_merged[code == df_merged['Code']]
    df_merged.set_index('Date', inplace=True)
    print("return 직전전")
    return df_merged[df_merged['Volume'] > 0]

now = datetime.now()
print(yesterday.date())
print("code:" + STOCK_CODE[0]);
for i in range(len(STOCK_CODE)):
    print("stock가져오기 함수 시작")
    df_game_stock = marcap_data("2018-01-01", end = now.date() ,code=STOCK_CODE[i])

    df_game_stock = df_game_stock.assign(Amount=df_game_stock['Amount'].astype('int64'), Marcap=df_game_stock['Marcap'].astype('int64'))
    print("stock가져오기 함수 종료")
    print(STOCK_NAME[0]);
    df_game_stock.to_csv(f"./crawling/data/stock_{STOCK_NAME[i]}_table.csv")

여기서 start와 end가 갖추어져있는데, 현재는 데이터를 다 가져와서 저렇게 되어 있지만 bigquery에 적재할 때는 덮어 쓸지 한줄만 추가할지 결정한 후에 소스코드를 수정할 생각이다.

OpenDart 소스코드 open api 가 필수적이다. 반드시 발급받아서 진행하기 바란다.

import OpenDartReader
import pandas as pd
import time

api_key = '발급받은 key'
dart = OpenDartReader(api_key)

reprts='11011'
corcodes = ['036570', '251270', '225570', '078340']
companys = ['ncsoft', 'netmarble', 'nexon', 'comtus']
years = [2018, 2019, 2020, 2021, 2022]

try:
    for i in range(4):
        for j in range(5):
                df = dart.finstate(corp=corcodes[i], bsns_year=years[j], reprt_code=reprts)
                df.to_csv('./data/finance_statement/{0}/{0}_{1}_files_{2}.csv'.format(companys[i], reprts, years[j]))

except Exception as e:
    print(e)