ドル円RSIアラートBotをPythonで作る｜LINE通知終了に対応

ドル円は毎日動きがあるので、子供が生まれる前は寝る前にチャートを確認するのが習慣でした。でも最近はそれすらできなくて、気づいたら大きく動いていたことが何度かあって。せめてRSIが特定の水準に達したらアラートだけ飛ばしてくれるBotを作ることにしました。

ただ、作り直すにあたって2つ、重要な前提が変わっていました。

1つ目は、LINE Notifyが2025年3月31日でサービス終了したこと。notify-api.line.meへのリクエストはすべてエラーになります。ネット上のBot記事の大半はこのAPIを使っているので、写経しても通知は飛びません。代替はLINE公式アカウントのMessaging APIです。

2つ目は、RSIの計算方法が1つではないこと。単純移動平均で書いたRSIと、MT4やTradingViewが使うWilder方式のRSIは、同じ14期間でも数値が数ポイント違います。閾値30で通知する設計なら、この差はそのまま通知の有無になります。

この記事では、その2点を踏まえた実装を書きます。あわせて、鳴りすぎを防ぐ状態管理も入れます。

📘 外部参考：Messaging API（LINE公式） ／ yfinance（GitHub）

RSIの計算方式で数値が変わる

RSIは「値上がり幅の平均 ÷（値上がり幅の平均＋値下がり幅の平均）」で作りますが、この「平均」の取り方に2通りあるのが混乱の元です。

方式	平均の取り方	特徴	採用例
SMA方式	単純移動平均	期間外のデータを完全に切る。反応が急	入門記事のコード
Wilder方式	平滑移動平均（α=1/n）	過去を指数的に薄めて残す。滑らか	MT4、TradingViewの既定

import numpy as np
import pandas as pd
import yfinance as yf

def rsi_sma(close: pd.Series, period: int = 14) -> pd.Series:
    delta = close.diff()
    gain = delta.clip(lower=0)
    loss = (-delta).clip(lower=0)
    avg_gain = gain.rolling(period).mean()
    avg_loss = loss.rolling(period).mean()
    rs = avg_gain / avg_loss.replace(0, np.nan)
    return (100 - 100 / (1 + rs)).fillna(100)

def rsi_wilder(close: pd.Series, period: int = 14) -> pd.Series:
    delta = close.diff()
    gain = delta.clip(lower=0)
    loss = (-delta).clip(lower=0)
    # Wilderの平滑化は alpha = 1/period の指数移動平均と同じ
    avg_gain = gain.ewm(alpha=1 / period, adjust=False, min_periods=period).mean()
    avg_loss = loss.ewm(alpha=1 / period, adjust=False, min_periods=period).mean()
    rs = avg_gain / avg_loss.replace(0, np.nan)
    return (100 - 100 / (1 + rs)).fillna(100)

2つの差を実際に測ります。

def compare_rsi(close: pd.Series, period: int = 14) -> dict:
    a, b = rsi_sma(close, period).dropna(), rsi_wilder(close, period).dropna()
    idx = a.index.intersection(b.index)
    a, b = a.loc[idx], b.loc[idx]
    diff = (a - b).abs()
    return {
        "mean_diff": round(float(diff.mean()), 2),
        "max_diff": round(float(diff.max()), 2),
        "sma_under30": int((a <= 30).sum()),
        "wilder_under30": int((b <= 30).sum()),
        "disagree_30": int(((a <= 30) != (b <= 30)).sum()),
    }

disagree_30が、片方だけが「売られすぎ」と判定した回数です。ここが0にならないので、どちらを使うかを決めないと、チャートを見て「30を割ってないじゃないか」と混乱します。MT4やTradingViewと揃えたいならrsi_wilderを使ってください。以降はこちらを使います。

ドル円のデータを取る

def load_usdjpy(interval: str = "1h", period: str = "60d") -> pd.Series:
    for ticker in ("USDJPY=X", "JPY=X"):
        df = yf.download(ticker, period=period, interval=interval,
                         auto_adjust=False, progress=False)
        if df.empty:
            continue
        if isinstance(df.columns, pd.MultiIndex):
            df.columns = df.columns.get_level_values(0)
        s = df["Close"].dropna()
        s.index = pd.to_datetime(s.index)
        return s.sort_index()
    raise RuntimeError("ドル円のデータを取得できません")

ティッカーを2つ試しているのは、時期によって片方が空を返すことがあるためです。どちらもドル円ですが、JPY=Xの方が安定している印象があります。

もう1つ注意点があります。yfinanceの1時間足は最新の足が「まだ確定していない足」です。形成途中の足でRSIを計算すると、数分後には値が変わります。確定足だけで判定するようにします。

def closed_bars(s: pd.Series, interval_minutes: int = 60) -> pd.Series:
    """最後の足がまだ形成中なら落とす。"""
    if len(s) < 2:
        return s
    last = s.index[-1]
    now = pd.Timestamp.now(tz=last.tz) if last.tz else pd.Timestamp.now()
    if (now - last).total_seconds() < interval_minutes * 60:
        return s.iloc[:-1]
    return s

LINE Messaging APIで通知する

LINE Notifyの代わりです。LINE公式アカウントを作り、Messaging APIを有効化して、チャネルアクセストークンと自分のユーザーIDを取得します。無料枠でも月に一定数のメッセージを送れます。

import os
import requests

LINE_TOKEN = os.getenv("LINE_CHANNEL_ACCESS_TOKEN")
LINE_USER_ID = os.getenv("LINE_USER_ID")

def notify_line(message: str) -> bool:
    if not (LINE_TOKEN and LINE_USER_ID):
        print("[dry-run]", message)
        return False
    r = requests.post(
        "https://api.line.me/v2/bot/message/push",
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {LINE_TOKEN}",
        },
        json={
            "to": LINE_USER_ID,
            "messages": [{"type": "text", "text": message[:900]}],
        },
        timeout=15,
    )
    if r.status_code != 200:
        print("LINE通知失敗", r.status_code, r.text[:200])
        return False
    return True

トークンが未設定なら画面に出すだけにしています。通知処理で例外を投げてBotごと止めないのが大事です。通知が飛ばないより、Botが死ぬ方が困ります。

LINEの設定が面倒なら、Discord Webhookの方が5分で終わります。URLを1つ発行するだけです。

def notify_discord(message: str) -> bool:
    url = os.getenv("DISCORD_WEBHOOK_URL")
    if not url:
        print("[dry-run]", message)
        return False
    r = requests.post(url, json={"content": message[:1900]}, timeout=15)
    return r.status_code in (200, 204)

鳴りすぎを止める

ここを入れないと実用になりません。RSIが29.8で行ったり来たりすると、1時間ごとに毎回通知が飛びます。「状態が変わったときだけ鳴らす」設計にします。

import json
from pathlib import Path

STATE_FILE = Path("rsi_state.json")

LOW, LOW_EXIT = 30, 35      # 30で入り、35を超えたら解除（ヒステリシス）
HIGH, HIGH_EXIT = 70, 65

def load_state() -> str:
    if STATE_FILE.exists():
        return json.loads(STATE_FILE.read_text()).get("zone", "neutral")
    return "neutral"

def save_state(zone: str) -> None:
    STATE_FILE.write_text(json.dumps({"zone": zone}), encoding="utf-8")

def next_zone(rsi: float, current: str) -> str:
    if current == "oversold":
        return "neutral" if rsi >= LOW_EXIT else "oversold"
    if current == "overbought":
        return "neutral" if rsi <= HIGH_EXIT else "overbought"
    if rsi <= LOW:
        return "oversold"
    if rsi >= HIGH:
        return "overbought"
    return "neutral"

解除の閾値を入る閾値からずらしているのが要点です（30で入って35で抜ける）。同じ30で判定すると、境界付近でバタつきます。これはヒステリシスと呼ばれる、サーモスタットなどでも使われる考え方です。

組み立てる

def run(interval: str = "1h", period_bars: int = 14) -> None:
    close = closed_bars(load_usdjpy(interval=interval))
    rsi = rsi_wilder(close, period_bars).dropna()
    if rsi.empty:
        print("RSIを計算できるだけの足がありません")
        return

    now_rsi = float(rsi.iloc[-1])
    price = float(close.iloc[-1])
    bar_time = close.index[-1]

    prev = load_state()
    zone = next_zone(now_rsi, prev)

    print(f"{bar_time} price={price:.3f} RSI={now_rsi:.1f} {prev} -> {zone}")

    if zone != prev and zone in ("oversold", "overbought"):
        label = "売られすぎ" if zone == "oversold" else "買われすぎ"
        notify_line(
            f"【ドル円 {label}】\n"
            f"RSI(14, {interval}) = {now_rsi:.1f}\n"
            f"価格 = {price:.3f}\n"
            f"足の時刻 = {bar_time:%Y-%m-%d %H:%M}"
        )
    save_state(zone)

if __name__ == "__main__":
    run()

通知文に足の時刻を必ず入れるようにしました。スマホで見たときに「これはいつの話か」が分からないと、判断できません。休日にデータが古いまま通知されても、時刻があれば気づけます。

定期実行

Windowsのタスクスケジューラに登録するのが一番シンプルです。バッチファイルを1つ用意します。

@echo off
cd /d C:\work\fx-alert
call .venv\Scripts\activate.bat
python rsi_alert.py >> log.txt 2>&1

ログを追記しているのは、動いていないことに気づくためです。通知が来ない理由が「シグナルが出ていない」のか「Botが落ちている」のか、ログがないと区別できません。僕は子供に電源を切られてBotが止まったことが2回あります。

常時稼働させたいならVPSですが、月数百円とはいえ最初は自宅PCで十分でした。落ちても実害がないのが、通知だけBotの良いところです。

まとめ

ドル円のRSIアラートBotは1〜2時間で作れます。ただし、いま作るなら押さえるべき点が3つあります。

LINE Notifyは終了しているので、Messaging APIかDiscord Webhookを使う。RSIはWilder方式で計算しないとチャートと数値が合わない。そしてヒステリシス付きの状態管理を入れないと、通知が鳴りすぎてミュートすることになる。

完全自動売買は怖いという人でも、通知だけなら実害がありません。RSIそのものの有効性を検証したい場合は、バックテスト実装の記事の枠組みにシグナルを差し替えれば測れます。次は通知と一緒にチャート画像も送るようにしたいと考えています。