edinet xbrl python 写経

# データをまとめて取得する関数
SUMMARY_TYPE = 2
def download_all_documents(date, save_path, doc_type_codes=['120', '130', '140', '150','160','170']): 
    params = {'date':date, 'type':SUMMARY_TYPE}
    doc_summary = get_submitted_summary(params)
    df_doc_summary = pd.DataFrame(doc_summary['results'])
    df_meta = pd.DataFrame(doc_summary['metadata'])

    # 対象となる報告書のみ抽出
    if len(df_doc_summary) >= 1:
        df_doc_summary = df_doc_summary.loc[df_doc_summary['docTypeCode'].isin(doc_type_codes)]

        # 一覧を保存
        if not os.path.exists(save_path + date):
            os.makedirs(save_path + date)
        df_doc_summary.to_csv(save_path + date + '/doc_summary.csv')

        # 書類を保存
        for _, doc in df_doc_summary.iterrows():
            download_document(doc['docID'], save_path + date + '/')
            open_zip_file(doc['docID'], save_path + date + '/')
        return df_doc_summary

    return df_doc_summary


・SUMMARY_TYPEを2に設定しておく。
・download_all_documents関数には引数をdate, save_path, doc_type_codeを指定する
・get_submitted_summaryにparamsを渡すとjsonが取得できるため、paramsを用意する
・doc_summaryに取ってきたjsonを入れる。
・df_doc_summaryに取ってきたjsonからresultsデータを取り出して代入
・df_metaにも同様に取ってきたjsonからmetaデータを取り出して代入

~~~~if文~~~~~~~
・len(df_doc_summary).......df_doc_summaryの長さが1以上だったら
 ・df_doc_summaryに、locで位置を指定して、ここではdf_doc_summary['docTypeCode'] で列を指定、.isinでdoc_type_codesが含まれているものを返す。それをdf_doc_summaryに代入
  
  locの記述について。ありがとうございます。
  Pandasで要素を抽出する方法(loc,iloc,iat,atの使い方) - DeepAge

   #一覧を保存
  ・os.path(ファイルやディレクトリが指定したパスに存在するかを確認したり、パスからファイル名や拡張子を取得するなどを行う際に利用される)にsave_pathとdateが存在していなかったら、
   ・os.makedirsでディレクトリを作成
  ・df_doc_dummaryを、save_pathとdateとdoc_summary.csvを渡してcsvファイルに書き出す。

  #書類を保存
  ・繰り返し処理を用いてdf_doc_summaryのデータフレームに入っているデータをiterrows()メソッドを用いて、1行ずつ、インデックス名(行名)とその行のデータ(pandas.Series型)のタプル(index, Series)を取得する。
    ・前回作成したdownload_document関数にdoc['docID']とパス名を渡す
    ・open_zip_fileで解凍。


# zipファイルを解凍する関数
def open_zip_file(doc_id, save_path):
    if not os.path.exists(save_path):
        os.makedirs(save_path + doc_id)

    with zipfile.ZipFile(save_path + doc_id + '.zip') as zip_f:
        zip_f.extractall(save_path + doc_id)

・この関数はdoc_idとsave_pathを渡して使用する関数
・もしos.pathにsave_pathが存在していなかったら
 ・save_pathにdoc_idを付け加えたディレクトリを作成

 ・with zipfile.ZipFile........ZipFileオブジェクトはclose()で閉じる必要があるが、with文を使うと中身が終わったときに自動で閉じてくれる。 .extractall().........ZipFileオブジェクトのextractall()メソッドで、ZIPファイルの中身がすべて解凍(展開)される。第一引数pathに展開先のディレクトリのパスを指定する。省略するとカレントディレクトリに解凍される。アリ10。
PythonでZIPファイルを圧縮・解凍するzipfile | note.nkmk.me
つまりzip_f.extractall()で中身を解凍して解凍先に展開。


# 日付でループするので、そのための設定をし、とりあえず2020年3月1日から2020年6月30日までに提出された報告書を取得する。
from datetime import datetime, timedelta

def date_range(start_date: datetime, end_date: datetime):
    diff = (end_date - start_date).days + 1
    return (start_date + timedelta(i) for i in range(diff))

start_date = datetime(2020, 3, 1)
end_date = datetime(2020, 6, 30)
# 取得する。今回は有価証券報告書のみに限定する。指定したフォルダに4か月間に提出された有価証券報告書が保存される。
doc_type_code = ['120']
save_path = 'original_data/'

for i, date in enumerate(date_range(start_date, end_date)):
    date_str = str(date)[:10]
    df_doc_summary = download_all_documents(date_str, save_path)
    if i == 0:
        df_doc_summary_all = df_doc_summary.copy()
    else:
        df_doc_summary_all = pd.concat([df_doc_summary_all, df_doc_summary])
# 上記で取得したデータを使用して、取得したいデータを抽出する方法
from edinet_xbrl.edinet_xbrl_parser import EdinetXbrlParser

# 以下の関数でkeyとcontext_refを指定し、pathで指定したファイルからデータを取得する。
def get_one_xbrl_data(path, key, context_ref):
    parser = EdinetXbrlParser()

    # 指定したxbrlファイルをパースする
    xbrl_file_path = path
    edinet_xbrl_object = parser.parse_file(xbrl_file_path)

    # データの取得
    data = edinet_xbrl_object.get_data_by_context_ref(key, context_ref)
    if data is not None:
        return data.get_value()
# 例えば今期の売上を取りたい場合は、
# keyがjpcrp_cor:NetSalesSummaryOfBusinessResultsで、contextRefがCurrentYearDurationなので、以下のように指定する。
import glob

seccode = 7816
# 文書IDと提出日を取得
doc_id = df_doc_summary_all.query(f'secCode=="{seccode}0"').docID.values[0]
date = df_doc_summary_all.query(f'secCode=="{seccode}0"').submitDateTime.values[0]

# XBRLを取得
file_path = save_path + date[:10] + '/' + doc_id + '/XBRL/PublicDoc/*.xbrl'
file_path = glob.glob(file_path)[0]

# データを取得
get_one_xbrl_data(file_path, key='jpcrp_cor:NetSalesSummaryOfBusinessResults', context_ref='CurrentYearDuration')

# テキスト情報も取得可能
get_one_xbrl_data(file_path, key='jpcrp_cor:BusinessRisksTextBlock', context_ref='FilingDateInstant')


# 欲しいデータをまとめて取得
account_df = pd.read_csv('account_list.csv').fillna('')

def get_xbrl_data(path, account_df):
    parser = EdinetXbrlParser()

    # parse xbrl file and get data container
    xbrl_file_path = path
    edinet_xbrl_object = parser.parse_file(xbrl_file_path)
    ## データの取得
    account_df['value'] = ''
    for idx, account in account_df.iterrows():
        key = account['key']
        context_ref = account['context_ref']
        data = edinet_xbrl_object.get_data_by_context_ref(key, context_ref)

        if data is not None:
            account_df.loc[idx, 'value'] = data.get_value()
            #print(key, context_ref, account['value'])
    return account_df


また新しい発見したら更新する。

ありがとうございます!
https://data-analytics.fun/2020/12/22/ufo-analysis-data-collection/