Spaces:
Paused
Paused
| """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, | |
| } | |