1．はじめに
どうも、趣味でデータ分析している猫背なエンジニアです。
余談ですが、チェーンソーマン公開されましたね...私も早く見に行きたいです😏
ちなみに私は岸部さん推しです（笑）
今回は、前回に引き続き兆しシリーズを進めていきたいと思います。本記事では前回のプロトタイプ（記事）のデータ収集部分を強化したので、記録していきます。
2. 収集ロジックの概要
■ 概要
前回の記事のデータ収集項で「暫定処置としてローカルに東証記載の銘柄をハンドコーディングで集めています。」と書いていました。これに関して改良・強化を行いました。
■ システム構成
銘柄番号の取得、辞書リスト更新、個別銘柄データ取得という順番でJPX内の最新のデータを取得します。
収集ロジックの概要を以下に添付します。
3. 収集器
■ 銘柄番号取得
ここでは、上場している企業名と銘柄番号が記載されているエクセルをダウンロードします。JPXのサイトで一般公開されているので、ありがたく使わせていただいています。これをダウンロードすることで上場廃止や新規上場などの最新情報を差分することができます。
一点注意なのが、xlsという旧型のエクセルを使用しているところです。さすが金融と思いました....。
# データリンクから対象のcsvを取得して保存
def download_tse_master():
# JPXの銘柄csvマスタページ
url = "https://www.jpx.co.jp/markets/statistics-equities/misc/01.html"
#ページ取得
res = requests.get(url)
res.raise_for_status()
# BeautifulSoupで解析
soup = BeautifulSoup(res.text, 'html.parser')
# Excelファイルリンクを探す（拡張子 .xls または .xlsx）
excel_link = None
for a in soup.find_all("a", href=True):
if ".xls" in a['href'] or ".xlsx" in a['href']:
excel_link = a['href']
break
if excel_link:
# 相対パスなら絶対URLに変換
if not excel_link.startswith("http"):
excel_link = "https://www.jpx.co.jp" + excel_link
save_dir = "D:/01.開発/05KK/finance_TSE/"
os.makedirs(save_dir, exist_ok=True) # フォルダがなければ作成
ext = os.path.splitext(excel_link)[1] # ".xls" or ".xlsx"
# 固定ファイル名に拡張子をつける
filename = "TSE_master" + ext
save_path = os.path.join(save_dir, filename)
with open(save_path, "wb") as f:
f.write(requests.get(excel_link).content)
print(f"✅ JPX銘柄情報 DL完了: {save_path}")
else:
print("❌ Excelリンクが見つかりませんでした")
■ 辞書リスト更新
ここでは、DLしたエクセルを辞書型に変換してyfinanceで読みやすい形式に変換している部分になります。いろんな変換の方法があると思いますが、私は辞書型に変換して銘柄番号と銘柄名を取得する方法がスタンダードになっているので、その方法をとっています。
ちなみに、さきほどDLしたエクセルの旧フォーマットに関してですが、一般的なpandasでは旧エクセルは読めないため、xlrdという追加ライブラリ(エンジン)が必要なようです。
# ----------------------------------------------------- #
# content : JPXのドキュメントから東証上場銘柄を更新 #
# ----------------------------------------------------- #
jpx_dataset_path = "D:/01.開発/05KK/finance_TSE/TSE_master.xls"
jpx_dataset_path_new = "D:/01.開発/05KK/finance_TSE/TSE_master.xlsx"
finance_csv_path = f"D:/01.開発/05KK/finance_TSE/TSE_dataset.csv"
def update_tse_master():
if jpx_dataset_path:
jpx_dataset = pd.read_excel(jpx_dataset_path, header=0, engine="xlrd")
elif jpx_dataset_path_new: # 新しいフォーマットに変わった時のため
jpx_dataset = pd.read_excel(jpx_dataset_path_new, header=0)
else:
print("❌ JPXデータファイルが見つかりません")
finance_dataset = jpx_dataset[['コード', '銘柄名']]
finance_dataset.to_csv(finance_csv_path, index=False, encoding='utf-8-sig')
print(f"✅ JPX銘柄情報 更新完了: {finance_csv_path}")
# ----------------------------------------------------- #
# content : コードと銘柄名を辞書型に変換する #
# ----------------------------------------------------- #
save_finance_path = f"D:/01.開発/05KK/finance_TSE/tickers_code.py"
# 銘柄辞書を作成（コードをkey, 銘柄名をvalueにする）
def get_tse_CodeAndName():
stocks_dict = {}
reader = pd.read_csv(finance_csv_path, encoding='utf-8').reset_index()
for _, row in reader.iterrows():
code = str(row['コード']).strip()
name = str(row['銘柄名']).strip()
stocks_dict[code] = name
# 辞書をJSONで保存
with open(save_finance_path, mode='w', encoding='utf-8') as file:
file.write("stock_dict = ")
json.dump(stocks_dict, file, ensure_ascii=False, indent=4)
print(f"✅ JPX銘柄辞書 更新完了: {save_finance_path}")
■ 個別銘柄データ取得・保存
前回の記事でもこの部分はあったのですが、強化版にしました。具体的には取得成功だけではなく、プログレスバーを追加してどのくらいの銘柄があって、進捗はどうなのかを追加しました。また、定期実行を目指しているのでlog取得機能も追加して運用・保守面でも強化しました。
def download_tse_stocks(period="1y", interval="1d"):
# stock_dict をループして yfinance で取得
for code, name in tqdm(Finance_code.stock_dict.items(),desc="株価データDL進捗",unit="銘柄"):
ticker = f"{code}.T" # yfinance 用のティッカーを作成
max_retry = 3 # 最大リトライ回数
attempt = 0
success = False
while attempt < max_retry and not success:
attempt += 1
data = yf.download(ticker, period=period, interval=interval, progress=False, auto_adjust=False)
if data.empty:
if attempt < max_retry:
time.sleep(3)
else:
success = True
csv_path = f"D:/01.開発/05KK/finance_TSE/finance_stock_data/{code}_{name}_stock.csv"
if isinstance(data.columns, pd.MultiIndex):
data.columns = [col[0] for col in data.columns]
data.to_csv(csv_path)
if not success:
log_save_dir = f"D:/01.開発/05KK/finance_TSE/log_file/"
os.makedirs(log_save_dir, exist_ok=True)
log_file_path = os.path.join(log_save_dir, "failed_downloads.log")
failed_tickers = []
failed_tickers.append(f"{code},{name},{ticker}")
# 失敗ティッカーをログファイルに保存
if failed_tickers:
with open(log_file_path, "w", encoding="utf-8") as f:
f.write("\n".join(failed_tickers))
print(f"❌ 取得失敗ティッカーを検出しました。")
print(f"logファイル： {log_file_path}")
else:
print("✅ すべてのティッカーを取得できました")
DL後のcsv群はこんな感じ👍
4.おわりに
今回は兆しシリーズのKK-FinancialEaterの収集を強化・改良というところで進めていきました。実際にハンドコーディングでデータを集めていたので、めんどくささが格段に減りました。次回は「スクリーニング1号機」の内部ロジックを追加していきたいと思います。
📈 兆しシリーズ
■ 原点(KK-Adam)
■ KK-FinancialEater
〇プロトタイプ編