下記の記事では、先日追加されたばかりのサンプルを使って、実際に財務情報を取得した。
how-to-make-stock-trading-system.dogwood008.com
今回ではそれを更に変更し、pickleとして保存して再度呼び出せるようにしたり、差分だけ自動検知してpickleに再保存するようにする。
from datetime import datetime from dateutil import tz import jquantsapi import pandas as pd import os import numpy as np from typing import Union, List, Optional DatetimeLike = Optional[Union[datetime, pd.Timestamp, str]] class Statements: def __init__(self, storage_dir_path: str, jqapi: jquantsapi.Client): self.storage_dir_path:str = storage_dir_path self._jqapi:jquantsapi.Client = jqapi self.statements_file:str = f"{self.storage_dir_path}/statements.pkl" self._cache_dir = f"{self.storage_dir_path}/raw_statements" def fetch_from_api_and_append(self, start_dt: DatetimeLike=None, end_dt: DatetimeLike=None): ''' Arguments ------------------- start_dt: DatetimeLike 取得開始日 end_dt: DatetimeLike 取得終了日 ''' previous_df = self.load_df() df = self.fetch_from_api(start_dt, end_dt) appended_df = pd.concat([previous_df, df]) unique_df = appended_df.loc[~appended_df.duplicated(subset='DisclosureNumber')] self.save_df(unique_df) return unique_df def fetch_from_api(self, start_dt: DatetimeLike=None, end_dt: DatetimeLike=None): ''' Original: J-Quants/jquants-api-client-python https://github.com/J-Quants/jquants-api-client-python/blob/da16a23d85c80a0106673f0a0deaec3437016418/examples/20220825-003-dividend.ipynb Arguments ------------------- start_dt: DatetimeLike 取得開始日 end_dt: DatetimeLike 取得終了日 ''' if not start_dt or not end_dt: # 過去3ヶ月に発表された財務情報を取得します now = pd.Timestamp.now(tz="Asia/Tokyo") start_dt = now - pd.Timedelta(90, unit="D") end_dt = now if end_dt.hour < 1: # データ更新時間前の場合は日付を1日ずらします。 end_dt -= pd.Timedelta(1, unit="D") if not os.path.isfile(self.statements_file): os.makedirs(self._cache_dir, exist_ok=True) df_s = self._jqapi.get_statements_range( start_dt=start_dt, end_dt=end_dt, cache_dir=self._cache_dir ) df = self._format_df(df_s) return df def _format_df(self, df_s: pd.DataFrame): ''' Original: J-Quants/jquants-api-client-python https://github.com/J-Quants/jquants-api-client-python/blob/da16a23d85c80a0106673f0a0deaec3437016418/examples/20220825-003-dividend.ipynb ''' # float64にするために"-"をnp.nanに置き換えます df_s.replace({"-": np.nan}, inplace=True) float_columns: List[str] = ['ResultDividendPerShareFiscalYearEnd', 'EarningsPerShare', 'ForecastDividendPerShareAnnual', 'ForecastEarningsPerShare'] for column in float_columns: df_s.loc[:, column] = pd.to_numeric(df_s.loc[:, column], errors='coerce') # 日付型に変換します date_columns: List[str] = ['DisclosedDate', 'CurrentPeriodEndDate', 'CurrentFiscalYearStartDate', 'CurrentFiscalYearEndDate'] for column in date_columns: df_s.loc[:, column] = pd.to_datetime( df_s[column], format="%Y-%m-%d" ) df_s.sort_values("DisclosedUnixTime", inplace=True) df_s.index = df.DisclosureNumber df_s.index.name = 'DisclosureNumberIndex' return df_s def save_df(self, df: pd.DataFrame) -> pd.DataFrame: df.to_pickle(self.statements_file) print(f"save file: {self.statements_file}") def load_df(self) -> pd.DataFrame: ''' Original: J-Quants/jquants-api-client-python https://github.com/J-Quants/jquants-api-client-python/blob/da16a23d85c80a0106673f0a0deaec3437016418/examples/20220825-003-dividend.ipynb ''' print(f"file exists: {self.statements_file}, loading") df_s = pd.read_pickle(self.statements_file) return df_s my_refresh_token: str = '(token)' cli = jquantsapi.Client(refresh_token=my_refresh_token) parent_path = '/content/drive/MyDrive/drive_ws/marketdata' statements = Statements(parent_path, cli) df = statements.fetch_from_api_and_append( start_dt='2017-01-01', end_dt='2022-12-31' )
これを今度は配当利回りのサンプルを使用して、解析してみようと思う。下記のファイルを参考にする。