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

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

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

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

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

# ざっくり要件
- 以下サイトで毎週火曜に更新される最新のPDFを使用する

https://www.jpx.co.jp/markets/statistics-equities/margin/05.html

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

## pipライブラリの比較検討

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

https://qiita.com/mczkzk/items/894110558fb890c930b5

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

- tabula-py

https://www.samoariblog.com/2021/04/tabula-py.html

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

- pdfminer3

https://qiita.com/daifuku10/items/3f99c56998f35419c7f3

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


# 完成品
インストールしたもの

pandas
urllib
pdfminer.six



```python

# -*- 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, LAParams
import requests


def extractTableData(pdf_path):
    laparams = LAParams(
        line_overlap=0.5,
        char_margin=0.5,
        line_margin=0.5,
        word_margin=0.5,
        boxes_flow=0.5,
        detect_vertical=False,
        all_texts=False,
    )
    df_output = pd.DataFrame(
        columns=[
            "type",
            "銘柄名コード",
            "コード",
            "新証券コード",
            "合計-売り残",
            "合計-売り前週比",
            "合計-買い残",
            "合計-買い前週比",
            "一般信用-売り残",
            "一般信用-売り前週比",
            "制度信用-売り残",
            "制度信用-売り前週比",
            "一般信用-買い残",
            "一般信用-買い前週比",
            "制度信用-買い残",
            "制度信用-買い前週比",
            "Type",
            "Market",
        ]
    )

    type, market = None, None
    type_mapping = {
        "貸借銘柄 loan trading issue": "margin",
        "制度信用銘柄 standardized margin trading issue": "system",
        "その他 other issues": "other",
    }
    market_mapping = {
        "プライム Prime": "Prime",
        "スタンダード Standard": "Standard",
        "グロース Growth": "Growth",
        "投信等 Investment trusts": "Investment",
    }

    for page, page_layout in enumerate(extract_pages(pdf_path, laparams=laparams)):
        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), text_line.x1, int(text_line.y1))
                    )

        df_x = pd.DataFrame(records, columns=["word", "x1", "x2", "y1"])
        df_x = df_x.sort_values(["y1", "x2"], ascending=[False, True])
        d_out, last_y1 = [], None

        for _, row in df_x.iterrows():
            type = type_mapping.get(row["word"], type)
            market = market_mapping.get(row["word"], market)
            if type == "ohter" or market == "Investment":
                continue

            if row["y1"] != last_y1:
                d_out, last_y1 = [], row["y1"]
            if (
                len(d_out) > 4 and row["word"][0] != "▲" and "▲" in row["word"]
            ):  # 数値が結合してる場合の処理
                a = row["word"].find("▲")
                col = [row["word"][:a], row["word"][a:]]
                d_out.extend(col)
            elif (
                len(d_out) == 1 and len(row["word"]) > 20 and row["word"][-5:].isdigit()
            ):  # 銘柄名とコードが結合してる場合の処理
                col = [row["word"][:-5], row["word"][-5:]]
                d_out.extend(col)
            else:
                d_out.append(row["word"])
            if len(d_out) == 16:
                if (
                    len(d_out[2]) > 25 and d_out[1][-5:].isdigit()
                ):  # 銘柄名が長すぎて銘柄名とコードの順序が逆転している場合の処理
                    d_out[1], d_out[2] = d_out[2], d_out[1]
                d_out.append(type)
                d_out.append(market)
                df_output.loc[len(df_output)] = d_out

    return df_output


def calculate_ratio(buy, sell):
    return round(buy / sell, 2) if sell != 0 else float("inf")


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[:-5]
    output_data["信用倍率"] = output_data.apply(
        lambda row: calculate_ratio(row["合計-買い残"], row["合計-売り残"]), axis=1
    )
    output_data["一般信用倍率"] = output_data.apply(
        lambda row: calculate_ratio(row["一般信用-買い残"], row["一般信用-売り残"]),
        axis=1,
    )
    output_data["制度信用倍率"] = output_data.apply(
        lambda row: calculate_ratio(row["制度信用-買い残"], row["制度信用-売り残"]),
        axis=1,
    )

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


def createFileName():
    print("ファイル名の生成")
    today = datetime.date.today()
    last_friday = today - datetime.timedelta(days=today.weekday() + 3)
    for i in range(5):
        date = (last_friday - datetime.timedelta(days=i)).strftime("%Y%m%d")
        file_name = "syumatsu" + date + "00.pdf"
        url = f"https://www.jpx.co.jp/markets/statistics-equities/margin/tvdivq0000001rnl-att/{file_name}"
        if requests.get(url).status_code == 200:
            break
    return file_name, date


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 fileClear(file_name):
    print("ファイルの削除開始")
    os.remove(file_name)
    print("ファイルの削除完了")


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


プログラムの流れはざっくりこんな感じ
1. 前週金曜日の日付からPDFのファイル名を特定
1. サイトからPDFファイルをDL
1. PDFファイルからpdfminer.sixを用いてデータを配列形式で抽出
1. CSVに出力
1. 抽出済みのPDFを退避フォルダに移動

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

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

```python
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に投げてくれたり、マージリクエストの作成と修正までして人間はレビューしてマージするだけにしてくれないかなーとも思いました。

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