株のシステムトレードをしよう - 1から始める株自動取引システムの作り方

株式をコンピュータに売買させる仕組みを少しずつ作っていきます。できあがってから公開ではなく、書いたら途中でも記事として即掲載して、後から固定ページにして体裁を整える方式で進めていきます。

J-Quants API Clientを使って、過去の財務情報を取得してDataFrameにする

下記の記事では、先日追加されたばかりのサンプルを使って、実際に財務情報を取得した。

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'
)

得られるDataFrame
得られるDataFrame

これを今度は配当利回りのサンプルを使用して、解析してみようと思う。下記のファイルを参考にする。

github.com

(C) 2020 dogwood008 禁無断転載 不許複製 Reprinting, reproducing are prohibited.