心ゆく日並

辛さを辛味に、苦しみを苦味に転じら れればこの日々も味わい深い

Python プログラム

EDINETのAPIを使って企業の財務情報を取得する④:PythonとEDINETのAPIを使って書類を取得する

投稿日:

ども♪マコトです。

EDINETのAPIを使って企業の財務情報を取得するの最終版です。

関数を使って、ファイルの取得日付、ファイルの形式、解凍の有無などを設定できるようにしてみました。

以下の部分を変更すれば、思った通りの財務情報を取得できるはずです。

#ファイルの取得日を設定
FileDate = "2019-10-01"
#すべて取得する0、有価証券報告書、半期報告書、四半期報告書のみを取得する 1
SRFlg = 1
#提出本文書及び監査報告書 1,PDF 2,替書面・添付文書 3,英文ファイル 4
DocType = 1
#zipを解凍しない 0、zipを解凍する 1
ZipFlg = 0

 

以下がすべてのプログラムです。

"""
EDINETのAPIを使って有価証券報告書などを取得する
"""
#書類一覧APIのやつ
import os
import sys
import time
import requests
import Cls_Connect as Conn
import json
from zipfile import ZipFile
from io import BytesIO
import shutil

#保存するディレクトリ
DOWNLOAD_SAVE_DIR = r"C:\data"

#ファイルの取得日を設定
FileDate = "2019-10-01"
#すべて取得する0、有価証券報告書、半期報告書、四半期報告書のみを取得する 1
SRFlg = 1
#提出本文書及び監査報告書 1,PDF 2,替書面・添付文書 3,英文ファイル 4
DocType = 1
#zipを解凍しない 0、zipを解凍する 1
ZipFlg = 0

def getDocumentlist():
    #バージョンが変わるたびに”v1"の部分を変更する必要がある模様
    url = "https://disclosure.edinet-fsa.go.jp/api/v1/documents.json"

    #必須パラメータのdateを入れる
    #出力結果を変えるためにtypeを2にする(type=1だとmetadataしか出力されない)
    params = {"date": FileDate, "type": 2}

    #このAPIはREST APIなのでverifyをFalseにして接続するときのSSLを無効化する
    res = requests.get(url, params=params, verify=False)

    json_data = json.loads(res.text)

    saveFileName = "EDINET-FSA" + FileDate + ".CSV"
    saveFilePath = os.path.join(DOWNLOAD_SAVE_DIR, saveFileName)

    # 書き出し用のファイルを開く
    out_file = open(saveFilePath,"w")
    #ヘッダー
    dataline = "連番,書類管理番号,提出者EDINETコード,提出者証券コード,提出者法人番号,提出者名,ファンドコード,府令コード,様式コード,書類種別コード,期間(自),期間(至),提出日時,提出書類概要,発行会社EDINETコー,"\
            "対象EDINETコード,子会社EDINETコード,臨報提出自由,新書類管理番号,操作日時,取下区分,書類情報修正区分,開示不開示区分,XBRL有無フラグ,PDF有無フラグ,代替書面・添付文章有無フラグ,英文ファイル有無フラグ" + '\n'
    out_file.write(dataline)

    for jsonline in json_data["results"]:
        dataline = str(jsonline["seqNumber"]) + ',' + str(jsonline["docID"]) + ',' + str(jsonline["edinetCode"]) + ',' + str(jsonline["secCode"]) + ',' + str(jsonline["JCN"]) + ',' + \
                str(jsonline["filerName"]) + ',' + str(jsonline["fundCode"]) + ',' + str(jsonline["ordinanceCode"]) + ',' + str(jsonline["formCode"]) + ',' + str(jsonline["docTypeCode"]) + ',' + \
                str(jsonline["periodStart"]) + ',' + str(jsonline["periodEnd"]) + ',' + str(jsonline["submitDateTime"]) + ',' + str(jsonline["docDescription"]) + ',' + str(jsonline["issuerEdinetCode"]) + ',' + \
                str(jsonline["subjectEdinetCode"]) + ',' + str(jsonline["subsidiaryEdinetCode"]) + ',' + str(jsonline["currentReportReason"]) + ',' + str(jsonline["parentDocID"]) + ',' + str(jsonline["opeDateTime"]) + ',' + \
                str(jsonline["withdrawalStatus"]) + ',' + str(jsonline["docInfoEditStatus"]) + ',' + str(jsonline["disclosureStatus"]) + ',' + str(jsonline["xbrlFlag"]) + ',' + str(jsonline["pdfFlag"]) + ',' + \
                str(jsonline["attachDocFlag"]) + ',' + str(jsonline["englishDocFlag"]) + '\n'
        out_file.write(dataline)

    #ファイルをクローズする
    out_file.close()

    return saveFilePath


def CheckFlg(word):
    if SRFlg == 0 or (SRFlg == 1 and (word[9] == '120' or word[9] == '130' or word[9] == '140' or word[9] == '150' or word[9] == '160' or word[9] == '170')):
        #フラグが立っている銘柄のみを取得する
        if DocType == 1:
            if word[23] == '1':
                flg = 1
            else:
                flg = 0
        elif DocType == 2:
            if word[24] == '1':
                flg = 1
            else:
                flg = 0
        elif DocType == 3:
            if word[25] == '1':
                flg = 1
            else:
                flg = 0
        elif DocType == 4:
            if word[26].strip("\n") == '1':
                flg = 1
            else:
                flg = 0
    else:
        flg = 0

    return flg


def download_xbrl_file(word,flg):
    if flg == 1 :
        #書類取得APIのやつ
        url = "https://disclosure.edinet-fsa.go.jp/api/v1/documents/%s"  % (word[1])

        params = {"type": DocType}

        #先ほどと同じ接続方法に加えて、取得ファイルが大きいときのために
        #stream=Trueにすることでメモリに優しい接続方法にしておく
        res = requests.get(url, params=params, verify=False, stream=True)

        if res.ok:

            if ZipFlg == 0 or DocType == "2":

                #出力先ファイル
                if DocType == 1:
                    saveFileName = str(word[2]) + "_" + str(word[5]) + "_" + str(word[13]) + ".zip"
                elif DocType == 2:
                    saveFileName = str(word[2]) + "_" + str(word[5]) + "_" + str(word[13]) + ".pdf"
                elif DocType == 3:
                    saveFileName = str(word[2]) + "_" + str(word[5]) + "_" + str(word[13]) + ".zip"
                elif DocType == 4:
                    saveFileName = str(word[2]) + "_" + str(word[5]) + "_" + str(word[13]) + ".zip"

                saveFilePath = os.path.join(DOWNLOAD_SAVE_DIR, saveFileName)

                with open(saveFilePath, 'wb') as f:
                    #念の為、小まめに書き込み、メモリに優しくしておく
                    for chunk in res.iter_content(chunk_size=1024):
                        f.write(chunk)
            else:
                #フォルダパス
                saveFolderPath = os.path.join(DOWNLOAD_SAVE_DIR, str(word[2]) + "_" + str(word[5]) + "_" + str(word[13]))
                #フォルダ削除
                if os.path.isdir(saveFolderPath):
                    shutil.rmtree(saveFolderPath)
                #ファイルを解凍
                f = ZipFile(BytesIO(res.content))
                f.extractall(saveFolderPath)
                f.close()

def main():

        #ドキュメントリストを作成
        saveFilePath = getDocumentlist()

        #ファイルを開く
        file = open(saveFilePath,"r")
        #ヘッダーを読まない
        next(file)
        # 元ファイルのレコード部分をreadlinesメソッドで全行を読み取る
        lines = file.readlines()

        # for文で1行ずつ取得
        for line in lines:
            #カンマで区切る
            word = line.split(',')

            #ファイルのチェック
            flg = CheckFlg(word)

            if flg == 1:
                #ファイルのダウンロード
                download_xbrl_file(word,flg)


if __name__=='__main__':
    try:
        # 実行時間を測るStart
        start = time.time()

        # プロキシの設定
        Con = Conn.Connect()
        os.environ["http_proxy"] = Con.http_proxy()
        os.environ["https_proxy"] = Con.https_proxy()

        #xbrlを取得するメイン処理
        main()
    except Exception:
        e = sys.exc_info()[1]
        print("エラー",e.args)

    finally:
        # 実行時間を測るEnd
        elapsed_time = time.time() - start
        print ("処理時間:{0}".format(elapsed_time) + "[sec]")

 







お・す・す・め・!

1

目次1 マコトのプロフィール1.1 自己紹介1.2 タイトルの由来1.3 サブタイトルの由来1.4 趣味1.5 ほしいも ...

-Python, プログラム

Copyright© 心ゆく日並 , 2020 AllRights Reserved.