【LINE以外】株価シグナルをDiscordに自動送信するPythonコード

自動化・運用

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

株価の売買シグナルをプログラムで検出するところまでは実装できた、という段階にいる方は多いはずです。

しかし、シグナルが発生しても気づけなければ意味がありません。リアルタイムに近い通知手段を確保することが、自動売買・裁量判断のどちらにおいても不可欠です。

これまで個人開発者の定番だったLINE Notifyは、2025年3月末でサービスを終了しました。「通知手段がなくなった」「代替を探しているが設定が難しそう」という声が後を絶ちません。

その原因は、多くの解説記事がLINE Notify前提で書かれており、乗り換え先の具体的なコードが不足している点にあります。

本記事では、Discord(ディスコード)のWebhook(ウェブフック)機能を使い、株価シグナルを自動送信するPythonコードを提供します。Webhookの作成手順から、移動平均クロスの検出、リッチな埋め込みメッセージ(Embed)の送信まで、すべてコピペで動く形にまとめました。

コードはそのまま動かした後、自分のシグナルロジックに差し替えて応用してください。

Discord Webhookによる通知の仕組み

Webhookとは何か

Webhook(ウェブフック)とは、外部のプログラムから特定のURLにHTTPリクエストを送るだけでメッセージを投稿できる仕組みです。Discordではサーバーのチャンネルごとに専用のWebhook URLを発行できます。

Botアカウントの作成やOAuth認証は不要です。URLさえあればPythonのrequestsライブラリ1つで通知を送れるため、個人用途の通知システムに最適です。

LINE Notifyとの違い

LINE Notifyはトークン発行後にLINEグループへ通知を送る仕組みでした。Discord Webhookとの主な違いを整理します。

項目 LINE Notify(終了済) Discord Webhook
認証方式 アクセストークン Webhook URL
メッセージ装飾 テキストのみ Embed(色・フィールド・画像対応)
レート制限 1,000回/時 30回/分(チャンネル単位)
料金 無料 無料
サービス継続性 2025年3月終了 Discord本体機能のため安定

Discord Webhookはメッセージの装飾性が高く、銘柄名・価格・シグナル種別を色分きで表示できます。通知としての視認性はLINE Notifyを上回ります。

Webhook URLの取得手順

Discordサーバーの管理権限が必要です。以下の手順でURLを取得してください。

  1. Discordでサーバーを作成する(既存サーバーでも可)
  2. 通知用チャンネルを作成し、歯車アイコンから「チャンネルの編集」を開く
  3. 左メニュー「連携サービス」→「ウェブフック」→「新しいウェブフック」を押す
  4. 名前を設定し「ウェブフックURLをコピー」を押す

取得したURLはhttps://discord.com/api/webhooks/で始まる長い文字列です。このURLが漏れると第三者がメッセージを送信できるため、公開リポジトリに直接書き込むことは危険です。環境変数か.envファイルで管理してください。

【コピペOK】株価シグナルをDiscordに送信するPythonコード

まず必要なライブラリをインストールします。


pip install requests yfinance pandas

以下がメインコードです。


import os
import json
from datetime import datetime, timedelta
from typing import Optional

import requests
import yfinance as yf
import pandas as pd

# ==============================
# 設定エリア
# ==============================
WEBHOOK_URL: str = os.environ.get(
    "DISCORD_WEBHOOK_URL",
    "https://discord.com/api/webhooks/YOUR_WEBHOOK_ID/YOUR_WEBHOOK_TOKEN",
)
TICKER_SYMBOL: str = "7203.T"  # トヨタ自動車(Yahoo Finance形式)
TICKER_LABEL: str = "トヨタ自動車(7203)"
SHORT_WINDOW: int = 5    # 短期移動平均の期間(日)
LONG_WINDOW: int = 25    # 長期移動平均の期間(日)
LOOKBACK_DAYS: int = 60  # 株価データ取得期間(日)
EMBED_COLOR_BUY: int = 0x00CC66   # 買いシグナルの色(緑)
EMBED_COLOR_SELL: int = 0xFF3333  # 売りシグナルの色(赤)


# ==============================
# 株価データ取得
# ==============================
def fetch_stock_data(ticker: str, days: int) -> pd.DataFrame:
    '株価データをYahoo Financeから取得する'
    end_date: datetime = datetime.now()
    start_date: datetime = end_date - timedelta(days=days)
    df: pd.DataFrame = yf.download(
        ticker,
        start=start_date.strftime("%Y-%m-%d"),
        end=end_date.strftime("%Y-%m-%d"),
        progress=False,
    )
    if df.empty:
        raise ValueError(f"株価データを取得できませんでした: {ticker}")
    return df


# ==============================
# シグナル検出
# ==============================
def detect_cross_signal(
    df: pd.DataFrame, short_window: int, long_window: int
) -> Optional[dict]:
    '移動平均クロスを検出しシグナル情報を返す'
    df = df.copy()
    df["SMA_SHORT"] = df["Close"].rolling(window=short_window).mean()
    df["SMA_LONG"] = df["Close"].rolling(window=long_window).mean()
    df.dropna(inplace=True)

    if len(df) < 2:
        return None

    prev = df.iloc[-2]
    curr = df.iloc[-1]

    prev_diff: float = float(prev["SMA_SHORT"] - prev["SMA_LONG"])
    curr_diff: float = float(curr["SMA_SHORT"] - curr["SMA_LONG"])

    signal: Optional[str] = None
    if prev_diff <= 0 < curr_diff:
        signal = "BUY"
    elif prev_diff >= 0 > curr_diff:
        signal = "SELL"

    if signal is None:
        return None

    return {
        "signal": signal,
        "close": float(curr["Close"]),
        "sma_short": float(curr["SMA_SHORT"]),
        "sma_long": float(curr["SMA_LONG"]),
        "date": str(curr.name.date()),
    }


# ==============================
# Discord通知送信
# ==============================
def build_embed(signal_info: dict, label: str) -> dict:
    'Discord Embed形式のペイロードを構築する'
    is_buy: bool = signal_info["signal"] == "BUY"
    color: int = EMBED_COLOR_BUY if is_buy else EMBED_COLOR_SELL
    signal_text: str = "🟢 ゴールデンクロス(買い)" if is_buy else "🔴 デッドクロス(売り)"

    embed: dict = {
        "title": f"📊 {label} シグナル検出",
        "description": signal_text,
        "color": color,
        "fields": [
            {"name": "日付", "value": signal_info["date"], "inline": True},
            {
                "name": "終値",
                "value": f"¥{signal_info['close']:,.1f}",
                "inline": True,
            },
            {
                "name": f"SMA{SHORT_WINDOW}",
                "value": f"¥{signal_info['sma_short']:,.1f}",
                "inline": True,
            },
            {
                "name": f"SMA{LONG_WINDOW}",
                "value": f"¥{signal_info['sma_long']:,.1f}",
                "inline": True,
            },
        ],
        "footer": {"text": "Stock Signal Bot | 投資は自己責任です"},
        "timestamp": datetime.utcnow().isoformat(),
    }
    return embed


def send_discord_notification(webhook_url: str, embed: dict) -> int:
    'Discord WebhookにEmbedメッセージを送信する'
    payload: dict = {"embeds":
}
    headers: dict = {"Content-Type": "application/json"}

    response = requests.post(
        webhook_url,
        data=json.dumps(payload),
        headers=headers,
        timeout=10,
    )
    response.raise_for_status()
    return response.status_code


# ==============================
# メイン処理
# ==============================
if __name__ == "__main__":
    print(f"[INFO] {TICKER_LABEL} の株価データを取得中...")
    stock_df: pd.DataFrame = fetch_stock_data(TICKER_SYMBOL, LOOKBACK_DAYS)

    print(f"[INFO] SMA{SHORT_WINDOW} × SMA{LONG_WINDOW} クロスを判定中...")
    result: Optional[dict] = detect_cross_signal(stock_df, SHORT_WINDOW, LONG_WINDOW)

    if result is None:
        print("[INFO] シグナルは検出されませんでした。")
    else:
        print(f"[SIGNAL] {result['signal']} を検出({result['date']})")
        embed_data: dict = build_embed(result, TICKER_LABEL)
        status: int = send_discord_notification(WEBHOOK_URL, embed_data)
        print(f"[INFO] Discord送信完了(ステータス: {status})")

コードの処理フロー解説

上記のコードは、以下の5ステップで構成されています。

* ステップ1 設定読み込み:Webhook URL・銘柄コード・移動平均期間などの定数をファイル冒頭で一元管理する

* ステップ2 株価取得yfinanceで指定銘柄の過去60日分のOHLCVデータをダウンロードする

* ステップ3 シグナル検出:短期SMA(5日)と長期SMA(25日)のクロスを直近2日分の差分から判定する

* ステップ4 Embed構築:買い・売りに応じて色とテキストを切り替え、Discord Embed形式の辞書を生成する

* ステップ5 Webhook送信requests.postでJSON形式のペイロードをDiscordに送信し、ステータスコードを確認する

detect_cross_signal関数を差し替えれば、RSI・MACD・ボリンジャーバンドなど任意のシグナルロジックに対応できます。

シグナル検出結果の読み方と通知メッセージの見方

Embedフィールドの意味

Discordに届くメッセージには4つのフィールドが表示されます。それぞれの意味を確認してください。

* 日付:シグナルが発生した営業日を示す

* 終値:その日の終値(Close)であり、エントリー価格の参考値になる

* SMA5:短期移動平均線の値で、直近のトレンド方向を示す

* SMA25:長期移動平均線の値で、中期トレンドの基準線になる

ゴールデンクロス(Golden Cross)は短期線が長期線を下から上に抜けた状態です。デッドクロス(Dead Cross)はその逆です。

シグナルを過信しないための注意点

移動平均クロスは遅行指標(Lagging Indicator)です。価格が動いた後にシグナルが出るため、すでにトレンドの一部を逃している場合があります。

レンジ相場ではクロスが頻発し、だましシグナルが連続します。通知が来たからといって即座にエントリーすることは危険です。出来高やRSI(Relative Strength Index:相対力指数)など、別の指標と組み合わせて判断してください。

【コピペOK】複数銘柄の一括監視と定期実行

複数銘柄への拡張方法

実運用では複数銘柄を同時に監視したいケースがほとんどです。設定エリアを以下のように辞書のリストに変更してください。


import os
import json
import time
from datetime import datetime, timedelta
from typing import Optional

import requests
import yfinance as yf
import pandas as pd

# ==============================
# 設定エリア(複数銘柄対応版)
# ==============================
WEBHOOK_URL: str = os.environ.get(
    "DISCORD_WEBHOOK_URL",
    "https://discord.com/api/webhooks/YOUR_WEBHOOK_ID/YOUR_WEBHOOK_TOKEN",
)
WATCHLIST: list[dict] = [
    {"ticker": "7203.T", "label": "トヨタ自動車(7203)"},
    {"ticker": "6758.T", "label": "ソニーG(6758)"},
    {"ticker": "9984.T", "label": "ソフトバンクG(9984)"},
]
SHORT_WINDOW: int = 5
LONG_WINDOW: int = 25
LOOKBACK_DAYS: int = 60
EMBED_COLOR_BUY: int = 0x00CC66
EMBED_COLOR_SELL: int = 0xFF3333
SEND_INTERVAL_SEC: float = 2.5  # Webhook送信間隔(秒)


# ==============================
# 株価データ取得
# ==============================
def fetch_stock_data(ticker: str, days: int) -> pd.DataFrame:
    '株価データをYahoo Financeから取得する'
    end_date: datetime = datetime.now()
    start_date: datetime = end_date - timedelta(days=days)
    df: pd.DataFrame = yf.download(
        ticker,
        start=start_date.strftime("%Y-%m-%d"),
        end=end_date.strftime("%Y-%m-%d"),
        progress=False,
    )
    if df.empty:
        raise ValueError(f"株価データを取得できませんでした: {ticker}")
    return df


# ==============================
# シグナル検出
# ==============================
def detect_cross_signal(
    df: pd.DataFrame, short_window: int, long_window: int
) -> Optional[dict]:
    '移動平均クロスを検出しシグナル情報を返す'
    df = df.copy()
    df["SMA_SHORT"] = df["Close"].rolling(window=short_window).mean()
    df["SMA_LONG"] = df["Close"].rolling(window=long_window).mean()
    df.dropna(inplace=True)

    if len(df) < 2:
        return None

    prev = df.iloc[-2]
    curr = df.iloc[-1]

    prev_diff: float = float(prev["SMA_SHORT"] - prev["SMA_LONG"])
    curr_diff: float = float(curr["SMA_SHORT"] - curr["SMA_LONG"])

    signal: Optional[str] = None
    if prev_diff <= 0 < curr_diff:
        signal = "BUY"
    elif prev_diff >= 0 > curr_diff:
        signal = "SELL"

    if signal is None:
        return None

    return {
        "signal": signal,
        "close": float(curr["Close"]),
        "sma_short": float(curr["SMA_SHORT"]),
        "sma_long": float(curr["SMA_LONG"]),
        "date": str(curr.name.date()),
    }


# ==============================
# Discord通知送信
# ==============================
def build_embed(signal_info: dict, label: str) -> dict:
    'Discord Embed形式のペイロードを構築する'
    is_buy: bool = signal_info["signal"] == "BUY"
    color: int = EMBED_COLOR_BUY if is_buy else EMBED_COLOR_SELL
    signal_text: str = "🟢 ゴールデンクロス(買い)" if is_buy else "🔴 デッドクロス(売り)"

    embed: dict = {
        "title": f"📊 {label} シグナル検出",
        "description": signal_text,
        "color": color,
        "fields": [
            {"name": "日付", "value": signal_info["date"], "inline": True},
            {
                "name": "終値",
                "value": f"¥{signal_info['close']:,.1f}",
                "inline": True,
            },
            {
                "name": f"SMA{SHORT_WINDOW}",
                "value": f"¥{signal_info['sma_short']:,.1f}",
                "inline": True,
            },
            {
                "name": f"SMA{LONG_WINDOW}",
                "value": f"¥{signal_info['sma_long']:,.1f}",
                "inline": True,
            },
        ],
        "footer": {"text": "Stock Signal Bot | 投資は自己責任です"},
        "timestamp": datetime.utcnow().isoformat(),
    }
    return embed


def send_discord_notification(webhook_url: str, embed: dict) -> int:
    'Discord WebhookにEmbedメッセージを送信する'
    payload: dict = {"embeds":
}
    headers: dict = {"Content-Type": "application/json"}

    response = requests.post(
        webhook_url,
        data=json.dumps(payload),
        headers=headers,
        timeout=10,
    )
    response.raise_for_status()
    return response.status_code


# ==============================
# メイン処理(複数銘柄一括チェック)
# ==============================
if __name__ == "__main__":
    signal_count: int = 0

    for item in WATCHLIST:
        ticker: str = item["ticker"]
        label: str = item["label"]

        try:
            print(f"[INFO] {label} のデータを取得中...")
            stock_df: pd.DataFrame = fetch_stock_data(ticker, LOOKBACK_DAYS)

            result: Optional[dict] = detect_cross_signal(
                stock_df, SHORT_WINDOW, LONG_WINDOW
            )

            if result is None:
                print(f"[INFO] {label}: シグナルなし")
                continue

            print(f"[SIGNAL] {label}: {result['signal']}({result['date']})")
            embed_data: dict = build_embed(result, label)
            status: int = send_discord_notification(WEBHOOK_URL, embed_data)
            print(f"[INFO] Discord送信完了(ステータス: {status})")
            signal_count += 1

            time.sleep(SEND_INTERVAL_SEC)

        except Exception as e:
            print(f"[ERROR] {label}: {e}")

    print(f"[DONE] 検出シグナル数: {signal_count}")

コードの処理フロー解説

上記のコードは、以下の4ステップで構成されています。

* ステップ1 ウォッチリスト定義WATCHLISTに辞書形式で銘柄コードと表示名を列挙する

* ステップ2 ループ処理:各銘柄について株価取得 → シグナル判定 → 通知送信を順次実行する

* ステップ3 レート制限対策SEND_INTERVAL_SEC(2.5秒)の待機を挟み、Discordのレート制限(30回/分)を回避する

* ステップ4 エラーハンドリング:1銘柄の失敗が他の銘柄に影響しないよう、try-exceptで個別に捕捉する

WATCHLISTに銘柄を追加するだけで監視対象を増やせます。cronやタスクスケジューラで毎日15時30分に実行すれば、場が閉まった直後に自動チェックが走ります。

cronによる定期実行の設定例

LinuxやmacOSでは、cronで毎営業日に自動実行できます。以下のコマンドでcronを編集してください。


# ==============================
# cron設定を編集
# ==============================
crontab -e

# ==============================
# 毎週月〜金の15:30にスクリプトを実行
# ==============================
# 30 15 * * 1-5 cd /home/user/stock-bot && DISCORD_WEBHOOK_URL="https://discord.com/api/webhooks/xxx/yyy" /usr/bin/python3 main.py >> /home/user/stock-bot/cron.log 2>&1

環境変数DISCORD_WEBHOOK_URLをcronの行内で指定している点に注目してください。.envファイルを使う場合はpython-dotenvの導入を検討してください。

よくあるエラーと対処法

401 Unauthorized または 404 Not Found

Webhook URLが無効な場合に発生します。URLをコピーし直すか、Webhookが削除されていないか確認してください。

以下を試してください。

* Discordのチャンネル設定 →「連携サービス」でWebhookが存在するか確認する

* URLの末尾にスペースや改行が混入していないか確認する

* 環境変数が正しくセットされているかecho $DISCORD_WEBHOOK_URLで確認する

429 Too Many Requests(レート制限)

短時間に大量のリクエストを送ると発生します。Discordのチャンネル単位のレート制限は約30回/分です。

以下を試してください。

* SEND_INTERVAL_SECを3〜5秒に引き上げる

* レスポンスヘッダのRetry-After値を読み取り、その秒数だけtime.sleepで待機する

* 10銘柄以上を監視する場合はバッチを分割し、複数チャンネルのWebhookに分散する

yfinanceで株価データが空になる

yf.downloadが空のDataFrameを返すケースです。原因はティッカーシンボルの誤りか、Yahoo Financeの一時的な障害です。

以下を試してください。

* ティッカーシンボルがYahoo Finance形式であるか確認する(日本株は末尾に.Tが必要)

* LOOKBACK_DAYSを90〜120に広げてリトライする

* yfinanceを最新版にアップデートする(pip install -U yfinance

まとめ

この記事では、Discord Webhookを使って株価シグナルをPythonで自動通知するシステムの構築方法を解説しました。

要点を整理します。

* Discord WebhookはURL1つで通知を送信でき、Bot作成やOAuth認証は不要

* LINE Notify終了後の代替として、Embed形式のリッチな通知が無料で使える

* 移動平均クロス(SMA5×SMA25)のシグナル検出コードはそのままコピペで動作する

* 複数銘柄の一括監視はWATCHLISTに辞書を追加するだけで拡張できる

* レート制限(30回/分)を考慮し、送信間隔を2.5秒以上空けることが安定運用の鍵

次のステップとして、シグナル検出ロジックの拡張に取り組んでください。detect_cross_signal関数をRSIやMACDベースのロジックに差し替えるだけで、通知の仕組みはそのまま再利用できます。

さらに、VPS(Virtual Private Server:仮想専用サーバー)上でcronを稼働させれば、自分のPCを起動していなくても24時間365日の自動監視が実現します。まずは1銘柄で動作確認し、段階的に監視対象を広げていくことを推奨します。

タイトルとURLをコピーしました