Building a Data-Driven Performance Management System

interface PerformanceFramework {
  objectiveMetrics: ObjectiveMetric[];
  behavioralCompetencies: BehavioralCompetency[];
  businessImpact: BusinessImpact;
}

interface ObjectiveMetric {
  name: string;
  dataSource: string;
  frequency: 'Real-time' | 'Weekly' | 'Monthly' | 'Quarterly';
  weight: number; // 0-1
  threshold: {
    exceptional: number;
    strong: number;
    meets: number;
    improvement: number;
  };
}

interface BehavioralCompetency {
  name: string;
  description: string;
  levelDefinitions: { [key: string]: string };
  assessmentMethod: '360' | 'Manager' | 'Peer' | 'Self';
}

interface BusinessImpact {
  revenue: number;
  efficiency: number;
  quality: number;
  innovation: number;
}

// Example framework for a software engineering role
const engineerFramework: PerformanceFramework = {
  objectiveMetrics: [
    {
      name: 'Code Quality Score',
      dataSource: 'SonarQube API',
      frequency: 'Weekly',
      weight: 0.25,
      threshold: {
        exceptional: 95,
        strong: 85,
        meets: 75,
        improvement: 65
      }
    },
    {
      name: 'Sprint Completion Rate',
      dataSource: 'Jira API',
      frequency: 'Monthly',
      weight: 0.20,
      threshold: {
        exceptional: 95,
        strong: 85,
        meets: 75,
        improvement: 65
      }
    },
    {
      name: 'Peer Review Participation',
      dataSource: 'GitHub API',
      frequency: 'Monthly',
      weight: 0.15,
      threshold: {
        exceptional: 20,
        strong: 15,
        meets: 10,
        improvement: 5
      }
    }
  ],
  behavioralCompetencies: [
    {
      name: 'Technical Leadership',
      description: 'Ability to guide technical decisions and mentor others',
      levelDefinitions: {
        '5': 'Sets technical strategy for org, mentors senior engineers',
        '4': 'Leads complex projects, mentors mid-level engineers',
        '3': 'Owns features independently, supports junior engineers',
        '2': 'Contributes to features with guidance',
        '1': 'Learning fundamentals'
      },
      assessmentMethod: '360'
    }
  ],
  businessImpact: {
    revenue: 0, // Calculated based on projects shipped
    efficiency: 0, // Calculated based on cycle time reduction
    quality: 0, // Calculated based on bug rates
    innovation: 0 // Calculated based on patents/novel solutions
  }
};

from dataclasses import dataclass
from typing import List, Dict
from enum import Enum

class MetricType(Enum):
    OUTPUT = "output"
    QUALITY = "quality"
    EFFICIENCY = "efficiency"
    IMPACT = "impact"

@dataclass
class PerformanceMetric:
    name: str
    metric_type: MetricType
    data_source: str
    calculation: str
    weight: float
    target: float
    unit: str

class MetricDesigner:
    """Design performance metrics for different roles"""

    @staticmethod
    def create_sales_metrics() -> List[PerformanceMetric]:
        return [
            PerformanceMetric(
                name="Revenue Closed",
                metric_type=MetricType.OUTPUT,
                data_source="Salesforce",
                calculation="SUM(closed_deals.amount)",
                weight=0.40,
                target=500000,
                unit="USD"
            ),
            PerformanceMetric(
                name="Win Rate",
                metric_type=MetricType.EFFICIENCY,
                data_source="Salesforce",
                calculation="closed_won / (closed_won + closed_lost)",
                weight=0.25,
                target=0.30,
                unit="percentage"
            ),
            PerformanceMetric(
                name="Customer Retention",
                metric_type=MetricType.QUALITY,
                data_source="Salesforce",
                calculation="retained_customers / total_customers",
                weight=0.20,
                target=0.90,
                unit="percentage"
            ),
            PerformanceMetric(
                name="Pipeline Generation",
                metric_type=MetricType.OUTPUT,
                data_source="Salesforce",
                calculation="SUM(new_opportunities.amount)",
                weight=0.15,
                target=1500000,
                unit="USD"
            )
        ]

    @staticmethod
    def create_engineer_metrics() -> List[PerformanceMetric]:
        return [
            PerformanceMetric(
                name="Story Points Delivered",
                metric_type=MetricType.OUTPUT,
                data_source="Jira",
                calculation="SUM(completed_stories.points)",
                weight=0.30,
                target=40,
                unit="points"
            ),
            PerformanceMetric(
                name="Code Quality Score",
                metric_type=MetricType.QUALITY,
                data_source="SonarQube",
                calculation="weighted_avg(maintainability, reliability, security)",
                weight=0.25,
                target=90,
                unit="score"
            ),
            PerformanceMetric(
                name="Cycle Time",
                metric_type=MetricType.EFFICIENCY,
                data_source="GitHub + Jira",
                calculation="AVG(deploy_date - start_date)",
                weight=0.20,
                target=5,
                unit="days"
            ),
            PerformanceMetric(
                name="Code Review Participation",
                metric_type=MetricType.QUALITY,
                data_source="GitHub",
                calculation="COUNT(pull_request_reviews)",
                weight=0.15,
                target=15,
                unit="reviews"
            ),
            PerformanceMetric(
                name="Production Incidents",
                metric_type=MetricType.QUALITY,
                data_source="PagerDuty",
                calculation="COUNT(incidents WHERE owner = engineer)",
                weight=0.10,
                target=2,
                unit="incidents"
            )
        ]

    @staticmethod
    def create_support_metrics() -> List[PerformanceMetric]:
        return [
            PerformanceMetric(
                name="Tickets Resolved",
                metric_type=MetricType.OUTPUT,
                data_source="Zendesk",
                calculation="COUNT(resolved_tickets)",
                weight=0.25,
                target=100,
                unit="tickets"
            ),
            PerformanceMetric(
                name="Customer Satisfaction",
                metric_type=MetricType.QUALITY,
                data_source="Zendesk",
                calculation="AVG(csat_score)",
                weight=0.35,
                target=4.5,
                unit="score"
            ),
            PerformanceMetric(
                name="First Response Time",
                metric_type=MetricType.EFFICIENCY,
                data_source="Zendesk",
                calculation="AVG(first_response_minutes)",
                weight=0.20,
                target=30,
                unit="minutes"
            ),
            PerformanceMetric(
                name="Resolution Time",
                metric_type=MetricType.EFFICIENCY,
                data_source="Zendesk",
                calculation="AVG(resolution_hours)",
                weight=0.20,
                target=24,
                unit="hours"
            )
        ]

import requests
import pandas as pd
from datetime import datetime, timedelta
from typing import Dict, List

class PerformanceDataCollector:
    """Automated collection of performance metrics from various sources"""

    def __init__(self, config: Dict[str, str]):
        self.jira_token = config.get('jira_token')
        self.github_token = config.get('github_token')
        self.salesforce_token = config.get('salesforce_token')
        self.sonarqube_token = config.get('sonarqube_token')

    def fetch_jira_metrics(self, employee_email: str, start_date: str, end_date: str) -> Dict:
        """Fetch Jira performance data"""
        jql = f'''
            assignee = "{employee_email}"
            AND status = Done
            AND resolved >= {start_date}
            AND resolved <= {end_date}
        '''

        response = requests.get(
            'https://your-domain.atlassian.net/rest/api/3/search',
            params={'jql': jql, 'fields': 'summary,storyPoints,resolution'},
            headers={'Authorization': f'Bearer {self.jira_token}'}
        )

        issues = response.json()['issues']

        return {
            'total_stories': len(issues),
            'story_points_delivered': sum(
                issue['fields'].get('storyPoints', 0) for issue in issues
            ),
            'completion_rate': len(issues) / (len(issues) + 5)  # Simplified
        }

    def fetch_github_metrics(self, username: str, start_date: str) -> Dict:
        """Fetch GitHub activity metrics"""
        query = f'''
        {{
          user(login: "{username}") {{
            contributionsCollection(from: "{start_date}") {{
              totalCommitContributions
              totalPullRequestReviewContributions
              totalPullRequestContributions
            }}
          }}
        }}
        '''

        response = requests.post(
            'https://api.github.com/graphql',
            json={'query': query},
            headers={'Authorization': f'Bearer {self.github_token}'}
        )

        data = response.json()['data']['user']['contributionsCollection']

        return {
            'commits': data['totalCommitContributions'],
            'prs_created': data['totalPullRequestContributions'],
            'pr_reviews': data['totalPullRequestReviewContributions']
        }

    def fetch_sonarqube_metrics(self, project_key: str) -> Dict:
        """Fetch code quality metrics"""
        response = requests.get(
            f'https://sonarqube.yourcompany.com/api/measures/component',
            params={
                'component': project_key,
                'metricKeys': 'reliability_rating,security_rating,sqale_rating,coverage'
            },
            headers={'Authorization': f'Bearer {self.sonarqube_token}'}
        )

        measures = response.json()['component']['measures']
        metrics = {m['metric']: float(m['value']) for m in measures}

        # Convert to 0-100 score
        quality_score = (
            (100 - metrics['reliability_rating'] * 20) * 0.30 +
            (100 - metrics['security_rating'] * 20) * 0.30 +
            (100 - metrics['sqale_rating'] * 20) * 0.20 +
            metrics['coverage'] * 0.20
        )

        return {
            'quality_score': quality_score,
            'coverage': metrics['coverage']
        }

    def aggregate_employee_metrics(self, employee: Dict, period: str = 'month') -> pd.DataFrame:
        """Aggregate all metrics for an employee"""
        end_date = datetime.now()
        start_date = end_date - timedelta(days=30 if period == 'month' else 90)

        jira_metrics = self.fetch_jira_metrics(
            employee['email'],
            start_date.strftime('%Y-%m-%d'),
            end_date.strftime('%Y-%m-%d')
        )

        github_metrics = self.fetch_github_metrics(
            employee['github_username'],
            start_date.strftime('%Y-%m-%dT%H:%M:%SZ')
        )

        sonarqube_metrics = self.fetch_sonarqube_metrics(
            employee['sonarqube_project']
        )

        # Combine all metrics
        metrics = {
            'employee_id': employee['id'],
            'employee_name': employee['name'],
            'period_start': start_date,
            'period_end': end_date,
            **jira_metrics,
            **github_metrics,
            **sonarqube_metrics
        }

        return pd.DataFrame([metrics])

# Usage example
config = {
    'jira_token': 'your_jira_token',
    'github_token': 'your_github_token',
    'sonarqube_token': 'your_sonarqube_token'
}

collector = PerformanceDataCollector(config)

employee = {
    'id': 'E001',
    'name': 'Jane Doe',
    'email': 'jane.doe@company.com',
    'github_username': 'janedoe',
    'sonarqube_project': 'project-alpha'
}

metrics_df = collector.aggregate_employee_metrics(employee)
print(metrics_df)

import plotly.graph_objects as go
import plotly.express as px
from plotly.subplots import make_subplots

class PerformanceDashboard:
    """Generate interactive performance dashboards"""

    def create_individual_dashboard(self, employee_metrics: pd.DataFrame) -> go.Figure:
        """Create dashboard for individual employee"""

        fig = make_subplots(
            rows=2, cols=2,
            subplot_titles=('Performance Trend', 'Metric Comparison',
                          'Peer Benchmark', 'Skill Development'),
            specs=[[{"type": "scatter"}, {"type": "bar"}],
                   [{"type": "box"}, {"type": "radar"}]]
        )

        # 1. Performance trend over time
        fig.add_trace(
            go.Scatter(
                x=employee_metrics['period_end'],
                y=employee_metrics['composite_score'],
                mode='lines+markers',
                name='Your Score',
                line=dict(color='#2563eb', width=3)
            ),
            row=1, col=1
        )

        # 2. Current metrics vs targets
        metrics = ['Quality', 'Efficiency', 'Output', 'Impact']
        actual = [85, 78, 92, 71]
        target = [80, 85, 85, 75]

        fig.add_trace(
            go.Bar(name='Actual', x=metrics, y=actual, marker_color='#2563eb'),
            row=1, col=2
        )
        fig.add_trace(
            go.Bar(name='Target', x=metrics, y=target, marker_color='#94a3b8'),
            row=1, col=2
        )

        # 3. Peer benchmark (anonymized)
        fig.add_trace(
            go.Box(
                y=[75, 78, 82, 85, 88, 90, 92, 95],  # Peer scores
                name='Peer Distribution',
                marker_color='#94a3b8'
            ),
            row=2, col=1
        )
        fig.add_trace(
            go.Scatter(
                x=['Peers'], y=[85],  # Employee's score
                mode='markers',
                name='Your Score',
                marker=dict(size=15, color='#2563eb')
            ),
            row=2, col=1
        )

        # 4. Skill radar chart
        skills = ['Technical', 'Leadership', 'Communication', 'Problem Solving', 'Innovation']
        current_level = [4, 3, 4, 5, 3]
        target_level = [5, 4, 4, 5, 4]

        fig.add_trace(
            go.Scatterpolar(
                r=current_level,
                theta=skills,
                fill='toself',
                name='Current',
                line_color='#2563eb'
            ),
            row=2, col=2
        )
        fig.add_trace(
            go.Scatterpolar(
                r=target_level,
                theta=skills,
                fill='toself',
                name='Target',
                line_color='#94a3b8',
                opacity=0.4
            ),
            row=2, col=2
        )

        fig.update_layout(
            title_text="Performance Dashboard - Current Quarter",
            showlegend=True,
            height=800
        )

        return fig

    def create_manager_dashboard(self, team_metrics: pd.DataFrame) -> go.Figure:
        """Create dashboard for team manager"""

        fig = make_subplots(
            rows=2, cols=2,
            subplot_titles=('Team Performance Distribution',
                          'Top & Bottom Performers',
                          'Metric Trends',
                          'Risk Indicators'),
            specs=[[{"type": "histogram"}, {"type": "bar"}],
                   [{"type": "scatter"}, {"type": "indicator"}]]
        )

        # 1. Performance distribution
        fig.add_trace(
            go.Histogram(
                x=team_metrics['composite_score'],
                nbinsx=20,
                name='Performance Distribution',
                marker_color='#2563eb'
            ),
            row=1, col=1
        )

        # 2. Top and bottom performers
        sorted_team = team_metrics.sort_values('composite_score', ascending=False)
        top_5 = sorted_team.head(5)
        bottom_5 = sorted_team.tail(5)

        fig.add_trace(
            go.Bar(
                x=top_5['employee_name'],
                y=top_5['composite_score'],
                name='Top 5',
                marker_color='#22c55e'
            ),
            row=1, col=2
        )
        fig.add_trace(
            go.Bar(
                x=bottom_5['employee_name'],
                y=bottom_5['composite_score'],
                name='Bottom 5',
                marker_color='#ef4444'
            ),
            row=1, col=2
        )

        # 3. Metric trends
        for metric in ['quality_score', 'efficiency_score', 'output_score']:
            fig.add_trace(
                go.Scatter(
                    x=team_metrics['period_end'],
                    y=team_metrics[metric],
                    mode='lines',
                    name=metric.replace('_score', '').title()
                ),
                row=2, col=1
            )

        # 4. Risk indicators
        at_risk_count = (team_metrics['composite_score'] < 70).sum()
        fig.add_trace(
            go.Indicator(
                mode="number+delta",
                value=at_risk_count,
                title={"text": "At-Risk Employees"},
                delta={'reference': 3, 'relative': True}
            ),
            row=2, col=2
        )

        fig.update_layout(
            title_text="Team Performance Dashboard",
            height=800
        )

        return fig

from typing import List
import numpy as np
from scipy import stats

class PerformanceCalibration:
    """Statistical calibration of performance ratings"""

    def __init__(self, target_distribution: Dict[str, float]):
        """
        target_distribution example:
        {
            'Exceptional': 0.10,  # Top 10%
            'Strong': 0.25,       # Next 25%
            'Meets': 0.50,        # Middle 50%
            'Improvement': 0.15   # Bottom 15%
        }
        """
        self.target_distribution = target_distribution

    def calculate_rating_inflation(self, ratings: List[str]) -> float:
        """Measure if manager is rating too high/low"""
        rating_values = {
            'Exceptional': 4,
            'Strong': 3,
            'Meets': 2,
            'Improvement': 1
        }

        actual_avg = np.mean([rating_values[r] for r in ratings])
        expected_avg = sum(
            rating_values[level] * pct
            for level, pct in self.target_distribution.items()
        )

        inflation = ((actual_avg - expected_avg) / expected_avg) * 100
        return inflation

    def suggest_rebalancing(self, ratings: List[str], scores: List[float]) -> Dict:
        """Suggest rating adjustments to match target distribution"""
        # Sort employees by score
        sorted_indices = np.argsort(scores)[::-1]  # Descending

        total_count = len(ratings)
        suggested_ratings = [None] * total_count

        # Assign ratings based on target distribution percentiles
        current_idx = 0
        for level, target_pct in self.target_distribution.items():
            count = int(total_count * target_pct)
            for i in range(count):
                if current_idx < total_count:
                    suggested_ratings[sorted_indices[current_idx]] = level
                    current_idx += 1

        # Identify changes needed
        changes = []
        for i, (current, suggested) in enumerate(zip(ratings, suggested_ratings)):
            if current != suggested:
                changes.append({
                    'employee_idx': i,
                    'current_rating': current,
                    'suggested_rating': suggested,
                    'score': scores[i]
                })

        return {
            'changes_needed': len(changes),
            'suggested_changes': changes,
            'inflation_before': self.calculate_rating_inflation(ratings),
            'inflation_after': self.calculate_rating_inflation(suggested_ratings)
        }

    def generate_calibration_report(self, manager_ratings: Dict[str, List]) -> pd.DataFrame:
        """Compare all managers' rating distributions"""
        report = []

        for manager_name, ratings in manager_ratings.items():
            distribution = {
                level: (ratings.count(level) / len(ratings)) * 100
                for level in ['Exceptional', 'Strong', 'Meets', 'Improvement']
            }

            report.append({
                'Manager': manager_name,
                'Team Size': len(ratings),
                'Inflation %': self.calculate_rating_inflation(ratings),
                **{f'{level} %': pct for level, pct in distribution.items()}
            })

        return pd.DataFrame(report)

# Usage
calibrator = PerformanceCalibration({
    'Exceptional': 0.10,
    'Strong': 0.25,
    'Meets': 0.50,
    'Improvement': 0.15
})

# Manager A's ratings
manager_a_ratings = ['Strong', 'Exceptional', 'Strong', 'Meets', 'Strong', 'Meets']
manager_a_scores = [85, 95, 82, 75, 88, 72]

rebalancing = calibrator.suggest_rebalancing(manager_a_ratings, manager_a_scores)
print(f"Rating inflation: {rebalancing['inflation_before']:.1f}%")
print(f"Changes needed: {rebalancing['changes_needed']}")

class CompensationEngine:
    """Calculate merit increases based on performance"""

    def __init__(self, budget_pct: float, compa_ratio_weight: float = 0.3):
        """
        budget_pct: Total merit budget as % of payroll (e.g., 0.04 for 4%)
        compa_ratio_weight: How much to consider position in pay range
        """
        self.budget_pct = budget_pct
        self.compa_ratio_weight = compa_ratio_weight

    def calculate_merit_increase(
        self,
        performance_rating: str,
        compa_ratio: float,
        salary: float
    ) -> Dict:
        """
        Calculate merit increase for employee

        compa_ratio = current_salary / midpoint_for_role
        < 0.80 = Below range
        0.80-1.20 = In range
        > 1.20 = Above range
        """

        # Base merit matrix
        base_merit = {
            'Exceptional': 0.06,  # 6%
            'Strong': 0.04,       # 4%
            'Meets': 0.03,        # 3%
            'Improvement': 0.00   # 0%
        }

        # Compa ratio adjustment
        compa_adjustments = {
            'low': 1.5,    # Multiply merit by 1.5 if below range
            'mid': 1.0,    # No adjustment if in range
            'high': 0.5    # Multiply merit by 0.5 if above range
        }

        # Determine compa ratio category
        if compa_ratio < 0.80:
            compa_category = 'low'
        elif compa_ratio > 1.20:
            compa_category = 'high'
        else:
            compa_category = 'mid'

        # Calculate final merit
        base_pct = base_merit[performance_rating]
        adjusted_pct = base_pct * compa_adjustments[compa_category]

        # Apply budget constraints
        final_pct = min(adjusted_pct, self.budget_pct * 2)  # Cap at 2x budget

        increase_amount = salary * final_pct
        new_salary = salary + increase_amount

        return {
            'base_merit_pct': base_pct,
            'compa_ratio': compa_ratio,
            'compa_category': compa_category,
            'final_merit_pct': final_pct,
            'increase_amount': increase_amount,
            'new_salary': new_salary
        }

    def optimize_merit_budget(
        self,
        employees: List[Dict]
    ) -> pd.DataFrame:
        """Optimize merit distribution across team within budget"""

        total_payroll = sum(e['salary'] for e in employees)
        total_budget = total_payroll * self.budget_pct

        # Calculate individual merits
        results = []
        for emp in employees:
            merit = self.calculate_merit_increase(
                emp['performance_rating'],
                emp['compa_ratio'],
                emp['salary']
            )

            results.append({
                'employee': emp['name'],
                'current_salary': emp['salary'],
                'performance': emp['performance_rating'],
                'compa_ratio': emp['compa_ratio'],
                'merit_pct': merit['final_merit_pct'],
                'increase': merit['increase_amount'],
                'new_salary': merit['new_salary']
            })

        results_df = pd.DataFrame(results)

        # Check if over budget
        total_spent = results_df['increase'].sum()
        budget_variance = total_spent - total_budget

        # If over budget, proportionally reduce
        if budget_variance > 0:
            reduction_factor = total_budget / total_spent
            results_df['increase'] = results_df['increase'] * reduction_factor
            results_df['merit_pct'] = results_df['increase'] / results_df['current_salary']
            results_df['new_salary'] = results_df['current_salary'] + results_df['increase']

        results_df['budget_used'] = (results_df['increase'].sum() / total_budget) * 100

        return results_df

# Example usage
engine = CompensationEngine(budget_pct=0.04)  # 4% merit budget

team = [
    {'name': 'Alice', 'salary': 120000, 'performance_rating': 'Exceptional', 'compa_ratio': 0.85},
    {'name': 'Bob', 'salary': 100000, 'performance_rating': 'Strong', 'compa_ratio': 1.00},
    {'name': 'Carol', 'salary': 90000, 'performance_rating': 'Meets', 'compa_ratio': 0.95},
    {'name': 'David', 'salary': 110000, 'performance_rating': 'Improvement', 'compa_ratio': 1.15}
]

merit_plan = engine.optimize_merit_budget(team)
print(merit_plan)

class PromotionReadinessModel:
    """Assess promotion readiness using multiple signals"""

    def calculate_readiness_score(self, employee_data: Dict) -> Dict:
        """
        Multi-factor promotion readiness assessment
        """

        # Factor 1: Performance history (40% weight)
        recent_ratings = employee_data['performance_ratings']  # Last 3 periods
        perf_score = self._score_performance(recent_ratings)

        # Factor 2: Skill level vs next level (25% weight)
        skill_gap_score = self._score_skill_gap(
            employee_data['current_skills'],
            employee_data['next_level_requirements']
        )

        # Factor 3: Leadership/scope expansion (20% weight)
        scope_score = self._score_scope_expansion(
            employee_data['projects_led'],
            employee_data['mentorship_count']
        )

        # Factor 4: Business impact (15% weight)
        impact_score = self._score_business_impact(
            employee_data['revenue_impact'],
            employee_data['cost_savings']
        )

        # Calculate composite score
        composite = (
            perf_score * 0.40 +
            skill_gap_score * 0.25 +
            scope_score * 0.20 +
            impact_score * 0.15
        )

        # Determine readiness level
        if composite >= 85:
            readiness = 'Ready Now'
        elif composite >= 70:
            readiness = 'Ready in 6 months'
        elif composite >= 55:
            readiness = 'Ready in 12 months'
        else:
            readiness = 'Not Ready (18+ months)'

        return {
            'composite_score': composite,
            'readiness': readiness,
            'performance_score': perf_score,
            'skill_gap_score': skill_gap_score,
            'scope_score': scope_score,
            'impact_score': impact_score,
            'development_areas': self._identify_gaps(
                skill_gap_score, scope_score, impact_score
            )
        }

    def _score_performance(self, ratings: List[str]) -> float:
        """Score based on recent performance trend"""
        rating_values = {'Exceptional': 100, 'Strong': 80, 'Meets': 60, 'Improvement': 30}
        scores = [rating_values[r] for r in ratings]

        # Weighted average (recent more important)
        weights = [0.5, 0.3, 0.2]  # Most recent = 50% weight
        weighted_score = sum(s * w for s, w in zip(scores, weights))

        return weighted_score

    def _score_skill_gap(self, current: List[Dict], required: List[Dict]) -> float:
        """Score how close to next level skill requirements"""
        total_gap = 0
        for req_skill in required:
            current_skill = next(
                (s for s in current if s['name'] == req_skill['name']),
                {'level': 0}
            )
            gap = max(0, req_skill['required_level'] - current_skill['level'])
            total_gap += gap

        max_possible_gap = len(required) * 2  # Assuming max 2 level gaps
        gap_ratio = 1 - (total_gap / max_possible_gap)

        return gap_ratio * 100

    def _score_scope_expansion(self, projects_led: int, mentees: int) -> float:
        """Score leadership and scope indicators"""
        project_score = min(projects_led * 20, 60)  # Cap at 60 points
        mentorship_score = min(mentees * 10, 40)    # Cap at 40 points

        return project_score + mentorship_score

    def _score_business_impact(self, revenue: float, cost_savings: float) -> float:
        """Score measurable business contribution"""
        # Normalize to 0-100 scale based on thresholds
        revenue_score = min((revenue / 500000) * 50, 50)
        savings_score = min((cost_savings / 100000) * 50, 50)

        return revenue_score + savings_score

    def _identify_gaps(
        self,
        skill_gap_score: float,
        scope_score: float,
        impact_score: float
    ) -> List[str]:
        """Identify which areas need development"""
        gaps = []

        if skill_gap_score < 70:
            gaps.append('Technical/functional skills')
        if scope_score < 60:
            gaps.append('Leadership and mentorship')
        if impact_score < 60:
            gaps.append('Business impact and strategic thinking')

        return gaps if gaps else ['None - ready for promotion']

class SystemHealthMonitor:
    """Monitor health of performance management system"""

    def generate_health_report(self, system_data: Dict) -> Dict:
        """Comprehensive health check"""

        metrics = {
            'adoption': {
                'managers_using_system': system_data['active_managers'],
                'total_managers': system_data['total_managers'],
                'rate': system_data['active_managers'] / system_data['total_managers'],
                'status': 'healthy' if (system_data['active_managers'] / system_data['total_managers']) > 0.95 else 'at_risk'
            },
            'data_quality': {
                'metrics_with_data': system_data['populated_metrics'],
                'total_metrics': system_data['total_metrics'],
                'completeness': system_data['populated_metrics'] / system_data['total_metrics'],
                'status': 'healthy' if (system_data['populated_metrics'] / system_data['total_metrics']) > 0.90 else 'at_risk'
            },
            'employee_satisfaction': {
                'nps_score': system_data['employee_nps'],
                'status': 'healthy' if system_data['employee_nps'] > 7 else 'at_risk'
            },
            'process_efficiency': {
                'avg_review_time_minutes': system_data['avg_review_time'],
                'target_time_minutes': 120,
                'status': 'healthy' if system_data['avg_review_time'] < 120 else 'at_risk'
            }
        }

        # Overall health
        healthy_count = sum(1 for m in metrics.values() if m['status'] == 'healthy')
        overall_health = 'Healthy' if healthy_count == len(metrics) else 'Needs Attention'

        return {
            'overall_health': overall_health,
            'metrics': metrics,
            'recommendations': self._generate_recommendations(metrics)
        }

    def _generate_recommendations(self, metrics: Dict) -> List[str]:
        """Generate improvement recommendations"""
        recommendations = []

        if metrics['adoption']['rate'] < 0.95:
            recommendations.append('Increase manager training and support')

        if metrics['data_quality']['completeness'] < 0.90:
            recommendations.append('Audit data integration pipelines')

        if metrics['employee_satisfaction']['nps_score'] < 7:
            recommendations.append('Survey employees on pain points')

        if metrics['process_efficiency']['avg_review_time_minutes'] > 120:
            recommendations.append('Simplify review process or provide templates')

        return recommendations if recommendations else ['System performing well - no actions needed']