J-Quants と EDINET の使い方|Python で財務データ自動取得

Chelsea-Labs #13 サムネイル

免責事項

本記事は投資助言を目的としたものではなく、技術・分析手法の紹介です。記事中のコード・データ取得手順は教育目的であり、特定の銘柄・金融商品の売買を推奨するものではありません。投資判断はご自身の責任で行ってください。J-Quants API・EDINET API の利用規約は変更される可能性があるため、実装時は各APIの公式ドキュメント・利用規約を必ず確認してください。本記事中に J-Quants API から取得した実データは掲載していません(利用規約に基づく方針)。

前回の記事#12では、両学長の高配当株スクリーニング基準を「製品スペックシート」の枠組みで再構築し、6基準を Python の spec_sheet_judge.py ミニマム実装で判定する方法を整理しました。今回からは「その判定器に渡す財務データを、どう自動で集めてくるか」というデータパイプライン編に入ります。

応用編の最終的な目標は「東証上場約3,900社の財務データを毎週自動で取得し、両学長基準でスクリーニング・罠銘柄除外・業種分散まで考慮した投資候補リストを生成するパイプライン」です。これを実現するため、本記事では2つの公的・準公的データソースの使い方を Python で実装します。

  • J-Quants API(日本取引所グループの公式 API): 株価・配当・財務サマリを構造化データで一括取得
  • EDINET API(金融庁の公式 API): 有価証券報告書(有報)の XBRL データから財務指標を抽出

本記事の前提と難易度(#08 で前処理を学んだ方向け)

  • Python の基本(関数・dict・例外処理)と requests での HTTP 呼び出しが出来ると読み進めやすい
  • API 認証・レート制限・JSON/XBRL の取り扱いに慣れていない場合、基礎編 #08(リスクとリターン)基礎編 #03(複利のPython可視化)で Python 環境構築のステップを先に踏んでおくと無理なく進めます
  • 本記事のコードは「最小実装」。本格運用(全銘柄取得・夜間バッチ化)は応用編 #15・#20 で扱います

筆者は製造業の開発現場で、複数の社内システム(生産管理・品質管理・在庫管理)からデータを取得して統合分析する業務を多く担当してきました。「2つ以上のシステムからデータを取って統合する」業務は、製造業 DX の現場でも投資データパイプラインでも本質的に同じ構造で、形式の違い・タイミングの違い・欠損値の扱いという3つの落とし穴に共通してハマります。本記事ではこの「3つの罠」を軸に、ベテランから学んだ教訓も交えて整理します。

本記事では 4 個の Python スニペット(各15-25行程度)を通じて、認証→取得→保存の最小パイプラインを完成させます。本格的なデータ統合(JOIN・正規化)は次回 #14 で DuckDB を使って深掘りします。

動作環境

  • Python 3.11+
  • requests 2.31+ / pandas 2.x / duckdb 0.10+ / pyarrow 14+
  • J-Quants API のフリープラン以上(要:JPX のメールアドレス登録)。フリープランは12週間遅延・スクリーニング基本機能のみのため、応用編 #15 で全銘柄スクリーニングを実装する際は Light(月1,650円)以上が必要
  • EDINET API は完全無料(API キー登録のみ、即日発行されるが営業日扱いの場合あり)

サンプルコードは 架空の銘柄コード(XXXX)環境変数経由のダミー認証情報で記述しています。実際の取得データは各 API の利用規約に基づき本記事には掲載していません(J-Quants は再配布・記事内直接掲載を制限する利用規約があるため、ご自身の環境で実行してご確認ください)。

結論:高配当株スクリーニングのデータ層は、J-Quants(市場価格・財務サマリ)と EDINET(有価証券報告書)の組み合わせで構築できる。「2つのデータソースを取って統合する」のは製造業DXの典型パターンと同じで、形式統一・取得タイミング・欠損処理の3点を最初に設計しておけば、後の分析パイプラインが破綻するリスクが大きく低減する。本記事の4個のスニペットを動かせば、自分の開発環境でデータパイプラインの第1段階が完成する。

目次

2つのデータソースの役割分担と取得アーキテクチャ

J-Quants と EDINET は、同じ「日本株の財務データ」を扱いますが、出所と粒度が異なります。両者の役割分担を最初に整理しておくことが、パイプライン設計の出発点になります。

項目J-Quants APIEDINET API
提供元日本取引所グループ(JPX)金融庁
主な対象東証上場銘柄の株価・配当・財務サマリ有価証券報告書・四半期報告書(XBRL)
形式JSON(構造化済み)XBRL(タグ付きXML、要パース)
更新頻度日次(株価)/ 四半期(財務)提出ベース(不定期)
認証Refresh Token + ID Token(2段階)API キー(単一・ヘッダ渡し)
料金Free〜月額9,900円(プラン別、後述)完全無料
データの粒度サマリ・標準化済み(扱いやすい)全項目・原データ(粒度細かいが扱い重い)

J-Quants のプラン別機能差(応用編どこから Light 必須か)

J-Quants はプランによって取得できるデータ範囲が大きく異なります。応用編シリーズで「全銘柄スクリーニング」「過去履歴での検証」を回す段階で、フリープランでは限界が来ます(料金は2026年5月時点の参考値で、最新は公式公開情報をご確認ください)。

プラン料金(月額)株価データ財務データ応用編シリーズで使える範囲
Free無料12週間遅延・全銘柄主要決算サマリ(限定的)本記事の「動作確認・学習用」のみ
Light1,650円5営業日遅延・全銘柄主要決算サマリ応用編 #15 の全銘柄スクリーニングはここから可能
Standard3,300円遅延なし・全銘柄四半期含む詳細財務毎週運用+直近データに基づくバックテスト
Premium9,900円すべて+詳細全項目・特殊データ本格運用・収益化視野(個人開発では通常不要)

継続運用コストの選択肢比較(個人投資家視点)

  • SBI証券・楽天証券・マネックス証券のスクリーニングツール: 無料。両学長基準のような「6条件で絞り込み」レベルなら追加費用ゼロで実用可能。ただし API は無く、自前のパイプライン化はできない
  • KABU+ (kabu.plus): 月額約1,000円。CSV ベース・全銘柄財務データ。API ではなく定期 CSV 配信のため、本記事のような自動取得パイプラインには向かない
  • J-Quants Light(1,650円)+ EDINET(無料): 本シリーズの推奨組み合わせ。API ベース・公式・継続性重視。年間20,000円の投資で、両学長基準の自動化+拡張性を確保

「無料で始めたい」なら証券会社のスクリーニングツール、「自動化&カスタマイズしたい」なら J-Quants の流れが、ユースケース別の最適解です。

スクリーニング6基準と取得元の対応マトリクス

記事 #12 で整理した6基準のうち、どれが J-Quants で取れて、どれが EDINET 必須なのかを最初に整理しておきます。6個全部がJ-Quants 1本で揃うわけではありません(v1 では「4-5 個カバー」と書きましたが、実態は次の表の通り、より厳密には 3 指標相当)。

基準主要項目J-Quants(標準サマリ)EDINET(XBRL)
1. 配当利回り1株配当 / 株価◎(配当 + 株価)○(配当)
2. 配当性向配当 / 純利益○(配当 + 純利益)○(同上)
3. 自己資本比率純資産 / 総資産△(プラン依存)◎(必須・原典)
4. EPS1株純利益◎(EarningsPerShare)
5. 連続増配年数過去配当の単調増加判定◎(過去配当履歴)
6. 営業キャッシュフローキャッシュフロー計算書×(標準サマリには無い)◎(必須・原典)

つまり、営業CFと自己資本比率(厳密値)は EDINET の有報 XBRL から取らないと完結しません。J-Quants をメインに、EDINET で精度補完する設計が必然です。

エンジニア的に言い換えると

J-Quants と EDINET の使い分けは、製造業 DX で言う 「ETL(Extract-Transform-Load)のステージング層と原データ層」の関係です:

  • J-Quants = ステージング層(前処理済み・速度優先・整形コストが低い)
  • EDINET = 原データ層(生のXBRL・正確性優先・整形コストが高い)
  • 速報レポートはステージング層で十分、監査対応や原典確認のときだけ原データ層に降りる、という運用設計

両者の関係が見えていれば、「なぜ2つも使うのか」が腑に落ちます。本記事のスニペット2はステージング層、スニペット3は原データ層に対応します。

本記事のパイプライン全体像

┌─────────────────┐    ┌─────────────────┐
│  J-Quants API   │    │   EDINET API    │
│ (株価・財務サマリ) │    │ (有報 XBRL)     │
│ ステージング層    │    │ 原データ層       │
└────────┬────────┘    └────────┬────────┘
         │  JSON                 │  XBRL
         ▼                       ▼
   [認証 + 取得]            [文書一覧 + XBRL DL + パース]
         │                       │
         └───────────┬───────────┘
                     ▼
            [DuckDB に統合書き込み]
                     │
                     ▼
            stocks.duckdb(次回 #14 で深掘り)

スニペット1:J-Quants API の認証(Refresh Token → ID Token)

J-Quants は2段階認証を採用しています。まず JPX に登録したメール/パスワードで Refresh Token(有効期限 1 週間)を取得し、それを使って ID Token(有効期限 24 時間)を取得します。実際の API リクエストには ID Token を Authorization ヘッダで添付します。

# jquants_auth.py — J-Quants API 認証
# 動作環境: Python 3.11+ / requests 2.31+
import os
import requests

BASE_URL = "https://api.jquants.com/v1"

def get_refresh_token(email: str, password: str) -> str:
    """JPX 登録のメール・パスワードから Refresh Token(1週間有効)を取得"""
    resp = requests.post(
        f"{BASE_URL}/token/auth_user",
        json={"mailaddress": email, "password": password},
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()["refreshToken"]

def get_id_token(refresh_token: str) -> str:
    """Refresh Token から ID Token(24時間有効)を取得

    トークンを URL に入れずに params= 経由で渡すことで、ログ・履歴への漏洩を防ぐ
    """
    resp = requests.post(
        f"{BASE_URL}/token/auth_refresh",
        params={"refreshtoken": refresh_token},
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()["idToken"]

if __name__ == "__main__":
    # 認証情報は環境変数経由(直書きは絶対禁止)
    rt = get_refresh_token(os.environ["JQUANTS_EMAIL"], os.environ["JQUANTS_PASSWORD"])
    it = get_id_token(rt)
    print(f"ID Token (truncated): {it[:20]}...")  # 実値はログ出力しない

形式の罠:認証情報の取り扱い

  • メール・パスワード・Refresh Token をコードに直書きしない。環境変数か .env ファイル経由で渡す(リポジトリ混入の事故が多発)
  • v1 では Refresh Token を URL クエリ文字列に含めていましたが、params= 経由に変更しました。requests 内部では同じ動作ですが、ログ・履歴・プロキシキャッシュへの漏洩リスクが下がります
  • 本プロジェクトの .env.example が参考になります

運用上は、Refresh Token を週1回更新するスケジュールID Token を取得時に24時間キャッシュする仕組みを組み合わせると、毎回のリクエストで認証 API を叩く必要がなくなります(API レート制限の節約にもなる)。本記事では最小実装に留め、キャッシュ層は応用編 #15 で扱います。

スニペット2:J-Quants で株価と財務サマリを取得する(ステージング層)

ID Token を取得したら、それを Authorization ヘッダに乗せて株価データ/prices/daily_quotes)と財務サマリ/fins/statements)を取得します。両方とも JSON で返ってくるので、pandas で DataFrame 化が容易です。

形式の罠:日付フォーマットとカラム名のキー不整合

  • 日付パラメータは YYYYMMDD(ハイフン無し)YYYY-MM-DD でも受理される場合がありますが、エンドポイント・プランによっては 400 エラーになります。本コードでは YYYYMMDD に統一
  • カラム名のキーが2エンドポイントで異なる: /prices/daily_quotesCode(5桁含む)、/fins/statementsLocalCode(4桁または5桁)。次回 #14 の JOIN ではこの差を必ず正規化(例: 末尾 0 を統一)
  • ページネーション: 大量データ時は pagination_key が返るので、None になるまで繰り返す(下記コードに実装)
# jquants_fetch.py — J-Quants 株価・財務サマリ取得(ページネーション対応)
# 動作環境: Python 3.11+ / requests 2.31+ / pandas 2.x
import time
import requests
import pandas as pd

BASE_URL = "https://api.jquants.com/v1"

def _paged_get(url: str, headers: dict, params: dict, items_key: str) -> list[dict]:
    """pagination_key が返る限り取得を継続する汎用ヘルパ"""
    rows: list[dict] = []
    while True:
        resp = requests.get(url, headers=headers, params=params, timeout=60)
        resp.raise_for_status()
        body = resp.json()
        rows.extend(body.get(items_key, []))
        next_key = body.get("pagination_key")
        if not next_key:
            break
        params = {**params, "pagination_key": next_key}
        time.sleep(0.3)  # レート制限対策(プランごとの上限は公式参照)
    return rows

def fetch_daily_quotes(id_token: str, code: str, from_date: str, to_date: str) -> pd.DataFrame:
    """日次株価を取得。日付は YYYYMMDD(ハイフン無し)。code は 4桁/5桁のいずれも可"""
    headers = {"Authorization": f"Bearer {id_token}"}
    params = {"code": code, "from": from_date, "to": to_date}
    rows = _paged_get(f"{BASE_URL}/prices/daily_quotes", headers, params, "daily_quotes")
    return pd.DataFrame(rows)

def fetch_statements(id_token: str, code: str) -> pd.DataFrame:
    """決算財務サマリを取得(過去全期分、ページネーション対応)"""
    headers = {"Authorization": f"Bearer {id_token}"}
    params = {"code": code}
    rows = _paged_get(f"{BASE_URL}/fins/statements", headers, params, "statements")
    return pd.DataFrame(rows)

# 使用例(架空の銘柄コード XXXX)
# id_token = ...  # スニペット1で取得済み
# prices = fetch_daily_quotes(id_token, "XXXX", "20250101", "20260430")
# print(prices.shape, prices.columns.tolist())
# # Code, Date, Open, High, Low, Close, Volume, ... のキャメルケースで返る
#
# stmts = fetch_statements(id_token, "XXXX")
# print(stmts[["DisclosedDate","TypeOfDocument","NetSales","Profit","EarningsPerShare","LocalCode"]].head())
# # ⚠ statements は LocalCode、prices は Code でキー名が違う点に注意

J-Quants のレスポンスはすでにキャメルケース(PascalCase)で標準化されています(例: NetSalesProfitEarningsPerShare)。これが「ステージング層」たる所以で、後段の処理が楽になります。一方で先ほどのマトリクス表の通り、営業CF・自己資本比率の厳密値は標準サマリには含まれないため、次のスニペット3で EDINET から取得します。

タイミングの罠:API レート制限と全銘柄取得の設計

東証上場約3,900社のデータを一気に取りに行くと API レート制限に引っかかる可能性があります。実運用では:

  • 銘柄リストを取得/listed/info)して loop で1銘柄ずつ /fins/statements を叩く(プランによっては「全銘柄一括取得」エンドポイントも存在)
  • 各リクエスト間に time.sleep(0.3〜0.5) 程度を挟む(プランごとの制限値は公式ドキュメント参照)
  • 夜間バッチで一回取得し、Parquet ファイルにキャッシュ。日中の分析はキャッシュから読む

「全銘柄一気に取得 → 失敗したら再取得」という設計は規約・コスト両面で NG です。取得時刻は as_of_date としてデータに保持し、後段で「どの時点のデータか」を判断できるようにしておくのが、タイミング罠の構造的回避策です。

スニペット3:EDINET API で有報の XBRL を取得する(原データ層)

EDINET は金融庁が運営する電子開示システムです。API v2(https://api.edinet-fsa.go.jp/api/v2)で、(1) 文書一覧の取得 → (2) 個別文書の ZIP ダウンロード → (3) ZIP 内の XBRL ファイルをパースという3段階で財務データを取得します。原データ層に該当するため、扱いはやや重めです。

# edinet_fetch.py — EDINET 有報の XBRL 取得(API v2)
# 動作環境: Python 3.11+ / requests 2.31+
import os
import zipfile
import io
import requests

EDINET_BASE = "https://api.edinet-fsa.go.jp/api/v2"

def _headers(api_key: str) -> dict:
    """EDINET API v2 の認証ヘッダ(Subscription-Key はクエリではなくヘッダで渡すのが推奨)"""
    return {"Ocp-Apim-Subscription-Key": api_key}

def list_documents(date: str, api_key: str) -> list[dict]:
    """指定日に提出された全文書のメタデータ一覧を取得(YYYY-MM-DD)

    type=2 でメタデータ(書類詳細含む)が返る
    """
    params = {"date": date, "type": 2}
    resp = requests.get(f"{EDINET_BASE}/documents.json", headers=_headers(api_key),
                        params=params, timeout=60)
    resp.raise_for_status()
    return resp.json()["results"]

def download_xbrl(doc_id: str, api_key: str) -> bytes:
    """指定 docID の XBRL ZIPファイルをダウンロード(type=1: 提出本文書 + XBRL)"""
    params = {"type": 1}
    resp = requests.get(f"{EDINET_BASE}/documents/{doc_id}", headers=_headers(api_key),
                        params=params, timeout=120)
    resp.raise_for_status()
    return resp.content  # ZIP バイナリ

def extract_xbrl_files(zip_bytes: bytes) -> dict[str, bytes]:
    """ZIPの中から .xbrl ファイルを取り出す。EDINET ZIP は PublicDoc / AuditDoc / Summary の階層が
    含まれることがあるため、すべてのパスをそのまま辞書のキーとして保持する"""
    with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf:
        return {n: zf.read(n) for n in zf.namelist() if n.endswith(".xbrl")}

# 使用例
# api_key = os.environ["EDINET_API_KEY"]
# docs = list_documents("2026-04-15", api_key)
# # 有価証券報告書のみ(docTypeCode == "120")
# yuho_docs = [d for d in docs if d.get("docTypeCode") == "120"]
# print(f"提出された有報数: {len(yuho_docs)}")
#
# # 1件取得して .xbrl 抽出
# zip_bin = download_xbrl(yuho_docs[0]["docID"], api_key)
# xbrl_files = extract_xbrl_files(zip_bin)
# print(f"XBRL ファイル数: {len(xbrl_files)}")

XBRL は eXtensible Business Reporting Language の略で、財務データの各項目をタグ付けした XMLです。タクソノミ(タグ体系)が会計基準ごとに分かれており、たとえば:

  • jppfs_cor: 日本基準(個別財務諸表のコア・ラベル)— 多くの日本企業の有報で使われる
  • jpigp_cor: IFRS(International Financial Reporting Standards)採用企業向け
  • jpcrp_cor: 基準横断の「企業内容開示府令」サマリ用ラベル(営業利益・売上高など、複数基準を吸収する集約タグ)
  • us-gaap: 米国会計基準採用企業向け

会社ごとに採用する基準が異なるため、「会社ごとのタクソノミの違いを統一する前処理」が必須です。jpcrp_cor のサマリで間に合うケースもあれば、業種固有の項目で jppfs_cor / jpigp_cor の原タグまで降りる必要があるケースもあります。

欠損の罠:XBRL パース実装の落とし穴

  • タクソノミの違い: 上記 jppfs_cor / jpigp_cor / jpcrp_cor / us-gaap は同じ「売上高」でもタグ名が違う。マッピングテーブルを最低3パターン用意する
  • 勘定科目の階層: 「売上高」と「連結売上高」が両方タグ付けされていて、どちらを採用するかの判定ロジックが必要
  • 欠損項目: 業種特性(金融業など)で「営業 CF」が無く「経常損益」しか無いケースもある。欠損は NaN として扱い、後段の判定器側で業種補正する。0 で埋めるのは誤った解釈の温床
  • パース実装: 自前で書くより arelle(XBRL汎用パーサー、業界標準)か、より EDINET 特化で扱いやすい edinet-xbrl の利用を推奨。本記事ではタグ抽出までを示し、本格パースは応用編 #14 で扱う
  • 本プロジェクトの参考実装: samplecode/EDINET_Summary_App-main/ に EDINET ダウンローダの実装サンプルがあるので参照してください

スニペット4:取得データを DuckDB に統合書き込みする

取得した J-Quants の DataFrame と EDINET の財務指標を、DuckDB ファイル(stocks.duckdb)に書き込みます。DuckDB は Python から直接扱える組み込み型 OLAP(Online Analytical Processing、集計分析特化型)データベースで、Parquet・CSV・DataFrame をそのままテーブル化できるのが強みです。次回 #14 で複数テーブルを JOIN・正規化する基盤になります。

# save_to_duckdb.py — 取得データを DuckDB に統合保存
# 動作環境: Python 3.11+ / duckdb 0.10+ / pandas 2.x / pyarrow 14+
import duckdb
import pandas as pd
from pathlib import Path

DB_PATH = Path("data/stocks.duckdb")

def save_jquants_data(prices: pd.DataFrame, statements: pd.DataFrame) -> None:
    """J-Quants から取得した株価・財務サマリを DuckDB に保存(idempotent: 同一キーは置換)"""
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    with duckdb.connect(str(DB_PATH)) as conn:
        # DataFrame を SQL から参照可能にする(DuckDB は明示的な register が安全)
        conn.register("prices_df", prices)
        conn.register("statements_df", statements)

        # スキーマがなければ作成
        conn.execute("CREATE TABLE IF NOT EXISTS prices_daily AS SELECT * FROM prices_df LIMIT 0")
        conn.execute("CREATE TABLE IF NOT EXISTS fins_statements AS SELECT * FROM statements_df LIMIT 0")

        # idempotent な置換: 該当銘柄の既存行を削除してから挿入
        # ⚠ prices は Code、statements は LocalCode でキー名が違う点に注意
        conn.execute("""DELETE FROM prices_daily
                        WHERE Code IN (SELECT DISTINCT Code FROM prices_df)""")
        conn.execute("INSERT INTO prices_daily SELECT * FROM prices_df")

        conn.execute("""DELETE FROM fins_statements
                        WHERE LocalCode IN (SELECT DISTINCT LocalCode FROM statements_df)""")
        conn.execute("INSERT INTO fins_statements SELECT * FROM statements_df")

        # 後始末(同名 DataFrame で別テーブルを作るときの混線防止)
        conn.unregister("prices_df")
        conn.unregister("statements_df")

def save_edinet_metrics(metrics_df: pd.DataFrame) -> None:
    """EDINET から抽出した財務指標を保存(営業CF、自己資本比率など)"""
    with duckdb.connect(str(DB_PATH)) as conn:
        conn.register("metrics_df", metrics_df)
        conn.execute("CREATE TABLE IF NOT EXISTS edinet_metrics AS SELECT * FROM metrics_df LIMIT 0")
        conn.execute("INSERT INTO edinet_metrics SELECT * FROM metrics_df")
        conn.unregister("metrics_df")

# 使用例
# save_jquants_data(prices, stmts)        # スニペット2で取得済み
# save_edinet_metrics(edinet_metrics_df)  # スニペット3 + #14 のパース結果
#
# # 保存後の確認
# with duckdb.connect(str(DB_PATH)) as conn:
#     print(conn.execute("SELECT COUNT(*) FROM prices_daily").fetchone())
#     print(conn.execute("SELECT COUNT(*) FROM fins_statements").fetchone())

エンジニア的に言い換えると

DuckDB の選定は、製造業 DX で言うと 「Excel から Access へ移行する代わりに、Excel と SQL Server の中間として組み込み DB を選ぶ」ような判断です。クラウドの Snowflake / BigQuery は規模・運用コストが大きすぎ、SQLite は OLTP(Online Transaction Processing、トランザクション特化型)向けで OLAP 処理に向かない。個人開発・軽量分析・〜数百GBクラスのデータには DuckDB がスイートスポットです。Phase B(GCS = Google Cloud Storage)以降のクラウド化までは、ローカル DuckDB で十分機能します(CLAUDE.md の Phase A 設計と完全整合)。

idempotent(冪等:何回実行しても同じ結果になる性質)な書き込みのため、本コードは「既存キーを削除してから挿入」というパターンを採用しています。並行実行時のデッドロック対策には DuckDB のトランザクション(BEGIN; ... COMMIT;)で囲む設計を応用編 #15 で扱います。

プロセスFMEAでデータパイプラインの故障モード × 検出指標を整理する

製造業の品質工学では、生産プロセスのリスクを プロセスFMEA(Process Failure Mode and Effects Analysis、工程故障モード影響解析) で体系化します。記事 #12 では「6基準を故障モード × 検出指標として再整理」する形で FMEA を投資判定に応用しました。同じ考え方は、データパイプラインの工程にも完全に適用できます。本記事の認証・取得・保存の各工程で起こり得る故障モードを、検出指標とセットで整理しておくと、運用時のトラブルシュートと早期警報設計が圧倒的に楽になります。

故障モード影響度検出指標事前対策
認証失敗(Token 期限切れ)HTTP 401 / 403取得時に Token を24時間キャッシュ+期限前再取得
レート制限超過HTTP 429 / 取得失敗増加リクエスト間 sleep / 全銘柄取得は夜間バッチ
ページネーション取りこぼし件数の不整合pagination_key を None になるまでループ
形式の不整合(Code vs LocalCode)JOIN 結果が空 or 重複取得層で銘柄キーを正規化(4/5桁統一)
タイミングのズレ「最新」が日付ごとに違う各データに as_of_date を持たせる
欠損値の誤解釈致命的0 と NaN が混在欠損は NaN 保持。0 埋めは禁止
XBRL タクソノミ不一致タグが見つからないjppfs_cor / jpigp_cor / jpcrp_cor のマッピング表

このプロセスFMEAは、応用編 #20 のパイプライン全体像までずっと使い続ける運用ドキュメントです。「動かして壊れたら直す」のではなく「壊れる前に検出系を仕込む」のが、製造業 DX 由来の品質保証アプローチです。

設計判断の記録:パイプラインの選択ポイントとトレードオフ

判断1:なぜ KABU+ ではなく J-Quants を採用するか

類似サービスに KABU+(民間運営の有料データ提供サービス)もあります。本シリーズでは J-Quants を主軸に据えました:

  • 採用理由: 公式 API(JPX 直営)=データの一次性・継続性が安心 / REST + JSON で扱いやすい / フリープランあり / 規約が明確
  • 採用したことで失うもの: KABU+ には独自集計指標(過去20年のクオリティスコアなど)があり、それらは取得できない。また CSV 一括ダウンロードの便利さも犠牲になる(J-Quants は API 個別取得が基本)。これらが必要になる将来の拡張時には KABU+ を補助的に組み合わせる前提

判断2:なぜ yfinance / Alpha Vantage を使わないか

yfinance は手軽ですが、Yahoo! Finance の利用規約変更や非公式スクレイピング扱いに左右されやすく、継続運用には不安定です。Alpha Vantage は無料枠の制限が厳しく、料金対効果が悪い。

  • 採用しないことで失うもの: yfinance の手軽さ(pip install 直後に動く)と、Alpha Vantage の経済指標データ。学習・プロトタイピング段階では yfinance 併用も検討余地あり(基礎編 #04 では実際に yfinance を使用)
  • 応用編で使わない理由: 「公式 API を選ぶ」を最初に固めると、規約変更やスクレイピング規制で全体が止まるリスクが減る

判断3:データを Parquet キャッシュ + DuckDB で扱う理由

取得した JSON をそのまま JSON ファイルで保存することもできますが、本シリーズでは Parquet(圧縮済みカラムナフォーマット = 列指向ストレージ)+ DuckDB の SQL クエリを採用します。

  • 採用理由: Parquet の圧縮効率(同じデータが JSON 比 5〜10 倍コンパクト)/ DuckDB の SQL クエリ速度(複数銘柄横断の集計が pandas より速い)/ Phase B 以降のクラウド化でも同じ Parquet が使える
  • 採用したことで失うもの: JSON のテキスト可読性(grep やエディタでの直接確認が困難)。デバッグ時のみ conn.execute("SELECT ... LIMIT 5").fetchdf() で覗く運用が必要。pyarrow / duckdb への依存が増える点もコストとして認識

本業の話:3つの社内システム統合で生産指示を誤発行した失敗談

筆者が製造業で 生産計画の最適化システムを構築する業務を担当していたとき、3つの社内システムからデータを統合する必要がありました——生産管理システム(出力は CSV、項目名は日本語)、品質管理システム(出力は XML、欧米の親会社由来で項目名は英語)、在庫管理システム(出力は Excel ファイル、形式は人手で月ごとに微妙に変わる)。

初稿の実装では、各システムからのデータをそのまま結合する方針で書き始めました。2週間は順調に動いていたものの、3週目で1週間分の生産指示の数値が誤発行される事態が発生し、本業現場に迷惑をかけました。具体的に何が起きたかの時系列:

  • 初日〜2週目: 3 システムのデータが「同じ日付・同じ単位」で来ていたため、何の問題も起きず動作(テストでも気づけない)
  • 3週目開始: 在庫管理 Excel の月次更新で、列順が変わった + 「在庫数 = NaN」が 0 として読み込まれた
  • 当日中: 「在庫切れ → 至急生産」の指示が約30品目で誤発行
  • 翌日〜1週間: 過剰生産の発覚 → 緊急在庫差し替え・原因調査・現場説明
  • 復旧後の改修: 取得層・統合層・分析層の3層分離アーキテクチャに書き直し、欠損は NaN として保持する設計に統一。改修後は同種のトラブル発生率が約8割減

レビュー会議で、ベテランエンジニアからの指摘:

  • 取得層と統合層を分けろ。各システムから取ったデータを共通の中間形式(DataFrame or Parquet)に揃える層を作ってから、初めて結合する」
  • タイミングのズレは設計レベルで吸収しろ。「最新の生産計画」と「最新の品質データ」は別の日付になる前提で、各データに as_of_date を持たせて使う側が判断する」
  • 欠損は欠損として保持しろ0 で埋めるのは「データがない=0」と解釈してよいときだけ。それ以外は NaN のまま、判定器側で業種補正する」

この経験が、本記事のパイプライン設計に直接反映されています:

  • J-Quants と EDINET の取得層を完全に分離(スニペット2 と スニペット3 が独立)
  • 両者を DuckDB の別テーブルに保存してから結合(スニペット4)。結合は次回 #14 で深掘り
  • EDINET の欠損財務項目(業種特性で存在しないもの)は NaN のまま保持、判定器で業種補正

取得・統合・分析の3層を最初から分ける」というアーキテクチャは、製造業の社内システム統合でも投資データパイプラインでも全く同じです。応用編 #20 のパイプライン全体像でも、この3層構造が骨格として残ります。

まとめ:データソースの選定と「3層分離」を最初に決めれば、自動化は破綻リスクが大きく低減する

  • J-Quants は速報レイヤー(ステージング層)、EDINET は精密レイヤー(原データ層)として使い分ける。週次のスクリーニングは J-Quants で完結、原典確認・詳細分析は EDINET の XBRL に降りる。指標 ↔ ソース対応マトリクスを最初に持っておけば、どこに何を取りに行くかで迷わない
  • 取得層・統合層・分析層を最初から分ける。各データソースから取得 → 共通形式に揃えて DuckDB に保存 → JOIN・正規化は次の段で行う、というレイヤー分離が破綻防止の本質。プロセスFMEA で工程ごとの故障モード × 検出指標を最初から書いておくと、自動化運用の安心感が桁違い
  • 形式・タイミング・欠損の3つの罠を設計レベルで吸収する。XBRL のタクソノミ違いは前処理で統一、データ更新時期のズレは as_of_date を持たせる、欠損は NaN として保持して判定器で業種補正

今日からできる3つのアクション(自動化のスタート)

  1. J-Quants API のフリープラン登録EDINET API キー発行を済ませる。両方とも数分〜営業日で完了。.envJQUANTS_EMAIL / JQUANTS_PASSWORD / EDINET_API_KEY を保存(リポジトリには絶対コミットしない)
  2. 本記事のスニペット1〜4を順に実行し、自分の興味のある銘柄1〜2社の株価・財務サマリ・有報 XBRLを取得 → data/stocks.duckdb に保存。次に SQL で確認(例: SELECT Code, Date, Close FROM prices_daily ORDER BY Date DESC LIMIT 10)。J-Quants のレスポンス構造(カラム名)と EDINET XBRL のタグ体系を実物で確認するのが、応用編 #14 以降の理解の土台になります
  3. 取得した1〜2社のデータを記事 #12 の spec_sheet_judge.py に流し込み、「PASS / CAUTION / FAIL」の判定を自動化してみる。この時点で「自動化された両学長基準スクリーニング」のミニマム版が完成します(残るのは全銘柄展開と業種補正で、それぞれ #14 / #15 / #18 で扱います)

次回予告:DuckDB でデータを統合して、両学長基準で計算可能な状態にする

次回(記事#14)では、本記事で取得した J-Quants データと EDINET 財務指標を DuckDB で結合・正規化します。スキーマ設計、JOIN 戦略、欠損値の扱い、業種別の代替指標マッピング——「分析に使える形に整える」フェーズの本番です。

  • 銘柄マスタ・株価・財務サマリ・EDINET 財務指標の4テーブルを設計するスキーマ設計図
  • 「業種別代替指標」(銀行は BIS 比率、REIT は LTV)をどう DuckDB で扱うかの設計
  • 記事#12 の spec_sheet_judge.py をそのまま使える状態のデータマートを作る

「製品開発DXエンジニアの投資術」シリーズ全体像

本記事は 応用編(記事#11〜#20)の第3回 です。応用編の DX フェーズマップでの位置づけ:

  • 導入・基準設計#11 なぜ高配当株か#12 6基準のスペックシート
  • Phase 1: 収集▶ イマココ #13 J-Quants・EDINETでデータ取得(本記事)
  • Phase 2: 前処理:#14 DuckDB でデータ統合
  • Phase 3: 分析:#15 両学長基準で Python スクリーニング → #16 配当推移の安定性 → #17 罠銘柄検知
  • Phase 4: 可視化/運用:#18 財務健全性の可視化 → #19 業種分散の FMEA
  • まとめ:#20 パイプライン全体像 + 発展編接続

前回 #12 スクリーニング基準本記事 #13 データ取得 | 次回 #14(公開予定)

関連記事(基礎編から): #03 複利のPython可視化#08 リスクとリターンの関係#09 NISA・iDeCoの設計

免責事項(再掲)

本記事は投資助言を目的としたものではなく、技術・分析手法の紹介です。コード・取得手順は教育目的であり、特定の銘柄・金融商品の売買を推奨するものではありません。投資判断はご自身の責任で行ってください。J-Quants API・EDINET API の利用規約・料金プラン・機能は変更される可能性があるため、実装時は各APIの公式ドキュメント・利用規約を必ず確認してください。本記事中に J-Quants API から取得した実データは掲載していません(利用規約に基づく方針)。

よかったらシェアしてね!
  • URLをコピーしました!
  • URLをコピーしました!

この記事を書いた人

コメント

コメントする

CAPTCHA


目次