ИИ для Обнаружения Аномалий Производительности в Тестировании

import numpy as np
from sklearn.preprocessing import StandardScaler
from datetime import datetime, timedelta

class PerformanceBaseline:
    def __init__(self, window_days=30):
        self.window_days = window_days
        self.scaler = StandardScaler()
        self.baseline_metrics = {}

    def train_baseline(self (как обсуждается в [AI Log Analysis: Intelligent Error Detection and Root Cause Analysis](/blog/ai-log-analysis)), metrics_data):
        """
        Обучает модель базовой линии на исторических данных производительности

        Args:
            metrics_data: DataFrame с колонками ['timestamp', 'response_time',
                         'throughput', 'error_rate', 'cpu_usage', 'memory_usage']
        """
        # Фильтрация данных для окна обучения
        cutoff_date = datetime.now() - timedelta(days=self.window_days)
        training_data (как обсуждается в [AI-Powered Security Testing: Finding Vulnerabilities Faster](/blog/ai-security-testing)) = metrics_data[metrics_data['timestamp'] >= cutoff_date]

        # Расчет статистических базовых линий для каждой метрики
        for metric in ['response_time', 'throughput', 'error_rate',
                       'cpu_usage', 'memory_usage']:
            self.baseline_metrics[metric] = {
                'mean': training_data[metric] (как обсуждается в [AI Test Metrics Analytics: Intelligent Analysis of QA Metrics](/blog/ai-test-metrics)).mean(),
                'std': training_data[metric].std(),
                'percentile_95': training_data[metric].quantile(0.95),
                'percentile_99': training_data[metric].quantile(0.99),
                'min': training_data[metric].min(),
                'max': training_data[metric].max()
            }

        return self.baseline_metrics

    def is_anomaly(self, current_value, metric_name, threshold_std=3):
        """
        Определяет, отклоняется ли текущее значение от базовой линии
        """
        baseline = self.baseline_metrics[metric_name]
        z_score = abs((current_value - baseline['mean']) / baseline['std'])

        return z_score > threshold_std, z_score

import pandas as pd
from sklearn.ensemble import RandomForestRegressor

class TemporalBaselineModel:
    def __init__(self):
        self.model = RandomForestRegressor(n_estimators=100, random_state=42)

    def extract_temporal_features(self, timestamp):
        """Извлекает временные признаки для распознавания паттернов"""
        return {
            'hour': timestamp.hour,
            'day_of_week': timestamp.dayofweek,
            'day_of_month': timestamp.day,
            'month': timestamp.month,
            'is_weekend': 1 if timestamp.dayofweek >= 5 else 0,
            'is_business_hours': 1 if 9 <= timestamp.hour <= 17 else 0
        }

    def train(self, historical_data):
        """Обучает модель для предсказания ожидаемой производительности на основе времени"""
        features = pd.DataFrame([
            self.extract_temporal_features(ts)
            for ts in historical_data['timestamp']
        ])

        self.model.fit(features, historical_data['response_time'])

    def predict_expected_performance(self, timestamp):
        """Предсказывает ожидаемое время отклика для данного timestamp"""
        features = pd.DataFrame([self.extract_temporal_features(timestamp)])
        return self.model.predict(features)[0]

from sklearn.ensemble import IsolationForest
import pandas as pd

class PerformanceAnomalyDetector:
    def __init__(self, contamination=0.1):
        self.model = IsolationForest(
            contamination=contamination,
            random_state=42,
            n_estimators=100
        )
        self.feature_columns = [
            'response_time', 'throughput', 'error_rate',
            'cpu_usage', 'memory_usage', 'db_query_time'
        ]

    def train(self, historical_metrics):
        """Обучает Isolation Forest на нормальных паттернах производительности"""
        X = historical_metrics[self.feature_columns]
        self.model.fit(X)

    def detect_anomalies(self, current_metrics):
        """
        Обнаруживает аномалии в текущих метриках

        Returns:
            predictions: -1 для аномалий, 1 для нормальных
            scores: оценки аномалий (ниже = более аномально)
        """
        X = current_metrics[self.feature_columns]
        predictions = self.model.predict(X)
        scores = self.model.score_samples(X)

        anomalies = current_metrics[predictions == -1].copy()
        anomalies['anomaly_score'] = scores[predictions == -1]

        return anomalies

    def explain_anomaly(self, anomaly_record):
        """Определяет, какие метрики больше всего повлияли на обнаружение аномалии"""
        contributions = {}

        for feature in self.feature_columns:
            baseline_mean = self.baseline_metrics[feature]['mean']
            baseline_std = self.baseline_metrics[feature]['std']
            current_value = anomaly_record[feature]

            deviation = abs((current_value - baseline_mean) / baseline_std)
            contributions[feature] = deviation

        # Сортировка по вкладу
        sorted_contributions = sorted(
            contributions.items(),
            key=lambda x: x[1],
            reverse=True
        )

        return sorted_contributions

import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
import numpy as np

class LSTMAnomalyDetector:
    def __init__(self, sequence_length=50):
        self.sequence_length = sequence_length
        self.model = None
        self.threshold = None

    def build_model(self, n_features):
        """Строит автоэнкодер LSTM для обнаружения аномалий"""
        model = Sequential([
            LSTM(64, activation='relu', input_shape=(self.sequence_length, n_features),
                 return_sequences=True),
            Dropout(0.2),
            LSTM(32, activation='relu', return_sequences=False),
            Dropout(0.2),
            Dense(32, activation='relu'),
            Dense(n_features)
        ])

        model.compile(optimizer='adam', loss='mse')
        self.model = model
        return model

    def create_sequences(self, data):
        """Преобразует данные временных рядов в последовательности"""
        sequences = []
        for i in range(len(data) - self.sequence_length):
            sequences.append(data[i:i + self.sequence_length])
        return np.array(sequences)

    def train(self, normal_data, epochs=50, batch_size=32):
        """Обучает LSTM на нормальных данных производительности"""
        n_features = normal_data.shape[1]

        if self.model is None:
            self.build_model(n_features)

        X_train = self.create_sequences(normal_data)

        # Обучение автоэнкодера для реконструкции нормальных паттернов
        self.model.fit(
            X_train,
            normal_data[self.sequence_length:],
            epochs=epochs,
            batch_size=batch_size,
            validation_split=0.1,
            verbose=0
        )

        # Расчет порога ошибки реконструкции
        predictions = self.model.predict(X_train)
        reconstruction_errors = np.mean(np.abs(predictions - normal_data[self.sequence_length:]), axis=1)
        self.threshold = np.percentile(reconstruction_errors, 95)

    def detect_anomalies(self, test_data):
        """Обнаруживает аномалии на основе ошибки реконструкции"""
        X_test = self.create_sequences(test_data)
        predictions = self.model.predict(X_test)

        reconstruction_errors = np.mean(np.abs(predictions - test_data[self.sequence_length:]), axis=1)

        anomalies = reconstruction_errors > self.threshold
        return anomalies, reconstruction_errors

from statsmodels.tsa.holtwinters import ExponentialSmoothing
from sklearn.metrics import mean_absolute_error
import warnings
warnings.filterwarnings('ignore')

class PerformanceTrendPredictor:
    def __init__(self):
        self.models = {}

    def train_predictor(self, metric_data, metric_name, seasonal_periods=24):
        """
        Обучает модель экспоненциального сглаживания для прогнозирования трендов

        Args:
            metric_data: Данные временных рядов для конкретной метрики
            metric_name: Название метрики (напр., 'response_time')
            seasonal_periods: Количество периодов в сезонном цикле (24 для почасовых данных)
        """
        model = ExponentialSmoothing(
            metric_data,
            seasonal_periods=seasonal_periods,
            trend='add',
            seasonal='add'
        ).fit()

        self.models[metric_name] = model
        return model

    def predict_future(self, metric_name, steps_ahead=24):
        """Предсказывает будущие значения для указанной метрики"""
        if metric_name not in self.models:
            raise ValueError(f"Нет обученной модели для {metric_name}")

        forecast = self.models[metric_name].forecast(steps=steps_ahead)
        return forecast

    def detect_degradation_trend(self, metric_name, threshold_slope=0.05):
        """
        Определяет, показывает ли метрика тренд деградации

        Returns:
            is_degrading: Булево значение, указывающее тренд деградации
            slope: Скорость деградации
            forecast: Предсказанные значения
        """
        forecast = self.predict_future(metric_name, steps_ahead=24)

        # Расчет наклона тренда
        time_steps = np.arange(len(forecast))
        slope = np.polyfit(time_steps, forecast, 1)[0]

        is_degrading = slope > threshold_slope

        return is_degrading, slope, forecast

from enum import Enum

class AlertSeverity(Enum):
    INFO = 1
    WARNING = 2
    CRITICAL = 3
    EMERGENCY = 4

class SmartAlertSystem:
    def __init__(self):
        self.alert_history = []
        self.suppression_rules = {}

    def classify_alert(self, anomaly_score, metric_name, impact_score):
        """
        Классифицирует серьезность оповещения на основе нескольких факторов

        Args:
            anomaly_score: Насколько аномальна метрика (0-100)
            metric_name: Название затронутой метрики
            impact_score: Оценка бизнес-влияния (0-100)
        """
        # Взвешенный расчет серьезности
        severity_score = (anomaly_score * 0.6) + (impact_score * 0.4)

        if severity_score >= 90:
            return AlertSeverity.EMERGENCY
        elif severity_score >= 70:
            return AlertSeverity.CRITICAL
        elif severity_score >= 40:
            return AlertSeverity.WARNING
        else:
            return AlertSeverity.INFO

    def should_suppress_alert(self, metric_name, current_time):
        """
        Определяет, должно ли оповещение быть подавлено на основе недавней истории
        """
        # Проверка усталости от оповещений (та же метрика, несколько оповещений за короткое время)
        recent_alerts = [
            a for a in self.alert_history
            if a['metric'] == metric_name
            and (current_time - a['timestamp']).seconds < 600  # 10 минут
        ]

        if len(recent_alerts) >= 3:
            return True  # Подавить для избежания усталости от оповещений

        return False

    def generate_alert(self, anomaly_data, root_cause_analysis):
        """
        Генерирует действенное оповещение с контекстом
        """
        alert = {
            'timestamp': anomaly_data['timestamp'],
            'severity': self.classify_alert(
                anomaly_data['score'],
                anomaly_data['metric'],
                anomaly_data['impact']
            ),
            'metric': anomaly_data['metric'],
            'current_value': anomaly_data['value'],
            'expected_value': anomaly_data['baseline'],
            'deviation_percent': anomaly_data['deviation'],
            'root_cause': root_cause_analysis,
            'recommended_actions': self.get_remediation_steps(anomaly_data['metric'])
        }

        return alert

    def get_remediation_steps(self, metric_name):
        """Предоставляет контекстно-специфичное руководство по устранению"""
        remediation_map = {
            'response_time': [
                'Проверить производительность запросов к базе данных',
                'Просмотреть недавние развертывания кода',
                'Проверить зависимости внешних API',
                'Проверить использование ресурсов сервера'
            ],
            'error_rate': [
                'Просмотреть логи приложения на наличие ошибок',
                'Проверить подключение к базе данных',
                'Проверить статус сторонних сервисов',
                'Просмотреть недавние изменения конфигурации'
            ],
            'throughput': [
                'Проверить конфигурацию балансировщика нагрузки',
                'Проверить политики автомасштабирования',
                'Просмотреть настройки ограничения скорости',
                'Проверить пропускную способность сети'
            ]
        }

        return remediation_map.get(metric_name, ['Исследовать аномалию метрики'])

from prometheus_client import Gauge, Counter
import requests

class PrometheusAnomalyIntegration:
    def __init__(self, prometheus_url, grafana_url):
        self.prometheus_url = prometheus_url
        self.grafana_url = grafana_url

        # Определение пользовательских метрик
        self.anomaly_score_gauge = Gauge(
            'performance_anomaly_score',
            'Текущая оценка аномалии для метрик производительности',
            ['metric_name', 'service']
        )

        self.anomaly_counter = Counter(
            'performance_anomalies_total',
            'Общее количество обнаруженных аномалий производительности',
            ['severity', 'metric_name']
        )

    def query_metrics(self, query, start_time, end_time):
        """Запрос исторических метрик из Prometheus"""
        params = {
            'query': query,
            'start': start_time,
            'end': end_time,
            'step': '1m'
        }

        response = requests.get(
            f"{self.prometheus_url}/api/v1/query_range",
            params=params
        )

        return response.json()['data']['result']

    def publish_anomaly_metrics(self, anomalies):
        """Публикация обнаруженных аномалий обратно в Prometheus"""
        for anomaly in anomalies:
            self.anomaly_score_gauge.labels(
                metric_name=anomaly['metric'],
                service=anomaly['service']
            ).set(anomaly['score'])

            self.anomaly_counter.labels(
                severity=anomaly['severity'].name,
                metric_name=anomaly['metric']
            ).inc()

    def create_grafana_annotation(self, anomaly):
        """Создание аннотации в Grafana для обнаруженной аномалии"""
        annotation = {
            'time': int(anomaly['timestamp'].timestamp() * 1000),
            'tags': ['anomaly', anomaly['severity'].name, anomaly['metric']],
            'text': f"Обнаружена аномалия: {anomaly['metric']} - {anomaly['description']}"
        }

        requests.post(
            f"{self.grafana_url}/api/annotations",
            json=annotation,
            headers={'Authorization': f'Bearer {self.grafana_token}'}
        )