AI異常検出API：業務データ品質監視とアラートシステムの実装ガイド

本システムは以下の3層構造で構築します：

• データ収集層：業務データベースやAPIからのリアルタイムデータ取得
• 異常検知層：HolySheep AI APIを活用したAI駆動の異常検知エンジン
• アラート通知層：異常検知時のSlack/メール/Webhook通知

Pythonによる異常検知APIの実装

まず、HolySheep AIのAPIクライアントをセットアップします。以下のコードは、リアルタイムでデータ品質を監視し、異常を検出する基盤となります。

import requests
import json
import time
from datetime import datetime
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, asdict
import statistics

@dataclass
class AnomalyResult:
    timestamp: str
    metric_name: str
    value: float
    expected_range: tuple
    is_anomaly: bool
    confidence: float
    severity: str
    recommendation: str

class HolySheepAnomalyDetector:
    """HolySheep AI APIを活用した異常検知システム"""
    
    BASE_URL = "https://api.holysheep.ai/v1"
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        self.historical_data: Dict[str, List[float]] = {}
        self.baseline_windows = 100  # ベースライン取得ウィンドウサイズ
    
    def analyze_with_ai(self, metrics: List[Dict[str, Any]]) -> Dict[str, Any]:
        """
        HolySheep AI APIを使用して複数メトリクスの異常分析を実行
        """
        prompt = self._build_analysis_prompt(metrics)
        
        payload = {
            "model": "gpt-4.1",
            "messages": [
                {
                    "role": "system",
                    "content": """あなたはデータ品質監視 전문가です。
metrics 配列内の各メトリクスについて異常度を0-1で評価し、
{
  "anomalies": [
    {
      "metric": "メトリクス名",
      "anomaly_score": 0.0-1.0,
      "severity": "low|medium|high|critical",
      "reason": "異常理由",
      "recommendation": "推奨対応"
    }
  ],
  "overall_health": 0.0-1.0,
  "summary": "全体サマリー"
}
形式のJSONを出力してください。"""
                },
                {
                    "role": "user", 
                    "content": prompt
                }
            ],
            "temperature": 0.1,
            "response_format": {"type": "json_object"}
        }
        
        response = requests.post(
            f"{self.BASE_URL}/chat/completions",
            headers=self.headers,
            json=payload,
            timeout=30
        )
        
        if response.status_code == 200:
            result = response.json()
            return json.loads(result["choices"][0]["message"]["content"])
        else:
            raise Exception(f"API Error: {response.status_code} - {response.text}")
    
    def _build_analysis_prompt(self, metrics: List[Dict[str, Any]]) -> str:
        """分析用プロンプトを構築"""
        metrics_text = "\n".join([
            f"- {m['name']}: 値={m['value']}, 単位={m.get('unit', 'N/A')}, "
            f"時間={m.get('timestamp', 'unknown')}"
            for m in metrics
        ])
        
        return f"""以下の業務メトリクスのデータを分析し、異常値を検出してください：

{metrics_text}

各メトリクスについて：
1.  статисти的に異常な値を特定
2.  異常の重大度（low/medium/high/critical）を判定
3.  異常の考えられる原因
4.  推奨される対応措施

JSON形式で回答してください。"""

使用例
api_key = "YOUR_HOLYSHEEP_API_KEY"
detector = HolySheepAnomalyDetector(api_key)

監視対象メトリクスの例
current_metrics = [
    {"name": "order_count", "value": 45, "unit": "件/分", "timestamp": datetime.now().isoformat()},
    {"name": "response_time", "value": 2500, "unit": "ms", "timestamp": datetime.now().isoformat()},
    {"name": "error_rate", "value": 8.5, "unit": "%", "timestamp": datetime.now().isoformat()},
    {"name": "cpu_usage", "value": 92, "unit": "%", "timestamp": datetime.now().isoformat()},
]

try:
    analysis_result = detector.analyze_with_ai(current_metrics)
    print(f"全体健康度: {analysis_result['overall_health']}")
    for anomaly in analysis_result['anomalies']:
        print(f"異常: {anomaly['metric']} - {anomaly['severity']} - {anomaly['reason']}")
except Exception as e:
    print(f"エラー: {e}")

リアルタイム品質監視ダッシュボードの実装

次に、Webhook通知機能を備えたリアルタイム監視ダッシュボードを構築します。

import asyncio
import aiohttp
from collections import deque
from typing import Deque, Callable
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class DataQualityMonitor:
    """データ品質リアルタイム監視システム"""
    
    def __init__(
        self,
        api_key: str,
        check_interval: int = 60,
        anomaly_threshold: float = 0.7
    ):
        self.api_key = api_key
        self.check_interval = check_interval
        self.anomaly_threshold = anomaly_threshold
        self.metric_history: Deque = deque(maxlen=1000)
        self.alert_callbacks: List[Callable] = []
        
    async def check_data_quality(self, data_batch: List[Dict]) -> Dict:
        """データバッチの品質チェックを実行"""
        
        validation_results = {
            "total_records": len(data_batch),
            "valid_records": 0,
            "invalid_records": 0,
            "quality_score": 0.0,
            "issues": []
        }
        
        # 基本バリデーション
        for idx, record in enumerate(data_batch):
            issues = self._validate_record(record, idx)
            if not issues:
                validation_results["valid_records"] += 1
            else:
                validation_results["invalid_records"] += 1
                validation_results["issues"].extend(issues)
        
        # 品質スコア計算
        if validation_results["total_records"] > 0:
            validation_results["quality_score"] = (
                validation_results["valid_records"] / 
                validation_results["total_records"]
            ) * 100
        
        # AIを活用した詳細な異常検知
        if validation_results["invalid_records"] > 0:
            ai_analysis = await self._ai_anomaly_detection(data_batch, validation_results)
            validation_results["ai_analysis"] = ai_analysis
        
        return validation_results
    
    def _validate_record(self, record: Dict, idx: int) -> List[str]:
        """单个レコードのバリデーション"""
        issues = []
        
        # null/nullチェック
        if record.get("value") is None:
            issues.append(f"[{idx}] valueがnullです")
        
        # 型チェック
        if "value" in record and not isinstance(record["value"], (int, float)):
            issues.append(f"[{idx}] valueが数値型ではありません: {type(record['value'])}")
        
        # 範囲チェック
        if "value" in record and isinstance(record["value"], (int, float)):
            if record["value"] < 0:
                issues.append(f"[{idx}] valueが負値です: {record['value']}")
            if record["value"] > 1000000:
                issues.append(f"[{idx}] valueが異常な高値です: {record['value']}")
        
        # 必須フィールドチェック
        required_fields = ["id", "timestamp", "value"]
        for field in required_fields:
            if field not in record:
                issues.append(f"[{idx}] 必須フィールド'{field}'が欠落しています")
        
        return issues
    
    async def _ai_anomaly_detection(
        self,
        data_batch: List[Dict],
        validation_results: Dict
    ) -> Dict:
        """HolySheep AI APIで異常パターンを検出"""
        
        prompt = f"""
データ品質チェック結果：
- 総レコード数: {validation_results['total_records']}
- 有効レコード: {validation_results['valid_records']}
- 無効レコード: {validation_results['invalid_records']}
- 品質スコア: {validation_results['quality_score']:.2f}%

検出された問題：
{chr(10).join(validation_results['issues'][:10])}

以下の観点から分析してください：
1. 問題の根本原因の推定
2. データソース別の問題分布
3. 時間帯やバッチとの関連性
4. 推奨される対処措施

{
  "root_cause": "推定される根本原因",
  "affected_sources": ["影響を受けたソース一覧"],
  "recommended_actions": ["推奨対処措施"],
  "risk_level": "low|medium|high|critical"
}
形式のJSONで回答してください。"""

        payload = {
            "model": "gpt-4.1",
            "messages": [
                {"role": "system", "content": "あなたはデータエンジニアリングの専門家です。"},
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.2,
            "response_format": {"type": "json_object"}
        }
        
        async with aiohttp.ClientSession() as session:
            async with session.post(
                "https://api.holysheep.ai/v1/chat/completions",
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json"
                },
                json=payload,
                timeout=aiohttp.ClientTimeout(total=30)
            ) as response:
                if response.status == 200:
                    result = await response.json()
                    return json.loads(result["choices"][0]["message"]["content"])
                else:
                    error_text = await response.text()
                    logger.error(f"AI分析エラー: {response.status} - {error_text}")
                    return {"error": error_text}
    
    def register_alert_callback(self, callback: Callable):
        """アラートコールバックを登録"""
        self.alert_callbacks.append(callback)
    
    async def send_alert(self, alert_data: Dict):
        """登録されたコールバックにアラートを送信"""
        for callback in self.alert_callbacks:
            try:
                await callback(alert_data)
            except Exception as e:
                logger.error(f"アラート送信エラー: {e}")
    
    async def start_monitoring(self):
        """継続的な監視を開始"""
        logger.info("データ品質監視を開始しました")
        
        while True:
            try:
                # サンプルデータの生成（実際のシステムではDBやAPIから取得）
                sample_data = self._generate_sample_data()
                
                # 品質チェック実行
                result = await self.check_data_quality(sample_data)
                
                # 異常検出時のアラート送信
                if result["quality_score"] < (1 - self.anomaly_threshold) * 100:
                    alert_data = {
                        "timestamp": datetime.now().isoformat(),
                        "quality_score": result["quality_score"],
                        "issues": result["issues"][:5],
                        "ai_analysis": result.get("ai_analysis", {})
                    }
                    await self.send_alert(alert_data)
                    logger.warning(f"品質異常検出: {result['quality_score']:.2f}%")
                
                self.metric_history.append(result)
                
            except Exception as e:
                logger.error(f"監視エラー: {e}")
            
            await asyncio.sleep(self.check_interval)
    
    def _generate_sample_data(self) -> List[Dict]:
        """サンプルデータ生成（実際のシステムでは別のデータソースを使用）"""
        import random
        return [
            {
                "id": f"rec_{i}",
                "timestamp": datetime.now().isoformat(),
                "value": random.choice([100, 200, None, 300, -1]) if random.random() < 0.1 else random.randint(50, 500)
            }
            for i in range(100)
        ]

Slack webhook通知コールバックの例
async def slack_notification(alert_data: Dict):
    """Slackに通知を送信"""
    webhook_url = "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"
    
    message = {
        "text": "🚨 データ品質アラート",
        "blocks": [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*品質スコア:* {alert_data['quality_score']:.2f}%\n*検出問題数:* {len(alert_data['issues'])}"
                }
            }
        ]
    }
    
    async with aiohttp.ClientSession() as session:
        await session.post(webhook_url, json=message)

監視の開始
monitor = DataQualityMonitor(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    check_interval=60,
    anomaly_threshold=0.7
)
monitor.register_alert_callback(slack_notification)

異常検知の実践例：ECサイトの注文データ監視

実際のビジネスシナリオとして、ECサイトの注文データ品質を監視するシステムを構築します。