パイプライン設計支援機能

[takaeda]

概要

時系列データ処理パイプラインの設計・試行錯誤を支援する機能の提案です。
toorPIAマップを「評価出口」として、パイプライン設計の妥当性を判断できる仕組みを目指します。

背景・動機

製造設備の予兆保全において、データ処理パイプラインの設計は極めて重要です。

• 時間スケール（移動平均窓幅等）の選択
• 値域を意識したセグメント分割（停止/稼働、高負荷/低負荷等）
• 処理ステップの組み合わせ

これらのパラメータは設備や対象事象ごとに異なり、試行錯誤が必要です。

従来のML的アプローチでは評価指標が抽象的になりがちですが、
toorPIAマップを使うことで以下の具体的な評価が可能になります：

1. クラスタ分離 - 島が明確に分離するか → セグメント設計の妥当性
2. 既知事象の検出 - 過去の異常がマップ上で識別できるか → 時間スケールの妥当性

提案する機能

1. パイプライン定義

pipeline:
  name: "bearing_monitoring_v1"
  steps:
    - moving_average: { window: "10min" }
    - zscore: { baseline: "7days" }
  
  variants:  # 比較検証用のバリエーション
    - name: "short_window"
      override:
        moving_average: { window: "5min" }
    - name: "long_window"  
      override:
        moving_average: { window: "30min" }

2. 評価指標の自動計算

• 既知イベントの検出率
• 検出タイミング（何時間前から兆候が見えたか）
• 誤検出率（異常がない期間での逸脱度）
• クラスタ分離度

3. 比較ビュー

複数のパイプラインバリエーションを同一期間で実行し、結果を並べて比較。

4. 複数時間スケール並行運用の設計支援

toorPIAの軽量性を活かし、「一つの最適解」ではなく
「複数の時間スケールで並行監視」という設計を支援。

multi_scale_monitoring:
  scales:
    - name: "short_term"
      window: "10min"
      target_events: ["急激な変化", "振動異常"]
    - name: "long_term"
      window: "2hour"
      target_events: ["緩やかな劣化トレンド"]

toorPIAとの連携

• マップ構築・追加プロットは既存toorPIA APIを使用
• IF-HUB側で検証セッションとマップIDの対応を管理
• 検証用マップは保持し、後から参照可能に

依存関係

• 仮想時刻による過去データ検証基盤 #20 （本機能の前提となる基盤）

期待される効果

• パイプライン設計の試行錯誤サイクルの短縮
• 「なぜこのパラメータを選んだか」の根拠明確化
• 現場への説明可能性の向上

[takaeda]

実装上の課題と注意点

前提

本機能は Issue #20（仮想時刻による過去データ検証基盤）の上に構築します。
#20の以下の機能が利用可能であることを前提とします：

• reference_timeパラメータ付きデータ取得
• 検証セッション管理
• toorPIA連携

---

1. パイプライン定義のYAML設計

処理ステップの抽象化

pipeline:
  steps:
    - type: moving_average
      params:
        window: "10min"
    
    - type: zscore
      params:
        baseline: "7days"
    
    - type: segment_filter
      params:
        condition: "value > 100"  # 条件式の評価が必要

実装上の課題

• 処理ステップの実行順序保証
• 各ステップのインターフェース統一
• 条件式の安全な評価（evalの危険性回避）

推奨アプローチ

// ステップの登録
const stepRegistry = {
  moving_average: (data, params) => movingAverage(data, params.window),
  zscore: (data, params) => calculateZscore(data, params.baseline),
  segment_filter: (data, params) => filterByCondition(data, params.condition)
};

// パイプライン実行
function executePipeline(data, steps) {
  return steps.reduce((acc, step) => {
    const handler = stepRegistry[step.type];
    if (!handler) throw new Error(`Unknown step: ${step.type}`);
    return handler(acc, step.params);
  }, data);
}

---

2. バリエーション生成と比較

オーバーライド機構

pipeline:
  name: "base"
  steps:
    - type: moving_average
      id: "ma"  # 参照用ID
      params:
        window: "10min"

variants:
  - name: "short_window"
    override:
      ma:  # idで参照
        window: "5min"
  
  - name: "long_window"
    override:
      ma:
        window: "30min"

実装

function applyOverride(baseSteps, override) {
  return baseSteps.map(step => {
    if (override[step.id]) {
      return { ...step, params: { ...step.params, ...override[step.id] } };
    }
    return step;
  });
}

---

3. 評価指標の計算

既知イベント検出率

function evaluateDetection(plotResults, knownEvents, threshold) {
  const detected = knownEvents.filter(event => {
    // イベント前後N時間のプロットで閾値超えがあるか
    const nearbyPlots = plotResults.filter(p => 
      Math.abs(p.virtualTime - event.time) < DETECTION_WINDOW
    );
    return nearbyPlots.some(p => p.deviation > threshold);
  });
  
  return {
    total: knownEvents.length,
    detected: detected.length,
    rate: detected.length / knownEvents.length
  };
}

検出リードタイム

function calculateLeadTime(plotResults, event, threshold) {
  // イベント前で最初に閾値を超えた時刻を探す
  const priorPlots = plotResults
    .filter(p => p.virtualTime < event.time)
    .sort((a, b) => b.virtualTime - a.virtualTime);
  
  for (const plot of priorPlots) {
    if (plot.deviation > threshold) {
      return event.time - plot.virtualTime;  // ミリ秒
    }
  }
  return null;  // 事前検出なし
}

誤検出率

function calculateFalsePositiveRate(plotResults, knownEvents, threshold) {
  // 既知イベント周辺以外で閾値超えした数
  const falsePositives = plotResults.filter(p => {
    if (p.deviation <= threshold) return false;
    
    // いずれかの既知イベントの近くか？
    const nearEvent = knownEvents.some(e => 
      Math.abs(p.virtualTime - e.time) < EVENT_BUFFER
    );
    return !nearEvent;
  });
  
  return {
    count: falsePositives.length,
    rate: falsePositives.length / plotResults.length
  };
}

---

4. 複数時間スケール並行運用

設計上のポイント

各スケールごとに独立したマップを持つため、検証セッション内で複数のmodel_timelineを管理する必要があります。

multi_scale:
  scales:
    - id: "short"
      window: "10min"
      model_rebuild: "daily@02:00"
    
    - id: "long"
      window: "2hour"
      model_rebuild: "weekly@sunday_02:00"

データ構造の拡張

// 単一スケール（#20の設計）
modelTimeline: [{ virtualTime, toorpiaMapId }]

// 複数スケール（#21で拡張）
modelTimelines: {
  "short": [{ virtualTime, toorpiaMapId }],
  "long": [{ virtualTime, toorpiaMapId }]
}

---

5. クラスタ分離度の評価

toorPIA側で取得可能な情報の確認が必要

取得可能な場合：

• クラスタ数
• 各クラスタのサイズ
• クラスタ間距離

取得困難な場合の代替案：

• 追加プロットの逸脱度分布を分析
• 逸脱度のbimodal分布 → クラスタ分離の兆候

---

6. ユーザーインターフェース

Phase 1: CLI

# パイプライン検証実行
ifhub-validate --scenario scenario.yaml --output results/

# バリエーション比較
ifhub-compare --results results/ --metrics detection_rate,lead_time

Phase 2: REST API

POST /api/validation/run
GET /api/validation/{session_id}/status
GET /api/validation/{session_id}/results
GET /api/validation/compare?sessions=id1,id2,id3

Phase 3: Web UI（将来）

• シナリオエディタ
• 結果の可視化ダッシュボード
• マップの並列表示

---

7. 実装優先順位

1. パイプライン定義とステップ実行 - 基本的な処理フローの確立
2. 既知イベント検出評価 - 最も実用的な評価指標
3. バリエーション比較 - パラメータ探索の効率化
4. 複数時間スケール - 高度な運用パターン対応

---

8. 既存コードとの統合ポイント

**gtags/**との関係

• gtagは「仮想タグ」、パイプラインは「処理フロー」
• パイプラインのステップとしてgtagを呼び出す可能性
• 重複機能（移動平均など）の整理が必要かもしれない

**fetcher/**との関係

• Fetcherの出力がパイプラインの入力
• パイプライン処理をFetcher内に組み込むか、外部で行うか設計判断が必要