Pythonで個別株の信用取引情報を取得する

はじめに

今年から新NISA制度が始まり、投資の話題が職場でもよく飛び交うようになりました。
私も大学卒業を機に株の投資を始め、株の銘柄選びや売買のタイミングを勉強をしています。
そのおかげか「この数字は今どうなってるんだろう？」と思うことが増えてきました。

私は楽天証券のISPEEDで取引をしていますが、これがなかなか便利で大抵の数字はISPEEDで確認できます。しかし、中には載ってない数字・指標もあり、気になったらググって調べてはいますが、これが結構大変だなーと感じてました。

ということで、自動化しました。

私が欲しい情報自体は、日本証券取引所のサイトにアップロードされているPDFに載っているため、これをDL,データ抽出すれば事足ります。
今回はこの作業を自動化し、ゆくゆくはFirebaseのCloud FunctionsにデプロイしてFireStoreにデータをストックし、適当なWebアプリをホストしていつでもどこでも検索できるようにしたいなーと考えてます。

ざっくり要件

• 以下サイトで毎週火曜に更新される最新のPDFを使用する

• p85分のPDFを読み取り、表形式のデータを配列形式で抽出する
• 最終的にFirebaseのCloud Fucntionsで動かすため、処理時間が500秒以内に収まるようにする

pipライブラリの比較検討

比較検討したライブラリは以下３つです。最終的にpdfminer.sixを採用しました。

• pdfminer.six

PDFから文字と文字の位置情報を抽出するライブラリ
抽出する際に表データのみを取得できるようにすることもできるので結構便利
85ページ分のデータを抜き出すのに2分程度かかる

• tabula-py

PDFから表データをそのまま抜き出すことができるので超便利
実行速度は爆速
もともとJavaのライブラリで、tabula-pyはラッパーのため動かすにはJavaのインストールが必須
Cloud FunctionでJavaとPythonを両方インストールして動かす、といのは多分NGっぽく、tabula-pyを使う場合はCloud RunにDocker環境デプロイして動かすことになりそうなので今回は対象外

• pdfminer3

pdfminer.sixとやってることは同じ
ただ動作速度がめちゃんこ遅く、85ページからデータ抽出をしようとしたところ50ページ読み込むのに30分以上かかった(途中で止めた)ので今回は対象外
速度が遅い原因については、一文字ずつエンコードのチェックとかして精度上げてるせいだ、みたいなこと書いてる記事見たので、レイアウトがバラバラ、ページ数が少ないPDFとかには強そう

完成品

インストールしたもの

pandas
urllib
pdfminer.six

# -*- coding: utf-8 -*-
import pandas as pd
import urllib.request
import datetime
import shutil
import os
from pdfminer.high_level import extract_pages
from pdfminer.layout import LTTextBoxHorizontal

def extractTableData(pdf_path):
    df_output = pd.DataFrame(columns=[
        '銘柄名コード', 'コード', '新証券コード', '合計-売り残', '合計-売り前週比',
        '合計-買い残', '合計-買い前週比', '一般信用-売り残', '一般信用-売り前週比',
        '制度信用-売り残', '制度信用-売り前週比', '一般信用-買い残', '一般信用-買い前週比',
        '制度信用-買い残', '制度信用-買い前週比'
    ])

    for page, page_layout in enumerate(extract_pages(pdf_path)):
        print(page)
        records = []

        for element in page_layout:
            if isinstance(element, LTTextBoxHorizontal):
                for text_line in element:
                    text = text_line.get_text().strip()
                    records.append((text, round(text_line.x1, 2), int(text_line.y1)))

        df_x = pd.DataFrame(records, columns=['word', 'x1', 'y1'])
        df_x = df_x.sort_values(['y1', 'x1'], ascending=[False, True])

        d_out = []
        last_y1 = None

        for _, row in df_x.iterrows():
            if row['y1'] != last_y1:
                d_out = []
                last_y1 = row['y1']
            d_out.append(row['word'])
            if len(d_out) == 15 and d_out[2].startswith("JP3") and d_out[0].startswith("B"):
                df_output.loc[len(df_output)] = d_out

    return df_output

def getData(file):
    print("データの取り込み開始")
    output_data = extractTableData(file)

    # 信用取引の買い残、売り残の文字列からコンマを除去し、数値に変換
    for col in ['合計-売り残', '合計-買い残', '一般信用-売り残', '一般信用-買い残', '制度信用-売り残', '制度信用-買い残']:
        output_data[col] = pd.to_numeric(output_data[col].str.replace(",", ""), errors='coerce')

    # ノイズ除去
    output_data = output_data.dropna(how='any')

    # 銘柄名の抽出
    output_data['銘柄名'] = output_data['銘柄名コード'].str[1:-5]
    output_data['信用倍率'] = output_data.apply(
        lambda row: round(row['合計-買い残'] / row['合計-売り残'], 2) if row['合計-売り残'] != 0 else float('inf'), axis=1
    )
    output_data['一般信用倍率'] = output_data.apply(
        lambda row: round(row['一般信用-買い残'] / row['一般信用-売り残'], 2) if row['一般信用-売り残'] != 0 else float('inf'), axis=1
    )
    output_data['制度信用倍率'] = output_data.apply(
        lambda row: round(row['制度信用-買い残'] / row['制度信用-売り残'], 2) if row['制度信用-売り残'] != 0 else float('inf'), axis=1
    )

    print("データの取り込み完了")
    return output_data

def createFileName():
    print("ファイル名の生成")
    today = datetime.date.today()
    last_friday = today - datetime.timedelta(days=today.weekday() + 3)
    file_name = 'syumatsu' + last_friday.strftime('%Y%m%d') + '00.pdf'
    return file_name

def fileDownload(file_name):
    print("ファイルDL開始")
    url = f"https://www.jpx.co.jp/markets/statistics-equities/margin/tvdivq0000001rnl-att/{file_name}"
    urllib.request.urlretrieve(url, file_name)
    print("ファイルDL完了")

def fileMove(file_name):
    print("ファイルの退避開始")
    os.makedirs("old", exist_ok=True)
    shutil.move(file_name, os.path.join('old', file_name))
    print("ファイルの退避完了")

if __name__ == "__main__":
    file_name = createFileName()
    fileDownload(file_name)
    output_data = getData(file_name)
    output_data.to_csv("extract_sample.csv", index=False)
    fileMove(file_name)
    

プログラムの流れはざっくりこんな感じ

1. 前週金曜日の日付からPDFのファイル名を特定
2. サイトからPDFファイルをDL
3. PDFファイルからpdfminer.sixを用いてデータを配列形式で抽出
4. CSVに出力
5. 抽出済みのPDFを退避フォルダに移動

PDFからデータを抽出する処理はextractTableDataで行っていて、各ページごとに以下の順で処理を実行してます。

1. [文字,x座標,y座標]で配列を作成
2. x座標、y座標でソート
3. ソートしたデータからy座標が同じになる15個の文字を２次元配列に放り込む（ついでに不要なデータをここで除外している）

def extractTableData(pdf_path):
    df_output = pd.DataFrame(columns=[
        '銘柄名コード', 'コード', '新証券コード', '合計-売り残', '合計-売り前週比',
        '合計-買い残', '合計-買い前週比', '一般信用-売り残', '一般信用-売り前週比',
        '制度信用-売り残', '制度信用-売り前週比', '一般信用-買い残', '一般信用-買い前週比',
        '制度信用-買い残', '制度信用-買い前週比'
    ])

    for page, page_layout in enumerate(extract_pages(pdf_path)):
        print(page)
        records = []

        for element in page_layout:
            if isinstance(element, LTTextBoxHorizontal):
                for text_line in element:
                    text = text_line.get_text().strip()
                    records.append((text, round(text_line.x1, 2), int(text_line.y1)))

        df_x = pd.DataFrame(records, columns=['word', 'x1', 'y1'])
        df_x = df_x.sort_values(['y1', 'x1'], ascending=[False, True])

        d_out = []
        last_y1 = None

        for _, row in df_x.iterrows():
            if row['y1'] != last_y1:
                d_out = []
                last_y1 = row['y1']
            d_out.append(row['word'])
            if len(d_out) == 15 and d_out[2].startswith("JP3") and d_out[0].startswith("B"):
                df_output.loc[len(df_output)] = d_out

    return df_output

おわりに

2,3年ぶりくらいにコードを書いたので、結構雑で力業になってしまいました...
実際プログラム書いたあと、ChatGPTに投げてリファクタリングしてもらいましたが、結構訂正されていて「あ、こんなに書けなくなってるんだ」と思いました。

ふと思ったのが、VSCodeでコード書いてるときとか、GitにコミットしたタイミングなどでChatGPTを噛ませて、リファクタの提案（こここうした方がいいですよ～みたいなの）をしてくれるといいなと思いました。
またGitのコードをChatGPTが確認して、潜在的なエラーやリファクタが必要な個所を検知してIssueに投げてくれたり、マージリクエストの作成と修正までして人間はレビューしてマージするだけにしてくれないかなーとも思いました。

今度このあたりのノウハウを実装してみて、記事にしてみようかと思います。