ども♪マコトです。
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]")
コメント