決算アラートBotの作り方｜Slack・Discord・LINE通知

決算発表日が近づいたらSlackやLINEに自動通知するBotを作ります。GitHub Actionsで毎朝実行すれば、スマホにプッシュ通知が届きます。

ただ、この手のBotには「気づかないうちに死んでいる」という固有の失敗があります。決算がない日は何も届かないので、Botが壊れて何も届かないのと区別がつきません。気づくのは決算を見逃したあとです。

この記事では通知の実装に加えて、壊れたら壊れたと分かる仕組みと同じ通知を7日連続で受け取らない仕組みまで作ります。

📘 外部参考：LINE Messaging API（公式） ／ Slack Incoming Webhooks（公式） ／ GitHub Actions ドキュメント

LINE Notifyは終了した。移行先を選ぶ

まず前提の確認です。LINE Notifyは2025年3月31日にサービスを終了しました。notify-api.line.meを叩くコードはもう動きません。ネット上の記事の大半がまだLINE Notifyのままなので注意してください。

通知先	無料枠	設定の手間	向いている用途
Discord Webhook	実質無制限	最小（URLを作るだけ）	日次のログ・定期通知
Slack Webhook	実質無制限	小（アプリ作成）	整形された通知
LINE Messaging API	月200通	大（公式アカウント作成）	本当に重要な通知だけ

LINE Messaging APIの無料枠（コミュニケーションプラン）は月200通で、しかも送信先の人数ぶんカウントされます。5人のグループに1回送れば5通です。上位プランは月5,000円からなので、個人のBotで気軽に超えるわけにはいきません。

そこで「日次の定期通知はDiscord、当日の決算だけLINE」という使い分けにします。1日1件でも月30通なので、200通の枠にじゅうぶん収まります。

通知層を差し替え可能にする

LINE Notifyの終了で全スクリプトが止まった経験から学んだのは、通知先を直接呼ばないということです。共通のインターフェースを挟みます。

from __future__ import annotations
import json
import os
from abc import ABC, abstractmethod

import requests

class Notifier(ABC):
    @abstractmethod
    def send(self, text: str, urgent: bool = False) -> bool: ...


class DiscordNotifier(Notifier):
    def __init__(self, webhook_url: str):
        self.url = webhook_url

    def send(self, text: str, urgent: bool = False) -> bool:
        r = requests.post(self.url, json={"content": text[:1900]}, timeout=15)
        return r.status_code in (200, 204)


class SlackNotifier(Notifier):
    def __init__(self, webhook_url: str):
        self.url = webhook_url

    def send(self, text: str, urgent: bool = False) -> bool:
        r = requests.post(self.url, json={"text": text}, timeout=15)
        return r.status_code == 200


class LineNotifier(Notifier):
    """LINE Messaging API。無料枠が月200通なので urgent のときだけ使う。"""

    def __init__(self, token: str, user_id: str):
        self.token, self.user_id = token, user_id

    def send(self, text: str, urgent: bool = False) -> bool:
        if not urgent:
            return True          # 通常通知では枠を消費しない
        r = requests.post(
            "https://api.line.me/v2/bot/message/push",
            headers={"Authorization": f"Bearer {self.token}",
                     "Content-Type": "application/json"},
            data=json.dumps({
                "to": self.user_id,
                "messages": [{"type": "text", "text": text[:4900]}],
            }),
            timeout=15)
        if r.status_code != 200:
            print(f"LINE送信失敗 {r.status_code}: {r.text}")
        return r.status_code == 200


class MultiNotifier(Notifier):
    def __init__(self, notifiers: list[Notifier]):
        self.notifiers = notifiers

    def send(self, text: str, urgent: bool = False) -> bool:
        # 1つ失敗しても他には送る
        return all([n.send(text, urgent) for n in self.notifiers])


def build_notifier() -> Notifier:
    ns: list[Notifier] = []
    if url := os.getenv("DISCORD_WEBHOOK_URL"):
        ns.append(DiscordNotifier(url))
    if url := os.getenv("SLACK_WEBHOOK_URL"):
        ns.append(SlackNotifier(url))
    if (tok := os.getenv("LINE_TOKEN")) and (uid := os.getenv("LINE_USER_ID")):
        ns.append(LineNotifier(tok, uid))
    if not ns:
        raise RuntimeError("通知先が1つも設定されていません")
    return MultiNotifier(ns)

urgentフラグでLINEを使うかどうかを切り替えるのがポイントです。「あと5日」はDiscordだけ、「今日が決算日」はLINEにも、という設計にできます。

LINEのユーザーIDは、LINE Developersコンソールの「Basic settings」にあるYour user IDを使います。自分にだけ送るならこれで足ります。

決算日を取得する

yfinanceのcalendarやget_earnings_dates()は、日本株ではしばしば空になります。米国株なら比較的取れます。

import datetime as dt
from zoneinfo import ZoneInfo

import pandas as pd
import yfinance as yf

JST = ZoneInfo("Asia/Tokyo")

class FetchError(Exception):
    """データ取得に失敗したことを、結果ゼロと区別するための例外。"""


def earnings_from_yf(tickers: list[str], days_ahead: int = 7) -> list[dict]:
    today = dt.datetime.now(JST).date()
    found, failed = [], []

    for tk in tickers:
        try:
            ed = yf.Ticker(tk).get_earnings_dates(limit=12)
            if ed is None or ed.empty:
                failed.append(tk)
                continue
            for ts in pd.to_datetime(ed.index):
                d = ts.tz_localize(None).date() if ts.tzinfo else ts.date()
                delta = (d - today).days
                if 0 <= delta <= days_ahead:
                    found.append({"ticker": tk, "date": d.isoformat(),
                                  "days": delta})
        except Exception as e:
            print(f"{tk}: 取得失敗 {type(e).__name__}: {e}")
            failed.append(tk)

    # 半分以上取れていないなら、それは「決算なし」ではなく「壊れている」
    if len(failed) > len(tickers) / 2:
        raise FetchError(f"{len(failed)}/{len(tickers)} 銘柄で取得失敗: {failed}")

    return sorted(found, key=lambda x: x["days"])

ここが元のコードとの決定的な違いです。元のコードはexcept: passで全部握り潰していました。yfinanceが仕様変更で全滅しても、Botは平然と「今後7日以内に決算はありません」と通知し続けます。

そうならないように、失敗が多すぎたら例外を投げます。「取れなかった」と「無かった」は別の事実です。

日本株を真面目にやるなら、JPX公式のJ-Quants APIの決算発表予定日を使ってください。全上場銘柄ぶんが確実に揃います。

def earnings_from_jquants(codes: list[str], api_key: str,
                          days_ahead: int = 7) -> list[dict]:
    """J-Quantsの決算発表予定日から抽出する。"""
    r = requests.get("https://api.jquants.com/v2/fins/announcement",
                     headers={"Authorization": f"Bearer {api_key}"},
                     timeout=30)
    r.raise_for_status()
    rows = r.json().get("announcement", [])
    if not rows:
        raise FetchError("J-Quantsから予定日を取得できませんでした")

    today = dt.datetime.now(JST).date()
    want = {c.replace(".T", "") for c in codes}
    out = []
    for row in rows:
        code = str(row.get("Code", ""))[:4]
        if code not in want:
            continue
        d = dt.date.fromisoformat(row["Date"])
        delta = (d - today).days
        if 0 <= delta <= days_ahead:
            out.append({"ticker": f"{code}.T", "date": d.isoformat(),
                        "days": delta})
    return sorted(out, key=lambda x: x["days"])

同じ通知を毎日受け取らない

毎日実行するBotで「7日以内の決算」を通知すると、同じ銘柄について7日連続で通知が来ます。3日で見なくなります。

解決策は、通知した組み合わせを記録しておくことです。

import sqlite3
from pathlib import Path

class SentLog:
    def __init__(self, path: str = "sent.db"):
        self.con = sqlite3.connect(path)
        self.con.execute("""
            CREATE TABLE IF NOT EXISTS sent(
                key TEXT PRIMARY KEY,
                at TEXT NOT NULL
            )""")
        self.con.commit()

    def is_new(self, key: str) -> bool:
        return self.con.execute(
            "SELECT 1 FROM sent WHERE key=?", (key,)).fetchone() is None

    def mark(self, key: str) -> None:
        with self.con:
            self.con.execute("INSERT OR IGNORE INTO sent VALUES (?,?)",
                             (key, dt.datetime.now(JST).isoformat()))

def alert_key(item: dict) -> str:
    """通知したい節目を決める。7日前と当日の2回だけ通知する。"""
    if item["days"] == 0:
        stage = "day0"
    elif item["days"] <= 3:
        stage = "day3"
    else:
        stage = "week"
    return f"{item['ticker']}-{item['date']}-{stage}"

節目を3段階に絞るのがコツです。「1週間前」「3日前」「当日」の3回だけ通知します。alert_keyに決算日そのものを含めているので、日程が変更されれば別のキーになり、ちゃんと再通知されます。

壊れたら分かるようにする

ここがこのBotでいちばん大事な部分です。

def heartbeat_path() -> Path:
    return Path("last_run.txt")

def run(tickers: list[str], days_ahead: int = 7) -> None:
    notifier = build_notifier()
    now = dt.datetime.now(JST)
    stamp = now.strftime("%Y-%m-%d %H:%M")

    try:
        items = earnings_from_yf(tickers, days_ahead)
    except FetchError as e:
        # 静かに失敗させない。必ず通知する
        notifier.send(f"⚠️ 決算Bot エラー\n{stamp}\n{e}", urgent=True)
        raise
    except Exception as e:
        notifier.send(f"⚠️ 決算Bot 想定外のエラー\n{stamp}\n"
                      f"{type(e).__name__}: {e}", urgent=True)
        raise

    log = SentLog()
    fresh = [it for it in items if log.is_new(alert_key(it))]

    if fresh:
        lines = [f"📅 決算アラート（{stamp}）"]
        for it in fresh:
            mark = "🔴" if it["days"] == 0 else "🟡" if it["days"] <= 3 else "🟢"
            when = "本日" if it["days"] == 0 else f"あと{it['days']}日"
            lines.append(f"{mark} {it['ticker']}  {it['date']}（{when}）")

        urgent = any(it["days"] == 0 for it in fresh)
        if notifier.send("\n".join(lines), urgent=urgent):
            for it in fresh:
                log.mark(alert_key(it))     # 送信成功してから記録する
    else:
        print(f"{stamp}: 新規の通知はありません（監視 {len(items)}件）")

    # 週1回、生存確認を送る
    if now.weekday() == 0:
        notifier.send(f"✅ 決算Bot 稼働中（{stamp}）\n"
                      f"監視銘柄 {len(tickers)}件 / 予定 {len(items)}件")

    heartbeat_path().write_text(now.isoformat(), encoding="utf-8")


if __name__ == "__main__":
    WATCHLIST = ["7203.T", "6501.T", "6758.T", "9984.T",
                 "AAPL", "NVDA", "MSFT"]
    run(WATCHLIST)

3つの仕掛けを入れています。

• エラーを必ず通知する。しかもurgent=TrueでLINEにも飛ばす。壊れたことは決算より重要です
• 週1回の生存確認。月曜に「稼働中」が来なければ、Botが止まっています
• 送信成功してから記録する。逆にすると、送信失敗したのに「送信済み」になって永久に届きません

GitHub Actionsで動かす

# .github/workflows/earnings_alert.yml
name: Earnings Alert Bot

on:
  schedule:
    - cron: '0 23 * * 0-4'      # UTC 23:00 = JST 翌8:00（月〜金）
  workflow_dispatch:            # 手動実行も可能にしておく

permissions:
  contents: write               # 状態ファイルをコミットするため

jobs:
  alert:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
          cache: pip

      - run: pip install -r requirements.txt

      - name: 通知済みDBを復元
        uses: actions/cache@v4
        with:
          path: sent.db
          key: sent-db-${{ github.run_id }}
          restore-keys: sent-db-

      - name: Bot実行
        env:
          DISCORD_WEBHOOK_URL: ${{ secrets.DISCORD_WEBHOOK_URL }}
          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
          LINE_TOKEN: ${{ secrets.LINE_TOKEN }}
          LINE_USER_ID: ${{ secrets.LINE_USER_ID }}
        run: python earnings_bot.py

GitHub Actionsには、知らないとハマる仕様が3つあります。

仕様	影響	対処
スケジュールが遅延する	数分〜30分ずれる	余裕のある時刻に設定
負荷が高いと実行がスキップされる	その日は通知なし	生存確認で気づく
60日間コミットがないと無効化	Botが静かに止まる	定期的にコミットするか、無効化通知メールを見逃さない

3つ目が特に厄介です。リポジトリに60日間コミットがないと、GitHubはスケジュール実行を自動で止めます。個人のBot用リポジトリはコミットが途絶えがちなので、まさにこれに当たります。生存確認の通知を入れておく理由の半分はこれです。

また、0時0分ちょうどのcronは世界中のジョブが集中するので、7分や23分といった半端な時刻にしておくと遅延が減ります。

Slackで見やすく整形する

def slack_blocks(items: list[dict]) -> list[dict]:
    blocks = [{
        "type": "header",
        "text": {"type": "plain_text", "text": "📅 決算アラート"},
    }]
    for it in items:
        mark = "🔴" if it["days"] == 0 else "🟡" if it["days"] <= 3 else "🟢"
        blocks.append({
            "type": "section",
            "fields": [
                {"type": "mrkdwn", "text": f"*銘柄*\n{mark} {it['ticker']}"},
                {"type": "mrkdwn", "text": f"*決算日*\n{it['date']}"},
                {"type": "mrkdwn", "text": f"*残り*\n{it['days']}日"},
            ],
        })
    return blocks

Block Kitは1メッセージ50ブロックまでという制限があります。監視銘柄が多いと超えるので、上のように1銘柄1ブロックにするなら16銘柄が上限です。超える場合は分割送信してください。

まとめ

• LINE Notifyは終了済み。移行先はDiscord・Slack・LINE Messaging API
• LINE Messaging APIの無料枠は月200通・送信人数でカウント。重要通知だけに使う
• 通知層をインターフェースで抽象化しておくと、サービス終了に強い
• except: passをやめる。「取れなかった」と「無かった」を区別する
• エラーは必ず通知し、週1回の生存確認を送る
• 通知済みを記録して同じ内容を連日送らない
• GitHub Actionsは60日でスケジュールが無効化される

この種のBotで一番怖いのは、動かなくなることではなく動かなくなったことに気づかないことです。「何も起きていない」と「壊れている」を区別できる設計にしておけば、安心して放っておけます。