カテゴリー: データ
python3 スクリプト
投稿日時:
投稿者: ry
コメント数: コメントはありません
特定の提出者の大量保有を取得するスクリプト
修正する場所は、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}")