はじめに
Pythonを使って企業が発表する適時開示の自動通知機能を作りました。
サービスの概要
東証が提供している「適時開示情報閲覧サービス」というwebから特定の企業の情報が開示されたら通知するという内容です。
全体のコード
from bs4 import BeautifulSoup
import urllib.request
import datetime
import MySQLdb
import json
import requests
import list
connection = MySQLdb.connect(
host='ホスト名', user='ユーザー名', passwd='パスワード', db='tdnet_db', charset='utf8')
cursor = connection.cursor()
cursor.execute("""CREATE TABLE IF NOT EXISTS tekijikaiji(
id INT auto_increment,
date VARCHAR(10) CHARACTER SET utf8 COLLATE utf8_general_ci,
company_code INT,
company_name VARCHAR(300) CHARACTER SET utf8 COLLATE utf8_general_ci,
company_title VARCHAR(300) CHARACTER SET utf8 COLLATE utf8_general_ci,
company_url VARCHAR(300) CHARACTER SET utf8 COLLATE utf8_general_ci,
PRIMARY KEY (id));""")
WEB_HOOK_URL = "https://hooks.slack.com/services/XXXXXXXXXXXXXXXXXXXXXX"
d_today = datetime.date.today()
sdate = d_today.strftime('%Y%m%d')
url = f"https://www.release.tdnet.info/inbs/I_list_001_{sdate}.html"
sub_url="https://www.release.tdnet.info/inbs/"
html = urllib.request.urlopen(urla)
bs = BeautifulSoup(html.read(),"html.parser")
tables = bs.find_all("table")[3]
trs=tables.find_all("tr")
for tr in trs:
date=tr.find("td",class_="kjTime").get_text()
code=tr.find("td",class_="kjCode").get_text()
name=tr.find("td",class_="kjName").get_text()
title=tr.find("td",class_="kjTitle")
title_text=title.get_text()
url=sub_url+title.find("a").get("href")
if code in list.code_list:
cursor.execute("SELECT * FROM tekijikaiji WHERE company_name = %s AND company_title = %s", (name,title_text))
if cursor.rowcount== 0:
requests.post(WEB_HOOK_URL, data=json.dumps({
"text" : name,
'attachments': [{
'title': title_text,
'title_link': url
}]
}))
cursor.execute("INSERT INTO tekijikaiji VALUES (%s, %s, %s, %s, %s, %s)", (0, date, code, name, title_text, url))
connection.commit()
code_list=["47430","70470","35570","22080","15624","18520","18980","47500","72010","65020"]
mainのファイルと、対象企業のコード番号が記載された外部ファイルの二つを作りました。
MySQLへの接続
⑴取得した情報をデータベースに保存するためにMySQLに接続
⑵tekijikaijiテーブルを作成
対象となる開示情報の開示時刻、銘柄コード、企業名、タイトル、URlが欲しいのでそれぞれのカラムをdate, company_code,company_name,company_title,company_urlとし、プライムキーをidとします。
#データベースtdnet_dbに接続
connection = MySQLdb.connect(
host='ホスト名', user='ユーザー名', passwd='パスワード', db='tdnet_db', charset='utf8')
cursor = connection.cursor()
#tekijikaijiテーブル作成
cursor.execute("""CREATE TABLE IF NOT EXISTS tekijikaiji(
id INT auto_increment,
date VARCHAR(10) CHARACTER SET utf8 COLLATE utf8_general_ci,
company_code INT,
company_name VARCHAR(300) CHARACTER SET utf8 COLLATE utf8_general_ci,
company_title VARCHAR(300) CHARACTER SET utf8 COLLATE utf8_general_ci,
company_url VARCHAR(300) CHARACTER SET utf8 COLLATE utf8_general_ci,
PRIMARY KEY (id));""")
開示情報の取得
⑴対象となるurlは"https://www.release.tdnet.info/inbs/I_list_001_20210308.html"
です。"20210308"の日付の部分は任意に変える必要があるので、標準ライブラリ「datetime」を使って必ず当日の日付が入るように設定します。
⑵時刻、コード、会社名、表題,URLを取得します。
#本日の日付をYYYYMMDDの形式に設定する
d_today = datetime.date.today()
sdate = d_today.strftime('%Y%m%d')
url = f"https://www.release.tdnet.info/inbs/I_list_001_{sdate}.html"
sub_url="https://www.release.tdnet.info/inbs/"
html = urllib.request.urlopen(urla)
bs = BeautifulSoup(html.read(),"html.parser")
tables = bs.find_all("table")[3]
trs=tables.find_all("tr")
#時刻、コード、会社名、表題、URLを取得
for tr in trs:
date=tr.find("td",class_="kjTime").get_text()
code=tr.find("td",class_="kjCode").get_text()
name=tr.find("td",class_="kjName").get_text()
title=tr.find("td",class_="kjTitle")
title_text=title.get_text()
url=sub_url+title.find("a").get("href")
対象企業の最新の開示情報があればDBに保存→通知
⑴取得したデータが対象企業のものであるかどうかをスクリーニングするために、対象企業のコードが記載された外部ファイルを読み込み、比較します。
⑵外部ファイルに記載されたコードのデータであれば、DBから同一のコードと開示タイトルがすでにあるかどうかをチェックします。チェック方法は、(1)で実行した処理件数を取得し、0の場合(つまり、DBにない情報=最新の情報)は、slackで通知し、DBに登録します。
あとはこれをwindowsならタスクスケジューラ、macならcronで5分ごとにプログラムを実行するように設定するだけです。
#対象企業かどうかチェック
if code in list.code_list:
cursor.execute("SELECT * FROM tekijikaiji WHERE company_name = %s AND company_title = %s", (name,title_text))
if cursor.rowcount== 0:
requests.post(WEB_HOOK_URL, data=json.dumps({
"text" : name,
'attachments': [{
'title': title_text,
'title_link': url
}]
}))
cursor.execute("INSERT INTO tekijikaiji VALUES (%s, %s, %s, %s, %s, %s)", (0, date, code, name, title_text, url))
connection.commit()
工夫した点
⑴TDnet用のwebAPIがありますが → https://webapi.yanoshin.jp/tdnet/
APIを利用する場合は情報がアップされて反映されるまでにタイムラグがありましたので、少し処理は多くなりますがAPIを使わない方法で行いました。
⑵取得したデータを保存、最新のものかどうかをチェックするために別の外部ファイルに保存して比較していましたが、勉強のためにMySQLを使ました。
⑶開発者以外の人でも簡単に対象企業を削除・追加できるようにしました。外部ファイルlist.pyにコードを追加・削除するだけで使用できます。