Employee Performance Analytics Hr

import pandas as pdimport 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}")
# Usagecreate_database('data/employees.csv', 'hr.db')

import sqlite3import 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

import matplotlib.pyplot as pltimport 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()

import osimport sqlite3import pandas as pdimport matplotlib.pyplot as pltimport seaborn as snsfrom 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!")
# Usageanalyzer = HRAnalyzer(    db_path='hr.db',    sql_path='src/queries.sql',    output_dir='outputs')analyzer.run_full_analysis()

def get_top_performers(db_path, n=10):    """Retrieve top N employees by efficiency."""    conn = sqlite3.connect(db_path)    query = """    SELECT         name,         department,         efficiency,         avg_rating    FROM employee_summary    ORDER BY efficiency DESC    LIMIT ?    """    top_performers = pd.read_sql_query(query, conn, params=(n,))    conn.close()    return top_performers
def get_department_trends(db_path, department):    """Get time-series data for specific department."""    conn = sqlite3.connect(db_path)    query = """    SELECT         date,         daily_tasks,         daily_hours,         daily_rating    FROM daily_productivity    WHERE department = ?    ORDER BY date    """    trends = pd.read_sql_query(query, conn, params=(department,))    conn.close()    return trends

def analyze_by_role(db_path, role_filter):    """Aggregate performance by specific role."""    conn = sqlite3.connect(db_path)    query = """    SELECT         role,        AVG(rating) as avg_rating,        AVG(tasks_completed) as avg_tasks,        AVG(hours_worked) as avg_hours    FROM employees    WHERE role LIKE ?    GROUP BY role    """    role_stats = pd.read_sql_query(query, conn, params=(f'%{role_filter}%',))    conn.close()    return role_stats

def calculate_workload_balance(emp_summary):    """Calculate workload balance score."""    emp_summary['workload_score'] = (        emp_summary['total_tasks'] / emp_summary['total_tasks'].max() * 0.4 +        emp_summary['total_hours'] / emp_summary['total_hours'].max() * 0.3 +        emp_summary['avg_rating'] / 5 * 0.3    )    return emp_summary

# Set global matplotlib styleplt.style.use('seaborn-v0_8-darkgrid')sns.set_palette("husl")
# Custom colorsDEPT_COLORS = {    'Engineering': '#3498db',    'Sales': '#e74c3c',    'Finance': '#2ecc71',    'HR': '#f39c12',    'Support': '#9b59b6'}

# For larger datasets, enable performance optimizationsdef optimize_database(db_path):    """Apply SQLite performance settings."""    conn = sqlite3.connect(db_path)    cursor = conn.cursor()        cursor.execute("PRAGMA journal_mode = WAL")    cursor.execute("PRAGMA synchronous = NORMAL")    cursor.execute("PRAGMA cache_size = 10000")    cursor.execute("CREATE INDEX IF NOT EXISTS idx_dept ON employees(department)")    cursor.execute("CREATE INDEX IF NOT EXISTS idx_date ON employees(date)")        conn.commit()    conn.close()

# Validate CSV before loadingrequired_columns = [    'employee_id', 'name', 'department', 'role',     'date', 'tasks_completed', 'hours_worked',     'rating', 'projects', 'absences']
df = pd.read_csv('data/employees.csv')missing = set(required_columns) - set(df.columns)if missing:    raise ValueError(f"Missing columns: {missing}")

# Check if views existdef verify_views(db_path):    conn = sqlite3.connect(db_path)    cursor = conn.cursor()    cursor.execute("SELECT name FROM sqlite_master WHERE type='view'")    views = [row[0] for row in cursor.fetchall()]    conn.close()        expected = ['department_kpis', 'employee_summary', 'daily_productivity']    missing = set(expected) - set(views)    if missing:        print(f"⚠ Missing views: {missing}. Re-run queries.sql")    else:        print("✓ All views exist")

# Safe efficiency calculationemp_summary['efficiency'] = emp_summary.apply(    lambda row: row['total_tasks'] / row['total_hours']     if row['total_hours'] > 0 else None,    axis=1)

# Ensure proper date formatdf['date'] = pd.to_datetime(df['date'], format='%Y-%m-%d', errors='coerce')df = df.dropna(subset=['date'])

def plot_monthly_trends(db_path, department):    """Monthly productivity trends for a department."""    conn = sqlite3.connect(db_path)    query = """    SELECT         strftime('%Y-%m', date) as month,        AVG(rating) as avg_rating,        SUM(tasks_completed) as total_tasks    FROM employees    WHERE department = ?    GROUP BY month    ORDER BY month    """    df = pd.read_sql_query(query, conn, params=(department,))    conn.close()        fig, ax1 = plt.subplots(figsize=(14, 7))    ax2 = ax1.twinx()        ax1.plot(df['month'], df['avg_rating'], 'b-', linewidth=2, label='Avg Rating')    ax2.bar(df['month'], df['total_tasks'], alpha=0.3, color='gray', label='Total Tasks')        ax1.set_xlabel('Month')    ax1.set_ylabel('Average Rating', color='b')    ax2.set_ylabel('Total Tasks', color='gray')    plt.title(f'{department} Performance Trends')    plt.xticks(rotation=45)    plt.tight_layout()    plt.show()

def compare_departments(dept_kpis):    """Generate department comparison report."""    comparison = dept_kpis[['department', 'avg_rating', 'tasks_per_hour', 'absence_rate']]        # Normalize metrics    for col in ['avg_rating', 'tasks_per_hour']:        comparison[f'{col}_norm'] = (            (comparison[col] - comparison[col].min()) /             (comparison[col].max() - comparison[col].min())        )        comparison['performance_index'] = (        comparison['avg_rating_norm'] * 0.5 +         comparison['tasks_per_hour_norm'] * 0.5    )        return comparison.sort_values('performance_index', ascending=False)