株価取得、スクレイピングスクリプト、Pythonコード、株探

 

目的：指定した日の株価を取得して、CSVとして出力する。2026/10/23

条件：株探サイトにアクセスし、株価を取得する。指定するのはコードと取得したい日を設定。過去約1年くらいは、株探にデータが残っている。なお、取得できない場合は、取得不可になる。

使用方法：スクリプト内の「# --- 設定項目」にコードと日付を設定。「# --- CSVファイルへの保存処理 ---」に、CSV出力先のパスを記入。フォルダは事前に作成しておくこと。

　実行はDOS窓やPowerShellを起動して、   python スクリプト名 で実行させる。

注意事項：Webスクレイピングを実行する場合、連続して大量にデータを取得すると相手方のサーバー等に負荷をかけて迷惑がかかりますので、スリープやディレイを設定して、連続してアクセスしないように設定してください。

作成した日以降の株探側のサイト仕様の変更が発生した場合は、その変更に合わせてスクリプトを変更する必要がある。今回GeminiProを使用した。正常に動作しない場祭は、再度GeminiProもしくはそれに類するAIにこのスクリプトを添付して、修正を指示させればよいはず。

import os
import re
import time
import datetime
import pandas as pd
import requests
from bs4 import BeautifulSoup

#修正履歴、2026/10/22　ページを遡って過去の株価を取得
#002番は当日のみに対応、003は作業中でCSV出力なし、005は完了ただし、未確認
#もし、005で実行してだめだったら、003で画面コピーして株価を取得
#その後005をGeminiProに聞いて修正して
# 006が最新

#なお株探は、約1年前（実際はもう少し過去の株価もある）くらいまで取得できる
#それ以上過去になると、一個ずつトレーディングビューで調べるかヤフーファイナンス(有料に加入)
#とかマーケットスピード2とかで調べる必要がある

# --- 設定項目 --- コード4桁、アルファベットコードも対応、日付はスラッシュアリも対応
input_data = """
3563	20260130
6869	20260130

"""
# ----------------

def normalize_date(date_str):
    current_year = datetime.datetime.now().year
    numbers = re.findall(r'\d+', date_str)
    
    if len(numbers) == 1 and len(numbers[0]) == 8:
        year = int(numbers[0][0:4])
        month = int(numbers[0][4:6])
        day = int(numbers[0][6:8])
    elif len(numbers) >= 2:
        if len(numbers[0]) == 4:
            year = int(numbers[0])
            month = int(numbers[1])
            day = int(numbers[2]) if len(numbers) > 2 else 1
        else:
            year = current_year
            month = int(numbers[0])
            day = int(numbers[1])
    else:
        raise ValueError(f"日付形式の解析に失敗しました: {date_str}")
        
    kabutan_date = f"{str(year)[2:4]}/{month:02d}/{day:02d}"
    output_date = f"{year}{month:02d}{day:02d}"
    
    return kabutan_date, output_date

def get_kabutan_data(code, target_date_formatted, session):
    company_name = "不明"
    close_price = None

    for page in range(1, 15):
        url = f"https://kabutan.jp/stock/kabuka?code={code}&ashi=day&page={page}"
        
        try:
            response = session.get(url, timeout=10)
            if response.status_code != 200:
                print(f"  [{code}] ページ{page}で取得失敗 (Status: {response.status_code})")
                break
                
            soup = BeautifulSoup(response.content, "html.parser")
            
            if company_name == "不明" and page == 1:
                name_element = soup.find("h2")
                if name_element:
                    company_name = name_element.text.strip().replace(str(code), "").replace("【株価】", "").strip()

            target_table = soup.find("table", class_="stock_kabuka_dwm")
            if not target_table:
                print(f"  [{code}] ページ{page}で時系列テーブルが見つかりません。探索を終了します。")
                break

            rows = target_table.find_all("tr")
            page_dates = [] 
            
            for row in rows:
                cells = row.find_all(["td", "th"])
                if len(cells) >= 5:
                    row_date = cells[0].text.strip()
                    page_dates.append(row_date)
                    
                    if target_date_formatted in row_date:
                        close_price = cells[4].text.strip().replace(",", "")
                        break
            
            if len(page_dates) > 1:
                print(f"    - ページ{page}: {page_dates[1]} 〜 {page_dates[-1]} を探索中")

            if close_price:
                print(f"    ★ ページ{page}で {target_date_formatted} の終値({close_price}円)を発見！")
                break 
                
            time.sleep(1.0)

        except Exception as e:
            print(f"  [{code}] ページ{page}でエラー: {e}")
            break

    return company_name, close_price

def main():
    results = []
    lines = [line.strip() for line in input_data.strip().split("\n") if line.strip()]
    
    session = requests.Session()
    session.headers.update({
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
    })
    
    print("スクレイピングを開始します...")
    for line in lines:
        parts = [p.strip() for p in re.split(r'[,\t]+', line) if p.strip()]
        if len(parts) < 2:
            parts = [p.strip() for p in re.split(r'\s+', line) if p.strip()]
            if len(parts) < 2:
                continue
            
        code, raw_date_str = parts[0], parts[1]
        
        try:
            kabutan_date, output_date = normalize_date(raw_date_str)
        except Exception as e:
            print(f"日付の変換に失敗したためスキップします ({raw_date_str}): {e}")
            continue
            
        print(f"データ取得中: コード {code} / 目的の日付: {kabutan_date}")
        
        company_name, close_price = get_kabutan_data(code, kabutan_date, session)
        
        results.append({
            "企業コード": code,
            "企業名": company_name,
            "終値": close_price if close_price else "取得失敗",
            "日付": output_date
        })
        
        time.sleep(1.0)
        
    df = pd.DataFrame(results)
    
    # --- 画面へのCSV形式出力 ---
    print("\n--- 出力結果 (CSV形式) ---")
    print(df.to_csv(index=False))
    
    # --- CSVファイルへの保存処理 ---
    output_dir = r"D:\株2025\株価スクレイピング\2026年"
    now = datetime.datetime.now()
    filename = now.strftime("%Y年%m月%d日_%H%M%S.csv")
    output_file = os.path.join(output_dir, filename)
    
    try:
        os.makedirs(output_dir, exist_ok=True)
        # utf-8-sig を指定することでエクセルで開いたときの文字化けを防ぐ
        df.to_csv(output_file, index=False, encoding="utf-8-sig")
        print(f"\n結果を '{output_file}' に保存しました。")
    except Exception as e:
        print(f"\nファイルの保存中にエラーが発生しました: {e}")

if __name__ == "__main__":
    main()