"""Fitz-enz HR metric calculations. Implements: - HCVA (Human Capital Value Added) - HCROI (Human Capital ROI) - QIPS (Quality, Innovation, Productivity, Service) - Five Indexes of Change - Employee Value composite """ from __future__ import annotations from typing import Any, Dict, List, Optional from hr_env.server.company import Company def compute_hcva(company: Company) -> float: """Human Capital Value Added = (Revenue - (Total_Costs - Employment_Cost)) / FTE. Measures the profit contribution per full-time equivalent employee, excluding employment costs from the cost base. """ revenue = company.compute_revenue() employment_cost = company.total_employment_cost # Non-employment operating costs ≈ 55% of revenue total_costs = revenue * 0.55 + employment_cost non_employment_costs = total_costs - employment_cost fte = max(1, company.total_headcount) return (revenue - non_employment_costs) / fte def compute_hcroi(company: Company) -> float: """Human Capital ROI = Revenue / Employment_Cost. Measures revenue generated per dollar of employment cost. """ employment_cost = max(1.0, company.total_employment_cost) return company.compute_revenue() / employment_cost def compute_qips(company: Company) -> Dict[str, float]: """Quality, Innovation, Productivity, Service composite. Returns individual components and weighted average. """ depts = company.departments # Quality: average performance score across company (normalized 0-1) all_active = company.all_active_employees if not all_active: return {"quality": 0, "innovation": 0, "productivity": 0, "service": 0, "composite": 0} avg_perf = sum(e.performance_score for e in all_active) / len(all_active) quality = avg_perf / 5.0 # Innovation: proxy from engineering skill breadth and training investment eng = depts.get("Engineering") if eng and eng.active_employees: unique_skills = set() for e in eng.active_employees: unique_skills.update(e.skills) avg_training = sum(e.training_hours for e in eng.active_employees) / len(eng.active_employees) innovation = min(1.0, (len(unique_skills) / 8) * 0.5 + (avg_training / 40) * 0.5) else: innovation = 0.2 # Productivity: revenue per employee relative to benchmark revenue_per_emp = company.compute_revenue() / max(1, len(all_active)) benchmark_per_emp = company.base_revenue / max(1, len(all_active)) productivity = min(1.0, max(0.0, revenue_per_emp / max(1, benchmark_per_emp))) # Service: proxy from avg engagement (customer-facing depts weighted higher) sales = depts.get("Sales") ops = depts.get("Operations") service_engagement = 0.0 service_count = 0 for dept_name, weight in [("Sales", 2.0), ("Operations", 1.5), ("HR", 1.0), ("Engineering", 0.5), ("Finance", 0.5)]: d = depts.get(dept_name) if d and d.active_employees: service_engagement += d.avg_engagement * weight * len(d.active_employees) service_count += weight * len(d.active_employees) service = (service_engagement / max(1, service_count)) / 100 if service_count > 0 else 0.5 # Composite: weighted average composite = quality * 0.25 + innovation * 0.25 + productivity * 0.30 + service * 0.20 return { "quality": round(quality, 4), "innovation": round(innovation, 4), "productivity": round(productivity, 4), "service": round(service, 4), "composite": round(composite, 4), } def compute_five_indexes( current: Dict[str, float], previous: Optional[Dict[str, float]], ) -> Dict[str, float]: """Five Indexes of Change: Cost, Time, Quantity, Quality, Human Reactions. Compares current quarter metrics to previous quarter. Returns percentage change for each index. """ if previous is None: return {"cost": 0.0, "time": 0.0, "quantity": 0.0, "quality": 0.0, "human_reactions": 0.0} def pct_change(curr: float, prev: float) -> float: if prev == 0: return 0.0 return (curr - prev) / abs(prev) return { "cost": round(pct_change(current.get("employment_cost", 0), previous.get("employment_cost", 0)), 4), "time": round(pct_change(current.get("time_to_fill", 30), previous.get("time_to_fill", 30)), 4), "quantity": round(pct_change(current.get("headcount", 0), previous.get("headcount", 0)), 4), "quality": round(pct_change(current.get("avg_performance", 0), previous.get("avg_performance", 0)), 4), "human_reactions": round(pct_change(current.get("avg_engagement", 0), previous.get("avg_engagement", 0)), 4), } def compute_employee_value(company: Company) -> float: """Employee Value = avg(Productivity + Promotability + Transferability + Retainability). Normalized to 0-1 scale. """ active = company.all_active_employees if not active: return 0.0 total = 0.0 for emp in active: productivity_proxy = emp.performance_score / 5.0 value = (productivity_proxy + emp.promotability + emp.transferability + emp.retainability) / 4.0 total += value return round(total / len(active), 4) def compute_all_metrics(company: Company, previous_snapshot: Optional[Dict] = None) -> Dict[str, Any]: """Compute all Fitz-enz metrics for the current quarter.""" hcva = compute_hcva(company) hcroi = compute_hcroi(company) qips = compute_qips(company) employee_value = compute_employee_value(company) # Build current snapshot for five indexes current_snapshot = { "employment_cost": company.total_employment_cost, "time_to_fill": 30, # Default, could be tracked "headcount": company.total_headcount, "avg_performance": sum(e.performance_score for e in company.all_active_employees) / max(1, len(company.all_active_employees)), "avg_engagement": sum(e.engagement for e in company.all_active_employees) / max(1, len(company.all_active_employees)), } five_indexes = compute_five_indexes(current_snapshot, previous_snapshot) return { "hcva": round(hcva, 2), "hcroi": round(hcroi, 4), "qips": qips, "five_indexes": five_indexes, "employee_value": employee_value, "snapshot": current_snapshot, }