コンテンツへ移動します
カテゴリー:

python3 スクリプト

投稿日時:
投稿者:

特定の提出者の大量保有を取得するスクリプト

修正する場所は、Eから始まる提出者のコード、END_DATEの日付、およびdays=の日数

import requests
import os
import datetime
import time
import jpholiday

# ✅ APIキーをファイルから読み込み
with open("/Users/kusakaberyoma/Pythonテスト/EDINET/apikey.txt", "r") as f:
    API_KEY = f.read().strip()
print(f"✅ APIキー読み込み確認:{API_KEY[:5]}...")

# ✅ 提出者コードと日付範囲
EDINET_CODE = "E31883"
END_DATE = datetime.date(2025, 4, 3)
START_DATE = END_DATE - datetime.timedelta(days=1190)

# ✅ 保存先フォルダ名(1フォルダに統一)
folder_name = f"{START_DATE.strftime('%Y%m%d')}-{END_DATE.strftime('%Y%m%d')}_Oasis"
SAVE_DIR = f"/Users/kusakaberyoma/Pythonテスト/EDINET/Oasis/{folder_name}"
os.makedirs(SAVE_DIR, exist_ok=True)

# ✅ ループで各営業日を処理
current = START_DATE
while current <= END_DATE:
    if current.weekday() >= 5 or jpholiday.is_holiday(current):
        current += datetime.timedelta(days=1)
        continue

    TARGET_DATE = current.isoformat()
    print(f"\n📅 対象日:{TARGET_DATE}")

    # 書類一覧 API
    url = "https://api.edinet-fsa.go.jp/api/v2/documents.json"
    headers = {"Ocp-Apim-Subscription-Key": API_KEY}
    params = {"date": TARGET_DATE, "type": 2}

    response = requests.get(url, headers=headers, params=params)
    if response.status_code != 200:
        print(f"❌ 書類一覧取得失敗: {response.status_code}")
        current += datetime.timedelta(days=1)
        time.sleep(5)
        continue

    results = response.json().get("results", [])

    # キーワードでフィルタ
    keywords = ["大量保有", "変更報告書", "特例対象株券", "訂正報告書"]
    target_docs = [
        doc for doc in results
        if doc.get("edinetCode") == EDINET_CODE and any(kw in doc.get("docDescription", "") for kw in keywords)
    ]

    if not target_docs:
        print("📭 対象書類なし(該当提出者)")
    else:
        print(f"📄 該当書類数:{len(target_docs)} 件")
        for doc in target_docs:
            doc_id = doc["docID"]
            description = doc.get("docDescription", "No_Description").replace(" ", "_")
            filename = f"{doc_id}_{description}.zip"
            filepath = os.path.join(SAVE_DIR, filename)

            if os.path.exists(filepath):
                print(f"✅ スキップ(既に存在): {filename}")
                continue

            print(f"⬇️ ダウンロード中: {filename}")
            download_url = f"https://api.edinet-fsa.go.jp/api/v2/documents/{doc_id}"
            res = requests.get(download_url, headers=headers, params={"type": 1})

            if res.status_code == 200:
                with open(filepath, "wb") as f:
                    f.write(res.content)
                print(f"✅ 保存完了: {filename}")
            else:
                print(f"⚠️ ダウンロード失敗({doc_id}): {res.status_code}")

    # 5秒待機
    time.sleep(5)
    current += datetime.timedelta(days=1)

print("\n🎉 1年分のOasis提出書類のダウンロード完了!")

ZIP取得したフォルダを指定して、CSVにするスクリプト

現状1人の保有者(共同保有者なし)で、Oasisの大量保有報告では成功しているもの。

import os
import zipfile
from lxml import etree
import pandas as pd

# パスの定義
zip_dir = "/Users/kusakaberyoma/Pythonテスト/EDINET/Oasis/2020412-20250403_Oasis"  # ZIPが入っているフォルダ
extract_path = "/Users/kusakaberyoma/Pythonテスト/EDINET/Oasis/temp_extract"
output_csv = "/Users/kusakaberyoma/Pythonテスト/EDINET/Oasis/Oasis_大量保有.csv"

# 抽出対象項目
target_labels = {
    "報告義務発生日": "A",
    "提出日": "B",
    "証券コード": "C",
    "発行者の名称": "D",
    "提出者及び共同保有者の総数": "E",
    "変更報告書提出事由": "F",
    "氏名又は名称": "G",
    "保有株券等の数": "H",
    "上記提出者の株券等保有割合(%)": "I",
    "株券等保有割合": "J",
    "直前の報告書に記載された株券等保有割合": "K",
    "取得資金合計": "M"
}

xbrl_tags = {
    "氏名又は名称": "jplvh_cor:Name",
    "保有株券等の数": "jplvh_cor:TotalNumberOfStocksEtcHeld"
}

# 出力用リスト
all_data = []

# ZIPファイルを一つずつ処理
for zip_filename in os.listdir(zip_dir):
    if not zip_filename.endswith(".zip"):
        continue

    zip_path = os.path.join(zip_dir, zip_filename)

    # 解凍
    if os.path.exists(extract_path):
        for root, dirs, files in os.walk(extract_path, topdown=False):
            for f in files:
                os.remove(os.path.join(root, f))
            for d in dirs:
                os.rmdir(os.path.join(root, d))
    else:
        os.makedirs(extract_path)

    with zipfile.ZipFile(zip_path, "r") as zip_ref:
        zip_ref.extractall(extract_path)

    # 一つのファイルごとの抽出データ
    data = {label: "" for label in target_labels.keys()}

    # HTMLパースで取得(氏名・保有株券等を除く)
    for root_dir, _, files in os.walk(extract_path):
        for file in files:
            if file.endswith(".htm") or file.endswith(".xhtml"):
                filepath = os.path.join(root_dir, file)
                try:
                    parser = etree.HTMLParser()
                    tree = etree.parse(filepath, parser)
                    rows = tree.xpath("//tr")

                    for row in rows:
                        cells = row.xpath(".//td")
                        if len(cells) < 2:
                            continue
                        label_text = "".join(cells[0].itertext()).strip().replace("\u3000", "").replace(" ", "")

                        for key in target_labels:
                            if key in ["氏名又は名称", "保有株券等の数"]:
                                continue
                            if key in label_text:
                                value_text = "".join(cells[1].itertext()).strip().replace(" ", "")
                                data[key] = value_text
                except Exception as e:
                    print(f"⚠️ HTMLパースエラー: {file} — {e}")

    # XBRLタグから氏名・保有株券等の数を取得
    for root_dir, _, files in os.walk(extract_path):
        for file in files:
            if file.endswith(".htm") or file.endswith(".xhtml"):
                filepath = os.path.join(root_dir, file)
                try:
                    with open(filepath, "rb") as f:
                        parser = etree.XMLParser(recover=True)
                        tree = etree.parse(f, parser)
                        root = tree.getroot()
                        nsmap = root.nsmap

                        for label, tag in xbrl_tags.items():
                            result = root.xpath(f".//ix:nonNumeric[@name='{tag}'] | .//ix:nonFraction[@name='{tag}']",
                                                namespaces={"ix": nsmap.get("ix", "http://www.xbrl.org/2013/inlineXBRL")})
                            if result:
                                data[label] = result[0].text.strip()
                except Exception as e:
                    print(f"⚠️ XBRLパースエラー: {file} — {e}")

    # データ格納
    all_data.append(data)

# 保存(Excelで文字化けしないUTF-8 BOM付き)
df = pd.DataFrame(all_data)
df.to_csv(output_csv, index=False, encoding="utf-8-sig")
print(f"✅ 抽出完了: {output_csv}")

コメントを残す

メールアドレスが公開されることはありません。 が付いている欄は必須項目です