😊
第二回：有価証券報告書を解析する機能を作ってみた
公開
2024-12-08
更新
2024-12-09
文章量
約6340字
この記事は、Python Advent Calendar 2024 シリーズ3の10日目の記事となります。
はじめに
みなさま、ご無沙汰してます。
気づいたら、第一回の記事から半年くらい経ってました・・。
サービス自体の開発はゆっくり進めてましたが、全然記事が書けていなかったので、
現状の進捗状況も含めつつ、一つ作った機能を紹介できればと思います。
※前回の記事で気分次第で作る機能変わるかも、と記載しておりましたが、思いっきり変わってます・・。LLMを使うという点は変わっていません。
現状のサービス開発状況
現状ですが、GCPクラウド上で下記のようなサービスを構築しています。
クライアントはブラウザ（Web UI）となっております。
LLMとのチャット機能
LLMに対してチャットが出来る機能。とりあえずバックエンドAPIが稼働するのをチェックする用にAPIとして構築
有価証券報告書の分析機能
有価証券報告書を元に、財務三表の分析・業績が今後どうなるか？をLLMに分析してもらう機能
今回の記事のメインテーマ
ログイン機能
メールアドレスとパスワードでログインするための機能。ログインしていない場合は、前述した機能が使えないように認証機能を追加。
ログ保存機能
チャットの会話履歴、分析で利用した入力データ、分析結果などなどをログとして、bigquery上に保存するようにした機能。後ほどモニタリングを行うための仕組みも考えており、そのためにログを保存しています。（プロンプトエンジニアリングを行い、ABテスト比較を行う際にも、精度比較をするための指標設計をし、ログを蓄積し、指標比較が必要となります）
インフラとしては、下記なようにバージョンアップされています。
※ステップ3となっているのは、ステップ1,2が実はこの前段階であったからなのですが、それは一旦割愛します。
認証周りについては、ちょっと自動的に構築するためのコード化が不十分なので、また第3回で詳細を触れればと思っていますが、アカウント管理をIdentity Platformを利用し、ユーザー認証後のトークン認証をIdentity Aware Proxyを使っております。IaPを利用することで、非認証のリクエストを弾くことが可能となります。（アクセス制御の実施）
※モバイルアプリなどのWeb Front以外のアクセスも今後考えているため、LB経由でIaPでアクセス制御を実施しています。
今回主題となるのは、「有価証券報告書の分析機能」となります。
ちなみに今作っているサービスの実装コードは下記のリポジトリにまとまっています。
※ドキュメント作成やIaCが間に合っていないので、作成中。
有価証券報告書の分析機能とは？
有価証券報告書は、端的にいうと、株式を発行する上場企業が定期的に開示する情報となります。情報の中には、企業の状況・事業の状況・財務三表などが含まれます。
企業への株式投資などの判断を行う際に、上記有価証券報告書を分析したりして、判断を行うケースがあります。ただし、慣れていないと、報告資料の量（100ページ以上超えるのは割と多い印象）も多いため、分析が大変です。
そこで、LLMを使って、有価証券報告書を解読してもらい、企業分析をしてもらうような機能があると良いな、と思ったのが、分析機能を作るきっかけとなります。
※geminiやGPTを使って、インターネットにアクセスさせての分析も可能ですが、その場合、ハルシネーションリスク（= 分析結果が明らかな嘘）も高まるため、企業の決算情報が含まれる有価証券報告書を入力とした分析を行いたいと考えました。
分析機能の要件
有価証券報告書の分析機能は以下のような処理フローとなります。
ブラウザ経由でサービスにアクセス（ログイン情報を入力し、サービスにログイン）
会社名をキーワード入力し（部分一致可能）、検索ボタンを押す
有価証券報告書の検索：サービス側でキーワード情報から、有価証券報告書を検索し、ヒットした資料を一覧として、ユーザーに表示する
ユーザーは、該当の有価証券報告書を選択し、分析ボタンを押す
有価証券報告書の分析：サービス側で、選択された有価証券報告書を分析し、分析結果をユーザーに表示する（このタイミングで、分析結果の保存や、入力となった有価証券報告書情報も保存）
1に関しては、ログイン認証もありますが、詳細は今回は省略します。
有価証券報告書の検索と保存
有価証券報告書は、各企業のHPに掲載されていたりしますが、実は金融庁で提供しているEDINETのサービスを使うことで、Webブラウザ経由・API経由で取得が可能です。（参考：https://www.fsa.go.jp/search/20130917.html）
API仕様書は下記のWebサイト経由でドキュメントとして見ることが出来ます。
※利用規約についても、上記サイト経由で閲覧可能です。（無料利用可能、商用利用も可能そうかな、多分）
無料でアカウント登録ができ、APIキーを発行して、RESTfulなAPIにアクセスすることで、有価証券報告書を取得することが出来ます。
アカウント登録やAPIキー発行については、下記の記事が参考になります。
EDINETではざっくり以下のようなAPIが提供されています。
※2024/12/08時点
書類一覧API（入力として、日付情報などが必要）
書類取得API（入力として、ドキュメントIDなどが必要）
上記を見ると分かるのですが、ドキュメントをキーワード情報から一覧取得するためのAPIがありません。
日付をキーとしての一覧取得はできますが、キーワード（ex. 会社名など）を使った検索には非対応となります。
自分で作るWebサービスについて、ユーザー目線だと、キーワード検索で有価証券報告書が一覧化できないと、使い勝手が悪いと思ったので、キーワード検索ができるように対応することにしました。以下のような処理を実施しています。
書類一覧情報をサービス内部に保存
デイリーでバッチジョブを実行し（CloudRun Job）、EDINET API経由で、書類一覧情報を取得
取得した一覧情報をデータベース（BigQuery）に保存
書類情報の検索
ユーザーのキーワード情報から、データベース（BigQuery）にクエリを投げて、検索処理を実施し、書類一覧を取得
書類（pdf）を分析タイミングで取得
分析時に、指定した書類情報から、書類URLを取得し、EDINETの書類取得API経由で、書類をダウンロード。ダウンロードした書類をGCSに保存。
保存した書類（有価証券報告書）をLLMを使って分析
pythonを用いたEDINETの書類一覧APIの利用コードは下記となります。
※DataFrameに一覧情報を格納しています。
class EdinetWrapper:
def __init__(self, api_key: str, output_folder: str = None) -> None:
self.__api_key = api_key
self.__output_folder = (
os.path.join(os.path.dirname(__file__), "output", datetime.now().strftime("%Y%m%d%H%M%S"))
if output_folder is None
else output_folder
)
os.makedirs(self.__output_folder, exist_ok=True)
def get_documents_info_dataframe(self, target_date: datetime) -> pd.DataFrame:
url = "https://disclosure.edinet-fsa.go.jp/api/v2/documents.json"
params = {
"date": target_date.strftime("%Y-%m-%d"),
"type": 2, # 2は有価証券報告書などの決算書類
"Subscription-Key": self.__api_key,
}
response = requests.get(url, params=params)
if response.status_code != 200:
raise Exception(f"failed to get document list! http status code is {response.status_code}")
json_data = response.json()
status_code = int(json_data["metadata"]["status"])
if status_code != 200:
raise Exception(f"failed to get document list! status code is {status_code}")
documents = json_data["results"]
df = pd.DataFrame(documents)
return dfまた、同様にEDINETの書類取得APIの利用コードは下記となります。
有価証券報告書のファイルフォーマットをpdfに指定して、ダウンロードしています。gcsへの保存処理は別実装となっています。
class EdinetWrapper:
def __init__(self, api_key: str, output_folder: str = None) -> None:
self.__api_key = api_key
self.__output_folder = (
os.path.join(os.path.dirname(__file__), "output", datetime.now().strftime("%Y%m%d%H%M%S"))
if output_folder is None
else output_folder
)
os.makedirs(self.__output_folder, exist_ok=True)
def get_document_url(self, doc_id: str) -> str:
return f"https://api.edinet-fsa.go.jp/api/v2/documents/{doc_id}"
def download_pdf_of_financial_report(self, doc_id: str) -> str:
url = self.get_document_url(doc_id=doc_id)
params = {"type": 2, "Subscription-Key": self.__api_key} # PDFを取得する場合は2を指定
try:
res = requests.get(url, params=params, verify=False)
output_path = os.path.join(self.__output_folder, f"{doc_id}.pdf")
if res.status_code != 200:
raise Exception(f"fail to download {doc_id} document. status code is {res.status_code}")
with open(output_path, "wb") as file_out:
file_out.write(res.content)
return output_path
except urllib.error.HTTPError as e:
if e.code >= 400:
sys.stderr.write(e.reason + "\n")
else:
raise eEDINET関連の実装コードは下記でまとめています。
有価証券報告書の分析
書類（有価証券報告書）がgcsに保存できたら、保存したgcs uriを使って、LLM（gemini v1.5 flash）に分析をさせます。
サンプルコードについては、下記のリポジトリのコード（controller -> agent）となります。
まとめ
今回は、有価証券報告書をLLMを使って分析するための機能について、EDINETを使って検索機能を実装した内容を含めて、記事に記載しました。
ログの保存周り、認証周りについても、次の記事で詳細触れられればと思います。
※何か不明点などあれば、コメントなどでいただけると幸いです。