Google Colab：EDINET 企業データの自動取得･スクリーニング

1. EDINET と XRBL とは
2. EDINET から XBRLファイルを取得して、データを CSV にまとめる
   1. EDINETコードと証券コードを読み込む
   2. 有価証券報告書の情報取得（1st step）
   3. 期間を指定してコードを実行する
   4. XBRLファイルのダウンロード（2nd step ①）
   5. XBRLファイルのデータを CSV にまとめる（2nd step ②）
   6. 期間を指定してコードを実行する
3. CSV から必要なデータを抽出してスクリーニングする
   1. EDINETコードと証券コードを読み込む
   2. 対象のファイルを指定
   3. ファイルを Google Colab にアップロード
   4. 必要なデータを抽出
   5. 抽出したデータを表示して確認
   6. PER、ROE、一株配当などでスクリーニング
4. ファイルのダウンロードがうまくいかない場合

１．EDINET と XRBL とは

前提知識のメモ。

上場企業の財務諸表が EDINET にあり、XBRLファイルで保存されている。掲載されているのは決算時点のデータ。

目次へ

２．EDINET から XBRLファイルを取得して、データを CSV にまとめる

この記事のコードを、Google Colab で上から順番に実行していく。

まずは、Google Colab に入っていないモジュール（python-xbrl）をインストール。

Google Colab: Cell-1

!pip install python-xbrl

次に、必要なモジュールを読み込む。

Google Colab: Cell-2

import os ,re ,csv ,json ,io ,time ,random
import codecs ,requests
import shutil ,multiprocessing

import pandas as pd
import xml.etree.ElementTree as ET

from tqdm import tqdm
from collections import defaultdict
from datetime import datetime ,timedelta
from dateutil.relativedelta import relativedelta
from zipfile import ZipFile
from xbrl import XBRLParser
from google.colab import files

目次へ

２－１．EDINETコードと証券コードを読み込む

上場している企業のデータだけ取得するために、EDINETのコードと証券コードを付け合わせる必要がある。EDINETのページで公開されている「EdinetcodeDlInfo.csv」というファイルを用いる。これは自動取得がむずかしいため、手動でダウンロードする必要がある。

EdinetcodeDlInfo.csv を入手

EDINETにアクセス。

トップページ下部の「EDINETタクソノミ及びコードリスト」をクリック。

リンク先のページ下部の「EDINETコードリスト」をクリックしてダウンロード。

Google Colab に EdinetcodeDlInfo.csv をアップロード

取得した「EdinetcodeDlInfo.csv」をサーバーやDropboxなどのクラウドサービスにアップロードして、以下のコードで Colab にダウンロード（アップロード）。

Google Colab: Cell-3

!wget https://●●●●/●●●/EdinetcodeDlInfo.csv

もしくは、Google Colab に直接アップロードすることもできる。その場合は上記のコードは不要。

EdinetcodeDlInfo.csv を読み込み

Google Colab: Cell-4

file_name = 'EdinetcodeDlInfo.csv'

DIR = os.getcwd()
file_path = os.path.join(DIR ,file_name)

with codecs.open(file_path, "r", "Shift-JIS", "ignore") as file:
      df_codes = pd.read_table( file ,sep=","
                              , skiprows=[0]
                              , index_col=['ＥＤＩＮＥＴコード']
                              , usecols=['ＥＤＩＮＥＴコード','提出者名','証券コード','提出者業種'] )
        
# ここに掲載されている証券コードは「7203」→「72030」のようになっている

目次へ

２－２．有価証券報告書の情報取得（1st step）

Google Colab: Cell-5

csv_tag = [ 'id' ,'title' ,'cd' ,'url' ,'update' ,'period' ]
encode_type = 'utf-8'
wait_time = 5

def get_link_info_str( ticker_symbol ,base_url ):
    url = base_url+ticker_symbol
    response = requests.get( url )
    return response.text

def get_link( tree ,namespace ,since ,until ):
    key ,yuho_dict = '' ,{}
    
    #entryタグ毎にforeach
    for el in tree.findall('.//'+namespace+'entry'):
      
        #titleタグに有価証券の文字があれば、後続処理を実行
        title = el.find( namespace+'title' ).text
        if not is_yuho(title): continue
          
        #日付の条件に合うデータを取得
        updated = el.find( namespace+'updated' ).text
        checked = time_check( updated ,since ,until )
        if not checked[ 'until' ] : continue
        if not checked[ 'since' ] : return yuho_dict
        
        # zipファイルのアドレスを辞書オブジェクトへ登録
        _id = el.find( namespace+'id' ).text
        link = el.find( './'+namespace+'link[@type="application/zip"]' )
        url = link.attrib[ 'href' ]
        cd = re.sub( r'^【(\w+)】.*',r"\1" ,title )
        period = re.sub( r'.*第(\w+)期.*',r"\1" ,title )
        
        #最新の報告書だけを登録する
        if len( yuho_dict )==0 :
            yuho_dict[_id] = { 'id':_id ,'title':title ,'cd':cd ,'url':url ,'update':updated ,'period':period }
            key = _id
        elif period_check( period ,yuho_dict[key]['period'] ,title ) :
            yuho_dict = {}
            yuho_dict[_id] = { 'id':_id ,'title':title ,'cd':cd ,'url':url ,'update':updated ,'period':period }
            key = _id
        
    return yuho_dict

def is_yuho( title ):
    if  u'有価証券報告書' in str(title)\
    and u'株式会社' in str(title)\
    and u'内国投資信託受益証券' not in str(title):
        return True
    else:
        return False

def is_stock( code ):
    if not pd.isnull( df_codes['証券コード'][code] ) :
        return True
    else:
        return False

def time_check( update ,since ,until ):
    updated_time = datetime.strptime( update ,'%Y-%m-%dT%H:%M:%S+09:00' )
    return { 'since':updated_time >= since ,'until':updated_time < until }
  
def period_check( period ,b4period ,title ):
    if period > b4period :
        return True
    elif period==b4period and u'訂正有価証券報告書' in str(title) :
        return True
    else :
        return False

def dump_file( file ,info_dict ,tag ,encode_type ) :
    with open( file ,'w' ,encoding=encode_type ) as of :
        writer = csv.DictWriter( of ,tag ,lineterminator='\n' )
        writer.writeheader()
        for key in info_dict :
            writer.writerow( info_dict[key] )

def craete_xbrl_url_json( since ,until ,str_period ):
    #有報キャッチャーWebServiceのAtomAPIアドレス<https://resource.ufocatch.com/>
    base_url = 'http://resource.ufocatch.com/atom/edinetx/query/'
    namespace = '{http://www.w3.org/2005/Atom}'

    #証券コードを持つものだけを対象にする
    target_df_codes = df_codes[df_codes['証券コード'].isnull()==False].index
    i ,length ,result_dict = 0 ,len( target_df_codes ) ,{}

    for t_symbol in tqdm( target_df_codes ) :
        
        count ,retry = 0 ,3
        while True:
            try : 
                #企業毎の有報へのデータへのリンク情報を取得
                response_string = get_link_info_str( t_symbol ,base_url )
                
                #xmlをparseするElementTreeを作成
                ET_tree = ET.fromstring( response_string )
                ET.register_namespace( '' ,namespace[1:-1] )
                break
                
            except Exception :
                print( f'{t_symbol}のperseに失敗しました。[{count}] \n {response_string}' )
                if count < retry : 
                    time.sleep( wait_time * random.random() )
                    count += 1
                    continue
                else : raise
        
        #downloadファイルの対象を取得
        info_dict = get_link( ET_tree ,namespace ,since ,until )
        
        #Request用のJson形式のファイルを作成
        if len(info_dict) > 0 : 
            #print( 'symbol('+str(i+1)+'/'+str(length)+'):'+t_symbol+', wrote csv...' )
            for key in info_dict : 
                result_dict[key] = info_dict[key]
            dump_file( 'dat_download' + str_period + '.csv' ,result_dict ,csv_tag ,encode_type )
            
        else : print( 'symbol('+str(i+1)+'/'+str(length)+'):'+t_symbol+', is not yuho or out of date range.' )
        i += 1
        
    print( 'complete a download!!' )
    files.download( 'dat_download' + str_period + '.csv' )

以下の記事を参考にした。

目次へ

２－３．期間を指定してコードを実行する

ここまでで作成したコードを、ここでまとめて実行する。処理が完了すると、自動的に CSVファイルがブラウザ経由でダウンロードされる。ファイル名はstr_period = since.strftime('_%y%m%d_')+until.strftime('%y%m%d')によって決まり、'dat_download'+str_period+'.csv'の形になる。

Google Colab を閉じずに「２－４．XBRLファイルのダウンロード」以降の処理を続ける場合はそのままで良いが、一旦、閉じる必要がある場合は、ダウンロードしたファイルを Google Colab にアップロードすることで、繰り返し同じ処理を行う必要がなくなる。

Google Colab: Cell-6

#df_codes = df_codes.loc["E02824":,]

if __name__=='__main__':
    since = datetime.strptime('2016-01-01' ,'%Y-%m-%d')
    until = datetime.strptime('2017-02-01' ,'%Y-%m-%d')
    str_period = since.strftime( '_%y%m%d_' )+until.strftime( '%y%m%d' )
    print( f'since:{ since } ,until:{ until } ({str_period})' )

    #一定期間に発生したデータ全てのURLを取得する
    craete_xbrl_url_json( since ,until ,str_period )

ここでは、「最終的にほしい期間」のデータをすべて取得してしまった方が良い。

目次へ

２－４．XBRLファイルのダウンロード（2nd step ①）

Google Colab: Cell-6

proc = 4    # TODO:マルチプロセス制御（初期値：1並列）
wait_time = 10

def make_directory( dir_name ) :
    if not os.path.exists( dir_name ) :
        os.mkdir( dir_name )

def download_all_xbrl_files( info_df ,str_period ) :
    # xbrl_filesのディレクトリが存在しないとき、作成する
    directory_path = os.getcwd()+'/xbrl_files'+str_period+'/'
    make_directory( directory_path )
    counter ,mp_dict = 0 ,{}

    #分散処理用にデータを設定
    for index ,row in info_df.iterrows() :
        mp_dict[ counter ] = row.to_dict()
        counter += 1

    #is_yuho関数に当てはまる全ての企業×有報情報IDごとに取得する
    p ,jobs = 0 ,[]
    while( p < proc ) :
        job = multiprocessing.Process( target=_download_xbrl_file ,args=( mp_dict ,directory_path ,p ,) )
        jobs.append( job )
        job.start()
        p += 1

    [ job.join() for job in jobs ]

def _download_xbrl_file( info_dicts ,directory_path ,p ) :
    no = p
    
    while( no < len(info_dicts) ) :
        info_dict = info_dicts[ no ]
        no += proc

        #証券CDごとのディレクトリ作成
        company_path = f'{ directory_path + info_dict["cd"] }/'
        make_directory( company_path )

        #証券コード×IR情報ごとのディレクトリ作成
        ir_path = company_path + info_dict['id']
        if os.path.exists( ir_path ) :
            continue
        make_directory( ir_path )
        
        #Requestによりファイルを取得して、Unzipする
        count ,retry = 0 ,3
        while True:
            r = requests.get( info_dict['url'] )
            if r.ok :
                z = ZipFile( io.BytesIO( r.content ) )
                z.extractall( ir_path ) # unzip the file and save files to path.
                print( f'Process({ str(p+1) }):downloaded:{ info_dict["update"] }({ info_dict["period"] })_{ info_dict["title"] }' )
                break
            else :
                print( f'Process({ str(p+1) }):download failed[{ count }]:{ info_dict["update"] }({ info_dict["period"] })_{ info_dict["title"] }' )
                if count < retry : 
                    count += 1
                    time.sleep( wait_time * random.random() )
                    continue
                else : raise

def download_csv_url( since ,until ,big_period ,small_period ) :
    search_directory = f'{ os.getcwd() }/'
    
    info_df = pd.read_csv( f'{ search_directory }dat_download{ big_period }.csv' ,parse_dates=['update'])
    info_df = info_df.loc[( info_df['update'] >= since ) & ( info_df['update'] < until )]
    
    if len(info_df) > 0 :
        download_all_xbrl_files( info_df ,small_period )
        
    print( 'complete a download!!' )

目次へ

２－５．XBRLファイルのデータを CSV にまとめる（2nd step ②）

Google Colab: Cell-7

default_tag = ['file_nm','element_id','amount']
custom_tag  = ['unit_ref','decimals','contextref']
encode_type = 'utf-8'

class XbrlParser(XBRLParser):
    def __init__(self, xbrl_filepath):
        self.xbrl_filepath = xbrl_filepath

    def parse_xbrl(self):
        # parse xbrl file
        with open(self.xbrl_filepath,'r',encoding='utf-8') as of:
            xbrl = XBRLParser.parse(of) # beautiful soup type object
        result_dicts = defaultdict(list)
        _idx = 0

        #print xbrl
        name_space = 'jp*'
        for node in xbrl.find_all(name=re.compile(name_space+':*')):
            if self.ignore_pattern(node):
                continue

            row_dict = defaultdict(list)
            #default tag
            row_dict['file_nm'] = self.xbrl_filepath.rsplit(os.sep, 1)[1]
            row_dict['element_id'] = node.name
            row_dict['amount'] = node.string

            #cutom tag
            for tag in custom_tag:
                row_dict[tag] = self.get_attrib_value(node,tag)

            result_dicts[_idx] = row_dict
            _idx += 1
        return result_dicts

    def ignore_pattern(self, node):
        if 'xsi:nil' in node.attrs:
            if node.attrs['xsi:nil']=='true':
                return True
        #結果が空の場合は対象外にする
        if not isinstance(node.string, str):
            return True
        # 改行コードが結果にある場合対象外にする
        if str(node.string).find(u'\n') > -1:
            return True
        #text文字列は解析に利用できないため、対象外
        if u'textblock' in str(node.name):
            return True
        return False

    def get_attrib_value(self, node, attrib):
        if attrib in node.attrs.keys():
            return node.attrs[attrib]
        else:
            return None

#ディレクトリ配下のファイルのうちis_xbrl_fileがTrueのもののみ取得する
def fild_all_files(directory ,str_period):
    for root, dirs, files in os.walk(directory):
        for file in files:
            if not is_xbrl_file(root ,file ,str_period):
                continue
            yield os.path.join(root ,file)

def is_xbrl_file(root_path ,file_name ,str_period):
    #xbrlファイルでなければ対象外
    if not file_name.endswith('.xbrl'):
        return False
    #AuditDocは対象外
    if u'AuditDoc' in str(root_path):
        return False
    if 'xbrl_files'+str_period in str(root_path):
        return True

def dump_file(writer ,info_dicts):
    _idx = 0
    while _idx < len(info_dicts):
        row_dict = info_dicts[_idx]
        writer.writerow(row_dict)
        _idx += 1

def xbrl_to_csv(str_period):
    base_path = os.getcwd()+'/xbrl_files'+str_period

    with open( 'eggs'+str_period+'.csv' ,'w' ,encoding=encode_type ) as of :
        resultCsvWriter = csv.DictWriter( of ,default_tag + custom_tag ,lineterminator='\n' )
        resultCsvWriter.writeheader()

        #対象ファイルを抽出
        xbrl_files = fild_all_files(base_path ,str_period)
        count = 0
        for xbrl_file in tqdm( xbrl_files ) :
            count += 1
            print( f'getting data...{xbrl_file}' )
            xp = XbrlParser( xbrl_file )
            info_dicts = xp.parse_xbrl()
            dump_file( resultCsvWriter ,info_dicts )
    
    print( 'complete a conversion!!' )
    files.download( 'eggs'+str_period+'.csv' )

以下の記事を参考にした。

目次へ

２－６．期間を指定してコードを実行する

2nd step のコードをまとめて実行する。ここでも、処理が完了すると CSVファイルがブラウザ経由でダウンロードされる。ファイル名はsmall_period = since.strftime('_%y%m%d_')+until.strftime('%y%m%d')によって決まり、'eggs'+str_period+'.csv'の形になる。

ダウンロードしたファイルを Google Colab にアップロードすることで、繰り返し同じ処理を行う必要がなくなる。

Google Colab: Cell-8

if __name__=='__main__' :
    start = datetime.strptime('2016-01-01' ,'%Y-%m-%d')
    end   = datetime.strptime('2017-02-01' ,'%Y-%m-%d')
    big_period = start.strftime('_%y%m%d_')+end.strftime('%y%m%d')

    month = 3
    since = start
    until = start + relativedelta( months=month )

    while True :
        small_period = since.strftime('_%y%m%d_')+until.strftime('%y%m%d')
        print( f'since:{ since } ,until:{ until } ({ small_period })' )

        download_csv_url( since ,until ,big_period ,small_period )
        xbrl_to_csv( small_period )

        since = until
        until = until + relativedelta( months=month )
        if since >= end : break

１年分の全上場企業のデータを取得しようとすると丸１日かかる。Google Colab側の制限があり効率が悪いため、指定した期間の、指定した間隔ごとに処理できるようにしている。

例えば、start = datetime.strptime('2018-01-01' ,'%Y-%m-%d')、end = datetime.strptime('2019-01-01' ,'%Y-%m-%d')、month = 3とすると、2018年の１年間を３ヶ月ごと処理する。つまり、出力されるファイルが４つできる。これで、途中で処理がとまるようなことがあっても、完了していないところから改めて処理を再開すれば良い。

目次へ

３．CSV から必要なデータを抽出してスクリーニングする

「２．EDINET から XBRLファイルを取得して、データを CSV にまとめる」の処理がもっとも時間を必要とする。後日あらためて処理を再開できるように、以下のコードは「ここから実行していけば事足りる」ようにしている。

あらためて、必要なモジュールを読み込むところからはじめる。

Google Colab: Cell-9

import os ,re ,codecs
import urllib.request
import pandas as pd
from datetime import datetime as dt
from tqdm import tqdm
from google.colab import files

目次へ

３－１．EDINETコードと証券コードを読み込む

ここは、「２－１．EDINETコードと証券コードを読み込む」とまったく同じ工程。詳しくは「２－１」を参照。

Google Colab: Cell-10

!wget https://●●●●/●●●/EdinetcodeDlInfo.csv

Google Colab: Cell-11

file_name = 'EdinetcodeDlInfo.csv'

DIR = os.getcwd()
file_path = os.path.join(DIR ,file_name)

with codecs.open(file_path, "r", "Shift-JIS", "ignore") as file:
      df_codes = pd.read_table( file ,sep=","
                              , skiprows=[0]
                              , index_col=['ＥＤＩＮＥＴコード']
                              , usecols=['ＥＤＩＮＥＴコード','提出者名','証券コード','提出者業種'] )
        
# ここに掲載されている証券コードは「7203」→「72030」のようになっている

目次へ

３－２．対象のファイルを指定

「２．EDINET から XBRLファイルを取得して、データを CSV にまとめる」で作成した CSV のファイル名と、処理完了時に新たに作成されるファイルの名称を指定する。以下の例では、参考にeggs_190101_190401.csvも入れているが、不要な場合は削除する。

Google Colab: Cell-12

fileNames = [ 'eggs_181001_190101.csv'
            , 'eggs_190101_190401.csv' ]
resultFileName = 'com_indices_190101.csv'

目次へ

３－３．ファイルを Google Colab にアップロード

どこか特定のサーバーなどにファイルを置いた場合は、以下のコードでダウンロードする。手動で直接ファイルをアップロードすることもできる。

Google Colab: Cell-13

def download_files(fileNames) :
    for fileName in fileNames :
        url = f'https://●●●●/●●●/{fileName}'
        try : 
            urllib.request.urlretrieve(url, fileName)
            print(f'{ url }をダウンロードしました。')
        except Exception :
            print(f'{ url }のダウンロードに失敗しました。')
            pass 
    
if __name__=='__main__' :
    download_files(fileNames)

目次へ

３－４．必要なデータを抽出

• 「３－２．対象のファイルを指定」で設定した CSV が対象
• 対象のファイルの中で、最新のデータだけが抽出される
• dict_cols = {}で指定している項目を抽出

Google Colab: Cell-14

#取得したい情報（複数あるものは上のものほど優先度が高い）
dict_cols = {
      '会社名'           : { 'element_id':['companynamecoverpage'] }
    , '提出日'           : { 'element_id':['filingdatecoverpage'] }
    , '当年度開始日'     : { 'element_id':['currentfiscalyearstartdatedei'] }
    , '発行済み株式数'   : { 'element_id':['totalnumberofissuedsharessummaryofbusinessresults'
                                          ,'totalnumberofissuedsharescommonstocksummaryofbusinessresults'] }
    , 'BPS'              : { 'element_id':['equityattributabletoownersofparentpershareusgaapsummaryofbusinessresults'
                                          ,'equitytoassetratioifrssummaryofbusinessresults'
                                          ,'netassetspersharesummaryofbusinessresults'] }
    , 'EPS'              : { 'element_id':['basicearningslosspershareusgaapsummaryofbusinessresults'
                                          ,'basicearningslosspershareifrssummaryofbusinessresults'
                                          ,'basicearningslosspersharesummaryofbusinessresults'] }
    , 'PER'              : { 'element_id':['priceearningsratiousgaapsummaryofbusinessresults'
                                          ,'priceearningsratioifrssummaryofbusinessresults'
                                          ,'priceearningsratiosummaryofbusinessresults'] }
    , '一株当たりの配当' : { 'element_id':['dividendpaidpersharesummaryofbusinessresults'
                                          ,'dividendpaidpersharecommonstocksummaryofbusinessresults'] } }



def check_date(new_dates ,b4_dates) :
    dt_new1 = dt.strptime( new_dates[0] ,'%Y-%m-%d' )
    dt_new2 = dt.strptime( new_dates[1] ,'%Y-%m-%d' )
    dt_b4_1 = dt.strptime( b4_dates[0] ,'%Y-%m-%d' )
    dt_b4_2 = dt.strptime( b4_dates[1] ,'%Y-%m-%d' )
    return dt_new1 >= dt_b4_1 and dt_new2 > dt_b4_2
  
def check_same_code(code ,ar_code) :
    if len(ar_code) > 0 :
        if code in ar_code : 
            return True
    else : 
        return False

def check_element(data ,col) :
    element_ids = dict_cols[col]['element_id']
    if col=='会社名' or col=='当年度開始日' or col=='提出日' :
        check1 = data[ data['element_id'].str.contains(element_ids[0]) ]
        if len(check1)==1 : 
            return check1['amount'].values[0]
        else : 
            return 0
    else :
        #当年度、連結
        for element_id in element_ids :
            check1 = data[ data['element_id'].str.contains(element_id) ]
            check2 = check1[ (check1['contextref'].str.contains('CurrentYear'))
                           &~(check1['contextref'].str.contains('NonConsolidated')) ] #~含まない
            if len(check2)==1 : break
        if len(check2)==1 : 
            return float( check2['amount'].values[0] )
        #当年度、個別
        for element_id in element_ids :
            check1 = data[ data['element_id'].str.contains(element_id) ]
            check2 = check1[ (check1['contextref'].str.contains('CurrentYear'))
                           & (check1['contextref'].str.contains('NonConsolidated')) ] #含む
            if len(check2)==1 : break
        if len(check2)==1 : 
            return float( check2['amount'].values[0] )
        #前年度、連結
        for element_id in element_ids :
            check1 = data[ data['element_id'].str.contains(element_id) ]
            check2 = check1[ (check1['contextref'].str.contains('Prior1Year'))
                           &~(check1['contextref'].str.contains('NonConsolidated')) ]
            if len(check2)==1 : break
        if len(check2)==1 : 
            return float( check2['amount'].values[0] )
        #前年度、個別
        for element_id in element_ids :
            check1 = data[ data['element_id'].str.contains(element_id) ]
            check2 = check1[ (check1['contextref'].str.contains('Prior1Year'))
                           & (check1['contextref'].str.contains('NonConsolidated')) ]
            if len(check2)==1 : break
        if len(check2)==1 : 
            return float( check2['amount'].values[0] )
        #ない場合は 0 とする
        return 0

    
    
def main2() :
 
    DIR = os.getcwd()
    filePaths = [ os.path.join(DIR ,fileName) for fileName in fileNames ]
    ar_eggs = [ pd.read_csv(filePath).drop_duplicates() for filePath in filePaths ]

    com_indices = pd.DataFrame()

    for eggs in ar_eggs :
        file_nms = eggs[ eggs['element_id']=='jpcrp_cor:companynamecoverpage' ].drop_duplicates()['file_nm'].tolist()

        for file_nm in tqdm(file_nms) :
            #[ 年度末の日付 ,アップロードの日付 ] を抽出、とれない → 次のループ
            new_dates = re.findall(r'20[0-9]{2}-[0-9]{2}-[0-9]{2}' ,file_nm)
            if len(new_dates)!=2 : continue

            edinet_code = re.search(r'E[0-9]{5}' ,file_nm).group(0)

            #同じ会社のデータがある場合
            if check_same_code(edinet_code ,com_indices.index.values) :
                b4_dates = re.findall(r'20[0-9]{2}-[0-9]{2}-[0-9]{2}' ,com_indices['file_nm'][edinet_code])

                #２つの日付を比較、最新なら古いデータを削除して処理を継続
                if check_date(new_dates ,b4_dates) : 
                    com_indices = com_indices.drop( index=[edinet_code] )

                #必要ないデータの → 次のループに
                else : continue

            #必要なデータを抽出
            check_data = eggs[ eggs['file_nm']==file_nm ]
            data = { col:check_element(check_data ,col) for col in dict_cols }
            data['証券コード'] = int( df_codes['証券コード'][edinet_code] / 10 ) #72030→7203の変換
            data['業種'] = df_codes['提出者業種'][edinet_code]
            data['file_nm'] = file_nm

            #抽出したデータを結合
            raw = pd.DataFrame( data ,index=[edinet_code] )
            com_indices = pd.concat([ com_indices ,raw ])

    com_indices['ROE'] = com_indices['EPS'] / com_indices['BPS']
    com_indices.to_csv( resultFileName )



if __name__=='__main__' :
    main2()
    files.download( resultFileName )

目次へ

３－５．抽出したデータを表示して確認

Google Colab: Cell-15

DIR = os.getcwd()
filePath = os.path.join( DIR ,resultFileName )
screening = pd.read_csv( resultFileName ,index_col=0 )
screening

目次へ

３－６．PER、ROE、一株配当などでスクリーニング

条件を指定してスクリーニングを行います。

Google Colab: Cell-16

screening = screening.copy()[ (screening['PER'].astype(float) <= 20)
                            & (screening['ROE'].astype(float) >= 0.15)
                            & (screening['一株当たりの配当'].astype(float) > 0) ]
screening[['証券コード','会社名','PER','ROE','一株当たりの配当']]

目次へ

４．ファイルのダウンロードがうまくいかない場合

ファイルのダウンロードでエラーが発生し、対処した。メモ。

chrome://settings/content/cookiesにアクセスし、「サードパーティの Cookie をブロックする」が無効になっていることを確認。

「許可」にcolab.research.google.comを追加。

目次へ

数分で検証できる！「Google Colab のテンプレート」をプレゼント

この記事を含め、過去に紹介してきた Google Colab のファイルをそのままプレゼントします。コピペする必要もなく、Googleドライブ上でまるごとコピーを作成して、そのまま使用することができます。

1. 単利を含むバックテスト 日経平均・為替 4時間足
2. 相関係数を算出してグラフを表示
3. TA-Lib でテクニカル分析を算出してグラフを表示
4. Ku-Chart を作成してグラフを表示
5. EDINETから財務データを取得してスクリーニング
6. 破産の確率を算出する

などなど！

必要なのはメールアドレスのご登録だけです！

※ 今後、Google Colab の記事を更新するごとに追加していく予定です

※ メルマガで行うのは、ほとんどがブログ更新のお知らせです。面白い、もしくは役に立つツールができたらお知らせすることもありますが、本業が忙しいのでせいぜい年に数回です。

※ 使用方法がわからない方はこの記事の最後にあるコメント欄からお問い合わせください。できる限りお答えします。コードの細かい部分も、ご質問いただければ出し惜しみせずお答えするつもりです！

数分で検証できる！「Google Colab のテンプレート」

誰でも、もっとも簡単に、さまざまな検証ができる Google Colaboratory のサンプル。

• Google Colab のテンプレートを
  ダウンロードする ≫