Pythonで財務分析1 ~EDINETからのデータ取得~
はじめに
昨年末から新NISAをきっかけに株式投資を始め、諸々勉強中の者です。
Udemyでこちらの講座に感化され、Pythonでデータ収集から可視化まで自動化できるのではと思い立ったのがことの発端です。
「はじめての財務分析（中級）～財務分析から戦略へ！解説3.5時間超の演習中心コース、多くの実例から自分の事業観を深めよう！」
ネットを探ると先人の方々の記事がたくさん見つかり、大いに参考にさせていただきました。
当面の目標
調べたい企業のROAツリーを作成して表示する
当初の目論見
大まかな流れとしては
- EDINETからデータを取得する
- ROAツリーの作成に必要なデータを抽出し別ファイルとして保存する
- Pythonの可視化ライブラリを使って2.で保存したファイルを読み込みROAツリー状にグラフを表示する
という目論見でした。
ところが結論としては「1.EDINETからデータを取得する」でまあまあ苦戦し、「2.ROAツリーの作成～～」でドはまりし、結局人力作業との組み合わせに落ち着く結果となりました。
本記事では「1.EDINETからデータを取得する」の内容を紹介します。
EDINET API
EDINETは金融庁が運営するサービスで、日本の上場企業の財務情報が公開されています。
2023年8月にEDINET API(Version2)が公開され、こちらのAPIを介してEDINETで公開されるデータをダウンロードできます。
EDINET APIの利用に当たっては予めユーザ登録を済ませ、APIキーの取得が必要です。
このあたりの手順はEDINET API仕様書に記載されているので、興味がある方はそちらをご参照ください。
EDINETからのデータ取得
先人の方々の記事でも言及されていましたが、特定の企業を絞ってデータをダウンロードするといった用途で使用する場合、工夫が必要です。
EDINET APIでは下記2種類のAPIが用意されています。
- 書類一覧API
- 書類取得API
書類一覧APIでは日付を指定してリクエストを投げ、その日付に提出された書類の書類管理番号を含むメタデータ一覧が返ってきます。
つまり、知りたい企業と関係のない企業の情報まで返ってきてしまいます。
なので受け取ったレスポンスに対して自前でフィルタリングする処理を記述する必要があります。(苦戦ポイントその１)
また、リクエストの引数に日付を指定するということは有価証券報告書が提出された日付を把握している必要があります。せっかくデータを自動取得したいのにわざわざEDINETにアクセスして過去10年分の提出日を調べてリストアップするのでは本末転倒です。
苦肉の策として「この企業は6月末にデータを提出しているから6月20日～6月30日の範囲でリクエストを投げれば欲しい情報が得られるだろう」という発想で進めることにしました。（苦戦ポイント２）
晴れて10年分の有価証券報告書の書類管理番号一覧が取得できたら次は書類取得APIの出番です。
こちらのAPIは比較的シンプルでAPI仕様書に従えば欲しい情報は得られます。
というわけでまあまあ苦戦したポイントは上述の2点で
- 書類一覧APIで受け取ったレスポンスに対して適切にフィルタリングする
- 10年分の有価証券報告書を取得するために日付の検索範囲に広がりを持たせてリクエストを繰り返す
といった工夫によりPythonでデータを自動取得するという当初の目論見を果たすことができました。
これらの工夫を織り込んだソースコードを公開します。このままで動くわけではないですが部分的にでも参考になれば幸いです。
class DocumentListApi:
"""
書類一覧APIに相当するクラス
"""
def __init__(self, api_key) -> None:
self.endpoint = "https://api.edinet-fsa.go.jp/api/v2/documents.json"
self.api_key = api_key
def get_document_indice_filername(self, edinet_code, start_month, start_day, days=10):
# 10年分の書類一覧を取得しedinet_codeが一致する有価証券報告書のdoc_idのみリストに格納して返す
doc_indice = []
filername = ""
date_l = self._get_datelist(start_month, start_day, days)
for s_date in date_l:
doc_meta = self._get_doc_metadata(edinet_code, s_date)
if doc_meta == None:
continue
doc_indice.append(doc_meta["docID"])
filername = doc_meta["filerName"]
return doc_indice, filername
def _get_doc_metadata(self, edinet_code, date):
url = f'{self.endpoint}?date={date}&type={2}&Subscription-Key={self.api_key}'
res = utils.get_with_sleep(url)
if res.status_code != 200:
print(f"bad request: {res.status_code}")
return None
j_res = res.json()
docs = j_res["results"]
f_docs = [doc for doc in docs if doc["edinetCode"] == edinet_code and "有価証券報告書" in doc["docDescription"]]
if len(f_docs) <= 0:
print("no data was found.")
return None
print(f"{len(f_docs)} document was found.")
return f_docs[0]
def _get_datelist(self, start_month, start_day, days=10):
dt_today = datetime.today()
start_year = dt_today.year - 10
end_year = dt_today.year if dt_today.month > 6 else dt_today.year - 1
date_l = []
for y in range(start_year, end_year+1):
start_date = datetime(y, start_month, start_day)
for i in range(0, days+1):
delta = timedelta(days=i)
date = start_date + delta
s_ymd = date.strftime("%Y-%m-%d")
date_l.append(s_ymd)
return date_l
class DocumentAcquisitionApi:
"""
書類取得APIに相当するクラス
"""
def __init__(self, api_key) -> None:
self.endpoint = "https://api.edinet-fsa.go.jp/api/v2/documents/"
self.api_key = api_key
def download(self, doc_id, savepath, data_type=5):
#data_type: 1:XBRL, 2:PDF, 3:代替書面・添付文書, 4:英文ファイル, 5:CSV
url = f'{self.endpoint}/{doc_id}?&type={data_type}&Subscription-Key={self.api_key}'
res = utils.get_with_sleep(url)
if res.status_code != 200:
print(f"bad request. {res.status_code}")
return
with open(savepath, "wb") as f:
for chunk in res.iter_content(chunk_size=1024):
f.write(chunk)
print(f"save file: {savepath}")
def main():
try:
# Edinetコードの入力受付
print("Enter edinet code. [Exxxxxx]")
edinet_code = input(">>> ")
# 文書番号一覧を取得
data_dpath = os.path.join(Path(__file__).parent.parent, "data", edinet_code)
doc_indice_fpath = os.path.join(data_dpath, "doc_indice.txt")
filername_fpath = os.path.join(data_dpath, "filername.txt")
doc_indice = []
filername = ""
if os.path.exists(doc_indice_fpath) == True and os.path.exists(filername_fpath) == True:
with open(doc_indice_fpath, "r", encoding="utf8") as f:
buff = f.readlines()
doc_indice = [l.rstrip("\n") for l in buff]
with open(filername_fpath, "r", encoding="utf8") as f:
filername = f.readline()
else:
print("Enter start month. [6]")
start_month = int(input(">>> "))
print("Enter start day. [20]")
start_day = int(input(">>> "))
print("Enter days. [10]")
days = int(input(">>> "))
doc_list_api = DocumentListApi(config.API_KEY)
doc_indice, filername = doc_list_api.get_document_indice_filername(edinet_code, start_month, start_day, days)
if len(doc_indice) <= 0:
print("no data was found.")
raise Exception()
# フォルダを作成
if os.path.exists(data_dpath) == False:
os.mkdir(data_dpath)
print(f"create directory: {data_dpath}")
raw_dpath = os.path.join(data_dpath, "00_raw")
os.mkdir(raw_dpath)
print(f"create directory: {raw_dpath}")
dst_dpath = os.path.join(data_dpath, "01_dst")
os.mkdir(dst_dpath)
print(f"create directory: {dst_dpath}")
with open(filername_fpath, "w", encoding="utf8") as f:
f.write(filername)
print(f"save file: {filername_fpath}")
with open(doc_indice_fpath, "w", encoding="utf8") as f:
for doc_id in doc_indice:
f.write(f"{doc_id}\n")
print(f"save file: {doc_indice_fpath}")
# 文書を取得
doc_acq_api = DocumentAcquisitionApi(config.API_KEY)
data_type = 1#1:xbrl, 2:pdf, 5:csv
ext = "pdf" if data_type == 2 else "zip"
for doc_id in doc_indice:
doc_fname = f"{doc_id}_{data_type}.{ext}"
doc_fpath = os.path.join(data_dpath, "00_raw", doc_fname)
doc_acq_api.download(doc_id, doc_fpath, data_type)
except Exception as ex:
print(ex)
Discussion