外れ値アラートを『連続N回』エスカレーションに切り替えてアラート疲労を解消した話

1
def check_anomaly(today_value: float, baseline: float) -> bool:
2
    if today_value > baseline * 2:
3
        send_alert(f"値が異常: {today_value}")
4
        return True
5
    return False

1
月曜 値=5.2  ← baseline=2.5、 2倍超え → 通知
2
火曜 値=2.1  ← 通常範囲、何も来ない
3
水曜 値=2.4  ← 通常範囲
4
...

1
月曜 値=5.2 → トリガー（カウント開始）
2
火曜 値=4.8 → 連続 2 回目
3
水曜 値=5.0 → 連続 3 回目 → 通知！

1
import json
2
from pathlib import Path
3
from datetime import date
4

5
STATE_FILE = Path("/var/state/anomaly_state.json")
6
CONSECUTIVE_THRESHOLD = 3
7

8
def load_state() -> dict:
9
    if not STATE_FILE.exists():
10
        return {"consecutive_days": 0, "last_anomaly_date": None}
11
    return json.loads(STATE_FILE.read_text())
12

13
def save_state(state: dict):
14
    STATE_FILE.write_text(json.dumps(state, default=str))
15

20 collapsed lines
16
def check_anomaly_with_consecutive(today_value: float, baseline: float, target_date: date):
17
    state = load_state()
18
    is_anomaly = today_value > baseline * 2
19

20
    if is_anomaly:
21
        state["consecutive_days"] += 1
22
        state["last_anomaly_date"] = str(target_date)
23

24
        if state["consecutive_days"] >= CONSECUTIVE_THRESHOLD:
25
            send_alert(
26
                f"値が {state['consecutive_days']} 日連続で異常: "
27
                f"今日={today_value}, baseline={baseline}"
28
            )
29
    else:
30
        # 連続ストリークが切れた
31
        if state["consecutive_days"] >= CONSECUTIVE_THRESHOLD:
32
            send_alert(f"異常状態が解消: {state['consecutive_days']} 日連続→正常へ")
33
        state["consecutive_days"] = 0
34

35
    save_state(state)

1
# Bad
2
STATE_FILE = Path("/tmp/anomaly_state.json")  # 再起動で消える
3

4
# Good
5
STATE_FILE = Path("/var/state/anomaly_state.json")  # 永続パス

1
state["consecutive_days"] += 1
2
send_alert(...)  # ← ここで例外
3
save_state(state)  # ← 保存されない

1
state["consecutive_days"] += 1
2
save_state(state)  # 先に保存
3
try:
4
    send_alert(...)
5
except Exception:
6
    logger.exception("alert failed but state saved")

1
def is_consecutive(last_date: date, today: date, max_gap_days: int = 1) -> bool:
2
    """営業日的な連続を判定（土日を許容）"""
3
    if last_date is None:
4
        return False
5
    gap = (today - last_date).days
6
    return gap <= max_gap_days

1
if state["consecutive_days"] >= CONSECUTIVE_THRESHOLD:
2
    send_alert("異常状態が解消")  # 必須
3
state["consecutive_days"] = 0

1
TRIP_THRESHOLD = 2.0   # baseline の 2.0 倍を超えたら異常
2
RESET_THRESHOLD = 1.5  # baseline の 1.5 倍を下回ったら解消
3

4
def check(value, baseline, currently_tripped: bool) -> bool:
5
    if not currently_tripped and value > baseline * TRIP_THRESHOLD:
6
        return True
7
    if currently_tripped and value < baseline * RESET_THRESHOLD:
8
        return False
9
    return currently_tripped  # 状態維持