Docker + JupyterLab + Plotlyで時系列データの異常値検知 Step 1〜3 を試す
はじめに
前回の記事からStep 1〜3 を考えてみます。
時系列データの異常値検知では、いきなり高度な機械学習モデルを使うよりも、まずは説明しやすい方法から始めた方が扱いやすいです。
特に、雨量・水位・流量・設備センサーのようなデータでは、最初に見るべきことは意外と地味です。
- 時刻が抜けていないか
- 欠測や重複がないか
- 物理的にあり得ない値が入っていないか
- 瞬間的なスパイクがないか
- 時間帯や周期を考えると不自然な値がないか
このあたりをきちんと見ておくと、後から予測モデルやAIモデルを追加するときにも整理しやすくなります。
今回は、第1段階のプロトタイプとして、次の3つを Docker 環境で動かします。
| Step | 内容 | 目的 |
|---|---|---|
| Step 1 | データ品質チェック | 欠測・重複・時刻抜けなど、入力データ自体の問題を確認する |
| Step 2 | ロバスト統計 | 移動中央値やMADを使って、スパイクや急変を検出する |
| Step 3 | 周期性・季節性対応 | 時間帯や周期を考えると不自然な値を検出する |
時系列グラフは Plotly を使い、Web画面とJupyterLabの両方で拡大・縮小・ホバー確認ができるようにします。
今回作るもの
構成は、Web画面、FastAPI、JupyterLabを Docker Compose でまとめて起動する形です。
timeseries_anomaly_step1_3_docker_jupyter_r004/
├─ app/
│ └─ main.py
├─ src/
│ └─ timeseries_anomaly/
│ ├─ pipeline.py
│ ├─ steps/
│ │ ├─ data_quality.py
│ │ ├─ robust_detector.py
│ │ └─ seasonal_detector.py
│ ├─ common/
│ │ ├─ io_utils.py
│ │ └─ plotting.py
│ └─ future/
│ ├─ step4_forecast_detector.py
│ └─ step5_multivariate_detector.py
├─ web/
│ ├─ index.html
│ ├─ style.css
│ └─ app.js
├─ notebooks/
│ └─ 01_step1_3_analysis.ipynb
├─ docker/
│ ├─ Dockerfile.api
│ ├─ Dockerfile.web
│ ├─ Dockerfile.jupyter
│ └─ nginx.conf
├─ data/
│ ├─ raw/
│ ├─ processed/
│ └─ sample/
│ └─ sample_timeseries.csv
├─ outputs/
├─ reports/
├─ configs/
│ └─ default_config.json
├─ scripts/
│ ├─ run_sample_analysis.py
│ └─ export_interactive_html.py
├─ docker-compose.yml
├─ requirements.txt
├─ run_app.bat
├─ stop_app.bat
└─ rebuild_app.bat
future/ は、今後 Step 4 の予測残差ベース、Step 5 の多変量・AI拡張を追加するための置き場所です。
最初から全部を詰め込まず、まず Step 1〜3 を安定して動かす方針にしています。
Dockerサービス構成
| サービス | 内容 | URL |
|---|---|---|
| web | ブラウザ画面 | http://localhost:8080 |
| api | FastAPI | http://localhost:8000/docs |
| jupyterlab | Notebook分析環境 | http://localhost:8888/lab?token=timeseries |
data/、outputs/、reports/ は共有フォルダーにしています。
Web画面で実行した結果を outputs/ に保存し、JupyterLabで読み直すこともできます。
Notebookで作ったPlotlyグラフを reports/ にHTML保存する使い方も想定しています。
docker-compose.yml
version: "3.9"
services:
api:
# FastAPIで異常検知パイプラインを実行するサービスです。
# Web画面からCSVをアップロードしたり、サンプルCSVを実行したりすると、
# このAPIがStep 1〜3をまとめて処理します。
build:
context: .
dockerfile: docker/Dockerfile.api
container_name: ts-anomaly-api
ports:
- "${API_PORT:-8000}:8000"
volumes:
# srcをマウントしておくと、Pythonコードを修正したあとに確認しやすくなります。
- ./src:/workspace/src
- ./app:/workspace/app
# データと出力をホスト側と共有します。
- ./data:/workspace/data
- ./outputs:/workspace/outputs
- ./reports:/workspace/reports
environment:
- PYTHONPATH=/workspace/src
command: >
uvicorn app.main:app
--host 0.0.0.0
--port 8000
--reload
web:
# 静的HTML/CSS/JavaScriptをnginxで配信するサービスです。
# Plotly.jsを使って、ブラウザ上でインタラクティブな時系列グラフを表示します。
build:
context: .
dockerfile: docker/Dockerfile.web
container_name: ts-anomaly-web
ports:
- "${WEB_PORT:-8080}:80"
volumes:
- ./web:/usr/share/nginx/html:ro
depends_on:
- api
jupyterlab:
# Notebook分析用のサービスです。
# Web画面と同じsrc/data/outputs/reportsを共有するため、
# 画面で見た結果をNotebook側でも追跡できます。
build:
context: .
dockerfile: docker/Dockerfile.jupyter
container_name: ts-anomaly-jupyter
ports:
- "${JUPYTER_PORT:-8888}:8888"
volumes:
- ./src:/workspace/src
- ./notebooks:/workspace/notebooks
- ./data:/workspace/data
- ./outputs:/workspace/outputs
- ./reports:/workspace/reports
environment:
- PYTHONPATH=/workspace/src
command: >
jupyter lab
--ip=0.0.0.0
--port=8888
--no-browser
--allow-root
--NotebookApp.token=${JUPYTER_TOKEN:-timeseries}
ポイントは、API、Web、JupyterLabが同じ data/ と outputs/ を見ていることです。
この構成にしておくと、Web画面でざっと確認し、気になる部分をNotebookで掘り下げる流れにしやすくなります。
起動方法
ZIPファイルをダウンロードします。
Docker Desktopを起動してから、プロジェクトフォルダーで以下を実行します。
run_app.bat
または、直接Docker Composeを使います。
docker compose up --build
起動後、以下を開きます。
Web画面:
http://localhost:8080
API Docs:
http://localhost:8000/docs
JupyterLab:
http://localhost:8888/lab?token=timeseries
Step 1：データ品質チェック
考え方
異常値検知というと、すぐにモデルやアルゴリズムを考えたくなります。
ただ、実データではその前に、欠測や時刻抜け、重複、物理的にあり得ない値を確認する必要があります。
例えば、10分間隔の水位データで30分空いている場合、その区間はモデル以前に「時刻抜け」として扱った方が自然です。
また、水位や雨量に負値が入っている場合も、まずは品質フラグとして残します。
data_quality.py
"""Step 1: データ品質チェック。
時系列データをモデルに入れる前に、まずデータそのものの状態を確認します。
この段階で、欠測、重複、時刻抜け、物理範囲外、長時間同じ値が続く状態を
品質フラグとして記録しておきます。
品質フラグを残しておくと、後段のStep 2・Step 3で検出された異常と、
入力データ由来の問題を分けて確認できます。
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
import numpy as np
import pandas as pd
@dataclass
class QualityConfig:
"""データ品質チェック用の設定。
timestamp_col:
入力CSVの時刻列名です。
例: "timestamp", "datetime", "time" など。
value_col:
異常検知の対象にする値の列名です。
例: "value", "water_level", "rainfall" など。
freq_minutes:
期待する時系列間隔です。
10分データなら10、1時間データなら60を指定します。
lower_limit / upper_limit:
物理的な下限・上限です。
雨量なら下限0、水位なら観測所ごとの上限値などを設定します。
チェック不要の場合は None にします。
flatline_points:
同じ値がこの点数以上続いた場合に FLATLINE とします。
センサー停止や通信側の値固定を見つけるための簡易チェックです。
"""
timestamp_col: str = "timestamp"
value_col: str = "value"
freq_minutes: int = 10
lower_limit: float | None = 0.0
upper_limit: float | None = 3.0
flatline_points: int = 12
def _append_flag(flag_text: str, new_flag: str) -> str:
"""品質フラグ文字列に新しいフラグを追加します。
複数の問題が同じ行に出ることがあるため、
フラグは "A;B;C" のようにセミコロン区切りで持たせます。
例:
OK → MISSING_VALUE
MISSING_VALUE → MISSING_VALUE;ABOVE_UPPER_LIMIT
"""
if not flag_text or flag_text == "OK":
return new_flag
parts = flag_text.split(";")
if new_flag in parts:
# 同じフラグを二重に付けないようにします。
return flag_text
return f"{flag_text};{new_flag}"
def prepare_timeseries(df: pd.DataFrame, config: QualityConfig) -> pd.DataFrame:
"""時刻列と値列を標準化し、時刻順に並べ替えます。
入力CSVでは列名がプロジェクトごとに変わりやすいため、
内部処理では "timestamp" と "value" にそろえています。
ここではまだ異常判定はせず、後続処理が扱いやすい形へ整えるだけです。
"""
if config.timestamp_col not in df.columns:
raise ValueError(f"時刻列 '{config.timestamp_col}' が見つかりません。")
if config.value_col not in df.columns:
raise ValueError(f"値列 '{config.value_col}' が見つかりません。")
out = df.copy()
# 後続処理では固定列名で扱いたいので、時刻列と値列を標準名にそろえます。
out = out.rename(columns={
config.timestamp_col: "timestamp",
config.value_col: "value",
})
# 時刻変換に失敗した値は NaT になります。
# 変換失敗そのものは、この後の check_data_quality() でフラグ化します。
out["timestamp"] = pd.to_datetime(out["timestamp"], errors="coerce")
# 数値変換できない値は NaN にします。
# 例えば文字列や空欄が混じっていても、後で MISSING_VALUE として確認できます。
out["value"] = pd.to_numeric(out["value"], errors="coerce")
# 時刻順に並べ替えます。
# mergesort は安定ソートなので、同じ時刻がある場合に元の順序を比較的保ちやすいです。
out = out.sort_values("timestamp", kind="mergesort").reset_index(drop=True)
return out
def check_data_quality(
df: pd.DataFrame,
config: QualityConfig,
) -> tuple[pd.DataFrame, dict[str, Any]]:
"""データ品質チェックを実行します。
この関数で見るのは、いわゆる「モデルで見つける異常」ではなく、
入力データとして先に確認しておきたい問題です。
検出する主な内容:
- 時刻変換失敗
- 値の欠測
- 時刻の重複
- 期待時間間隔からの時刻抜け
- 物理下限・上限の超過
- 同一値が長時間続く状態
戻り値:
品質フラグを付けたDataFrameと、件数をまとめたsummaryを返します。
"""
out = df.copy()
out["quality_flags"] = "OK"
# 時刻が変換できなかった行です。
invalid_time = out["timestamp"].isna()
# 値が欠測、または数値化できなかった行です。
missing_value = out["value"].isna()
out.loc[invalid_time, "quality_flags"] = out.loc[invalid_time, "quality_flags"].apply(
lambda x: _append_flag(x, "INVALID_TIMESTAMP")
)
out.loc[missing_value, "quality_flags"] = out.loc[missing_value, "quality_flags"].apply(
lambda x: _append_flag(x, "MISSING_VALUE")
)
# 同じ時刻が複数ある場合は、重複している全行にフラグを付けます。
# keep=False にすることで、1件目だけでなく重複対象すべてを確認できます。
duplicated = out["timestamp"].duplicated(keep=False) & out["timestamp"].notna()
out.loc[duplicated, "quality_flags"] = out.loc[duplicated, "quality_flags"].apply(
lambda x: _append_flag(x, "DUPLICATE_TIMESTAMP")
)
# 物理的な下限・上限を確認します。
# 雨量なら下限0、水位なら観測所ごとの上限値などを想定します。
if config.lower_limit is not None:
below = out["value"] < config.lower_limit
out.loc[below, "quality_flags"] = out.loc[below, "quality_flags"].apply(
lambda x: _append_flag(x, "BELOW_LOWER_LIMIT")
)
if config.upper_limit is not None:
above = out["value"] > config.upper_limit
out.loc[above, "quality_flags"] = out.loc[above, "quality_flags"].apply(
lambda x: _append_flag(x, "ABOVE_UPPER_LIMIT")
)
# 時刻抜けを確認します。
# 例えば10分間隔のデータで30分空いた場合、ギャップ直後の行にフラグを付けます。
expected_delta = pd.Timedelta(minutes=config.freq_minutes)
valid_unique = (
out.dropna(subset=["timestamp"])
.drop_duplicates(subset=["timestamp"], keep="first")
.sort_values("timestamp")
.reset_index()
)
time_gaps: list[dict[str, str | float]] = []
if len(valid_unique) >= 2:
diffs = valid_unique["timestamp"].diff()
# 少し余裕を持たせるため、期待間隔の1.5倍を超えた場合にギャップとします。
gap_positions = np.where(diffs > expected_delta * 1.5)[0]
for pos in gap_positions:
current_ts = valid_unique.loc[pos, "timestamp"]
prev_ts = valid_unique.loc[pos - 1, "timestamp"]
original_index = valid_unique.loc[pos, "index"]
out.loc[original_index, "quality_flags"] = _append_flag(
out.loc[original_index, "quality_flags"],
"TIME_GAP_BEFORE",
)
time_gaps.append({
"before": prev_ts.isoformat(),
"after": current_ts.isoformat(),
"gap_minutes": float((current_ts - prev_ts).total_seconds() / 60.0),
})
# 長時間同じ値が続く状態を検出します。
# センサー値が固定されたままになるケースを拾うための簡易チェックです。
flatline = pd.Series(False, index=out.index)
if config.flatline_points > 1 and len(out) >= config.flatline_points:
# 微小な丸め誤差でグループが分かれないよう、いったん丸めます。
values = out["value"].round(6)
# 値が変わった場所でグループIDを増やします。
group_id = values.ne(values.shift()).cumsum()
# 各連続区間の長さを求めます。
run_length = values.groupby(group_id).transform("size")
flatline = values.notna() & (run_length >= config.flatline_points)
out.loc[flatline, "quality_flags"] = out.loc[flatline, "quality_flags"].apply(
lambda x: _append_flag(x, "FLATLINE")
)
out["quality_is_anomaly"] = out["quality_flags"].ne("OK")
summary = {
"total_rows": int(len(out)),
"invalid_timestamp_count": int(invalid_time.sum()),
"missing_value_count": int(missing_value.sum()),
"duplicate_timestamp_count": int(duplicated.sum()),
"time_gap_count": int(len(time_gaps)),
"flatline_count": int(flatline.sum()),
"time_gaps": time_gaps[:20],
"quality_anomaly_count": int(out["quality_is_anomaly"].sum()),
}
return out, summary
Step 2：ロバスト統計による異常検知
考え方
平均と標準偏差を使った外れ値検出は分かりやすいのですが、外れ値自身に平均値や標準偏差が引っ張られることがあります。
そこで、この段階では以下を使います。
- 移動中央値
- MAD
- 前時刻との差分
- 差分に対するMADスコア
MAD は Median Absolute Deviation の略です。
平均ではなく中央値を基準にするため、スパイクのような値に比較的強くなります。
robust_detector.py
"""Step 2: ロバスト統計による簡易異常検知。
ここでは、移動中央値、MAD、前時刻との差分を使います。
目的は、単発のスパイクや急な落ち込みを、なるべく説明しやすい形で拾うことです。
平均値と標準偏差を使う方法もありますが、外れ値が混じると基準値自体が動いてしまうため、
第1段階のプロトタイプでは中央値とMADを使う方が扱いやすいです。
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
import numpy as np
import pandas as pd
@dataclass
class RobustConfig:
"""ロバスト統計検出用の設定。
rolling_window:
移動中央値とMADを計算する窓幅です。
10分データで13点なら、前後およそ1時間程度を見ます。
mad_threshold:
観測値が移動中央値からどれくらい外れたら異常とするかのしきい値です。
diff_score_threshold:
前時刻との差分がどれくらい大きければ急変とみなすかのしきい値です。
"""
rolling_window: int = 13
mad_threshold: float = 3.5
diff_score_threshold: float = 6.0
def _safe_mad_score(
value: pd.Series,
center: pd.Series,
mad: pd.Series,
) -> pd.Series:
"""MADを使ったロバストスコアを計算します。
score = |value - center| / (1.4826 * MAD)
1.4826 は、MADを標準偏差相当に近づけるためによく使われる係数です。
MADが0の場所ではゼロ割が起きるため、いったんNaNに置き換えてから0で埋めます。
"""
diff = (value - center).abs()
# MADが0の場所は計算できないため、NaNにしてから後で0埋めします。
mad_safe = mad.replace(0, np.nan)
score = diff / (1.4826 * mad_safe)
# infやNaNはグラフ表示や後続処理で扱いづらいので0にそろえます。
return score.replace([np.inf, -np.inf], np.nan).fillna(0.0)
def detect_robust_anomalies(
df: pd.DataFrame,
config: RobustConfig,
) -> tuple[pd.DataFrame, dict[str, Any]]:
"""移動中央値・MAD・急変チェックによる異常検知を実行します。
入力は Step 1 の結果DataFrameです。
この関数では、元の値を消さずに、以下の列を追加します。
- rolling_median
- rolling_mad
- robust_score
- diff_score
- robust_is_anomaly
"""
out = df.copy()
value = out["value"]
# 窓幅は最低3点にします。
# また、center=Trueで前後対称に見るため、偶数なら奇数へ補正します。
window = max(3, int(config.rolling_window))
if window % 2 == 0:
window += 1
# 移動中央値を通常値の目安として使います。
rolling_median = value.rolling(
window=window,
center=True,
min_periods=1,
).median()
# 移動中央値からの絶対偏差を求め、その中央値をMADとして使います。
abs_dev = (value - rolling_median).abs()
rolling_mad = abs_dev.rolling(
window=window,
center=True,
min_periods=1,
).median()
robust_score = _safe_mad_score(value, rolling_median, rolling_mad)
# 値そのものの外れだけでなく、前時刻からの急変も見ます。
# 例えば、水位の絶対値は正常範囲でも、変化量が急すぎる場合を拾うためです。
diff = value.diff().abs()
diff_center = diff.rolling(
window=window,
center=True,
min_periods=1,
).median()
diff_mad = (diff - diff_center).abs().rolling(
window=window,
center=True,
min_periods=1,
).median()
diff_score = _safe_mad_score(diff, diff_center, diff_mad)
out["rolling_median"] = rolling_median
out["rolling_mad"] = rolling_mad
out["robust_score"] = robust_score
out["diff_score"] = diff_score
# どちらかのスコアがしきい値を超えたら、Step 2の異常候補とします。
out["robust_is_anomaly"] = (
(robust_score > config.mad_threshold)
| (diff_score > config.diff_score_threshold)
)
summary = {
"rolling_window": int(window),
"mad_threshold": float(config.mad_threshold),
"diff_score_threshold": float(config.diff_score_threshold),
"robust_anomaly_count": int(out["robust_is_anomaly"].sum()),
}
return out, summary
Step 3：周期性・季節性を考慮した異常検知
考え方
値だけを見ると普通でも、時間帯や周期を考えると不自然な値があります。
例えば、水位が 1.2 m という値だけを見ると正常でも、大雨が降っている時間帯にまったく水位が反応しない場合は、文脈的におかしい可能性があります。
このStepでは、次の2つを使います。
- 周期位置ごとの中央値
- STL分解による期待値と残差
seasonal_detector.py
"""Step 3: 周期性・季節性を考慮した異常検知。
ここでは、値そのものの大きさだけではなく、
「その時刻や周期を考えると自然か」を見ます。
例えば10分データで1日周期を見る場合、1日は144点です。
同じ周期位置の過去値から通常値を作り、そこから外れているかを確認します。
さらに、statsmodelsのSTL分解が使える場合は、トレンド+季節成分から期待値を作ります。
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
import numpy as np
import pandas as pd
try:
from statsmodels.tsa.seasonal import STL
except Exception:
# statsmodelsが入っていない環境でも、周期位置ベースの検出だけは動かせるようにします。
STL = None
@dataclass
class SeasonalConfig:
"""周期性・季節性検出用の設定。
period:
周期の点数です。
10分データで1日周期なら 24 * 6 = 144 です。
seasonal_threshold:
STL残差スコアのしきい値です。
time_context_threshold:
周期位置ごとの通常値から外れた場合のしきい値です。
use_stl:
Trueの場合、STL分解も使います。
データ数が足りない場合やstatsmodelsが無い場合は、自動的にスキップします。
"""
period: int = 144
seasonal_threshold: float = 3.5
time_context_threshold: float = 3.5
use_stl: bool = True
def _robust_zscore(series: pd.Series) -> pd.Series:
"""中央値とMADを使ってロバストなZスコアを計算します。
通常のZスコアは平均と標準偏差を使いますが、
ここでは外れ値に強くするため中央値とMADを使います。
"""
median = series.median(skipna=True)
mad = (series - median).abs().median(skipna=True)
if pd.isna(mad) or mad == 0:
return pd.Series(0.0, index=series.index)
score = (series - median).abs() / (1.4826 * mad)
return score.replace([np.inf, -np.inf], np.nan).fillna(0.0)
def _time_context_expected(
df: pd.DataFrame,
period: int,
) -> tuple[pd.Series, pd.Series]:
"""周期内の位置ごとの中央値から期待値とスコアを作成します。
例:
10分データで period=144 の場合、
00:00, 00:10, 00:20 ... という1日の位置ごとに通常値を作ります。
ここでは簡易実装として、行番号から cycle_pos を作ります。
実運用では、時刻そのものから「時刻・曜日・月」などの特徴量を作る方法も使えます。
"""
out = df.copy()
# 周期内の位置を作ります。
# 10分データ・1日周期なら 0〜143 の値が繰り返されます。
out["cycle_pos"] = np.arange(len(out)) % max(2, period)
# 同じ周期位置にある値の中央値を、その位置の通常値とします。
expected = out.groupby("cycle_pos")["value"].transform("median")
# 通常値からのズレを求めます。
residual = out["value"] - expected
# 周期位置ごとに、残差がどれくらい大きいかをロバストスコア化します。
score = residual.groupby(out["cycle_pos"]).transform(_robust_zscore)
return expected, score
def _stl_residual_score(
value: pd.Series,
period: int,
use_stl: bool,
) -> tuple[pd.Series, pd.Series]:
"""STL分解による期待値と残差スコアを計算します。
STLでは、時系列をおおまかに以下へ分けます。
観測値 = トレンド + 季節成分 + 残差
このうち、トレンド + 季節成分 を期待値として扱い、
残差が大きい点を異常候補として見ます。
"""
if not use_stl or STL is None or len(value.dropna()) < period * 2:
# データが短い場合、STL分解は無理に使いません。
return pd.Series(np.nan, index=value.index), pd.Series(0.0, index=value.index)
# STLは欠測をそのまま扱いにくいため、線形補間した系列で分解します。
# 欠測そのものはStep 1でフラグ化済みなので、ここでは分解の安定性を優先します。
filled = value.interpolate(limit_direction="both")
result = STL(filled, period=period, robust=True).fit()
expected = result.trend + result.seasonal
residual = filled - expected
score = _robust_zscore(residual)
return expected, score
def detect_seasonal_anomalies(
df: pd.DataFrame,
config: SeasonalConfig,
) -> tuple[pd.DataFrame, dict[str, Any]]:
"""周期性・季節性を考慮した異常検知を実行します。
Step 2までの結果DataFrameに、以下の列を追加します。
- context_expected
- context_residual
- seasonal_score
- stl_expected
- stl_residual
- stl_score
- seasonal_is_anomaly
"""
out = df.copy()
# 周期位置ごとの通常値とスコアです。
context_expected, context_score = _time_context_expected(out, config.period)
# STL分解に基づく期待値とスコアです。
stl_expected, stl_score = _stl_residual_score(
out["value"],
config.period,
config.use_stl,
)
out["context_expected"] = context_expected
out["context_residual"] = out["value"] - context_expected
out["seasonal_score"] = context_score
out["stl_expected"] = stl_expected
out["stl_residual"] = out["value"] - stl_expected
out["stl_score"] = stl_score
# 周期位置ベース、またはSTL残差ベースのどちらかで大きく外れた場合に、
# Step 3の文脈異常候補とします。
out["seasonal_is_anomaly"] = (
(out["seasonal_score"] > config.time_context_threshold)
| (out["stl_score"] > config.seasonal_threshold)
)
summary = {
"period": int(config.period),
"time_context_threshold": float(config.time_context_threshold),
"seasonal_threshold": float(config.seasonal_threshold),
"use_stl": bool(config.use_stl and STL is not None),
"seasonal_anomaly_count": int(out["seasonal_is_anomaly"].sum()),
}
return out, summary
Step 1〜3をつなぐパイプライン
考え方
個別の検出処理をバラバラに呼ぶと、Web画面やNotebookから使いにくくなります。
そこで、Step 1〜3をまとめて実行する run_pipeline() を用意しています。
この関数が返す結果を、API、Notebook、HTML出力スクリプトで共通利用します。
pipeline.py
"""Step 1〜3をまとめて実行するパイプライン。
Web画面、FastAPI、JupyterLab、バッチ実行で同じ処理を使えるように、
ここで設定と実行順序をまとめています。
処理の流れ:
1. 時刻列・値列の標準化
2. Step 1 データ品質チェック
3. Step 2 ロバスト統計
4. Step 3 周期性・季節性対応
5. 最終異常フラグと異常理由の作成
6. 必要に応じてCSV保存
"""
from __future__ import annotations
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any
import pandas as pd
from .steps.data_quality import QualityConfig, check_data_quality, prepare_timeseries
from .steps.robust_detector import RobustConfig, detect_robust_anomalies
from .steps.seasonal_detector import SeasonalConfig, detect_seasonal_anomalies
@dataclass
class PipelineConfig:
"""異常検知パイプライン全体の設定。
この設定だけを変えれば、Web画面、Notebook、スクリプトで同じ条件を使えます。
しきい値の調整は、まずこのクラスの項目を見ると分かりやすいです。
"""
timestamp_col: str = "timestamp"
value_col: str = "value"
# Step 1: データ品質チェック
freq_minutes: int = 10
lower_limit: float | None = 0.0
upper_limit: float | None = 3.0
flatline_points: int = 12
# Step 2: ロバスト統計
rolling_window: int = 13
mad_threshold: float = 3.5
diff_score_threshold: float = 6.0
# Step 3: 周期性・季節性
seasonal_period: int = 144
seasonal_threshold: float = 3.5
time_context_threshold: float = 3.5
use_stl: bool = True
def _build_reason(row: pd.Series) -> str:
"""各行の異常理由を人が読める文字列にします。
複数のStepで異常候補になった場合は、理由をセミコロンでつなぎます。
画面のホバー表示やCSV確認で、どのStepが反応したか分かるようにしています。
"""
reasons: list[str] = []
if bool(row.get("quality_is_anomaly", False)):
reasons.append(f"Step1:{row.get('quality_flags', '')}")
if bool(row.get("robust_is_anomaly", False)):
reasons.append("Step2:ROBUST_OR_DIFF_ANOMALY")
if bool(row.get("seasonal_is_anomaly", False)):
reasons.append("Step3:SEASONAL_CONTEXT_ANOMALY")
return ";".join([r for r in reasons if r]) or "OK"
def run_pipeline(
df: pd.DataFrame,
config: PipelineConfig,
output_dir: Path | None = None,
) -> dict[str, Any]:
"""Step 1〜3の異常検知処理をまとめて実行します。
Parameters
----------
df:
入力CSVを読み込んだDataFrameです。
config:
パイプライン全体の設定です。
output_dir:
結果CSVを保存するフォルダーです。
Noneの場合は保存しません。
Returns
-------
dict[str, Any]
Web画面やNotebookで使いやすいように、summary、records、anomaly_records、
result_dfをまとめて返します。
"""
# Stepごとの設定クラスへ振り分けます。
# PipelineConfigだけを外側に見せ、内部では各Step専用の設定へ変換します。
quality_config = QualityConfig(
timestamp_col=config.timestamp_col,
value_col=config.value_col,
freq_minutes=config.freq_minutes,
lower_limit=config.lower_limit,
upper_limit=config.upper_limit,
flatline_points=config.flatline_points,
)
robust_config = RobustConfig(
rolling_window=config.rolling_window,
mad_threshold=config.mad_threshold,
diff_score_threshold=config.diff_score_threshold,
)
seasonal_config = SeasonalConfig(
period=config.seasonal_period,
seasonal_threshold=config.seasonal_threshold,
time_context_threshold=config.time_context_threshold,
use_stl=config.use_stl,
)
# Step 1 の前処理です。
# 入力列名を timestamp/value にそろえ、時刻順に並べ替えます。
prepared = prepare_timeseries(df, quality_config)
# Step 1: 品質チェック
q_df, q_summary = check_data_quality(prepared, quality_config)
# Step 2: ロバスト統計
r_df, r_summary = detect_robust_anomalies(q_df, robust_config)
# Step 3: 周期性・季節性
s_df, s_summary = detect_seasonal_anomalies(r_df, seasonal_config)
# どれか1つのStepで異常になれば、最終的な異常候補とします。
s_df["final_is_anomaly"] = (
s_df["quality_is_anomaly"]
| s_df["robust_is_anomaly"]
| s_df["seasonal_is_anomaly"]
)
# 画面やCSVで理由を確認しやすいように、文字列の説明を作ります。
s_df["anomaly_reason"] = s_df.apply(_build_reason, axis=1)
output_path: str | None = None
if output_dir is not None:
output_dir.mkdir(parents=True, exist_ok=True)
filename = pd.Timestamp.now().strftime("anomaly_result_%Y%m%d_%H%M%S.csv")
out_path = output_dir / filename
# Excelで開くことも想定して utf-8-sig にしています。
s_df.to_csv(out_path, index=False, encoding="utf-8-sig")
output_path = str(out_path)
# Web画面へ返す列を絞ります。
# 全列を返すと重くなるため、表示に使う列だけにしています。
display_cols = [
"timestamp",
"value",
"quality_flags",
"quality_is_anomaly",
"rolling_median",
"robust_score",
"diff_score",
"robust_is_anomaly",
"context_expected",
"context_residual",
"seasonal_score",
"stl_expected",
"stl_score",
"seasonal_is_anomaly",
"final_is_anomaly",
"anomaly_reason",
]
display_cols = [c for c in display_cols if c in s_df.columns]
records_df = s_df[display_cols].copy()
# JSONで返しやすいように、timestampは文字列へ変換します。
if pd.api.types.is_datetime64_any_dtype(records_df["timestamp"]):
records_df["timestamp"] = records_df["timestamp"].dt.strftime("%Y-%m-%d %H:%M:%S")
# NaNはJSONのnullに近い形で扱えるよう、Noneへ寄せます。
records_df = records_df.where(pd.notnull(records_df), None)
# Web画面で重くなりすぎないよう、全体表示は末尾3000件にしています。
anomaly_records = (
records_df[records_df["final_is_anomaly"] == True]
.head(500)
.to_dict(orient="records")
)
all_records = records_df.tail(3000).to_dict(orient="records")
summary = {
"config": asdict(config),
"total_rows": int(len(s_df)),
"final_anomaly_count": int(s_df["final_is_anomaly"].sum()),
"quality": q_summary,
"robust": r_summary,
"seasonal": s_summary,
"output_path": output_path,
}
return {
"summary": summary,
"records": all_records,
"anomaly_records": anomaly_records,
"result_df": s_df,
}
Plotlyでインタラクティブに確認する
考え方
異常値検知では、異常点の前後を見たいことが多いです。
そのため、グラフは静止画よりも、拡大・移動・ホバー確認ができる方が便利です。
今回の構成では、JupyterLab側では Plotly Python、Web画面側では Plotly.js を使っています。
plotting.py
"""Plotlyによるインタラクティブ可視化ユーティリティ。
JupyterLab上で時系列グラフを拡大・縮小・パン・ホバー確認できるようにするための
共通関数をまとめています。
Web画面では Plotly.js を使い、Notebookではこのモジュールを使います。
表示する系列名やホバー内容をなるべくそろえておくと、
Web画面とNotebookで見比べやすくなります。
"""
from __future__ import annotations
from pathlib import Path
import pandas as pd
import plotly.graph_objects as go
def create_interactive_timeseries_figure(
df: pd.DataFrame,
timestamp_col: str = "timestamp",
value_col: str = "value",
title: str = "Step 1〜3 異常検知結果",
) -> go.Figure:
"""Step 1〜3の結果DataFrameからPlotly Figureを作成します。
表示する主な系列:
- 観測値
- 移動中央値
- 周期文脈の期待値
- STL期待値
- 異常候補
戻り値は Plotly Figure なので、Notebook上では fig.show() で表示できます。
"""
plot_df = df.copy()
# 念のため、時刻列をdatetime型に変換します。
# 文字列のままでもPlotlyは表示できますが、datetime型の方が範囲操作が安定します。
plot_df[timestamp_col] = pd.to_datetime(plot_df[timestamp_col], errors="coerce")
fig = go.Figure()
# 観測値です。まずは元データの形を確認できるよう、線で表示します。
fig.add_trace(
go.Scatter(
x=plot_df[timestamp_col],
y=plot_df[value_col],
mode="lines",
name="観測値",
line=dict(width=2.5, color="#0f63ce"),
hovertemplate="時刻: %{x}<br>観測値: %{y:.4f}<extra></extra>",
)
)
# Step 2で使った移動中央値です。
# 観測値と重ねると、スパイクや局所的なズレを確認しやすくなります。
if "rolling_median" in plot_df.columns:
fig.add_trace(
go.Scatter(
x=plot_df[timestamp_col],
y=plot_df["rolling_median"],
mode="lines",
name="移動中央値",
line=dict(width=2, color="#64748b", dash="dash"),
hovertemplate="時刻: %{x}<br>移動中央値: %{y:.4f}<extra></extra>",
)
)
# Step 3の周期位置ベースの期待値です。
# 初期表示で線が多くなりすぎないよう、legendonlyにしています。
if "context_expected" in plot_df.columns:
fig.add_trace(
go.Scatter(
x=plot_df[timestamp_col],
y=plot_df["context_expected"],
mode="lines",
name="周期文脈の期待値",
line=dict(width=1.8, color="#16a34a", dash="dot"),
visible="legendonly",
hovertemplate="時刻: %{x}<br>周期文脈期待値: %{y:.4f}<extra></extra>",
)
)
# STL分解から作った期待値です。
# こちらも必要なときに凡例からONにする想定です。
if "stl_expected" in plot_df.columns:
fig.add_trace(
go.Scatter(
x=plot_df[timestamp_col],
y=plot_df["stl_expected"],
mode="lines",
name="STL期待値",
line=dict(width=1.8, color="#f97316", dash="dashdot"),
visible="legendonly",
hovertemplate="時刻: %{x}<br>STL期待値: %{y:.4f}<extra></extra>",
)
)
# 最終的な異常候補です。
# ホバーに理由とスコアを入れておくと、グラフ上で原因を追いやすくなります。
if "final_is_anomaly" in plot_df.columns:
anomaly_df = plot_df[plot_df["final_is_anomaly"]].copy()
hover_text = anomaly_df.apply(
lambda r: (
f"時刻: {r.get(timestamp_col)}<br>"
f"値: {r.get(value_col)}<br>"
f"理由: {r.get('anomaly_reason', '')}<br>"
f"robust: {r.get('robust_score', '')}<br>"
f"seasonal: {r.get('seasonal_score', '')}<br>"
f"stl: {r.get('stl_score', '')}"
),
axis=1,
)
fig.add_trace(
go.Scatter(
x=anomaly_df[timestamp_col],
y=anomaly_df[value_col],
mode="markers",
name="異常候補",
marker=dict(size=9, color="#dc2626", symbol="x", line=dict(width=2)),
text=hover_text,
hovertemplate="%{text}<extra></extra>",
)
)
# 範囲スライダーと簡単な期間ボタンを付けます。
# 長い時系列でも、全体を見てから一部を拡大しやすくなります。
fig.update_layout(
title=title,
template="plotly_white",
height=560,
hovermode="x unified",
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="left", x=0),
margin=dict(l=60, r=30, t=80, b=70),
xaxis=dict(
title="時刻",
rangeslider=dict(visible=True),
rangeselector=dict(
buttons=[
dict(count=6, label="6h", step="hour", stepmode="backward"),
dict(count=1, label="1d", step="day", stepmode="backward"),
dict(step="all", label="全体"),
]
),
),
yaxis=dict(title="値", zeroline=False),
)
return fig
def save_interactive_html(fig: go.Figure, output_path: str | Path) -> Path:
"""Plotly Figureを単体HTMLとして保存します。
HTMLとして保存しておくと、Notebookを開かなくてもブラウザだけで確認できます。
Qiita記事用のスクリーンショットを撮るときにも便利です。
"""
output_path = Path(output_path)
output_path.parent.mkdir(parents=True, exist_ok=True)
# include_plotlyjs="cdn" にするとHTMLが軽くなります。
# 完全オフラインで配布したい場合は、include_plotlyjs=True に変更します。
fig.write_html(output_path, include_plotlyjs="cdn", full_html=True)
return output_path
Web画面側のPlotly.js表示
Web画面では、APIから返された records を使って Plotly.js で描画します。
Notebookと同じく、観測値、移動中央値、周期文脈の期待値、STL期待値、異常候補を重ねます。
web/app.js の主要部分
/**
* APIから返された値を数値配列に変換します。
*
* Plotlyに undefined や NaN をそのまま渡すと、表示やホバーが崩れることがあります。
* そのため、数値として扱えない値は null にそろえます。
*/
function numericArray(records, key) {
return records.map((r) => {
const v = Number(r[key]);
return Number.isFinite(v) ? v : null;
});
}
/**
* 異常点のホバー表示に使うテキストを作成します。
*
* グラフ上で異常候補にマウスを合わせたとき、
* 値だけでなく、どのStepが反応したかを確認できるようにしています。
*/
function anomalyHoverText(records) {
return records.map((r) => {
const reason = r.anomaly_reason || 'OK';
const robust = r.robust_score ?? '';
const seasonal = r.seasonal_score ?? '';
const stl = r.stl_score ?? '';
return [
`時刻: ${r.timestamp}`,
`値: ${r.value}`,
`理由: ${reason}`,
`robust: ${robust}`,
`seasonal: ${seasonal}`,
`stl: ${stl}`,
].join('<br>');
});
}
/**
* Plotly.jsでインタラクティブな時系列グラフを描画します。
*
* 主な操作:
* - マウスホイールによるズーム
* - ドラッグによる範囲拡大
* - panモードでの移動
* - 下部レンジスライダーによる期間変更
* - 凡例クリックによる系列ON/OFF
* - 異常点ホバーによる理由確認
* - カメラアイコンからPNG保存
*/
function drawChart(records) {
const chart = document.getElementById('chart');
if (!window.Plotly) {
chart.innerHTML = '<div class="plotly-error">Plotly.jsを読み込めませんでした。</div>';
return;
}
if (!records || records.length === 0) {
chart.innerHTML = '<div class="plotly-error">表示するデータがありません。</div>';
return;
}
const x = records.map((r) => r.timestamp);
const values = numericArray(records, 'value');
const rollingMedian = numericArray(records, 'rolling_median');
const contextExpected = numericArray(records, 'context_expected');
const stlExpected = numericArray(records, 'stl_expected');
const anomalies = records.filter((r) => r.final_is_anomaly);
const anomalyX = anomalies.map((r) => r.timestamp);
const anomalyY = anomalies.map((r) => Number.isFinite(Number(r.value)) ? Number(r.value) : null);
const anomalyText = anomalyHoverText(anomalies);
const traces = [
{
x,
y: values,
type: 'scatter',
mode: 'lines',
name: '観測値',
line: { width: 2.5, color: '#0f63ce' },
hovertemplate: '時刻: %{x}<br>観測値: %{y:.4f}<extra></extra>',
},
{
x,
y: rollingMedian,
type: 'scatter',
mode: 'lines',
name: '移動中央値',
line: { width: 2, color: '#64748b', dash: 'dash' },
hovertemplate: '時刻: %{x}<br>移動中央値: %{y:.4f}<extra></extra>',
},
{
x,
y: contextExpected,
type: 'scatter',
mode: 'lines',
name: '周期文脈の期待値',
line: { width: 1.8, color: '#16a34a', dash: 'dot' },
// 初期表示では非表示にしておきます。
// 凡例をクリックすると表示できます。
visible: 'legendonly',
hovertemplate: '時刻: %{x}<br>周期文脈期待値: %{y:.4f}<extra></extra>',
},
{
x,
y: stlExpected,
type: 'scatter',
mode: 'lines',
name: 'STL期待値',
line: { width: 1.8, color: '#f97316', dash: 'dashdot' },
visible: 'legendonly',
hovertemplate: '時刻: %{x}<br>STL期待値: %{y:.4f}<extra></extra>',
},
{
x: anomalyX,
y: anomalyY,
type: 'scatter',
mode: 'markers',
name: '異常候補',
marker: {
size: 9,
color: '#dc2626',
symbol: 'x',
line: { width: 2 },
},
text: anomalyText,
hovertemplate: '%{text}<extra></extra>',
},
];
const layout = {
title: { text: 'Step 1〜3 異常検知結果', x: 0.02, xanchor: 'left' },
autosize: true,
height: 520,
margin: { l: 60, r: 26, t: 58, b: 70 },
paper_bgcolor: '#ffffff',
plot_bgcolor: '#fbfdff',
hovermode: 'x unified',
legend: {
orientation: 'h',
y: 1.13,
x: 0,
bgcolor: 'rgba(255,255,255,0.75)',
},
xaxis: {
title: '時刻',
rangeslider: { visible: true, thickness: 0.12 },
rangeselector: {
buttons: [
{ count: 6, label: '6h', step: 'hour', stepmode: 'backward' },
{ count: 1, label: '1d', step: 'day', stepmode: 'backward' },
{ step: 'all', label: '全体' },
],
},
gridcolor: '#e2e8f0',
},
yaxis: {
title: '値',
zeroline: false,
gridcolor: '#e2e8f0',
},
};
const config = {
responsive: true,
// マウスホイールでズームできるようにします。
scrollZoom: true,
displaylogo: false,
// 今回は時系列を見る用途なので、投げ縄選択などは外しています。
modeBarButtonsToRemove: ['lasso2d', 'select2d'],
toImageButtonOptions: {
format: 'png',
filename: 'timeseries_anomaly_plotly',
height: 720,
width: 1280,
scale: 1,
},
};
Plotly.react(chart, traces, layout, config);
}
JupyterLabでの確認
Notebookでは、サンプルCSVを読み込んでパイプラインを実行します。
from pathlib import Path
import pandas as pd
from timeseries_anomaly.pipeline import PipelineConfig, run_pipeline
from timeseries_anomaly.common.plotting import (
create_interactive_timeseries_figure,
save_interactive_html,
)
# サンプルデータを読み込みます。
# 実データを使う場合は、このパスを data/raw/ 配下のCSVへ変更します。
sample_csv = Path("data/sample/sample_timeseries.csv")
df = pd.read_csv(sample_csv)
# まずはデフォルト設定で動かします。
# しきい値や周期は、後のセルで調整できるようにしています。
config = PipelineConfig(
timestamp_col="timestamp",
value_col="value",
freq_minutes=10,
lower_limit=0.0,
upper_limit=3.0,
flatline_points=12,
rolling_window=13,
mad_threshold=3.5,
diff_score_threshold=6.0,
seasonal_period=144,
seasonal_threshold=3.5,
time_context_threshold=3.5,
use_stl=True,
)
# Step 1〜3をまとめて実行します。
# output_dirを指定すると、結果CSVが outputs/ に保存されます。
result = run_pipeline(
df,
config,
output_dir=Path("outputs"),
)
# サマリーを確認します。
result["summary"]
異常候補だけを表で見る場合は、次のようにします。
res = result["result_df"]
# final_is_anomaly が True の行だけを取り出します。
# まずは上位30件を見て、どのStepが反応しているか確認します。
anomaly_df = res[res["final_is_anomaly"]].copy()
anomaly_df[[
"timestamp",
"value",
"quality_flags",
"robust_score",
"seasonal_score",
"stl_score",
"anomaly_reason",
]].head(30)
Plotlyで表示します。
fig = create_interactive_timeseries_figure(
res,
timestamp_col="timestamp",
value_col="value",
title="時系列異常値検知 Step 1〜3",
)
fig.show()
HTMLとして保存します。
save_interactive_html(
fig,
"reports/step1_3_interactive_timeseries.html",
)
結果CSV
実行結果は outputs/ に保存されます。
outputs/anomaly_result_YYYYMMDD_HHMMSS.csv
主な列は以下です。
| 列名 | 内容 |
|---|---|
| quality_flags | Step 1 の品質フラグ |
| quality_is_anomaly | Step 1 で異常候補になったか |
| rolling_median | Step 2 の移動中央値 |
| robust_score | Step 2 のロバストスコア |
| diff_score | Step 2 の急変スコア |
| robust_is_anomaly | Step 2 で異常候補になったか |
| context_expected | Step 3 の周期位置ベース期待値 |
| context_residual | 観測値と周期期待値の差 |
| seasonal_score | 周期文脈スコア |
| stl_expected | STL分解から作った期待値 |
| stl_score | STL残差スコア |
| seasonal_is_anomaly | Step 3 で異常候補になったか |
| final_is_anomaly | Step 1〜3のいずれかで異常候補になったか |
| anomaly_reason | 異常理由 |
オフライン利用時の注意
Web画面では、web/index.html で Plotly.js を CDN から読み込んでいます。
<script src="https://cdn.plot.ly/plotly-2.35.2.min.js"></script>
インターネットに接続できない環境で完全オフライン運用する場合は、Plotly.jsを以下へ配置します。
web/vendor/plotly.min.js
そのうえで、index.html の読み込み先を以下のように変更します。
<script src="vendor/plotly.min.js"></script>
JupyterLab側は Python パッケージの plotly をインストールしているため、Notebook内で対話的なグラフを利用できます。
この段階でできること
この構成では、以下を確認できます。
- CSVの時刻・値列を指定して分析
- 欠測、重複、時刻抜け、物理範囲外の検出
- 移動中央値とMADによるスパイク検出
- 前時刻との差分による急変検出
- 周期性・季節性を考慮した文脈異常検出
- Plotlyによる拡大・縮小・ホバー確認
- 結果CSV保存
- Notebookでの追加分析
- HTMLレポート保存
今後の拡張方針
この r004 では Step 1〜3 に限定しています。
次の段階では、予測残差ベースを追加します。
Step 4：予測残差ベース
- 正常値を予測する
- 実測値との差を residual として計算する
- residual が大きい区間を異常候補にする
さらにその次は、多変量・変化点・AI拡張です。
Step 5：多変量・変化点・AI拡張
- PCA
- Isolation Forest
- 変化点検出
- AutoEncoder
- Transformer
最初から高度なモデルを入れるより、Step 1〜3で「データの状態」と「検出理由」を確認できるようにしておく方が、後の拡張もしやすくなります。
まとめ
今回は、時系列異常値検知の第1段階として、Docker + JupyterLab + Plotly の環境を作りました。
実装した内容は以下です。
Step 1：データ品質チェック
Step 2：ロバスト統計
Step 3：周期性・季節性対応
Plotlyを使っているので、Web画面とNotebookの両方で、異常候補の周辺を拡大しながら確認できます。
まずはこの段階で、欠測やスパイク、時間帯を考えると不自然な値を拾えるようにしておくと、次のStep 4やStep 5に進めやすくなります。
参考リンク
-
pandas
https://pandas.pydata.org/ -
statsmodels
https://www.statsmodels.org/ -
Plotly Python
https://plotly.com/python/ -
Plotly.js
https://plotly.com/javascript/ -
JupyterLab
https://jupyter.org/ -
FastAPI
https://fastapi.tiangolo.com/ -
Docker Compose
https://docs.docker.com/compose/