J-Quants APIで製造メーカー株のMACDを自動チェックする【Python実装・トヨタ・日立対応】

トヨタの決算前後で動きに乗り遅れたことが何度もありました。子供が寝た後にスマホで確認しても、そのときにはもう終わっている。だから「製造メーカー株のMACDが動いたら自動で通知してくれる仕組み」を作ることにしました。

最初はyfinanceで組んでいたのですが、日本株では取得できない日があったり、値が微妙にずれたりという問題に何度かぶつかりました。そこで乗り換えたのが、日本取引所グループが公式に提供しているJ-Quants APIです。

この記事では、データソースとしてのJ-Quants APIの使い方を中心に、MACDの自動チェック機能を作るまでを解説します。yfinanceとの違いや使い分けにも触れるので、日本株を扱うなら知っておいて損はない内容です。

※本記事のコードや情報は執筆時点の仕様に基づいています。投資は自己責任であり、必ずデモ環境や少額資金でテストした上で運用してください。

なぜJ-Quants APIを使うのか

結論から言うと、「日本株のデータを、公式が保証する形で取れる」という一点に尽きます。J-Quantsは東京証券取引所を運営するJPXグループが提供しているサービスです。

	yfinance	J-Quants API
提供元	非公式（スクレイピング的）	JPXグループ公式
費用	無料	無料プランあり
登録	不要	必要
日本株の網羅性	△ 取れない銘柄がある	◎ 全上場銘柄
財務データ	△ 断片的	◎ 決算情報が充実
過去データ	長期間	無料プランは約2年分
速報性	比較的早い	遅延あり
安定性	△ 仕様変更が多い	◎ 公式なので安定

使い分けの結論は明確です。

• とりあえず試したい・米国株も見る → yfinance
• 日本株を網羅的に・決算データも使う → J-Quants API
• リアルタイムで売買判断したい → どちらも不向き。証券会社のAPIを使う

3番目は重要です。J-Quantsのデータには遅延があるため、ザラ場中のリアルタイム判断には使えません。大引け後の分析や、翌日の戦略立案が主な用途になります。

データ取得ツール全体の比較はPython株価データ取得ツール徹底比較にまとめています。

僕が実際に乗り換えを決めた直接のきっかけは、yfinanceで特定の銘柄だけデータが取れない日があったことでした。監視対象6銘柄のうち1つが空で返ってきて、しかもエラーではなく空のDataFrameなので気づきにくい。

毎日動かす仕組みでこれをやられると、「シグナルが出なかった」のか「データが取れなかった」のかが判別できません。公式提供のAPIに移せば、少なくともデータの欠落は自分の実装の問題に絞り込めます。

📘 外部参考：J-Quants 公式サイト ／ J-Quants API ドキュメント

登録と認証

利用には登録が必要です。無料プランでも過去2年分のデータが使えるので、個人の分析用途なら十分に実用的です。

• 公式サイトから利用申し込みを行う
• プランを選択する（まずは無料プランで問題ない）
• ダッシュボードからAPIキーを発行する

ここで1点、注意が必要です。ネット上の古い記事には「リフレッシュトークンを取得して…」という手順が載っていますが、これは旧バージョンの方式です。現在はダッシュボードで発行するAPIキーに統一されています。

認証で詰まったら、まず公式ドキュメントの最新版を確認してください。ここは仕様が変わりやすい部分です。

# .env に保存する（Gitには絶対に上げない）
JQUANTS_API_KEY=your_api_key_here
DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/xxxxx

APIキーの管理方法はPython自動売買の開発環境セットアップ完全版にまとめています。コードへの直書きは絶対に避けてください。

無料プランで何ができるのかも、先に把握しておくと計画が立てやすくなります。

• 日次株価: 全上場銘柄。過去約2年分
• 上場銘柄一覧: 銘柄コード・名称・業種・市場区分
• 財務情報: 決算発表日と主要な財務数値
• 取引カレンダー: 営業日と休場日の判定に使える

日々のテクニカル監視という用途であれば、無料プランで必要なものはほぼ揃います。有料プランが必要になるのは、長期のバックテストや、より詳細なデータを扱う段階に入ってからです。

株価データを取得する

pip install requests pandas numpy python-dotenv

APIを直接叩く形で実装します。専用のクライアントライブラリもありますが、素のrequestsで書いておくと仕様変更に対応しやすいという利点があります。

import os
import datetime as dt

import requests
import pandas as pd
import numpy as np
from dotenv import load_dotenv

load_dotenv()

BASE_URL = "https://api.jquants.com/v1"
API_KEY = os.getenv("JQUANTS_API_KEY", "")


def _headers():
    if not API_KEY:
        raise RuntimeError("JQUANTS_API_KEY が設定されていません")
    return {"Authorization": f"Bearer {API_KEY}"}


def fetch_prices(code, days=200):
    """日次株価を取得する

    code : 銘柄コード（例: "72030" または "7203"）
    """
    end = dt.date.today()
    start = end - dt.timedelta(days=days)

    params = {
        "code": code,
        "from": start.strftime("%Y-%m-%d"),
        "to": end.strftime("%Y-%m-%d"),
    }

    rows, pagination_key = [], None
    while True:
        if pagination_key:
            params["pagination_key"] = pagination_key

        res = requests.get(f"{BASE_URL}/prices/daily_quotes",
                           headers=_headers(), params=params, timeout=30)
        res.raise_for_status()
        data = res.json()

        rows.extend(data.get("daily_quotes", []))
        pagination_key = data.get("pagination_key")
        if not pagination_key:
            break

    if not rows:
        return pd.DataFrame()

    df = pd.DataFrame(rows)
    df["Date"] = pd.to_datetime(df["Date"])
    df = df.set_index("Date").sort_index()

    # 分割調整済みの列を優先して使う
    for col in ("AdjustmentClose", "Close"):
        if col in df.columns:
            df["close"] = pd.to_numeric(df[col], errors="coerce")
            break

    return df.dropna(subset=["close"])

実装で押さえるべき点が2つあります。

• ページネーション対応: 期間が長いとデータが分割して返ってくるので、pagination_key がなくなるまで繰り返す
• 調整済み株価を使う: AdjustmentClose は株式分割を調整済み。生の Close を使うと分割日に大暴落が記録される

2番目を忘れると、テクニカル指標がめちゃくちゃになります。株式分割は日本株でも頻繁に行われるので、必ず調整済みの列を使ってください。

MACDを計算する

取得したデータからMACDを計算します。ライブラリを使わなくても、pandasだけで数行で書けます。

def macd(close, fast=12, slow=26, signal=9):
    """MACD・シグナル線・ヒストグラムを返す"""
    ema_fast = close.ewm(span=fast, adjust=False).mean()
    ema_slow = close.ewm(span=slow, adjust=False).mean()

    macd_line = ema_fast - ema_slow
    signal_line = macd_line.ewm(span=signal, adjust=False).mean()
    histogram = macd_line - signal_line

    return pd.DataFrame({
        "macd": macd_line,
        "signal": signal_line,
        "hist": histogram,
    })


def detect_cross(hist):
    """ヒストグラムの符号が変わった瞬間を検出する"""
    golden = (hist > 0) & (hist.shift(1) <= 0)   # 好転
    dead = (hist < 0) & (hist.shift(1) >= 0)    # 悪化
    return golden, dead

ここでも「状態」ではなく「変化」を見ているのが重要です。「ヒストグラムがプラス」を条件にすると毎日通知が飛びますが、「マイナスからプラスに転じた瞬間」なら数週間に1回程度に収まります。

MACDの読み方そのものはMACDを計算してトレンド転換シグナルを検知する方法で詳しく扱っています。

複数銘柄をまとめてチェックする

製造メーカー株を対象に、まとめて確認する形にします。

import time

MAKERS = {
    "7203": "トヨタ自動車",
    "6501": "日立製作所",
    "6902": "デンソー",
    "6367": "ダイキン工業",
    "6301": "小松製作所",
    "7011": "三菱重工業",
}


def check_all(codes, sleep=0.5):
    """全銘柄のMACDを確認して、変化があったものを返す"""
    signals, errors = [], []

    for code, name in codes.items():
        try:
            df = fetch_prices(code)
            if len(df) < 60:
                errors.append(f"{name}: データ不足 ({len(df)}件)")
                continue

            m = macd(df["close"])
            golden, dead = detect_cross(m["hist"])

            latest_close = float(df["close"].iloc[-1])
            latest_hist = float(m["hist"].iloc[-1])

            if bool(golden.iloc[-1]):
                signals.append({
                    "銘柄": name, "コード": code, "種別": "好転",
                    "終値": latest_close, "ヒストグラム": latest_hist,
                })
            elif bool(dead.iloc[-1]):
                signals.append({
                    "銘柄": name, "コード": code, "種別": "悪化",
                    "終値": latest_close, "ヒストグラム": latest_hist,
                })

        except requests.HTTPError as e:
            errors.append(f"{name}: HTTP {e.response.status_code}")
        except Exception as e:
            errors.append(f"{name}: {e}")

        time.sleep(sleep)      # APIへの負荷を抑える

    return signals, errors


signals, errors = check_all(MAKERS)
print(f"シグナル: {len(signals)} 件 / エラー: {len(errors)} 件")
for s in signals:
    print(f"  {s['銘柄']}: {s['種別']} (終値 {s['終値']:,.0f}円)")

time.sleep() を挟んでいるのは、短時間に大量のリクエストを送らないためです。APIには利用上の制限があるので、余裕を持った間隔で呼ぶのがマナーであり、自分のためでもあります。

また、1銘柄のエラーで全体を止めない設計にしています。エラーは errors に集めて後でまとめて報告する形です。

例外の捕まえ方を2段階にしているのにも理由があります。

• requests.HTTPError: ステータスコードが分かるので、認証切れ（401）と一時的な障害（500系）を区別できる
• それ以外の Exception: データ形式の変化など、想定外の問題を拾う

この区別があると、通知を見ただけで「自分が直すべきか、待てば直るか」が判断できます。401が並んでいればAPIキーの問題、503が散発するならサーバー側の一時的な不調、という具合です。

通知して仕上げる

WEBHOOK_URL = os.getenv("DISCORD_WEBHOOK_URL", "")


def notify(signals, errors):
    """結果をまとめて1通で送る"""
    if not WEBHOOK_URL:
        print("通知先が未設定です")
        return

    if signals:
        lines = []
        for s in signals:
            mark = "📈" if s["種別"] == "好転" else "📉"
            lines.append(
                f"{mark} **{s['銘柄']}** ({s['コード']})  "
                f"{s['終値']:,.0f}円  hist={s['ヒストグラム']:+.2f}"
            )
        title = f"MACD変化 {len(signals)}件"
        color = 0x2ECC71 if any(s["種別"] == "好転" for s in signals) else 0xE74C3C
    else:
        lines = ["本日は変化なし"]
        title = "✅ チェック完了"
        color = 0x95A5A6

    if errors:
        lines.append("")
        lines.append("⚠️ **エラー**")
        lines.extend(f"・{e}" for e in errors)

    embed = {
        "title": title,
        "description": "\n".join(lines),
        "color": color,
        "footer": {"text": f"{dt.datetime.now():%Y-%m-%d %H:%M} 実行"},
    }
    try:
        requests.post(WEBHOOK_URL, json={"embeds": }, timeout=15)
    except requests.RequestException as e:
        print(f"通知失敗: {e}")


if __name__ == "__main__":
    signals, errors = check_all(MAKERS)
    notify(signals, errors)

設計上のポイントは「シグナルが0件でも必ず送る」ことです。何も届かない日があると、システムが止まっているのか本当に変化がないのかが判別できません。

もう1つ、複数の変化を1通にまとめて送っているのも意図的です。銘柄ごとに個別通知すると、6銘柄で6回スマホが鳴ります。まとめれば1回で済み、全体像も把握しやすくなります。

通知の実装詳細はDiscordにアラートをPythonで送る方法にまとめています。あとはcronで毎日自動実行する設定を入れれば完成です。

J-Quantsならではの使い方

ここからが、yfinanceではできない部分です。J-Quantsは決算情報や銘柄一覧も取得できるので、株価だけの分析から一歩進めます。

決算発表日を取得して回避する

テクニカル指標は、決算をまたぐとまったく通用しなくなります。発表日を事前に知っておけば、その前後のシグナルを無効化できます。

def fetch_statements(code):
    """財務情報（決算）を取得する"""
    res = requests.get(f"{BASE_URL}/fins/statements",
                       headers=_headers(), params={"code": code}, timeout=30)
    res.raise_for_status()

    rows = res.json().get("statements", [])
    if not rows:
        return pd.DataFrame()

    df = pd.DataFrame(rows)
    if "DisclosedDate" in df.columns:
        df["DisclosedDate"] = pd.to_datetime(df["DisclosedDate"])
        df = df.sort_values("DisclosedDate")
    return df


def near_earnings(code, days=3):
    """直近の決算発表から days 日以内かを判定する"""
    df = fetch_statements(code)
    if df.empty or "DisclosedDate" not in df.columns:
        return False

    last = df["DisclosedDate"].max()
    elapsed = (pd.Timestamp.today().normalize() - last).days
    return 0 <= elapsed <= days

これを使って、決算直後の銘柄はシグナルを出さないという条件を加えられます。「テクニカルは好転しているのに、実は決算が悪くて下げている」という罠を避けられます。

上場銘柄一覧から対象を絞り込む

銘柄コードを手打ちで管理するのは限界があります。業種で絞り込んで自動的にリストを作るほうが実用的です。

def fetch_listed():
    """上場銘柄一覧を取得する"""
    res = requests.get(f"{BASE_URL}/listed/info",
                       headers=_headers(), timeout=30)
    res.raise_for_status()
    return pd.DataFrame(res.json().get("info", []))


def filter_by_sector(keyword, market=None):
    """業種名で銘柄を絞り込む"""
    df = fetch_listed()
    if df.empty:
        return df

    sector_col = next(
        (c for c in df.columns if "Sector" in c and "Name" in c), None)
    if not sector_col:
        return df

    result = df[df[sector_col].astype(str).str.contains(keyword, na=False)]

    if market:
        market_col = next(
            (c for c in df.columns if "Market" in c and "Name" in c), None)
        if market_col:
            result = result[result[market_col].astype(str)
                            .str.contains(market, na=False)]
    return result


# 例: 輸送用機器（自動車関連）の銘柄を抽出する
makers = filter_by_sector("輸送用機器")
print(f"該当銘柄: {len(makers)} 件")

これで「東証プライムの輸送用機器を全部監視する」といった運用ができるようになります。yfinanceでは銘柄リスト自体を自分で用意する必要があるので、ここは明確な優位点です。

つまずきやすいところ

認証エラーが出る

最も多いトラブルです。原因は限られています。

def diagnose():
    """認証まわりを診断する"""
    if not API_KEY:
        print("✗ APIキーが読み込めていません")
        print("  → .env の場所と load_dotenv() を確認してください")
        return

    print(f"✓ APIキーを読み込みました（先頭6文字: {API_KEY[:6]}...）")

    try:
        res = requests.get(f"{BASE_URL}/listed/info",
                           headers=_headers(), timeout=20)
        if res.status_code == 200:
            print("✓ 認証成功")
        elif res.status_code == 401:
            print("✗ 認証失敗（401）")
            print("  → キーが無効か、古い認証方式のコードを使っています")
        elif res.status_code == 403:
            print("✗ 権限なし（403）")
            print("  → プランで許可されていないデータかもしれません")
        else:
            print(f"✗ HTTP {res.status_code}: {res.text[:200]}")
    except requests.RequestException as e:
        print(f"✗ 通信エラー: {e}")


diagnose()

401が返る場合、古い記事を参考にしてリフレッシュトークン方式で実装している可能性が高いです。現在の方式は公式ドキュメントで確認してください。

銘柄コードの桁数

J-Quantsでは5桁のコード（末尾に0を付けた形）が使われることがあります。4桁で指定して結果が空になる場合は、こちらを試してください。

def normalize_code(code):
    """4桁・5桁のどちらでも扱えるようにする"""
    code = str(code).strip()
    return code if len(code) == 5 else code + "0"


for c in ("7203", "72030"):
    print(f"{c} → {normalize_code(c)}")

無料プランの期間制限

無料プランで取得できるのは約2年分です。長期のバックテストをしたい場合は、有料プランを検討するか、yfinanceと併用することになります。

ただし、MACDのようなテクニカル指標の日常的な監視であれば2年分でまったく問題ありません。用途によって使い分けるのが現実的です。

まとめ

J-Quants APIを使って製造メーカー株のMACDを自動チェックする仕組みをまとめました。

• J-QuantsはJPXグループ公式。日本株の網羅性と決算データが強み
• ただし遅延があるためリアルタイム売買には不向き。大引け後の分析向け
• 認証はダッシュボードで発行するAPIキー。古い記事のトークン方式は使えない
• 株価は調整済みの列を使う。生の終値だと分割日に大暴落が記録される
• MACDは「状態」ではなく「符号が変わった瞬間」を検知する
• 複数の変化は1通にまとめて送る。0件でも稼働報告として送る
• 決算発表日と銘柄一覧が取れるのがyfinanceにない強み

乗り換えて一番良かったのは、「データが取れなかった」という理由でシステムが止まらなくなったことでした。公式提供という安心感は、毎日動かす仕組みを作るうえで想像以上に価値があります。

一方で、登録が必要なぶん最初のハードルはあります。まずはyfinanceで仕組みを作り、日本株を本格的に扱うようになってから移行する——という順序が、無理がなくておすすめです。

移行の際は、データ取得部分を関数に切り出しておくと作業が楽になります。fetch_prices() の中身だけ差し替えれば、指標の計算や通知のコードはそのまま使えるからです。データソースは今後も変わりうるものだと考えて、依存する部分を1か所に閉じ込めておく設計にしておくことをおすすめします。