Как понять что данные подвержены сильным изменениям во времени?

Выявление временной нестабильности данных (Concept Drift)

Время́нная нестабильность данных (concept drift) — это один из главных вызовов в аналитике. Это означает, что паттерны в данных меняются со временем, и модели/гипотезы, которые работали раньше, перестают быть верными. Давайте разберемся, как это обнаружить.

1. Визуальный анализ: смотрим на графики

Первый способ — построить временной ряд и посмотреть, есть ли структурные изменения.

-- Таблица с дневными метриками
CREATE TABLE daily_metrics (
    date DATE,
    metric_name VARCHAR,
    value DECIMAL,
    created_at TIMESTAMPTZ
);

-- Запрос для анализа временного ряда
SELECT 
    date,
    AVG(value) OVER (ORDER BY date ROWS BETWEEN 7 PRECEDING AND CURRENT ROW) as avg_7d,
    AVG(value) OVER (ORDER BY date ROWS BETWEEN 30 PRECEDING AND CURRENT ROW) as avg_30d,
    value,
    ABS(value - LAG(value) OVER (ORDER BY date)) as daily_change,
    ROUND(100.0 * ABS(value - LAG(value) OVER (ORDER BY date)) / 
        LAG(value) OVER (ORDER BY date), 2) as pct_change
FROM daily_metrics
WHERE metric_name = 'conversion_rate'
ORDER BY date DESC
LIMIT 90;

На что смотрим:

• Резкие скачки (spikes)
• Смены тренда
• Изменение волатильности
• Сезонность, которая появляется/исчезает

2. Статистические тесты на стабильность

Тест Адфуллера (Augmented Dickey-Fuller) для стационарности

from statsmodels.tsa.stattools import adfuller
import pandas as pd

# Загружаем данные
data = pd.read_sql("""
    SELECT date, value 
    FROM daily_metrics 
    WHERE metric_name = 'churn_rate'
    ORDER BY date
""", conn)

# Тест на стационарность
result = adfuller(data['value'], autolag='AIC')
print(f'ADF Statistic: {result[0]}')
print(f'p-value: {result[1]}')
print(f'Critical Values:')
for key, value in result[4].items():
    print(f'\t{key}: {value:.3f}')

# Интерпретация:
# p-value < 0.05 -> временной ряд СТАЦИОНАРНЫЙ (стабильный)
# p-value > 0.05 -> временной ряд НЕ СТАЦИОНАРНЫЙ (нестабильный)

if result[1] < 0.05:
    print("✅ Данные стабильны (нет тренда)")
else:
    print("⚠️ Данные содержат тренд (нестабильны)")

3. Анализ волатильности

Волатильность (стандартное отклонение) часто меняется во времени. Это индикатор изменения.

-- Анализ волатильности по периодам
WITH monthly_stats AS (
    SELECT 
        DATE_TRUNC('month', date) as month,
        AVG(value) as avg_value,
        STDDEV_POP(value) as stddev_value,
        COUNT(*) as sample_size,
        MIN(value) as min_value,
        MAX(value) as max_value
    FROM daily_metrics
    WHERE metric_name = 'daily_revenue'
    GROUP BY 1
)
SELECT
    month,
    ROUND(avg_value, 2) as avg_value,
    ROUND(stddev_value, 2) as volatility,
    ROUND(100.0 * stddev_value / avg_value, 2) as cv_percent,  -- coefficient of variation
    max_value - min_value as range,
    sample_size
FROM monthly_stats
ORDER BY month DESC;

-- Если CV (coefficient of variation) растет,
-- это означает, что данные становятся менее предсказуемы

4. CUSUM (Cumulative Sum Control Chart) для обнаружения сдвигов

Этот метод хорошо обнаруживает постепенные изменения в среднем.

import numpy as np
import pandas as pd
from scipy import stats

def detect_cusum_shift(series, threshold=5, drift=1):
    """
    CUSUM для обнаружения постепенного сдвига в данных
    """
    # Стандартизируем данные
    mean = series.mean()
    std = series.std()
    normalized = (series - mean) / std
    
    # Рассчитываем CUSUM
    cusum_pos = np.zeros(len(normalized))
    cusum_neg = np.zeros(len(normalized))
    
    for i in range(1, len(normalized)):
        cusum_pos[i] = max(0, cusum_pos[i-1] + normalized[i] - drift)
        cusum_neg[i] = max(0, cusum_neg[i-1] - normalized[i] - drift)
    
    # Находим точки, где превышен порог
    shift_points = np.where((cusum_pos > threshold) | (cusum_neg > threshold))[0]
    
    return shift_points, cusum_pos, cusum_neg

# Пример использования
data = pd.read_sql(
    "SELECT date, value FROM daily_metrics WHERE metric_name = 'dau'",
    conn
).set_index('date')

shift_points, cusum_pos, cusum_neg = detect_cusum_shift(data['value'])

if len(shift_points) > 0:
    print(f"⚠️ Обнаружены точки сдвига в индексах: {shift_points}")
    for point in shift_points:
        print(f"  День {data.index[point]}: потенциальное изменение")
else:
    print("✅ Значительных сдвигов не обнаружено")

5. Анализ по когортам/сегментам

Нестабильность может быть локальной (в одном сегменте), а не глобальной.

-- Анализ стабильности по платформам
WITH platform_metrics AS (
    SELECT
        DATE_TRUNC('week', date) as week,
        platform,
        AVG(conversion_rate) as avg_cr,
        STDDEV_POP(conversion_rate) as volatility,
        COUNT(*) as days_count
    FROM daily_platform_metrics
    GROUP BY 1, 2
)
SELECT
    week,
    platform,
    ROUND(avg_cr, 4) as cr,
    ROUND(volatility, 4) as volatility,
    ROUND(100.0 * volatility / NULLIF(avg_cr, 0), 2) as cv_percent
FROM platform_metrics
WHERE week >= CURRENT_DATE - INTERVAL '3 months'
ORDER BY week DESC, platform;

-- Если для iOS волатильность 0.5%, а для Android 5%,
-- это указывает на нестабильность для Android

6. Сравнение распределений (Distribution Shift)

Даже если среднее не изменилось, распределение может измениться.

from scipy.stats import ks_2samp, wasserstein_distance
import pandas as pd

# Данные за два периода
period1 = pd.read_sql(
    "SELECT value FROM daily_metrics WHERE date BETWEEN '2024-01-01' AND '2024-02-01'",
    conn
)['value']

period2 = pd.read_sql(
    "SELECT value FROM daily_metrics WHERE date BETWEEN '2024-03-01' AND '2024-04-01'",
    conn
)['value']

# Тест Колмогорова-Смирнова (KS test)
ks_statistic, ks_pvalue = ks_2samp(period1, period2)

print(f"KS Statistic: {ks_statistic:.4f}")
print(f"p-value: {ks_pvalue:.4f}")

if ks_pvalue < 0.05:
    print("⚠️ Распределения ЗНАЧИТЕЛЬНО отличаются (distribution shift)")
else:
    print("✅ Распределения похожи")

# Wasserstein Distance (более интерпретируемая метрика)
wd = wasserstein_distance(period1, period2)
print(f"Wasserstein Distance: {wd:.4f}")
print(f"Это означает, что в среднем значения отличаются на {wd:.2f}")

7. Проверка на выбросы и аномалии

Выбросы часто указывают на изменение в системе.

import numpy as np
from scipy.stats import zscore

def detect_anomalies(series, threshold=3):
    """
    Z-score для обнаружения аномалий
    """
    z_scores = np.abs(zscore(series))
    anomalies = np.where(z_scores > threshold)[0]
    
    return anomalies, z_scores

# Более продвинутый метод: Isolation Forest
from sklearn.ensemble import IsolationForest

def detect_anomalies_isolation_forest(series, contamination=0.05):
    """
    Isolation Forest для обнаружения аномалий
    """
    X = series.values.reshape(-1, 1)
    iso_forest = IsolationForest(contamination=contamination, random_state=42)
    predictions = iso_forest.fit_predict(X)
    
    anomalies = np.where(predictions == -1)[0]
    return anomalies

# Пример
data = pd.read_sql(
    "SELECT date, value FROM daily_metrics ORDER BY date",
    conn
)

anomalies = detect_anomalies_isolation_forest(data['value'], contamination=0.05)

print(f"Обнаружено {len(anomalies)} аномальных дней:")
for idx in anomalies:
    print(f"  {data.iloc[idx]['date']}: {data.iloc[idx]['value']}")

8. Практический SQL-скрипт для мониторинга

-- Комплексный скрипт мониторинга нестабильности
WITH daily_stats AS (
    SELECT
        date,
        AVG(value) OVER (ORDER BY date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) as avg_7d,
        STDDEV_POP(value) OVER (ORDER BY date ROWS BETWEEN 29 PRECEDING AND CURRENT ROW) as stddev_30d,
        value,
        LAG(value) OVER (ORDER BY date) as prev_value,
        LAG(value, 7) OVER (ORDER BY date) as prev_week_value
    FROM daily_metrics
    WHERE metric_name = 'active_users'
),
anomaly_detection AS (
    SELECT
        date,
        value,
        avg_7d,
        stddev_30d,
        ROUND(100.0 * ABS(value - avg_7d) / NULLIF(stddev_30d, 0), 2) as z_score,
        ROUND(100.0 * ABS(value - prev_value) / NULLIF(prev_value, 0), 2) as daily_change_pct,
        ROUND(100.0 * ABS(value - prev_week_value) / NULLIF(prev_week_value, 0), 2) as weekly_change_pct,
        CASE
            WHEN ABS(value - avg_7d) > 2 * stddev_30d THEN 'ANOMALY'
            WHEN ABS(value - prev_value) > 0.1 * prev_value THEN 'SPIKE'
            ELSE 'NORMAL'
        END as status
    FROM daily_stats
    WHERE date >= CURRENT_DATE - INTERVAL '30 days'
)
SELECT
    date,
    ROUND(value::NUMERIC, 2) as value,
    z_score,
    daily_change_pct,
    weekly_change_pct,
    status
FROM anomaly_detection
WHERE status != 'NORMAL' OR z_score > 2
ORDER BY date DESC;

9. Индикаторы для уведомлений

Рекомендуется настроить алерты на:

def check_data_stability(metrics_data):
    """
    Проверяет стабильность данных и возвращает алерты
    """
    alerts = []
    
    # Проверка 1: Резкий скачок
    daily_change = abs(metrics_data.iloc[-1] - metrics_data.iloc[-2]) / metrics_data.iloc[-2]
    if daily_change > 0.20:  # Больше 20%
        alerts.append(f"⚠️ ALERT: Резкое изменение на {daily_change*100:.1f}%")
    
    # Проверка 2: Тренд менее 7 дней
    recent_7d = metrics_data.iloc[-7:]
    if recent_7d.max() > recent_7d.mean() * 1.5:
        alerts.append("⚠️ ALERT: Волатильность увеличилась")
    
    # Проверка 3: Все значения выше/ниже исторического
    historical_mean = metrics_data.iloc[:-7].mean()
    recent_mean = metrics_data.iloc[-7:].mean()
    if recent_mean > historical_mean * 1.25:
        alerts.append(f"⚠️ ALERT: Сдвиг среднего на {(recent_mean/historical_mean-1)*100:.1f}%")
    
    return alerts

10. Root Cause Analysis: поиск причин

Если обнаружена нестабильность, нужно понять почему:

-- Ищем корреляцию с внешними событиями
SELECT
    dm.date,
    dm.value as metric_value,
    e.event_type,
    e.impact_scope,
    CORR(dm.value, CASE WHEN e.event_type IS NOT NULL THEN 1 ELSE 0 END) 
        OVER (ORDER BY dm.date ROWS BETWEEN 7 PRECEDING AND CURRENT ROW) as correlation
FROM daily_metrics dm
LEFT JOIN external_events e ON dm.date BETWEEN e.date AND e.date + INTERVAL '7 days'
WHERE dm.metric_name = 'revenue'
ORDER BY dm.date DESC
LIMIT 30;

-- События, которые могут вызвать изменения:
-- - Развертывание новой версии приложения
-- - Маркетинговая кампания
-- - Техническое обслуживание (downtime)
-- - Изменение в алгоритме рекомендаций
-- - Конкурентное действие

Чеклист обнаружения нестабильности

• ✅ Визуально смотрю график временного ряда за 90+ дней
• ✅ Рассчитываю скользящее среднее (7d и 30d)
• ✅ Проверяю тест Адфуллера на стационарность
• ✅ Анализирую волатильность по периодам
• ✅ Применяю CUSUM для обнаружения постепенных сдвигов
• ✅ Ищу аномалии через z-score или Isolation Forest
• ✅ Сравниваю распределения за разные периоды (KS test)
• ✅ Проверяю по подсегментам (платформа, регион, когорта)
• ✅ Ищу корреляцию с известными событиями
• ✅ Мониторю через дашборд с алертами

Заключение

Нестабильность данных может быть обнаружена несколькими способами: от простого визуального анализа до продвинутых статистических тестов. Главное — регулярно мониторить ключевые метрики и быстро реагировать на изменения, чтобы модели и гипотезы оставались актуальными.