employee-performance-analytics-hr

# Employee Performance Analytics HR Skill > Skill by [ara.so](https://ara.so) — Data Skills collection. ## Overview Employee Performance Analytics is a Python and SQL-based HR analytics tool that transforms employee data into actionable insights. It uses SQLite for KPI aggregation and pandas/matplotlib for visualization, generating departmental performance reports, efficiency metrics, and workload analysis. The project provides an end-to-end analytics pipeline: data loading → SQL feature engineering → Python analysis → visualization exports. ## Installation ```bash # Clone the repository git clone https://github.com/AmirhosseinHonardoust/Employee-Performance-Analytics.git cd Employee-Performance-Analytics # Create virtual environment python -m venv .venv source .venv/bin/activate # On Windows: .venv\Scripts\activate # Install dependencies pip install -r requirements.txt ``` ### Dependencies ```txt pandas>=2.0.0 matplotlib>=3.7.0 seaborn>=0.12.0 sqlite3 # Built-in with Python numpy>=1.24.0 ``` ## Project Structure ``` employee-performance-analytics/ ├── data/ │ └── employees.csv # Raw employee data ├── src/ │ ├── create_db.py # CSV to SQLite loader │ ├── queries.sql # SQL KPI queries │ ├── analyze_performance.py # Main analysis script │ └── utils.py # Helper functions └── outputs/ ├── department_kpis.csv ├── performance_summary.csv └── charts/ # Generated visualizations ``` ## Data Schema The project expects employee data with these columns: | Column | Type | Description | |--------|------|-------------| | `employee_id` | int | Unique identifier | | `name` | string | Employee name | | `department` | string | Department (Engineering, Sales, etc.) | | `role` | string | Job title | | `date` | date | Record date (YYYY-MM-DD) | | `tasks_completed` | int | Daily tasks completed | | `hours_worked` | float | Hours worked | | `rating` | float | Performance rating (1-5) | | `projects` | int | Active projects | | `absences` | int | 1 if absent, 0 otherwise | ## Key Commands ### 1. Load Data into SQLite ```bash python src/create_db.py --csv data/employees.csv --db hr.db ``` **Options:** - `--csv`: Path to input CSV file - `--db`: Output SQLite database path - `--table`: Table name (default: `employees`) ### 2. Run Performance Analysis ```bash python src/analyze_performance.py --db hr.db --sql src/queries.sql --outdir outputs ``` **Options:** - `--db`: Path to SQLite database - `--sql`: Path to SQL queries file - `--outdir`: Output directory for CSV reports and charts ## Core SQL Queries The `queries.sql` file contains three main analytical views: ### Department KPIs ```sql -- Department-level performance metrics CREATE VIEW IF NOT EXISTS department_kpis AS SELECT department, COUNT(DISTINCT employee_id) AS employee_count, AVG(rating) AS avg_rating, SUM(tasks_completed) AS total_tasks, SUM(hours_worked) AS total_hours, ROUND(AVG(CAST(absences AS FLOAT)), 2) AS absence_rate, ROUND(SUM(tasks_completed) * 1.0 / SUM(hours_worked), 2) AS tasks_per_hour FROM employees GROUP BY department ORDER BY avg_rating DESC; ``` ### Employee Summary ```sql -- Individual employee performance aggregation CREATE VIEW IF NOT EXISTS employee_summary AS SELECT employee_id, name, department, role, SUM(tasks_completed) AS total_tasks, SUM(hours_worked) AS total_hours, AVG(rating) AS avg_rating, COUNT(DISTINCT projects) AS project_count, SUM(absences) AS total_absences, ROUND(SUM(tasks_completed) * 1.0 / SUM(hours_worked), 2) AS efficiency FROM employees GROUP BY employee_id, name, department, role ORDER BY efficiency DESC; ``` ### Daily Productivity ```sql -- Day-by-day productivity tracking CREATE VIEW IF NOT EXISTS daily_productivity AS SELECT date, department, SUM(tasks_completed) AS daily_tasks, SUM(hours_worked) AS daily_hours, AVG(rating) AS daily_rating FROM employees GROUP BY date, department ORDER BY date, department; ``` ## Python API Usage ### Creating Database from CSV ```python import pandas as pd import sqlite3 def create_database(csv_path, db_path, table_name='employees'): """Load CSV into SQLite database.""" df = pd.read_csv(csv_path) # Data validation required_cols = ['employee_id', 'name', 'department', 'date', 'tasks_completed', 'hours_worked', 'rating'] assert all(col in df.columns for col in required_cols), "Missing required columns" # Create database conn = sqlite3.connect(db_path) df.to_sql(table_name, conn, if_exists='replace', index=False) conn.close() print(f"✓ Database created: {db_path}") # Usage create_database('data/employees.csv', 'hr.db') ``` ### Running SQL Queries ```python import sqlite3 import pandas as pd def execute_sql_file(db_path, sql_file_path): """Execute SQL script and return results.""" conn = sqlite3.connect(db_path) with open(sql_file_path, 'r') as f: sql_script = f.read() # Execute all statements cursor = conn.cursor() cursor.executescript(sql_script) conn.commit() # Fetch view results dept_kpis = pd.read_sql_query("SELECT * FROM department_kpis", conn) emp_summary = pd.read_sql_query("SELECT * FROM employee_summary", conn) daily_prod = pd.read_sql_query("SELECT * FROM daily_productivity", conn) conn.close() return dept_kpis, emp_summary, daily_prod ``` ### Generating Visualizations ```python import matplotlib.pyplot as plt import seaborn as sns def plot_department_ratings(dept_kpis, output_path): """Bar chart of average rating by department.""" plt.figure(figsize=(12, 7)) sns.barplot( data=dept_kpis, x='department', y='avg_rating', palette='viridis' ) plt.title('Average Performance Rating by Department', fontsize=16, weight='bold') plt.xlabel('Department', fontsize=12) plt.ylabel('Average Rating', fontsize=12) plt.xticks(rotation=45, ha='right') plt.ylim(0, 5) plt.grid(axis='y', alpha=0.3) plt.tight_layout() plt.savefig(output_path, dpi=300, bbox_inches='tight') plt.close() def plot_performance_vs_hours(emp_summary, output_path): """Scatter plot of tasks vs hours worked.""" plt.figure(figsize=(12, 7)) scatter = plt.scatter( emp_summary['total_hours'], emp_summary['total_tasks'], c=emp_summary['avg_rating'], cmap='RdYlGn', s=100, alpha=0.6, edgecolors='black' ) plt.colorbar(scatter, label='Avg Rating') plt.title('Tasks Completed vs Hours Worked', fontsize=16, weight='bold') plt.xlabel('Total Hours Worked', fontsize=12) plt.ylabel('Total Tasks Completed', fontsize=12) plt.grid(alpha=0.3) plt.tight_layout() plt.savefig(output_path, dpi=300, bbox_inches='tight') plt.close() def plot_efficiency_distribution(emp_summary, output_path): """Histogram of task completion rate.""" plt.figure(figsize=(12, 7)) plt.hist( emp_summary['efficiency'].dropna(), bins=30, color='steelblue', edgecolor='black', alpha=0.7 ) plt.axvline( emp_summary['efficiency'].median(), color='red', linestyle='--', linewidth=2, label=f"Median: {emp_summary['efficiency'].median():.2f}" ) plt.title('Task Completion Rate Distribution', fontsize=16, weight='bold') plt.xlabel('Tasks per Hour', fontsize=12) plt.ylabel('Number of Employees', fontsize=12) plt.legend() plt.grid(alpha=0.3) plt.tight_layout() plt.savefig(output_path, dpi=300, bbox_inches='tight') plt.close() ``` ## Complete Analysis Pipeline ```python import os import sqlite3 import pandas as pd import matplotlib.pyplot as plt import seaborn as sns from pathlib import Path class HRAnalyzer: """Complete HR analytics pipeline.""" def __init__(self, db_path, sql_path, output_dir): self.db_path = db_path self.sql_path = sql_path self.output_dir = Path(output_dir) self.output_dir.mkdir(parents=True, exist_ok=True) (self.output_dir / 'charts').mkdir(exist_ok=True) def load_data(self): """Execute SQL and load analytical views.""" conn = sqlite3.connect(self.db_path) # Execute SQL script with open(self.sql_path, 'r') as f: conn.executescript(f.read()) # Load views self.dept_kpis = pd.read_sql_query("SELECT * FROM department_kpis", conn) self.emp_summary = pd.read_sql_query("SELECT * FROM employee_summary", conn) self.daily_prod = pd.read_sql_query("SELECT * FROM daily_productivity", conn) conn.close() print("✓ Data loaded from SQL views") def export_reports(self): """Save CSV reports.""" self.dept_kpis.to_csv( self.output_dir / 'department_kpis.csv', index=False ) self.emp_summary.to_csv( self.output_dir / 'performance_summary.csv', index=False ) print(f"✓ Reports saved to {self.output_dir}") def generate_visualizations(self): """Create all performance charts.""" charts_dir = self.output_dir / 'charts' # Department ratings plt.figure(figsize=(12, 7)) sns.barplot(data=self.dept_kpis, x='department', y='avg_rating', palette='viridis') plt.title('Average Rating by Department', fontsize=16, weight='bold') plt.xticks(rotation=45, ha='right') plt.tight_layout() plt.savefig(charts_dir / 'avg_rating_by_department.png', dpi=300) plt.close() # Performance vs hours plt.figure(figsize=(12, 7)) plt.scatter( self.emp_summary['total_hours'], self.emp_summary['total_tasks'], c=self.emp_summary['avg_rating'], cmap='RdYlGn', s=100, alpha=0.6 ) plt.colorbar(label='Avg Rating') plt.title('Performance vs Hours Worked', fontsize=16, weight='bold') plt.xlabel('Total Hours') plt.ylabel('Total Tasks') plt.tight_layout() plt.savefig(charts_dir / 'performance_vs_hours.png', dpi=300) plt.close() # Efficiency distribution plt.figure(figsize=(12, 7)) plt.hist(self.emp_summary['efficiency'].dropna(), bins=30, color='steelblue', edgecolor='black') plt.axvline(self.emp_summary['efficiency'].median(), color='red', linestyle='--', linewidth=2) plt.title('Task Completion Rate Distribution', fontsize=16, weight='bold') plt.xlabel('Tasks per Hour') plt.ylabel('Count') plt.tight_layout() plt.savefig(charts_dir / 'task_completion_rate.png', dpi=300) plt.close() print(f"✓ Visualizations saved to {charts_dir}") def run_full_analysis(self): """Execute complete analytics pipeline.""" self.load_data() self.export_reports() self.generate_visualizations() print("✓ Analysis complete!") # Usage analyzer = HRAnalyzer( db_path='hr.db', sql_path='src/queries.sql', output_dir='outputs' ) analyzer.run_full_analysis() ``` ## Common Patterns ### Custom KPI Queries ```python def get_top_performers(db_path, n=10):