Hr Business Partner

#!/usr/bin/env python3
"""
Compensation Analyzer - Analyze compensation data for pay equity and band alignment.

Reads employee compensation data and performs compa-ratio analysis, pay gap
calculations by demographic group, band positioning analysis, and outlier detection.
Uses standard library only -- no statsmodels or scipy required.

Usage:
    python compensation_analyzer.py --file comp_data.csv
    python compensation_analyzer.py --file comp_data.csv --json
    python compensation_analyzer.py --file comp_data.csv --group gender

Input CSV columns:
    employee_id     - Unique employee identifier
    department      - Department name
    level           - Job level (e.g., IC1, IC2, IC3, M1)
    salary          - Current annual base salary
    band_min        - Compensation band minimum for role/level
    band_max        - Compensation band maximum for role/level
    band_midpoint   - Compensation band midpoint (optional, calculated if absent)
    gender          - Gender (optional, for equity analysis)
    ethnicity       - Ethnicity (optional, for equity analysis)
    tenure_years    - Years of tenure (optional, for controlled analysis)
    performance     - Last performance rating 1-5 (optional, for controlled analysis)
    location        - Location (optional, for segmentation)

Output: Compa-ratio analysis, pay equity gaps, band positioning, outliers, and recommendations.
"""

import argparse
import csv
import json
import math
import os
import sys
from collections import defaultdict


MIN_GROUP_SIZE = 5  # Minimum group size for equity analysis


def read_csv(path: str) -> list:
    if not os.path.isfile(path):
        print(f"Error: File not found: {path}", file=sys.stderr)
        sys.exit(1)
    with open(path, "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        rows = list(reader)
    required = {"employee_id", "salary", "level"}
    if rows:
        missing = required - set(rows[0].keys())
        if missing:
            print(f"Error: Missing required columns: {', '.join(missing)}", file=sys.stderr)
            sys.exit(1)
    return rows


def safe_float(val: str, default: float = 0.0) -> float:
    try:
        return float(val)
    except (ValueError, TypeError):
        return default


def compute_compa_ratios(rows: list) -> list:
    """Compute compa-ratio for each employee."""
    results = []
    for row in rows:
        salary = safe_float(row.get("salary"))
        band_min = safe_float(row.get("band_min"))
        band_max = safe_float(row.get("band_max"))
        midpoint = safe_float(row.get("band_midpoint"))

        if not midpoint and band_min and band_max:
            midpoint = (band_min + band_max) / 2

        compa_ratio = salary / midpoint if midpoint > 0 else 0
        # Band position: 0% = at min, 100% = at max
        band_range = band_max - band_min if band_max > band_min else 1
        band_position = (salary - band_min) / band_range * 100 if band_range > 0 else 50

        # Outlier flags
        below_band = salary < band_min if band_min > 0 else False
        above_band = salary > band_max if band_max > 0 else False

        results.append({
            "employee_id": row["employee_id"],
            "department": row.get("department", ""),
            "level": row.get("level", ""),
            "salary": salary,
            "band_min": band_min,
            "band_midpoint": round(midpoint, 2),
            "band_max": band_max,
            "compa_ratio": round(compa_ratio, 3),
            "band_position_pct": round(band_position, 1),
            "below_band": below_band,
            "above_band": above_band,
            "gender": row.get("gender", ""),
            "ethnicity": row.get("ethnicity", ""),
            "tenure_years": safe_float(row.get("tenure_years")),
            "performance": safe_float(row.get("performance")),
            "location": row.get("location", ""),
        })

    return results


def compute_summary_stats(employees: list) -> dict:
    """Compute overall summary statistics."""
    salaries = [e["salary"] for e in employees if e["salary"] > 0]
    compa_ratios = [e["compa_ratio"] for e in employees if e["compa_ratio"] > 0]

    def percentile(values, pct):
        if not values:
            return 0
        s = sorted(values)
        k = (len(s) - 1) * pct / 100
        f = math.floor(k)
        c = math.ceil(k)
        if f == c:
            return s[int(k)]
        return s[f] * (c - k) + s[c] * (k - f)

    below = sum(1 for e in employees if e["below_band"])
    above = sum(1 for e in employees if e["above_band"])

    return {
        "total_employees": len(employees),
        "avg_salary": round(sum(salaries) / max(1, len(salaries)), 2),
        "median_salary": round(percentile(salaries, 50), 2),
        "avg_compa_ratio": round(sum(compa_ratios) / max(1, len(compa_ratios)), 3),
        "median_compa_ratio": round(percentile(compa_ratios, 50), 3),
        "p25_compa_ratio": round(percentile(compa_ratios, 25), 3),
        "p75_compa_ratio": round(percentile(compa_ratios, 75), 3),
        "below_band_count": below,
        "above_band_count": above,
        "within_band_count": len(employees) - below - above,
        "below_band_pct": round(below / max(1, len(employees)) * 100, 1),
        "above_band_pct": round(above / max(1, len(employees)) * 100, 1),
    }


def compute_level_analysis(employees: list) -> list:
    """Analyze compensation by level."""
    level_data = defaultdict(list)
    for e in employees:
        if e["level"]:
            level_data[e["level"]].append(e)

    results = []
    for level, emps in sorted(level_data.items()):
        salaries = [e["salary"] for e in emps]
        compa_ratios = [e["compa_ratio"] for e in emps if e["compa_ratio"] > 0]
        below = sum(1 for e in emps if e["below_band"])
        above = sum(1 for e in emps if e["above_band"])

        results.append({
            "level": level,
            "count": len(emps),
            "avg_salary": round(sum(salaries) / max(1, len(salaries))),
            "avg_compa_ratio": round(sum(compa_ratios) / max(1, len(compa_ratios)), 3),
            "below_band": below,
            "above_band": above,
        })

    return results


def compute_equity_analysis(employees: list, group_field: str) -> dict:
    """Compute pay equity analysis by demographic group."""
    groups = defaultdict(list)
    for e in employees:
        group_val = e.get(group_field, "").strip()
        if group_val:
            groups[group_val].append(e)

    # Filter groups below minimum size
    valid_groups = {k: v for k, v in groups.items() if len(v) >= MIN_GROUP_SIZE}
    suppressed = {k: len(v) for k, v in groups.items() if len(v) < MIN_GROUP_SIZE}

    if len(valid_groups) < 2:
        return {
            "field": group_field,
            "sufficient_data": False,
            "reason": f"Need at least 2 groups with {MIN_GROUP_SIZE}+ members for equity analysis",
            "suppressed_groups": suppressed,
        }

    # Raw gap analysis
    group_stats = {}
    for group, emps in valid_groups.items():
        salaries = [e["salary"] for e in emps]
        compa_ratios = [e["compa_ratio"] for e in emps if e["compa_ratio"] > 0]
        group_stats[group] = {
            "group": group,
            "count": len(emps),
            "avg_salary": round(sum(salaries) / len(salaries)),
            "avg_compa_ratio": round(sum(compa_ratios) / max(1, len(compa_ratios)), 3),
        }

    # Calculate gaps relative to highest-paid group
    max_salary_group = max(group_stats.values(), key=lambda x: x["avg_salary"])
    ref_salary = max_salary_group["avg_salary"]

    gaps = []
    for group, stats in group_stats.items():
        gap_pct = round((stats["avg_salary"] - ref_salary) / ref_salary * 100, 1) if ref_salary > 0 else 0
        gaps.append({
            **stats,
            "raw_gap_pct": gap_pct,
            "is_reference": group == max_salary_group["group"],
        })

    gaps.sort(key=lambda x: x["raw_gap_pct"])

    # Controlled gap: group by level, then compute within-level gaps
    level_gaps = []
    level_groups = defaultdict(lambda: defaultdict(list))
    for e in employees:
        group_val = e.get(group_field, "").strip()
        if group_val and group_val in valid_groups:
            level_groups[e["level"]][group_val].append(e["salary"])

    for level, grps in level_groups.items():
        if len(grps) >= 2:
            grp_avgs = {}
            for g, sals in grps.items():
                if len(sals) >= MIN_GROUP_SIZE:
                    grp_avgs[g] = sum(sals) / len(sals)
            if len(grp_avgs) >= 2:
                max_avg = max(grp_avgs.values())
                for g, avg in grp_avgs.items():
                    gap = round((avg - max_avg) / max_avg * 100, 1) if max_avg > 0 else 0
                    if gap != 0:
                        level_gaps.append({
                            "level": level,
                            "group": g,
                            "avg_salary": round(avg),
                            "controlled_gap_pct": gap,
                        })

    return {
        "field": group_field,
        "sufficient_data": True,
        "reference_group": max_salary_group["group"],
        "group_comparison": gaps,
        "controlled_gaps_by_level": level_gaps,
        "suppressed_groups": suppressed,
    }


def find_outliers(employees: list) -> list:
    """Find compensation outliers."""
    outliers = []
    for e in employees:
        reasons = []
        if e["below_band"]:
            reasons.append(f"Below band minimum (salary ${e['salary']:,.0f} vs min ${e['band_min']:,.0f})")
        if e["above_band"]:
            reasons.append(f"Above band maximum (salary ${e['salary']:,.0f} vs max ${e['band_max']:,.0f})")
        if e["compa_ratio"] > 0 and (e["compa_ratio"] < 0.85 or e["compa_ratio"] > 1.15):
            reasons.append(f"Compa-ratio {e['compa_ratio']:.3f} outside 0.85-1.15 range")

        if reasons:
            outliers.append({
                "employee_id": e["employee_id"],
                "department": e["department"],
                "level": e["level"],
                "salary": e["salary"],
                "compa_ratio": e["compa_ratio"],
                "reasons": reasons,
            })

    outliers.sort(key=lambda x: abs(x["compa_ratio"] - 1.0), reverse=True)
    return outliers


def build_recommendations(summary: dict, equity: dict, outliers: list) -> list:
    """Generate recommendations."""
    recs = []

    if summary["below_band_pct"] > 5:
        recs.append(
            f"{summary['below_band_count']} employees ({summary['below_band_pct']}%) are below band minimum. "
            "Prioritize market adjustments to bring these employees to at least band minimum within the next review cycle."
        )

    if summary["avg_compa_ratio"] < 0.93:
        recs.append(
            f"Average compa-ratio is {summary['avg_compa_ratio']:.3f}, indicating the organization is paying "
            "below band midpoints. Review market data currency and consider whether bands need updating or salaries need adjusting."
        )

    if equity.get("sufficient_data") and equity.get("controlled_gaps_by_level"):
        significant_gaps = [g for g in equity["controlled_gaps_by_level"] if abs(g["controlled_gap_pct"]) > 5]
        if significant_gaps:
            recs.append(
                f"Found {len(significant_gaps)} level-controlled pay gaps exceeding 5%. "
                "Conduct a detailed pay equity review with Legal and Total Rewards to determine root causes and remediation."
            )

    if len(outliers) > 0:
        recs.append(
            f"{len(outliers)} employees flagged as compensation outliers. "
            "Review each case with the relevant HRBP and manager to determine if adjustments are needed."
        )

    if not recs:
        recs.append("Compensation is well-aligned across the organization. Continue monitoring quarterly.")

    return recs


def format_human(summary: dict, levels: list, equity: dict, outliers: list, recommendations: list) -> str:
    """Format results for human-readable output."""
    lines = []
    lines.append("=" * 70)
    lines.append("COMPENSATION ANALYSIS REPORT")
    lines.append("=" * 70)
    lines.append("")
    lines.append(f"  Total Employees:       {summary['total_employees']}")
    lines.append(f"  Average Salary:        ${summary['avg_salary']:,.0f}")
    lines.append(f"  Median Salary:         ${summary['median_salary']:,.0f}")
    lines.append(f"  Average Compa-Ratio:   {summary['avg_compa_ratio']:.3f}")
    lines.append(f"  Median Compa-Ratio:    {summary['median_compa_ratio']:.3f}")
    lines.append(f"  Below Band:            {summary['below_band_count']} ({summary['below_band_pct']}%)")
    lines.append(f"  Above Band:            {summary['above_band_count']} ({summary['above_band_pct']}%)")
    lines.append(f"  Within Band:           {summary['within_band_count']}")

    lines.append("")
    lines.append("-" * 70)
    lines.append("BY LEVEL")
    lines.append("-" * 70)
    lines.append(f"  {'Level':<10} {'Count':>6} {'Avg Salary':>12} {'Avg CR':>8} {'Below':>6} {'Above':>6}")
    lines.append(f"  {'-'*10} {'-'*6} {'-'*12} {'-'*8} {'-'*6} {'-'*6}")
    for lv in levels:
        lines.append(
            f"  {lv['level']:<10} {lv['count']:>6} ${lv['avg_salary']:>10,} {lv['avg_compa_ratio']:>8.3f} "
            f"{lv['below_band']:>6} {lv['above_band']:>6}"
        )

    if equity.get("sufficient_data"):
        lines.append("")
        lines.append("-" * 70)
        lines.append(f"PAY EQUITY ANALYSIS (by {equity['field']})")
        lines.append("-" * 70)
        lines.append(f"  Reference group: {equity['reference_group']}")
        lines.append(f"  {'Group':<20} {'Count':>6} {'Avg Salary':>12} {'Avg CR':>8} {'Raw Gap':>8}")
        lines.append(f"  {'-'*20} {'-'*6} {'-'*12} {'-'*8} {'-'*8}")
        for g in equity["group_comparison"]:
            ref = " (ref)" if g["is_reference"] else ""
            lines.append(
                f"  {g['group']:<20} {g['count']:>6} ${g['avg_salary']:>10,} {g['avg_compa_ratio']:>8.3f} "
                f"{g['raw_gap_pct']:>+7.1f}%{ref}"
            )

        if equity.get("controlled_gaps_by_level"):
            lines.append("\n  Controlled gaps (within-level):")
            for g in equity["controlled_gaps_by_level"]:
                lines.append(f"    {g['level']}: {g['group']} gap = {g['controlled_gap_pct']:+.1f}% (avg ${g['avg_salary']:,})")

    if outliers:
        lines.append("")
        lines.append("-" * 70)
        lines.append(f"OUTLIERS ({len(outliers)} flagged)")
        lines.append("-" * 70)
        for o in outliers[:15]:
            lines.append(f"  {o['employee_id']} | {o['department']} | {o['level']} | ${o['salary']:,.0f} | CR: {o['compa_ratio']:.3f}")
            for r in o["reasons"]:
                lines.append(f"    - {r}")

    lines.append("")
    lines.append("-" * 70)
    lines.append("RECOMMENDATIONS")
    lines.append("-" * 70)
    for i, rec in enumerate(recommendations, 1):
        lines.append(f"  {i}. {rec}")

    lines.append("")
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Analyze compensation data for pay equity and band alignment."
    )
    parser.add_argument("--file", required=True, help="Path to compensation data CSV")
    parser.add_argument("--group", default="gender", help="Demographic field for equity analysis (default: gender)")
    parser.add_argument("--json", action="store_true", help="Output results as JSON")
    args = parser.parse_args()

    rows = read_csv(args.file)
    if not rows:
        print("Error: No data found in CSV file.", file=sys.stderr)
        sys.exit(1)

    employees = compute_compa_ratios(rows)
    summary = compute_summary_stats(employees)
    levels = compute_level_analysis(employees)
    equity = compute_equity_analysis(employees, args.group)
    outliers = find_outliers(employees)
    recommendations = build_recommendations(summary, equity, outliers)

    if args.json:
        output = {
            "summary": summary,
            "level_analysis": levels,
            "equity_analysis": equity,
            "outliers": outliers,
            "recommendations": recommendations,
        }
        print(json.dumps(output, indent=2))
    else:
        print(format_human(summary, levels, equity, outliers, recommendations))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Org Health Scorer - Score organizational health from workforce metrics.

Reads a CSV of organizational metrics per department/team and computes a
composite health score across six dimensions: retention, engagement,
performance, development, compensation, and diversity.

Usage:
    python org_health_scorer.py --file org_metrics.csv
    python org_health_scorer.py --file org_metrics.csv --json
    python org_health_scorer.py --file org_metrics.csv --threshold 60

Input CSV columns:
    department              - Department or team name
    headcount               - Total headcount
    voluntary_turnover_pct  - Voluntary turnover rate (e.g., 12 for 12%)
    regrettable_turnover_pct - Regrettable turnover as % of total exits
    engagement_score        - Engagement survey score (1-100 scale)
    enps                    - Employee Net Promoter Score (-100 to 100)
    high_performer_pct      - Percentage rated as high performers
    low_performer_pct       - Percentage rated as low performers
    promotion_rate_pct      - Annual promotion rate
    training_hours_avg      - Average training hours per employee
    compa_ratio_avg         - Average compa-ratio (actual/midpoint)
    open_role_pct           - Open roles as % of headcount
    diversity_pct           - Underrepresented group representation %
    manager_score           - Manager effectiveness score (1-5)
    span_of_control         - Average direct reports per manager

Output: Per-department health scores with dimension breakdowns and recommendations.
"""

import argparse
import csv
import json
import os
import sys


# --- Scoring functions for each dimension (0-100) ---

def score_retention(turnover_pct: float, regrettable_pct: float) -> tuple:
    """Score retention health."""
    # Voluntary turnover: <10% excellent, 10-15% good, 15-20% concerning, >20% critical
    if turnover_pct <= 8:
        turn_score = 100
    elif turnover_pct <= 12:
        turn_score = 85
    elif turnover_pct <= 15:
        turn_score = 70
    elif turnover_pct <= 20:
        turn_score = 50
    elif turnover_pct <= 25:
        turn_score = 30
    else:
        turn_score = 15

    # Regrettable: <20% excellent, 20-30% good, >30% concerning
    if regrettable_pct <= 15:
        reg_score = 100
    elif regrettable_pct <= 25:
        reg_score = 80
    elif regrettable_pct <= 35:
        reg_score = 60
    else:
        reg_score = 35

    score = round(turn_score * 0.6 + reg_score * 0.4)
    issues = []
    if turnover_pct > 15:
        issues.append(f"Voluntary turnover at {turnover_pct:.1f}% exceeds 15% threshold")
    if regrettable_pct > 30:
        issues.append(f"Regrettable turnover at {regrettable_pct:.1f}% exceeds 30% threshold")
    return score, issues


def score_engagement(engagement: float, enps: float) -> tuple:
    """Score engagement health."""
    # Engagement: 80+ excellent, 70-80 good, 60-70 fair, <60 poor
    if engagement >= 80:
        eng_score = 100
    elif engagement >= 70:
        eng_score = 80
    elif engagement >= 60:
        eng_score = 60
    elif engagement >= 50:
        eng_score = 40
    else:
        eng_score = 20

    # eNPS: 40+ excellent, 20-40 good, 0-20 fair, <0 poor
    if enps >= 40:
        enps_score = 100
    elif enps >= 20:
        enps_score = 80
    elif enps >= 0:
        enps_score = 55
    elif enps >= -20:
        enps_score = 30
    else:
        enps_score = 15

    score = round(eng_score * 0.6 + enps_score * 0.4)
    issues = []
    if engagement < 65:
        issues.append(f"Engagement score {engagement:.0f} below 65 threshold")
    if enps < 10:
        issues.append(f"eNPS at {enps:.0f} below minimum of 10")
    return score, issues


def score_performance(high_perf_pct: float, low_perf_pct: float) -> tuple:
    """Score performance distribution health."""
    # High performers: 15-25% is healthy
    if 15 <= high_perf_pct <= 25:
        hp_score = 100
    elif 10 <= high_perf_pct < 15 or 25 < high_perf_pct <= 35:
        hp_score = 75
    elif high_perf_pct > 35:
        hp_score = 50  # Rating inflation
    else:
        hp_score = 50

    # Low performers: 5-10% is healthy (shows differentiation)
    if 5 <= low_perf_pct <= 10:
        lp_score = 100
    elif 3 <= low_perf_pct < 5:
        lp_score = 80
    elif low_perf_pct < 3:
        lp_score = 60  # Possible lack of differentiation
    elif low_perf_pct <= 20:
        lp_score = 50
    else:
        lp_score = 30

    score = round(hp_score * 0.6 + lp_score * 0.4)
    issues = []
    if high_perf_pct > 35:
        issues.append(f"High performer rate {high_perf_pct:.1f}% suggests rating inflation")
    if low_perf_pct < 3:
        issues.append(f"Low performer rate {low_perf_pct:.1f}% suggests lack of differentiation")
    if low_perf_pct > 15:
        issues.append(f"Low performer rate {low_perf_pct:.1f}% is elevated")
    return score, issues


def score_development(promotion_rate: float, training_hours: float) -> tuple:
    """Score development investment health."""
    # Promotion rate: 8-12% is healthy
    if 8 <= promotion_rate <= 15:
        promo_score = 100
    elif 5 <= promotion_rate < 8:
        promo_score = 70
    elif promotion_rate > 15:
        promo_score = 70  # Possible title inflation
    else:
        promo_score = 40

    # Training hours: 40+ excellent, 20-40 good, <20 concerning
    if training_hours >= 40:
        train_score = 100
    elif training_hours >= 25:
        train_score = 80
    elif training_hours >= 15:
        train_score = 60
    else:
        train_score = 35

    score = round(promo_score * 0.5 + train_score * 0.5)
    issues = []
    if promotion_rate < 5:
        issues.append(f"Promotion rate {promotion_rate:.1f}% below 5% may indicate career stagnation")
    if training_hours < 15:
        issues.append(f"Average training {training_hours:.0f} hrs below 15 hr minimum")
    return score, issues


def score_compensation(compa_ratio: float) -> tuple:
    """Score compensation health."""
    # Target: 0.95-1.05
    if 0.95 <= compa_ratio <= 1.05:
        score = 100
    elif 0.90 <= compa_ratio < 0.95 or 1.05 < compa_ratio <= 1.10:
        score = 80
    elif 0.85 <= compa_ratio < 0.90 or 1.10 < compa_ratio <= 1.15:
        score = 60
    elif 0.80 <= compa_ratio < 0.85:
        score = 40
    else:
        score = 25

    issues = []
    if compa_ratio < 0.90:
        issues.append(f"Compa-ratio {compa_ratio:.2f} significantly below midpoint")
    if compa_ratio > 1.10:
        issues.append(f"Compa-ratio {compa_ratio:.2f} significantly above midpoint")
    return score, issues


def score_structure(open_role_pct: float, span: float, manager_score: float) -> tuple:
    """Score organizational structure health."""
    # Open roles: <5% healthy, 5-10% manageable, >10% strained
    if open_role_pct <= 5:
        open_score = 100
    elif open_role_pct <= 10:
        open_score = 75
    elif open_role_pct <= 15:
        open_score = 50
    else:
        open_score = 25

    # Span of control: 5-8 is optimal
    if 5 <= span <= 8:
        span_score = 100
    elif 4 <= span < 5 or 8 < span <= 10:
        span_score = 75
    elif 3 <= span < 4 or 10 < span <= 12:
        span_score = 50
    else:
        span_score = 30

    # Manager score: 4.0+ excellent
    if manager_score >= 4.0:
        mgr_score = 100
    elif manager_score >= 3.5:
        mgr_score = 75
    elif manager_score >= 3.0:
        mgr_score = 50
    else:
        mgr_score = 30

    score = round(open_score * 0.3 + span_score * 0.3 + mgr_score * 0.4)
    issues = []
    if open_role_pct > 10:
        issues.append(f"Open role rate {open_role_pct:.1f}% indicates capacity strain")
    if span < 4 or span > 10:
        issues.append(f"Span of control {span:.1f} outside 5-8 optimal range")
    if manager_score < 3.5:
        issues.append(f"Manager effectiveness {manager_score:.1f} below 3.5 threshold")
    return score, issues


DIMENSION_WEIGHTS = {
    "retention": 25,
    "engagement": 25,
    "performance": 15,
    "development": 15,
    "compensation": 10,
    "structure": 10,
}


def safe_float(val: str, default: float = 0.0) -> float:
    try:
        return float(val)
    except (ValueError, TypeError):
        return default


def read_csv_file(path: str) -> list:
    if not os.path.isfile(path):
        print(f"Error: File not found: {path}", file=sys.stderr)
        sys.exit(1)
    with open(path, "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        rows = list(reader)
    required = {"department", "headcount"}
    if rows:
        missing = required - set(rows[0].keys())
        if missing:
            print(f"Error: Missing required columns: {', '.join(missing)}", file=sys.stderr)
            sys.exit(1)
    return rows


def score_department(row: dict) -> dict:
    """Score a single department."""
    dept = row["department"]
    hc = safe_float(row.get("headcount", 0))

    retention_score, retention_issues = score_retention(
        safe_float(row.get("voluntary_turnover_pct", 12)),
        safe_float(row.get("regrettable_turnover_pct", 25)),
    )
    engagement_s, engagement_issues = score_engagement(
        safe_float(row.get("engagement_score", 70)),
        safe_float(row.get("enps", 20)),
    )
    performance_s, performance_issues = score_performance(
        safe_float(row.get("high_performer_pct", 18)),
        safe_float(row.get("low_performer_pct", 7)),
    )
    development_s, development_issues = score_development(
        safe_float(row.get("promotion_rate_pct", 10)),
        safe_float(row.get("training_hours_avg", 25)),
    )
    compensation_s, compensation_issues = score_compensation(
        safe_float(row.get("compa_ratio_avg", 1.0)),
    )
    structure_s, structure_issues = score_structure(
        safe_float(row.get("open_role_pct", 5)),
        safe_float(row.get("span_of_control", 6)),
        safe_float(row.get("manager_score", 3.8)),
    )

    # Weighted overall
    overall = round(
        retention_score * DIMENSION_WEIGHTS["retention"] / 100
        + engagement_s * DIMENSION_WEIGHTS["engagement"] / 100
        + performance_s * DIMENSION_WEIGHTS["performance"] / 100
        + development_s * DIMENSION_WEIGHTS["development"] / 100
        + compensation_s * DIMENSION_WEIGHTS["compensation"] / 100
        + structure_s * DIMENSION_WEIGHTS["structure"] / 100
    )

    # Health level
    if overall >= 80:
        health = "HEALTHY"
    elif overall >= 65:
        health = "WATCH"
    elif overall >= 50:
        health = "AT_RISK"
    else:
        health = "CRITICAL"

    all_issues = retention_issues + engagement_issues + performance_issues + development_issues + compensation_issues + structure_issues

    return {
        "department": dept,
        "headcount": int(hc),
        "overall_score": overall,
        "health_level": health,
        "dimensions": {
            "retention": {"score": retention_score, "weight": DIMENSION_WEIGHTS["retention"], "issues": retention_issues},
            "engagement": {"score": engagement_s, "weight": DIMENSION_WEIGHTS["engagement"], "issues": engagement_issues},
            "performance": {"score": performance_s, "weight": DIMENSION_WEIGHTS["performance"], "issues": performance_issues},
            "development": {"score": development_s, "weight": DIMENSION_WEIGHTS["development"], "issues": development_issues},
            "compensation": {"score": compensation_s, "weight": DIMENSION_WEIGHTS["compensation"], "issues": compensation_issues},
            "structure": {"score": structure_s, "weight": DIMENSION_WEIGHTS["structure"], "issues": structure_issues},
        },
        "all_issues": all_issues,
    }


def compute_org_summary(results: list) -> dict:
    """Compute org-level summary."""
    total_hc = sum(r["headcount"] for r in results)
    # Weighted average by headcount
    if total_hc > 0:
        weighted_score = sum(r["overall_score"] * r["headcount"] for r in results) / total_hc
    else:
        weighted_score = sum(r["overall_score"] for r in results) / max(1, len(results))

    health_dist = {"HEALTHY": 0, "WATCH": 0, "AT_RISK": 0, "CRITICAL": 0}
    for r in results:
        health_dist[r["health_level"]] += 1

    return {
        "total_departments": len(results),
        "total_headcount": int(total_hc),
        "org_health_score": round(weighted_score),
        "health_distribution": health_dist,
    }


def format_human(results: list, summary: dict, threshold: float) -> str:
    """Format results for human-readable output."""
    lines = []
    lines.append("=" * 70)
    lines.append("ORGANIZATIONAL HEALTH REPORT")
    lines.append("=" * 70)
    lines.append("")

    health_label = "HEALTHY" if summary["org_health_score"] >= 80 else "WATCH" if summary["org_health_score"] >= 65 else "AT RISK" if summary["org_health_score"] >= 50 else "CRITICAL"
    lines.append(f"  Org Health Score:      {summary['org_health_score']}/100 ({health_label})")
    lines.append(f"  Total Headcount:       {summary['total_headcount']}")
    lines.append(f"  Departments Assessed:  {summary['total_departments']}")
    dist = summary["health_distribution"]
    lines.append(f"  Health Distribution:   Healthy: {dist['HEALTHY']} | Watch: {dist['WATCH']} | At Risk: {dist['AT_RISK']} | Critical: {dist['CRITICAL']}")
    lines.append("")

    lines.append("-" * 70)
    lines.append("DEPARTMENT SCORES")
    lines.append("-" * 70)
    lines.append(f"  {'Department':<22} {'HC':>5} {'Score':>6} {'Status':<10} {'Ret':>4} {'Eng':>4} {'Perf':>5} {'Dev':>4} {'Comp':>5} {'Str':>4}")
    lines.append(f"  {'-'*22} {'-'*5} {'-'*6} {'-'*10} {'-'*4} {'-'*4} {'-'*5} {'-'*4} {'-'*5} {'-'*4}")

    for r in sorted(results, key=lambda x: x["overall_score"]):
        d = r["dimensions"]
        lines.append(
            f"  {r['department']:<22} {r['headcount']:>5} {r['overall_score']:>5}/100 {r['health_level']:<10} "
            f"{d['retention']['score']:>4} {d['engagement']['score']:>4} {d['performance']['score']:>5} "
            f"{d['development']['score']:>4} {d['compensation']['score']:>5} {d['structure']['score']:>4}"
        )

    # Issues for departments below threshold
    flagged = [r for r in results if r["overall_score"] < threshold]
    if flagged:
        lines.append("")
        lines.append("-" * 70)
        lines.append(f"ISSUES (Departments Below {threshold} Threshold)")
        lines.append("-" * 70)
        for r in sorted(flagged, key=lambda x: x["overall_score"]):
            lines.append(f"\n  {r['department']} ({r['overall_score']}/100 - {r['health_level']})")
            for issue in r["all_issues"]:
                lines.append(f"    - {issue}")

    lines.append("")
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Score organizational health from workforce metrics."
    )
    parser.add_argument("--file", required=True, help="Path to org metrics CSV")
    parser.add_argument("--threshold", type=float, default=65, help="Score threshold for flagging (default: 65)")
    parser.add_argument("--json", action="store_true", help="Output results as JSON")
    args = parser.parse_args()

    rows = read_csv_file(args.file)
    if not rows:
        print("Error: No data found in CSV file.", file=sys.stderr)
        sys.exit(1)

    results = [score_department(row) for row in rows]
    summary = compute_org_summary(results)

    if args.json:
        output = {"summary": summary, "departments": results}
        print(json.dumps(output, indent=2))
    else:
        print(format_human(results, summary, args.threshold))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Workforce Dashboard - Generate HR metrics dashboard from workforce data.

Reads a workforce CSV and produces a comprehensive dashboard covering
headcount, attrition, tenure distribution, performance distribution,
diversity metrics, and department-level breakdowns.

Usage:
    python workforce_dashboard.py --file workforce.csv
    python workforce_dashboard.py --file workforce.csv --json
    python workforce_dashboard.py --file workforce.csv --period Q1-2026

Input CSV columns:
    employee_id     - Unique employee identifier
    department      - Department name
    level           - Job level
    status          - Employment status (Active, Terminated, Resigned, etc.)
    hire_date       - Date of hire (YYYY-MM-DD)
    term_date       - Termination date if applicable (YYYY-MM-DD)
    term_type       - Termination type (Voluntary, Involuntary, blank if active)
    salary          - Current annual salary
    performance     - Last performance rating (1-5)
    gender          - Gender (optional)
    ethnicity       - Ethnicity (optional)
    location        - Office location (optional)
    manager_id      - Manager's employee ID (optional)

Output: Workforce dashboard with headcount, attrition, tenure, performance, diversity metrics.
"""

import argparse
import csv
import json
import os
import sys
from collections import defaultdict
from datetime import datetime, date


def read_csv(path: str) -> list:
    if not os.path.isfile(path):
        print(f"Error: File not found: {path}", file=sys.stderr)
        sys.exit(1)
    with open(path, "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        rows = list(reader)
    required = {"employee_id", "department", "status"}
    if rows:
        missing = required - set(rows[0].keys())
        if missing:
            print(f"Error: Missing required columns: {', '.join(missing)}", file=sys.stderr)
            sys.exit(1)
    return rows


def parse_date(val: str) -> date:
    if not val or not val.strip():
        return None
    for fmt in ("%Y-%m-%d", "%m/%d/%Y", "%d/%m/%Y"):
        try:
            return datetime.strptime(val.strip(), fmt).date()
        except ValueError:
            continue
    return None


def safe_float(val: str, default: float = 0.0) -> float:
    try:
        return float(val)
    except (ValueError, TypeError):
        return default


def compute_headcount(rows: list) -> dict:
    """Compute headcount metrics."""
    active = [r for r in rows if r.get("status", "").strip().lower() == "active"]
    terminated = [r for r in rows if r.get("status", "").strip().lower() in ("terminated", "resigned")]

    dept_counts = defaultdict(int)
    level_counts = defaultdict(int)
    location_counts = defaultdict(int)

    for r in active:
        dept_counts[r.get("department", "Unknown")] += 1
        level_counts[r.get("level", "Unknown")] += 1
        loc = r.get("location", "Unknown") or "Unknown"
        location_counts[loc] += 1

    return {
        "total_active": len(active),
        "total_terminated": len(terminated),
        "total_records": len(rows),
        "by_department": dict(sorted(dept_counts.items(), key=lambda x: -x[1])),
        "by_level": dict(sorted(level_counts.items())),
        "by_location": dict(sorted(location_counts.items(), key=lambda x: -x[1])),
    }


def compute_attrition(rows: list) -> dict:
    """Compute attrition metrics."""
    today = date.today()
    active = [r for r in rows if r.get("status", "").strip().lower() == "active"]
    terminated = [r for r in rows if r.get("status", "").strip().lower() in ("terminated", "resigned")]

    voluntary = [r for r in terminated if r.get("term_type", "").strip().lower() == "voluntary"]
    involuntary = [r for r in terminated if r.get("term_type", "").strip().lower() == "involuntary"]

    avg_hc = (len(active) + len(active) + len(terminated)) / 2  # Approximation

    total_rate = round(len(terminated) / max(1, avg_hc) * 100, 1)
    voluntary_rate = round(len(voluntary) / max(1, avg_hc) * 100, 1)
    involuntary_rate = round(len(involuntary) / max(1, avg_hc) * 100, 1)

    # Attrition by department
    dept_attrition = defaultdict(lambda: {"active": 0, "terminated": 0})
    for r in active:
        dept_attrition[r.get("department", "Unknown")]["active"] += 1
    for r in terminated:
        dept_attrition[r.get("department", "Unknown")]["terminated"] += 1

    dept_rates = {}
    for dept, counts in dept_attrition.items():
        total = counts["active"] + counts["terminated"]
        dept_rates[dept] = round(counts["terminated"] / max(1, total) * 100, 1)

    return {
        "total_exits": len(terminated),
        "voluntary_exits": len(voluntary),
        "involuntary_exits": len(involuntary),
        "total_attrition_rate": total_rate,
        "voluntary_attrition_rate": voluntary_rate,
        "involuntary_attrition_rate": involuntary_rate,
        "by_department": dict(sorted(dept_rates.items(), key=lambda x: -x[1])),
    }


def compute_tenure(rows: list) -> dict:
    """Compute tenure distribution."""
    today = date.today()
    active = [r for r in rows if r.get("status", "").strip().lower() == "active"]

    tenures = []
    for r in active:
        hire = parse_date(r.get("hire_date", ""))
        if hire:
            months = (today.year - hire.year) * 12 + (today.month - hire.month)
            tenures.append(months)

    if not tenures:
        return {"avg_tenure_months": 0, "distribution": {}}

    bands = {
        "< 6 months": 0,
        "6-12 months": 0,
        "1-2 years": 0,
        "2-5 years": 0,
        "5-10 years": 0,
        "10+ years": 0,
    }

    for t in tenures:
        if t < 6:
            bands["< 6 months"] += 1
        elif t < 12:
            bands["6-12 months"] += 1
        elif t < 24:
            bands["1-2 years"] += 1
        elif t < 60:
            bands["2-5 years"] += 1
        elif t < 120:
            bands["5-10 years"] += 1
        else:
            bands["10+ years"] += 1

    avg_tenure = sum(tenures) / len(tenures)
    sorted_t = sorted(tenures)
    median_tenure = sorted_t[len(sorted_t) // 2]

    return {
        "avg_tenure_months": round(avg_tenure, 1),
        "median_tenure_months": median_tenure,
        "distribution": bands,
    }


def compute_performance(rows: list) -> dict:
    """Compute performance distribution."""
    active = [r for r in rows if r.get("status", "").strip().lower() == "active"]

    ratings = []
    for r in active:
        perf = safe_float(r.get("performance"))
        if 1 <= perf <= 5:
            ratings.append(perf)

    if not ratings:
        return {"avg_rating": 0, "distribution": {}}

    dist = {1: 0, 2: 0, 3: 0, 4: 0, 5: 0}
    for r in ratings:
        dist[int(round(r))] += 1

    total = len(ratings)
    dist_pct = {k: round(v / total * 100, 1) for k, v in dist.items()}

    high_performers = sum(1 for r in ratings if r >= 4) / total * 100
    low_performers = sum(1 for r in ratings if r <= 2) / total * 100

    return {
        "avg_rating": round(sum(ratings) / total, 2),
        "rated_count": total,
        "distribution": dist,
        "distribution_pct": dist_pct,
        "high_performer_pct": round(high_performers, 1),
        "low_performer_pct": round(low_performers, 1),
    }


def compute_diversity(rows: list) -> dict:
    """Compute diversity metrics."""
    active = [r for r in rows if r.get("status", "").strip().lower() == "active"]

    gender_counts = defaultdict(int)
    ethnicity_counts = defaultdict(int)
    gender_by_level = defaultdict(lambda: defaultdict(int))

    for r in active:
        gender = r.get("gender", "").strip()
        ethnicity = r.get("ethnicity", "").strip()
        level = r.get("level", "Unknown")

        if gender:
            gender_counts[gender] += 1
            gender_by_level[level][gender] += 1
        if ethnicity:
            ethnicity_counts[ethnicity] += 1

    total = len(active)

    def to_pct(counts):
        return {k: round(v / max(1, total) * 100, 1) for k, v in sorted(counts.items(), key=lambda x: -x[1])}

    return {
        "gender_distribution": to_pct(gender_counts),
        "ethnicity_distribution": to_pct(ethnicity_counts),
        "gender_by_level": {level: dict(genders) for level, genders in sorted(gender_by_level.items())},
    }


def compute_compensation_summary(rows: list) -> dict:
    """Compute basic compensation summary."""
    active = [r for r in rows if r.get("status", "").strip().lower() == "active"]
    salaries = [safe_float(r.get("salary")) for r in active if safe_float(r.get("salary")) > 0]

    if not salaries:
        return {"avg_salary": 0, "total_payroll": 0}

    s = sorted(salaries)
    n = len(s)

    return {
        "avg_salary": round(sum(s) / n),
        "median_salary": round(s[n // 2]),
        "min_salary": round(s[0]),
        "max_salary": round(s[-1]),
        "total_payroll": round(sum(s)),
        "employee_count": n,
    }


def compute_manager_spans(rows: list) -> dict:
    """Compute span of control metrics."""
    active = [r for r in rows if r.get("status", "").strip().lower() == "active"]

    manager_counts = defaultdict(int)
    for r in active:
        mgr = r.get("manager_id", "").strip()
        if mgr:
            manager_counts[mgr] += 1

    if not manager_counts:
        return {"avg_span": 0, "managers": 0}

    spans = list(manager_counts.values())
    avg_span = sum(spans) / len(spans)

    span_dist = {"1-3": 0, "4-6": 0, "7-9": 0, "10-12": 0, "13+": 0}
    for s in spans:
        if s <= 3:
            span_dist["1-3"] += 1
        elif s <= 6:
            span_dist["4-6"] += 1
        elif s <= 9:
            span_dist["7-9"] += 1
        elif s <= 12:
            span_dist["10-12"] += 1
        else:
            span_dist["13+"] += 1

    return {
        "total_managers": len(manager_counts),
        "avg_span_of_control": round(avg_span, 1),
        "max_span": max(spans),
        "min_span": min(spans),
        "span_distribution": span_dist,
    }


def format_human(headcount: dict, attrition: dict, tenure: dict, performance: dict,
                 diversity: dict, comp: dict, spans: dict, period: str) -> str:
    """Format dashboard for human-readable output."""
    lines = []
    lines.append("=" * 70)
    lines.append(f"WORKFORCE DASHBOARD{' - ' + period if period else ''}")
    lines.append("=" * 70)

    # Headcount
    lines.append("")
    lines.append("-" * 70)
    lines.append("HEADCOUNT")
    lines.append("-" * 70)
    lines.append(f"  Active Employees:  {headcount['total_active']}")
    lines.append(f"  Total Records:     {headcount['total_records']}")
    lines.append("")
    lines.append("  By Department:")
    for dept, count in headcount["by_department"].items():
        pct = round(count / max(1, headcount["total_active"]) * 100, 1)
        bar = "#" * int(pct / 2)
        lines.append(f"    {dept:<25} {count:>5} ({pct:>5.1f}%)  {bar}")

    lines.append("")
    lines.append("  By Level:")
    for level, count in headcount["by_level"].items():
        lines.append(f"    {level:<15} {count:>5}")

    # Attrition
    lines.append("")
    lines.append("-" * 70)
    lines.append("ATTRITION")
    lines.append("-" * 70)
    lines.append(f"  Total Exits:       {attrition['total_exits']}")
    lines.append(f"  Voluntary:         {attrition['voluntary_exits']} ({attrition['voluntary_attrition_rate']}%)")
    lines.append(f"  Involuntary:       {attrition['involuntary_exits']} ({attrition['involuntary_attrition_rate']}%)")
    lines.append(f"  Total Rate:        {attrition['total_attrition_rate']}%")
    if attrition["by_department"]:
        lines.append("")
        lines.append("  By Department:")
        for dept, rate in attrition["by_department"].items():
            flag = " <<<" if rate > 15 else ""
            lines.append(f"    {dept:<25} {rate:>5.1f}%{flag}")

    # Tenure
    lines.append("")
    lines.append("-" * 70)
    lines.append("TENURE DISTRIBUTION")
    lines.append("-" * 70)
    lines.append(f"  Average Tenure:    {tenure['avg_tenure_months']:.1f} months ({tenure['avg_tenure_months']/12:.1f} years)")
    lines.append(f"  Median Tenure:     {tenure.get('median_tenure_months', 0)} months")
    if tenure.get("distribution"):
        for band, count in tenure["distribution"].items():
            lines.append(f"    {band:<20} {count:>5}")

    # Performance
    lines.append("")
    lines.append("-" * 70)
    lines.append("PERFORMANCE DISTRIBUTION")
    lines.append("-" * 70)
    lines.append(f"  Average Rating:    {performance.get('avg_rating', 0):.2f}/5.0")
    lines.append(f"  High Performers:   {performance.get('high_performer_pct', 0):.1f}%")
    lines.append(f"  Low Performers:    {performance.get('low_performer_pct', 0):.1f}%")
    if performance.get("distribution_pct"):
        for rating, pct in performance["distribution_pct"].items():
            bar = "#" * int(pct / 2)
            lines.append(f"    Rating {rating}: {pct:>5.1f}%  {bar}")

    # Compensation
    if comp.get("avg_salary"):
        lines.append("")
        lines.append("-" * 70)
        lines.append("COMPENSATION SUMMARY")
        lines.append("-" * 70)
        lines.append(f"  Average Salary:    ${comp['avg_salary']:,}")
        lines.append(f"  Median Salary:     ${comp['median_salary']:,}")
        lines.append(f"  Salary Range:      ${comp['min_salary']:,} - ${comp['max_salary']:,}")
        lines.append(f"  Total Payroll:     ${comp['total_payroll']:,}")

    # Diversity
    if diversity.get("gender_distribution"):
        lines.append("")
        lines.append("-" * 70)
        lines.append("DIVERSITY METRICS")
        lines.append("-" * 70)
        lines.append("  Gender:")
        for gender, pct in diversity["gender_distribution"].items():
            lines.append(f"    {gender:<20} {pct:>5.1f}%")
        if diversity.get("ethnicity_distribution"):
            lines.append("  Ethnicity:")
            for eth, pct in diversity["ethnicity_distribution"].items():
                lines.append(f"    {eth:<20} {pct:>5.1f}%")

    # Span of control
    if spans.get("total_managers"):
        lines.append("")
        lines.append("-" * 70)
        lines.append("SPAN OF CONTROL")
        lines.append("-" * 70)
        lines.append(f"  Total Managers:    {spans['total_managers']}")
        lines.append(f"  Avg Span:          {spans['avg_span_of_control']:.1f}")
        lines.append(f"  Range:             {spans['min_span']} - {spans['max_span']}")

    lines.append("")
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Generate workforce metrics dashboard from HR data."
    )
    parser.add_argument("--file", required=True, help="Path to workforce data CSV")
    parser.add_argument("--period", default=None, help="Reporting period label (e.g., Q1-2026)")
    parser.add_argument("--json", action="store_true", help="Output results as JSON")
    args = parser.parse_args()

    rows = read_csv(args.file)
    if not rows:
        print("Error: No data found in CSV file.", file=sys.stderr)
        sys.exit(1)

    headcount = compute_headcount(rows)
    attrition = compute_attrition(rows)
    tenure = compute_tenure(rows)
    performance = compute_performance(rows)
    diversity = compute_diversity(rows)
    comp = compute_compensation_summary(rows)
    spans = compute_manager_spans(rows)

    if args.json:
        output = {
            "period": args.period,
            "headcount": headcount,
            "attrition": attrition,
            "tenure": tenure,
            "performance": performance,
            "diversity": diversity,
            "compensation": comp,
            "span_of_control": spans,
        }
        print(json.dumps(output, indent=2))
    else:
        print(format_human(headcount, attrition, tenure, performance, diversity, comp, spans, args.period))


if __name__ == "__main__":
    main()