◆はじめに
ここでは、財務情報を取得して解析、そして投資までのSTEPを紹介します。
今後は取得した財務情報を解析するまでに道筋を追って紹介していく予定です。
最終的には、公開企業の倒産確率を求めたり、独自ファクターを用いた株価予測モデルの構築などを行い、実際に投資を行っていくところまでやるのが目標です。
関連記事：
◆まずは財務情報の取得（一般的な方法）
投資を行うにあたって、まず財務情報を取得・解析・分析を行っていきます。
◇財務データの宝庫（EDINETとは）
Wikipediaより、EDINETとは、“Electronic Disclosure for Investors’ NETwork”の略称であり、財務情報の電子情報開示システムのことをいいます。
平たくいえば、財務情報をお国が管理・公示するためのホームページです。
◇企業は、誰のもの? EDINETでの公開義務
「企業は、誰のもの？」。
従業員のものか。経営者のものか。はたまた購買を行う利用者のためのものなのか。
会計学のうえで答えは１つで株主のものです。
したがって、企業は会計上、株主の利益を守るために情報を開示する義務を追います。
それをディスクロージャー制度（企業内容等開示）といいます。
ただし、原則として全ての企業がディスクロージャー制度の対象となるわけではなく、株主が多い企業が対象にディスクロージャー制度の対象となります。たとえば、上場企業などがそれらに当てはまります。
したがって、EDINETには上場企業など株主の多い企業の情報が全て掲載されます。
◇経営コンサルタントの財務分析（人依存・手作業）
経営コンサルタントは、EDINETやHPに公表される財務諸表を目で熟読して財務分析を行います。
なんとも原始的ですが、IT武装のない経営コンサルタントはその眼力をつかい、一人ひとりが感覚でノウハウをつけて、それら企業評価を行うのが一般的です。
IT業界では、それを人依存と揶揄する人もいるでしょう。けれども、機械的に財務分析を行うというのが、日本ではほとんどなされていないのも現実です。
それは、財務諸表を読みこむ難しさが背景にあります。たとえば、企業が財テクなどにより数値操作することがなされ、財務諸表の実情を知るには難読なものであることがあげられます。
◆財務情報を取得（機械的な方法）
◇財務データGETに超便利（有報キャッチャー）
企業の情報を機械的に取得するために、ここでは有報キャッチャーを利用します。
有報キャッチャーはＩＲ情報を取得するためのAPIを、一般企業が公開しているものです。
有報キャッチャーでは、HTTPのGETを行いレスポンスとして規定のXML文書を返却します。
そのXML文書からIR情報を取得するURLを取得し、法定開示情報（XBRL添付）の一覧の取得して解析を行っていきます
◇財務分析はやりたいけど、なかなか面倒な人は、TeCAを使おう
IT技術者でなければ、有報キャッチャーのAPIを利用したアプリケーションTeCAなどを利用すると良いでしょう。
コードを意識せずに、財務分析を行うことができます。
この記事は、IT技術者を駆使して解析を行う記事のため、こちらの説明は割愛します。
◇XBRLをPythonで一括取得する！
以下の記事を参考にPythonでデータを取得するプログラムを作り直しました。
上記はPython2系で動くようなアプリですが、3系で動くように変更しています。
その他にも、IRデータベースが作成できるように一括で取得するような定義に変更しています。
※初めて作成したpythonコードなので、コードの汚さはご了承下さい。
- 取得予定のデータをすべてJsonファイルにして一度dump
- 別処理でjsonファイルからデータを取得してdownload
- 処理したjsonファイルは、oldフォルダにmoveして同一処理を行わない
- 最新のデータ等日付指定で引き抜けるよう修正
- コメントを追加
- ディレクトリが存在しなければ作成するように変更
- マルチプロセス処理化（proc変数を変えれば、その分マルチプロセスを起動）★注意：やりすぎはDOS攻撃になるので節度を持って実行して下さい
以下をPythonの3系で実行していただければ、IR情報が取得できます
TODO部分で2015/07/17までに公示されたデータを作成するようにプログラムされていますが、
一度、必要な情報まで取得して、その後は、日付は外渡しにして継続的に最新データのみをアップデートしてもらえればと思います。
import requests
import xml.etree.ElementTree as ET
import json
import os
import io
import re
import multiprocessing
from collections import defaultdict
from datetime import datetime as dt
proc = 1 # TODO:マルチプロセス制御（初期値：1並列）
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):
yuho_dict = defaultdict(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
if not time_check(updated,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)
yuho_dict[_id] = {'id':_id,'title':title,'cd':cd,'url':url,'update':updated}
return yuho_dict
def is_yuho(title):
if u'有価証券報告書' in str(title):
return True
else:
return False
def time_check(update,since):
updated_time = dt.strptime(update, '%Y-%m-%dT%H:%M:%S+09:00')
return updated_time>=since
def make_directory(dir_name):
if not os.path.exists(dir_name):
os.mkdir(dir_name)
def write_download_info(ofname,info_dict):
with open(ofname,'w') as of:
json.dump(info_dict, of, indent=4)
def multi_controller(since):
p = 0
jobs = []
while(p < proc):
job = multiprocessing.Process(target=craete_xbrl_url_json, args=(since,p,))
jobs.append(job)
job.start()
p += 1
[job.join() for job in jobs]
def craete_xbrl_url_json(since,p):
#有報キャッチャーWebServiceのAtomAPIアドレス<http://resource.ufocatch.com/>
base_url = 'http://resource.ufocatch.com/atom/edinetx/query/'
namespace = '{http://www.w3.org/2005/Atom}'
#有報キャッチャーのページ
page = 1 + p
count = 0
while True:
#文字列変換
t_symbol = str(page)
print('page:'+t_symbol + ', loading...')
#企業毎の有報へのデータへのリンク情報を取得
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])
#downloadファイルの対象を取得
info_dict = get_link(ET_tree,namespace,since)
count += len(info_dict)
if len(info_dict) == 0 :
#取得データがなくなり次第、処理終了
print('process' + str(p) + ':complete a download!! [' + str(count) + ']')
break
#Request用のJson形式のファイルを作成
json_directory=os.getcwd()+'/downloaded_info'
make_directory(json_directory)
ofname = json_directory+'/dat_download_'+t_symbol+'.json'
write_download_info(ofname,info_dict)
page += proc
if __name__=='__main__':
#TODO 取得データの末尾（以下例では、2015/07/17以降のデータを取得）
since = dt.strptime('2015-07-17','%Y-%m-%d')
print('since:', since)
#一定期間に発生したデータ全てのURLを取得する
multi_controller(since)
上記で、取得対象のURLを格納したJsonを作成したら、次はXBRLをダウンロードする！
import requests
import json
import os
import io
import shutil
import multiprocessing
from collections import defaultdict
from zipfile import ZipFile
proc = 1 # TODO:マルチプロセス制御（初期値：1並列）
def make_directory(dir_name):
if not os.path.exists(dir_name):
os.mkdir(dir_name)
def download_all_xbrl_files(info_dicts):
# xbrl_filesのディレクトリが存在しないとき、作成する
directory_path = os.getcwd()+'/xbrl_files/'
make_directory(directory_path)
mp_dict = defaultdict(dict)
counter = 0
for _id, info_dict in info_dicts.items():
#分散処理用にデータを設定
mp_dict[counter] = info_dict
counter += 1
p = 0
jobs = []
while(p < proc):
job = multiprocessing.Process(target=_download_xbrl_file, args=(mp_dict,p,))
jobs.append(job)
job.start()
p += 1
[job.join() for job in jobs]
#is_yuho関数に当てはまる全ての企業×有報情報IDごとに取得する
def _download_xbrl_file(info_dicts,p):
no = p
directory_path = os.getcwd()+'/xbrl_files/'
while(no < len(info_dicts)):
info_dict = info_dicts[no]
no += proc
# 証券CDごとのディレクトリ作成
company_path = directory_path + info_dict['cd'] + '/'
ir_path = company_path + info_dict['id']
make_directory(company_path)
# 証券コード×IR情報ごとのディレクトリ作成
if os.path.exists(ir_path):
continue
make_directory(ir_path)
print('Process('+str(p + 1) + '):downloading:' + info_dict['update']+'_'+info_dict['title'])
url = info_dict['url']
r = requests.get(url)
if r.ok:
#Requestによりファイルを取得して、Unzipする
r = requests.get(url)
z = ZipFile(io.BytesIO(r.content))
z.extractall(ir_path) # unzip the file and save files to path.
def download_json_url():
search_directory = os.getcwd()+'/downloaded_info/'
old_directory = search_directory + 'old/'
make_directory(old_directory)
count = 0
for file_name in os.listdir(search_directory):
if not u'.json' in file_name :continue
print(file_name + ' loading...')
with open(search_directory + file_name,'r') as of:
info_dict = json.load(of)
print('downliading:' + str(count) + '~' + str(count+len(info_dict)) )
count += len(info_dict)
download_all_xbrl_files(info_dict)
shutil.move(search_directory + file_name, old_directory + file_name)
if __name__=='__main__':
download_json_url()