AI-система детекции burnout сотрудников по цифровому следу

AI-система детекции burnout сотрудников по цифровому следу

Эта система отличается от системы мониторинга 1053 фокусом: не активность в корпоративных инструментах, а цифровой след в задачах, коммуникациях и продуктивности — с акцентом на ранние предвестники, обнаруживаемые за 4-8 недель до полного выгорания.

Предвестники выгорания по исследованиям

Маслах-шкала и их цифровые прокси:

maslach_dimensions = {
    'emotional_exhaustion': {
        'description': 'Истощение эмоциональных ресурсов',
        'digital_proxy': [
            'поздний старт рабочего дня (сдвиг на 1+ ч)',
            'снижение инициативных сообщений (не ответные, а созданные самим)',
            'рост времени на задачу при той же сложности'
        ]
    },
    'depersonalization': {
        'description': 'Цинизм, дистанция от работы',
        'digital_proxy': [
            'снижение качества документации (длина, форматирование)',
            'меньше добровольных code review',
            'рост времени ответа на сообщения коллег'
        ]
    },
    'reduced_accomplishment': {
        'description': 'Ощущение неэффективности',
        'digital_proxy': [
            'рост незакрытых задач при постоянном создании',
            'частые возвраты задач на доработку',
            'снижение commit-размера и рост исправительных коммитов'
        ]
    }
}

Feature Engineering из Git и Task Tracker

Анализ паттернов разработчика:

import pandas as pd
import numpy as np
from datetime import timedelta

def extract_developer_burnout_features(developer_id: str,
                                        git_log: pd.DataFrame,
                                        jira_events: pd.DataFrame,
                                        lookback_weeks: int = 8) -> dict:
    """
    Рассчитываем признаки за последние 8 недель.
    Сравниваем с baseline этого же разработчика из предыдущих 3 месяцев.
    """
    dev_commits = git_log[git_log['author_id'] == developer_id]
    dev_tasks = jira_events[jira_events['assignee_id'] == developer_id]

    # Паттерны коммитов
    recent_commits = dev_commits[
        dev_commits['timestamp'] >= pd.Timestamp.now() - timedelta(weeks=lookback_weeks)
    ]

    # Время коммитов — признак переработок
    commit_hours = recent_commits['timestamp'].dt.hour
    late_commits_ratio = (commit_hours >= 20).mean()
    weekend_commits = recent_commits['timestamp'].dt.dayofweek.isin([5, 6]).mean()

    # Размер коммитов — снижение говорит о микро-сдвигах или потере фокуса
    avg_lines_changed = recent_commits['lines_changed'].mean() if len(recent_commits) > 0 else 0

    # Fix-коммиты — растёт ли доля исправлений?
    fix_commit_ratio = recent_commits['message'].str.lower().str.contains(
        'fix|hotfix|revert|bugfix', na=False
    ).mean()

    # Задачи
    recent_tasks = dev_tasks[
        dev_tasks['event_timestamp'] >= pd.Timestamp.now() - timedelta(weeks=lookback_weeks)
    ]
    tasks_created = len(recent_tasks[recent_tasks['event'] == 'created'])
    tasks_closed = len(recent_tasks[recent_tasks['event'] == 'closed'])
    reopened_ratio = len(recent_tasks[recent_tasks['event'] == 'reopened']) / (tasks_created + 1)

    # Задержки задач
    overdue_tasks = recent_tasks[
        (recent_tasks['event'] == 'due_date_exceeded') |
        (recent_tasks['actual_days'] > recent_tasks['estimated_days'] * 1.5)
    ]
    overdue_ratio = len(overdue_tasks) / (tasks_created + 1)

    return {
        'late_commits_ratio': round(late_commits_ratio, 3),
        'weekend_work_ratio': round(weekend_commits, 3),
        'avg_commit_size_lines': round(avg_lines_changed, 1),
        'fix_commit_ratio': round(fix_commit_ratio, 3),
        'task_completion_rate': round(tasks_closed / (tasks_created + 1), 3),
        'task_reopen_ratio': round(reopened_ratio, 3),
        'task_overdue_ratio': round(overdue_ratio, 3)
    }

Временной тренд — ключевой сигнал

Детекция ухудшения паттернов:

def detect_burnout_trajectory(weekly_metrics: pd.DataFrame,
                                developer_id: str) -> dict:
    """
    Одна плохая неделя — не выгорание.
    Прогрессивное ухудшение за 4+ недель = сигнал.
    """
    dev_data = weekly_metrics[weekly_metrics['developer_id'] == developer_id].sort_values('week')

    if len(dev_data) < 6:
        return {'status': 'insufficient_history'}

    recent = dev_data.tail(8)  # последние 8 недель

    # Тренды ключевых метрик
    x = np.arange(len(recent))
    burnout_indicators = {}

    metrics_to_trend = {
        'task_completion_rate': 'decreasing',  # должен снижаться для сигнала
        'task_overdue_ratio': 'increasing',
        'fix_commit_ratio': 'increasing',
        'late_commits_ratio': 'increasing',
        'weekend_work_ratio': 'increasing'
    }

    negative_trends = 0
    for metric, direction in metrics_to_trend.items():
        if metric not in recent.columns:
            continue
        slope = np.polyfit(x, recent[metric].values, 1)[0]
        is_bad = (direction == 'increasing' and slope > 0.01) or \
                 (direction == 'decreasing' and slope < -0.01)
        burnout_indicators[f'{metric}_trend'] = slope
        if is_bad:
            negative_trends += 1

    # Ускорение — последние 4 недели хуже, чем первые 4
    first_half = dev_data.tail(8).head(4)
    second_half = dev_data.tail(4)

    acceleration_score = 0
    for metric in ['task_overdue_ratio', 'fix_commit_ratio']:
        if metric in first_half.columns:
            delta = second_half[metric].mean() - first_half[metric].mean()
            if delta > 0.05:
                acceleration_score += 1

    burnout_risk = (negative_trends / len(metrics_to_trend) * 0.6 +
                    min(1, acceleration_score / 2) * 0.4)

    return {
        'developer_id': developer_id,
        'burnout_risk_score': round(burnout_risk, 3),
        'risk_level': 'high' if burnout_risk > 0.65 else ('medium' if burnout_risk > 0.35 else 'low'),
        'negative_trend_count': negative_trends,
        'acceleration_detected': acceleration_score >= 2,
        'indicators': burnout_indicators
    }

Рекомендации менеджеру

Контекстуализированные действия:

def generate_manager_recommendations(burnout_result: dict,
                                       employee_context: dict) -> list:
    risk = burnout_result['risk_level']
    indicators = burnout_result.get('indicators', {})
    recommendations = []

    if risk == 'high':
        recommendations.append({
            'priority': 1,
            'action': '1:1 встреча в ближайшие 2-3 дня',
            'script': 'Проверьте нагрузку, попросите оценить уровень стресса (шкала 1-10)'
        })
        if indicators.get('task_overdue_ratio_trend', 0) > 0.03:
            recommendations.append({
                'priority': 2,
                'action': 'Ревизия бэклога',
                'script': 'Совместно пересмотреть приоритеты, часть задач делегировать'
            })

    if indicators.get('weekend_work_ratio_trend', 0) > 0.02:
        recommendations.append({
            'priority': 2,
            'action': 'Обсудить переработки',
            'script': 'Убедиться что работа в выходные не систематическая, выяснить причины'
        })

    if indicators.get('late_commits_ratio_trend', 0) > 0.02:
        recommendations.append({
            'priority': 3,
            'action': 'Проверить распределение нагрузки',
            'script': 'Поздние часы могут говорить о нереалистичных дедлайнах'
        })

    return recommendations

Этика и конфиденциальность: Система работает в аналитическом режиме: HR видит агрегированный риск, без доступа к конкретным коммитам или задачам. Персональные данные защищены GDPR/152-ФЗ. Сотрудник уведомлён о мониторинге (consent). Данные не используются для KPI или оценки производительности.

Сроки: Git + Jira features + risk score + HR dashboard — 4-5 недель. Trajectory analysis, acceleration detection, ML-модель на размеченных данных, manager recommendations engine — 2-3 месяца.