- 前説：これまでもやってきたが…無様なスクリプトが気になり…。
- 今回の工夫：全体を「関数型」風に書き換えました。
- PDF解析の流れ
- 今回のスクリプトは複数日対応にしていません。
- Python:PDFMinerで解析し、CSV保存するスクリプト。
この記事では、ネット上にPDF形式である「東京証券取引所日報」からテキスト抽出する作業を、Python+PDFMinerでおこなったことをとりあげます。最後にPythonスクリプトを載せました。
前説：これまでもやってきたが…無様なスクリプトが気になり…。
２年ほど前にPythonで書いたとりあえず動くスクリプトができたので記事をUPしました。
色々、反省があった（＝怪しい）ことはすでに別記事に載せました。
最初のスクリプトよりは多少マシにはなったのかと思います。「関数型」風の書き方に変えることができ、可読性が上がったと自覚（≒幻想）しています。
今回の工夫：全体を「関数型」風に書き換えました。
爺、1980年代前半からN88-BASICに触る機会があり、「手続き型」をやってました。かなりブランクがあり、（更に(略)）Pythonを使い始めたときも、そのようなお作法でした。
「関数」は、同じ処理を何回も使うようなときに使う、という思い込みがありました。「VTuberサプーが教える！ Python 初心者のコード／プロのコード」を読み、書き方を変えねばと思い至りました。
これまで書き散らかしてきたいくつかのスクリプトの中でも、今回のスクリプトは、自分でも何をやっているかよくわからないまま作ったもので、改修も遅めに…。
「関数型」風にやっと書き換えることができました。これについては、Micrsoft:Copilotにずいぶんと助けてもらいました。その先の「Class」を使ったところまで教えてもらいましたが、チョット飛躍し過ぎで、爺的には十分対応できませんでした。あわせて、「Class」を使うとimportした部品がさまざまな警告（エラーではないが型がヨロシクナイなど）表示にさらされるということで、先送りしました。
ということで、今回は「関数型」風な書き方のスクリプトです。
PDF解析の流れ
- すでに自分でデータベース化している東京証券取引所日報の最新日付取得
- 東証サイトへアクセスし、前項日付より新しいものがあったらダウンロード（DL）：複数日付あれば繰り返す
- DLしたPDFをPDFMinerで解析し、必要なデータを取得し、CSVで保存
- 前項のCSVをデータベースに保存
という流れです。
爺はデータベースとしてMySQLを使っています。SQLiteでも使えそうです。一日分は約4000銘柄です。
今回のスクリプトは複数日対応にしていません。
東証サイトからPDFをDLするのもPythonで行っています。爺用は、保存先に複数のPDFがあればfor文で処理してますが今回のスクリプトは単PDFに対応するようにしてます。
必要な方は、for文を適当に組み込んでくださいね。
事前に、PDFMner.six をインストールしておく必要があります。ターミナル画面でやるおまじないみたいなことです。
Python:PDFMinerで解析し、CSV保存するスクリプト。
ご利用は計画的に、自己責任でお願いしますね。誤動作などの結果何か起こっても知りませぬ（キッパリ）。
Python環境や、ファイル保存場所などはご自身のPCに合わせてくださいね。
# PDFMiner.sixで、東証株式相場表（日報、PDF）を解析し、CSVにし保存
# 前提：解析対象PDFは指定フォルダへDL済
# 前提：CSV保管用フォルダ作成済
# ライブラリ
import time
from datetime import datetime
# import sys
import dataclasses
from io import StringIO
from pdfminer.pdfparser import PDFParser
from pdfminer.pdfdocument import PDFDocument
from pdfminer.pdfinterp import PDFPageInterpreter, PDFResourceManager
from pdfminer.layout import LAParams
from pdfminer.converter import TextConverter
from pdfminer.pdfpage import PDFPage
import re
import csv
# クラス
@dataclasses.dataclass
class st:
pdf_folder : str = 'C:\\DLしたPDFを保管しているフォルダ\\'
csv_folder : str = 'C:\\CSV保管用フォルダ\\'
pro : str = 'TOKYO PRO Market銘柄'
unit: str = "千株[thous.shs.]\n千口/千個[thous.units.]\n"
class num:
len17 : int = 17
# 関数
def text_replace(txt): # 不要文字削除、銘柄コード先頭文字位置取得
d_pos = []
iter_pos = []
text_page = txt.replace(st.unit, "")
# text_page = txt.replace("千株[thous.shs.]\n千口/千個[thous.units.]\n", "")
text_page = text_page.replace("\t\t ", "\t")
# 単頁内最初の銘柄コード以前のテキストは、正規表現で結果的に無視される
# 銘柄コードの文字始終位置、合致文字列の一日分を納める
iter_pos = re.finditer(r"\t[0-9|A-C]{4}\t{1,3}.", text_page)
for def_i in iter_pos: # 銘柄コード先頭文字位置
d_pos.append(def_i.start() + 1)
return (text_page, d_pos) # ある程度整ったテキスト・データ、銘柄毎の先頭文字位置（銘柄コード）
def delComma_str2num(txt): # 加重平均と売買株数用 整形と数値化
txt = (
txt.replace(",", "") if re.search(r"\d\,\d", txt) else txt
) # 三桁区切りコンマ削除
r_txt = float(txt) if re.match(r"^[0-9]", txt) else txt
return r_txt
def set_parser_and_doc_of_PDF(pdf_path):
fp = open(pdf_path, 'rb')
parser = PDFParser( fp)
doc = PDFDocument(parser)
rs_manager = PDFResourceManager()
parms = LAParams(boxes_flow = None)
outfp = StringIO()
device = TextConverter(rs_manager, outfp, laparams = parms)
interpreter = PDFPageInterpreter(rs_manager, device)
return doc, interpreter, outfp
def get_text_from_doc_by_each_pages(pdf_path, interpreter, outfp):
fp = open(pdf_path, "rb")
for p, pdf_page in enumerate(PDFPage.get_pages(fp)):
text_on_a_page = ""
interpreter.process_page(pdf_page)
text_on_a_page = (outfp.getvalue()).replace("\n\n", "\t")
if re.search(st.pro, text_on_a_page):
text_on_a_page = text_on_a_page.split(st.pro)[0]
break
# text_on_a_page = text_on_a_page.split(st.pro)[0]
# pre_text:全頁テキスト div_pos_of_cds:銘柄コードの先頭文字位置
pre_text, div_pos_of_each_cds = text_replace(text_on_a_page)
# print(f'{p}\t\r')
outfp.close()
fp.close()
return pre_text, div_pos_of_each_cds
def slice_and_erase_for_shaping(date_pdf, pre_text, div_pos_of_each_cds):
pdf_date = datetime.strptime(date_pdf, '%Y%m%d').strftime('%Y%m%d')
# 銘柄情報「pre_text」をslice等で整理・整形し銘柄別に配列格納
cds_data = []
# print(pre_text) #必要情報OK
for a_pos in div_pos_of_each_cds[-3::-1]: # 逆順スライス ※正順は文字削除すると文字位置が狂う
# print(a_pos) # 必要位置情報OK
# 単銘柄内データ抽出
a_cd_data = pre_text[a_pos:]
a_cd_data = a_cd_data.replace("\n","")
a_cd_data = a_cd_data.split("\t")
a_cd_data if a_cd_data[1].isdecimal() else a_cd_data.pop(1)
# print(a_cd_data) # \t区切りされた株価ほかのデタOK
if len(a_cd_data) < num.len17: # 要素数不足
continue
if a_cd_data[1].startswith("1"): #第2項目=1:取引単位を指す位置
a_cd_clms =[]
a_cd_clms.append(pdf_date)
a_cd_clms.append(str(a_cd_data[0])) # 銘柄コード
a_cd_clms.append(int(a_cd_data[1])) # 取引単位
a_cd_clms = erase_comma(a_cd_data, 3, 11, a_cd_clms) # 前場始値～後場終値
# 項目（気配、前日比）スキップ
a_cd_clms = erase_comma(a_cd_data, 13, 15, a_cd_clms) # vwap 売買高
# 売買代金 スキップ
# print(a_cd_clms)
cds_data.append(a_cd_clms)
sorted_data = sorted(cds_data, key = lambda x: x[1]) # 銘柄コード昇順
pre_text = pre_text[: a_pos - len(pre_text)] # 処理済み1銘柄分のテキスト削除
return sorted_data
def erase_comma(a_cd_data, s, e, a_cd_clms):
for i in a_cd_data[s:e]:
i = i.replace(",", "") if re.search(r"\d\,\d", i) else i # 三桁コンマ削除
i = str(i) if re.match(r"^[0-9]", i) else "\0"
a_cd_clms.append(i)
return a_cd_clms
def save_data_as_csv(sorted_data, csv_path):
f = open(f"{csv_path}", "w", encoding="utf_8", newline="")
writer = csv.writer(
f, delimiter="\t", quoting=csv.QUOTE_NONNUMERIC
) # 項目間はtab=\t
writer.writerows(sorted_data)
f.close()
return '保存'
def main(time_s):
a_pdf = '20250130' # ここが解析対象のPDFの名前の日付部分。実際に作業対象にする名前に変更してください。
pdf_path = f'{st.pdf_folder}{a_pdf}.pdf'
csv_path = f'{st.csv_folder}{a_pdf}.csv'
doc, interpreter, outfp = set_parser_and_doc_of_PDF(pdf_path)
print(f'\n経過時間：{time.time() - time_s} 秒\t オブジェクト取得済')
pre_text, div_pos_of_each_cds = get_text_from_doc_by_each_pages(pdf_path, interpreter, outfp)
print(f'経過時間：{time.time() - time_s} 秒\t テキスト取得済')
sorted_data = slice_and_erase_for_shaping(a_pdf, pre_text, div_pos_of_each_cds)
print(f'経過時間：{time.time() - time_s} 秒\t リスト整形済')
save_data_as_csv(sorted_data, csv_path)
print(f'経過時間：{time.time() - time_s} 秒\t CSV保存済')
print(f'\t\t 日付:{a_pdf} CSV保存迄の作業 完了\n')
return 'done'
if __name__ == '__main__':
time_s = time.time()
main(time_s)
time_m = time.time()
print(f'経過時間：{time_m - time_s} 秒 PDF→CSV 全作業 終了\n')
Python作法の違いなどは優しく教えていただければ喜びます。
