ドル円のボリンジャーバンドをPythonで実装してLINEに自動アラートを送る方法

自動化・運用

ボリンジャーバンドの外に価格が飛び出したのに、気づいたのは翌日——という経験を何度もしました。子育て中だと相場を見張る時間がありません。「バンド逸脱を自動検知してスマホに飛ばす」仕組みを作ったのは、そういう事情からです。

ただ、作ってみて分かったのは「検知するのは簡単だが、通知の設計が難しい」ということでした。単純に条件へ合致するたび送っていると、1日に何十件も飛んできて、結局見なくなります。

この記事では、ドル円のボリンジャーバンドをPythonで実装し、「本当に知らせるべき瞬間だけ通知する」ところまで作り込みます。指標の計算そのものより、アラートとして成立させるための工夫に重点を置きました。

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

ボリンジャーバンドが示すもの

ボリンジャーバンドは、移動平均線の上下に標準偏差を足し引きした帯です。値動きのばらつきに応じて、帯の幅が自動的に伸び縮みします。

中心線(ミドルバンド) = 20期間の移動平均
上限(+2σ) = 中心線 + 標準偏差 × 2
下限(−2σ) = 中心線 − 標準偏差 × 2

正規分布を前提とすれば、価格が±2σの内側に収まる確率は約95%です。つまりバンドの外に出るのは「珍しい状態」ということになります。

ただし、この「珍しい」の解釈が2通りあるのが厄介なところです。

解釈考え方取る行動有効な相場
逆張り行き過ぎたので戻る−2σで買いレンジ相場
順張り強い動きが始まった+2σで買いトレンド相場

正反対の行動が導かれます。どちらが正しいかは相場環境次第で、これを見誤ると損失につながります。だからこそ「通知はするが、判断は人間がする」という設計が現実的だと考えています。

また、為替は株価リターン以上に裾が厚く、±2σを超える頻度は理論値より高くなります。この点は歪度・尖度で「暴落しやすさ」を数値化する方法で検証しています。

もう1つ、ボリンジャーバンドで見るべきなのがバンドの幅そのものです。価格が上限・下限に触れたかどうかばかりに目が行きがちですが、帯が広がっているか狭まっているかのほうが、実は情報量が多いことがあります。

  • 帯が狭い(スクイーズ): 値動きが小さい。大きな動きの前触れとされる
  • 帯が広がり始めた: 動きが本格化しつつある局面
  • 帯が非常に広い: すでに大きく動いた後。逸脱しても驚きは小さい

この視点があると、通知の設計が変わります。「帯が狭いときにバンドを抜けた」のは意味のある変化ですが、「もともと帯が広い状態でわずかに逸脱した」のはほぼノイズです。後半ではこの区別をコードに落とし込みます。

📘 外部参考Bollinger Bands(Investopedia)

Pythonで計算する

pip install yfinance pandas numpy requests python-dotenv matplotlib
import yfinance as yf
import numpy as np
import pandas as pd


def bollinger(close, period=20, sigma=2.0):
    """ボリンジャーバンドを計算する"""
    ma = close.rolling(period).mean()
    sd = close.rolling(period).std()

    df = pd.DataFrame({
        "close": close,
        "middle": ma,
        "upper": ma + sd * sigma,
        "lower": ma - sd * sigma,
    })

    # バンド内での位置を0〜1で表す(%B)
    width = df["upper"] - df["lower"]
    df["percent_b"] = (df["close"] - df["lower"]) / width.replace(0, np.nan)

    # バンドの幅(中心線に対する比率)
    df["band_width"] = width / df["middle"] * 100
    return df


px = yf.download("JPY=X", period="6mo", auto_adjust=True,
                 progress=False)["Close"].dropna()
bb = bollinger(px).dropna()
print(bb.tail(5).round(3).to_string())

ここで2つの補助指標を一緒に計算しているのがポイントです。単にバンドの値を出すだけでは、通知の判断材料として足りません。

  • %B(パーセントB): バンド内の位置。0が下限、1が上限、1超なら上抜け
  • バンド幅: 帯の広さ。狭いときは大きな動きの前触れとされる

%Bを使うと、「どのくらい飛び出したか」を1つの数字で扱えます。1.05なら少しはみ出した程度、1.3なら大きく逸脱している、といった具合です。通貨ペアが変わっても同じ尺度で比較できるのも利点です。

「通知すべき瞬間」を定義する

ここが本記事の核心です。単純に「バンドの外にいる」ことを条件にすると、通知が止まらなくなります。

理由は明快で、一度バンドを割るとそのまま数日〜数週間、外側に張り付き続けることがあるからです。実際に数えてみましょう。

# バンド外にいた期間の長さを調べる
outside = (bb["percent_b"] < 0) | (bb["percent_b"] > 1)
groups = (outside != outside.shift()).cumsum()
streaks = outside.groupby(groups).sum()
streaks = streaks[streaks > 0]

print(f"バンド外に出た局面 : {len(streaks)} 回")
print(f"平均の継続日数     : {float(streaks.mean()):.1f} 日")
print(f"最長の継続日数     : {int(streaks.max())} 日")
print(f"\n『外にいる日』を全部通知すると: {int(outside.sum())} 件")
print(f"『外に出た瞬間』だけなら       : {len(streaks)} 件")

この差が決定的です。「状態」ではなく「変化」を通知する——これだけで件数が数分の1になります。

def detect_events(bb, squeeze_pct=20):
    """通知に値する『変化』だけを抽出する"""
    df = bb.copy()
    pb = df["percent_b"]

    # ① バンドの外に「出た瞬間」(前日は内側だった)
    df["break_upper"] = (pb > 1) & (pb.shift(1) <= 1)
    df["break_lower"] = (pb < 0) & (pb.shift(1) >= 0)

    # ② バンドの中に「戻った瞬間」(逆張りの手仕舞い候補)
    df["back_inside"] = (pb.between(0, 1)) & (~pb.shift(1).between(0, 1))

    # ③ バンド幅が過去に比べて極端に狭い(大きな動きの前触れ)
    threshold = df["band_width"].rolling(120).quantile(squeeze_pct / 100)
    df["squeeze"] = (df["band_width"] < threshold) & \
                    (df["band_width"].shift(1) >= threshold.shift(1))

    return df


ev = detect_events(bb)
for name, label in [("break_upper", "上限突破"), ("break_lower", "下限割れ"),
                    ("back_inside", "バンド内に復帰"), ("squeeze", "収縮")]:
    print(f"{label:<14s}: {int(ev[name].sum()):>3d} 回")

3つ目の「収縮(スクイーズ)」を入れているのは、バンドが狭まっているときは大きな動きが近いとされるためです。逸脱してから慌てるより、その前に身構えられるほうが実用的でした。

ノイズを削る条件を足す

「出た瞬間」に絞ってもまだ多い場合、逸脱の大きさで足切りします。ほんの少しはみ出しただけの動きは、たいてい意味がありません。

def filter_signals(df, min_excess=0.05, min_width=None):
    """弱いシグナルを除外する

    min_excess : %Bがこの値以上はみ出していることを要求する
    min_width  : バンド幅の下限(狭すぎるバンドの逸脱は無視する)
    """
    pb = df["percent_b"]

    strong_upper = df["break_upper"] & (pb > 1 + min_excess)
    strong_lower = df["break_lower"] & (pb < -min_excess)

    if min_width is not None:
        wide_enough = df["band_width"] >= min_width
        strong_upper &= wide_enough
        strong_lower &= wide_enough

    return strong_upper, strong_lower


# 効果を確認する
up_all, low_all = ev["break_upper"], ev["break_lower"]
up_f, low_f = filter_signals(ev, min_excess=0.05,
                             min_width=float(ev["band_width"].median() * 0.7))

print(f"絞り込み前 : 上{int(up_all.sum())}件 / 下{int(low_all.sum())}件")
print(f"絞り込み後 : 上{int(up_f.sum())}件 / 下{int(low_f.sum())}件")

min_width の条件が地味に効きます。相場が凪いでいるとバンドが極端に狭くなり、わずかな動きでも簡単に逸脱するからです。この状態での通知はほぼノイズなので、除外します。

通知を送る

検知できたら通知します。ここでは受け取りやすさを優先し、状況が一目で分かる形に整えます。

import os
import requests
from dotenv import load_dotenv

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


def notify(title, lines, color=0x3498DB):
    """整形した通知を送る"""
    if not WEBHOOK_URL:
        print(f"[通知先未設定]\n{title}\n" + "\n".join(lines))
        return False

    embed = {
        "title": title,
        "description": "\n".join(lines),
        "color": color,
    }
    try:
        r = requests.post(WEBHOOK_URL, json={"embeds": }, timeout=10)
        r.raise_for_status()
        return True
    except requests.RequestException as e:
        print(f"送信失敗: {e}")
        return False


def build_message(pair_name, row):
    """通知本文を組み立てる"""
    return [
        f"**現在値** {float(row['close']):.3f}",
        f"**上限** {float(row['upper']):.3f} / "
        f"**中心** {float(row['middle']):.3f} / "
        f"**下限** {float(row['lower']):.3f}",
        f"**%B** {float(row['percent_b']):.3f}  "
        f"**バンド幅** {float(row['band_width']):.2f}%",
    ]

通知の実装そのものはDiscordにアラートをPythonで送る方法で詳しく扱っています。LINEを使いたい場合も、送信部分を差し替えるだけで同じように動きます。

同じ通知を繰り返さない

「変化」に絞っても、行ったり来たりを繰り返す局面では連発します。一定時間は同じ種類の通知を送らない仕組みを入れます。

import json
import datetime as dt
from pathlib import Path

STATE = Path(__file__).resolve().parent / "bb_state.json"


def should_send(key, cooldown_hours=12):
    """同じシグナルを短時間に繰り返さない"""
    state = json.loads(STATE.read_text(encoding="utf-8")) if STATE.exists() else {}
    now = dt.datetime.now()

    last = state.get(key)
    if last:
        hours = (now - dt.datetime.fromisoformat(last)).total_seconds() / 3600
        if hours < cooldown_hours:
            return False

    state[key] = now.isoformat()
    STATE.write_text(json.dumps(state, ensure_ascii=False, indent=1),
                     encoding="utf-8")
    return True

全体をつなげる

ここまでの部品を組み合わせて、毎日回せる形にします。

PAIRS = {
    "JPY=X": "ドル円",
    "EURJPY=X": "ユーロ円",
    "GBPJPY=X": "ポンド円",
}

COLOR_UP, COLOR_DOWN, COLOR_INFO = 0xE74C3C, 0x2ECC71, 0xF1C40F


def check_pair(symbol, name):
    """1通貨ペアをチェックして、必要なら通知する"""
    try:
        px = yf.download(symbol, period="6mo", auto_adjust=True,
                         progress=False)["Close"].dropna()
        if len(px) < 130:
            print(f"{name}: データ不足")
            return

        bb = bollinger(px)
        ev = detect_events(bb).dropna()
        up, low = filter_signals(
            ev, min_excess=0.05,
            min_width=float(ev["band_width"].median() * 0.7))

        latest = ev.iloc[-1]
        today = ev.index[-1]

        if bool(up.iloc[-1]) and should_send(f"{symbol}:upper"):
            notify(f"📈 {name} 上限突破", build_message(name, latest), COLOR_UP)
            print(f"{name}: 上限突破を通知しました")

        elif bool(low.iloc[-1]) and should_send(f"{symbol}:lower"):
            notify(f"📉 {name} 下限割れ", build_message(name, latest), COLOR_DOWN)
            print(f"{name}: 下限割れを通知しました")

        elif bool(latest["squeeze"]) and should_send(f"{symbol}:squeeze", 48):
            notify(f"⚡ {name} バンド収縮",
                   build_message(name, latest) +
                   ["", "変動が小さくなっています。大きな動きに備えてください"],
                   COLOR_INFO)
            print(f"{name}: 収縮を通知しました")

        else:
            print(f"{name}: 変化なし (%B={float(latest['percent_b']):.3f})")

    except Exception as e:
        print(f"{name}: エラー {e}")


if __name__ == "__main__":
    print(f"=== {dt.datetime.now():%Y-%m-%d %H:%M} チェック開始 ===")
    for symbol, name in PAIRS.items():
        check_pair(symbol, name)

1通貨ペアの処理を try/except で囲んでいるのがポイントです。1つの取得に失敗しても、他のペアのチェックは続きます。自動実行では、一部の失敗で全体が止まらない設計が重要です。

これを毎日決まった時刻に動かせば完成です。設定方法はcronで毎日自動実行する方法Windowsタスクスケジューラでの自動実行にまとめています。

実行タイミングについては、通貨ペアの性質を踏まえて決めます。株式と違い為替は24時間動いているので、「大引け後」に相当する明確な区切りがありません。

実行時刻狙い向いている使い方
朝7時前後NY市場の終了後前日の総括として確認する
夕方17時前後欧州市場の開始前これからの動きに備える
21時前後NY市場の開始前最も動く時間帯の直前
1時間ごと常時監視通知が増えるので絞り込み必須

僕は朝7時と21時の1日2回にしています。日足で判断している以上、それ以上の頻度で確認しても新しい情報は増えません。チェック頻度は使っている足の粒度に合わせる——これが無駄な通知を減らす基本です。

運用してみて分かったこと

通知は「判断材料」であって「指示」ではない

最初は「−2σ割れ=買い」と機械的に考えていましたが、これは危険でした。下落トレンドの最中なら、割ったまま延々と下げ続けます。

いまは通知にトレンドの向きも一緒に載せるようにしています。逆張りしてよい局面かを、通知を見た時点で判断できるようにするためです。

def add_context(df, ma_long=75):
    """通知に添えるトレンド情報を作る"""
    ma = df["close"].rolling(ma_long).mean()
    above = df["close"] > ma

    latest_above = bool(above.iloc[-1])
    ratio = float(above.tail(ma_long).mean())

    if ratio > 0.75:
        trend = "強い上昇基調"
    elif ratio < 0.25:
        trend = "強い下降基調"
    else:
        trend = "方向感なし(レンジ)"

    hint = "逆張りが機能しやすい局面です" if 0.25 < ratio < 0.75 \
        else "トレンド中のため逆張りは危険です"

    return [f"**トレンド** {trend}", f"_{hint}_"]

この一言があるだけで、通知を見た瞬間の判断がかなり変わります。相場環境による使い分けは市場レジーム検出で戦略を自動切り替えする方法でも扱っています。

為替では指標発表に注意する

雇用統計やFOMCの直後は、バンドを大きく逸脱します。しかしこれは「行き過ぎ」ではなく「新しい水準への移行」であることが多く、逆張りすると轢かれます。

実務的な対処は単純で、重要指標の前後はシグナルを無効にすることです。カレンダーを見て手動で止めるだけでも十分に効果があります。

期間とσの値をむやみに変えない

通知が多いと感じると、σを2.5や3に上げたくなります。しかしそれは「シグナルを減らす」だけで、質が上がるわけではありません

# σを変えたときの発生頻度を確認する
for s in (1.5, 2.0, 2.5, 3.0):
    b = bollinger(px, sigma=s).dropna()
    outside = ((b["percent_b"] < 0) | (b["percent_b"] > 1))
    print(f"σ={s}: バンド外 {int(outside.sum()):>3d}日 "
          f"({float(outside.mean()) * 100:>4.1f}%)")

数を減らしたいなら、σを上げるより本記事の「変化のみ通知」とフィルターで絞るほうが理にかなっています。σは20期間・2σという標準的な設定のまま使うのが、他人と会話が通じる意味でも無難です。

まとめ

ドル円のボリンジャーバンドを自動監視し、実用的なアラートとして仕上げる方法をまとめました。

  • バンド逸脱は逆張りとも順張りとも解釈できる。通知は判断材料であって指示ではない
  • %Bとバンド幅を一緒に計算しておくと、通知の判断材料が揃う
  • 「状態」ではなく「変化」を通知する。これだけで件数が数分の1になる
  • バンドが極端に狭いときの逸脱はほぼノイズ。バンド幅で足切りする
  • 収縮(スクイーズ)の検知を入れると、動く前に身構えられる
  • 通知にはトレンドの向きも添える。逆張りしてよい局面かが分かる
  • 通知が多いときはσを上げるのではなく、変化とフィルターで絞る

作ってみて一番実感したのは、「検知する技術」より「通知しない判断」のほうが難しいということでした。条件に合致するものを全部送るのは簡単ですが、それでは人間が処理しきれません。

いまの設定だと、ドル円で通知が来るのは週に1〜2回程度です。少ないと感じるかもしれませんが、来たときは必ず見ます。それこそが、自動化の目的だったはずだと思っています。

まずは本記事の bollinger()detect_events() をコピーして、自分が見ている通貨ペアで走らせてみてください。「バンド外にいた日」と「バンドの外に出た日」の件数を比べるだけでも、通知設計の勘所がつかめるはずです。

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