Day4
# -*- coding:utf-8 -*- #! usr/bin/env python3.7 from datetime import date import os import sys import requests import json from zipfile import ZipFile import pandas as pd import glob # XBRLをpython形式に変換するライブラリのフォルダパス sys.path.append(r'任意のフォルダパス') from xbrl_proc import read_xbrl_from_zip import urllib3 from urllib3.exceptions import InsecureRequestWarning urllib3.disable_warnings(InsecureRequestWarning) def get_list(start, end, company_list): """指定した期間、報告書種類、会社で報告書を取得し、取得したファイルのパスの辞書を返す""" #取得期間の日付リストを作成 day_term = [start + timedelta(days=i) for i in range((end - start).days)] # データ抜出時に使用する、有価証券報告書および四半期報告書のコードの設定 ordinance_code = "010" form_code_quart = "043000" # 四半期報告書 form_code_securities = "030000" # 有価証券報告書 # EDINETのAPIで、書類一覧を取得し、各日ごとに必要な書類の項目の抜き出し quart_list = [] # 四半期報告書のリスト securities_list = [] # 有価証券報告書のリスト print('EDINETへのアクセスを開始') for i,day in enumerate(day_term): url = "https://disclosure.edinet-fsa.go.jp/api/v1/documents.json" params = {"data": day, "type": 2} # 進捗表示 if i % 50 == 0: print('{i}日目:{day}を開始'.format(i, day)) # EDINETからの1日の書類一覧を取得 res = requests.get(url, params=params, verify=False) # 必要な書類の項目の抜き出し if res.ok: json_data = res.json() for data in json_data['results']: # 指定した会社の、指定した書類の抜き出し if data['ordinanceCode'] == ordinance_code and data['formCode'] == form_code_quart and data['filerName'].replace('株式会社', '') in company_list: quart_list.append(data) elif data['ordinanceCode'] == ordinance_code and data['formCode'] == form_code_securities and data['filerName'].replace('株式会社', '') in company_list: securities_list.append(data) else: print('アクセス失敗かも{day}'.format(day)) list_dic = {'四半期報告書':quart_list, '有価証券報告書':securities_list} return list_dic
day_term = [start + timedelta(days=i) for i in range((end - start).days)]
・リスト内包表記
list = [function(i) for i in range(params)]
・繰り返し処理
for i,day in enumerate(day_term)
enumerate関数がここで使えるのは今日の学び。
・辞書
params = {"data": day, "type": 2}
・フォーマット
print('{i}日目:{day}を開始'.format(i, day))
json_data = res.json()
・replaceと、繰り返し処理のデータ表示のしかたの例
for sth in data['result'] sth['content_in_result'] == sth[''].replace(' ',' ')
・リストを辞書に含めることができる
quart_list = []
securities_list = []
list_dic = {'四半期報告書':quart_list, '有価証券報告書':securities_list}
def get_zip(list_dic, quart_dir_path, securities_dir_path): """取得したいデータをzipファイルで取得してファイルパスのリストを返す""" dir_path_dic = {'四半期報告書':quart_dir_path, '有価証券報告書':securities_dir_path} file_path_dic = {'四半期報告書':[], '有価証券報告書':[]} # ダウンロードした有価証券報告書のパスを格納する辞書 for key in list_dic.keys(): # 既にzipをDLしている場合のため、既存んおdocIDリストを取得 files = os.listdir(dir_path_dic[key]) existing_docID_list = [file.split('.')[0].split('_')[1] for file in files if os.path.isfile(os.path.join(dir_path_dic[key], file))] print('{key}ファイルのDLを開始'.format(key)) for i, doc in enumerate(list_dic[key]): # zipファイルパスのリストの作成 file_name = doc['filerName'].replace('株式会社', '') + '_' + doc['docID'] file_path = os.path.join(dir_path_dic[key], file_name + ".zip") file_path_dic[key].append(file_path) # 所有していないファイルの場合はDLを行う if doc['docID'] not in existing_docID_list: # ファイルを取得 url_zip = "https://disclosure.edinet-fsa.go.jp/api/v1/documents/" + doc['docID'] params_zip = {"type":1} # 進捗表示 if i % 100 == 0: print('{i}ファイル目を開始'.format(i)) # データのDL res_zip = requests.get(url_zip, params=params_zip, verify=False, stream=True) # zipとして保存 if res_zip.status_code == 200: with open(file_path, 'wb') as f: for chunk in res_zip.iter_content(chunk_size=1024): if chunk: f.write(chunk) f.flush() return file_path_dic
・keys()とは、おそらく、辞書のkeyだな！
for key in file_path_dic.keys():
・これこれ、俺が気になったコードは。まず、リスト内包表記、でifの条件分岐をリストに入れることができるのか。
・split("第一引数"、"第二引数")があって、第一引数にはどこで区切るか。第二引数には初めから数えてどのくらい区切るか、を指定する。で[0]は一番初めのもの。それをさらにsplitして[1]から表示する。というような気がする。わからんが。わかったら直していこう！
existing_docID_list = [file.split('.')[0].split('_')[1] for file in files if os.path.isfile(os.path.join(dir_path_dic[key], file))] list = [something for x in Xs if sth]
次のコードへ。
def zip_to_df(file_path_dic): """ダウンロードしたzipをdfに変換して各会社のdicにして返す""" all_df_dic = {} for key in file_path_dic.keys(): print('{key}データの変換を開始'.format(key)) df_dic = {} for i, company_zip in enumerate(file_path_dic[key]): # 進捗表示 if i % 100 == 0: print('{i}ファイル目を開始'.format(i)) company_name = os.path.splittext(os.path.basename(company_zip))[0].split('_')[0] doc_name = ps.path.splittext(os.path.basename(company_zip))[0].split('_')[1] if company_name not in df_dic: # 会社が辞書中に存在しない場合 df_dic[company_name] = {} df_dic[company_name] = read_xbrl_from_zip(company_zip)[0] elif company_name in df_dic: # 会社が辞書中に存在する場合 df_dic[company_name] = pd.concat( [df_dic[company_name],read_xbrl_from_zip(company_zip)[0]]) all_df_dic[key] = df_dic return all_df_dic # 会社のリストを読み込み company_list = ['ホウスイ','カネコ種苗'] # 取得期間の設定:直近n日分 delta_day = 800 end = date.today() start = date.today() - timedelta(days=delta_day) # ダウンロードしたデータのフォルダパス quart_dir_path = r'四半期報告書を保存する任意のフォルダパス' securities_dir_path = r'有価証券報告書を保存する任意のフォルダパス' # XBRLデータの取得 list_dic = get_list(start,end,company_list) file_path_dic = get_zip(list_dic, quart_dir_path, securities_dir_path) # XBRLからデータ形式を変換 df_dic = zip_to_df(file_path_list) # 営業利益部分抽出 quart = all_df_dic['四半期報告書']['カネコ種苗'] quart_oi = quart[(quart['tag'] == 'OperatingIncome') & (quart['context'] == 'CurrentYTDDuration')] quart_oi[['第N期', '終了日', '値']].sort_values('終了日') quart_oi = quart_oi.rename({'終了日':'Date', '値':'OperatingIncome'}, axis=1).reset_index(drop=True) quart_oi['type'] = '四半期報告書' securities = all_df_dic['四半期報告書']['カネコ種苗'] securities_oi = securities[(securities['tag'] == 'OperatingIncome') & (securities['context'] == 'CurrentYTDDuration')] securities_oi[['第N期', '終了日', '値']].sort_values('終了日') securities_oi = securities_oi.rename({'終了日':'Date', '値':'OperatingIncome'}, axis=1).reset_index(drop=True) securities_oi['type'] = '四半期報告書' oi_df = pd.concat([quart_oi, securities_oi]).sort_values('Date').reset_index(drop=True) # 各四半期ごとの営業利益や、前年同期比の売上高営業利益率の変化を計算 oi_df['tempOperatingIncome'] = oi_df['OperatingIncome'] - oi_df['OperatingIncome'].shift(1) oi_df['CalcOperatingIncome'] = oi_df['OperatingIncome'].mask(oi_df['第N期'] == 0, oi_df['tempOperatingIncome']) oi_df['CalcOperatingIncome'] = oi_df['CalcOperatingIncome'].mask(oi_df['第N期'] == 3, oi_df['tempOperatingIncome']) oi_df['CalcOperatingIncome'] = oi_df['CalcOperatingIncome'].mask(oi_df['第N期'] == 2, oi_df['tempOperatingIncome']) oi_df['CalcOperatingIncome_YoYchangerate'] = oi_df['CalcOperatingIncome'] / oi_df['CalcOperatingIncome'].shift(4) # 棒グラフで可視化 oi_df.plot.bar(x='Date', y='CalcOperatingIncome', rot=45) oi_df.plot.bar(x='Date', y='CalcOperatingIncome_YoYchangerate', rot=45)
明日からは短めのコードをかくこと。
その中で一番ためになったコードを説明してみることにする。
以上！
