PythonでEDINET・TDNETから有価証券報告書・四半期報告書・決算短信のXBRLをダウンロードする

タイトルがクソ長い．．
金融界隈で定量的な分析やデータサイエンスをやっている9uantです．
twitterもやってるので，興味ある方はぜひフォローしていただけると！

タイトルの通り，決算書類のXBRLを手早くダウンロードするためのコードを共有する．
解説も追々書いていきたい．
以下の2ステップをとる．

• XBRLへのリンクをDataFrame化する
• DataFrameからXBRLのzipファイルをダウンロードする

import os
import glob
import shutil
import re
import time
from datetime import date, timedelta, datetime
from dateutil.relativedelta import relativedelta

import requests
from bs4 import BeautifulSoup
import urllib3
from urllib3.exceptions import InsecureRequestWarning
urllib3.disable_warnings(InsecureRequestWarning)
from selenium import webdriver
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.ui import Select
from selenium.webdriver.common.by import By
import zipfile

import numpy as np
import pandas as pd
import json

XBRLへのリンクをDataFrame化する

EDINET

EDINETにはAPIが存在するため容易．

def edinet_xbrl_link(annual=True, quarter=True, codes=None, year=0,month=0,day=0):
    '''
    特定の企業の，もしくは全ての有価証券報告書・四半期報告書のXBRLのリンクのDataFrameを作成する
    
    Parameters:
        annual: bool, default True
            True の場合に有価証券報告書を取得する
        quarter: bool, default Ture
            True の場合に四半期報告書を取得する
        codes: None, int, float, str, or list (codes[code]=int, float,or str), default None
            None の場合に全ての企業のデータを取得する
            銘柄コードを指定すると，それらの企業のデータのみを取得する:point_up_tone4:
        year, month, day: int, default 0
            現在から何日前までのデータを取得するかを指定する(最大5年)
            
    Returns:
        database: pandas.DataFrame
            database['code']: str
                5桁の証券コード
            database['type']: str
                'annual' or 'quarter'
            database['date']: datetime.date
                公開日
            database['title']: str
                表題
            database['URL']: str
                XBRLのzipファイルをダウンロードするURL
    '''
    edinet_url = "https://disclosure.edinet-fsa.go.jp/api/v1/documents.json"
    
    # codesを文字型の配列に統一する．
    if codes != None:
        if type(codes) in (str, int, float):
            codes = [int(codes)]
        for code in codes:
            # 4桁の証券コードを5桁に変換
            if len(str(int(code)))==4:
                code = str(int(code))+'0'

    # datetime型でfor文を回す
    def date_range(start, stop, step = timedelta(1)):
        current = start
        while current < stop:
            yield current
            current += step
            
    # 結果を格納するDataFrameを用意
    database = pd.DataFrame(index=[], columns=['code','type','date','title','URL'])
    
    for d in date_range(date.today()-relativedelta(years=year, months=month, days=day)+relativedelta(days=1), date.today()+relativedelta(days=1)):
        # EDINET API にアクセス
        d_str = d.strftime('%Y-%m-%d')
        params = {'date' : d_str, 'type' : 2}
        res = requests.get(edinet_url, params=params, verify=False)
        json_res = json.loads(res.text)
        time.sleep(5)
        
        # 正常にアクセスできない場合
        if json_res['metadata']['status']!='200':
            print(d_str, 'not accessible')
            continue
            
        print(d_str, json_res['metadata']['resultset']['count'])# 日付と件数を表示
        
        # 0件の場合
        if len(json_res['results'])==0:
            continue
        
        df = pd.DataFrame(json_res['results'])[['docID', 'secCode', 'ordinanceCode', 'formCode','docDescription']]
        df.dropna(subset=['docID'], inplace=True)
        df.dropna(subset=['secCode'], inplace=True)
        df.rename(columns={'secCode': 'code', 'docDescription': 'title'}, inplace=True)
        df['date'] = d
        df['URL'] = df['docID']
        df['URL'] = "https://disclosure.edinet-fsa.go.jp/api/v1/documents/" + df['URL']
        
        # 指定された証券コードのみを抽出
        if codes != None:
            df = df[df['code'] in codes]
            
        if annual == True:
            df1 = df[(df['ordinanceCode']=='010') & (df['formCode']=='030000')]
            df1['type'] = 'annual'
            database = pd.concat([database, df1[['code', 'type', 'date','title', 'URL']]], axis=0, join='outer').reset_index(drop=True)
            
        if quarter == True:
            df2 = df[(df['ordinanceCode']=='010') & (df['formCode']=='043000')]
            df2['type'] = 'quarter'
            database = pd.concat([database, df2[['code', 'type', 'date','title', 'URL']]], axis=0, join='outer').reset_index(drop=True)
            
    return database

TDNET

TDNETからのデータの収集にはseleniumを用いる．
フリーワード検索結果が200件までしか表示されないため，証券コードから検索する関数と，日付から検索する関数を別々に作成した．

def tdnet_xbrl_link_by_code(codes):
    '''
    指定された企業の決算短信をXBRLへのリンクのDataFrameを作成する
    
    Parameters:
        codes: None, int, float, str, or list (codes[code]=int, float,or str), default None
            None の場合に全ての企業のデータを取得する
            
    Returns:
        database: pandas.DataFrame
            database['code']: str
                5桁の証券コード
            database['type']: str
                'annual' or 'quarter'
            database['date']: datetime.date
                公開日
            database['title']: str
                表題
            database['URL']: str
                XBRLのzipファイルをダウンロードするURL
    '''

    # codesを文字型の配列に統一する．
    if type(codes) in (str, int, float):
        codes = [int(codes)]
    for i, code in enumerate(codes):
        # 4桁の証券コードを5桁に変換
        if len(str(int(code)))==4:
            codes[i] = str(int(code))+'0'
    
    database = pd.DataFrame(index=[], columns=['code','type','date','title','URL'])
    
    for code in codes:
        # ブラウザを起動する
        chromeOptions = webdriver.ChromeOptions()
        chromeOptions.add_argument('--headless') # ブラウザ非表示
        driver = webdriver.Chrome(options=chromeOptions)
        driver.get("https://www.release.tdnet.info/onsf/TDJFSearch/I_head")

        # 検索ワードを送る
        duration = driver.find_element_by_name('t0')
        select = Select(duration)
        select.options[-1].click()
        inputElement = driver.find_element_by_id("freewordtxt")
        inputElement.send_keys(code)
        inputElement.send_keys(Keys.RETURN)
        time.sleep(5)

        # 検索結果が表示されたフレームに移動
        iframe = driver.find_element_by_name("mainlist")
        driver.switch_to.frame(iframe)
        
        # 検索結果が0件の場合に処理を終える
        if driver.find_element_by_id("contentwrapper").text == '該当する適時開示情報が見つかりませんでした。':
            return database

        # 検索結果の表の各行からデータを読み取る
        table = driver.find_element_by_id("maintable")
        trs = table.find_elements(By.TAG_NAME, "tr")
        for i in range(len(trs)):
            title = trs[i].find_elements(By.TAG_NAME, "td")[3].text

            # 訂正書類でなく，XBRLが存在する，指定された企業の決算短信を選択
            if ('決算短信' in title) and ('訂正' not in title) and (len(trs[i].find_elements(By.TAG_NAME, "td")[4].text)!=0) and (code==trs[i].find_elements(By.TAG_NAME, "td")[1].text):
                date = trs[i].find_elements(By.TAG_NAME, "td")[0].text[:10]
                date = datetime.strptime(date, '%Y/%m/%d').date()
                url = trs[i].find_elements(By.TAG_NAME, "td")[4].find_element_by_tag_name("a").get_attribute("href")
                database = database.append(pd.Series([code,'brief',date,title,url], index=database.columns), ignore_index=True)

        driver.quit()

        return database

def tdnet_xbrl_link_by_date(date=None):
    '''
    指定された日付，もしくは全ての決算短信をXBRLへのリンクのDataFrameを作成する
    
    Parameters:
        date: None or str ('yyyy/mm/dd'), default None
            None の場合に全ての日付のデータを取得する
            
    Returns:
        database: pandas.DataFrame
            database['code']: str
                5桁の証券コード
            database['type']: str
                'annual' or 'quarter'
            database['date']: datetime.date
                公開日
            database['title']: str
                表題
            database['URL']: str
                XBRLのzipファイルをダウンロードするURL
    '''
    database = pd.DataFrame(index=[], columns=['code','type','date','title','URL'])
    
    # ブラウザを起動する
    chromeOptions = webdriver.ChromeOptions()
    chromeOptions.add_argument('--headless') # ブラウザ非表示
    driver = webdriver.Chrome(options=chromeOptions)
    driver.get("https://www.release.tdnet.info/inbs/I_main_00.html")
    
    duration = driver.find_element_by_name('daylist')
    select = Select(duration)
    for i in range(1, len(select.options)):
        driver.get("https://www.release.tdnet.info/inbs/I_main_00.html")
        duration = driver.find_element_by_name('daylist')
        select = Select(duration)
        d = datetime.strptime(select.options[i].text[:10], '%Y/%m/%d').date()
        print(select.options[i].text)
        
        if (date == None) or (date == select.options[i].text[:10]):
            select.options[i].click()
            time.sleep(5)

            # 検索結果が表示されたフレームに移動
            iframe = driver.find_element_by_id("main_list")
            driver.switch_to.frame(iframe)

            # 検索結果が0件の場合に処理を終える
            if driver.find_element_by_id("kaiji-text-1").text!='に開示された情報':
                continue

            # 最後のページまで処理を続ける
            while True:
                # 検索結果の表の各行からデータを読み取る
                table = driver.find_element_by_id("main-list-table")
                trs = table.find_elements(By.TAG_NAME, "tr")
                for i in range(len(trs)):
                    title = trs[i].find_elements(By.TAG_NAME, "td")[3].text

                    # 訂正書類でなく，XBRLが存在する，指定された企業の決算短信を選択
                    if ('決算短信' in title) and ('訂正' not in title) and (len(trs[i].find_elements(By.TAG_NAME, "td")[4].text)!=0):
                        code = trs[i].find_elements(By.TAG_NAME, "td")[1].text
                        url = trs[i].find_elements(By.TAG_NAME, "td")[4].find_element_by_tag_name("a").get_attribute("href")
                        database = database.append(pd.Series([code, 'brief', d, title,url], index=database.columns), ignore_index=True)

                if len(driver.find_element_by_class_name("pager-R").text)!=0:
                    driver.find_element_by_class_name("pager-R").click()
                    time.sleep(5)
                else:
                    # 「次へ」の文字が存在しない場合に処理を終了する
                    break
    driver.quit()
    return database

DataFrameからXBRLをダウンロードする

def dl_xbrl_zip(codes=None, database):
    '''
    XBRLへのリンクをリスト化したDataFrameを参照して，XBRLのzipファイルをダウンロードする
    
    Parameters:
        codes: None, int, float, str, or list (codes[code]=int, float,or str), default None
            None の場合に全ての企業のXBRLを取得する
            
        database: pandas.DataFrame
            database['code']: str
                5桁の証券コード
            database['type']: str
                'annual' or 'quarter'
            database['date']: datetime.date
                公開日
            database['title']: str
                表題
            database['URL']: str
                XBRLのzipファイルをダウンロードするURL
            
    Returns:
        None
    '''
    database.dropna(subset=['code'], inplace=True)
    database = database.reset_index(drop=True)
    
    # codesを文字型の配列に統一する
    if codes == None:
        codes = [None]
    else:
        if type(codes) in (str, int, float):
            codes = [int(codes)]
        for i, code in enumerate(codes):
            # 4桁の証券コードを5桁に変換
            if len(str(int(code)))==4:
                codes[i] = str(int(code))+'0'
            
    for code in codes:
        if code == None:
            df_company = database
        else:
            df_company = database[database['code']==code]
            df_company = df_company.reset_index(drop=True)
        
        # 証券コードをディレクトリ名とする
        dir_path = database.loc[i,'code']
        if os.path.exists(dir_path)==False:
            os.mkdir(dir_path)
            
        # 抽出したリストの各行からXBRLへのリンクからzipファイルをダウンロード
        for i in range(df_company.shape[0]):
            # EDINETへアクセスする場合
            if (df_company.loc[i,'type'] == 'annual') or (df_company.loc[i,'type'] == 'quarter'):
                params = {"type": 1}
                res = requests.get(df_company.loc[i,'URL'], params=params, stream=True)
                
                if df_company.loc[i,'type'] == 'annual':
                    # 有価証券報告書のファイル名は"yyyy_0.zip"
                    filename = dir_path + r'/' + df_company.loc[i,'date'][:4] + r"_0.zip"
                elif df_company.loc[i,'type'] == 'quarter':
                    if re.search('期第', df_company.loc[i,'title']) == None:
                        # 第何期か不明の四半期報告書のファイル名は"yyyy_unknown_docID.zip"
                        filename = dir_path + r'/' + df_company.loc[i,'date'][:4] + r'_unknown_' + df_company.loc[i,'URL'][-8:] + r'.zip'
                    else:
                        # 四半期報告書のファイル名は"yyyy_quarter.zip"
                        filename = dir_path + r'/' + df_company.loc[i,'date'][:4] + r'_' + df_company.loc[i,'title'][re.search('期第', df_company.loc[i,'title']).end()] + r'.zip'
                            
            # TDNETへアクセスする場合
            elif df_company.loc[i,'type'] == 'brief':
                res = requests.get(df_company.loc[i,'URL'], stream=True)
                
                # 空白文字を埋める
                s_list = df_company.loc[i,'title'].split()
                s = ''
                for i in s_list:
                    s += i
                
                filename = df_company.loc[i,'date'][:4] + r'_' + s[re.search('期第', s).end()] + r'_brief.zip'
            
            # 同名のzipファイルが存在する場合，上書きはしない
            if os.path.exists(filename):
                print(df_company.loc[i,'code'],df_company.loc[i,'date'],'already exists')
                continue
                
            # 正常にアクセスできた場合のみzipファイルをダウンロード
            if res.status_code == 200:
                with open(filename, 'wb') as file:
                    for chunk in res.iter_content(chunk_size=1024):
                        file.write(chunk)
                    print(df_company.loc[i,'code'],df_company.loc[i,'date'],'saved')
                            
    print('done!')
    return None