ドル円のアラートBotを作ったあと、実際に15分おきで回してみたら、想定していなかった問題が次々に出てきました。土日に古い価格で通知が飛ぶ。同じシグナルで4回連続鳴る。3日間止まっていたのに気づかない。
この記事は、アラートを「作る」ではなく「回し続ける」ための実装です。RSIの計算やLINE通知の書き方そのものはRSIアラートBotの記事に書いたので、こちらでは定期実行に固有の話を扱います。
最初に前提を1つ。LINE Notifyは2025年3月31日でサービスを終了しました。notify-api.line.meはもう応答しません。以前この記事で紹介していたコードも動かないので、Messaging APIかDiscord Webhookに置き換えてあります。
15分おき実行で起きる4つの問題
| 問題 | 症状 | 対策 |
|---|---|---|
| 市場が閉まっている | 土日に金曜の価格で通知 | 足の時刻の鮮度をチェック |
| 同じ足で複数回実行 | 1時間足なら4回連続で鳴る | 処理済みの足の時刻を記録 |
| 通信エラーで落ちる | 気づかず何日も停止 | 例外を握って記録、死活監視 |
| 多重起動 | 前回が終わる前に次が動く | ロックファイル |
2番目が特に厄介です。1時間足のRSIを15分おきにチェックすると、同じ足を4回評価します。状態管理を入れていても、足が確定していない間に閾値を出入りすると、そのたびに鳴ります。
足の鮮度と処理済み判定
import json
import os
from datetime import datetime, timezone
from pathlib import Path
import pandas as pd
import yfinance as yf
STATE = Path("alert_state.json")
def load_state() -> dict:
if STATE.exists():
return json.loads(STATE.read_text(encoding="utf-8"))
return {}
def save_state(d: dict) -> None:
STATE.write_text(json.dumps(d, ensure_ascii=False, indent=2), encoding="utf-8")
def load_fx(ticker: str = "JPY=X", interval: str = "1h", period: str = "10d") -> pd.DataFrame:
df = yf.download(ticker, period=period, interval=interval,
auto_adjust=False, progress=False)
if df.empty:
raise RuntimeError(f"{ticker}: データ取得失敗")
if isinstance(df.columns, pd.MultiIndex):
df.columns = df.columns.get_level_values(0)
df = df.dropna(subset=["Close"])
df.index = pd.to_datetime(df.index, utc=True)
return df.sort_index()
def is_fresh(last_bar: pd.Timestamp, interval_minutes: int = 60,
tolerance: float = 2.5) -> bool:
"""最新足が古すぎないか。休場やAPI停止を検出する。"""
age = (pd.Timestamp.now(tz=timezone.utc) - last_bar).total_seconds() / 60
return age <= interval_minutes * toleranceis_freshで休場を判定しています。曜日で判定する方法もありますが、祝日や臨時の休止を全部書くのは無理です。「データが更新されていないなら市場が動いていない」と考える方が、シンプルで漏れがありません。
def already_handled(state: dict, key: str, bar_time: pd.Timestamp) -> bool:
return state.get(key, {}).get("last_bar") == bar_time.isoformat()
def mark_handled(state: dict, key: str, bar_time: pd.Timestamp, zone: str) -> None:
state.setdefault(key, {})
state[key]["last_bar"] = bar_time.isoformat()
state[key]["zone"] = zone足の時刻をキーにして記録します。これで「同じ足では1回しか通知しない」が保証されます。冪等性と呼ばれる考え方で、定期実行するスクリプトでは必須です。何度実行しても結果が変わらないようにしておけば、実行間隔を変えても安全です。
複数通貨ペア・複数指標に対応する
import numpy as np
WATCH = {
"ドル円": {"ticker": "JPY=X", "interval": "1h"},
"ユーロ円": {"ticker": "EURJPY=X", "interval": "1h"},
"ポンド円": {"ticker": "GBPJPY=X", "interval": "1h"},
}
def rsi_wilder(close: pd.Series, period: int = 14) -> pd.Series:
delta = close.diff()
gain, loss = delta.clip(lower=0), (-delta).clip(lower=0)
ag = gain.ewm(alpha=1 / period, adjust=False, min_periods=period).mean()
al = loss.ewm(alpha=1 / period, adjust=False, min_periods=period).mean()
return (100 - 100 / (1 + ag / al.replace(0, np.nan))).fillna(100)
def macd_hist(close: pd.Series, fast: int = 12, slow: int = 26, sig: int = 9) -> pd.Series:
line = close.ewm(span=fast, adjust=False).mean() - close.ewm(span=slow, adjust=False).mean()
return line - line.ewm(span=sig, adjust=False).mean()
def evaluate(df: pd.DataFrame) -> list[str]:
close = df["Close"]
rsi = rsi_wilder(close)
hist = macd_hist(close)
if len(rsi.dropna()) < 2 or len(hist) < 2:
return []
msgs = []
r = float(rsi.iloc[-1])
if r <= 30:
msgs.append(f"RSI売られすぎ {r:.1f}")
elif r >= 70:
msgs.append(f"RSI買われすぎ {r:.1f}")
h0, h1 = float(hist.iloc[-2]), float(hist.iloc[-1])
if h0 < 0 <= h1:
msgs.append("MACDゴールデンクロス")
elif h0 > 0 >= h1:
msgs.append("MACDデッドクロス")
return msgsMACDをtaライブラリではなく素のpandasで書いています。外部ライブラリを1つ減らすと、バージョン差で動かなくなる可能性がそのぶん減ります。定期実行するスクリプトでは、依存は少ない方が安全でした。
落ちない・気づける本体
import logging
import traceback
logging.basicConfig(
filename="alert.log",
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
encoding="utf-8",
)
def run_once() -> None:
state = load_state()
errors = []
for name, cfg in WATCH.items():
try:
df = load_fx(cfg["ticker"], cfg["interval"])
bar = df.index[-1]
if not is_fresh(bar):
logging.info("%s: 休場または停止中 (last=%s)", name, bar)
continue
if already_handled(state, name, bar):
logging.info("%s: 処理済みの足 (%s)", name, bar)
continue
msgs = evaluate(df)
price = float(df["Close"].iloc[-1])
if msgs:
text = (f"【{name}】{price:.3f}\n"
+ "\n".join(msgs)
+ f"\n足: {bar.tz_convert('Asia/Tokyo'):%m/%d %H:%M}")
notify(text)
logging.info("通知: %s", text.replace("\n", " / "))
mark_handled(state, name, bar, ",".join(msgs))
except Exception as e:
errors.append(f"{name}: {type(e).__name__} {e}")
logging.error("%s\n%s", name, traceback.format_exc())
state["_last_run"] = datetime.now(timezone.utc).isoformat()
state["_last_errors"] = errors
save_state(state)通貨ペアごとにtryで囲んでいるのが要点です。1つのペアでエラーが出ても、残りは処理されます。まとめて囲むと、ユーロ円の取得失敗でドル円の通知まで止まります。
_last_runを残しているのは、次の死活監視のためです。
止まっていることに気づく
通知が来ないとき、それが「シグナルがない」のか「Botが死んでいる」のかは区別できません。1日1回、生きていることを通知します。
def heartbeat(state: dict, hours: int = 24) -> None:
last = state.get("_heartbeat")
now = datetime.now(timezone.utc)
if last and (now - datetime.fromisoformat(last)).total_seconds() < hours * 3600:
return
errs = state.get("_last_errors") or []
notify(
f"[稼働確認] アラートBotは動作中です\n"
f"監視: {len(WATCH)}ペア\n"
f"直近のエラー: {len(errs)}件"
+ ("\n" + "\n".join(errs[:3]) if errs else "")
)
state["_heartbeat"] = now.isoformat()毎朝1通、「動いています」という通知が来ます。それが来なかった日は、Botか自宅PCが止まっています。地味ですが、これを入れてから停止に気づけるようになりました。以前は3日ほど止まったまま「今週は静かだな」と思っていたことがあります。
多重起動を防ぐ
LOCK = Path("alert.lock")
def acquire_lock(max_age_sec: int = 600) -> bool:
if LOCK.exists():
age = datetime.now().timestamp() - LOCK.stat().st_mtime
if age < max_age_sec:
logging.warning("前回の実行が継続中のためスキップ")
return False
LOCK.unlink() # 古いロックは異常終了とみなす
LOCK.write_text(str(os.getpid()))
return True
def release_lock() -> None:
LOCK.unlink(missing_ok=True)
if __name__ == "__main__":
if acquire_lock():
try:
run_once()
heartbeat(load_state())
finally:
release_lock()古いロックを期限で無効化しているのが大事です。異常終了でロックが残ると、それ以降ずっとスキップされます。安全装置が原因で止まるのが、この手の仕組みでいちばん気づきにくい故障です。
タスクスケジューラで15分おきに実行する
Windowsのタスクスケジューラは「基本タスクの作成」だと15分間隔を選べません。詳細設定から繰り返しを指定します。
- タスクスケジューラを開き、「タスクの作成」（基本タスクではない方）を選ぶ
- 「全般」でユーザーがログオンしていなくても実行するにチェック
- 「トリガー」→新規→「毎日」を選び、「繰り返し間隔」を15分、「継続時間」を無期限に設定
- 「操作」でバッチファイルを指定
- 「設定」で「タスクを停止するまでの時間」を1時間に短縮しておく
@echo off
cd /d C:\work\fx-alert
call .venv\Scripts\activate.bat
python alert.py
exit /b %errorlevel%作業ディレクトリをcd /dで明示しないと、状態ファイルやログがSystem32配下に作られます。手動実行では動くのにタスクからだと動かない、という定番のハマりどころです。
macOSやLinuxならcronで*/15 * * * *です。常時稼働させたいならVPSですが、通知だけのBotなら自宅PCで十分でした。落ちても実害はありません。
通知の送信部分
import requests
def notify(text: str) -> bool:
"""LINE Messaging API。未設定ならDiscord、それもなければ標準出力。"""
token, uid = os.getenv("LINE_CHANNEL_ACCESS_TOKEN"), os.getenv("LINE_USER_ID")
if token and uid:
r = requests.post(
"https://api.line.me/v2/bot/message/push",
headers={"Authorization": f"Bearer {token}"},
json={"to": uid, "messages": [{"type": "text", "text": text[:900]}]},
timeout=15,
)
if r.status_code == 200:
return True
logging.error("LINE失敗 %s %s", r.status_code, r.text[:200])
hook = os.getenv("DISCORD_WEBHOOK_URL")
if hook:
return requests.post(hook, json={"content": text[:1900]},
timeout=15).status_code in (200, 204)
print(text)
return FalseLINEが失敗したらDiscordに落ちる二段構えにしています。通知が本命なので、経路は複数持っておく方が安心です。トークンはos.getenvで読み、コードに直書きしません。
まとめ
アラートを作るのは1〜2時間ですが、回し続けるにはもう少し手が要りました。押さえたのは5つです。
足の鮮度で休場を判定する。足の時刻を記録して同じ足で二重通知しない。通貨ペアごとに例外を握って全体を止めない。1日1回のハートビートで停止に気づく。ロックファイルで多重起動を防ぐ。
そしてLINE Notifyは終了しているので、Messaging APIかDiscord Webhookに移行してください。次は複数通貨ペアの相関を見て、同じ材料で全部鳴るのを抑える仕組みを試す予定です。
