SQL 異常検知 — MAD・EWMA・連続異常の応用

WITH base AS (
  SELECT
    dt, amount,
    AVG(amount)    OVER w AS base_avg,
    STDDEV(amount) OVER w AS base_stddev
  FROM  daily_sales
  WINDOW w AS (
    ORDER BY dt
    ROWS BETWEEN 5 PRECEDING AND 1 PRECEDING  -- 当日除外: 1〜5日前
  )
)
SELECT
  dt, amount,
  ROUND(base_avg, 2) AS base_avg,
  ROUND(base_stddev, 2) AS base_stddev,
  ROUND(
    (amount - base_avg) / NULLIF(base_stddev, 0),  -- 過去基準に対する標準化
    2
  ) AS z_score,
  CASE
    WHEN ABS((amount - base_avg) / NULLIF(base_stddev, 0)) >= 3
    THEN 'anomaly'
    ELSE 'normal'
  END AS anomaly_flag
FROM  base
ORDER BY dt;

/*
  実行順序（SQLの論理的な評価順）:
  1. CTE base          → CTE を定義（過去窓でウィンドウ関数を評価）
  2. 外側クエリ        → z-score を評価し anomaly/normal を判定
  3. ORDER BY dt       → 並び替えて出力
*/

WITH med AS (                       -- ① 全期間の中央値
  SELECT PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY amount) AS median_val
  FROM  daily_sales
),
dev AS (                              -- ② 各行の中央値からの絶対偏差
  SELECT
    s.dt, s.amount, m.median_val,
    ABS(s.amount - m.median_val) AS abs_dev
  FROM  daily_sales s
  CROSS JOIN med m                    -- 1行の median を全行へ展開
),
mad AS (                              -- ③ 絶対偏差の中央値 = MAD
  SELECT PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY abs_dev) AS mad_val
  FROM  dev
)
SELECT
  d.dt, d.amount, d.median_val, m.mad_val,
  ROUND(
    (0.6745 * (d.amount - d.median_val) / NULLIF(m.mad_val, 0))::numeric,  -- 修正z-score
    2
  ) AS mod_z,
  CASE
    WHEN ABS(0.6745 * (d.amount - d.median_val) / NULLIF(m.mad_val, 0)) > 3.5
    THEN 'anomaly'
    ELSE 'normal'
  END AS flag
FROM  dev d
CROSS JOIN mad m
ORDER BY d.dt;

/*
  実行順序（SQLの論理的な評価順）:
  1. CTE med    → CTE を定義（中央値を集計）
  2. CTE dev    → CTE を定義（絶対偏差を評価）
  3. CTE mad    → CTE を定義（絶対偏差の中央値を集計）
  4. 外側クエリ → 修正z-scoreを評価し anomaly/normal を判定
  5. ORDER BY   → 並び替えて出力
*/

WITH RECURSIVE seq AS (        -- 行に連番(rn)を付与
  SELECT
    dt, amount,
    ROW_NUMBER() OVER (ORDER BY dt) AS rn
  FROM  daily_sales
),
ewma AS (
  -- ① 非再帰項（アンカー）: 最初の行は ema = amount
  SELECT rn, dt, amount, amount::numeric AS ema
  FROM  seq
  WHERE rn = 1

  UNION ALL

  -- ② 再帰項: 前回 ema を引き継いで今回 ema を計算
  SELECT
    s.rn, s.dt, s.amount,
    0.5 * s.amount + 0.5 * e.ema  -- α=0.5
  FROM  ewma e
  JOIN  seq s ON s.rn = e.rn + 1  -- 1つ後ろの行へ進む
)
SELECT
  dt, amount,
  ROUND(ema, 2) AS ewma
FROM  ewma
ORDER BY rn;

/*
  実行順序（SQLの論理的な評価順）:
  1. CTE seq            → CTE を定義（ROW_NUMBER で連番付与）
  2. CTE ewma（再帰CTE）:
     アンカー部を評価（起点）
     再帰（子を順に展開）
     追加行なしで停止
  3. 外側クエリを評価   → ROUND / ORDER BY で並び替えて出力
*/

WITH flagged AS (
  -- ① 高水準フラグを付与
  SELECT dt, amount,
    CASE WHEN amount >= 200 THEN 1 ELSE 0 END AS is_high
  FROM  daily_sales
),
rownums AS (
  -- ② 2種類の連番を計算（全体連番とis_high別連番）
  SELECT dt, amount, is_high,
    ROW_NUMBER() OVER (ORDER BY dt) AS rn_all,
    ROW_NUMBER() OVER (PARTITION BY is_high ORDER BY dt) AS rn_flag
  FROM  flagged
),
grouped AS (
  -- ③ 連番の差分で連続区間ID(grp)を算出
  SELECT dt, amount, is_high,
    (rn_all - rn_flag) AS grp
  FROM  rownums
),
runs AS (
  -- ④ 各島の長さ(連続日数)を集計
  SELECT dt, amount, is_high, grp,
    COUNT(*) OVER (PARTITION BY is_high, grp) AS run_length
  FROM  grouped
)
SELECT
  dt, amount, is_high, run_length,
  CASE
    WHEN is_high = 1 AND run_length >= 3 THEN 'sustained_anomaly'
    WHEN is_high = 1                     THEN 'noise'
    ELSE 'normal'
  END AS alert_type
FROM  runs
ORDER BY dt;

/*
  実行順序（SQLの論理的な評価順）:
  1. CTE flagged  → CTE を定義（高額フラグを付与）
  2. CTE rownums  → CTE を定義（連番を付与）
  3. CTE grouped  → CTE を定義（連番差で島を識別）
  4. CTE runs     → CTE を定義（島ごとの連続日数を集計）
  5. 外側 CASE    → 列を評価（区分を判定）
  6. ORDER BY dt  → 並び替えて出力
*/