자동 취득한 주가 데이터를 BigQuery에 저장

취득 방법은 이쪽 → Selenium을 사용하여 주가 데이터 자동 획득

소개

CSV 데이터는 백업 의미를 포함하여 Cloud Storage에 저장
Cloud Storage에서 CSV 데이터를 임시 테이블로 생성
작성한 임시 테이블의 레코드를 가공하여 본 테이블을 작성

Cloud Storage에 저장

Cloud Storage, BigQuery를 사용하기 위해 패키지 설치

$ pip install google-cloud-bigquery
$ pip install google-cloud-storage

패키지 가져오기, 클라이언트 만들기

...
from google.cloud import bigquery, storage
from google.cloud.exceptions import NotFound

...
class YahooDownload:
    def __init__(self):
        self.str_client = storage.Client()
        self.bq_client = bigquery.Client() 

CSV 업로드
GCP 콘솔에서 버킷을 만들고, 이번에는 yahoo-finance-download라는 이름으로 버킷을 만듭니다.

...
    def upload(self):
        bucket = str_client.bucket("weekend-hackathon-stock")
        for _, v in self.download_list.iterrows():
            stock_no = v["stock_no"]
            blob = bucket.blob(f"yahoo-finance-download/{stock_no}.T.csv")
            blob.upload_from_filename(f"{self.download_dir}/{stock_no}.T.csv")

처리 추가

...
    def main(self):
        try:
            self.login()
            self.download()
            self.upload()
            pass
        except TimeoutException as ex:
            print(ex)
            pass
        except Exception as e:
            print(e)
            pass
        finally:
            time.sleep(5)
            self.driver.close()
            self.driver.quit()

실행하면 Cloud Storage에 업로드됨

BigQuery에 저장

BigQuery에서 bq 명령을 사용하여 버킷 만들기

$ bq --location=US mk --dataset --description "yahoo_finance_download dataset using" \
weekend-hackathon:yahoo_finance_download

포인트로서는 자동 취득한 CSV 데이터의 일자 포맷이 BigQuery의 DATE 포맷과 일치하지 않기 때문에 일시 테이블에 일자를 STRING로 작성한 후에 일자를 PARSE_DATE('%Y/%m/%e', trade_at) 로 퍼스 해 본 테이블에 데이터 추가하지 않으면 안 된다

취득 CSV

日付,始値,高値,安値,終値,出来高,調整後終値
2021/1/22,423,425,423,424,5200,424
2021/1/21,425,425,422,422,2500,422

BigQuery 추가

...
    def save(self):
        for _, v in self.download_list.iterrows():
            stock_no = v["stock_no"]

            # 一時テーブル準備
            external_config = bigquery.ExternalConfig("CSV")
            external_config.source_uris = [
                f"gs://weekend-hackathon-stock/yahoo-finance-download/{stock_no}.T.csv"
            ]
            external_config.schema = [
                bigquery.SchemaField("trade_at", "STRING", "REQUIRED"),
                bigquery.SchemaField("open", "FLOAT", "REQUIRED"),
                bigquery.SchemaField("high", "FLOAT", "REQUIRED"),
                bigquery.SchemaField("low", "FLOAT", "REQUIRED"),
                bigquery.SchemaField("close", "FLOAT", "REQUIRED"),
                bigquery.SchemaField("volume", "INTEGER", "REQUIRED"),
                bigquery.SchemaField("adj_close", "FLOAT", "REQUIRED"),
            ]
            external_config.options.skip_leading_rows = 1

            table_id = f"tmp_{stock_no}"

            # 一時テーブル参照結果を本テーブルに追加            
            job_config = bigquery.QueryJobConfig(
                destination = (
                    f"weekend-hackathon.yahoo_finance_download.{stock_no}"
                ),
                write_disposition="WRITE_TRUNCATE",
            )

            job_config.table_definitions = {table_id: external_config}
            sql = f"SELECT PARSE_DATE('%Y/%m/%e', trade_at) AS trad_at, open, high, low, close, volume, adj_close FROM {table_id}"
            query_job = self.bq_client.query(sql, job_config=job_config)
            query_job.result()

처리 추가

...
    def main(self):
        try:
            self.login()
            self.download()
            self.upload()
            self.save()
            pass
        except TimeoutException as ex:
            print(ex)
            pass
        except Exception as e:
            print(e)
            pass
        finally:
            time.sleep(5)
            self.driver.close()
            self.driver.quit()

실행하면 BigQuery에 주식 데이터가 저장됩니다.


다음 번에는 BigQuery 데이터를 처리, 분석하고 Streamlit을 사용하여 앱처럼 보입니다.
좋아! 라고 생각하면 LGTM 부탁합니다

【PR】 주말 하카손이라는 이벤트하고 있습니다! → htps : // 우에 켄 d는 c인가. 그리고 ysc Rechin. jp/아보 t/